From 7be919ded16f7edca25a589fd77774c96b4be19b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 6 Nov 2026 01:25:33 -0800 Subject: [PATCH] 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 --- weed/command/filer_remote_gateway_buckets.go | 4 +++- weed/command/filer_remote_sync_dir.go | 18 ++++++++++++++++- weed/command/filer_remote_sync_dir_test.go | 21 ++++++++++++++++++++ 3 files changed, 41 insertions(+), 2 deletions(-) diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go index f35f6bfd39..1a03a1c150 100644 --- a/weed/command/filer_remote_gateway_buckets.go +++ b/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 diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 49f21df0a6..99986e94e5 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/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) { diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 1ced1240cd..1275f44c78 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/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++ {