mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-07 04:27:49 +01:00
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 <noreply@anthropic.com>
This commit is contained in:
1 parent
51f603b5ec
commit
c7338fb68f
2 files changed
+14
-2
No files matched your search
@@ -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),
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user