amadan

carlos / platform Public

Merge pull request #199 from carlosframework/worktree-184-wake-singleflight

activator: failed wakes are singleflight — the initiator waits on its own entry (#184)
Paul Campbell d7077d173d539bf5bcb8b4fceed54de0bb8e7435 parent f5c5d07parent dbc3e52
2 files changed, +63 −2
  • internal/activator/activator.go +8 −2
  • internal/activator/activator_test.go +55 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index f44b976fc..2ec94c275 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -444,11 +444,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 86d696090..62bad5c92 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -808,6 +808,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) {