MB-74454: Fix hangs in drop keys error cases

index/scorch/merge.go

@@ -103,6 +103,12 @@ OUTER:
 				// lets get started
 				startTime := time.Now()
 				err := s.planMergeAtSnapshot(ctrlMsg, ourSnapshot)
+
+				done, hasDone := ctrlMsg.ctx.Value(mergeDoneKey).(chan error)
+				if hasDone {
+					done <- err
+				}
+
 				if err != nil {
 					atomic.StoreUint64(&s.iStats.mergeEpoch, 0)
 					if err == segment.ErrClosed {
@@ -128,6 +134,10 @@ OUTER:
 					))
 					_ = ourSnapshot.DecRef()
 					atomic.AddUint64(&s.stats.TotFileMergeLoopErr, 1)
+					// the error has been reported, so the request is not retried
+					if hasDone {
+						ctrlMsg = nil
+					}
 					continue OUTER
 				}
 
@@ -312,10 +322,8 @@ func (s *Scorch) planMergeAtSnapshot(ctrlMsg *mergerCtrl, ourSnapshot *IndexSnap
 			atomic.AddUint64(&s.stats.TotFileMergePlanErr, 1)
 			return fmt.Errorf("merge planning err: %v", err)
 		}
-	}
-
-	// default to making a merge plan if a custom one is not provided
-	if mergePlan == nil || len(mergePlan.Tasks) == 0 {
+	} else {
+		// default to making a merge plan if a custom one is not provided
 		// build list of persisted segments in this snapshot
 		var onlyPersistedSnapshots []mergeplan.Segment
 		for _, segmentSnapshot := range ourSnapshot.segment {
@@ -333,12 +341,12 @@ func (s *Scorch) planMergeAtSnapshot(ctrlMsg *mergerCtrl, ourSnapshot *IndexSnap
 			atomic.AddUint64(&s.stats.TotFileMergePlanErr, 1)
 			return fmt.Errorf("merge planning err: %v", err)
 		}
+	}
 
-		if mergePlan == nil || len(mergePlan.Tasks) == 0 {
-			// nothing to do
-			atomic.AddUint64(&s.stats.TotFileMergePlanNone, 1)
-			return nil
-		}
+	if mergePlan == nil || len(mergePlan.Tasks) == 0 {
+		// nothing to do
+		atomic.AddUint64(&s.stats.TotFileMergePlanNone, 1)
+		return nil
 	}
 
 	atomic.AddUint64(&s.stats.TotFileMergePlanOk, 1)
@@ -349,13 +357,6 @@ func (s *Scorch) planMergeAtSnapshot(ctrlMsg *mergerCtrl, ourSnapshot *IndexSnap
 	go cw.listen()
 
 	var err error
-	defer func() {
-		// send error to done channel if present
-		if done, ok := cw.ctx.Value(mergeDoneKey).(chan error); ok {
-			done <- err
-		}
-	}()
-
 	numBatches := len(mergePlan.Tasks)
 	mergeBatches := make([]*mergeBatch, numBatches)
 	defer func() {

index/scorch/scorch.go

@@ -1386,7 +1386,7 @@ func (s *Scorch) DropFileWriterIDs(ids map[string]struct{}) error {
 	s.rootLock.Lock()
 	// create a done channel to ensure success of merge
 	ctx := context.Background()
-	doneCh := make(chan error)
+	doneCh := make(chan error, 1)
 	ctx = context.WithValue(ctx, mergeDoneKey, doneCh)
 
 	// PARTIAL ROLLBACK WILL NOT BE SUPPORTED DURING THIS OPERATION
@@ -1472,15 +1472,22 @@ func (s *Scorch) DropFileWriterIDs(ids map[string]struct{}) error {
 	// any races
 	ctx = context.WithValue(ctx, mergePlanFuncKey, mergePlanner)
 
-	// trigger the merge with the force merge plan
-	s.forceMergeRequestCh <- &mergerCtrl{
-		ctx: ctx,
-	}
 	s.rootLock.Unlock()
 
+	// trigger the merge with the force merge plan. Sent without holding
+	// rootLock, as the merger may need it to finish a merge already queued
+	select {
+	case s.forceMergeRequestCh <- &mergerCtrl{ctx: ctx}:
+	case <-s.closeCh:
+		return segment.ErrClosed
+	}
+
 	// blockingly wait for merge to complete
-	err = <-doneCh
-	close(doneCh)
+	select {
+	case err = <-doneCh:
+	case <-s.closeCh:
+		return segment.ErrClosed
+	}
 	if err != nil {
 		return err
 	}
@@ -1498,6 +1505,33 @@ func (s *Scorch) DropFileWriterIDs(ids map[string]struct{}) error {
 // returns an error after a timeout. It does so by checking the index directory every 5 seconds
 // for the presence of the given files, and returns once they are no longer present.
 func (s *Scorch) waitTillFileCleanup(filePaths []string) error {
+	pending := make(map[string]struct{}, len(filePaths))
+	for _, filePath := range filePaths {
+		pending[filePath] = struct{}{}
+	}
+
+	// reports whether any of the given files is still in the index directory
+	filesPresent := func() (bool, error) {
+		files, err := os.ReadDir(s.path)
+		if err != nil {
+			return false, err
+		}
+		for _, f := range files {
+			if _, ok := pending[f.Name()]; ok {
+				return true, nil
+			}
+		}
+		return false, nil
+	}
+
+	if len(pending) == 0 {
+		return nil
+	}
+	present, err := filesPresent()
+	if err != nil || !present {
+		return err
+	}
+
 	ticker := time.NewTicker(5 * time.Second)
 	defer ticker.Stop()
 
@@ -1506,23 +1540,14 @@ func (s *Scorch) waitTillFileCleanup(filePaths []string) error {
 	for {
 		select {
 		case <-ticker.C:
-			files, err := os.ReadDir(s.path)
-			if err != nil {
+			present, err := filesPresent()
+			if err != nil || !present {
 				return err
 			}
-			for _, f := range files {
-				fname := f.Name()
-				if filepath.Ext(fname) == ".zap" {
-					for _, filePath := range filePaths {
-						if fname == filePath {
-							continue
-						}
-					}
-				}
-			}
-			return nil
 		case <-timeout:
 			return fmt.Errorf("timeout waiting for file cleanup for files: %v", filePaths)
+		case <-s.closeCh:
+			return segment.ErrClosed
 		}
 	}
 }

index_impl.go

@@ -516,9 +516,9 @@ func (i *indexImpl) FileWriterIDsInUse() (map[string]struct{}, error) {
 func (i *indexImpl) DropFileWriterIDs(ids map[string]struct{}) error {
 	i.mutex.Lock()
 	if _, ok := ids[i.meta.fileReader.Id()]; ok {
-		var err error
-		err = i.meta.UpdateWriter(i.path)
+		err := i.meta.UpdateWriter(i.path)
 		if err != nil {
+			i.mutex.Unlock()
 			return err
 		}
 	}