filer.remote.sync: bounded filer lookups, one chunk cache per process (#12441)

* filer.remote.sync: bounded filer lookups, one chunk cache per process

The filer lookups on the sync path (currentEntry, isSuperseded,
uploadCurrentEntry, liveRemoteEntry, the mount directory read) ran with no
deadline. A lookup that never answers keeps its job alive, the job keeps
its path in the MetadataProcessor conflict index, AddSyncJob waits on the
condition and the daemon stops consuming events with no log line. Each
lookup now carries a 30 s deadline, so the job fails and the processor's
stop-and-resubscribe path runs.

remoteUploadReader built a chunk cache per upload; the cache's worker
goroutine is never stopped, so each upload leaked one goroutine and its
cache (21k goroutines, ~150 MB/h under lock-file churn, OOM at the memory
limit). One chunk cache per process.

(cherry picked from commit ff77fd222772d4f3b68f34d5ee39a66684f90e78)

* filer.remote.sync: a failed mount lookup at startup never resolves to now

collectLastSyncOffset returned time.Now() when the mounted directory lookup
failed, which skips every event since the saved offset. The saved offset is
used instead; without one the daemon exits so a restart retries.

(cherry picked from commit 8fd3124f5b3d3f8c3a291263a710047b4aff81ae)

* filer.remote.sync: no upload on an unanswered supersession lookup; bounded offset read

retriedWriteFile returns the supersession lookup's error instead of
treating an unanswered lookup as "not superseded": in a replay an older
snapshot could otherwise overwrite a newer remote object. The processor
retries the event. isSuperseded keeps its boolean form for the other call
sites. GetSyncOffset's KvGet carries a 30 s deadline so a startup that falls
back to the saved offset cannot hang on it.

TestGatewayUncachedRewriteIsStillUploaded asserted the upload on a failed
lookup; it now asserts the withheld upload and the lookup error.

(cherry picked from commit 8fc2e718f7cd8322615b4e40007be903ade2dbfa)

* filer.remote.sync test: keep the slices import the other gateway tests use

* filer.remote.sync: an unanswered lookup after a failed attempt ends the retry loop; an unread offset never resolves to now minus timeAgo

After a failed upload attempt the supersession lookup's error ended in
isSuperseded's boolean and the loop could upload the snapshot again
unverified. The lookup error now ends the loop (errLookupUnanswered, not
retried) and the processor retries the event. With -timeAgo set, a failed
read of the saved offset resolved to now minus timeAgo, skipping everything
older; the daemon exits instead so a restart retries.

(cherry picked from commit 55ca2a88e01d7b556d494a404b9cc263c27b2566)

* filer.remote.sync: an unresolved start offset returns to the retry loop instead of exiting

resolveLastSyncOffset returns the lookup or offset-read error; both callers
(directory and buckets) hand it to util.RetryUntil, which retries with
backoff instead of the process exiting.

* filer.remote.sync: keep doc comments on the functions they describe

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 952d80c4d3)

* filer.remote.sync: isSuperseded returns the lookup error itself

Its only production callers needed the error, so the boolean wrapper was
left to the tests. The tests now assert the error on a failed lookup.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 51f603b5ec)

* filer.remote.sync: bound the stamp written after each upload

updateLocalEntry's UpdateEntry was the last filer call on the sync path
without a deadline. A filer that never answers the stamp keeps the job and
its path in the conflict index the same way an unanswered lookup did. The
stamp carries IF_ENTRY_EQUAL, so a retried event after a timeout either
stamps again or skips a stamp that already landed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit c7338fb68f)

* filer.remote.sync: an unread offset never resolves to the mount's creation time

With the offset read now bounded, a timed-out read fell through to the
directory's creation time and replayed every event already synced. The
error goes back to the startup retry loop, as with -timeAgo; a missing
offset still reads as zero without an error.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 0a1b6c2944)

* filer.remote.sync: stamp a copy of the event entry

updateLocalEntry set the RemoteEntry and dropped the rename marker on the
caller's entry before the RPC. A transient stamp failure retries the same
event in place, which then read as already synced and skipped both the
upload and the stamp. The stub filer now applies a successful update the
way the filer does, since tests had relied on that shared object.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 1f332e5866)

* filer.remote.sync: a directory not created yet starts now, as before

A gateway started before the first bucket looked up a missing buckets
directory, had no saved offset, and returned an error to the retry loop
on every attempt until a bucket appeared. A missing directory with no
saved offset has nothing to replay, so it starts now as it did before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit c31497a9a3)

* filer.remote.sync: take the start time before the mount lookup

A bucket created between a not-found lookup and the end of the offset
read had its creation before a start taken afterwards, so the gateway
never created its remote bucket.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit f535fca8f6)

* filer.remote.sync: save the resume position when a subscription ends

A failed event ends the subscription with ErrResubscribe, and the retry
loop runs startup again, recomputing -timeAgo from a later now. The
periodic checkpoint only runs as events arrive, so a failure before the
first one could start the retry after the event it failed on. The resume
position is now saved whenever the subscription returns.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 7be919ded1)

* remote_storage: bound the sync offset write like the read

The offset write runs in the subscription's checkpoint callback and, now,
when a subscription ends; a filer that never answers it stalled either.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit f888ac6b8f)

* filer.remote.sync: say that any saved offset wins over the creation time

The creation time in seconds was compared with the offset in
milliseconds, which made any saved offset win. That is the behaviour to
keep: remote.unmount recreates the directory and then saves the offset,
usually in the same second, so a seconds comparison would start at the
creation second and replay the purge. The comparison now states it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit a0d60ee00f)

* filer.remote.sync: keep the resume position for the retry when its write fails

The retry in the same process now resumes from the position the ended
subscription returned, as filer.backup does, so a failed offset write
followed by an empty read cannot recompute -timeAgo past a failed event.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 39ef551bcf)

* filer.remote.sync test: give the same-second saved offset a fractional part

With whole seconds, returning the creation second resolved to the same
time and passed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
(cherry picked from commit 4bb74382a6)

---------

Co-authored-by: james <james@medable.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/command/filer_remote_gateway.go

