amadan

carlos / platform Public

console: ending a policy holding cancels its attempts first; the stand-down stops acquisition (#328, #329)

Review fix round 1 for Task 8. An explicit release does not close the
lease's Lost, so each holding now owns its attempt contexts and cancels
them before its lease is released, on every path that ends it.
ReleasePolicyWriter sets a stopping state that refuses further
acquisition, cancels an acquisition in flight and waits, inside its 5 s
budget, for it to give back a lease it won anyway. The writer's log
lines go through an instance seam the tests capture.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 878dab6d43bdf7e82e0a70b7edb33635e3f37ebb parent 389c85d
3 files changed, +399 −86
  • internal/console/policylease.go +188 −62
  • internal/console/policylease_test.go +197 −17
  • internal/console/server.go +14 −7
diff --git a/internal/console/policylease.go b/internal/console/policylease.go
index cf8d7b33a..7b8a5eba0 100644
--- a/internal/console/policylease.go
+++ b/internal/console/policylease.go
@@ -37,10 +37,15 @@
// acquire before that expiry, so an attempt either reaches the store while
// this console still holds the lock or is cancelled. A request the network
// holds back past its own cancellation is the residual; a later holder's
-// conditional write then sees a changed ETag (Task 9).
+// conditional write then sees a changed ETag (Task 9). Ending a holding, for
+// any reason, cancels every attempt context issued under it before its lease
+// is released: an explicit release does not close the lease's Lost, so
+// without that an attempt could outlive the lock.
//
// Release is reachable: cmd/console calls ReleasePolicyWriter on SIGTERM, so
-// a rolled console's successor need not wait out the TTL. Where the console's
+// a rolled console's successor need not wait out the TTL. It stops the writer
+// for good, cancels an acquisition in flight and waits, inside its 5 s budget,
+// for that acquisition to give back a lease it won anyway. Where the console's
// role is denied s3:DeleteObject on the bucket (the flagship's store IAM Deny;
// plan 2026-10-21-instance-policy-1-foundation, "Boundary verification"), the
// release fails, is logged, and the successor waits up to lease.LeaseTTL, 15
@@ -112,9 +117,27 @@ type policyHolding struct {
lease PolicyLease
}
+// policyHold is the server's side of the current holding: the holding and
+// the cancel of every attempt context issued under it, so ending the holding
+// ends its attempts before its lease is released. Guarded by policyStateMu.
+type policyHold struct {
+ h policyHolding
+ attempts map[uint64]context.CancelFunc
+ nextID uint64
+}
+
+// policyAcquisition is an acquisition in flight: ReleasePolicyWriter cancels
+// it and waits for done, which closes only after a lease it won anyway has
+// been released.
+type policyAcquisition struct {
+ cancel context.CancelFunc
+ done chan struct{}
+}
+
// policyWriter returns the current holding, acquiring the lease if there is
// none or the last one was lost. The caller keeps the holding for its whole
// write and checks it with policyAttempt before every signing and publication.
+// After ReleasePolicyWriter it refuses for good: the console is stopping.
func (s *Server) policyWriter(ctx context.Context) (policyHolding, *doorRefusal) {
if s.cfg.Bucket == "" {
return policyHolding{n: 1, sole: true}, nil
@@ -122,72 +145,97 @@ func (s *Server) policyWriter(ctx context.Context) (policyHolding, *doorRefusal)
if s.cfg.PolicyLease == nil {
return policyHolding{}, &doorRefusal{http.StatusServiceUnavailable, policyLeaseMissingMsg}
}
+ unavailable := &doorRefusal{http.StatusServiceUnavailable, policyLeaseUnavailableMsg}
s.policyLeaseMu.Lock()
defer s.policyLeaseMu.Unlock()
- if cur := s.policyCur.Load(); cur != nil {
- err := cur.lease.Err()
- if err == nil {
- return *cur, nil
+
+ s.policyStateMu.Lock()
+ if s.policyStopping {
+ s.policyStateMu.Unlock()
+ return policyHolding{}, unavailable
+ }
+ var ended *policyHold
+ var why error
+ if cur := s.policyCur; cur != nil {
+ if why = cur.h.lease.Err(); why == nil {
+ h := cur.h
+ s.policyStateMu.Unlock()
+ return h, nil
}
- s.endPolicyHolding(cur, err)
+ ended = s.detachPolicyHoldLocked()
}
+ actx, cancel := context.WithCancel(ctx)
+ acq := &policyAcquisition{cancel: cancel, done: make(chan struct{})}
+ s.policyAcquiring = acq
+ s.policyStateMu.Unlock()
+ defer func() {
+ cancel()
+ close(acq.done)
+ }()
+ s.endPolicyHold(ended, why)
+
ttl := s.cfg.PolicyLeaseTTL
if ttl == 0 {
ttl = lease.LeaseTTL
}
- l, err := s.cfg.PolicyLease(ctx, s.cfg.Bucket, appPolicyLeasePrefix, ttl)
+ l, err := s.cfg.PolicyLease(actx, s.cfg.Bucket, appPolicyLeasePrefix, ttl)
if err == nil && l == nil {
err = errors.New("the leaser returned no lease")
}
- if err != nil {
- var held *lease.HeldError
- if errors.As(err, &held) {
- return policyHolding{}, &doorRefusal{http.StatusServiceUnavailable,
- fmt.Sprintf(policyLeaseHeldMsg, held.Owner, held.ExpiresAt.UTC().Format(time.RFC3339))}
- }
- log.Printf("console: policy writer lease: %v", err)
- return policyHolding{}, &doorRefusal{http.StatusServiceUnavailable, policyLeaseUnavailableMsg}
+
+ s.policyStateMu.Lock()
+ s.policyAcquiring = nil
+ stopping := s.policyStopping
+ if err == nil && !stopping {
+ s.policyN++
+ s.policyCur = &policyHold{h: policyHolding{n: s.policyN, lease: l}}
+ h := s.policyCur.h
+ s.policyStateMu.Unlock()
+ return h, nil
+ }
+ s.policyStateMu.Unlock()
+
+ if err == nil {
+ // Won while the console was stopping: give it straight back, before
+ // done closes, so the stand-down never leaves it behind.
+ s.policyLogf("console: policy writer lease: acquired while stopping; releasing it")
+ releasePolicyLease(context.Background(), l, s.policyLogf)
+ return policyHolding{}, unavailable
}
- s.policyN++
- h := &policyHolding{n: s.policyN, lease: l}
- s.policyCur.Store(h)
- return *h, nil
+ var held *lease.HeldError
+ if errors.As(err, &held) {
+ return policyHolding{}, &doorRefusal{http.StatusServiceUnavailable,
+ fmt.Sprintf(policyLeaseHeldMsg, held.Owner, held.ExpiresAt.UTC().Format(time.RFC3339))}
+ }
+ s.policyLogf("console: policy writer lease: %v", err)
+ return policyHolding{}, unavailable
}
// policyAttempt is the check before each signing and each publication
// attempt (Astra plan R3-1). It refuses unless h is still the current holding
// and its lease is healthy with at least policyAttemptWindow left before its
// safe deadline; otherwise it returns the context the attempt must run under,
-// which ends at the window, at the safe deadline, or when the lease is lost.
-// An attempt cut off by that context is this same refusal, not a store
-// failure. A sole holding is never refused and its context is ctx itself.
+// which ends at the window, at the safe deadline, when the lease is lost, or
+// when the holding ends (before its lease is released). An attempt cut off by
+// that context is this same refusal, not a store failure. A sole holding is
+// never refused and its context is ctx itself.
func (s *Server) policyAttempt(ctx context.Context, h policyHolding) (context.Context, context.CancelFunc, *doorRefusal) {
if h.sole {
return ctx, func() {}, nil
}
lost := &doorRefusal{http.StatusServiceUnavailable, policyLeaseLostMsg}
- cur := s.policyCur.Load()
- if cur == nil || cur.n != h.n || h.lease == nil {
- return ctx, func() {}, lost
- }
- if err := h.lease.Err(); err != nil {
- s.endPolicyHolding(cur, err)
- return ctx, func() {}, lost
- }
- select {
- case <-h.lease.Lost():
- s.endPolicyHolding(cur, errors.New("the lease reported lost"))
- return ctx, func() {}, lost
- default:
- }
- safe, err := h.lease.SafeUntil()
- if err != nil {
- s.endPolicyHolding(cur, err)
+ s.policyStateMu.Lock()
+ cur := s.policyCur
+ if cur == nil || cur.h.n != h.n || h.lease == nil {
+ s.policyStateMu.Unlock()
return ctx, func() {}, lost
}
now := time.Now()
- if safe.Sub(now) < policyAttemptWindow {
- s.endPolicyHolding(cur, fmt.Errorf("safe until %s, less than %s away", safe.UTC().Format(time.RFC3339), policyAttemptWindow))
+ safe, why := policyLeaseHealth(h.lease, now)
+ if why != nil {
+ ended := s.detachPolicyHoldLocked()
+ s.policyStateMu.Unlock()
+ s.endPolicyHold(ended, why)
return ctx, func() {}, lost
}
deadline := now.Add(policyAttemptWindow)
@@ -195,6 +243,13 @@ func (s *Server) policyAttempt(ctx context.Context, h policyHolding) (context.Co
deadline = safe
}
actx, cancel := context.WithDeadline(ctx, deadline)
+ id := cur.nextID
+ cur.nextID++
+ if cur.attempts == nil {
+ cur.attempts = map[uint64]context.CancelFunc{}
+ }
+ cur.attempts[id] = cancel
+ s.policyStateMu.Unlock()
go func() {
select {
case <-h.lease.Lost():
@@ -202,11 +257,37 @@ func (s *Server) policyAttempt(ctx context.Context, h policyHolding) (context.Co
case <-actx.Done():
}
}()
- return actx, cancel, nil
+ return actx, func() {
+ s.policyStateMu.Lock()
+ delete(cur.attempts, id)
+ s.policyStateMu.Unlock()
+ cancel()
+ }, nil
}
-// policyHeld reports, without blocking, whether this console holds the
-// writer lease and its renewal is healthy. A sole console always does.
+// policyLeaseHealth is policyAttempt's liveness check: renewal healthy, not
+// reported lost, and the safe deadline at least policyAttemptWindow away.
+func policyLeaseHealth(l PolicyLease, now time.Time) (time.Time, error) {
+ if err := l.Err(); err != nil {
+ return time.Time{}, err
+ }
+ select {
+ case <-l.Lost():
+ return time.Time{}, errors.New("the lease reported lost")
+ default:
+ }
+ safe, err := l.SafeUntil()
+ if err != nil {
+ return time.Time{}, err
+ }
+ if safe.Sub(now) < policyAttemptWindow {
+ return time.Time{}, fmt.Errorf("safe until %s, less than %s away", safe.UTC().Format(time.RFC3339), policyAttemptWindow)
+ }
+ return safe, nil
+}
+
+// policyHeld reports, without waiting on any I/O, whether this console holds
+// the writer lease and its renewal is healthy. A sole console always does.
func (s *Server) policyHeld() bool {
if s.cfg.Bucket == "" {
return true
@@ -214,36 +295,81 @@ func (s *Server) policyHeld() bool {
if s.cfg.PolicyLease == nil {
return false
}
- cur := s.policyCur.Load()
- return cur != nil && cur.lease.Err() == nil
+ s.policyStateMu.Lock()
+ cur := s.policyCur
+ s.policyStateMu.Unlock()
+ return cur != nil && cur.h.lease.Err() == nil
}
-// ReleasePolicyWriter ends the holding and releases its lease within
-// policyLeaseReleaseBudget, so a successor console need not wait out the TTL.
-// Nothing held is a no-op. It does not take policyLeaseMu, so a stand-down
-// never waits on an acquisition in flight.
+// ReleasePolicyWriter is the stand-down: within policyLeaseReleaseBudget it
+// stops all further acquisition, ends the holding (cancelling its attempts
+// first) and releases its lease, and cancels any acquisition in flight,
+// waiting for it to give back a lease it won anyway. A successor console then
+// need not wait out the TTL. It does not take policyLeaseMu, so it never
+// queues behind an acquisition. Nothing held is a no-op apart from the stop.
func (s *Server) ReleasePolicyWriter(ctx context.Context) {
- cur := s.policyCur.Swap(nil)
+ ctx, cancel := context.WithTimeout(ctx, policyLeaseReleaseBudget)
+ defer cancel()
+ s.policyStateMu.Lock()
+ s.policyStopping = true
+ cur := s.detachPolicyHoldLocked()
+ acq := s.policyAcquiring
+ s.policyStateMu.Unlock()
+ if acq != nil {
+ acq.cancel()
+ }
+ if cur != nil {
+ releasePolicyLease(ctx, cur.h.lease, s.policyLogf)
+ }
+ if acq != nil {
+ select {
+ case <-acq.done:
+ case <-ctx.Done():
+ s.policyLogf("console: policy writer lease: an acquisition in flight outlived the release budget")
+ }
+ }
+}
+
+// detachPolicyHoldLocked ends the current holding under policyStateMu:
+// nothing can bind to it from here on, and every attempt context issued
+// under it is cancelled now, before any release. The caller releases the
+// returned hold's lease outside the lock (endPolicyHold or ReleasePolicyWriter).
+func (s *Server) detachPolicyHoldLocked() *policyHold {
+ cur := s.policyCur
if cur == nil {
- return
+ return nil
+ }
+ s.policyCur = nil
+ for _, cancel := range cur.attempts {
+ cancel()
}
- releasePolicyLease(ctx, cur.lease)
+ cur.attempts = nil
+ return cur
}
-// endPolicyHolding ends cur if it is still the current holding, logging why
-// and releasing its lease best-effort. Only the caller that ends it releases.
-func (s *Server) endPolicyHolding(cur *policyHolding, why error) {
- if !s.policyCur.CompareAndSwap(cur, nil) {
+// endPolicyHold logs why a detached holding ended and releases its lease
+// best-effort. nil is a no-op.
+func (s *Server) endPolicyHold(ended *policyHold, why error) {
+ if ended == nil {
return
}
- log.Printf("console: policy writer lease lost: %v", why)
- releasePolicyLease(context.Background(), cur.lease)
+ s.policyLogf("console: policy writer lease lost: %v", why)
+ releasePolicyLease(context.Background(), ended.h.lease, s.policyLogf)
}
-func releasePolicyLease(ctx context.Context, l PolicyLease) {
+func releasePolicyLease(ctx context.Context, l PolicyLease, logf func(string, ...any)) {
ctx, cancel := context.WithTimeout(ctx, policyLeaseReleaseBudget)
defer cancel()
if err := l.Release(ctx); err != nil {
- log.Printf("console: policy writer lease release: %v", err)
+ logf("console: policy writer lease release: %v", err)
+ }
+}
+
+// policyLogf is log.Printf unless a test set policyLog.
+func (s *Server) policyLogf(format string, args ...any) {
+ if s.policyLog != nil {
+ s.policyLog(format, args...)
+ return
}
+ log.Printf(format, args...)
}
diff --git a/internal/console/policylease_test.go b/internal/console/policylease_test.go
index 563939912..4fb319d82 100644
--- a/internal/console/policylease_test.go
+++ b/internal/console/policylease_test.go
@@ -21,6 +21,8 @@ type fakePolicyLease struct {
safeErr error
lost chan struct{}
releases int
+ // onRelease, when set, runs at the start of Release.
+ onRelease func()
}
func newFakePolicyLease() *fakePolicyLease {
@@ -35,11 +37,21 @@ func (f *fakePolicyLease) Err() error {
func (f *fakePolicyLease) Release(context.Context) error {
f.mu.Lock()
- defer f.mu.Unlock()
+ hook := f.onRelease
f.releases++
+ f.mu.Unlock()
+ if hook != nil {
+ hook()
+ }
return nil
}
+func (f *fakePolicyLease) setOnRelease(hook func()) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.onRelease = hook
+}
+
func (f *fakePolicyLease) Lost() <-chan struct{} { return f.lost }
func (f *fakePolicyLease) SafeUntil() (time.Time, error) {
@@ -66,23 +78,33 @@ type policyLeaseCall struct {
}
// fakePolicyLeaser records its calls and hands out a fresh fake lease per
-// call, or fails with err when it is set.
+// call, or fails with err when it is set. With entered set, a call signals it
+// and then blocks until its context ends, and still returns a lease: the
+// worst case, an acquisition that won just as it was cancelled.
type fakePolicyLeaser struct {
- mu sync.Mutex
- calls []policyLeaseCall
- leases []*fakePolicyLease
- err error
+ mu sync.Mutex
+ calls []policyLeaseCall
+ leases []*fakePolicyLease
+ err error
+ entered chan struct{}
}
-func (f *fakePolicyLeaser) leaser(_ context.Context, bucket, prefix string, ttl time.Duration) (PolicyLease, error) {
+func (f *fakePolicyLeaser) leaser(ctx context.Context, bucket, prefix string, ttl time.Duration) (PolicyLease, error) {
f.mu.Lock()
- defer f.mu.Unlock()
f.calls = append(f.calls, policyLeaseCall{bucket, prefix, ttl})
- if f.err != nil {
- return nil, f.err
+ err, entered := f.err, f.entered
+ f.mu.Unlock()
+ if err != nil {
+ return nil, err
+ }
+ if entered != nil {
+ close(entered)
+ <-ctx.Done()
}
l := newFakePolicyLease()
+ f.mu.Lock()
f.leases = append(f.leases, l)
+ f.mu.Unlock()
return l, nil
}
@@ -98,8 +120,34 @@ func (f *fakePolicyLeaser) lease(i int) *fakePolicyLease {
return f.leases[i]
}
-func policyLeaseServer(f *fakePolicyLeaser) *Server {
- return &Server{cfg: Config{Bucket: "deploy", PolicyLease: f.leaser}}
+// policyLogRecorder captures the writer's log lines so the expected ones
+// are asserted on rather than printed.
+type policyLogRecorder struct {
+ mu sync.Mutex
+ lines []string
+}
+
+func (r *policyLogRecorder) logf(format string, args ...any) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ r.lines = append(r.lines, fmt.Sprintf(format, args...))
+}
+
+func (r *policyLogRecorder) has(t *testing.T, sub string) {
+ t.Helper()
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ for _, l := range r.lines {
+ if strings.Contains(l, sub) {
+ return
+ }
+ }
+ t.Errorf("no log line contains %q; got %q", sub, r.lines)
+}
+
+func policyLeaseServer(f *fakePolicyLeaser) (*Server, *policyLogRecorder) {
+ rec := &policyLogRecorder{}
+ return &Server{cfg: Config{Bucket: "deploy", PolicyLease: f.leaser}, policyLog: rec.logf}, rec
}
func wantPolicyRefusal(t *testing.T, ref *doorRefusal, msg string) {
@@ -144,7 +192,7 @@ func TestPolicyWriterRefusesWithoutALeaser(t *testing.T) {
func TestPolicyWriterAcquiresLazilyAndKeepsTheHolding(t *testing.T) {
t.Parallel()
f := &fakePolicyLeaser{}
- s := policyLeaseServer(f)
+ s, rec := policyLeaseServer(f)
if f.callCount() != 0 || s.policyHeld() {
t.Fatalf("a new server: %d calls, held %v", f.callCount(), s.policyHeld())
}
@@ -186,13 +234,14 @@ func TestPolicyWriterAcquiresLazilyAndKeepsTheHolding(t *testing.T) {
if !s.policyHeld() {
t.Fatal("policyHeld false on the fresh holding")
}
+ rec.has(t, "console: policy writer lease lost: renewal lost")
}
func TestPolicyWriterNamesTheHolder(t *testing.T) {
t.Parallel()
t1 := time.Date(2026, 10, 25, 14, 30, 0, 0, time.FixedZone("x", 3600))
f := &fakePolicyLeaser{err: fmt.Errorf("acquire lease: %w", &lease.HeldError{Owner: "console-b:4242", ExpiresAt: t1})}
- s := policyLeaseServer(f)
+ s, rec := policyLeaseServer(f)
_, ref := s.policyWriter(context.Background())
if ref == nil || ref.Status != http.StatusServiceUnavailable {
t.Fatalf("refusal %+v, want 503", ref)
@@ -212,13 +261,13 @@ func TestPolicyWriterNamesTheHolder(t *testing.T) {
f.mu.Unlock()
_, ref = s.policyWriter(context.Background())
wantPolicyRefusal(t, ref, policyLeaseUnavailableMsg)
+ rec.has(t, "console: policy writer lease: dial tcp: no route to host")
}
func TestReleasePolicyWriterEndsTheHolding(t *testing.T) {
t.Parallel()
f := &fakePolicyLeaser{}
- s := policyLeaseServer(f)
- s.ReleasePolicyWriter(context.Background()) // nothing held: a no-op
+ s, _ := policyLeaseServer(f)
if _, ref := s.policyWriter(context.Background()); ref != nil {
t.Fatalf("refused: %+v", ref)
}
@@ -233,6 +282,16 @@ func TestReleasePolicyWriterEndsTheHolding(t *testing.T) {
if got := f.lease(0).released(); got != 1 {
t.Fatalf("a second ReleasePolicyWriter released again: %d", got)
}
+
+ // Nothing held: nothing to release, and the console still stops.
+ idle := &fakePolicyLeaser{}
+ is, _ := policyLeaseServer(idle)
+ is.ReleasePolicyWriter(context.Background())
+ _, ref := is.policyWriter(context.Background())
+ wantPolicyRefusal(t, ref, policyLeaseUnavailableMsg)
+ if idle.callCount() != 0 {
+ t.Fatalf("a stopped console called the leaser %d times", idle.callCount())
+ }
}
// Astra plan R3-1. Fails if any check is dropped.
@@ -253,7 +312,7 @@ func TestPolicyAttemptBindsToTheSameLiveHolding(t *testing.T) {
cancel()
f := &fakePolicyLeaser{}
- s := policyLeaseServer(f)
+ s, rec := policyLeaseServer(f)
attempt := func(h policyHolding) (context.Context, *doorRefusal) {
t.Helper()
actx, cancel, ref := s.policyAttempt(bg, h)
@@ -356,4 +415,125 @@ func TestPolicyAttemptBindsToTheSameLiveHolding(t *testing.T) {
// And a lease already lost refuses the next attempt outright.
_, ref = attempt(h4)
wantPolicyRefusal(t, ref, policyLeaseLostMsg)
+ for _, why := range []string{"lost: safe until", "lost: lease spent", "lost: renewal lost", "lost: the lease reported lost"} {
+ rec.has(t, "console: policy writer lease "+why)
+ }
+}
+
+// Review fix round 1, item 1: ending a holding cancels every attempt issued
+// under it before its lease is released, on each path that ends one. The
+// real lease's Lost does not close on an explicit release, so without this an
+// attempt could still be writing after another console took the lock.
+func TestPolicyHoldingEndCancelsItsAttemptsFirst(t *testing.T) {
+ t.Parallel()
+ bg := context.Background()
+ f := &fakePolicyLeaser{}
+ s, _ := policyLeaseServer(f)
+
+ // wantCancelledAtRelease arms l so that its Release records whether each
+ // of the attempt contexts had already ended.
+ wantCancelledAtRelease := func(l *fakePolicyLease, ctxs ...context.Context) func() {
+ var mu sync.Mutex
+ var live []int
+ called := false
+ l.setOnRelease(func() {
+ mu.Lock()
+ defer mu.Unlock()
+ called = true
+ for i, c := range ctxs {
+ if c.Err() == nil {
+ live = append(live, i)
+ }
+ }
+ })
+ return func() {
+ t.Helper()
+ mu.Lock()
+ defer mu.Unlock()
+ if !called {
+ t.Fatal("the lease was not released")
+ }
+ if len(live) > 0 {
+ t.Fatalf("attempts %v were still live when the lease was released", live)
+ }
+ }
+ }
+ attempt := func(h policyHolding) context.Context {
+ t.Helper()
+ actx, cancel, ref := s.policyAttempt(bg, h)
+ if ref != nil {
+ t.Fatalf("refused: %+v", ref)
+ }
+ t.Cleanup(cancel)
+ return actx
+ }
+
+ // The safe window running short ends the holding from another attempt.
+ h1, _ := s.policyWriter(bg)
+ a, b := attempt(h1), attempt(h1)
+ check := wantCancelledAtRelease(f.lease(0), a, b)
+ f.lease(0).set(nil, time.Now().Add(8*time.Second), nil)
+ if _, _, ref := s.policyAttempt(bg, h1); ref == nil {
+ t.Fatal("a short safe window was accepted")
+ }
+ check()
+
+ // A lost renewal ends it from policyWriter.
+ h2, _ := s.policyWriter(bg)
+ c := attempt(h2)
+ check = wantCancelledAtRelease(f.lease(1), c)
+ f.lease(1).set(errors.New("renewal lost"), time.Now().Add(10*time.Minute), nil)
+ if h3, ref := s.policyWriter(bg); ref != nil || h3.n != 3 {
+ t.Fatalf("holding %+v, refusal %+v", h3, ref)
+ }
+ check()
+
+ // The stand-down ends it from ReleasePolicyWriter.
+ h3, _ := s.policyWriter(bg)
+ d := attempt(h3)
+ check = wantCancelledAtRelease(f.lease(2), d)
+ s.ReleasePolicyWriter(bg)
+ check()
+ if !errors.Is(d.Err(), context.Canceled) {
+ t.Fatalf("attempt context: %v, want canceled", d.Err())
+ }
+}
+
+// Review fix round 1, item 2: a stand-down while an acquisition is in flight
+// cancels it, waits for it, and releases the lease it won anyway before
+// returning; and nothing acquires afterwards. The leaser blocks until its
+// context ends, so the order is fixed without sleeps.
+func TestReleasePolicyWriterStopsAnAcquisitionInFlight(t *testing.T) {
+ t.Parallel()
+ bg := context.Background()
+ f := &fakePolicyLeaser{entered: make(chan struct{})}
+ s, rec := policyLeaseServer(f)
+
+ type result struct {
+ h policyHolding
+ ref *doorRefusal
+ }
+ res := make(chan result, 1)
+ go func() {
+ h, ref := s.policyWriter(bg)
+ res <- result{h, ref}
+ }()
+ <-f.entered
+ s.ReleasePolicyWriter(bg)
+
+ if got := f.lease(0).released(); got != 1 {
+ t.Fatalf("the lease won during the stand-down was released %d times before it returned, want 1", got)
+ }
+ r := <-res
+ wantPolicyRefusal(t, r.ref, policyLeaseUnavailableMsg)
+ if s.policyHeld() {
+ t.Fatal("policyHeld true after the stand-down")
+ }
+ rec.has(t, "acquired while stopping")
+
+ _, ref := s.policyWriter(bg)
+ wantPolicyRefusal(t, ref, policyLeaseUnavailableMsg)
+ if f.callCount() != 1 {
+ t.Fatalf("the leaser was called %d times; a stopping console must not acquire", f.callCount())
+ }
}
diff --git a/internal/console/server.go b/internal/console/server.go
index 99e089fdc..14ee6ced2 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -979,13 +979,20 @@ type Server struct {
doorPublishMu sync.Mutex
doorPublish map[string]*sync.Mutex
// The app policy writer lease (policylease.go). policyLeaseMu serializes
- // acquisition and the holding count policyN; policyCur is the current
- // holding, nil when none, read without the mutex so policyHeld and
- // policyAttempt never wait on an acquisition. The zero Server needs no
- // setup.
- policyLeaseMu sync.Mutex
- policyN uint64
- policyCur atomic.Pointer[policyHolding]
+ // acquisition and is held across it. policyStateMu guards the rest and is
+ // never held across I/O, so policyHeld, policyAttempt and
+ // ReleasePolicyWriter never wait on an acquisition: policyN is the last
+ // holding's number, policyCur the current holding (nil when none),
+ // policyAcquiring the acquisition in flight, policyStopping set for good
+ // by ReleasePolicyWriter. policyLog is a test seam (nil: log.Printf). The
+ // zero Server needs no setup.
+ policyLeaseMu sync.Mutex
+ policyStateMu sync.Mutex
+ policyN uint64
+ policyCur *policyHold
+ policyAcquiring *policyAcquisition
+ policyStopping bool
+ policyLog func(format string, args ...any)
// sendApp is how the alert watcher (alertwatch.go) delivers keymail:
// km.SendApp in production, a capture in tests. A field rather than a
// direct call so an alertPass assertion needs no keymail server.