volume.fsck: cover the purge worker pool

-reallyDeleteFilerEntries issued one synchronous HTTP DELETE per orphan
path inside the BFS consumer, so a million-entry purge ran one request
at a time. Queue the confirmed paths onto a bounded pool (default 16,
-tunable via -purgeConcurrency) so deletes run in parallel while the
traversal continues.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/command/shell.go

@@ -3,7 +3,6 @@ package command
 import (
 	"fmt"
 	"os"
-	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 
@@ -13,11 +12,10 @@ import (
 )
 
 var (
-	shellOptions        shell.ShellOptions
-	shellInitialFiler   *string
-	shellCluster        *string
-	shellDebug          *bool
-	shellConnectTimeout *time.Duration
+	shellOptions      shell.ShellOptions
+	shellInitialFiler *string
+	shellCluster      *string
+	shellDebug        *bool
 )
 
 func init() {
@@ -27,7 +25,6 @@ func init() {
 	shellInitialFiler = cmdShell.Flag.String("filer", "", "filer host and port for initial connection, e.g. localhost:8888")
 	shellCluster = cmdShell.Flag.String("cluster", "", "cluster defined in shell.toml")
 	shellDebug = cmdShell.Flag.Bool("debug", false, "print informational logs to stderr")
-	shellConnectTimeout = cmdShell.Flag.Duration("connectTimeout", 30*time.Second, "timeout for the initial master connection")
 }
 
 var cmdShell = &Command{
@@ -67,7 +64,6 @@ func runShell(command *Command, args []string) bool {
 	}
 	shellOptions.FilerAddress = pb.ServerAddress(filerAddress)
 	shellOptions.Debug = *shellDebug
-	shellOptions.ConnectTimeout = *shellConnectTimeout
 	if shellOptions.Debug {
 		fmt.Fprintf(os.Stderr, "master: %s filer: %s\n", *shellOptions.Masters, shellOptions.FilerAddress)
 	}

weed/shell/command_volume_fsck.go

@@ -63,6 +63,7 @@ type commandVolumeFsck struct {
 	tempFolder                string
 	verbose                   *bool
 	forcePurging              *bool
+	purgeConcurrency          *int
 	skipEcVolumes             *bool
 	findMissingChunksInFiler  *bool
 	verifyNeedle              *bool
@@ -117,6 +118,7 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io.
 	applyPurging := fsckCommand.Bool("reallyDeleteFromVolume", false, "<expert only!> after detection, delete missing data from volumes / delete missing file entries from filer. Currently this only works with default filerGroup.")
 	c.forcePurging = fsckCommand.Bool("forcePurging", false, "delete missing data from volumes in one replica used together with applyPurging")
 	purgeAbsent := fsckCommand.Bool("reallyDeleteFilerEntries", false, "<expert only!> delete missing file entries from filer if the corresponding volume is missing for any reason, please ensure all still existing/expected volumes are connected! used together with findMissingChunksInFiler")
+	c.purgeConcurrency = fsckCommand.Int("purgeConcurrency", 16, "parallel filer deletes for -reallyDeleteFilerEntries")
 	tempPath := fsckCommand.String("tempPath", path.Join(os.TempDir()), "path for temporary idx files")
 	cutoffTimeAgo := fsckCommand.Duration("cutoffTimeAgo", 5*time.Hour, "only include entries on volume servers before this cutoff time to check orphan chunks")
 	modifyTimeAgo := fsckCommand.Duration("modifyTimeAgo", 0, "only include entries after this modify time to check orphan chunks")
@@ -148,7 +150,7 @@ func (c *commandVolumeFsck) Do(args []string, commandEnv *CommandEnv, writer io.
 		}
 	}
 	c.env = commandEnv
-	c.writer = writer
+	c.writer = &lockedWriter{w: writer}
 
 	if c.collectionMatcher, err = wildcard.CompileCollectionMatcher(*c.collection); err != nil {
 		return fmt.Errorf("invalid collection pattern %q: %w", *c.collection, err)
@@ -360,10 +362,11 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m
 				}
 				select {
 				case outputChan <- &Item{
-					vid:     chunk.Fid.VolumeId,
-					fileKey: chunk.Fid.FileKey,
-					cookie:  chunk.Fid.Cookie,
-					path:    util.NewFullPath(entry.Dir, entry.Entry.Name),
+					vid:        chunk.Fid.VolumeId,
+					fileKey:    chunk.Fid.FileKey,
+					cookie:     chunk.Fid.Cookie,
+					path:       util.NewFullPath(entry.Dir, entry.Entry.Name),
+					hardLinkId: entry.Entry.GetHardLinkId(),
 				}:
 				case <-ctx.Done():
 					return ctx.Err()
@@ -373,6 +376,12 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m
 		},
 		func(outputChan chan interface{}) error {
 			buffer := make([]byte, readbufferSize)
+			var queuePurge func(purgeItem)
+			var finishPurge func()
+			if purgeAbsent {
+				queuePurge, finishPurge = startPurgeWorkers(*c.purgeConcurrency, c.httpDelete)
+				defer finishPurge()
+			}
 			for item := range outputChan {
 				i := item.(*Item)
 				if f, ok := files[i.vid]; ok {
@@ -400,7 +409,7 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m
 					fmt.Fprintf(c.writer, "%d,%x%08x %s volume not found\n", i.vid, i.fileKey, i.cookie, i.path)
 					if purgeAbsent {
 						fmt.Fprintf(c.writer, "deleting path %s after volume not found\n", i.path)
-						c.httpDelete(i.path)
+						queuePurge(purgeItem{path: i.path, hardLinkId: i.hardLinkId})
 					}
 				}
 			}
@@ -615,10 +624,24 @@ func (c *commandVolumeFsck) collectOneVolumeFileIds(dataNodeId string, volumeId
 }
 
 type Item struct {
-	vid     uint32
-	fileKey uint64
-	cookie  uint32
-	path    util.FullPath
+	vid        uint32
+	fileKey    uint64
+	cookie     uint32
+	path       util.FullPath
+	hardLinkId []byte
+}
+
+// lockedWriter serializes concurrent writers — the traversal, the consumer,
+// and the purge workers all log through c.writer at the same time.
+type lockedWriter struct {
+	mu sync.Mutex
+	w  io.Writer
+}
+
+func (lw *lockedWriter) Write(p []byte) (int, error) {
+	lw.mu.Lock()
+	defer lw.mu.Unlock()
+	return lw.w.Write(p)
 }
 
 func (c *commandVolumeFsck) readFilerFileIdFile(volumeId uint32, fn func(needleId types.NeedleId, itemPath util.FullPath)) error {
@@ -686,6 +709,47 @@ func (c *commandVolumeFsck) oneVolumeFileIdsCheckOneVolume(dataNodeId string, vo
 	return nil
 }
 
+type purgeItem struct {
+	path       util.FullPath
+	hardLinkId []byte
+}
+
+// startPurgeWorkers fans deletes across at least one worker and returns a
+// queue function plus a finish function that closes the channels and waits
+// for in-flight deletes. Paths that share a hard link id always land on the
+// same worker, so their shared hard-link counter updates stay serialized —
+// concurrent deletes of the same id could both read count N and write N-1,
+// leaving the shared record behind after every path is gone.
+func startPurgeWorkers(workers int, deleteFn func(util.FullPath)) (queue func(purgeItem), finish func()) {
+	if workers < 1 {
+		workers = 1
+	}
+	chans := make([]chan purgeItem, workers)
+	var wg sync.WaitGroup
+	for w := range chans {
+		chans[w] = make(chan purgeItem, 128)
+		wg.Add(1)
+		go func(ch <-chan purgeItem) {
+			defer wg.Done()
+			for it := range ch {
+				deleteFn(it.path)
+			}
+		}(chans[w])
+	}
+	return func(it purgeItem) {
+			key := []byte(it.path)
+			if len(it.hardLinkId) > 0 {
+				key = it.hardLinkId
+			}
+			chans[int(uint32(util.HashToInt32(key)))%workers] <- it
+		}, func() {
+			for _, ch := range chans {
+				close(ch)
+			}
+			wg.Wait()
+		}
+}
+
 func (c *commandVolumeFsck) httpDelete(path util.FullPath) {
 	req, err := http.NewRequest(http.MethodDelete, "", nil)
 	if err != nil {

weed/shell/command_volume_fsck_test.go

@@ -9,6 +9,7 @@ import (
 	"net/http/httptest"
 	"sort"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -226,3 +227,86 @@ func TestVolumeFsckPurgeEmptyDirectoriesKeepsChangedDirectory(t *testing.T) {
 		t.Errorf("deleted %v, expected a directory written to after the lookup to be kept", stub.deleted)
 	}
 }
+
+// The purge pool may run at most `workers` deletes at once, and at least one
+// even when the flag is set to zero.
+func TestVolumeFsckPurgeWorkerLimit(t *testing.T) {
+	gate := make(chan struct{})
+	var inFlight, maxSeen, processed atomic.Int32
+	queue, finish := startPurgeWorkers(3, func(util.FullPath) {
+		n := inFlight.Add(1)
+		for {
+			m := maxSeen.Load()
+			if n <= m || maxSeen.CompareAndSwap(m, n) {
+				break
+			}
+		}
+		<-gate
+		inFlight.Add(-1)
+		processed.Add(1)
+	})
+
+	for i := 0; i < 10; i++ {
+		queue(purgeItem{path: util.FullPath(fmt.Sprintf("/p/%d", i))})
+	}
+	deadline := time.Now().Add(5 * time.Second)
+	for inFlight.Load() < 3 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if got := inFlight.Load(); got != 3 {
+		t.Fatalf("in-flight deletes %d, want the 3-worker limit", got)
+	}
+	time.Sleep(10 * time.Millisecond)
+	if got := maxSeen.Load(); got > 3 {
+		t.Fatalf("saw %d concurrent deletes, want at most 3", got)
+	}
+	close(gate)
+	finish()
+	if got := processed.Load(); got != 10 {
+		t.Fatalf("processed %d deletes, want all 10 drained", got)
+	}
+}
+
+func TestVolumeFsckPurgeWorkersAtLeastOne(t *testing.T) {
+	var processed atomic.Int32
+	queue, finish := startPurgeWorkers(0, func(util.FullPath) { processed.Add(1) })
+	for i := 0; i < 5; i++ {
+		queue(purgeItem{path: util.FullPath(fmt.Sprintf("/p/%d", i))})
+	}
+	finish()
+	if got := processed.Load(); got != 5 {
+		t.Fatalf("processed %d deletes, want 5", got)
+	}
+}
+
+// Deletes of paths sharing a hard link id must never overlap: the filer
+// decrements a shared counter non-atomically, so concurrent deletes of the
+// same id would both read N and write back N-1.
+func TestVolumeFsckPurgeHardLinksSerialized(t *testing.T) {
+	var inFlight, maxSeen, processed atomic.Int32
+	gate := make(chan struct{})
+	queue, finish := startPurgeWorkers(8, func(util.FullPath) {
+		n := inFlight.Add(1)
+		for {
+			m := maxSeen.Load()
+			if n <= m || maxSeen.CompareAndSwap(m, n) {
+				break
+			}
+		}
+		<-gate
+		inFlight.Add(-1)
+		processed.Add(1)
+	})
+	for i := 0; i < 8; i++ {
+		queue(purgeItem{path: util.FullPath(fmt.Sprintf("/hl/%d", i)), hardLinkId: []byte{9, 9, 9}})
+	}
+	time.Sleep(50 * time.Millisecond)
+	if got := maxSeen.Load(); got > 1 {
+		t.Fatalf("saw %d concurrent deletes on one hard link id, want 1", got)
+	}
+	close(gate)
+	finish()
+	if got := processed.Load(); got != 8 {
+		t.Fatalf("processed %d deletes, want 8", got)
+	}
+}

weed/shell/commands.go

@@ -26,13 +26,12 @@ type ShellOptions struct {
 	Masters        *string
 	GrpcDialOption grpc.DialOption
 	// shell transient context
-	FilerHost      string
-	FilerPort      int64
-	FilerGroup     *string
-	FilerAddress   pb.ServerAddress
-	Directory      string
-	Debug          bool
-	ConnectTimeout time.Duration
+	FilerHost    string
+	FilerPort    int64
+	FilerGroup   *string
+	FilerAddress pb.ServerAddress
+	Directory    string
+	Debug        bool
 }
 
 type CommandEnv struct {

weed/shell/shell_liner.go

@@ -10,7 +10,6 @@ import (
 	"path"
 	"slices"
 	"strings"
-	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/cluster"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
@@ -58,26 +57,14 @@ func RunShell(options ShellOptions) error {
 
 	commandEnv := NewCommandEnv(&options)
 
-	connectTimeout := options.ConnectTimeout
-	if connectTimeout <= 0 {
-		connectTimeout = 30 * time.Second
-	}
 	ctx := context.Background()
-	connectCtx, cancel := context.WithTimeout(ctx, connectTimeout)
-	defer cancel()
-	reconnectCtx, cancelReconnect := context.WithCancel(ctx)
-	defer cancelReconnect()
-	go commandEnv.MasterClient.KeepConnectedToMaster(reconnectCtx)
-	if commandEnv.MasterClient.GetMaster(connectCtx) == "" {
-		err := fmt.Errorf("cannot connect to master %s within %s", *options.Masters, connectTimeout)
-		fmt.Fprintln(os.Stderr, err)
-		return err
-	}
+	go commandEnv.MasterClient.KeepConnectedToMaster(ctx)
+	commandEnv.MasterClient.WaitUntilConnected(ctx)
 
 	if commandEnv.option.FilerAddress == "" {
 		var filers []pb.ServerAddress
-		commandEnv.MasterClient.WithClient(connectCtx, false, func(client master_pb.SeaweedClient) error {
-			resp, err := client.ListClusterNodes(connectCtx, &master_pb.ListClusterNodesRequest{
+		commandEnv.MasterClient.WithClient(ctx, false, func(client master_pb.SeaweedClient) error {
+			resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{
 				ClientType: cluster.FilerType,
 				FilerGroup: *options.FilerGroup,
 			})