@@ -31,6 +31,7 @@ type RemoteGatewayOptions struct {
 	bucketsDir  string
 	clientId    int32
 	clientEpoch int32
+	resumeTsNs  int64
 }
 
 var _ = filer_pb.FilerClient(&RemoteGatewayOptions{})

weed/command/filer_remote_gateway_buckets.go

@@ -33,7 +33,10 @@ func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSo
 		return err
 	}
 
-	lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, *option.timeAgo)
+	lastOffsetTs, err := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, *option.timeAgo, option.resumeTsNs)
+	if err != nil {
+		return err
+	}
 	processor := NewMetadataProcessor(eachEntryFunc, 128, lastOffsetTs.UnixNano())
 
 	var lastLogTsNs = time.Now().UnixNano()
@@ -70,7 +73,9 @@ func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSo
 		Resubscribe: processor.ResubscribeCh(),
 	}
 
-	return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset)
+	err = pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset)
+	option.resumeTsNs = saveResumeOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, processor)
+	return err
 }
 
 // makeBucketedEventProcessor is the gateway half of the sync protocol

weed/command/filer_remote_gateway_buckets_test.go

@@ -2,6 +2,7 @@ package command
 
 import (
 	"slices"
+	"strings"
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -63,11 +64,13 @@ func TestGatewayDroppedLocalCopyIsNotUploaded(t *testing.T) {
 	}
 }
 
-// TestGatewayUncachedRewriteIsStillUploaded: a rewrite of an uncached entry
-// holds local data, so the dropped-copy skip does not apply and the rewrite
-// uploads. The stamp cannot reach a filer in the test, so the call errors;
-// the upload is what is asserted.
-func TestGatewayUncachedRewriteIsStillUploaded(t *testing.T) {
+// TestGatewayUncachedRewriteWaitsForTheSupersessionLookup: a rewrite of an
+// uncached entry holds local data, so the dropped-copy skip does not apply
+// and the rewrite is an upload. The lookup that guards the upload cannot
+// reach a filer in the test, so the event fails with that error and nothing
+// is uploaded: an unanswered lookup never lets a snapshot overwrite the
+// remote object, and the processor retries the event.
+func TestGatewayUncachedRewriteWaitsForTheSupersessionLookup(t *testing.T) {
 	option, remote := gatewayTestOption("gatewaytest-rewrite")
 	t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-rewrite") })
 	process, err := option.makeBucketedEventProcessor(&source.FilerSource{})
@@ -77,14 +80,12 @@ func TestGatewayUncachedRewriteIsStillUploaded(t *testing.T) {
 
 	rewrite := entryWith(droppedCopyName, droppedCopyStamp(), chunk("4,02", "e2"))
 	resp := droppedCopyUpdate(droppedCopyEntry(droppedCopyStamp()), rewrite)
-	if err := process(resp); err != nil {
-		t.Logf("stamp against the test filer fails as expected: %v", err)
+	err = process(resp)
+	if err == nil || !strings.Contains(err.Error(), "supersession lookup") {
+		t.Fatalf("process = %v, want the supersession lookup error", err)
 	}
-	if len(remote.writes) != 1 {
-		t.Fatalf("writes = %+v, want the rewrite uploaded once", remote.writes)
-	}
-	if !slices.Equal(remote.writtenChunks[0], []string{"4,02"}) {
-		t.Errorf("uploaded chunks %v, want the rewrite's [4,02]", remote.writtenChunks[0])
+	if len(remote.writes) != 0 {
+		t.Fatalf("writes = %+v, want none until the lookup answers", remote.writes)
 	}
 }
 

weed/command/filer_remote_sync.go

@@ -22,6 +22,7 @@ type RemoteSyncOptions struct {
 	dir                *string
 	clientId           int32
 	clientEpoch        int32
+	resumeTsNs         int64
 }
 
 var _ = filer_pb.FilerClient(&RemoteSyncOptions{})

weed/command/filer_remote_sync_dir.go

@@ -62,7 +62,10 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour
 		return err
 	}
 
-	lastOffsetTs := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, *option.timeAgo)
+	lastOffsetTs, err := collectLastSyncOffset(option, option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, *option.timeAgo, option.resumeTsNs)
+	if err != nil {
+		return err
+	}
 	processor := NewMetadataProcessor(eachEntryFunc, 128, lastOffsetTs.UnixNano())
 
 	var lastLogTsNs = time.Now().UnixNano()
@@ -105,7 +108,25 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour
 		Resubscribe: processor.ResubscribeCh(),
 	}
 
-	return pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset)
+	err = pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset)
+	option.resumeTsNs = saveResumeOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, processor)
+	return err
+}
+
+// saveResumeOffset records where the next subscription resumes once this one
+// ends. The periodic checkpoint only runs as events arrive, and the retry
+// recomputes -timeAgo from now, so a failure before the first checkpoint
+// would otherwise start the retry after the event it failed on. The offset is
+// returned for the retry in this process, which must not depend on the write.
+func saveResumeOffset(grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, dir string, processor *MetadataProcessor) int64 {
+	offsetTsNs := resumeTsNsFor(processor)
+	if offsetTsNs <= 0 {
+		return 0
+	}
+	if err := remote_storage.SetSyncOffset(grpcDialOption, filerAddress, dir, offsetTsNs); err != nil {
+		glog.Errorf("save the sync offset of %s: %v", dir, err)
+	}
+	return offsetTsNs
 }
 
 func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) {
@@ -612,10 +633,21 @@ func releaseRenamedSource(filerSource filer_pb.FilerClient, client remote_storag
 	return releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry)
 }
 
