diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go index f5d8c71c73..4699fbfd2b 100644 --- a/weed/server/volume_grpc_admin.go +++ b/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) diff --git a/weed/shell/command_fs_verify.go b/weed/shell/command_fs_verify.go index 92aed41698..32128b4b6f 100644 --- a/weed/shell/command_fs_verify.go +++ b/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 " 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() diff --git a/weed/shell/command_fs_verify_test.go b/weed/shell/command_fs_verify_test.go new file mode 100644 index 0000000000..5c997ff229 --- /dev/null +++ b/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") + } + }) +}