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>

weed/command/filer_remote_gateway_buckets.go

@@ -73,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)
+	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_sync_dir.go

@@ -108,7 +108,23 @@ 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)
+	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.
+func saveResumeOffset(grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, dir string, processor *MetadataProcessor) {
+	offsetTsNs := resumeTsNsFor(processor)
+	if offsetTsNs <= 0 {
+		return
+	}
+	if err := remote_storage.SetSyncOffset(grpcDialOption, filerAddress, dir, offsetTsNs); err != nil {
+		glog.Errorf("save the sync offset of %s: %v", dir, err)
+	}
 }
 
 func (option *RemoteSyncOptions) makeEventProcessor(remoteStorage *remote_pb.RemoteConf, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, filerSource *source.FilerSource) (pb.ProcessMetadataFunc, error) {

weed/command/filer_remote_sync_dir_test.go

@@ -695,6 +695,27 @@ func TestResolveLastSyncOffsetReturnsTheErrorToTheRetryLoop(t *testing.T) {
 	}
 }
 
+// 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)
+	}
+}
+
 func TestRemoteUploadReaderSharesOneChunkCache(t *testing.T) {
 	entry := entryWith("blob.bin", nil, chunk("3,01", "e1"))
 	for i := 0; i < 3; i++ {