+// remoteSyncLookupTimeout bounds a filer lookup on the sync path. A lookup
+// that never answers keeps its job alive, the job keeps its path in the
+// processor's conflict index, and the stream stops behind it.
+var remoteSyncLookupTimeout = 30 * time.Second
+
+func lookupContext() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), remoteSyncLookupTimeout)
+}
+
 // currentEntry returns what the filer holds at dir/name now, nil when the
 // entry is gone.
 func currentEntry(filerSource filer_pb.FilerClient, dir, name string) (*filer_pb.Entry, error) {
-	current, _, _, err := filer_pb.GetEntry(context.Background(), filerSource, util.NewFullPath(dir, name))
+	ctx, cancel := lookupContext()
+	defer cancel()
+	current, _, _, err := filer_pb.GetEntry(ctx, filerSource, util.NewFullPath(dir, name))
 	if errors.Is(err, filer_pb.ErrNotFound) {
 		return nil, nil
 	}
@@ -743,7 +775,9 @@ func openRemoteObject(client remote_storage.RemoteStorageClient, loc *remote_pb.
 // finishes a rename: its stamp can already describe the destination (a sync
 // plus remote.uncache in the meantime), and only then is it complete.
 func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, oldSource renameSource, dest *remote_pb.RemoteStorageLocation, storageClass string, sourceOf func(*filer_pb.Entry, renameSource) renameSource) error {
-	current, _, _, err := filer_pb.GetEntry(context.Background(), filerSource, util.NewFullPath(dir, name))
+	ctx, cancel := lookupContext()
+	defer cancel()
+	current, _, _, err := filer_pb.GetEntry(ctx, filerSource, util.NewFullPath(dir, name))
 	if errors.Is(err, filer_pb.ErrNotFound) {
 		return nil
 	}
@@ -958,28 +992,35 @@ func readRemoteStorageConf(filerClient filer_pb.FilerClient, storageName string)
 // after it in the log is never persisted. Skipping is safe: the event that
 // superseded this one follows in the log, and the delete removes the remote
 // object or the later write uploads the current content. A lookup that fails
-// for any other reason keeps the write failure, so the event is retried.
-func isSuperseded(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry) bool {
-	current, _, _, err := filer_pb.GetEntry(context.Background(), filerClient, util.NewFullPath(dir, entry.Name))
+// for any other reason is returned: no upload starts on a lookup that did not
+// answer, or an older snapshot in a replay overwrites a newer remote object.
+func isSuperseded(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry) (bool, error) {
+	ctx, cancel := lookupContext()
+	defer cancel()
+	current, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.NewFullPath(dir, entry.Name))
 	if errors.Is(err, filer_pb.ErrNotFound) {
-		return true
+		return true, nil
 	}
 	if err != nil {
-		return false
+		return false, err
 	}
 	if !filer.HasData(entry) && filer.HasData(current) {
 		// The event described an entry without data (remote-only, or empty);
 		// the filer has written to it since. Uploading the snapshot would put
 		// an empty object where the write belongs.
-		return true
+		return true, nil
 	}
 	if len(entry.Content) > 0 || len(current.Content) > 0 {
-		return !bytes.Equal(entry.Content, current.Content)
+		return !bytes.Equal(entry.Content, current.Content), nil
 	}
 	return len(filer.DoMinusChunks(entry.GetChunks(), current.GetChunks())) > 0 ||
-		len(filer.DoMinusChunks(current.GetChunks(), entry.GetChunks())) > 0
+		len(filer.DoMinusChunks(current.GetChunks(), entry.GetChunks())) > 0, nil
 }
 
+// remoteUploadChunkCache is one per process: a chunk cache runs a worker
+// goroutine that nothing stops, so a cache per upload leaks one per file.
+var remoteUploadChunkCache = chunk_cache.NewChunkCacheInMemory(8)
+
 // remoteUploadReader streams the entry's full declared size through
 // ChunkReadAt, which zero-fills any range no chunk view covers: a sparse
 // file, a mid-write snapshot flushed while its chunk uploads were still
@@ -999,7 +1040,7 @@ func remoteUploadReader(filerClient filer_pb.FilerClient, entry *filer_pb.Entry)
 		return nil, nil, err
 	}
 	readerAt := filer.NewChunkReaderAtFromClient(context.Background(),
-		filer.NewReaderCache(8, chunk_cache.NewChunkCacheInMemory(8), lookupFileIdFn, nil),
+		filer.NewReaderCache(8, remoteUploadChunkCache, lookupFileIdFn, nil),
 		chunkViews, size, filer.DefaultPrefetchCount)
 	return io.NewSectionReader(readerAt, 0, size), readerAt.Close, nil
 }
@@ -1010,6 +1051,10 @@ func remoteUploadReader(filerClient filer_pb.FilerClient, entry *filer_pb.Entry)
 // follows in the log and uploads them.
 var errSuperseded = errors.New("superseded or incomplete since the event was logged")
 
+// errLookupUnanswered marks an upload stopped because the filer did not
+// answer the supersession lookup; the processor retries the event.
+var errLookupUnanswered = errors.New("supersession lookup unanswered")
+
 // retriedWriteFile uploads the entry, retrying transient failures. The filer is
 // asked whether the entry is superseded before the first attempt and after
 // every failed one, and the upload stops at once when it is. Before: a backlog
