amadan
activator: a failed wake's initiator must wait on its own entry (#184)
The wake initiator unlocked, spawned the wake, and RE-READ the entries map to wait. A wake that failed inside that one-preemption window had already deleted its entry: the initiator found nothing and started a second wake — two lease acquires and two S3 restore attempts per request, doubled error events, and the returned error from the second failure. Seen only as a loaded-suite flake (#184) because the window is that narrow. The initiator now falls through to the waiting arm holding the lock, waiting on the entry it created; the new stress test drives 300 failed wakes under lock churn and -race, and fails on the old shape in every run (310-317 acquires for 300 requests). Closes #184.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2 files changed,
+63
−2
internal/activator/activator.go+8 −2internal/activator/activator_test.go+55 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index ce345e4ae..fd4b36d0e 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -435,11 +435,17 @@ func (a *Service) beginRequest(ctx context.Context, rt registry.Route) (release |
| // coalesce on e.ready rather than racing a second lease claim |
| // (singleflight). The wake runs detached so even its initiator is |
| // held to the same bounded wait as everyone queued behind it. |
| + // |
| + // FALL THROUGH to the switch holding the lock — the initiator |
| + // waits on the entry IT created. The old shape unlocked and |
| + // re-read the map, and a wake that FAILED inside that one- |
| + // preemption window had already deleted its entry: the initiator |
| + // found nothing and started a second wake — two lease acquires |
| + // and two restore attempts per request (#184, seen only as a |
| + // loaded-suite test flake because the window is that narrow). |
| e = &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})} |
| a.entries[rt.Host] = e |
| - a.mu.Unlock() |
| go a.wake(e) |
| - continue |
| } |
| switch e.state { |
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go| index 1416eb235..cfdea20a7 100644 |
| --- a/internal/activator/activator_test.go |
| +++ b/internal/activator/activator_test.go |
| @@ -805,6 +805,61 @@ func TestActivatorWakeFailuresReleaseLeaseOnce(t *testing.T) { |
| } |
| } |
| +// The singleflight hole behind flake #184: the wake INITIATOR must wait on |
| +// the entry it created, never re-read the map — a fast-FAILING wake deletes |
| +// its entry, and an initiator that looks the host up again finds nothing and |
| +// starts a second wake: two lease acquires, two restore attempts, doubled |
| +// error events, for one request. The window is one preemption wide (between |
| +// the initiator's unlock and its next lock), so this drives many failed |
| +// wakes under -race and contending goroutines and asserts the invariant |
| +// acquires == requests exactly. |
| +func TestActivatorFailedWakeIsSingleflight(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Minute) |
| + b.restoreErr = errors.New("s3 down") |
| + rt := testRoute("singleflight.example.com") |
| + |
| + // Background lock churn widens the initiator's unlock→lock window the |
| + // way a loaded test suite does. |
| + stopChurn := make(chan struct{}) |
| + var churn sync.WaitGroup |
| + for i := 0; i < 4; i++ { |
| + churn.Add(1) |
| + go func() { |
| + defer churn.Done() |
| + for { |
| + select { |
| + case <-stopChurn: |
| + return |
| + default: |
| + a.mu.Lock() |
| + _ = len(a.entries) //nolint:staticcheck // hold briefly |
| + a.mu.Unlock() |
| + } |
| + } |
| + }() |
| + } |
| + |
| + const requests = 300 |
| + for i := 0; i < requests; i++ { |
| + if _, err := a.beginRequest(context.Background(), rt); err == nil { |
| + t.Fatal("wake succeeded despite a failing restore") |
| + } |
| + } |
| + close(stopChurn) |
| + churn.Wait() |
| + |
| + b.mu.Lock() |
| + acquires, releases := b.acquires, b.releases |
| + b.mu.Unlock() |
| + if acquires != requests || releases != requests { |
| + t.Fatalf("acquires=%d releases=%d for %d requests — a failed wake was not singleflight (#184)", |
| + acquires, releases, requests) |
| + } |
| + if a.entryCount() != 0 { |
| + t.Fatal("failed wakes left entries behind") |
| + } |
| +} |
| + |
| // A restore failure never reaches the seed: seeding a host whose data we |
| // couldn't materialize is precisely the "half-seeded instance" case. |
| func TestActivatorRestoreFailureSkipsSeed(t *testing.T) { |