filer.sync: skip a rewritten source version regardless of its mtime order (#12366)

* filer.sync: a chunk the live source entry dropped is gone whatever its mtime says

sourceDroppedChunk refused to judge a 404'd chunk when the live source entry
carried an older mtime than the replayed version, treating that as a rollback.
Timestamp-preserving writers (cp -p, rsync -a, tar extracts) produce exactly
that order on every rewrite, so the vacuumed chunk sat in the retry loop for
the full missingSourceChunkGrace before being written off. Whether the live
entry still lists the chunk is independent of mtime order.

* filer.sync: treat a rewritten source entry as moved on regardless of its mtime

sourceVersion only compared chunk sets when the live entry was at least as
new as the replayed one, so a live entry stamped with an older mtime but
holding different chunks was never recognized as the current version. The
sink then fetched the replayed chunks, hit 404 on each vacuumed one, and the
same mtime guard in onReplicateChunkError kept the event from being skipped.
Whether the source rewrote the file is a chunk-set question, not an mtime
one; the live version is synced in its place as before.

* filer.sync: apply a substituted live snapshot even under a newer destination mtime

After sourceVersion swapped in the live entry, UpdateEntry re-ran
chooseUpdateAction on it and a destination whose mtime was above the live
entry's read as a late update, so the current source content was never
written. The live snapshot is the source's current state, not a replay, so
mtime order says nothing about it; repair from it unless the destination
already holds it.

* filer.sync: a live entry committed before the replay does not supersede it

Dropping the mtime guard let any chunk-set difference read as a rewrite,
including a live lookup that lags the event: a peer filer's event reaches
subscribers before the local store applies it, so the lookup can still
return the version the replay replaces. sourceVersion then synced that
older version over the destination and consumed the event. The meta-log
commit stamps order the two versions where mtime cannot, so a live entry
committed before the replay proves nothing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* filer.sync: check commit order before the mtime supersession

A lagging live entry can carry a newer mtime than the replay, as when the
replayed version was stamped back to an older time, so sourceSupersedes
took it as newer and the older version was synced in place of the replay.
Rule out a live entry committed before the replay first.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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]

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

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.