@@ -1019,18 +1064,30 @@ var errSuperseded = errors.New("superseded or incomplete since the event was log
 // still holds is worth sending. After: a dead chunk reads as a transient
 // "RequestError" from the SDK, and waiting out the backoff on it buys nothing.
 func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource filer_pb.FilerClient, dir string, newEntry *filer_pb.Entry, dest *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) {
-	if isSuperseded(filerSource, dir, newEntry) {
+	if superseded, lookupErr := isSuperseded(filerSource, dir, newEntry); lookupErr != nil {
+		return nil, fmt.Errorf("%s: supersession lookup: %w", util.NewFullPath(dir, newEntry.Name), lookupErr)
+	} else if superseded {
 		return nil, fmt.Errorf("%s %w", util.NewFullPath(dir, newEntry.Name), errSuperseded)
 	}
+	// A failed attempt is retried only once the filer has answered that the
+	// entry is still the one the event described; an unanswered lookup
+	// ends the loop so no further attempt uploads an unverified snapshot.
+	afterFailure := func(attemptErr error) error {
+		superseded, lookupErr := isSuperseded(filerSource, dir, newEntry)
+		if lookupErr != nil {
+			return fmt.Errorf("%s %w: %w (after: %v)", util.NewFullPath(dir, newEntry.Name), errLookupUnanswered, lookupErr, attemptErr)
+		}
+		if superseded {
+			return fmt.Errorf("%s %w: %w", util.NewFullPath(dir, newEntry.Name), errSuperseded, attemptErr)
+		}
+		return attemptErr
+	}
 	err = util.RetryOnError("writeFile", func(err error) bool {
-		return !errors.Is(err, errSuperseded) && util.IsTransientError(err)
+		return !errors.Is(err, errSuperseded) && !errors.Is(err, errLookupUnanswered) && util.IsTransientError(err)
 	}, func() error {
 		reader, closeFn, readerErr := remoteUploadReader(filerSource, newEntry)
 		if readerErr != nil {
-			if isSuperseded(filerSource, dir, newEntry) {
-				return fmt.Errorf("%s %w: %w", util.NewFullPath(dir, newEntry.Name), errSuperseded, readerErr)
-			}
-			return readerErr
+			return afterFailure(readerErr)
 		}
 		if closeFn != nil {
 			defer closeFn()
@@ -1038,10 +1095,10 @@ func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource fil
 		glog.V(0).Infof("create %s", remote_storage.FormatLocation(dest))
 		var writeErr error
 		remoteEntry, writeErr = client.WriteFile(dest, newEntry, reader)
-		if writeErr != nil && isSuperseded(filerSource, dir, newEntry) {
-			return fmt.Errorf("%s %w: %w", util.NewFullPath(dir, newEntry.Name), errSuperseded, writeErr)
+		if writeErr != nil {
+			return afterFailure(writeErr)
 		}
-		return writeErr
+		return nil
 	})
 	if err != nil && !errors.Is(err, errSuperseded) {
 		glog.Errorf("write to %s: %v", dest, err)
@@ -1049,38 +1106,79 @@ func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource fil
 	return
 }
 
-func collectLastSyncOffset(filerClient filer_pb.FilerClient, grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, mountedDir string, timeAgo time.Duration) time.Time {
-	// 1. specified by timeAgo
-	// 2. last offset timestamp for this directory
-	// 3. directory creation time
+func collectLastSyncOffset(filerClient filer_pb.FilerClient, grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, mountedDir string, timeAgo time.Duration, resumeTsNs int64) (time.Time, error) {
+	if resumeTsNs > 0 {
+		// the previous attempt pinned where it stopped, which can sit behind
+		// both the saved offset and a recomputed -timeAgo window
+		glog.V(0).Infof("resuming previous attempt at %v", time.Unix(0, resumeTsNs))
+		return time.Unix(0, resumeTsNs), nil
+	}
+	return resolveLastSyncOffset(mountedDir, timeAgo,
+		func() (*filer_pb.Entry, error) {
+			ctx, cancel := lookupContext()
+			defer cancel()
+			entry, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.FullPath(mountedDir))
+			return entry, err
+		},
+		func() (int64, error) { return remote_storage.GetSyncOffset(grpcDialOption, filerAddress, mountedDir) })
+}
+
+// resolveLastSyncOffset picks the position to subscribe from:
+// 1. now minus timeAgo, or the saved offset when it is older
+// 2. the saved offset for this directory
+// 3. the directory's creation time
+// A mount lookup that fails never resolves to now: the saved offset still
+// orders the replay, and without one the error goes back to the caller's
+// retry loop instead of skipping every event since the failure. Only a
+// directory that does not exist yet, with no saved offset, starts now.
+func resolveLastSyncOffset(mountedDir string, timeAgo time.Duration, lookupMount func() (*filer_pb.Entry, error), readOffset func() (int64, error)) (time.Time, error) {
+	startedAt := time.Now()
 	var lastOffsetTs time.Time
 	if timeAgo == 0 {
-		mountedDirEntry, _, _, err := filer_pb.GetEntry(context.Background(), filerClient, util.FullPath(mountedDir))
+		mountedDirEntry, err := lookupMount()
 		if err != nil {
-			glog.V(0).Infof("get mounted directory %s: %v", mountedDir, err)
-			return time.Now()
+			lastOffsetTsNs, offsetErr := readOffset()
+			if offsetErr == nil && lastOffsetTsNs > 0 {
+				glog.V(0).Infof("get mounted directory %s: %v; resume from the saved offset %v", mountedDir, err, time.Unix(0, lastOffsetTsNs))
+				return time.Unix(0, lastOffsetTsNs), nil
+			}
+			if offsetErr == nil && errors.Is(err, filer_pb.ErrNotFound) {
+				return startedAt, nil
+			}
+			return time.Time{}, fmt.Errorf("get mounted directory %s: %w; no saved sync offset (%v)", mountedDir, err, offsetErr)
 		}
 
-		lastOffsetTsNs, err := remote_storage.GetSyncOffset(grpcDialOption, filerAddress, mountedDir)
+		lastOffsetTsNs, err := readOffset()
+		if err != nil {
+			return time.Time{}, fmt.Errorf("read the sync offset of %s: %w", mountedDir, err)
+		}
 		if mountedDirEntry != nil {
-			if err == nil && mountedDirEntry.Attributes.Crtime < lastOffsetTsNs/1000000 {
+			// a saved offset wins even over a newer creation time: remote.unmount
+			// saves one after recreating the directory, often in the same second
+			if lastOffsetTsNs > 0 {
 				lastOffsetTs = time.Unix(0, lastOffsetTsNs)
 				glog.V(0).Infof("resume from %v", lastOffsetTs)
 			} else {
 				lastOffsetTs = time.Unix(mountedDirEntry.Attributes.Crtime, 0)
 			}
 		} else {
-			lastOffsetTs = time.Now()
+			lastOffsetTs = startedAt
 		}
 	} else {
-		lastOffsetTs = time.Now().Add(-timeAgo)
-		if lastOffsetTsNs, err := remote_storage.GetSyncOffset(grpcDialOption, filerAddress, mountedDir); err == nil && lastOffsetTsNs > 0 {
+		lastOffsetTs = startedAt.Add(-timeAgo)
+		lastOffsetTsNs, err := readOffset()
+		if err != nil {
+			// an unread offset must not resolve to now minus timeAgo: the
+			// saved one may be older, and everything between would be skipped
+			return time.Time{}, fmt.Errorf("read the sync offset of %s: %w", mountedDir, err)
+		}
+		if lastOffsetTsNs > 0 {
 			if savedOffsetTs := time.Unix(0, lastOffsetTsNs); savedOffsetTs.Before(lastOffsetTs) {
 				lastOffsetTs = savedOffsetTs
 			}
 		}
 	}
-	return lastOffsetTs
+	return lastOffsetTs, nil
 }
 
 func toRemoteStorageLocation(mountDir, sourcePath util.FullPath, remoteMountLocation *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation {
@@ -1126,7 +1224,9 @@ func liveRemoteEntry(filerClient filer_pb.FilerClient, dir string, entry *filer_
 	if entry.RemoteEntry != nil {
 		return entry.RemoteEntry, nil
 	}
-	current, _, _, err := filer_pb.GetEntry(context.Background(), filerClient, util.NewFullPath(dir, entry.Name))
+	ctx, cancel := lookupContext()
+	defer cancel()
+	current, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.NewFullPath(dir, entry.Name))
 	if err != nil {
 		return nil, err
 	}
@@ -1174,10 +1274,13 @@ func updateLocalEntry(filerClient filer_pb.FilerClient, dir string, entry *filer
 	remoteEntry = proto.Clone(remoteEntry).(*filer_pb.RemoteEntry)
 	remoteEntry.LastLocalSyncTsNs = time.Now().UnixNano()
 	expected := proto.Clone(entry).(*filer_pb.Entry)
+	entry = proto.Clone(entry).(*filer_pb.Entry)
 	entry.RemoteEntry = remoteEntry
 	delete(entry.Extended, filer.ExtRemoteRenameSourceKey)
 	err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		_, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
+		ctx, cancel := lookupContext()
+		defer cancel()
+		_, err := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
 			Directory: dir,
 			Entry:     entry,
 			Condition: ifEntryEqual(expected),

weed/command/filer_remote_sync_dir_test.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"runtime"
 	"slices"
 	"strings"
 	"testing"
@@ -455,6 +456,11 @@ func (c *stubFilerClient) UpdateEntry(_ context.Context, req *filer_pb.UpdateEnt
 	if c.updateErr != nil {
 		return &filer_pb.UpdateEntryResponse{}, c.updateErr
 	}
+	if key := req.Directory + "/" + req.Entry.GetName(); c.entries[key] != nil {
+		c.entries[key] = req.Entry
+	} else if c.entry != nil && c.entry.Name == req.Entry.GetName() {
+		c.entry = req.Entry
+	}
 	return &filer_pb.UpdateEntryResponse{}, nil
 }
 
@@ -537,6 +543,230 @@ func entryWith(name string, remote *filer_pb.RemoteEntry, chunks ...*filer_pb.Fi
 	return &filer_pb.Entry{Name: name, Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}, RemoteEntry: remote, Chunks: chunks}
 }
 
+// blockingFilerClient answers a lookup only when its context ends.
+type blockingFilerClient struct {
+	stubFilerClient
+}
+
+func (c *blockingFilerClient) LookupDirectoryEntry(ctx context.Context, _ *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	c.lookups++
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func (c *blockingFilerClient) UpdateEntry(ctx context.Context, _ *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) {
+	c.updates++
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func (c *blockingFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+	return fn(c)
+}
+
+func TestFilerLookupsOnTheSyncPathCarryADeadline(t *testing.T) {
+	remoteSyncLookupTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { remoteSyncLookupTimeout = 30 * time.Second })
+	const dir = "/buckets/ingest"
+	event := entryWith("lease.json", nil, chunk("3,01", "e1"))
+
+	started := time.Now()
+	current, err := currentEntry(&blockingFilerClient{}, dir, event.Name)
+	if current != nil || !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("currentEntry = %v, %v; want nil, deadline exceeded", current, err)
+	}
+	if superseded, lookupErr := isSuperseded(&blockingFilerClient{}, dir, event); superseded || !errors.Is(lookupErr, context.DeadlineExceeded) {
+		t.Errorf("isSuperseded = %v, %v on a lookup that timed out; want false, deadline exceeded (no upload starts on an unanswered lookup)", superseded, lookupErr)
+	}
+	if _, writeErr := retriedWriteFile(nil, &blockingFilerClient{}, dir, event, &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "b", Path: "/lease.json"}); !errors.Is(writeErr, context.DeadlineExceeded) {
+		t.Errorf("retriedWriteFile = %v on an unanswered supersession lookup; want the lookup error, not an upload", writeErr)
+	}
+	stamp := &blockingFilerClient{}
+	if stampErr := updateLocalEntry(stamp, dir, event, &filer_pb.RemoteEntry{StorageName: "b2"}); stamp.updates != 1 || !errors.Is(stampErr, context.DeadlineExceeded) {
+		t.Errorf("updateLocalEntry = %v after %d updates on an unanswered stamp; want deadline exceeded after 1", stampErr, stamp.updates)
+	}
+	if elapsed := time.Since(started); elapsed > 5*time.Second {
+		t.Fatalf("four timed-out filer calls took %v; the deadline did not bound them", elapsed)
+	}
+}
+
+// answerThenBlockFilerClient answers the first lookup and blocks every later
+// one until its context ends.
+type answerThenBlockFilerClient struct {
+	stubFilerClient
+	entry *filer_pb.Entry
+}
+
+func (c *answerThenBlockFilerClient) LookupDirectoryEntry(ctx context.Context, _ *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) {
+	c.lookups++
+	if c.lookups == 1 {
+		return &filer_pb.LookupDirectoryEntryResponse{Entry: c.entry}, nil
+	}
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+func (c *answerThenBlockFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error {
+	return fn(c)
+}
+
+func TestRetriedWriteFileStopsWhenTheLookupAfterAFailureIsUnanswered(t *testing.T) {
+	remoteSyncLookupTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { remoteSyncLookupTimeout = 30 * time.Second })
+	const dir = "/buckets/ingest"
+	event := entryWith("lease.json", nil, chunk("3,01", "e1"))
+	dest := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "tier", Path: "/" + event.Name}
+	transient := errors.New("RequestError: send request failed\ncaused by: Put \"https://s3.example.com/tier/x\": 503 Service Unavailable")
+	remote := &failingRemote{err: transient}
+	filerClient := &answerThenBlockFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"))}
+
+	started := time.Now()
+	_, err := retriedWriteFile(remote, filerClient, dir, event, dest)
+	if !errors.Is(err, errLookupUnanswered) || !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("err = %v, want the unanswered-lookup error wrapping the deadline", err)
+	}
+	if remote.writes != 1 {
+		t.Errorf("wrote %d times, want 1: no further attempt until the filer answers", remote.writes)
+	}
+	if filerClient.lookups != 2 {
+		t.Errorf("looked up the filer %d times, want one before the attempt and one after it failed", filerClient.lookups)
+	}
+	if elapsed := time.Since(started); elapsed > 5*time.Second {
+		t.Fatalf("took %v; the retry loop did not stop on the unanswered lookup", elapsed)
+	}
+}
+
+func TestResolveLastSyncOffsetKeepsTheSavedOffsetWhenTheLookupFails(t *testing.T) {
+	saved := time.Date(2026, 11, 4, 17, 57, 10, 123_000_000, time.UTC)
+	lookupFails := func() (*filer_pb.Entry, error) { return nil, context.DeadlineExceeded }
+	savedOffset := func() (int64, error) { return saved.UnixNano(), nil }
+
+	got, err := resolveLastSyncOffset("/buckets/x", 0, lookupFails, savedOffset)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !got.Equal(saved) {
+		t.Fatalf("resolved %v, want the saved offset %v: a failed mount lookup must not start the replay at now", got, saved)
+	}
+
+	// a mount that answers resumes from the saved offset when it postdates the directory
+	entry := &filer_pb.Entry{Name: "x", Attributes: &filer_pb.FuseAttributes{Crtime: saved.Add(-time.Hour).Unix()}}
+	got, err = resolveLastSyncOffset("/buckets/x", 0, func() (*filer_pb.Entry, error) { return entry, nil }, savedOffset)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !got.Equal(saved) {
+		t.Fatalf("resolved %v, want %v", got, saved)
+	}
+
+	// remote.unmount recreates the directory, then saves the offset in the same
+	// second: starting at the creation second would replay the unmount's purge
+	unmounted := &filer_pb.Entry{Name: "x", Attributes: &filer_pb.FuseAttributes{Crtime: saved.Unix()}}
+	got, err = resolveLastSyncOffset("/buckets/x", 0, func() (*filer_pb.Entry, error) { return unmounted, nil }, savedOffset)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !got.Equal(saved) {
+		t.Fatalf("resolved %v, want the saved offset %v", got, saved)
+	}
+}
+
+func TestResolveLastSyncOffsetReturnsTheErrorToTheRetryLoop(t *testing.T) {
+	lookupFails := func() (*filer_pb.Entry, error) { return nil, context.DeadlineExceeded }
+	noOffset := func() (int64, error) { return 0, nil }
+	readFails := func() (int64, error) { return 0, errors.New("filer unavailable") }
+
+	// no saved offset to fall back on: the error goes back to the caller
+	if _, err := resolveLastSyncOffset("/buckets/x", 0, lookupFails, noOffset); !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("got %v, want the lookup error: without a saved offset the caller's retry loop decides, not an exit", err)
+	}
+
+	// no directory and no saved offset yet: a first run starts no later than
+	// the lookup, so a bucket created while the offset is read is replayed
+	started := time.Now()
+	var lookedUp time.Time
+	notFound := func() (*filer_pb.Entry, error) {
+		lookedUp = time.Now()
+		return nil, fmt.Errorf("lookup /buckets: %w", filer_pb.ErrNotFound)
+	}
+	slowNoOffset := func() (int64, error) { time.Sleep(10 * time.Millisecond); return 0, nil }
+	if got, err := resolveLastSyncOffset("/buckets", 0, notFound, slowNoOffset); err != nil || got.Before(started) || got.After(lookedUp) {
+		t.Fatalf("got %v, %v; want a start between %v and the lookup at %v", got, err, started, lookedUp)
+	}
+
+	// the mount answers but the saved offset is unreadable: never its creation time
+	mount := &filer_pb.Entry{Name: "x", Attributes: &filer_pb.FuseAttributes{Crtime: 1}}
+	if _, err := resolveLastSyncOffset("/buckets/x", 0, func() (*filer_pb.Entry, error) { return mount, nil }, readFails); err == nil || !strings.Contains(err.Error(), "filer unavailable") {
+		t.Fatalf("got %v, want the offset read error: the creation time would replay everything already synced", err)
+	}
+
+	// -timeAgo set, the saved offset unreadable: never now minus timeAgo
+	if _, err := resolveLastSyncOffset("/buckets/x", time.Hour, func() (*filer_pb.Entry, error) { return nil, nil }, readFails); err == nil || !strings.Contains(err.Error(), "filer unavailable") {
+		t.Fatalf("got %v, want the offset read error", err)
+	}
+}
+
+// TestSaveResumeOffsetKeepsTheStartForTheRetry: a subscription that ends
+// before its first checkpoint still records where it started, so the retry's
+// -timeAgo, recomputed from a later now, cannot start after it.
+func TestSaveResumeOffsetKeepsTheStartForTheRetry(t *testing.T) {
+	filerAddress, dialOption := startKvFiler(t)
+	const dir = "/buckets/x"
+	start := time.Now().Add(-time.Hour)
+	processor := NewMetadataProcessor(func(*filer_pb.SubscribeMetadataResponse) error { return nil }, 1, start.UnixNano())
+	saveResumeOffset(dialOption, filerAddress, dir, processor)
+
+	got, err := resolveLastSyncOffset(dir, time.Hour, func() (*filer_pb.Entry, error) { return nil, nil }, func() (int64, error) {
+		return remote_storage.GetSyncOffset(dialOption, filerAddress, dir)
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.After(start) {
+		t.Fatalf("the retry starts at %v, after the first subscription's start %v", got, start)
+	}
+
+	// the write fails and the retry reads no saved offset: the position
+	// returned for this process still holds the start
+	failingAddress, failingDialOption := startFiler(t, &failingKvFilerServer{})
+	resumeTsNs := saveResumeOffset(failingDialOption, failingAddress, dir, processor)
+	emptyAddress, emptyDialOption := startKvFiler(t)
+	got, err = collectLastSyncOffset(&stubFilerClient{}, emptyDialOption, emptyAddress, dir, time.Hour, resumeTsNs)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.After(start) {
+		t.Fatalf("after a failed write the retry starts at %v, after the first subscription's start %v", got, start)
+	}
+}
+
+func TestRemoteUploadReaderSharesOneChunkCache(t *testing.T) {
+	entry := entryWith("blob.bin", nil, chunk("3,01", "e1"))
+	for i := 0; i < 3; i++ {
+		if _, closeFn, err := remoteUploadReader(&stubFilerClient{}, entry); err != nil {
+			t.Fatal(err)
+		} else if closeFn != nil {
+			_ = closeFn()
+		}
+	}
+	runtime.GC()
+	before := runtime.NumGoroutine()
+	for i := 0; i < 200; i++ {
+		_, closeFn, err := remoteUploadReader(&stubFilerClient{}, entry)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if closeFn != nil {
+			_ = closeFn()
+		}
+	}
+	runtime.GC()
+	time.Sleep(100 * time.Millisecond)
+	if after := runtime.NumGoroutine(); after-before > 20 {
+		t.Fatalf("200 upload readers left %d goroutines behind (%d -> %d); one chunk cache per reader leaks its worker", after-before, before, after)
+	}
+}
+
 // TestIsSuperseded decides what a failed upload means for its event (#11148).
 // A replay from an earlier offset re-emits creates for entries the filer has
 // since deleted or rewritten; their chunks are gone, so the upload can never
@@ -571,12 +801,12 @@ func TestIsSuperseded(t *testing.T) {
 		{name: "still as described", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"), chunk("3,02", "e2"))}, want: false},
 		{name: "still as described, since replicated", filer: &stubFilerClient{entry: entryWith(event.Name, stamped, chunk("3,01", "e1"), chunk("3,02", "e2"))}, want: false},
 		{name: "appended since", filer: &stubFilerClient{entry: entryWith(event.Name, nil, chunk("3,01", "e1"), chunk("3,02", "e2"), chunk("3,03", "e3"))}, want: true},
-		{name: "filer lookup failed", filer: &stubFilerClient{err: errors.New("rpc error: code = Unavailable")}, want: false},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := isSuperseded(tt.filer, dir, event); got != tt.want {
-				t.Errorf("isSuperseded = %v, want %v", got, tt.want)
+			got, err := isSuperseded(tt.filer, dir, event)
+			if err != nil || got != tt.want {
+				t.Errorf("isSuperseded = %v, %v; want %v, nil", got, err, tt.want)
 			}
 			if tt.filer.lookups != 1 {
 				t.Errorf("looked up the filer %d times, want 1", tt.filer.lookups)
@@ -584,19 +814,35 @@ func TestIsSuperseded(t *testing.T) {
 		})
 	}
 
+	t.Run("filer lookup failed", func(t *testing.T) {
+		lookupErr := errors.New("rpc error: code = Unavailable")
+		if got, err := isSuperseded(&stubFilerClient{err: lookupErr}, dir, event); got || !errors.Is(err, lookupErr) {
+			t.Errorf("isSuperseded = %v, %v; want false and the lookup error", got, err)
+		}
+	})
+
+	superseded := func(t *testing.T, filer *stubFilerClient, entry *filer_pb.Entry) bool {
+		t.Helper()
+		got, err := isSuperseded(filer, dir, entry)
+		if err != nil {
+			t.Fatalf("isSuperseded: %v", err)
+		}
+		return got
+	}
+
 	t.Run("written since: the event described an entry without data", func(t *testing.T) {
 		remoteOnly := entryWith("video.mp4", &filer_pb.RemoteEntry{StorageName: "b2", RemoteSize: 20971520, RemoteMtime: 1786096669})
-		if !isSuperseded(&stubFilerClient{entry: entryWith("video.mp4", remoteOnly.RemoteEntry, chunk("3,09", "e9"))}, dir, remoteOnly) {
+		if !superseded(t, &stubFilerClient{entry: entryWith("video.mp4", remoteOnly.RemoteEntry, chunk("3,09", "e9"))}, remoteOnly) {
 			t.Error("isSuperseded = false, want true: the filer wrote local data to a remote-only entry")
 		}
-		if isSuperseded(&stubFilerClient{entry: entryWith("video.mp4", remoteOnly.RemoteEntry)}, dir, remoteOnly) {
+		if superseded(t, &stubFilerClient{entry: entryWith("video.mp4", remoteOnly.RemoteEntry)}, remoteOnly) {
 			t.Error("isSuperseded = true, want false: the entry is still remote-only")
 		}
 		empty := &filer_pb.Entry{Name: "touch.txt", Attributes: &filer_pb.FuseAttributes{Mtime: 1786096669}}
-		if !isSuperseded(&stubFilerClient{entry: &filer_pb.Entry{Name: "touch.txt", Content: []byte("now")}}, dir, empty) {
+		if !superseded(t, &stubFilerClient{entry: &filer_pb.Entry{Name: "touch.txt", Content: []byte("now")}}, empty) {
 			t.Error("isSuperseded = false, want true: the empty file was written to")
 		}
-		if isSuperseded(&stubFilerClient{entry: &filer_pb.Entry{Name: "touch.txt"}}, dir, empty) {
+		if superseded(t, &stubFilerClient{entry: &filer_pb.Entry{Name: "touch.txt"}}, empty) {
 			t.Error("isSuperseded = true, want false: the file is still empty")
 		}
 	})
@@ -604,20 +850,20 @@ func TestIsSuperseded(t *testing.T) {
 	t.Run("mid-write snapshot with a hole, filled in since", func(t *testing.T) {
 		snapshot := entryWith("IMG_6526.mov", nil, chunkAt("222,01", "e1", 0), chunkAt("220,03", "e3", 2048))
 		filled := entryWith("IMG_6526.mov", nil, chunkAt("222,01", "e1", 0), chunkAt("216,02", "e2", 1024), chunkAt("220,03", "e3", 2048))
-		if !isSuperseded(&stubFilerClient{entry: filled}, dir, snapshot) {
+		if !superseded(t, &stubFilerClient{entry: filled}, snapshot) {
 			t.Error("isSuperseded = false, want true: the filer holds the finished write, the snapshot can never be read past its hole")
 		}
-		if isSuperseded(&stubFilerClient{entry: filled}, dir, filled) {
+		if superseded(t, &stubFilerClient{entry: filled}, filled) {
 			t.Error("isSuperseded = true, want false for the event that describes the finished write")
 		}
 	})
 
 	t.Run("inline content rewritten since", func(t *testing.T) {
 		event := &filer_pb.Entry{Name: "note.txt", Content: []byte("v1")}
-		if !isSuperseded(&stubFilerClient{entry: &filer_pb.Entry{Name: "note.txt", Content: []byte("v2")}}, dir, event) {
+		if !superseded(t, &stubFilerClient{entry: &filer_pb.Entry{Name: "note.txt", Content: []byte("v2")}}, event) {
 			t.Error("isSuperseded = false, want true for an entry whose inline content has changed")
 		}
-		if isSuperseded(&stubFilerClient{entry: &filer_pb.Entry{Name: "note.txt", Content: []byte("v1")}}, dir, event) {
+		if superseded(t, &stubFilerClient{entry: &filer_pb.Entry{Name: "note.txt", Content: []byte("v1")}}, event) {
 			t.Error("isSuperseded = true, want false for an entry whose inline content is unchanged")
 		}
 	})
@@ -1366,6 +1612,28 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) {
 	})
 }
 
+// TestUpdateLocalEntryLeavesTheEventEntryAlone: a transient stamp failure
+// retries the same event in place, so the event must still read as unsynced
+// and keep its rename source, or the retry skips the upload and the stamp.
+func TestUpdateLocalEntryLeavesTheEventEntryAlone(t *testing.T) {
+	const dir = "/buckets/b/vol"
+	event := entryWith("state.db", nil, chunk("3,01", "e1"))
+	event.Extended = map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("src")}
+	filerClient := &stubFilerClient{updateErr: status.Error(codes.Unavailable, "filer is shutting down")}
+	if err := updateLocalEntry(filerClient, dir, event, &filer_pb.RemoteEntry{StorageName: "gcs", RemoteSize: 7}); err == nil {
+		t.Fatal("err = nil, want the transient failure")
+	}
+	if event.RemoteEntry != nil || len(event.Extended[filer.ExtRemoteRenameSourceKey]) == 0 {
+		t.Errorf("event entry = %v, want it untouched by the failed stamp", event)
+	}
+	if !shouldSendToRemote(event) {
+		t.Error("shouldSendToRemote = false after a failed stamp: the in-place retry would skip the upload")
+	}
+	if sent := filerClient.updated.GetEntry(); sent.GetRemoteEntry() == nil || len(sent.GetExtended()[filer.ExtRemoteRenameSourceKey]) > 0 {
+		t.Errorf("stamp request entry = %v, want the remote entry set and the rename source dropped", sent)
+	}
+}
+
 // TestUpdateLocalEntrySkipsDeletedEntry: the stamp that follows an upload lands
 // on an entry a later event already deleted. The filer answers "not found"; the
 // delete's own event follows in the log, so the stamp is skipped rather than
@@ -1651,7 +1919,7 @@ func TestCompleteRemoteOnlyRenameClearsThePendingSourceFirst(t *testing.T) {
 
 	t.Run("a marker still in place keeps the source", func(t *testing.T) {
 		// the stamp did not land: the filer still has the marker
-		filerClient := &stubFilerClient{entry: pending()}
+		filerClient := &stubFilerClient{entry: pending(), updateErr: status.Error(codes.FailedPrecondition, "entry changed")}
 		remote := &recordingRemote{objects: objects(), statETag: `"copy"`}
 		if err := completeRemoteOnlyRename(filerClient, remote, "/buckets/b/dst", pending(), renameSource{path: "/buckets/b/src/a.jpg", loc: oldDest}, dest, ""); err == nil {
 			t.Fatal("err = nil, want the event retried")

weed/remote_storage/track_sync_offset.go

@@ -3,6 +3,7 @@ package remote_storage
 import (
 	"context"
 	"errors"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
@@ -23,10 +24,16 @@ func SyncOffsetKey(dir string) []byte {
 	return syncKey
 }
 
+// syncOffsetTimeout bounds an offset read or write: one that never answers
+// would hold the daemon's start, or its event stream behind the checkpoint.
+const syncOffsetTimeout = 30 * time.Second
+
 func GetSyncOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, dir string) (lastOffsetTsNs int64, readErr error) {
 
 	readErr = pb.WithFilerClient(false, 0, filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
-		resp, err := client.KvGet(context.Background(), &filer_pb.KvGetRequest{Key: SyncOffsetKey(dir)})
+		ctx, cancel := context.WithTimeout(context.Background(), syncOffsetTimeout)
+		defer cancel()
+		resp, err := client.KvGet(ctx, &filer_pb.KvGetRequest{Key: SyncOffsetKey(dir)})
 		if err != nil {
 			return err
 		}
@@ -54,7 +61,9 @@ func SetSyncOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, dir s
 		valueBuf := make([]byte, 8)
 		util.Uint64toBytes(valueBuf, uint64(offsetTsNs))
 
-		resp, err := client.KvPut(context.Background(), &filer_pb.KvPutRequest{
+		ctx, cancel := context.WithTimeout(context.Background(), syncOffsetTimeout)
+		defer cancel()
+		resp, err := client.KvPut(ctx, &filer_pb.KvPutRequest{
 			Key:   SyncOffsetKey(dir),
 			Value: valueBuf,
 		})