shell: fs.verify -pruneEntries deletes entries whose needles are lost (#11338)

* shell: fs.verify -pruneEntries deletes entries whose needles are lost

* shell: harden fs.verify -pruneEntries guards; VolumeNeedleStatus returns NotFound for absent needles

* shell: resolve chunk manifests in fs.verify metadata path; require confirmed deletion before counting prunes

* shell: anchor fs.verify legacy missing-needle error matching

* shell: keep fs.verify metadata scan alive on manifest resolution failures

* shell: classify EC missing needles and keep manifest failures unverified

VolumeNeedleStatus now canonicalizes erasure_coding.NotFoundError to
codes.NotFound, so absent needles in EC volumes reach the prune path
through the same stable contract as regular volumes. The client-side
isNeedleMissingError keeps recognizing the legacy wrapped EC shape
("locate in local ec volume: ... needle not found") for mixed-version
clusters.

A chunk manifest that fails to resolve is now an entry-level
verification failure even when the raw top-level chunks are healthy:
the file is not fully readable without the manifest. Raw chunks are
still verified on a resolution failure so a missing top-level manifest
needle is classified and can be pruned. The per-entry logic is
extracted into resolveAndVerify for testability.

* shell: trim fs.verify prune comments

---------

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

weed/server/volume_grpc_admin.go

@@ -24,6 +24,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
@@ -497,10 +498,13 @@ func (vs *VolumeServer) VolumeNeedleStatus(ctx context.Context, req *volume_serv
 		count, err = vs.store.ReadVolumeNeedle(volumeId, n, nil, nil)
 	}
 	if err != nil {
+		if errors.Is(err, storage.ErrorNotFound) || errors.Is(err, erasure_coding.NotFoundError) {
+			return nil, status.Errorf(codes.NotFound, "needle not found %d", n.Id)
+		}
 		return nil, err
 	}
 	if count < 0 {
-		return nil, fmt.Errorf("needle not found %d", n.Id)
+		return nil, status.Errorf(codes.NotFound, "needle not found %d", n.Id)
 	}
 
 	resp.NeedleId = uint64(n.Id)

weed/shell/command_fs_verify.go

@@ -22,6 +22,8 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 	"go.uber.org/atomic"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 func init() {
@@ -34,6 +36,7 @@ type commandFsVerify struct {
 	volumeIds          map[uint32][]pb.ServerAddress
 	verbose            *bool
 	metadataFromLog    *bool
+	pruneEntries       *bool
 	concurrency        *int
 	modifyTimeAgoAtSec int64
 	writer             io.Writer
@@ -48,7 +51,12 @@ func (c *commandFsVerify) Name() string {
 func (c *commandFsVerify) Help() string {
 	return `recursively verify all files under a directory
 
-	fs.verify [-v] [-modifyTimeAgo 1h] /buckets/dir
+	fs.verify [-v] [-modifyTimeAgo 1h] [-pruneEntries] [-concurrency 4] /buckets/dir
+
+	-pruneEntries deletes filer entries whose needles are missing from every
+	volume location holding the volume (data lost, e.g. after a volume
+	truncation). The delete only applies when the entry is unchanged since
+	verification, so a re-uploaded file is never pruned.
 
 `
 }
@@ -65,6 +73,7 @@ func (c *commandFsVerify) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 	modifyTimeAgo := fsVerifyCommand.Duration("modifyTimeAgo", 0, "only include files after this modify time to verify")
 	c.concurrency = fsVerifyCommand.Int("concurrency", 0, "number of parallel verification per volume server")
 	c.metadataFromLog = fsVerifyCommand.Bool("metadataFromLog", false, "Using  filer log to get metadata")
+	c.pruneEntries = fsVerifyCommand.Bool("pruneEntries", false, "delete filer entries whose needles are missing from all volume locations (data lost); a changed entry is never deleted")
 	if err = fsVerifyCommand.Parse(args); err != nil {
 		return err
 	}
@@ -96,18 +105,15 @@ func (c *commandFsVerify) Do(args []string, commandEnv *CommandEnv, writer io.Wr
 			defer close(c.waitChan[volumeServerStr])
 		}
 	}
-	var fCount, eCount uint64
+	var fCount, eCount, pCount uint64
 	if *c.metadataFromLog {
 		var wg sync.WaitGroup
-		fCount, eCount, err = c.verifyProcessMetadata(path, &wg)
+		fCount, eCount, pCount, err = c.verifyProcessMetadata(path, &wg)
 		wg.Wait()
-		if err != nil {
-			return err
-		}
 	} else {
-		fCount, eCount, err = c.verifyTraverseBfs(path)
+		fCount, eCount, pCount, err = c.verifyTraverseBfs(path)
 	}
-	fmt.Fprintf(writer, "verified %d files, error %d files \n", fCount, eCount)
+	fmt.Fprintf(writer, "verified %d files, error %d files, pruned %d entries \n", fCount, eCount, pCount)
 	return err
 }
 
@@ -154,11 +160,35 @@ func (c *commandFsVerify) verifyChunk(volumeServer pb.ServerAddress, fileId *fil
 }
 
 type ItemEntry struct {
-	chunks []*filer_pb.FileChunk
-	path   util.FullPath
+	chunks    []*filer_pb.FileChunk
+	rawChunks []*filer_pb.FileChunk
+	path      util.FullPath
+	mtimeSec  int64
+	md5       []byte
 }
 
-func (c *commandFsVerify) verifyProcessMetadata(path string, wg *sync.WaitGroup) (fileCount uint64, errCount uint64, err error) {
+// resolveAndVerify expands chunk manifests and verifies the entry's needles.
+// An unresolvable manifest is an entry-level failure even when raw chunks are
+// healthy; raw chunks are still checked so a missing manifest needle can be pruned.
+func (c *commandFsVerify) resolveAndVerify(entryPath string, chunks []*filer_pb.FileChunk, errorChunksCount *atomic.Uint64, wg *sync.WaitGroup) (verified bool, hasMissingNeedles bool) {
+	dataChunks := chunks
+	manifestResolveFailed := false
+	if resolved, manifestChunks, resolveErr := filer.ResolveChunkManifest(context.Background(), filer.LookupFn(c.env), chunks, 0, math.MaxInt64, nil); resolveErr == nil {
+		dataChunks = append(resolved, manifestChunks...)
+	} else {
+		manifestResolveFailed = true
+		fmt.Fprintf(c.writer, "file: %s failed to resolve chunk manifest (%v), verifying raw chunks\n", entryPath, resolveErr)
+	}
+	verified, hasMissingNeedles = c.verifyEntry(entryPath, dataChunks, errorChunksCount, wg)
+	if manifestResolveFailed && verified {
+		// an unreadable manifest is an entry-level failure even with healthy raw needles
+		verified = false
+		errorChunksCount.CompareAndSwap(0, 1)
+	}
+	return verified, hasMissingNeedles
+}
+
+func (c *commandFsVerify) verifyProcessMetadata(path string, wg *sync.WaitGroup) (fileCount uint64, errCount uint64, prunedCount uint64, err error) {
 	processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
 		message := resp.EventNotification
 		if resp.EventNotification.NewEntry == nil {
@@ -170,7 +200,8 @@ func (c *commandFsVerify) verifyProcessMetadata(path string, wg *sync.WaitGroup)
 		}
 		entryPath := fmt.Sprintf("%s/%s", message.NewParentPath, message.NewEntry.Name)
 		errorChunksCount := atomic.NewUint64(0)
-		if !c.verifyEntry(entryPath, message.NewEntry.Chunks, errorChunksCount, wg) {
+		verified, hasMissingNeedles := c.resolveAndVerify(entryPath, message.NewEntry.Chunks, errorChunksCount, wg)
+		if !verified {
 			if err = c.env.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 				entryResp, errReq := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
 					Directory: message.NewParentPath,
@@ -186,6 +217,18 @@ func (c *commandFsVerify) verifyProcessMetadata(path string, wg *sync.WaitGroup)
 					bytes.Equal(entryResp.Entry.Attributes.Md5, message.NewEntry.Attributes.Md5) {
 					fmt.Fprintf(c.writer, "file: %s needles:%d failed:%d\n", entryPath, chunkCount, errorChunksCount.Load())
 					errCount++
+					if *c.pruneEntries && hasMissingNeedles {
+						pruned, pruneErr := c.pruneEntry(
+							util.NewFullPath(message.NewParentPath, message.NewEntry.Name),
+							message.NewEntry.Attributes.Mtime,
+							message.NewEntry.Attributes.Md5,
+							message.NewEntry.Chunks)
+						if pruneErr != nil {
+							fmt.Fprintf(c.writer, "prune %s failed: %v\n", entryPath, pruneErr)
+						} else if pruned {
+							prunedCount++
+						}
+					}
 				}
 				return nil
 			}); err != nil {
@@ -211,17 +254,58 @@ func (c *commandFsVerify) verifyProcessMetadata(path string, wg *sync.WaitGroup)
 		StopTsNs:               time.Now().UnixNano(),
 		EventErrorType:         pb.DontLogError,
 	}
-	return fileCount, errCount, pb.FollowMetadata(c.env.option.FilerAddress, c.env.option.GrpcDialOption, metadataFollowOption, processEventFn)
+	err = pb.FollowMetadata(c.env.option.FilerAddress, c.env.option.GrpcDialOption, metadataFollowOption, processEventFn)
+	return fileCount, errCount, prunedCount, err
 }
 
-func (c *commandFsVerify) verifyEntry(path string, chunks []*filer_pb.FileChunk, errorCount *atomic.Uint64, wg *sync.WaitGroup) bool {
+// isNeedleMissingError reports whether a VolumeNeedleStatus error means the
+// needle data is lost at that location. It must NOT match "volume not found".
+// Newer Go servers and the Rust server return codes.NotFound; older Go
+// servers return codes.Unknown with "needle not found <id>" or "EOF" (and EC
+// volumes wrap it as "locate in local ec volume: ... needle not found").
+// Legacy shapes are matched anchored so unrelated errors don't classify as missing.
+func isNeedleMissingError(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := err.Error()
+	if st, ok := status.FromError(err); ok {
+		switch st.Code() {
+		case codes.NotFound:
+			return strings.HasPrefix(st.Message(), "needle not found ")
+		case codes.Unknown:
+			if st.Message() != "" {
+				msg = st.Message()
+			}
+		default:
+			return false
+		}
+	}
+	return strings.HasPrefix(msg, "needle not found ") || msg == "EOF" ||
+		strings.Contains(msg, "locate in local ec volume:") && strings.HasSuffix(msg, "needle not found")
+}
+
+// verifyEntry verifies all chunks of an entry. It returns verified=false when
+// any chunk failed. hasMissingNeedles is true when at least one needle is
+// missing from every volume location holding its volume, i.e. the data is
+// lost rather than a transient or routing error.
+func (c *commandFsVerify) verifyEntry(path string, chunks []*filer_pb.FileChunk, errorCount *atomic.Uint64, wg *sync.WaitGroup) (verified bool, hasMissingNeedles bool) {
 	fileMsg := fmt.Sprintf("file:%s", path)
 	itemIsVerifed := atomic.NewBool(true)
+	hasMissingNeedlesFlag := atomic.NewBool(false)
+	// tracks this entry's in-flight verifications so the missing-needle
+	// decision at the end of each chunk is deterministic
+	itemWg := &sync.WaitGroup{}
+
 	for _, chunk := range chunks {
 		if volumeIds, ok := c.volumeIds[chunk.Fid.VolumeId]; ok {
+			chunkMissingLocations := atomic.NewUint64(0)
 			for _, volumeServer := range volumeIds {
 				if *c.concurrency == 0 {
 					if err := c.verifyChunk(volumeServer, chunk.Fid); err != nil {
+						if isNeedleMissingError(err) {
+							chunkMissingLocations.Add(1)
+						}
 						if !(*c.metadataFromLog && strings.HasSuffix(err.Error(), "not found")) {
 							fmt.Fprintf(c.writer, "%s failed verify fileId %s: %+v, at volume server %v\n",
 								fileMsg, chunk.GetFileIdString(), err, volumeServer)
@@ -246,10 +330,15 @@ func (c *commandFsVerify) verifyEntry(path string, chunks []*filer_pb.FileChunk,
 					continue
 				}
 				wg.Add(1)
+				itemWg.Add(1)
 				waitChan <- struct{}{}
-				go func(fChunk *filer_pb.FileChunk, path string, volumeServer pb.ServerAddress, msg string) {
+				go func(fChunk *filer_pb.FileChunk, path string, volumeServer pb.ServerAddress, msg string, chunkMissingLocations *atomic.Uint64) {
 					defer wg.Done()
+					defer itemWg.Done()
 					if err := c.verifyChunk(volumeServer, fChunk.Fid); err != nil {
+						if isNeedleMissingError(err) {
+							chunkMissingLocations.Add(1)
+						}
 						if !(*c.metadataFromLog && strings.HasSuffix(err.Error(), "not found")) {
 							fmt.Fprintf(c.writer, "%s failed verify fileId %s: %+v, at volume server %v\n",
 								msg, fChunk.GetFileIdString(), err, volumeServer)
@@ -260,7 +349,11 @@ func (c *commandFsVerify) verifyEntry(path string, chunks []*filer_pb.FileChunk,
 						}
 					}
 					<-waitChan
-				}(chunk, path, volumeServer, fileMsg)
+				}(chunk, path, volumeServer, fileMsg, chunkMissingLocations)
+			}
+			itemWg.Wait()
+			if chunkMissingLocations.Load() == uint64(len(volumeIds)) {
+				hasMissingNeedlesFlag.Store(true)
 			}
 		} else {
 			if !*c.metadataFromLog {
@@ -275,12 +368,108 @@ func (c *commandFsVerify) verifyEntry(path string, chunks []*filer_pb.FileChunk,
 			break
 		}
 	}
-	return itemIsVerifed.Load()
+	return itemIsVerifed.Load(), hasMissingNeedlesFlag.Load()
 }
 
-func (c *commandFsVerify) verifyTraverseBfs(path string) (fileCount uint64, errCount uint64, err error) {
+// pruneEntry deletes an entry whose needles are gone, unless the entry
+// changed since verification. Three guards protect a concurrent repair:
+// re-lookup with mtime/md5 comparison, an identity check of the chunk set
+// (an append or rewrite in the same second as the original mtime changes the
+// chunks but not the timestamps), and IfNotModifiedAfter on the delete. The
+// entry is only counted as pruned when a follow-up lookup confirms it is gone.
+func (c *commandFsVerify) pruneEntry(path util.FullPath, mtimeSec int64, md5 []byte, expectedChunks []*filer_pb.FileChunk) (pruned bool, err error) {
+	dir, name := path.DirAndName()
+	expectedIds := chunkFingerprints(expectedChunks)
+	lookupErr := c.env.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		lookupResp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
+			Directory: dir,
+			Name:      name,
+		})
+		if err != nil {
+			if strings.Contains(err.Error(), "no entry is found in filer store") {
+				return nil // already gone
+			}
+			return err
+		}
+		att := lookupResp.Entry.GetAttributes()
+		if att.GetMtime() != mtimeSec {
+			fmt.Fprintf(c.writer, "skip pruning %s: entry changed since verification\n", path)
+			return nil
+		}
+		if len(md5) > 0 && !bytes.Equal(att.GetMd5(), md5) {
+			fmt.Fprintf(c.writer, "skip pruning %s: entry changed since verification\n", path)
+			return nil
+		}
+		if !chunksEqual(chunkFingerprints(lookupResp.Entry.GetChunks()), expectedIds) {
+			fmt.Fprintf(c.writer, "skip pruning %s: entry changed since verification\n", path)
+			return nil
+		}
+		deleteResp, err := client.DeleteEntry(context.Background(), &filer_pb.DeleteEntryRequest{
+			Directory:          dir,
+			Name:               name,
+			IfNotModifiedAfter: mtimeSec,
+		})
+		if err != nil {
+			return err
+		}
+		if deleteResp.Error != "" {
+			return fmt.Errorf("delete entry %s: %s", path, deleteResp.Error)
+		}
+		// the filer silently skips the delete when the entry advanced past
+		// IfNotModifiedAfter between our lookup and the delete; confirm
+		confirm, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
+			Directory: dir,
+			Name:      name,
+		})
+		if err != nil {
+			if strings.Contains(err.Error(), "no entry is found in filer store") {
+				pruned = true
+				fmt.Fprintf(c.writer, "pruned entry with missing needles: %s\n", path)
+				return nil
+			}
+			// the delete outcome is unknown — never count it as pruned
+			fmt.Fprintf(c.writer, "skip pruning %s: delete not confirmed (%v)\n", path, err)
+			return nil
+		}
+		if confirm.Entry != nil {
+			fmt.Fprintf(c.writer, "skip pruning %s: entry changed since verification\n", path)
+			return nil
+		}
+		pruned = true
+		fmt.Fprintf(c.writer, "pruned entry with missing needles: %s\n", path)
+		return nil
+	})
+	return pruned, lookupErr
+}
+
+// chunkFingerprints maps the file id of every chunk to a set, so two chunk
+// lists count as identical when they reference the same needles.
+func chunkFingerprints(chunks []*filer_pb.FileChunk) map[string]struct{} {
+	ids := make(map[string]struct{}, len(chunks))
+	for _, chunk := range chunks {
+		if chunk == nil || chunk.Fid == nil {
+			continue
+		}
+		ids[chunk.GetFileIdString()] = struct{}{}
+	}
+	return ids
+}
+
+func chunksEqual(a, b map[string]struct{}) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for id := range a {
+		if _, ok := b[id]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
+func (c *commandFsVerify) verifyTraverseBfs(path string) (fileCount uint64, errCount uint64, prunedCount uint64, err error) {
 	timeNowAtSec := time.Now().Unix()
-	return fileCount, errCount, doTraverseBfsAndSaving(c.env, c.writer, path, false, false,
+	return fileCount, errCount, prunedCount, doTraverseBfsAndSaving(c.env, c.writer, path, false, false,
 		func(ctx context.Context, entry *filer_pb.FullEntry, outputChan chan interface{}) (err error) {
 			if c.modifyTimeAgoAtSec > 0 {
 				if entry.Entry.Attributes != nil && c.modifyTimeAgoAtSec < timeNowAtSec-entry.Entry.Attributes.Mtime {
@@ -295,8 +484,11 @@ func (c *commandFsVerify) verifyTraverseBfs(path string) (fileCount uint64, errC
 			if len(dataChunks) > 0 {
 				select {
 				case outputChan <- &ItemEntry{
-					chunks: dataChunks,
-					path:   util.NewFullPath(entry.Dir, entry.Entry.Name),
+					chunks:    dataChunks,
+					rawChunks: entry.Entry.GetChunks(),
+					path:      util.NewFullPath(entry.Dir, entry.Entry.Name),
+					mtimeSec:  entry.Entry.GetAttributes().GetMtime(),
+					md5:       entry.Entry.GetAttributes().GetMd5(),
 				}:
 				case <-ctx.Done():
 					return ctx.Err()
@@ -310,11 +502,18 @@ func (c *commandFsVerify) verifyTraverseBfs(path string) (fileCount uint64, errC
 			for itemEntry := range outputChan {
 				i := itemEntry.(*ItemEntry)
 				itemPath := string(i.path)
-				if c.verifyEntry(itemPath, i.chunks, itemErrCount, &wg) {
+				if verified, hasMissingNeedles := c.verifyEntry(itemPath, i.chunks, itemErrCount, &wg); verified {
 					if *c.verbose {
 						fmt.Fprintf(c.writer, "file: %s needles:%d verified\n", itemPath, len(i.chunks))
 					}
 					fileCount++
+				} else if *c.pruneEntries && hasMissingNeedles {
+					pruned, pruneErr := c.pruneEntry(i.path, i.mtimeSec, i.md5, i.rawChunks)
+					if pruneErr != nil {
+						fmt.Fprintf(c.writer, "prune %s failed: %v\n", itemPath, pruneErr)
+					} else if pruned {
+						prunedCount++
+					}
 				}
 			}
 			wg.Wait()

weed/shell/command_fs_verify_test.go

@@ -0,0 +1,348 @@
+package shell
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"go.uber.org/atomic"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+)
+
+type fsVerifyTestVolumeServer struct {
+	volume_server_pb.UnimplementedVolumeServerServer
+	missingNeedles map[uint64]bool
+}
+
+func (s *fsVerifyTestVolumeServer) VolumeNeedleStatus(_ context.Context, req *volume_server_pb.VolumeNeedleStatusRequest) (*volume_server_pb.VolumeNeedleStatusResponse, error) {
+	if s.missingNeedles[req.NeedleId] {
+		return nil, fmt.Errorf("needle not found %d", req.NeedleId)
+	}
+	return &volume_server_pb.VolumeNeedleStatusResponse{NeedleId: req.NeedleId}, nil
+}
+
+type fsVerifyTestFilerServer struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	entry   *filer_pb.Entry
+	deleted []*filer_pb.DeleteEntryRequest
+}
+
+func (s *fsVerifyTestFilerServer) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	if s.entry == nil {
+		return nil, fmt.Errorf("no entry is found in filer store")
+	}
+	return &filer_pb.LookupDirectoryEntryResponse{Entry: s.entry}, nil
+}
+
+func (s *fsVerifyTestFilerServer) DeleteEntry(_ context.Context, req *filer_pb.DeleteEntryRequest) (*filer_pb.DeleteEntryResponse, error) {
+	s.deleted = append(s.deleted, req)
+	// mirror the real filer: an entry newer than IfNotModifiedAfter is
+	// silently skipped, no error and no Error field
+	if s.entry != nil && req.IfNotModifiedAfter > 0 && s.entry.Attributes.GetMtime() > req.IfNotModifiedAfter {
+		return &filer_pb.DeleteEntryResponse{}, nil
+	}
+	s.entry = nil
+	return &filer_pb.DeleteEntryResponse{}, nil
+}
+
+var fsVerifyTestPortCounter atomic.Int64
+
+func newFsVerifyTestCommandEnv(t *testing.T, filerServer filer_pb.SeaweedFilerServer, volumeServer volume_server_pb.VolumeServerServer) (*CommandEnv, pb.ServerAddress, func()) {
+	t.Helper()
+
+	socketDir, err := os.MkdirTemp("", "swverify-")
+	if err != nil {
+		t.Fatalf("create socket dir: %v", err)
+	}
+	t.Cleanup(func() { _ = os.RemoveAll(socketDir) })
+
+	filerListener, err := net.Listen("unix", filepath.Join(socketDir, "filer.sock"))
+	if err != nil {
+		t.Fatalf("listen filer socket: %v", err)
+	}
+	filerGrpc := grpc.NewServer()
+	filer_pb.RegisterSeaweedFilerServer(filerGrpc, filerServer)
+	go func() { _ = filerGrpc.Serve(filerListener) }()
+
+	volumeListener, err := net.Listen("unix", filepath.Join(socketDir, "volume.sock"))
+	if err != nil {
+		t.Fatalf("listen volume socket: %v", err)
+	}
+	volumeGrpc := grpc.NewServer()
+	volume_server_pb.RegisterVolumeServerServer(volumeGrpc, volumeServer)
+	go func() { _ = volumeGrpc.Serve(volumeListener) }()
+
+	// unique ports per env: pb caches gRPC connections by address, so
+	// reusing a port across envs would dial a dead cached connection
+	filerPort := int(fsVerifyTestPortCounter.Add(1))
+	volumePort := int(fsVerifyTestPortCounter.Add(1))
+	pb.RegisterLocalGrpcSocket("127.0.0.1", filerPort, filepath.Join(socketDir, "filer.sock"))
+	pb.RegisterLocalGrpcSocket("127.0.0.1", volumePort, filepath.Join(socketDir, "volume.sock"))
+
+	cleanup := func() {
+		filerGrpc.Stop()
+		volumeGrpc.Stop()
+		_ = filerListener.Close()
+		_ = volumeListener.Close()
+	}
+
+	commandEnv := &CommandEnv{
+		option: &ShellOptions{
+			FilerAddress:   pb.ServerAddress(fmt.Sprintf("127.0.0.1:8888.%d", filerPort)),
+			GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()),
+			Directory:      "/",
+		},
+	}
+	volumeLocation := pb.NewServerAddress("127.0.0.1", 19080, volumePort)
+	return commandEnv, volumeLocation, cleanup
+}
+
+func newFsVerifyTestCommand(t *testing.T, commandEnv *CommandEnv, output *bytes.Buffer, volumeLocations ...pb.ServerAddress) *commandFsVerify {
+	t.Helper()
+	pruneEntries := false
+	metadataFromLog := false
+	verbose := false
+	concurrency := 0
+	c := &commandFsVerify{
+		env:             commandEnv,
+		writer:          output,
+		volumeIds:       make(map[uint32][]pb.ServerAddress),
+		verbose:         &verbose,
+		metadataFromLog: &metadataFromLog,
+		pruneEntries:    &pruneEntries,
+		concurrency:     &concurrency,
+		waitChan:        make(map[string]chan struct{}),
+	}
+	for _, loc := range volumeLocations {
+		c.volumeIds[7] = append(c.volumeIds[7], loc)
+		c.waitChan[string(loc)] = make(chan struct{}, 4)
+	}
+	return c
+}
+
+func TestFsVerifyIsNeedleMissingError(t *testing.T) {
+	if !isNeedleMissingError(fmt.Errorf("needle not found 42")) {
+		t.Error("needle not found must classify as missing")
+	}
+	// truncated volume files surface as EOF wrapped in gRPC code Unknown
+	if !isNeedleMissingError(status.Error(codes.Unknown, "EOF")) {
+		t.Error("EOF must classify as missing (truncated needle data)")
+	}
+	// anchored matching: unrelated errors merely containing the substrings
+	// must not classify as missing
+	if isNeedleMissingError(status.Error(codes.Unknown, "unexpected EOF while reading trailer")) {
+		t.Error("unrelated EOF-containing errors must NOT classify as missing")
+	}
+	if isNeedleMissingError(fmt.Errorf("rpc error: code = Unknown desc = read needle: context deadline exceeded, stream EOF")) {
+		t.Error("deadline errors mentioning EOF must NOT classify as missing")
+	}
+	if isNeedleMissingError(status.Error(codes.NotFound, "needle not foundxyz")) {
+		t.Error("malformed needle-not-found prefix must NOT classify as missing")
+	}
+	// new Go servers and the Rust volume server answer with code NotFound
+	if !isNeedleMissingError(status.Error(codes.NotFound, "needle not found 42")) {
+		t.Error("NotFound needle-not-found must classify as missing")
+	}
+	// a NotFound for the volume itself says nothing about the needle
+	if isNeedleMissingError(status.Error(codes.NotFound, "volume 7 not found")) {
+		t.Error("NotFound volume-not-found must NOT classify as missing")
+	}
+	if isNeedleMissingError(status.Error(codes.Unavailable, "connection refused")) {
+		t.Error("transport errors must NOT classify as missing")
+	}
+	if isNeedleMissingError(fmt.Errorf("volume not found 7")) {
+		t.Error("volume not found must NOT classify as missing (it says nothing about the needle)")
+	}
+	// older Go servers wrap EC's NotFoundError as code Unknown
+	if !isNeedleMissingError(status.Error(codes.Unknown, "locate in local ec volume: FindNeedleFromEcx: needle not found")) {
+		t.Error("EC wrapped needle-not-found must classify as missing")
+	}
+	if isNeedleMissingError(status.Error(codes.Unknown, "locate in local ec volume: ReadEcShardIntervals: shard 3 missing")) {
+		t.Error("EC errors that do not end in needle not found must NOT classify as missing")
+	}
+	if isNeedleMissingError(nil) {
+		t.Error("nil error must not classify as missing")
+	}
+}
+
+// A needle gone from every location holding the volume means the data is lost,
+// so the entry is prunable. One healthy location keeps it (reads can still
+// succeed there).
+func TestFsVerifyVerifyEntryPrunability(t *testing.T) {
+	deadChunk := &filer_pb.FileChunk{Fid: &filer_pb.FileId{VolumeId: 7, FileKey: 42}}
+
+	t.Run("missing on all locations", func(t *testing.T) {
+		commandEnv, loc, cleanup := newFsVerifyTestCommandEnv(t,
+			&fsVerifyTestFilerServer{},
+			&fsVerifyTestVolumeServer{missingNeedles: map[uint64]bool{42: true}})
+		defer cleanup()
+		c := newFsVerifyTestCommand(t, commandEnv, &bytes.Buffer{}, loc)
+
+		verified, prunable := c.verifyEntry("/buckets/b/file", []*filer_pb.FileChunk{deadChunk}, atomic.NewUint64(0), &sync.WaitGroup{})
+		if verified {
+			t.Error("expected verification failure for missing needle")
+		}
+		if !prunable {
+			t.Error("needle missing on the only location must be prunable")
+		}
+	})
+
+	t.Run("healthy on one of two locations", func(t *testing.T) {
+		commandEnv, loc, cleanup := newFsVerifyTestCommandEnv(t,
+			&fsVerifyTestFilerServer{},
+			&fsVerifyTestVolumeServer{})
+		defer cleanup()
+		c := newFsVerifyTestCommand(t, commandEnv, &bytes.Buffer{}, loc, loc)
+
+		verified, prunable := c.verifyEntry("/buckets/b/file", []*filer_pb.FileChunk{deadChunk}, atomic.NewUint64(0), &sync.WaitGroup{})
+		if !verified {
+			t.Error("needle present on one location should not fail verification")
+		}
+		if prunable {
+			t.Error("needle present somewhere must not be prunable")
+		}
+	})
+
+	t.Run("concurrent path matches sequential", func(t *testing.T) {
+		commandEnv, loc, cleanup := newFsVerifyTestCommandEnv(t,
+			&fsVerifyTestFilerServer{},
+			&fsVerifyTestVolumeServer{missingNeedles: map[uint64]bool{42: true}})
+		defer cleanup()
+		c := newFsVerifyTestCommand(t, commandEnv, &bytes.Buffer{}, loc)
+		concurrency := 2
+		c.concurrency = &concurrency
+
+		verified, prunable := c.verifyEntry("/buckets/b/file", []*filer_pb.FileChunk{deadChunk}, atomic.NewUint64(0), &sync.WaitGroup{})
+		if verified || !prunable {
+			t.Errorf("concurrent verification = (%v, %v), want (false, true)", verified, prunable)
+		}
+	})
+}
+
+// Pruning must only delete the entry when it did not change since the
+// verification pass — a re-upload fixes the file and must survive.
+func TestFsVerifyPruneEntryGuards(t *testing.T) {
+	filer := &fsVerifyTestFilerServer{entry: &filer_pb.Entry{
+		Name: "file",
+		Attributes: &filer_pb.FuseAttributes{
+			Mtime: 100,
+			Md5:   []byte("md5"),
+		},
+		Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 7, FileKey: 42}}},
+	}}
+	commandEnv, _, cleanup := newFsVerifyTestCommandEnv(t, filer, &fsVerifyTestVolumeServer{})
+	defer cleanup()
+	c := newFsVerifyTestCommand(t, commandEnv, &bytes.Buffer{})
+
+	chunks := []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 7, FileKey: 42}}}
+
+	// unchanged entry: deleted
+	pruned, err := c.pruneEntry(util.NewFullPath("/buckets/b", "file"), 100, []byte("md5"), chunks)
+	if err != nil || !pruned {
+		t.Fatalf("expected prune of unchanged entry, got pruned=%v err=%v", pruned, err)
+	}
+
+	// re-uploaded entry (new mtime/md5): must survive
+	filer.entry = &filer_pb.Entry{Name: "file", Attributes: &filer_pb.FuseAttributes{Mtime: 200, Md5: []byte("new")}}
+	pruned, err = c.pruneEntry(util.NewFullPath("/buckets/b", "file"), 100, []byte("md5"), chunks)
+	if err != nil || pruned {
+		t.Fatalf("changed entry must not be pruned, got pruned=%v err=%v", pruned, err)
+	}
+
+	// same-second rewrite: timestamps identical but the chunk set differs —
+	// the new upload must survive
+	filer.entry = &filer_pb.Entry{
+		Name:       "file",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 100, Md5: []byte("md5")},
+		Chunks:     []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 7, FileKey: 999}}},
+	}
+	pruned, err = c.pruneEntry(util.NewFullPath("/buckets/b", "file"), 100, []byte("md5"), chunks)
+	if err != nil || pruned {
+		t.Fatalf("same-second rewrite must not be pruned, got pruned=%v err=%v", pruned, err)
+	}
+
+	// identical chunk set (order-insensitive): prunes
+	filer.entry = &filer_pb.Entry{
+		Name:       "file",
+		Attributes: &filer_pb.FuseAttributes{Mtime: 100, Md5: []byte("md5")},
+		Chunks:     []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 7, FileKey: 42}}},
+	}
+	pruned, err = c.pruneEntry(util.NewFullPath("/buckets/b", "file"), 100, []byte("md5"), chunks)
+	if err != nil || !pruned {
+		t.Fatalf("expected prune when chunk set matches, got pruned=%v err=%v", pruned, err)
+	}
+
+	// entry already gone: no error, no delete
+	filer.entry = nil
+	pruned, err = c.pruneEntry(util.NewFullPath("/buckets/b", "file"), 100, nil, chunks)
+	if err != nil || pruned {
+		t.Fatalf("missing entry must be a no-op, got pruned=%v err=%v", pruned, err)
+	}
+}
+
+// An unresolvable chunk manifest is an entry-level verification failure even
+// when the raw manifest needle is healthy; raw chunks are still checked so a
+// missing manifest needle can be pruned.
+func TestFsVerifyManifestResolutionFailure(t *testing.T) {
+	t.Run("healthy raw needle but unresolvable manifest is not verified", func(t *testing.T) {
+		// IsChunkManifest forces ResolveChunkManifest to read children via
+		// LookupFn; the test filer does not implement LookupVolume, so
+		// resolution fails. Needle 100 is present, so verifyEntry alone
+		// would report verified=true — the manifest failure must override.
+		manifestChunk := &filer_pb.FileChunk{
+			Fid:             &filer_pb.FileId{VolumeId: 7, FileKey: 100},
+			IsChunkManifest: true,
+			Size:            100,
+		}
+		commandEnv, loc, cleanup := newFsVerifyTestCommandEnv(t,
+			&fsVerifyTestFilerServer{},
+			&fsVerifyTestVolumeServer{})
+		defer cleanup()
+		c := newFsVerifyTestCommand(t, commandEnv, &bytes.Buffer{}, loc)
+
+		verified, hasMissing := c.resolveAndVerify("/buckets/b/file",
+			[]*filer_pb.FileChunk{manifestChunk}, atomic.NewUint64(0), &sync.WaitGroup{})
+		if verified {
+			t.Error("entry with an unresolvable manifest must not count as verified")
+		}
+		if hasMissing {
+			t.Error("healthy raw manifest needle must not be classified as missing")
+		}
+	})
+
+	t.Run("missing manifest needle is prunable", func(t *testing.T) {
+		// The manifest needle itself is gone: resolution fails AND verifyEntry
+		// classifies the raw needle as missing, so the entry is prunable.
+		manifestChunk := &filer_pb.FileChunk{
+			Fid:             &filer_pb.FileId{VolumeId: 7, FileKey: 100},
+			IsChunkManifest: true,
+			Size:            100,
+		}
+		commandEnv, loc, cleanup := newFsVerifyTestCommandEnv(t,
+			&fsVerifyTestFilerServer{},
+			&fsVerifyTestVolumeServer{missingNeedles: map[uint64]bool{100: true}})
+		defer cleanup()
+		c := newFsVerifyTestCommand(t, commandEnv, &bytes.Buffer{}, loc)
+
+		verified, hasMissing := c.resolveAndVerify("/buckets/b/file",
+			[]*filer_pb.FileChunk{manifestChunk}, atomic.NewUint64(0), &sync.WaitGroup{})
+		if verified {
+			t.Error("entry with a missing manifest needle must not count as verified")
+		}
+		if !hasMissing {
+			t.Error("missing manifest needle must be classified as prunable")
+		}
+	})
+}