diff --git a/weed/shell/command_fs_distribute_chunks.go b/weed/shell/command_fs_distribute_chunks.go index 17c6f4fdd9..bb05ce6217 100644 --- a/weed/shell/command_fs_distribute_chunks.go +++ b/weed/shell/command_fs_distribute_chunks.go @@ -23,6 +23,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/util" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" + "google.golang.org/protobuf/proto" ) func init() { @@ -116,6 +117,7 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr if err != nil { return fmt.Errorf("lookup %s: %v", path, err) } + readEntry := proto.Clone(entry).(*filer_pb.Entry) chunks, hadManifest, err := resolveEntryDataChunks(commandEnv, entry) if err != nil { @@ -180,6 +182,11 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr fmt.Fprintf(writer, "\nExecuting redistribution...\n") defer util_http.GetGlobalHttpClient().CloseIdleConnections() + readFids := make(map[string]bool, len(chunks)) + for _, chunk := range chunks { + readFids[chunk.GetFileIdString()] = true + } + movedCount, err := executeChunkMoves(commandEnv, writer, moves, volumeInfoMap) if err != nil { return err @@ -191,8 +198,14 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr finalChunks := chunks if hadManifest { - // no chunk deleter here: a failed fold reports the blobs it saved - remanifested, manErr := filer.MaybeManifestize(newShellSaveAsChunk(commandEnv), nil, chunks) + dropSaved := func(saved []*filer_pb.FileChunk) { + var blobs []orphanedNeedle + for _, chunk := range saved { + blobs = append(blobs, orphanedNeedle{volumeId: chunk.GetFid().GetVolumeId(), fileId: chunk.GetFileIdString()}) + } + deleteOrphanedNeedles(commandEnv, util.FullPath(path), blobs, true) + } + remanifested, manErr := filer.MaybeManifestize(newShellSaveAsChunk(commandEnv), dropSaved, chunks) if manErr != nil { fmt.Fprintf(writer, "WARNING: re-manifestize failed: %v. Writing flat chunk list.\n", manErr) } else { @@ -203,12 +216,24 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr entry.Chunks = finalChunks fmt.Fprintf(writer, "\nUpdating filer metadata (%d chunks moved)...\n", movedCount) + var stale bool err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{ - Directory: dir, - Entry: entry, - }) + var uErr error + stale, uErr = updateEntryIfUnchanged(context.Background(), client, dir, entry, readEntry) + return uErr }) + if stale { + var copies []orphanedNeedle + for _, chunk := range slices.Concat(chunks, finalChunks) { + if fid := chunk.GetFileIdString(); !readFids[fid] { + readFids[fid] = true + copies = append(copies, orphanedNeedle{volumeId: chunk.GetFid().GetVolumeId(), fileId: fid}) + } + } + deleteOrphanedNeedles(commandEnv, util.FullPath(path), copies, true) + fmt.Fprintf(writer, "%s was changed or deleted during redistribution; the filer kept that version and the moved copies were deleted.\n", path) + return err + } if err != nil { fmt.Fprintf(writer, "FAILED to update filer metadata: %v\n", err) fmt.Fprintf(writer, "WARNING: New chunks were uploaded but metadata was not updated.\n") diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 99b6140943..26059da18e 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/weed/shell/command_fs_merge_volumes.go @@ -6,6 +6,7 @@ import ( "flag" "fmt" "io" + "math" "net/http" "sort" "strconv" @@ -21,6 +22,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util/wildcard" "github.com/seaweedfs/seaweedfs/weed/wdclient" "golang.org/x/exp/maps" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" "github.com/seaweedfs/seaweedfs/weed/operation" @@ -54,6 +57,9 @@ func (c *commandFsMergeVolumes) Help() string { -toVolumeId accepts a comma-separated list. With -fromVolumeId, the source chunks are distributed across the listed volumes by remaining capacity, so a volume that does not fit into any single target can still be cleared. + + -apply needs the shell lock (run "lock" first), so two runs never move the + same chunks at once. ` } @@ -72,6 +78,11 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer if err = fsMergeVolumesCommand.Parse(args); err != nil { return err } + if *apply { + if err = commandEnv.confirmIsLocked(args); err != nil { + return err + } + } dir := *dirArg if dir != "/" { @@ -171,7 +182,16 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer // look like mergeVolumes hadn't done anything. Track the old fids // and delete them below after the filer update commits, so the // filer never points at a fid we already deleted. - var movedSources []orphanedNeedle + var moved rewrittenNeedles + // The update only lands while the filer still holds the entry as + // the walk read it, so keep that copy before the first chunk is + // rewritten in place. + var readEntry *filer_pb.Entry + keepReadEntry := func() { + if readEntry == nil { + readEntry = proto.Clone(entry).(*filer_pb.Entry) + } + } for i, chunk := range entry.Chunks { if chunk.IsChunkManifest { if !c.manifestMayReferencePlan(plan, planCollections, needle.VolumeId(chunk.Fid.VolumeId)) { @@ -190,14 +210,18 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer if !changed || !*apply { continue } + keepReadEntry() entry.Chunks[i] = newChunk entryChanged = true - movedSources = append(movedSources, rewritten.sources...) + moved.sources = append(moved.sources, rewritten.sources...) // The old manifest needle is always orphaned when we // replace it with a freshly uploaded one, even when the // rewrite was triggered by sub-chunk moves rather than the // manifest volume itself being in the plan. - movedSources = append(movedSources, orphanedNeedle{volumeId: oldManifestVid, fileId: oldManifestFid}) + moved.sources = append(moved.sources, orphanedNeedle{volumeId: oldManifestVid, fileId: oldManifestFid}) + moved.targets = append(moved.targets, rewritten.targets...) + moved.targets = append(moved.targets, orphanedNeedle{volumeId: newChunk.Fid.VolumeId, fileId: newChunk.GetFileIdString()}) + moved.reservations = append(moved.reservations, rewritten.reservations...) continue } @@ -206,6 +230,7 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer continue } seen.record(chunkVolumeId) + keepReadEntry() oldFid := chunk.GetFileIdString() oldVid := chunk.Fid.VolumeId @@ -224,20 +249,29 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer continue } entryChanged = true - movedSources = append(movedSources, orphanedNeedle{volumeId: oldVid, fileId: oldFid}) + moved.sources = append(moved.sources, orphanedNeedle{volumeId: oldVid, fileId: oldFid}) + moved.targets = append(moved.targets, orphanedNeedle{volumeId: uint32(toVolumeId), fileId: chunk.GetFileIdString()}) + moved.reservations = append(moved.reservations, reservation{src: chunkVolumeId, dst: toVolumeId, size: chunk.Size}) } if entryChanged { - if uErr := filer_pb.UpdateEntry(context.Background(), filerClient, &filer_pb.UpdateEntryRequest{ - Directory: string(parentPath), - Entry: entry, - }); uErr != nil { + stale, uErr := updateEntryIfUnchanged(context.Background(), filerClient, string(parentPath), entry, readEntry) + if stale { + fmt.Printf("skip %s: changed during merge\n", entryPath) + moved.targets = unreferencedTargets(context.Background(), filerClient, lookupFn, parentPath, entry.Name, moved.targets) + moved.abandon(commandEnv, entryPath, plan) + if len(entry.HardLinkId) > 0 { + processedHardLinks.Delete(string(entry.HardLinkId)) + } + return nil + } + if uErr != nil { fmt.Printf("failed to update %s: %v\n", entryPath, uErr) // Filer still references the source fids. Deleting them // now would lose data — abandon the cleanup for this // entry and let fsck reconcile later. return nil } - deleteOrphanedNeedles(commandEnv, entryPath, movedSources, false) + deleteOrphanedNeedles(commandEnv, entryPath, moved.sources, false) } return nil }) @@ -313,6 +347,61 @@ type rewrittenNeedles struct { reservations []reservation } +// abandon drops a rewrite the filer never took: the copies on the target +// volumes belong to nobody, and the plan gets back the room they held, or a +// multi-target merge keeps counting bytes it just deleted and skips later +// chunks for no reason. +func (r rewrittenNeedles) abandon(commandEnv *CommandEnv, entryPath util.FullPath, plan *mergePlan) { + deleteOrphanedNeedles(commandEnv, entryPath, r.targets, true) + for _, res := range r.reservations { + plan.release(res.src, res.dst, res.size) + } +} + +// updateEntryIfUnchanged writes entry back only while the filer still holds +// readEntry. An unconditional write would revert a change made since the read +// and the filer would delete the newer chunks as garbage. stale reports that +// the filer refused because the entry was changed or deleted, so nothing was +// written and the copies made for this update are referenced by nothing. +func updateEntryIfUnchanged(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string, entry, readEntry *filer_pb.Entry) (stale bool, err error) { + err = filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{ + Directory: dir, + Entry: entry, + Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{ + {Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: readEntry}, + }}, + }) + switch status.Code(err) { + case codes.FailedPrecondition, codes.NotFound: + return true, err + } + return false, err +} + +// unreferencedTargets drops the copies the filer's current entry points at. +// moveChunk keeps the key and cookie, so an overlapping run that moved the same +// chunk to the same volume wrote the same fid, and its committed update is what +// refused this one. When the entry cannot be read, no copy is reported. +func unreferencedTargets(ctx context.Context, client filer_pb.SeaweedFilerClient, lookupFn wdclient.LookupFileIdFunctionType, dir util.FullPath, name string, targets []orphanedNeedle) []orphanedNeedle { + resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: string(dir), Name: name}) + if errors.Is(err, filer_pb.ErrNotFound) { + return targets + } + if err != nil { + return nil + } + chunks := resp.Entry.GetChunks() + dataChunks, manifestChunks, err := filer.ResolveChunkManifest(ctx, lookupFn, chunks, 0, math.MaxInt64, nil) + if err != nil { + return nil + } + referenced := make(map[string]bool) + for _, chunk := range slices.Concat(chunks, dataChunks, manifestChunks) { + referenced[chunk.GetFileIdString()] = true + } + return slices.DeleteFunc(targets, func(t orphanedNeedle) bool { return referenced[t.fileId] }) +} + // deleteOrphanedNeedles fans out BatchDelete RPCs to every replica of each // needle's volume. Errors are logged but never returned — the data is already // orphan at this point, so a failed cleanup just leaves work for a later fsck. @@ -740,16 +829,8 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk( var rewritten rewrittenNeedles // Every path out of here that is not the successful return abandons the - // rewrite with the filer still pointing at the old manifest, so whatever - // already landed on the target volumes belongs to nobody. Give the plan - // back the room those copies were holding, or a multi-target merge keeps - // counting bytes it just deleted and skips later chunks for no reason. - abandon := func() { - deleteOrphanedNeedles(commandEnv, entryPath, rewritten.targets, true) - for _, r := range rewritten.reservations { - plan.release(r.src, r.dst, r.size) - } - } + // rewrite with the filer still pointing at the old manifest. + abandon := func() { rewritten.abandon(commandEnv, entryPath, plan) } anySubChanged := false for i, sub := range subChunks { if sub.IsChunkManifest { diff --git a/weed/shell/command_fs_merge_volumes_test.go b/weed/shell/command_fs_merge_volumes_test.go index d02dd60b71..37500e2d45 100644 --- a/weed/shell/command_fs_merge_volumes_test.go +++ b/weed/shell/command_fs_merge_volumes_test.go @@ -1,12 +1,22 @@ package shell import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "slices" "strings" "sync" "testing" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" ) func newMergeCmd(limitMB uint64, vols ...*master_pb.VolumeInformationMessage) *commandFsMergeVolumes { @@ -394,3 +404,112 @@ func TestWarnUnreferencedSources_ConcurrentRecording(t *testing.T) { t.Fatalf("expected no warning when all needles were seen, got %q", sb.String()) } } + +type conditionalUpdateFiler struct { + filer_pb.SeaweedFilerClient + err error + req *filer_pb.UpdateEntryRequest + current *filer_pb.Entry + lookupErr error + lookupReq *filer_pb.LookupDirectoryEntryRequest +} + +func (s *conditionalUpdateFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + s.lookupReq = req + if s.lookupErr != nil { + return nil, s.lookupErr + } + return &filer_pb.LookupDirectoryEntryResponse{Entry: s.current}, nil +} + +func (s *conditionalUpdateFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + s.req = req + if s.err != nil { + return nil, s.err + } + return &filer_pb.UpdateEntryResponse{}, nil +} + +// The write-back is conditioned on the entry as it was read, and only a refusal +// that proves nothing was written counts as stale: the caller then deletes the +// copies it made, which would lose data if the update might have landed. +func TestUpdateEntryIfUnchanged(t *testing.T) { + read := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 3, FileKey: 1, Cookie: 7}}}} + moved := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 9, FileKey: 1, Cookie: 7}}}} + + for _, tc := range []struct { + name string + err error + stale bool + }{ + {"committed", nil, false}, + {"changed", status.Error(codes.FailedPrecondition, "precondition failed"), true}, + {"deleted", status.Error(codes.NotFound, "not found"), true}, + {"outcome unknown", status.Error(codes.Unavailable, "filer down"), false}, + } { + t.Run(tc.name, func(t *testing.T) { + fake := &conditionalUpdateFiler{err: tc.err} + stale, err := updateEntryIfUnchanged(context.Background(), fake, "/dir", moved, read) + if stale != tc.stale || (err != nil) != (tc.err != nil) { + t.Fatalf("stale=%v err=%v, want stale=%v err=%v", stale, err, tc.stale, tc.err) + } + clauses := fake.req.GetCondition().GetClauses() + if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL || + !proto.Equal(clauses[0].ExpectedEntry, read) || !proto.Equal(fake.req.Entry, moved) { + t.Fatalf("request %v must write the moved entry conditioned on the read one", fake.req) + } + }) + } +} + +// A refused update may have been refused by an overlapping run that moved the +// same chunk to the same volume: moveChunk keeps the key and cookie, so that +// run's committed entry points at the very fid this run would delete. +func TestUnreferencedTargets(t *testing.T) { + committed := orphanedNeedle{volumeId: 9, fileId: "9,0101000007"} + own := orphanedNeedle{volumeId: 9, fileId: "9,0202000007"} + manifestCopy := orphanedNeedle{volumeId: 9, fileId: "9,0303000007"} + targets := func() []orphanedNeedle { return []orphanedNeedle{committed, own, manifestCopy} } + direct := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: committed.fileId}}} + + blob, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: []*filer_pb.FileChunk{{FileId: committed.fileId, Size: 10}}}) + if err != nil { + t.Fatal(err) + } + volumeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write(blob) + })) + defer volumeServer.Close() + lookupFn := func(_ context.Context, fileId string) ([]string, error) { + if fileId != manifestCopy.fileId { + return nil, fmt.Errorf("volume of %s unavailable", fileId) + } + return []string{volumeServer.URL + "/" + fileId}, nil + } + viaManifest := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: manifestCopy.fileId, IsChunkManifest: true, Size: 10}}} + unreadableManifest := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: "9,0404000007", IsChunkManifest: true, Size: 10}}} + + for _, tc := range []struct { + name string + current *filer_pb.Entry + lookupErr error + want []orphanedNeedle + }{ + {"another run committed one copy", direct, nil, []orphanedNeedle{own, manifestCopy}}, + {"another run committed copies through a manifest", viaManifest, nil, []orphanedNeedle{own}}, + {"manifest unreadable", unreadableManifest, nil, nil}, + {"entry deleted", direct, status.Error(codes.NotFound, "not found"), targets()}, + {"entry unreadable", direct, status.Error(codes.Unavailable, "filer down"), nil}, + } { + t.Run(tc.name, func(t *testing.T) { + fake := &conditionalUpdateFiler{current: tc.current, lookupErr: tc.lookupErr} + got := unreferencedTargets(context.Background(), fake, lookupFn, "/dir", "f", targets()) + if !slices.Equal(got, tc.want) { + t.Fatalf("got %v, want %v", got, tc.want) + } + if fake.lookupReq.GetDirectory() != "/dir" || fake.lookupReq.GetName() != "f" { + t.Fatalf("looked up %v, want /dir f", fake.lookupReq) + } + }) + } +} diff --git a/weed/shell/command_fs_meta_change_volume_id.go b/weed/shell/command_fs_meta_change_volume_id.go index ef5a9386a2..23bedd5f99 100644 --- a/weed/shell/command_fs_meta_change_volume_id.go +++ b/weed/shell/command_fs_meta_change_volume_id.go @@ -12,6 +12,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/protobuf/proto" ) func init() { @@ -78,6 +79,7 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv, return filer_pb.TraverseBfs(context.Background(), commandEnv, util.FullPath(*dir), func(parentPath util.FullPath, entry *filer_pb.Entry) error { if !entry.IsDirectory { var hasChanges bool + var readEntry *filer_pb.Entry for _, chunk := range entry.Chunks { if chunk.IsChunkManifest { fmt.Printf("Change volume id for large file is not implemented yet: %s/%s\n", parentPath, entry.Name) @@ -85,6 +87,9 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv, } chunkVolumeId := chunk.Fid.VolumeId if toVolumeId, found := mapping[needle.VolumeId(chunkVolumeId)]; found { + if readEntry == nil { + readEntry = proto.Clone(entry).(*filer_pb.Entry) + } hasChanges = true chunk.Fid.VolumeId = uint32(toVolumeId) chunk.FileId = "" @@ -93,10 +98,9 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv, if hasChanges { println("Updating", parentPath, entry.Name) if *applyChanges { - if updateErr := filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{ - Directory: string(parentPath), - Entry: entry, - }); updateErr != nil { + if stale, updateErr := updateEntryIfUnchanged(context.Background(), client, string(parentPath), entry, readEntry); stale { + fmt.Printf("skip %s/%s: changed since it was read\n", parentPath, entry.Name) + } else if updateErr != nil { fmt.Printf("failed to update %s/%s: %v\n", parentPath, entry.Name, updateErr) } } diff --git a/weed/shell/command_s3_versions_audit.go b/weed/shell/command_s3_versions_audit.go index 1be0b16ae1..c7b5aad164 100644 --- a/weed/shell/command_s3_versions_audit.go +++ b/weed/shell/command_s3_versions_audit.go @@ -13,6 +13,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/protobuf/proto" ) func init() { @@ -228,6 +229,7 @@ func healStrandedPointer(ctx context.Context, client filer_pb.SeaweedFilerClient if entry.Extended == nil { return nil } + readEntry := proto.Clone(entry).(*filer_pb.Entry) delete(entry.Extended, s3_constants.ExtLatestVersionIdKey) delete(entry.Extended, s3_constants.ExtLatestVersionFileNameKey) // Also clear the cached list metadata so a stale size/mtime/etag @@ -237,10 +239,9 @@ func healStrandedPointer(ctx context.Context, client filer_pb.SeaweedFilerClient delete(entry.Extended, s3_constants.ExtLatestVersionETagKey) delete(entry.Extended, s3_constants.ExtLatestVersionOwnerKey) delete(entry.Extended, s3_constants.ExtLatestVersionIsDeleteMarker) - _, err := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{ - Directory: string(parentPath), - Entry: entry, - }) + // A version written since the audit read the directory has moved the + // pointer, and clearing it would hide that version. + _, err := updateEntryIfUnchanged(ctx, client, string(parentPath), entry, readEntry) return err } diff --git a/weed/shell/command_s3_versions_audit_test.go b/weed/shell/command_s3_versions_audit_test.go new file mode 100644 index 0000000000..b32c1051e4 --- /dev/null +++ b/weed/shell/command_s3_versions_audit_test.go @@ -0,0 +1,37 @@ +package shell + +import ( + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" +) + +// The heal clears the pointer only while the directory still names the file +// the audit found missing; a version written since then has moved it. +func TestHealStrandedPointerIsConditional(t *testing.T) { + fake := &conditionalUpdateFiler{} + entry := &filer_pb.Entry{Name: "obj.versions", IsDirectory: true, Extended: map[string][]byte{ + s3_constants.ExtLatestVersionIdKey: []byte("v1"), + s3_constants.ExtLatestVersionFileNameKey: []byte("v_v1"), + "keep": []byte("x"), + }} + if err := healStrandedPointer(context.Background(), fake, "/buckets/b", entry); err != nil { + t.Fatalf("heal: %v", err) + } + + clauses := fake.req.GetCondition().GetClauses() + if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL { + t.Fatalf("heal must be conditioned on the audited entry, got %v", fake.req.GetCondition()) + } + if string(clauses[0].ExpectedEntry.Extended[s3_constants.ExtLatestVersionIdKey]) != "v1" { + t.Fatalf("expected entry must hold the pointer the audit saw, got %v", clauses[0].ExpectedEntry.Extended) + } + written := fake.req.Entry.Extended + _, idKept := written[s3_constants.ExtLatestVersionIdKey] + _, nameKept := written[s3_constants.ExtLatestVersionFileNameKey] + if idKept || nameKept || string(written["keep"]) != "x" { + t.Fatalf("written entry must clear only the pointer, got %v", written) + } +}