operations: fix hang when deleting files and a fatal error occurs

When a fatal error such as --max-delete being reached stopped the
deletions, the deleters returned without receiving the rest of the
objects, so whatever was sending them blocked forever once the
channel filled up.

For example "rclone delete --max-delete 1" on a directory with more
files than --checkers hung instead of returning the error.

The deleters now keep receiving objects without deleting them after a
fatal error.

fs/operations/operations.go

@@ -608,6 +608,10 @@ func DeleteFilesWithBackupDir(ctx context.Context, toBeDeleted fs.ObjectsChan, b
 		go func() {
 			defer wg.Done()
 			for dst := range toBeDeleted {
+				// Empty the channel on fatal error
+				if fatalErrorCount.Load() != 0 {
+					continue
+				}
 				err := DeleteFileWithBackupDir(ctx, dst, backupDir)
 				if err != nil {
 					errorCount.Add(1)
@@ -616,7 +620,6 @@ func DeleteFilesWithBackupDir(ctx context.Context, toBeDeleted fs.ObjectsChan, b
 					if fserrors.IsFatalError(err) {
 						fs.Errorf(dst, "Got fatal error on delete: %s", err)
 						fatalErrorCount.Add(1)
-						return
 					}
 				}
 			}
@@ -1593,7 +1596,7 @@ func Rmdirs(ctx context.Context, f fs.Fs, dir string, leaveRoot bool) error {
 		}
 		fs.Debugf(nil, "removing %d level %d directories", len(dirs), level)
 		sort.Strings(dirs)
-		g, gCtx := errgroup.WithContext(ctx)
+		g, gCtx := errgroup.WithContext(context.Background())
 		g.SetLimit(ci.Checkers)
 		for _, dir := range dirs {
 			// End early if error
@@ -2494,7 +2497,7 @@ func DirMove(ctx context.Context, f fs.Fs, srcRemote, dstRemote string) (err err
 		newPath string
 	}
 	renames := make(chan rename, ci.Checkers)
-	g, gCtx := errgroup.WithContext(context.Background())
+	g, gCtx := errgroup.WithContext(ctx)
 	for range ci.Checkers {
 		g.Go(func() error {
 			for job := range renames {

fs/operations/operations_test.go

@@ -427,6 +427,32 @@ func TestDelete(t *testing.T) {
 	r.CheckRemoteItems(t, file3)
 }
 
+// Check Delete doesn't hang when a fatal error stops the deletions
+// before all the objects have been sent to the deleters
+func TestDeleteFatalError(t *testing.T) {
+	ctx := context.Background()
+	ctx, ci := fs.AddConfig(ctx)
+	ci.Checkers = 2
+	ci.MaxDelete = 1
+	r := fstest.NewRun(t)
+	// More files than the deleters' channel can hold
+	for i := range 20 {
+		r.WriteObject(ctx, fmt.Sprintf("file%d", i), "x", t1)
+	}
+
+	done := make(chan error, 1)
+	go func() {
+		done <- operations.Delete(ctx, r.Fremote)
+	}()
+	select {
+	case err := <-done:
+		require.Error(t, err)
+		assert.True(t, fserrors.IsFatalError(err), err)
+	case <-time.After(30 * time.Second):
+		t.Fatal("Delete didn't return - deadlocked sending to the deleters")
+	}
+}
+
 func isChunker(f fs.Fs) bool {
 	return strings.HasPrefix(f.Name(), "TestChunker")
 }