diff --git a/weed/replication/sink/filersink/fetch_write.go b/weed/replication/sink/filersink/fetch_write.go index d67f7aa853..47b381cbc8 100644 --- a/weed/replication/sink/filersink/fetch_write.go +++ b/weed/replication/sink/filersink/fetch_write.go @@ -295,7 +295,7 @@ func (fs *FilerSink) manifestResolveRetryGate(path string, sourceMtimeNs int64, chunkName, path, missingSourceChunkGrace, resolveErr) return false } - if isSourceChunkMissing(resolveErr) && fs.sourceDroppedChunk(path, chunkName, sourceMtimeNs) { + if isSourceChunkMissing(resolveErr) && fs.sourceDroppedChunk(path, chunkName) { glog.V(0).Infof("source no longer lists manifest %s for %s, skipping the obsolete version: %v", chunkName, path, resolveErr) return false @@ -522,7 +522,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, if anchorId == "" { anchorId = sourceChunk.GetFileIdString() } - if fs.sourceDroppedChunk(path, anchorId, sourceMtimeNs) { + if fs.sourceDroppedChunk(path, anchorId) { glog.V(0).Infof("source no longer lists %s for %s, skipping the obsolete version: %v", anchorId, path, retryErr) return false @@ -669,16 +669,16 @@ func (fs *FilerSink) hasSourceNewerVersion(targetPath string, sourceMtimeNs int6 // sourceDroppedChunk reports whether the live source entry still exists but no // longer references fileId — a version rewritten in place, which the mtime -// supersession check cannot see. Only a live entry at least as new as the -// replayed version can prove that; an older one means the source was rolled -// back and says nothing about whether the chunk is still needed. -func (fs *FilerSink) sourceDroppedChunk(targetPath string, fileId string, sourceMtimeNs int64) bool { +// supersession check cannot see. The live entry's mtime does not matter: a +// client that preserves timestamps stamps a rewrite with an older mtime, and +// the chunk it stopped referencing is just as gone. +func (fs *FilerSink) sourceDroppedChunk(targetPath string, fileId string) bool { sourcePath, ok := fs.targetPathToSourcePath(targetPath) if !ok { return false } live, err := fs.sourceEntry(sourcePath) - if err != nil || live == nil || sourceMtimeNs <= 0 || getEntryMtimeNs(live) < sourceMtimeNs { + if err != nil || live == nil { return false } return !chunkFileIds(live.GetChunks())[fileId] diff --git a/weed/replication/sink/filersink/fetch_write_test.go b/weed/replication/sink/filersink/fetch_write_test.go index 0e3cb4fd7e..75dc53c061 100644 --- a/weed/replication/sink/filersink/fetch_write_test.go +++ b/weed/replication/sink/filersink/fetch_write_test.go @@ -824,15 +824,15 @@ func TestSourceDroppedChunk(t *testing.T) { } fs := &FilerSink{filerSource: startSourceFilerWithEntry(t, live, ""), dir: "/dst"} - if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", 5*int64(time.Second)) { + if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc") { t.Fatal("live entry no longer referencing the chunk must count as dropped") } - if fs.sourceDroppedChunk("/dst/x.bin", "5617,02def", 5*int64(time.Second)) { + if fs.sourceDroppedChunk("/dst/x.bin", "5617,02def") { t.Fatal("a chunk the live entry still lists must not count as dropped") } live.Attributes.Mtime = 4 - if fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", 5*int64(time.Second)) { - t.Fatal("a live entry older than the replayed version cannot prove the chunk dropped") + if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc") { + t.Fatal("a live entry stamped older than the replayed version still dropped the chunk") } } @@ -859,9 +859,28 @@ func TestSourceMovedOn(t *testing.T) { t.Fatal("newer mtime is still moved on") } live.Attributes.Mtime = 4 - replayed.Chunks = []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}} if fs.sourceMovedOn("/dst/x.bin", replayed) { - t.Fatal("an older live entry cannot supersede the replayed version") + t.Fatal("identical chunks under an older mtime is a timestamp change, not a new version") + } + replayed.Chunks = []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}} + if !fs.sourceMovedOn("/dst/x.bin", replayed) { + t.Fatal("a live entry stamped older but holding different chunks has moved on") + } + live.CommitTsNs, replayed.CommitTsNs = 20, 10 + if !fs.sourceMovedOn("/dst/x.bin", replayed) { + t.Fatal("a rewrite committed after the replay has moved on whatever its mtime") + } + live.CommitTsNs = 5 + if fs.sourceMovedOn("/dst/x.bin", replayed) { + t.Fatal("a live entry committed before the replay lags the event and proves nothing") + } + live.CommitTsNs = 0 + if fs.sourceMovedOn("/dst/x.bin", replayed) { + t.Fatal("an unstamped live entry predates a stamped replay") + } + live.CommitTsNs, live.Attributes.Mtime = 5, 6 + if fs.sourceMovedOn("/dst/x.bin", replayed) { + t.Fatal("a lagging live entry with a newer mtime still predates the replay") } } @@ -935,6 +954,53 @@ func TestFetchAndWriteStopsWhenSourceDroppedChunk(t *testing.T) { } } +// Mtime order says nothing about which version is current: a client that +// preserves timestamps (cp -p, rsync -a, tar) rewrites a file and then stamps +// it with an older mtime, so the live entry reads older than the replay whose +// chunk it no longer lists. The missing chunk is still gone for good. +func TestFetchAndWriteStopsWhenOlderLiveEntryDroppedChunk(t *testing.T) { + prevRetryWaitTime := util.RetryWaitTime + util.RetryWaitTime = 100 * time.Millisecond + t.Cleanup(func() { util.RetryWaitTime = prevRetryWaitTime }) + + gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "not found", http.StatusNotFound) + })) + defer gone.Close() + + live := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: 4}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,02def", Size: 10}}, + } + filerSrc := startSourceFilerWithEntry(t, live, strings.TrimPrefix(gone.URL, "http://"), "5617") + + fs := &FilerSink{ + filerSource: filerSrc, + dir: "/dst", + executor: util.NewLimitedConcurrentExecutor(1), + } + fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient)) + + done := make(chan error, 1) + go func() { + _, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, + "/dst/x.bin", 5*int64(time.Second), "") + done <- fetchErr + }() + + select { + case fetchErr := <-done: + if !errors.Is(fetchErr, util_http.ErrNotFound) { + t.Fatalf("expected the 404 to propagate, got %v", fetchErr) + } + if errors.Is(fetchErr, errSourceChunkMissing) { + t.Fatalf("an obsolete version must not wait for the give-up gate: %v", fetchErr) + } + case <-time.After(10 * time.Second): + t.Fatal("fetchAndWrite kept retrying a chunk the live entry no longer lists") + } +} + // A missing needle that lives inside a manifest is judged by whether the live // entry still lists the manifest, not the child: the manifest id is the // identity of its contents, so a live entry that kept it still needs the child. @@ -1199,6 +1265,144 @@ func TestUpdateEntrySyncsLiveVersion(t *testing.T) { } } +// A timestamp-preserving rewrite leaves the live source older than the replay +// whose chunks it dropped, and older than a destination that already took an +// earlier version. The destination must still converge on the live version +// instead of fetching chunks the source vacuumed or keeping its newer-stamped +// stale bytes. +func TestUpdateEntrySyncsOlderLiveVersion(t *testing.T) { + var volumeLookups atomic.Int32 + existing := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "4,00aaa", SourceFileId: "5617,00aaa", Size: 10}}, + } + oldEntry := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,00aaa", Size: 10}}, + } + live := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 4}, + Content: []byte("live"), + } + filerSrc := startSourceFilerServer(t, &sourceFilerServer{entry: live, 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: 6}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}}, + } + if _, err := fs.UpdateEntry("/dst/x.bin", oldEntry, "", replayed, true, nil); err != nil { + t.Fatalf("syncing the live version should succeed: %v", err) + } + 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() != 4 { + t.Fatalf("the destination must get the live version, got %+v", updated) + } + if volumeLookups.Load() != 0 { + t.Fatal("the superseded version's chunks must not be fetched") + } +} + +// A peer filer's event reaches subscribers before the local store applies it, +// so the live lookup can still return the version the replay replaces. That +// lagging entry was committed earlier and must not be synced over the replay. +func TestUpdateEntryAppliesReplayAheadOfLiveLookup(t *testing.T) { + var volumeLookups atomic.Int32 + existing := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "4,00aaa", SourceFileId: "5617,00aaa", Size: 10}}, + } + oldEntry := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 5}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,00aaa", Size: 10}}, + CommitTsNs: 5, + } + filerSrc := startSourceFilerServer(t, &sourceFilerServer{entry: oldEntry, 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: 6}, + Content: []byte("new"), + CommitTsNs: 10, + } + if _, err := fs.UpdateEntry("/dst/x.bin", oldEntry, "", replayed, true, nil); err != nil { + t.Fatalf("applying the replay should succeed: %v", err) + } + updated := dst.lastUpdate.Load() + if updated == nil || string(updated.GetContent()) != "new" || len(updated.GetChunks()) != 0 { + t.Fatalf("the destination must get the replayed version, got %+v", updated) + } + if volumeLookups.Load() != 0 { + t.Fatal("the lagging live entry's chunks must not be fetched") + } +} + +// Once the destination already holds the live version, the same stale replay +// must not re-replicate it. +func TestUpdateEntryKeepsDestinationHoldingLiveVersion(t *testing.T) { + live := &filer_pb.Entry{ + Name: "x.bin", + Attributes: &filer_pb.FuseAttributes{Mtime: 4}, + Content: []byte("live"), + } + filerSrc := startSourceFilerServer(t, &sourceFilerServer{entry: live}) + dst, dstAddr := startRecordingFiler(t, live) + + 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: 3}, + Chunks: []*filer_pb.FileChunk{{FileId: "5617,00aaa", Size: 10}}, + } + 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.UpdateEntry("/dst/x.bin", oldEntry, "", replayed, true, nil); err != nil { + t.Fatalf("a destination holding the live version should be left alone: %v", err) + } + if dst.updates.Load() != 0 { + t.Fatal("a destination already holding the live version must not be rewritten") + } +} + // 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 diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index 4a4d7ea1c4..6ecc92f3f0 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/weed/replication/sink/filersink/filer_sink.go @@ -419,11 +419,14 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent } 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 + // The live snapshot is the source's current state, not a replay: + // a newer destination mtime does not make it stale, and the + // replayed oldEntry predates it, so an incremental diff could + // resolve manifests the overwrite already vacuumed. Repair from + // the snapshot unless the destination already holds it. + action = updateRepair + if destinationHoldsContent(existingEntry, live) { + action = updateSkip } } } @@ -570,20 +573,25 @@ func (fs *FilerSink) onReplicateChunkError(key string, entry *filer_pb.Entry, er // 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. +// rewritten so it no longer references the same chunks — the last is invisible +// to the mtime check, and holds whatever mtime the rewrite was stamped with, +// since timestamp-preserving writers make a newer version read older. A failed +// lookup proves nothing, and neither does a live entry committed before the +// replayed version: the lookup lags the event, as when a peer filer's event +// reaches subscribers before the local store applies it. func (fs *FilerSink) sourceVersion(targetPath string, entry *filer_pb.Entry) (live *filer_pb.Entry, movedOn bool) { sourcePath, ok := fs.targetPathToSourcePath(targetPath) if !ok { return nil, false } - replayedMtimeNs := getEntryMtimeNs(entry) live, lookupErr := fs.sourceEntry(sourcePath) - if sourceSupersedes(sourcePath, live, lookupErr, replayedMtimeNs) { + if lookupErr == nil && live != nil && committedBefore(live, entry) { + return nil, false + } + if sourceSupersedes(sourcePath, live, lookupErr, getEntryMtimeNs(entry)) { return live, true } - if lookupErr != nil || live == nil || replayedMtimeNs <= 0 || getEntryMtimeNs(live) < replayedMtimeNs { + if lookupErr != nil || live == nil { return nil, false } liveIds := chunkFileIds(live.GetChunks()) @@ -599,6 +607,13 @@ func (fs *FilerSink) sourceVersion(targetPath string, entry *filer_pb.Entry) (li return nil, false } +// committedBefore reports whether the source committed live before replayed. +// Both stamps come from the source's meta log, so they order versions that +// mtime cannot; an unstamped live entry predates a stamped replay. +func committedBefore(live, replayed *filer_pb.Entry) bool { + return replayed.GetCommitTsNs() > 0 && live.GetCommitTsNs() < replayed.GetCommitTsNs() +} + // 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.