From 2bcb748a67917fc37e3471a1f65975121e5c1bf8 Mon Sep 17 00:00:00 2001 From: james Date: Thu, 5 Nov 2026 20:03:42 -0700 Subject: [PATCH] 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) --- weed/command/filer_remote_sync_dir.go | 38 +++++++++++++----- weed/command/filer_remote_sync_dir_test.go | 46 ++++++++++++++++++++++ 2 files changed, 75 insertions(+), 9 deletions(-) diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index ef975eb3d1..989b9a36a5 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -1041,6 +1041,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 @@ -1055,15 +1059,25 @@ func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource fil } 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 := supersessionLookup(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() @@ -1071,10 +1085,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) @@ -1126,7 +1140,13 @@ func resolveLastSyncOffset(mountedDir string, timeAgo time.Duration, lookupMount } } else { lastOffsetTs = time.Now().Add(-timeAgo) - if lastOffsetTsNs, err := readOffset(); err == nil && lastOffsetTsNs > 0 { + 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 + glog.Fatalf("read the sync offset of %s: %v", mountedDir, err) + } + if lastOffsetTsNs > 0 { if savedOffsetTs := time.Unix(0, lastOffsetTsNs); savedOffsetTs.Before(lastOffsetTs) { lastOffsetTs = savedOffsetTs } diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index cd05b168c5..776db25b6b 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -591,6 +591,52 @@ func TestFilerLookupsOnTheSyncPathCarryADeadline(t *testing.T) { } } +// 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, 0, time.UTC) lookupFails := func() (*filer_pb.Entry, error) { return nil, context.DeadlineExceeded }