filer.sync: stop chunk retries with the same supersession check the skip uses (#12367)

* filer.sync: pass the replayed entry down to the chunk retry gates The fetch, manifest-resolve, and manifest-upload retry gates only received the replayed version's mtime, which is not enough to order it against the live source entry. Pass the entry itself; the gates still compare mtime only, so behavior is unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer.sync: stop chunk retries with the same supersession check the skip uses The fetch, manifest-resolve, and manifest-upload retry gates stopped early only for a gone or strictly newer source entry. A transient error therefore kept retrying a version the source had rewritten under an older mtime until the read recovered, while a live lookup that lags the replay could stop the retries on its newer mtime alone. The error handler that runs once a gate stops already decides with sourceMovedOn, which orders by commit timestamp and compares chunk sets; the gates now use it too, and hasSourceNewerVersion has no callers left. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer.sync: a lagging live lookup cannot prove a chunk dropped A peer filer's event reaches subscribers before the source store applies it, so the live lookup can return the version the replay replaces. That entry never listed the replay's new chunk, so sourceDroppedChunk read a 404 on it as an obsolete version and cut the missing-chunk wait short. Like sourceVersion, it now ignores a live entry committed before the replayed version. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer.sync: test that a replicate failure on a superseded version is skipped Covers onReplicateChunkError with a transient error for a deleted source entry, a later commit with an older mtime, and an unchanged source, and points the unconfirmed-supersession test comment at it instead of the helper test. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer.sync: test the manifest gate against a missing volume and a lagging lookup Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-04 11:07:35 +01:00 · 2026-11-03 10:43:22 -08:00
1 parent 8461ae2fe9
commit cadb614630
@@ -104,7 +104,7 @@ func (g *missingSourceChunkGate) wrap(err error) error {
 // topLevelId anchors nested chunks to the entry-level chunk that contains them:
 // supersession is judged by whether the live entry still lists that chunk, since
 // a manifest's id is the identity of its contents.
-func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (replicatedChunks []*filer_pb.FileChunk, err error) {
+func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_pb.FileChunk, path string, replayed *filer_pb.Entry, topLevelId string) (replicatedChunks []*filer_pb.FileChunk, err error) {
 	if len(sourceChunks) == 0 {
 		return
 	}
@@ -155,7 +155,7 @@ func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_
 			if anchor == "" {
 				anchor = sourceChunk.GetFileIdString()
 			}
-			replicatedChunk, replicateErr := fs.replicateOneManifestChunk(ctx, sourceChunk, path, sourceMtimeNs, anchor)
+			replicatedChunk, replicateErr := fs.replicateOneManifestChunk(ctx, sourceChunk, path, replayed, anchor)
 			if replicateErr != nil {
 				setError(replicateErr)
 				break
@@ -173,7 +173,7 @@ func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_
 			defer wg.Done()
 			var replicatedChunk *filer_pb.FileChunk
 			retryErr := util.Retry("replicate chunks", func() error {
-				chunk, e := fs.replicateOneChunk(source, path, sourceMtimeNs, topLevelId)
+				chunk, e := fs.replicateOneChunk(source, path, replayed, topLevelId)
 				if e != nil {
 					return e
 				}
@@ -196,9 +196,9 @@ func (fs *FilerSink) replicateChunks(ctx context.Context, sourceChunks []*filer_
 	return
 }

-func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (*filer_pb.FileChunk, error) {
+func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path string, replayed *filer_pb.Entry, topLevelId string) (*filer_pb.FileChunk, error) {

-	fileId, err := fs.fetchAndWrite(sourceChunk, path, sourceMtimeNs, topLevelId)
+	fileId, err := fs.fetchAndWrite(sourceChunk, path, replayed, topLevelId)
 	if err != nil {
 		return nil, fmt.Errorf("copy %s: %w", sourceChunk.GetFileIdString(), err)
 	}
@@ -217,7 +217,7 @@ func (fs *FilerSink) replicateOneChunk(sourceChunk *filer_pb.FileChunk, path str
 	}, nil
 }

-func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (*filer_pb.FileChunk, error) {
+func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk *filer_pb.FileChunk, path string, replayed *filer_pb.Entry, topLevelId string) (*filer_pb.FileChunk, error) {
 	// LookupFileId can transiently fail during a write burst (volume not yet
 	// registered). Retry like the data-chunk path does; the gate stops on
 	// supersession or, when that cannot be checked, after a few attempts.
@@ -231,12 +231,12 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk
 		}
 		resolvedChunks = rc
 		return nil
-	}, fs.manifestResolveRetryGate(path, sourceMtimeNs, topLevelId, missingGate))
+	}, fs.manifestResolveRetryGate(path, replayed, topLevelId, missingGate))
 	if err != nil {
 		return nil, fmt.Errorf("resolve manifest %s: %w", sourceChunk.GetFileIdString(), missingGate.wrap(err))
 	}

-	replicatedResolvedChunks, err := fs.replicateChunks(ctx, resolvedChunks, path, sourceMtimeNs, topLevelId)
+	replicatedResolvedChunks, err := fs.replicateChunks(ctx, resolvedChunks, path, replayed, topLevelId)
 	if err != nil {
 		return nil, fmt.Errorf("replicate manifest data chunks %s: %w", sourceChunk.GetFileIdString(), err)
 	}
@@ -253,7 +253,7 @@ func (fs *FilerSink) replicateOneManifestChunk(ctx context.Context, sourceChunk
 		return nil, fmt.Errorf("marshal manifest %s: %w", sourceChunk.GetFileIdString(), err)
 	}

-	manifestFileId, err := fs.uploadManifestChunk(path, sourceMtimeNs, sourceChunk.GetFileIdString(), manifestData)
+	manifestFileId, err := fs.uploadManifestChunk(path, replayed, sourceChunk.GetFileIdString(), manifestData)
 	if err != nil {
 		return nil, err
 	}
@@ -282,11 +282,11 @@ const maxUnverifiableResolveAttempts = 3
 // dated target keys don't map back to a source path) — propagating lets
 // filer.backup decide with the event's real source key instead of spinning
 // here forever.
-func (fs *FilerSink) manifestResolveRetryGate(path string, sourceMtimeNs int64, chunkName string, missingGate *missingSourceChunkGate) func(error) bool {
+func (fs *FilerSink) manifestResolveRetryGate(path string, replayed *filer_pb.Entry, chunkName string, missingGate *missingSourceChunkGate) func(error) bool {
 	_, canCheckSupersession := fs.targetPathToSourcePath(path)
 	attempts := 0
 	return func(resolveErr error) (shouldContinue bool) {
-		if fs.hasSourceNewerVersion(path, sourceMtimeNs) {
+		if fs.sourceMovedOn(path, replayed) {
 			glog.V(1).Infof("skip retrying stale source manifest %s for %s: %v", chunkName, path, resolveErr)
 			return false
 		}
@@ -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) {
+		if isSourceChunkMissing(resolveErr) && fs.sourceDroppedChunk(path, chunkName, replayed) {
 			glog.V(0).Infof("source no longer lists manifest %s for %s, skipping the obsolete version: %v",
 				chunkName, path, resolveErr)
 			return false
@@ -330,7 +330,7 @@ func isTransientResolveError(err error) bool {
 		(strings.Contains(errStr, "volume id") && strings.Contains(errStr, "not found"))
 }

-func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourceFileId string, manifestData []byte) (fileId string, err error) {
+func (fs *FilerSink) uploadManifestChunk(path string, replayed *filer_pb.Entry, sourceFileId string, manifestData []byte) (fileId string, err error) {
 	uploader, err := fs.getUploader()
 	if err != nil {
 		glog.V(0).Infof("upload manifest data %v: %v", sourceFileId, err)
@@ -375,7 +375,7 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourc
 		fileId = currentFileId
 		return nil
 	}, func(uploadErr error) (shouldContinue bool) {
-		if fs.hasSourceNewerVersion(path, sourceMtimeNs) {
+		if fs.sourceMovedOn(path, replayed) {
 			glog.V(1).Infof("skip retrying stale source manifest %s for %s: %v", sourceFileId, path, uploadErr)
 			return false
 		}
@@ -389,7 +389,7 @@ func (fs *FilerSink) uploadManifestChunk(path string, sourceMtimeNs int64, sourc
 	return fileId, nil
 }

-func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, sourceMtimeNs int64, topLevelId string) (fileId string, err error) {
+func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string, replayed *filer_pb.Entry, topLevelId string) (fileId string, err error) {
 	uploader, err := fs.getUploader()
 	if err != nil {
 		glog.V(0).Infof("upload source data %v: %v", sourceChunk.GetFileIdString(), err)
@@ -505,7 +505,7 @@ func (fs *FilerSink) fetchAndWrite(sourceChunk *filer_pb.FileChunk, path string,
 			transferStatus.mu.Unlock()
 			return false
 		}
-		if fs.hasSourceNewerVersion(path, sourceMtimeNs) {
+		if fs.sourceMovedOn(path, replayed) {
 			glog.V(1).Infof("skip retrying stale source %s for %s: %v", sourceChunk.GetFileIdString(), path, retryErr)
 			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) {
+			if fs.sourceDroppedChunk(path, anchorId, replayed) {
 				glog.V(0).Infof("source no longer lists %s for %s, skipping the obsolete version: %v",
 					anchorId, path, retryErr)
 				return false
@@ -653,32 +653,19 @@ func (fs *FilerSink) sourceStillServesChunks() bool {
 	return true
 }

-// hasSourceNewerVersion reports whether the source's current entry for targetPath
-// has moved past sourceMtimeNs — gone, or a strictly-newer mtime — meaning the
-// version being replayed is stale. The lookup runs regardless of sourceMtimeNs so
-// a deleted source is detected even when the replayed mtime is epoch/unset.
-func (fs *FilerSink) hasSourceNewerVersion(targetPath string, sourceMtimeNs int64) bool {
-	sourcePath, ok := fs.targetPathToSourcePath(targetPath)
-	if !ok {
-		return false
-	}
-	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
 // longer references fileId — a version rewritten in place, which the mtime
 // 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 {
+// the chunk it stopped referencing is just as gone. A live entry committed
+// before the replayed version lags the event and proves nothing.
+func (fs *FilerSink) sourceDroppedChunk(targetPath string, fileId string, replayed *filer_pb.Entry) bool {
 	sourcePath, ok := fs.targetPathToSourcePath(targetPath)
 	if !ok {
 		return false
 	}
 	live, err := fs.sourceEntry(sourcePath)
-	if err != nil || live == nil {
+	if err != nil || live == nil || committedBefore(live, replayed) {
 		return false
 	}
 	return !chunkFileIds(live.GetChunks())[fileId]
@@ -233,7 +233,7 @@ func TestFetchAndWriteRejectsZeroByteSource(t *testing.T) {
 		err    error
 	}, 1)
 	go func() {
-		gotFileId, gotErr := fs.fetchAndWrite(sourceChunk, "/dst/index.bin", 0, "")
+		gotFileId, gotErr := fs.fetchAndWrite(sourceChunk, "/dst/index.bin", nil, "")
 		done <- struct {
 			fileId string
 			err    error
@@ -331,7 +331,7 @@ func TestReplicateChunksPreservesSizeMismatchSentinel(t *testing.T) {

 	sourceChunks := []*filer_pb.FileChunk{{FileId: fid, Size: expectedSize}}

-	_, err := fs.replicateChunks(nil, sourceChunks, "/dst/index.bin", 0, "")
+	_, err := fs.replicateChunks(nil, sourceChunks, "/dst/index.bin", nil, "")
 	if err == nil {
 		t.Fatal("expected error from replicateChunks, got nil")
 	}
@@ -396,7 +396,7 @@ func TestSourceSupersedesEpochMtime(t *testing.T) {
 // attempts and propagate instead of spinning forever.
 func TestManifestResolveRetryGateUnverifiableSupersessionBounded(t *testing.T) {
 	fs := &FilerSink{isIncremental: true, dir: "/backup"}
-	gate := fs.manifestResolveRetryGate("/backup/2026-07-10/buckets/x/f.pt", 123, "3,01abc", fs.newMissingSourceChunkGate("3,01abc"))
+	gate := fs.manifestResolveRetryGate("/backup/2026-07-10/buckets/x/f.pt", entryWithMtimeNs(123), "3,01abc", fs.newMissingSourceChunkGate("3,01abc"))
 	resolveErr := errors.New("LookupFileId volume id 3: not found")
 	for i := 1; i < maxUnverifiableResolveAttempts; i++ {
 		if !gate(resolveErr) {
@@ -414,7 +414,7 @@ func TestManifestResolveRetryGateUnverifiableSupersessionBounded(t *testing.T) {
 // retrying until the source is superseded.
 func TestManifestResolveRetryGateNonTransientPropagates(t *testing.T) {
 	fs := &FilerSink{dir: "/backup"}
-	gate := fs.manifestResolveRetryGate("/backup/buckets/x/f.pt", 123, "3,01abc", fs.newMissingSourceChunkGate("3,01abc"))
+	gate := fs.manifestResolveRetryGate("/backup/buckets/x/f.pt", entryWithMtimeNs(123), "3,01abc", fs.newMissingSourceChunkGate("3,01abc"))
 	permanentErrs := []error{
 		errors.New("fail to unmarshal manifest 3,01abc: proto: cannot parse invalid wire-format data"),
 		errors.New("invalid fileId abc"),
@@ -616,7 +616,7 @@ func TestFetchAndWriteStopsOnMissingSourceChunk(t *testing.T) {

 	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), "")
+		_, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10}, "/dst/x.bin", live, "")
 		done <- fetchErr
 	}()

@@ -800,7 +800,7 @@ func TestFetchAndWriteMissingSourceChunkUnverifiableSupersession(t *testing.T) {
 	done := make(chan error, 1)
 	go func() {
 		_, fetchErr := fs.fetchAndWrite(&filer_pb.FileChunk{FileId: "5617,01abc", Size: 10},
-			"/backup/2026-07-10/x.bin", 5*int64(time.Second), "")
+			"/backup/2026-07-10/x.bin", entryWithMtimeNs(5*int64(time.Second)), "")
 		done <- fetchErr
 	}()

@@ -823,17 +823,34 @@ func TestSourceDroppedChunk(t *testing.T) {
 		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,02def", Size: 10}},
 	}
 	fs := &FilerSink{filerSource: startSourceFilerWithEntry(t, live, ""), dir: "/dst"}
+	replayed := entryWithMtimeNs(5 * int64(time.Second))

-	if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc") {
+	if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", replayed) {
 		t.Fatal("live entry no longer referencing the chunk must count as dropped")
 	}
-	if fs.sourceDroppedChunk("/dst/x.bin", "5617,02def") {
+	if fs.sourceDroppedChunk("/dst/x.bin", "5617,02def", replayed) {
 		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") {
+	if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", replayed) {
 		t.Fatal("a live entry stamped older than the replayed version still dropped the chunk")
 	}
+
+	replayed.CommitTsNs, live.CommitTsNs = 100, 200
+	if !fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", replayed) {
+		t.Fatal("a live entry committed after the replay dropped the chunk")
+	}
+	live.CommitTsNs, live.Attributes.Mtime = 50, 6
+	if fs.sourceDroppedChunk("/dst/x.bin", "5617,01abc", replayed) {
+		t.Fatal("a live entry committed before the replay lags the event and cannot prove the chunk dropped")
+	}
+}
+
+func entryWithMtimeNs(mtimeNs int64) *filer_pb.Entry {
+	return &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{
+		Mtime:   mtimeNs / int64(time.Second),
+		MtimeNs: int32(mtimeNs % int64(time.Second)),
+	}}
 }

 func TestSourceMovedOn(t *testing.T) {
@@ -937,7 +954,7 @@ func TestFetchAndWriteStopsWhenSourceDroppedChunk(t *testing.T) {
 	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), "")
+			"/dst/x.bin", entryWithMtimeNs(5*int64(time.Second)), "")
 		done <- fetchErr
 	}()

@@ -984,7 +1001,7 @@ func TestFetchAndWriteStopsWhenOlderLiveEntryDroppedChunk(t *testing.T) {
 	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), "")
+			"/dst/x.bin", entryWithMtimeNs(5*int64(time.Second)), "")
 		done <- fetchErr
 	}()

@@ -1035,7 +1052,7 @@ func TestFetchAndWriteManifestChildKeepsRetrying(t *testing.T) {
 	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), "5617,09fed")
+			"/dst/x.bin", live, "5617,09fed")
 		done <- fetchErr
 	}()

@@ -1049,6 +1066,165 @@ func TestFetchAndWriteManifestChildKeepsRetrying(t *testing.T) {
 	}
 }

+// A transient read error must not keep retrying a version the source has
+// since committed over, even when the later version carries an older mtime.
+func TestFetchAndWriteStopsOnTransientErrorWhenCommittedOver(t *testing.T) {
+	refused, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	refusedAddr := refused.Addr().String()
+	refused.Close()
+
+	live, replayed := committedOverWithOlderMtime()
+	fs := &FilerSink{
+		filerSource: startSourceFilerWithEntry(t, live, refusedAddr, "5617"),
+		dir:         "/dst",
+		executor:    util.NewLimitedConcurrentExecutor(1),
+	}
+	fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient))
+
+	done := make(chan error, 1)
+	go func() {
+		_, fetchErr := fs.fetchAndWrite(replayed.Chunks[0], "/dst/x.bin", replayed, "")
+		done <- fetchErr
+	}()
+
+	select {
+	case fetchErr := <-done:
+		if fetchErr == nil {
+			t.Fatal("a read against a refused volume server cannot succeed")
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("fetchAndWrite kept retrying a version the source committed over")
+	}
+}
+
+// The resolve gate stops on the same supersession check, and like it ignores a
+// live lookup that lags the replay: an entry committed earlier proves nothing,
+// even with a newer mtime or without the replayed manifest.
+func TestManifestResolveRetryGateOrdersByCommit(t *testing.T) {
+	transient := errors.New("dial tcp: connection refused")
+	live, replayed := committedOverWithOlderMtime()
+	manifestId := replayed.Chunks[0].GetFileIdString()
+
+	fs := &FilerSink{filerSource: startSourceFilerWithEntry(t, live, ""), dir: "/dst"}
+	gate := fs.manifestResolveRetryGate("/dst/x.bin", replayed, manifestId, fs.newMissingSourceChunkGate(manifestId))
+	if gate(transient) {
+		t.Fatal("a transient resolve error must not keep retrying a version the source committed over")
+	}
+
+	lagging := &filer_pb.Entry{
+		Attributes: &filer_pb.FuseAttributes{Mtime: 6},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,00aaa", Size: 10}},
+		CommitTsNs: 50,
+	}
+	fs = &FilerSink{filerSource: startSourceFilerWithEntry(t, lagging, ""), dir: "/dst"}
+	gate = fs.manifestResolveRetryGate("/dst/x.bin", replayed, manifestId, fs.newMissingSourceChunkGate(manifestId))
+	if !gate(transient) {
+		t.Fatal("a live lookup committed before the replay must not stop its retries")
+	}
+
+	volumeGone := fmt.Errorf("LookupFileId locate volume id 5617: %w", source.ErrVolumeNotFound)
+	gate = fs.manifestResolveRetryGate("/dst/x.bin", replayed, manifestId, fs.newMissingSourceChunkGate(manifestId))
+	if !gate(volumeGone) {
+		t.Fatal("a live lookup committed before the replay must not cut a missing volume's wait short")
+	}
+}
+
+// A replicate failure on a version the source has moved past is skipped: the
+// later event carries the current content. The error here is transient, not a
+// missing chunk, so only the supersession check can turn it into a skip.
+func TestOnReplicateChunkErrorSkipsSupersededVersion(t *testing.T) {
+	transient := errors.New("copy 5617,01abc: dial tcp: connection refused")
+	live, replayed := committedOverWithOlderMtime()
+
+	t.Run("source entry gone", func(t *testing.T) {
+		fs := &FilerSink{filerSource: startSourceFilerServer(t, &sourceFilerServer{
+			lookupFn: func(dir, name string) (*filer_pb.Entry, error) {
+				return nil, filer_pb.ErrNotFound
+			},
+		}), dir: "/dst"}
+		if err := fs.onReplicateChunkError("/dst/x.bin", replayed, transient); err != nil {
+			t.Fatalf("a deleted source entry must skip the replay, got %v", err)
+		}
+	})
+
+	t.Run("committed over with an older mtime", func(t *testing.T) {
+		fs := &FilerSink{filerSource: startSourceFilerWithEntry(t, live, ""), dir: "/dst"}
+		if err := fs.onReplicateChunkError("/dst/x.bin", replayed, transient); err != nil {
+			t.Fatalf("a later commit that dropped the replayed chunks must skip the replay, got %v", err)
+		}
+	})
+
+	t.Run("source unchanged", func(t *testing.T) {
+		fs := &FilerSink{filerSource: startSourceFilerWithEntry(t, replayed, ""), dir: "/dst"}
+		if err := fs.onReplicateChunkError("/dst/x.bin", replayed, transient); !errors.Is(err, transient) {
+			t.Fatalf("a failure on the current version must propagate, got %v", err)
+		}
+	})
+}
+
+func committedOverWithOlderMtime() (live, replayed *filer_pb.Entry) {
+	live = &filer_pb.Entry{
+		Attributes: &filer_pb.FuseAttributes{Mtime: 4},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,02def", Size: 10}},
+		CommitTsNs: 200,
+	}
+	replayed = &filer_pb.Entry{
+		Attributes: &filer_pb.FuseAttributes{Mtime: 5},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,01abc", Size: 10}},
+		CommitTsNs: 100,
+	}
+	return live, replayed
+}
+
+// A peer filer's event can reach the sink before the source store applies it,
+// so the live lookup returns the version the replay replaces, which never
+// listed the replay's chunk. That lagging entry cannot prove the chunk dropped:
+// the 404 must be waited out like any other, not cut short.
+func TestFetchAndWriteLaggingLookupKeepsRetrying(t *testing.T) {
+	prevRetryWaitTime, prevGrace := util.RetryWaitTime, missingSourceChunkGrace
+	util.RetryWaitTime = 100 * time.Millisecond
+	missingSourceChunkGrace = 500 * time.Millisecond
+	t.Cleanup(func() {
+		util.RetryWaitTime, missingSourceChunkGrace = prevRetryWaitTime, prevGrace
+	})
+
+	gone := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "not found", http.StatusNotFound)
+	}))
+	defer gone.Close()
+
+	lagging := &filer_pb.Entry{
+		Attributes: &filer_pb.FuseAttributes{Mtime: 4},
+		Chunks:     []*filer_pb.FileChunk{{FileId: "5617,00aaa", Size: 10}},
+		CommitTsNs: 50,
+	}
+	_, replayed := committedOverWithOlderMtime()
+	fs := &FilerSink{
+		filerSource: startSourceFilerWithEntry(t, lagging, strings.TrimPrefix(gone.URL, "http://"), "5617"),
+		dir:         "/dst",
+		executor:    util.NewLimitedConcurrentExecutor(1),
+	}
+	fs.SetUploader(operation.NewUploaderWithHttpClient(http.DefaultClient))
+
+	done := make(chan error, 1)
+	go func() {
+		_, fetchErr := fs.fetchAndWrite(replayed.Chunks[0], "/dst/x.bin", replayed, "")
+		done <- fetchErr
+	}()
+
+	select {
+	case fetchErr := <-done:
+		if !errors.Is(fetchErr, errSourceChunkMissing) {
+			t.Fatalf("a lagging live lookup must not cut the missing-chunk wait short, got %v", fetchErr)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("the missing chunk must still expire via 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.
@@ -338,7 +338,7 @@ func (fs *FilerSink) createEntry(key string, entry *filer_pb.Entry, signatures [
 			}
 		}

-		replicatedChunks, err := fs.replicateChunks(context.Background(), entry.GetChunks(), key, getEntryMtimeNs(entry), "")
+		replicatedChunks, err := fs.replicateChunks(context.Background(), entry.GetChunks(), key, entry, "")

 		if err != nil {
 			// don't swallow: committing mismatched bytes would propagate corruption
@@ -439,7 +439,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 	case updateRepair:
 		glog.Warningf("repair %s: destination diverged from source (dst %d chunks, src %d chunks); re-replicating full source content",
 			key, len(existingEntry.GetChunks()), len(newEntry.GetChunks()))
-		replicatedChunks, err := fs.replicateChunks(context.Background(), newEntry.GetChunks(), targetKey, getEntryMtimeNs(newEntry), "")
+		replicatedChunks, err := fs.replicateChunks(context.Background(), newEntry.GetChunks(), targetKey, newEntry, "")
 		if err != nil {
 			if errors.Is(err, errChunkSizeMismatch) {
 				return true, fs.onCorruptChunk(targetKey, newEntry, err)
@@ -467,7 +467,7 @@ func (fs *FilerSink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParent
 		}

 		// replicate the chunks that are new in the source
-		replicatedChunks, err := fs.replicateChunks(context.Background(), newChunks, targetKey, getEntryMtimeNs(newEntry), "")
+		replicatedChunks, err := fs.replicateChunks(context.Background(), newChunks, targetKey, newEntry, "")
 		if err != nil {
 			if errors.Is(err, errChunkSizeMismatch) {
 				return true, fs.onCorruptChunk(targetKey, newEntry, err)
@@ -108,7 +108,7 @@ func TestGetEntryMtimeNs(t *testing.T) {
 // onCorruptChunk must surface the error, not skip, when it can't confirm
 // supersession (no source filer here) — otherwise it could drop a live file.
 func TestOnCorruptChunkRefusesWhenSupersessionUnconfirmed(t *testing.T) {
-	fs := &FilerSink{} // filerSource nil => hasSourceNewerVersion is always false
+	fs := &FilerSink{} // filerSource nil => sourceMovedOn is always false
 	sentinel := errors.New("corrupt chunk: read 0 bytes, source says 107")

 	got := fs.onCorruptChunk("/buckets/x/config",
@@ -123,14 +123,14 @@ func TestOnCorruptChunkRefusesWhenSupersessionUnconfirmed(t *testing.T) {
 // onReplicateChunkError must NOT swallow a transient/partial replicate failure
 // (a volume-id lookup race or unretried manifest-chunk resolve during a write
 // burst) when it cannot confirm the source has moved on. Here filerSource is
-// nil, so hasSourceNewerVersion is always false — the "supersession unconfirmed"
+// nil, so sourceMovedOn is always false — the "supersession unconfirmed"
 // case. Returning nil would let the metadata subscription advance its offset
 // past an event whose file was never written, silently dropping it (the
 // filer.backup partial-landing bug). It must propagate so the offset stays put
-// and the event is reprocessed. Genuine source-404 (gone/superseded) skipping is
-// covered by TestSourceSupersedes, which gates the return-nil branch.
+// and the event is reprocessed. The gone or superseded skip is covered by
+// TestOnReplicateChunkErrorSkipsSupersededVersion.
 func TestOnReplicateChunkErrorPropagatesWhenSupersessionUnconfirmed(t *testing.T) {
-	fs := &FilerSink{} // filerSource nil => hasSourceNewerVersion always false
+	fs := &FilerSink{} // filerSource nil => sourceMovedOn always false
 	sentinel := errors.New("replicate manifest data chunks 7,02def: volume id 7 not found")

 	got := fs.onReplicateChunkError("/buckets/x/model.pt",