Handle concurrent access to the inFlight list

repository/master_index.go

@@ -139,18 +139,19 @@ func (mi *MasterIndex) Current() *Index {
 	return newIdx
 }
 
-// AddInFlight add the given IDs to the list of in-flight IDs.
-func (mi *MasterIndex) AddInFlight(IDs ...backend.ID) {
+// AddInFlight add the given ID to the list of in-flight IDs. An error is
+// returned when the ID is already in the list.
+func (mi *MasterIndex) AddInFlight(id backend.ID) error {
 	mi.inFlight.Lock()
 	defer mi.inFlight.Unlock()
 
-	ids := backend.IDs(IDs)
-
-	debug.Log("MasterIndex.AddInFlight", "adding %v", ids)
-
-	for _, id := range ids {
-		mi.inFlight.Insert(id)
+	debug.Log("MasterIndex.AddInFlight", "adding %v", id)
+	if mi.inFlight.Has(id) {
+		return fmt.Errorf("%v is already in flight", id)
 	}
+
+	mi.inFlight.Insert(id)
+	return nil
 }
 
 // IsInFlight returns true iff the id is contained in the list of in-flight IDs.

repository/repository.go

@@ -305,15 +305,9 @@ func (r *Repository) SaveAndEncrypt(t pack.BlobType, data []byte, id *backend.ID
 		return backend.ID{}, err
 	}
 
-	// check if this id is already been saved by another goroutine
-	if r.idx.IsInFlight(*id) {
-		debug.Log("Repo.Save", "blob %v is already being saved", id.Str())
-		return *id, nil
-	}
-
 	// add this id to the list of in-flight chunk ids.
 	debug.Log("Repo.Save", "add %v to list of in-flight IDs", id.Str())
-	r.idx.AddInFlight(*id)
+	err = r.idx.AddInFlight(*id)
 	if err != nil {
 		debug.Log("Repo.Save", "another goroutine is already working on %v (%v) does already exist", id.Str, t)
 		return *id, nil