From 8a2fec8610ad85b0b2221d88822cb36e4c129fa0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 29 Oct 2026 13:37:27 -0700 Subject: [PATCH] filer.sync: skip superseded replays before fetching any chunks (#12212) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * filer.sync: skip superseded replays before fetching any chunks On a directory under heavy overwrite, most entries the sync replays are already superseded on the source. The sink only learned that after chunk reads failed — or worse, after fetching and uploading bytes a later event replaces — so the backlog drained slowly and the log filled with 404s. CreateEntry and UpdateEntry now check sourceMovedOn before touching a chunk: one entry lookup skips a stale replay entirely. A mid-replication supersession is still caught by the existing retry gates. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * replication: bound the source liveness probes with a timeout The supersession checks added a source entry lookup to the replication path, and every lookup ran on context.Background() with no deadline. A source that accepts the call but whose metadata store stalls would hold the sync forever — before a single chunk fetch, even when every chunk is still readable. Bound the probe at 30s; an unanswered lookup proves nothing, so replication proceeds and the existing retry gates decide. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * replication: sync the live version a superseded replay points at Skipping a stale replay assumed a later event always re-carries the current content. It does not when the superseding write was itself replicated in from another cluster: those events are dropped before they reach a filer sink, so the destination could be left missing the file. When the source reports a live entry newer than the replay, commit that version through the normal path instead — equivalent to applying the later event, with no dependence on its delivery. A gone source entry still skips, since the destination already matches the source. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * replication: reconcile moved entry content after a native rename AtomicRenameEntry relocates whatever version the destination held. If the preceding update replay was skipped because the source had already moved to the new name, the moved entry keeps the old bytes and — with no further writes — stays stale forever. MoveEntry now runs the same reconciliation as an in-place update against the rename's snapshot, so the moved entry picks up pending content changes. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * replication: repair from the live snapshot a superseded replay resolves to An incremental diff against the replayed oldEntry can resolve manifests the overwrite already vacuumed, which fails every retry of the event. --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/command/filer_sync.go | 2 +- weed/command/filer_sync_process_test.go | 2 +- weed/replication/replicator.go | 2 +- weed/replication/replicator_test.go | 2 +- .../replication/sink/filersink/fetch_write.go | 20 +- .../sink/filersink/fetch_write_test.go | 382 +++++++++++++++++- weed/replication/sink/filersink/filer_sink.go | 79 +++- weed/replication/sink/replication_sink.go | 2 +- 8 files changed, 464 insertions(+), 27 deletions(-) diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go index 85e1b533af..ae432ac538 100644 --- a/weed/command/filer_sync.go +++ b/weed/command/filer_sync.go @@ -768,7 +768,7 @@ func genProcessFunction(sourcePath string, targetPath string, excludePaths []str oldKey := buildKey(dataSink, message, targetPath, sourceOldKey, sourcePath) if mover, ok := dataSink.(sink.EntryMover); ok { // native atomic move: no re-copy, no descendant gap, no chunk leak. - return mover.MoveEntry(oldKey, newKey, message.NewEntry, message.Signatures) + return mover.MoveEntry(oldKey, newKey, message.OldEntry, message.NewEntry, message.Signatures) } // no native move: create the new entry first, then delete the // old, so a crash between the two leaves the entry visible. diff --git a/weed/command/filer_sync_process_test.go b/weed/command/filer_sync_process_test.go index 73d760cae8..132902134a 100644 --- a/weed/command/filer_sync_process_test.go +++ b/weed/command/filer_sync_process_test.go @@ -91,7 +91,7 @@ type movingSyncSink struct { var _ sink.EntryMover = (*movingSyncSink)(nil) -func (s *movingSyncSink) MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error { +func (s *movingSyncSink) MoveEntry(oldKey, newKey string, oldEntry, newEntry *filer_pb.Entry, signatures []int32) error { s.moveOldKeys = append(s.moveOldKeys, oldKey) s.moveNewKeys = append(s.moveNewKeys, newKey) s.ordered = append(s.ordered, "move") diff --git a/weed/replication/replicator.go b/weed/replication/replicator.go index e6261e2d64..c3ad29fb7d 100644 --- a/weed/replication/replicator.go +++ b/weed/replication/replicator.go @@ -116,7 +116,7 @@ func (r *Replicator) Replicate(ctx context.Context, key string, message *filer_p // A real move: the path changed. UpdateEntry cannot move an entry. if mover, ok := r.sink.(sink.EntryMover); ok { glog.V(4).Infof("moving %v => %v", oldSinkKey, newSinkKey) - return mover.MoveEntry(oldSinkKey, newSinkKey, newEntry, message.Signatures) + return mover.MoveEntry(oldSinkKey, newSinkKey, oldEntry, newEntry, message.Signatures) } // Sinks without a native move: create at the new key first, then delete the // old, so a crash between the two leaves the entry visible under both names diff --git a/weed/replication/replicator_test.go b/weed/replication/replicator_test.go index 1aab7da950..bdfcb843ec 100644 --- a/weed/replication/replicator_test.go +++ b/weed/replication/replicator_test.go @@ -96,7 +96,7 @@ type movingSink struct { var _ sink.EntryMover = (*movingSink)(nil) -func (s *movingSink) MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error { +func (s *movingSink) MoveEntry(oldKey, newKey string, oldEntry, newEntry *filer_pb.Entry, signatures []int32) error { s.moveCalls = append(s.moveCalls, moveCall{oldKey: oldKey, newKey: newKey}) s.ordered = append(s.ordered, "move") return nil diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index 2675975dd8..d67f7aa853 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -35,6 +35,20 @@ import ( // still exists is registered again. var missingSourceChunkGrace = 30 * time.Minute +// sourceLookupTimeout bounds a best-effort source entry probe: a source that +// accepts the call but whose metadata store never answers must stall +// replication only this long, not forever. A variable so tests can shorten it. +var sourceLookupTimeout = 30 * time.Second + +// sourceEntry returns the source's live entry at sourcePath within +// sourceLookupTimeout. +func (fs *FilerSink) sourceEntry(sourcePath util.FullPath) (*filer_pb.Entry, error) { + ctx, cancel := context.WithTimeout(context.Background(), sourceLookupTimeout) + defer cancel() + entry, _, _, err := filer_pb.GetEntry(ctx, fs.filerSource, sourcePath) + return entry, err +} + // errSourceChunkMissing is a permanent replication failure: the source cluster // no longer holds the chunk's data, so no later attempt can fetch it. var errSourceChunkMissing = errors.New("source chunk missing") @@ -648,7 +662,9 @@ func (fs *FilerSink) hasSourceNewerVersion(targetPath string, sourceMtimeNs int6 if !ok { return false } - return SourceSupersedes(context.Background(), fs.filerSource, sourcePath, sourceMtimeNs) + ctx, cancel := context.WithTimeout(context.Background(), sourceLookupTimeout) + defer cancel() + return SourceSupersedes(ctx, fs.filerSource, sourcePath, sourceMtimeNs) } // sourceDroppedChunk reports whether the live source entry still exists but no @@ -661,7 +677,7 @@ func (fs *FilerSink) sourceDroppedChunk(targetPath string, fileId string, source if !ok { return false } - live, _, _, err := filer_pb.GetEntry(context.Background(), fs.filerSource, sourcePath) + live, err := fs.sourceEntry(sourcePath) if err != nil || live == nil || sourceMtimeNs <= 0 || getEntryMtimeNs(live) < sourceMtimeNs { return false } diff --git a/weed/replication/sink/filersink/fetch_write_test.go b/weed/replication/sink/filersink/fetch_write_test.go index 05eb1855a0..0e3cb4fd7e 100644 --- a/weed/replication/sink/filersink/fetch_write_test.go +++ b/weed/replication/sink/filersink/fetch_write_test.go @@ -438,13 +438,19 @@ func TestManifestResolveRetryGateNonTransientPropagates(t *testing.T) { // that has vacuumed the volume away — or lost every replica of it. type sourceFilerServer struct { filer_pb.UnimplementedSeaweedFilerServer - mtime int64 - volumeUrl string - resolvable []string - entry *filer_pb.Entry + mtime int64 + volumeUrl string + resolvable []string + entry *filer_pb.Entry + lookupDelay time.Duration + lookupFn func(dir, name string) (*filer_pb.Entry, error) + volumeLookups *atomic.Int32 } func (s *sourceFilerServer) LookupVolume(ctx context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) { + if s.volumeLookups != nil { + s.volumeLookups.Add(int32(len(req.VolumeIds))) + } locationsMap := make(map[string]*filer_pb.Locations) for _, vid := range req.VolumeIds { if slices.Contains(s.resolvable, vid) { @@ -457,6 +463,17 @@ func (s *sourceFilerServer) LookupVolume(ctx context.Context, req *filer_pb.Look } func (s *sourceFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + if s.lookupDelay > 0 { + select { + case <-time.After(s.lookupDelay): + case <-ctx.Done(): + return nil, ctx.Err() + } + } + if s.lookupFn != nil { + entry, err := s.lookupFn(req.Directory, req.Name) + return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, err + } if s.entry != nil { return &filer_pb.LookupDirectoryEntryResponse{Entry: s.entry}, nil } @@ -498,13 +515,18 @@ func startSourceFiler(t *testing.T, mtime int64, volumeUrl string, resolvable .. } func startSourceFilerWithEntry(t *testing.T, entry *filer_pb.Entry, volumeUrl string, resolvable ...string) *source.FilerSource { + t.Helper() + return startSourceFilerServer(t, &sourceFilerServer{volumeUrl: volumeUrl, resolvable: resolvable, entry: entry}) +} + +func startSourceFilerServer(t *testing.T, impl *sourceFilerServer) *source.FilerSource { t.Helper() listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatalf("listen: %v", err) } server := grpc.NewServer() - filer_pb.RegisterSeaweedFilerServer(server, &sourceFilerServer{volumeUrl: volumeUrl, resolvable: resolvable, entry: entry}) + filer_pb.RegisterSeaweedFilerServer(server, impl) go server.Serve(listener) t.Cleanup(server.Stop) @@ -517,6 +539,56 @@ func startSourceFilerWithEntry(t *testing.T, entry *filer_pb.Entry, volumeUrl st return filerSrc } +// recordingFilerServer is a destination filer that counts CreateEntry and +// UpdateEntry calls and answers entry lookups from a fixed entry. +type recordingFilerServer struct { + filer_pb.UnimplementedSeaweedFilerServer + existing *filer_pb.Entry + creates atomic.Int32 + updates atomic.Int32 + renames atomic.Int32 + lastCreate atomic.Pointer[filer_pb.Entry] + lastUpdate atomic.Pointer[filer_pb.Entry] +} + +func (s *recordingFilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + if s.existing == nil { + return &filer_pb.LookupDirectoryEntryResponse{}, filer_pb.ErrNotFound + } + return &filer_pb.LookupDirectoryEntryResponse{Entry: s.existing}, nil +} + +func (s *recordingFilerServer) CreateEntry(ctx context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { + s.creates.Add(1) + s.lastCreate.Store(req.Entry) + return &filer_pb.CreateEntryResponse{}, nil +} + +func (s *recordingFilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + s.updates.Add(1) + s.lastUpdate.Store(req.Entry) + return &filer_pb.UpdateEntryResponse{}, nil +} + +func (s *recordingFilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.AtomicRenameEntryRequest) (*filer_pb.AtomicRenameEntryResponse, error) { + s.renames.Add(1) + return &filer_pb.AtomicRenameEntryResponse{}, nil +} + +func startRecordingFiler(t *testing.T, existing *filer_pb.Entry) (*recordingFilerServer, string) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + impl := &recordingFilerServer{existing: existing} + server := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(server, impl) + go server.Serve(listener) + t.Cleanup(server.Stop) + return impl, listener.Addr().String() +} + // A chunk the source cluster cannot produce must stop being retried once the // grace period is up. Before, the retry loop ran forever: the sync job never // completed, so it held its slot and pinned the offset watermark at the event @@ -793,6 +865,30 @@ func TestSourceMovedOn(t *testing.T) { } } +// A source that accepts the lookup but never answers must not hang the +// supersession probe: the bounded context gives up and replication continues. +func TestSourceMovedOnStalledLookupContinues(t *testing.T) { + prev := sourceLookupTimeout + sourceLookupTimeout = 200 * time.Millisecond + t.Cleanup(func() { sourceLookupTimeout = prev }) + + replayed := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}}, + } + fs := &FilerSink{ + filerSource: startSourceFilerServer(t, &sourceFilerServer{lookupDelay: 30 * time.Second}), + dir: "/dst", + } + start := time.Now() + if fs.sourceMovedOn("/dst/x.bin", replayed) { + t.Fatal("an unanswered lookup cannot prove supersession") + } + if time.Since(start) > 10*time.Second { + t.Fatal("the supersession probe must give up at sourceLookupTimeout") + } +} + // A replayed version the source rewrote in place must not wait out the // missing-chunk grace period: the live entry no longer lists the chunk, so the // loop stops on the first 404. @@ -886,3 +982,279 @@ func TestFetchAndWriteManifestChildKeepsRetrying(t *testing.T) { t.Fatal("a still-listed manifest's missing child must not retry past the gate") } } + +// A replayed version whose source entry is gone must not cost a single chunk +// read or destination write: on a busy directory every skipped fetch is one +// that would have 404'd or, worse, uploaded bytes a later event replaces. +func TestCreateEntrySkipsSupersededVersion(t *testing.T) { + var volumeLookups atomic.Int32 + filerSrc := startSourceFilerServer(t, &sourceFilerServer{ + lookupFn: func(dir, name string) (*filer_pb.Entry, error) { + return nil, filer_pb.ErrNotFound + }, + volumeLookups: &volumeLookups, + }) + dst, dstAddr := startRecordingFiler(t, nil) + + fs := &FilerSink{ + filerSource: filerSrc, + dir: "/dst", + grpcAddress: dstAddr, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + + replayed := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}}, + } + if err := fs.CreateEntry("/dst/x.bin", replayed, nil); err != nil { + t.Fatalf("superseded replay should be skipped, not fail: %v", err) + } + if dst.creates.Load() != 0 { + t.Fatal("a superseded version must not be committed to the destination") + } + if volumeLookups.Load() != 0 { + t.Fatal("a superseded version must not fetch any chunks") + } +} + +// The same skip must not fire while the live source still matches the replayed +// version: replication proceeds and the chunk fetch is attempted. +func TestCreateEntryReplicatesCurrentVersion(t *testing.T) { + var volumeLookups atomic.Int32 + replayed := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}}, + } + filerSrc := startSourceFilerServer(t, &sourceFilerServer{entry: replayed, volumeLookups: &volumeLookups}) + dst, dstAddr := startRecordingFiler(t, nil) + + fs := &FilerSink{ + filerSource: filerSrc, + dir: "/dst", + grpcAddress: dstAddr, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + + // The source has no volume to serve the chunk, so replication fails — the + // point is that it was attempted, not skipped as stale. A short grace lets + // the give-up path return instead of holding the test for the full wait. + prevGrace := missingSourceChunkGrace + missingSourceChunkGrace = 200 * time.Millisecond + t.Cleanup(func() { missingSourceChunkGrace = prevGrace }) + + if err := fs.CreateEntry("/dst/x.bin", replayed, nil); err == nil { + t.Fatal("replication of a chunk the source cannot serve must fail") + } + if volumeLookups.Load() == 0 { + t.Fatal("a version the source still serves must be replicated, not skipped") + } + if dst.creates.Load() != 0 { + t.Fatal("no destination write is expected while chunks cannot be fetched") + } +} + +// UpdateEntry takes the same pre-flight skip: a replayed version whose source +// entry is gone must not re-replicate or update the destination. +func TestUpdateEntrySkipsSupersededVersion(t *testing.T) { + var volumeLookups atomic.Int32 + existing := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 4}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,00aaa", Size: 10}}, + } + filerSrc := startSourceFilerServer(t, &sourceFilerServer{ + lookupFn: func(dir, name string) (*filer_pb.Entry, error) { + return nil, filer_pb.ErrNotFound + }, + volumeLookups: &volumeLookups, + }) + dst, dstAddr := startRecordingFiler(t, existing) + + fs := &FilerSink{ + filerSource: filerSrc, + dir: "/dst", + grpcAddress: dstAddr, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + + replayed := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}}, + } + found, err := fs.UpdateEntry("/dst/x.bin", existing, "", replayed, true, nil) + if err != nil { + t.Fatalf("superseded replay should be skipped, not fail: %v", err) + } + if !found { + t.Fatal("the existing entry was looked up; foundExistingEntry must report it") + } + if dst.updates.Load() != 0 { + t.Fatal("a superseded version must not be committed to the destination") + } + if volumeLookups.Load() != 0 { + t.Fatal("a superseded version must not fetch any chunks") + } +} + +// A superseded replay is replaced by the live version the source reports: the +// superseding event may never be delivered to this sink, so the live content +// must be committed now — still without fetching the replayed chunks. +func TestCreateEntrySyncsLiveVersion(t *testing.T) { + var volumeLookups atomic.Int32 + live := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 6}, + Content: []byte("live"), + } + filerSrc := startSourceFilerServer(t, &sourceFilerServer{entry: live, volumeLookups: &volumeLookups}) + dst, dstAddr := startRecordingFiler(t, nil) + + fs := &FilerSink{ + filerSource: filerSrc, + dir: "/dst", + grpcAddress: dstAddr, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + + replayed := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}}, + } + if err := fs.CreateEntry("/dst/x.bin", replayed, nil); err != nil { + t.Fatalf("syncing the live version should succeed: %v", err) + } + if dst.creates.Load() != 1 { + t.Fatal("the live version must be committed to the destination") + } + created := dst.lastCreate.Load() + if created == nil || string(created.GetContent()) != "live" || created.GetAttributes().GetMtime() != 6 { + t.Fatalf("the destination must get the live version, got %+v", created) + } + if volumeLookups.Load() != 0 { + t.Fatal("the superseded version's chunks must not be fetched") + } +} + +// UpdateEntry applies the same substitution: a superseded replay commits the +// live source version through the normal update path. +func TestUpdateEntrySyncsLiveVersion(t *testing.T) { + existing := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 4}, + Chunks: []*filer_pb.FileChunk{{FileId: "4,00aaa", SourceFileId: "5617,00aaa", Size: 10}}, + } + oldEntry := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 4}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,00aaa", Size: 10}}, + } + live := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 6}, + Content: []byte("live"), + } + filerSrc := startSourceFilerServer(t, &sourceFilerServer{entry: live}) + dst, dstAddr := startRecordingFiler(t, existing) + + fs := &FilerSink{ + filerSource: filerSrc, + dir: "/dst", + grpcAddress: dstAddr, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + + replayed := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}}, + } + found, err := fs.UpdateEntry("/dst/x.bin", oldEntry, "", replayed, true, nil) + if err != nil { + t.Fatalf("syncing the live version should succeed: %v", err) + } + if !found { + t.Fatal("the existing entry was looked up; foundExistingEntry must report it") + } + if dst.updates.Load() != 1 { + t.Fatal("the live version must be committed to the destination") + } + updated := dst.lastUpdate.Load() + if updated == nil || string(updated.GetContent()) != "live" || updated.GetAttributes().GetMtime() != 6 { + t.Fatalf("the destination must get the live version, got %+v", updated) + } +} + +// A rename carries whatever version the destination happened to hold. When the +// preceding content update was skipped — the source had already moved to the +// new name — the moved entry keeps the old bytes. MoveEntry must reconcile the +// moved entry with the event's snapshot, like an in-place update. +func TestMoveEntryReconcilesMovedContent(t *testing.T) { + existing := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 4}, + Chunks: []*filer_pb.FileChunk{{FileId: "4,0v1d", SourceFileId: "5617,0v1", Size: 10}}, + } + dst, dstAddr := startRecordingFiler(t, existing) + + live := &filer_pb.Entry{ + Name: "y.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 6}, + Content: []byte("v2"), + } + filerSrc := startSourceFilerServer(t, &sourceFilerServer{ + lookupFn: func(dir, name string) (*filer_pb.Entry, error) { + if name == "y.bin" { + return live, nil + } + return nil, filer_pb.ErrNotFound + }, + }) + + fs := &FilerSink{ + filerSource: filerSrc, + dir: "/dst", + grpcAddress: dstAddr, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + + oldEntry := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 6}, + Content: []byte("v2"), + } + newEntry := &filer_pb.Entry{ + Name: "y.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 6}, + Content: []byte("v2"), + } + if err := fs.MoveEntry("/dst/x.bin", "/dst/y.bin", oldEntry, newEntry, nil); err != nil { + t.Fatalf("move with reconcile should succeed: %v", err) + } + if dst.renames.Load() != 1 { + t.Fatal("the entry must be relocated with the native rename") + } + if dst.updates.Load() != 1 { + t.Fatal("the moved entry must be reconciled with the event's snapshot") + } + updated := dst.lastUpdate.Load() + if updated == nil || string(updated.GetContent()) != "v2" || updated.GetAttributes().GetMtime() != 6 { + t.Fatalf("the moved entry must carry the rename's content, got %+v", updated) + } +} diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index a032c84513..67657acb62 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/weed/replication/sink/filersink/filer_sink.go @@ -223,7 +223,7 @@ var _ sink.EntryMover = (*FilerSink)(nil) // is re-checked with a direct lookup rather than inferred from the rename error, // so a rolled-back move that left the old entry intact propagates for retry // instead of being mistaken for "gone". -func (fs *FilerSink) MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error { +func (fs *FilerSink) MoveEntry(oldKey, newKey string, oldEntry, newEntry *filer_pb.Entry, signatures []int32) error { oldDir, oldName := util.FullPath(oldKey).DirAndName() newDir, newName := util.FullPath(newKey).DirAndName() @@ -238,7 +238,15 @@ func (fs *FilerSink) MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, return err }) if err == nil { - return nil + // The rename relocates whatever version the destination held, which can + // predate the event's snapshot — e.g. a queued update skipped after the + // source already moved to the new name. Reconcile content the same way + // an in-place update does. + if oldEntry == nil { + return nil + } + _, uerr := fs.UpdateEntry(newKey, oldEntry, newDir, newEntry, true, signatures) + return uerr } if missing, lookupErr := fs.entryMissing(oldKey); lookupErr == nil && missing { glog.V(2).Infof("move %s => %s: old path gone, creating %s", oldKey, newKey, newKey) @@ -294,6 +302,20 @@ func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures [ key, len(resp.Entry.GetChunks()), len(entry.GetChunks())) } + // A version the source already moved past is superseded by a later + // event, so fetching its chunks only feeds a write that is obsolete on + // arrival — skip before any read. When the source holds a newer live + // version, replicate it directly: the superseding event may never be + // delivered to this sink. + if live, movedOn := fs.sourceVersion(key, entry); movedOn { + if live == nil { + glog.Warningf("skip stale entry %s, source superseded before replicate", key) + return nil + } + glog.Warningf("entry %s superseded before replicate; syncing live version", key) + entry = live + } + replicatedChunks, err := fs.replicateChunks(context.Background(), entry.GetChunks(), key, getEntryMtimeNs(entry), "") if err != nil { @@ -366,7 +388,25 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent // now, which for a rename is newParentPath/newEntry.Name, not the old key. targetKey := updatedEntryKey(key, newParentPath, newEntry) - switch chooseUpdateAction(existingEntry, oldEntry, newEntry) { + action := chooseUpdateAction(existingEntry, oldEntry, newEntry) + if action != updateSkip { + if live, movedOn := fs.sourceVersion(targetKey, newEntry); movedOn { + if live == nil { + glog.Warningf("skip stale entry %s, source superseded before replicate", targetKey) + return true, nil + } + glog.Warningf("entry %s superseded before replicate; syncing live version", targetKey) + newEntry = live + // The replayed oldEntry predates the live version, so an + // incremental diff could resolve manifests the overwrite already + // vacuumed; repair from the live snapshot instead. + if action = chooseUpdateAction(existingEntry, oldEntry, newEntry); action == updateNormal { + action = updateRepair + } + } + } + + switch action { case updateSkip: // a newer, complete version already landed (out-of-order); leave it glog.V(2).Infof("late updates %s", key) @@ -506,34 +546,43 @@ func (fs *FilerSink) onReplicateChunkError(key string, entry *filer_pb.Entry, er return fmt.Errorf("replicate entry chunks %s: %w", key, err) } -// sourceMovedOn reports whether the live source entry moved past the replayed -// version: gone, strictly newer, or rewritten in place so it no longer -// references the same chunks — the last is invisible to the mtime check. A live -// entry older than the replayed version proves nothing about it. -func (fs *FilerSink) sourceMovedOn(targetPath string, entry *filer_pb.Entry) bool { +// sourceVersion returns the source's live entry for targetPath and whether it +// moved past the replayed version: gone (live == nil), strictly newer, or +// rewritten in place so it no longer references the same chunks — the last is +// invisible to the mtime check. A live entry older than the replayed version +// proves nothing about it, and neither does a failed lookup. +func (fs *FilerSink) sourceVersion(targetPath string, entry *filer_pb.Entry) (live *filer_pb.Entry, movedOn bool) { sourcePath, ok := fs.targetPathToSourcePath(targetPath) if !ok { - return false + return nil, false } replayedMtimeNs := getEntryMtimeNs(entry) - live, _, _, lookupErr := filer_pb.GetEntry(context.Background(), fs.filerSource, sourcePath) + live, lookupErr := fs.sourceEntry(sourcePath) if sourceSupersedes(sourcePath, live, lookupErr, replayedMtimeNs) { - return true + return live, true } if lookupErr != nil || live == nil || replayedMtimeNs <= 0 || getEntryMtimeNs(live) < replayedMtimeNs { - return false + return nil, false } liveIds := chunkFileIds(live.GetChunks()) replayedIds := chunkFileIds(entry.GetChunks()) if len(liveIds) != len(replayedIds) { - return true + return live, true } for id := range replayedIds { if !liveIds[id] { - return true + return live, true } } - return false + return nil, false +} + +// sourceMovedOn reports whether the live source entry moved past the replayed +// version: gone, strictly newer, or rewritten in place so it no longer +// references the same chunks — the last is invisible to the mtime check. +func (fs *FilerSink) sourceMovedOn(targetPath string, entry *filer_pb.Entry) bool { + _, movedOn := fs.sourceVersion(targetPath, entry) + return movedOn } // updateAction decides how UpdateEntry reconciles an incoming source version with diff --git a/weed/replication/sink/replication_sink.go b/weed/replication/sink/replication_sink.go index 9528f4fb6e..a3352e14ba 100644 --- a/weed/replication/sink/replication_sink.go +++ b/weed/replication/sink/replication_sink.go @@ -29,7 +29,7 @@ type ReplicationSink interface { // committed destination, a directory move never deletes descendants before they // are recreated, and the entry's chunks are neither re-copied nor leaked. type EntryMover interface { - MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error + MoveEntry(oldKey, newKey string, oldEntry, newEntry *filer_pb.Entry, signatures []int32) error } // MetadataOnlyDeleter is an optional capability for sinks whose own delete