shell: warn when fs.mergeVolumes source holds only orphan needles (#11310)

* shell: warn when fs.mergeVolumes source holds only orphan needles

fs.mergeVolumes traverses filer entries, so a source volume whose
needles are all orphans — filer entries lost to a crashed write or a
wiped filer store — produces only the plan header and exits 0: no move,
no skip, no error. Operators read that as a successful merge while the
real cleanup (volume.fsck) never runs, and dat>idx volumes keep coming
back read-only after restarts.

Count the source-volume needles seen during traversal and, when a plan
source was never seen but its index still reports needles, print a
warning pointing at volume.fsck. Dry-run warns too.

* shell: make needle counting concurrency-safe and count manifest sub-chunks

TraverseBfs runs its callbacks from five workers, so the plain
needlesSeen map raced between source-heavy merges (fatal concurrent
map writes). All increments now funnel through a mutex-guarded
recordSeen closure.

Manifest sub-chunks that live on planned source volumes are now
recorded too — rewriteManifestChunk visits them (including dry-run
and capacity-skipped ones) but previously never marked their source,
which produced false 'orphan needles' warnings for sources whose
chunks were all reached through manifests.

* shell: extract sourceNeedleCounter so the concurrency test covers the production path

The orphan-warning recording was a closure local to Do, so
TestWarnUnreferencedSources_ConcurrentRecording could only exercise a
test-local copy of it — a regression in the production mutex would pass
the test. Lift the map and mutex into a sourceNeedleCounter type with
record/count methods and use it from Do and the test, so the -race test
now drives the actual recording path. Trim the verbose comments added
with the warning while here.

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/shell/command_fs_merge_volumes.go

@@ -146,7 +146,9 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
 		}
 	}
 
-	return commandEnv.WithFilerClient(false, func(filerClient filer_pb.SeaweedFilerClient) error {
+	seen := newSourceNeedleCounter()
+
+	if err := commandEnv.WithFilerClient(false, func(filerClient filer_pb.SeaweedFilerClient) error {
 		return filer_pb.TraverseBfs(context.Background(), commandEnv, util.FullPath(dir), func(parentPath util.FullPath, entry *filer_pb.Entry) error {
 			if entry.IsDirectory {
 				return nil
@@ -176,7 +178,10 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
 					}
 					oldManifestFid := chunk.GetFileIdString()
 					oldManifestVid := chunk.Fid.VolumeId
-					newChunk, changed, rewritten, mErr := c.rewriteManifestChunk(context.Background(), commandEnv, lookupFn, plan, entryPath, chunk, *apply)
+					if vid := needle.VolumeId(oldManifestVid); plan.isSource(vid) {
+						seen.record(vid)
+					}
+					newChunk, changed, rewritten, mErr := c.rewriteManifestChunk(context.Background(), commandEnv, lookupFn, plan, entryPath, chunk, *apply, seen.record)
 					if mErr != nil {
 						fmt.Printf("failed to rewrite manifest %s(%s): %v\n", entryPath, oldManifestFid, mErr)
 						continue
@@ -199,6 +204,7 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
 				if !plan.isSource(chunkVolumeId) {
 					continue
 				}
+				seen.record(chunkVolumeId)
 
 				oldFid := chunk.GetFileIdString()
 				oldVid := chunk.Fid.VolumeId
@@ -234,7 +240,51 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer
 			}
 			return nil
 		})
-	})
+	}); err != nil {
+		return err
+	}
+
+	c.warnUnreferencedSources(writer, plan, seen, dir)
+	return nil
+}
+
+// warnUnreferencedSources warns when a plan source's index still holds needles
+// but no filer entry referenced any during traversal — the merge moves nothing
+// and the real cleanup is volume.fsck.
+func (c *commandFsMergeVolumes) warnUnreferencedSources(writer io.Writer, plan *mergePlan, seen *sourceNeedleCounter, dir string) {
+	for src := range plan.targets {
+		if seen.count(src) > 0 {
+			continue
+		}
+		info := c.volumes[src]
+		if info == nil || info.FileCount == 0 {
+			continue
+		}
+		fmt.Fprintf(writer, "warning: volume %d has %d needle(s) in its index but no filer entries reference them under %s — nothing merged (orphan needles? run volume.fsck)\n", src, info.FileCount, dir)
+	}
+}
+
+// sourceNeedleCounter records plan-source needles seen during filer traversal.
+// TraverseBfs runs callbacks on concurrent workers, so record is mutex-guarded.
+type sourceNeedleCounter struct {
+	mu   sync.Mutex
+	seen map[needle.VolumeId]int
+}
+
+func newSourceNeedleCounter() *sourceNeedleCounter {
+	return &sourceNeedleCounter{seen: make(map[needle.VolumeId]int)}
+}
+
+func (s *sourceNeedleCounter) record(vid needle.VolumeId) {
+	s.mu.Lock()
+	s.seen[vid]++
+	s.mu.Unlock()
+}
+
+func (s *sourceNeedleCounter) count(vid needle.VolumeId) int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.seen[vid]
 }
 
 // orphanedNeedle is a needle no filer entry references any more: either a
