From e4c0acfa61561d8be99ff9f38b7bca694b15fe49 Mon Sep 17 00:00:00 2001 From: james Date: Thu, 5 Nov 2026 19:53:49 -0700 Subject: [PATCH] filer.remote.sync: no upload on an unanswered supersession lookup; bounded offset read retriedWriteFile returns the supersession lookup's error instead of treating an unanswered lookup as "not superseded": in a replay an older snapshot could otherwise overwrite a newer remote object. The processor retries the event. isSuperseded keeps its boolean form for the other call sites. GetSyncOffset's KvGet carries a 30 s deadline so a startup that falls back to the saved offset cannot hang on it. TestGatewayUncachedRewriteIsStillUploaded asserted the upload on a failed lookup; it now asserts the withheld upload and the lookup error. (cherry picked from commit 8fc2e718f7cd8322615b4e40007be903ade2dbfa) --- .../filer_remote_gateway_buckets_test.go | 26 +++++++++---------- weed/command/filer_remote_sync_dir.go | 20 +++++++++++--- weed/command/filer_remote_sync_dir_test.go | 7 +++-- weed/remote_storage/track_sync_offset.go | 6 ++++- 4 files changed, 40 insertions(+), 19 deletions(-) diff --git a/weed/command/filer_remote_gateway_buckets_test.go b/weed/command/filer_remote_gateway_buckets_test.go index 253b765189..0d2a9136ba 100644 --- a/weed/command/filer_remote_gateway_buckets_test.go +++ b/weed/command/filer_remote_gateway_buckets_test.go @@ -1,7 +1,7 @@ package command import ( - "slices" + "strings" "testing" "github.com/seaweedfs/seaweedfs/weed/filer" @@ -63,11 +63,13 @@ func TestGatewayDroppedLocalCopyIsNotUploaded(t *testing.T) { } } -// TestGatewayUncachedRewriteIsStillUploaded: a rewrite of an uncached entry -// holds local data, so the dropped-copy skip does not apply and the rewrite -// uploads. The stamp cannot reach a filer in the test, so the call errors; -// the upload is what is asserted. -func TestGatewayUncachedRewriteIsStillUploaded(t *testing.T) { +// TestGatewayUncachedRewriteWaitsForTheSupersessionLookup: a rewrite of an +// uncached entry holds local data, so the dropped-copy skip does not apply +// and the rewrite is an upload. The lookup that guards the upload cannot +// reach a filer in the test, so the event fails with that error and nothing +// is uploaded: an unanswered lookup never lets a snapshot overwrite the +// remote object, and the processor retries the event. +func TestGatewayUncachedRewriteWaitsForTheSupersessionLookup(t *testing.T) { option, remote := gatewayTestOption("gatewaytest-rewrite") t.Cleanup(func() { delete(remote_storage.RemoteStorageClientMakers, "gatewaytest-rewrite") }) process, err := option.makeBucketedEventProcessor(&source.FilerSource{}) @@ -77,14 +79,12 @@ func TestGatewayUncachedRewriteIsStillUploaded(t *testing.T) { rewrite := entryWith(droppedCopyName, droppedCopyStamp(), chunk("4,02", "e2")) resp := droppedCopyUpdate(droppedCopyEntry(droppedCopyStamp()), rewrite) - if err := process(resp); err != nil { - t.Logf("stamp against the test filer fails as expected: %v", err) + err = process(resp) + if err == nil || !strings.Contains(err.Error(), "supersession lookup") { + t.Fatalf("process = %v, want the supersession lookup error", err) } - if len(remote.writes) != 1 { - t.Fatalf("writes = %+v, want the rewrite uploaded once", remote.writes) - } - if !slices.Equal(remote.writtenChunks[0], []string{"4,02"}) { - t.Errorf("uploaded chunks %v, want the rewrite's [4,02]", remote.writtenChunks[0]) + if len(remote.writes) != 0 { + t.Fatalf("writes = %+v, want none until the lookup answers", remote.writes) } } diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 952ad5690e..ef975eb3d1 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -973,15 +973,27 @@ func readRemoteStorageConf(filerClient filer_pb.FilerClient, storageName string) // object or the later write uploads the current content. A lookup that fails // for any other reason keeps the write failure, so the event is retried. func isSuperseded(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry) bool { + superseded, _ := supersessionLookup(filerClient, dir, entry) + return superseded +} + +// supersessionLookup is isSuperseded with the lookup failure surfaced: an +// upload must not start on a lookup that did not answer, or an older +// snapshot in a replay overwrites a newer remote object. +func supersessionLookup(filerClient filer_pb.FilerClient, dir string, entry *filer_pb.Entry) (bool, error) { ctx, cancel := lookupContext() defer cancel() current, _, _, err := filer_pb.GetEntry(ctx, filerClient, util.NewFullPath(dir, entry.Name)) if errors.Is(err, filer_pb.ErrNotFound) { - return true + return true, nil } if err != nil { - return false + return false, err } + return entryDiffers(entry, current), nil +} + +func entryDiffers(entry, current *filer_pb.Entry) bool { if !filer.HasData(entry) && filer.HasData(current) { // The event described an entry without data (remote-only, or empty); // the filer has written to it since. Uploading the snapshot would put @@ -1038,7 +1050,9 @@ var errSuperseded = errors.New("superseded or incomplete since the event was log // still holds is worth sending. After: a dead chunk reads as a transient // "RequestError" from the SDK, and waiting out the backoff on it buys nothing. func retriedWriteFile(client remote_storage.RemoteStorageClient, filerSource filer_pb.FilerClient, dir string, newEntry *filer_pb.Entry, dest *remote_pb.RemoteStorageLocation) (remoteEntry *filer_pb.RemoteEntry, err error) { - if isSuperseded(filerSource, dir, newEntry) { + if superseded, lookupErr := supersessionLookup(filerSource, dir, newEntry); lookupErr != nil { + return nil, fmt.Errorf("%s: supersession lookup: %w", util.NewFullPath(dir, newEntry.Name), lookupErr) + } else if superseded { return nil, fmt.Errorf("%s %w", util.NewFullPath(dir, newEntry.Name), errSuperseded) } err = util.RetryOnError("writeFile", func(err error) bool { diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index ad041ebff4..cd05b168c5 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -580,8 +580,11 @@ func TestFilerLookupsOnTheSyncPathCarryADeadline(t *testing.T) { if current != nil || !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("currentEntry = %v, %v; want nil, deadline exceeded", current, err) } - if isSuperseded(&blockingFilerClient{}, dir, event) { - t.Error("isSuperseded = true on a lookup that timed out; want false (the write failure is kept and retried)") + if superseded, lookupErr := supersessionLookup(&blockingFilerClient{}, dir, event); superseded || !errors.Is(lookupErr, context.DeadlineExceeded) { + t.Errorf("supersessionLookup = %v, %v on a lookup that timed out; want false, deadline exceeded (no upload starts on an unanswered lookup)", superseded, lookupErr) + } + if _, writeErr := retriedWriteFile(nil, &blockingFilerClient{}, dir, event, &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "b", Path: "/lease.json"}); !errors.Is(writeErr, context.DeadlineExceeded) { + t.Errorf("retriedWriteFile = %v on an unanswered supersession lookup; want the lookup error, not an upload", writeErr) } if elapsed := time.Since(started); elapsed > 5*time.Second { t.Fatalf("two timed-out lookups took %v; the deadline did not bound them", elapsed) diff --git a/weed/remote_storage/track_sync_offset.go b/weed/remote_storage/track_sync_offset.go index 7fa77a468a..20faedbbd0 100644 --- a/weed/remote_storage/track_sync_offset.go +++ b/weed/remote_storage/track_sync_offset.go @@ -3,6 +3,7 @@ package remote_storage import ( "context" "errors" + "time" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -26,7 +27,10 @@ func SyncOffsetKey(dir string) []byte { func GetSyncOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, dir string) (lastOffsetTsNs int64, readErr error) { readErr = pb.WithFilerClient(false, 0, filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - resp, err := client.KvGet(context.Background(), &filer_pb.KvGetRequest{Key: SyncOffsetKey(dir)}) + // bounded: a read that never answers would hold the daemon's start + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + resp, err := client.KvGet(ctx, &filer_pb.KvGetRequest{Key: SyncOffsetKey(dir)}) if err != nil { return err }