amadan
host: the request-store cancellation regression waits on a signal, bounded (Task 8 round 2)
Astra's re-review: the polling loop read mwBucket.lists while the request's goroutine wrote it, with no synchronization and no timeout. The bucket now closes a started channel once the blocking request reaches it; the test waits in a bounded select that also fails on an early end, and reads the counter only after the request's result is received.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 file changed,
+22
−10
internal/host/moveswakesource_test.go+22 −10
diff --git a/internal/host/moveswakesource_test.go b/internal/host/moveswakesource_test.go| index 364470790..9d4411ebe 100644 |
| --- a/internal/host/moveswakesource_test.go |
| +++ b/internal/host/moveswakesource_test.go |
| @@ -380,20 +380,24 @@ func (s *mwDataSource) Open(_ context.Context, _ release.Store, data lease.Versi |
| return ClaimMovesOwners{}, nil, nil |
| } |
| -// mwBucket is the claim's data bucket, recording each listing's context |
| -// and, when block is set, holding the listing until its context ends. |
| +// mwBucket is the claim's data bucket, recording each listing's context. |
| +// When started is set, a listing closes it once it has reached the bucket |
| +// and then holds until its context ends. The test sets started before the |
| +// listing's goroutine starts, and reads the counters only after that |
| +// goroutine's result has been received, so neither side races. |
| type mwBucket struct { |
| *leasetest.VersionedBucket |
| lists int |
| deadlines []time.Time |
| - block bool |
| + started chan struct{} |
| } |
| func (b *mwBucket) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) { |
| b.lists++ |
| d, _ := ctx.Deadline() |
| b.deadlines = append(b.deadlines, d) |
| - if b.block { |
| + if b.started != nil { |
| + close(b.started) |
| <-ctx.Done() |
| return nil, ctx.Err() |
| } |
| @@ -436,15 +440,20 @@ func TestMovesOwnersReadTheBucketThroughTheRequestStore(t *testing.T) { |
| t.Fatalf("a refused identity's read: %v", err) |
| } |
| refuse = nil |
| - inner.block = true |
| + // A request in flight: wait (bounded) until it has reached the |
| + // bucket, or has ended early, then cancel the operation. |
| + started := make(chan struct{}) |
| + inner.started = started |
| ctx, cancel := context.WithCancel(t.Context()) |
| + defer cancel() |
| done := make(chan error, 1) |
| go func() { _, err := src.data.ListVersions(ctx, "data", "instances/"); done <- err }() |
| - for { |
| - if l := func() int { return inner.lists }(); l == 2 { |
| - break |
| - } |
| - time.Sleep(time.Millisecond) |
| + select { |
| + case <-started: |
| + case err := <-done: |
| + t.Fatalf("the request ended before it reached the bucket: %v", err) |
| + case <-time.After(10 * time.Second): |
| + t.Fatal("the request never reached the bucket") |
| } |
| cancel() |
| select { |
| @@ -455,6 +464,9 @@ func TestMovesOwnersReadTheBucketThroughTheRequestStore(t *testing.T) { |
| case <-time.After(10 * time.Second): |
| t.Fatal("the operation's cancellation did not reach the request") |
| } |
| + if inner.lists != 2 { |
| + t.Fatalf("%d bucket calls, want the first and the cancelled one", inner.lists) |
| + } |
| } |
| // TestMovesSourceOpenRefusalIsLoggedOnce (Task 8 fix round 1, |