@@ -671,6 +721,7 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk(
 	entryPath util.FullPath,
 	chunk *filer_pb.FileChunk,
 	apply bool,
+	recordSeen func(needle.VolumeId),
 ) (*filer_pb.FileChunk, bool, rewrittenNeedles, error) {
 	if !chunk.IsChunkManifest {
 		return chunk, false, rewrittenNeedles{}, fmt.Errorf("not a manifest chunk: %s", chunk.GetFileIdString())
@@ -698,7 +749,10 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk(
 		if sub.IsChunkManifest {
 			oldSubManifestFid := sub.GetFileIdString()
 			oldSubManifestVid := sub.Fid.VolumeId
-			newSub, changed, nested, rErr := c.rewriteManifestChunk(ctx, commandEnv, lookupFn, plan, entryPath, sub, apply)
+			if vid := needle.VolumeId(oldSubManifestVid); plan.isSource(vid) {
+				recordSeen(vid)
+			}
+			newSub, changed, nested, rErr := c.rewriteManifestChunk(ctx, commandEnv, lookupFn, plan, entryPath, sub, apply, recordSeen)
 			if rErr != nil {
 				abandon()
 				return chunk, false, rewrittenNeedles{}, rErr
@@ -722,6 +776,7 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk(
 		if !plan.isSource(subVid) {
 			continue
 		}
+		recordSeen(subVid)
 		oldSubFid := sub.GetFileIdString()
 		oldSubVid := sub.Fid.VolumeId
 		toVid, ok := plan.allocate(subVid, sub.Size)

weed/shell/command_fs_merge_volumes_test.go

@@ -2,6 +2,7 @@ package shell
 
 import (
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
@@ -271,3 +272,77 @@ func TestManifestMayReferencePlan(t *testing.T) {
 		}
 	}
 }
+
+// A source whose index still holds needles but that the traversal never saw is
+// made entirely of orphan needles — the merge moves nothing, but the operator
+// must be told the real cleanup is fsck.
+func TestWarnUnreferencedSources(t *testing.T) {
+	vol := &master_pb.VolumeInformationMessage{Id: 203, FileCount: 18}
+	c := newMergeCmd(250000, vol)
+	plan := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{
+		needle.VolumeId(203): {needle.VolumeId(187)},
+	}}
+
+	var sb strings.Builder
+	c.warnUnreferencedSources(&sb, plan, newSourceNeedleCounter(), "/buckets/test")
+	out := sb.String()
+	if !strings.Contains(out, "volume 203") || !strings.Contains(out, "orphan needles") {
+		t.Fatalf("expected orphan warning, got %q", out)
+	}
+
+	// Needles seen during traversal: no warning.
+	seen := newSourceNeedleCounter()
+	for i := 0; i < 7; i++ {
+		seen.record(needle.VolumeId(203))
+	}
+	sb.Reset()
+	c.warnUnreferencedSources(&sb, plan, seen, "/buckets/test")
+	if sb.Len() != 0 {
+		t.Fatalf("expected no warning when needles were seen, got %q", sb.String())
+	}
+
+	// Zero FileCount (truly empty volume): nothing to report.
+	emptyVol := &master_pb.VolumeInformationMessage{Id: 204}
+	c2 := newMergeCmd(250000, emptyVol)
+	plan2 := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{
+		needle.VolumeId(204): {needle.VolumeId(187)},
+	}}
+	sb.Reset()
+	c2.warnUnreferencedSources(&sb, plan2, newSourceNeedleCounter(), "/buckets/test")
+	if sb.Len() != 0 {
+		t.Fatalf("expected no warning for empty volume, got %q", sb.String())
+	}
+}
+
+// Concurrent BFS workers increment the production counter through record — the
+// count must stay correct under -race with source-heavy inputs.
+func TestWarnUnreferencedSources_ConcurrentRecording(t *testing.T) {
+	vol := &master_pb.VolumeInformationMessage{Id: 203, FileCount: 1000}
+	c := newMergeCmd(250000, vol)
+	plan := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{
+		needle.VolumeId(203): {needle.VolumeId(187)},
+	}}
+
+	seen := newSourceNeedleCounter()
+	var wg sync.WaitGroup
+	for w := 0; w < 5; w++ { // TraverseBfs worker count
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := 0; i < 200; i++ {
+				seen.record(needle.VolumeId(203))
+			}
+		}()
+	}
+	wg.Wait()
+
+	if got := seen.count(needle.VolumeId(203)); got != 1000 {
+		t.Fatalf("lost increments under concurrency: %d", got)
+	}
+
+	var sb strings.Builder
+	c.warnUnreferencedSources(&sb, plan, seen, "/buckets/test")
+	if sb.Len() != 0 {
+		t.Fatalf("expected no warning when all needles were seen, got %q", sb.String())
+	}
+}