From c7338fb68f57b6aebf5ed5c1f55468c96446128b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 6 Nov 2026 00:52:11 -0800 Subject: [PATCH] filer.remote.sync: bound the stamp written after each upload updateLocalEntry's UpdateEntry was the last filer call on the sync path without a deadline. A filer that never answers the stamp keeps the job and its path in the conflict index the same way an unanswered lookup did. The stamp carries IF_ENTRY_EQUAL, so a retried event after a timeout either stamps again or skips a stamp that already landed. Co-Authored-By: Claude Opus 5.5 --- weed/command/filer_remote_sync_dir.go | 4 +++- weed/command/filer_remote_sync_dir_test.go | 12 +++++++++++- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 55ab750e82..b5c97af92a 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -1243,7 +1243,9 @@ func updateLocalEntry(filerClient filer_pb.FilerClient, dir string, entry *filer entry.RemoteEntry = remoteEntry delete(entry.Extended, filer.ExtRemoteRenameSourceKey) err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - _, err := client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{ + ctx, cancel := lookupContext() + defer cancel() + _, err := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{ Directory: dir, Entry: entry, Condition: ifEntryEqual(expected), diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index da0320c785..ccbd25166b 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -549,6 +549,12 @@ func (c *blockingFilerClient) LookupDirectoryEntry(ctx context.Context, _ *filer return nil, ctx.Err() } +func (c *blockingFilerClient) UpdateEntry(ctx context.Context, _ *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + c.updates++ + <-ctx.Done() + return nil, ctx.Err() +} + func (c *blockingFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error { return fn(c) } @@ -570,8 +576,12 @@ func TestFilerLookupsOnTheSyncPathCarryADeadline(t *testing.T) { 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) } + stamp := &blockingFilerClient{} + if stampErr := updateLocalEntry(stamp, dir, event, &filer_pb.RemoteEntry{StorageName: "b2"}); stamp.updates != 1 || !errors.Is(stampErr, context.DeadlineExceeded) { + t.Errorf("updateLocalEntry = %v after %d updates on an unanswered stamp; want deadline exceeded after 1", stampErr, stamp.updates) + } if elapsed := time.Since(started); elapsed > 5*time.Second { - t.Fatalf("two timed-out lookups took %v; the deadline did not bound them", elapsed) + t.Fatalf("four timed-out filer calls took %v; the deadline did not bound them", elapsed) } }