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 }