From 39ef551bcfed8db8ae6476649e8ca42ba824fcc5 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 6 Nov 2026 01:43:44 -0800 Subject: [PATCH] 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 --- weed/command/filer_remote_gateway.go | 1 + weed/command/filer_remote_gateway_buckets.go | 4 ++-- weed/command/filer_remote_sync.go | 1 + weed/command/filer_remote_sync_dir.go | 20 ++++++++++++++------ weed/command/filer_remote_sync_dir_test.go | 13 +++++++++++++ 5 files changed, 31 insertions(+), 8 deletions(-) diff --git a/weed/command/filer_remote_gateway.go b/weed/command/filer_remote_gateway.go index 4f03b7927f..040e5b0555 100644 --- a/weed/command/filer_remote_gateway.go +++ b/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{}) diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go index 1a03a1c150..7a9453acfb 100644 --- a/weed/command/filer_remote_gateway_buckets.go +++ b/weed/command/filer_remote_gateway_buckets.go @@ -33,7 +33,7 @@ func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSo return err } - lastOffsetTs, err := 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 } @@ -74,7 +74,7 @@ func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSo } err = pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset) - saveResumeOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, processor) + option.resumeTsNs = saveResumeOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, processor) return err } diff --git a/weed/command/filer_remote_sync.go b/weed/command/filer_remote_sync.go index 9e96f6d069..60e683454d 100644 --- a/weed/command/filer_remote_sync.go +++ b/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{}) diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 5e3cd8cd5c..e7629157b2 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -62,7 +62,7 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour return err } - lastOffsetTs, err := 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 } @@ -109,22 +109,24 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour } err = pb.FollowMetadata(pb.ServerAddress(*option.filerAddress), option.grpcDialOption, metadataFollowOption, processEventFnWithOffset) - saveResumeOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, processor) + 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. -func saveResumeOffset(grpcDialOption grpc.DialOption, filerAddress pb.ServerAddress, dir string, processor *MetadataProcessor) { +// 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 + 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) { @@ -1104,7 +1106,13 @@ 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, error) { +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() diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 0c2d65c90e..a89f056aec 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -725,6 +725,19 @@ func TestSaveResumeOffsetKeepsTheStartForTheRetry(t *testing.T) { 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) {