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