amadan
activator: an aborted legacy start keeps its lease until its unit is proven stopped; Shutdown polls pending stops while joining wakes; a failed lease release is retried (#292)
Astra's review of 9a8e295c (REJECT), three findings: P1. Cancelling the client's /start does not cancel the start job systemd queued behind an edge restart, and with swaps off doWake then released the lease, so the job could later launch a writer no lease covers. A failed start of a nested host's original run unit (generation 0) is now marked legacyStartUncertain (text and errors.Is unchanged), and the failed wake's cleanup stops the unit through instance-control's /stop on a budget of its own (legacyStopBudget) and proves it gone (stop answered, which replaces any queued start job; unit inactive or failed; cgroup empty or gone) before the lease goes. Unproven, the wake is retained in unprovenWakes with its lease, the reaper retries the same proof, and at worst the lease is left to its TTL. ExecBackend.LegacyRunGone, forwarded by SplitBackend. P2. Shutdown's wake joins now run finishPendingStops on the ticker, so a stop left pending is finished and its lease given back while a wake that ignores cancellation is still being joined. P3. lease.Held keeps the record a failed Release could not delete and RetryRelease re-issues the same conditional (If-Match) delete; a record already gone or no longer ours resolves it. The activator's wake bail-outs and the unproven-wake reaper give leases back through giveBack, which keeps a failed release as an obligation retried by finishPendingStops (the sweep and Shutdown's polling) within the remaining budget; Shutdown names any release still unresolved, and any wake still unproven, when it runs out. Tests: TestShutdownAbortsAWakeAndReleasesItsLease (now asserts no release over a queued job), TestAnAbortedStartNotProvenStoppedKeepsItsLease, TestACancelledLegacyStartIsMarkedAndItsStopClearsTheQueuedJob, TestLegacyRunGoneRefusesWhileTheUnitRuns, TestShutdownFinishesPendingStopsWhileJoiningWakes, TestShutdownRetriesAFailedLeaseRelease, TestShutdownLogsALeaseReleaseItCouldNotResolve, and the lease package's TestRetryRelease*. The fake now models a queued job that survives client cancellation and a release that fails.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
11 files changed,
+924
−68
internal/activator/activator.go+161 −30internal/activator/activator_test.go+71 −7internal/activator/backend_exec.go+68 −1internal/activator/backend_legacy_test.go+108 −0internal/activator/backend_unit.go+15 −0internal/activator/containment.go+51 −0internal/activator/maintenance.go+27 −2internal/activator/opset.go+21 −6internal/activator/shutdown_lease_test.go+210 −21internal/lease/lease.go+52 −1internal/lease/release_retry_test.go+140 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index 932aa804f..ec5721a16 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -253,6 +253,89 @@ func releaseLease(ctx context.Context, lease Lease) error { |
| return lease.Release(rctx) |
| } |
| +// releaseRetrier is the OPTIONAL half of Lease that a failed Release |
| +// leaves something to retry with (#292, Astra P3): lease.Held.Release |
| +// spends its handle even when the DELETE fails, so the same conditional |
| +// delete is re-issued through RetryRelease. execLease has it. |
| +type releaseRetrier interface { |
| + RetryRelease(ctx context.Context) error |
| +} |
| + |
| +// owedRelease is a lease whose release failed: kept, and retried by |
| +// finishPendingStops (the sweep, and Shutdown's polling) until it lands or |
| +// Shutdown's budget ends. busy marks a retry in flight; lastLog paces the |
| +// failure lines. Both under a.mu. |
| +type owedRelease struct { |
| + host string |
| + lease releaseRetrier |
| + busy bool |
| + lastLog time.Time |
| +} |
| + |
| +// giveBack is releaseLease for a lease the agent holds for host, with a |
| +// failure kept as an obligation instead of forgotten (#292, Astra P3). |
| +// It reports whether the release landed now. |
| +func (a *Service) giveBack(ctx context.Context, host string, lease Lease) bool { |
| + err := releaseLease(ctx, lease) |
| + if err != nil { |
| + a.oweRelease(host, lease, err) |
| + } |
| + return err == nil |
| +} |
| + |
| +// oweRelease records a failed release of host's lease for retry. A lease |
| +// that cannot retry is left to its TTL, as before. Never called with a.mu |
| +// held. |
| +func (a *Service) oweRelease(host string, lease Lease, err error) { |
| + r, ok := lease.(releaseRetrier) |
| + if !ok { |
| + log.Printf("activator: %s: giving the lease back failed: %v (the lease's TTL frees it)", host, err) |
| + return |
| + } |
| + log.Printf("activator: %s: giving the lease back failed: %v; retrying it", host, err) |
| + a.mu.Lock() |
| + if a.owedReleases == nil { |
| + a.owedReleases = make(map[*owedRelease]struct{}) |
| + } |
| + a.owedReleases[&owedRelease{host: host, lease: r, lastLog: time.Now()}] = struct{}{} |
| + a.mu.Unlock() |
| +} |
| + |
| +// retryOwedReleasesLocked starts one retry of every owed release that has |
| +// none in flight. The caller holds a.mu. |
| +func (a *Service) retryOwedReleasesLocked() { |
| + for o := range a.owedReleases { |
| + if o.busy { |
| + continue |
| + } |
| + o.busy = true |
| + go a.retryOwedRelease(o) |
| + } |
| +} |
| + |
| +// retryOwedRelease is one conditional retry, under the release budget of |
| +// its own. Landed, or no longer ours to release, the obligation goes. |
| +func (a *Service) retryOwedRelease(o *owedRelease) { |
| + ctx, cancel := context.WithTimeout(context.Background(), leaseReleaseBudget) |
| + err := o.lease.RetryRelease(ctx) |
| + cancel() |
| + a.mu.Lock() |
| + o.busy = false |
| + logIt := err == nil || time.Since(o.lastLog) >= 30*time.Second |
| + if err == nil { |
| + delete(a.owedReleases, o) |
| + } else if logIt { |
| + o.lastLog = time.Now() |
| + } |
| + a.mu.Unlock() |
| + switch { |
| + case err == nil: |
| + log.Printf("activator: %s: lease given back on retry", o.host) |
| + case logIt: |
| + log.Printf("activator: %s: retrying the lease's release: %v", o.host, err) |
| + } |
| +} |
| + |
| // leaseHealth is the OPTIONAL half of Lease: a held lease that can say |
| // whether it is still held (no failed renewal, not spent). execLease has it |
| // (lease.Held.Err). A hold's pinned wake asks it at its exec, before the |
| @@ -1106,6 +1189,9 @@ type Service struct { |
| // the host's next wake until the sweep proves it |
| // (reapUnprovenWakesLocked). Under a.mu. |
| unprovenWakes map[string]*unprovenWake |
| + // owedReleases are lease releases that failed and are retried until |
| + // they land (giveBack, #292). Under a.mu. |
| + owedReleases map[*owedRelease]struct{} |
| // outcomeFences are planned-wake outcomes taken but not yet delivered |
| // to OnGen, per host, made lazily (final review F3): a failed or |
| // fallen-back first wake's, from before its entry goes until afterWake's |
| @@ -1894,6 +1980,10 @@ type wokenInstance struct { |
| // kill was not proven (genStartUnproven). cleanupFailedWake keeps the |
| // lease until the sweep proves it, whatever a.gens says. |
| genUnproven bool |
| + // legacyUncertain: the spawn was a start of the host's original run |
| + // unit that root may still carry out (legacyStartUncertain, #292). |
| + // cleanupFailedWake keeps the lease until the unit is proven gone. |
| + legacyUncertain bool |
| // pend is the build a cold wake captured and published as pending at |
| // step 2b (settleWakeIdentity; ruling R-COLDPUBLISH). wake() drops it in |
| // the critical section that ends the wake: by then the entry's |
| @@ -2084,7 +2174,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| if err != nil { |
| if lease != nil { |
| w.stopOutcome.uncertain = true |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease) |
| } |
| var held *errLeaseHeld |
| if errors.As(err, &held) { |
| @@ -2095,7 +2185,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| } |
| if ctx.Err() != nil || !a.wakeAllowed(host) { |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease) |
| return w, ErrFenced |
| } |
| // 1.5. Flag off, the host cgroup is proven empty under the lease before |
| @@ -2131,12 +2221,12 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| log.Printf("activator: %s: %v; leaving the lease to its TTL", host, err) |
| w.stopOutcome.uncertain = true |
| } else { |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease) |
| } |
| return w, fmt.Errorf("restore %s: %w", host, err) |
| } |
| if ctx.Err() != nil || !a.wakeAllowed(host) { |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease) |
| return w, ErrFenced |
| } |
| @@ -2157,7 +2247,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| if plan.planned() { |
| next, err := a.checkPlan(ctx, e, rt, plan) |
| if err != nil { |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease) |
| return w, fmt.Errorf("start %s: %w", host, err) |
| } |
| plan = next |
| @@ -2170,7 +2260,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| // file, so the link is read once more just before the start. It |
| // narrows a repoint after the wait; it cannot close it. |
| if v := a.liveVersion(host); v != held { |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease) |
| return w, fmt.Errorf("start %s: %w: %s is on %q, wants %s", host, errStaleLink, host, v, held) |
| } |
| } |
| @@ -2178,7 +2268,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| var idErr error |
| id, w.pend, idErr = a.settleWakeIdentity(ctx, ib, e, rt, held, heldID) |
| if idErr != nil { |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease) |
| return w, fmt.Errorf("start %s: %w", host, idErr) |
| } |
| if id.StagedPath == "" { |
| @@ -2216,7 +2306,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| // the same way stop() leaves ErrStillRunning. |
| log.Printf("activator: %s: %v; leaving the lease to its TTL", host, err) |
| } else { |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease) |
| } |
| return w, fmt.Errorf("seed %s: %w", host, err) |
| } |
| @@ -2249,7 +2339,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| token string |
| ) |
| if ctx.Err() != nil || !a.wakeAllowed(host) { |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease) |
| return w, ErrFenced |
| } |
| w.stopOutcome.uncertain = true |
| @@ -2268,7 +2358,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| if err != nil { |
| // w goes back with the error, so wake() drops step 2b's |
| // pending build (w.pend) and records this stop's outcome. |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease) |
| return w, fmt.Errorf("start %s: %w", host, err) |
| } |
| // Published onto the entry NOW, not at the awake transition: the |
| @@ -2368,10 +2458,14 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| // running with no process handle (plan 2's startUnitAt). Its lease |
| // is kept, process or not (P2-13). |
| w.genUnproven = genStartUnproven(err) // swapsim:M9 |
| - if proc != nil || w.genUnproven { |
| + // A legacy run unit's start that root may still carry out (a start |
| + // job systemd queued, which cancelling the client does not cancel) |
| + // is stopped and proven gone before its lease goes (#292, P1). |
| + w.legacyUncertain = legacyStartMayRun(err) |
| + if proc != nil || w.genUnproven || w.legacyUncertain { |
| a.cleanupFailedWake(ctx, rt, w, lease) |
| } else { |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease) |
| } |
| return w, fmt.Errorf("start %s: %w", host, err) |
| } |
| @@ -3990,6 +4084,8 @@ func (a *Service) finishPendingStops() { |
| // Failed wakes whose generation is not proven stopped hold their lease |
| // until it is (zero-downtime spec §7.4). |
| a.reapUnprovenWakesLocked() |
| + // Lease releases that failed are retried on the same tick (#292). |
| + a.retryOwedReleasesLocked() |
| a.reapStraySidecarsLocked() |
| for _, e := range a.entries { |
| // Members first (zero-downtime spec §7.4): a generation whose stop |
| @@ -4941,11 +5037,29 @@ func (a *Service) Shutdown(ctx context.Context) { |
| } |
| } |
| a.mu.Unlock() |
| + // One ticker for every wait below, the wake joins included (#292, |
| + // Astra P2): a wake that ignores its cancellation can hold a join until |
| + // the deadline, and a stop left pending meanwhile must still be |
| + // finished, and its lease given back, as soon as its process is proven |
| + // gone, not once every wake has joined. |
| + poll := time.NewTicker(100 * time.Millisecond) |
| + defer poll.Stop() |
| + pollPending := func() { |
| + a.mu.Lock() |
| + a.finishPendingStops() |
| + a.mu.Unlock() |
| + } |
| for _, e := range waking { |
| - select { |
| - case <-e.ready: |
| - case <-ctx.Done(): |
| - return |
| + for joined := false; !joined; { |
| + select { |
| + case <-e.ready: |
| + joined = true |
| + case <-poll.C: |
| + pollPending() |
| + case <-ctx.Done(): |
| + a.logUnresolvedAtExit() |
| + return |
| + } |
| } |
| a.mu.Lock() |
| if a.entries[e.host] == e && e.state == actStopping { |
| @@ -4953,18 +5067,15 @@ func (a *Service) Shutdown(ctx context.Context) { |
| } |
| a.mu.Unlock() |
| } |
| - poll := time.NewTicker(100 * time.Millisecond) |
| - defer poll.Stop() |
| for _, w := range waits { |
| for done := false; !done; { |
| select { |
| case <-w: |
| done = true |
| case <-poll.C: |
| - a.mu.Lock() |
| - a.finishPendingStops() |
| - a.mu.Unlock() |
| + pollPending() |
| case <-ctx.Done(): |
| + a.logUnresolvedAtExit() |
| return |
| } |
| } |
| @@ -4983,9 +5094,13 @@ func (a *Service) Shutdown(ctx context.Context) { |
| // exception to dormancy from the task 2 fix). That flag-off retention |
| // proves only the leader's exit, not its descendants (final review F6, |
| // deferred). With neither, the map is empty and this returns at once. |
| + // |
| + // Lease releases that failed (#292, Astra P3) are retried here the same |
| + // way, inside the same budget (retryOwedReleasesLocked, also inside |
| + // finishPendingStops), and named at exit if they are still owed. |
| for { |
| a.mu.Lock() |
| - left := len(a.unprovenWakes) |
| + left := len(a.unprovenWakes) + len(a.owedReleases) |
| if left > 0 { |
| a.finishPendingStops() |
| } |
| @@ -4996,19 +5111,35 @@ func (a *Service) Shutdown(ctx context.Context) { |
| select { |
| case <-poll.C: |
| case <-ctx.Done(): |
| - a.mu.Lock() |
| - var held []string |
| - for host, u := range a.unprovenWakes { |
| - held = append(held, fmt.Sprintf("%s (generation %d)", host, u.gen.N)) |
| - } |
| - a.mu.Unlock() |
| - slices.Sort(held) |
| - log.Printf("activator: shutdown: out of time with %d failed wake(s) not proven stopped, their leases still held: %s", len(held), strings.Join(held, ", ")) |
| + a.logUnresolvedAtExit() |
| return |
| } |
| } |
| } |
| +// logUnresolvedAtExit names what Shutdown's budget ran out on: failed |
| +// wakes not proven stopped, whose leases are kept, and lease releases that |
| +// failed and were not resolved by a retry. The TTL frees either. |
| +func (a *Service) logUnresolvedAtExit() { |
| + a.mu.Lock() |
| + var held, owed []string |
| + for host, u := range a.unprovenWakes { |
| + held = append(held, fmt.Sprintf("%s (generation %d)", host, u.gen.N)) |
| + } |
| + for o := range a.owedReleases { |
| + owed = append(owed, o.host) |
| + } |
| + a.mu.Unlock() |
| + if len(held) > 0 { |
| + slices.Sort(held) |
| + log.Printf("activator: shutdown: out of time with %d failed wake(s) not proven stopped, their leases still held: %s", len(held), strings.Join(held, ", ")) |
| + } |
| + if len(owed) > 0 { |
| + slices.Sort(owed) |
| + log.Printf("activator: shutdown: out of time with %d lease release(s) unresolved, held until their TTL: %s", len(owed), strings.Join(owed, ", ")) |
| + } |
| +} |
| + |
| // ServeHTTP answers one proxied request for a hibernating instance: wake (or |
| // join a wake), proxy, stamp activity. Lease contention and wake timeouts are |
| // 503s — "sleeping hosts are slow to wake", never a 5xx cascade of retries. |
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go| index 93998a83d..c6b218703 100644 |
| --- a/internal/activator/activator_test.go |
| +++ b/internal/activator/activator_test.go |
| @@ -243,6 +243,8 @@ func (p *fakeProc) exit() { |
| type fakeLease struct { |
| rep *fakeRep |
| b *fakeBackend // records the sync interval the wake threads through |
| + // owed: a Release failed and RetryRelease has not yet landed. Under b.mu. |
| + owed bool |
| } |
| func (l *fakeLease) Replicate(_ context.Context, _ string, syncInterval time.Duration) (Rep, error) { |
| @@ -272,10 +274,53 @@ func (l *fakeLease) Err() error { |
| func (l *fakeLease) Release(ctx context.Context) error { |
| if l.b != nil { |
| l.b.mu.Lock() |
| - l.b.releases++ |
| + defer l.b.mu.Unlock() |
| l.b.releaseCtxErrs = append(l.b.releaseCtxErrs, ctx.Err()) |
| - l.b.mu.Unlock() |
| + if len(l.b.releaseErrs) > 0 { |
| + err := l.b.releaseErrs[0] |
| + l.b.releaseErrs = l.b.releaseErrs[1:] |
| + l.owed = true |
| + return err |
| + } |
| + l.b.releases++ |
| + l.b.queuedAtRelease = append(l.b.queuedAtRelease, l.b.queuedJobs) |
| + } |
| + return nil |
| +} |
| + |
| +// RetryRelease is lease.Held's: it discharges what a failed Release left |
| +// owed, and counts as the release once it lands. |
| +func (l *fakeLease) RetryRelease(context.Context) error { |
| + if l.b == nil { |
| + return nil |
| + } |
| + l.b.mu.Lock() |
| + defer l.b.mu.Unlock() |
| + l.b.retries++ |
| + if !l.owed { |
| + return nil |
| + } |
| + if len(l.b.retryErrs) > 0 { |
| + err := l.b.retryErrs[0] |
| + l.b.retryErrs = l.b.retryErrs[1:] |
| + return err |
| + } |
| + l.owed = false |
| + l.b.releases++ |
| + l.b.queuedAtRelease = append(l.b.queuedAtRelease, l.b.queuedJobs) |
| + return nil |
| +} |
| + |
| +// LegacyRunGone is ExecBackend's: the legacy run unit's stop, which |
| +// replaces a queued start job, and its proof. |
| +func (b *fakeBackend) LegacyRunGone(context.Context, registry.Route) error { |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + b.legacyStops++ |
| + if b.legacyStopErr != nil { |
| + return b.legacyStopErr |
| } |
| + b.queuedJobs = 0 |
| return nil |
| } |
| @@ -307,11 +352,27 @@ type fakeBackend struct { |
| startErr error |
| startGate chan struct{} // when non-nil, Start() blocks until closed |
| // startQueued, when non-nil, blocks Start() until it is closed or the |
| - // wake's context ends, and a start whose context ended returns no |
| - // process. It models a nested start whose systemctl start job systemd |
| - // has queued behind an edge restart: the instance-control client gives |
| - // up when its context does (#292). |
| + // wake's context ends. It models a nested legacy start whose systemctl |
| + // start job systemd has queued behind an edge restart (#292). The |
| + // instance-control client gives up when its context does, but the |
| + // queued job SURVIVES that: a cancelled start counts one in queuedJobs |
| + // and answers as ExecBackend does (no process, legacyStartUncertain). |
| + // Only LegacyRunGone, the unit's stop, clears it. |
| startQueued chan struct{} |
| + queuedJobs int |
| + // legacyStops counts LegacyRunGone calls; legacyStopErr is what they |
| + // answer (an unproven stop) while set. |
| + legacyStops int |
| + legacyStopErr error |
| + // queuedAtRelease is queuedJobs at each lease release that landed: a |
| + // release over a still-queued job is the single-writer violation. |
| + queuedAtRelease []int |
| + // releaseErrs fails the next Release calls, in order (a store that |
| + // refuses or times out the DELETE); retryErrs does the same for |
| + // RetryRelease. retries counts RetryRelease calls. |
| + releaseErrs []error |
| + retryErrs []error |
| + retries int |
| reps []*fakeRep |
| procs []*fakeProc |
| // syncIntervals records, per wake, the replication-class tick doWake |
| @@ -498,7 +559,10 @@ func (b *fakeBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin |
| select { |
| case <-queued: |
| case <-ctx.Done(): |
| - return nil, "", ctx.Err() |
| + b.mu.Lock() |
| + b.queuedJobs++ |
| + b.mu.Unlock() |
| + return nil, "", markLegacyStartUncertain(markStartUncertain(fmt.Errorf("%s: start unit: %w", rt.Host, ctx.Err()))) |
| } |
| } |
| ver := "" |
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index 798479ab6..ddaa90770 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -728,6 +728,10 @@ func (l *execLease) Replicate(ctx context.Context, dbPath string, syncInterval t |
| func (l *execLease) Release(ctx context.Context) error { return l.held.Release(ctx) } |
| +// RetryRelease is releaseRetrier: the held lease's conditional retry of a |
| +// release that failed (#292). |
| +func (l *execLease) RetryRelease(ctx context.Context) error { return l.held.RetryRelease(ctx) } |
| + |
| // Err is the held lease's liveness (leaseHealth): a failed renewal, a |
| // spent handle or a poisoned one. |
| func (l *execLease) Err() error { return l.held.Err() } |
| @@ -2417,7 +2421,11 @@ func (b *ExecBackend) spawnUnitFrom(ctx context.Context, rt registry.Route, sock |
| return fail(markStartUncertain(fmt.Errorf("%s: start unit: %w", rt.Host, err))) |
| } |
| if err != nil { |
| - return fail(markStartUncertain(fmt.Errorf("%s: start unit: %w", rt.Host, err))) |
| + // The original run unit (generation 0). Root was asked to start |
| + // it, and a start job systemd queued (behind an edge restart, say) |
| + // outlives this call's cancellation, so the wake's cleanup stops |
| + // the unit and proves it gone before the lease goes (#292). |
| + return fail(markLegacyStartUncertain(markStartUncertain(fmt.Errorf("%s: start unit: %w", rt.Host, err)))) |
| } |
| if inv == "" { |
| // The server read the invocation after the start and found none: |
| @@ -2489,6 +2497,65 @@ func (b *ExecBackend) startUnitAt(ctx context.Context, rt registry.Route, sock s |
| return up, spawnVer, nil |
| } |
| +// LegacyRunGone stops rt's original run unit (carlos-instance@, generation |
| +// 0) and proves it gone (#292, Astra P1): the failed-wake cleanup's proof |
| +// before an aborted start's lease goes back. The stop goes through |
| +// instance-control's /stop, which runs `systemctl stop`. systemd keeps one |
| +// job per unit and a stop job replaces a queued start job, so a successful |
| +// stop also means that no start job is still queued, even while the edge |
| +// restart that queued it is in progress. Then the unit must read inactive |
| +// (or failed) with its cgroup empty or gone, the facts the exit observer |
| +// trusts (observe, design §6.5). Bounded by ctx, which the caller gives a |
| +// budget of its own; anything short of the proof is an error, and the |
| +// caller keeps the lease. |
| +func (b *ExecBackend) LegacyRunGone(ctx context.Context, rt registry.Route) error { |
| + if b.Control == nil { |
| + return fmt.Errorf("%s: instance control channel not configured", rt.Host) |
| + } |
| + unit := unitName(instancectl.RoleRun, rt.Host, 0) |
| + if err := b.Control.Stop(ctx, rt.Host, instancectl.RoleRun); err != nil { |
| + return fmt.Errorf("stop %s: %w", unit, err) |
| + } |
| + state := b.UnitState |
| + if state == nil { |
| + state = systemctlUnitState |
| + } |
| + events := b.CgroupEvents |
| + if events == nil { |
| + events = cgroupEventsUnder(b.cgroupRoot()) |
| + } |
| + poll := b.ObservePoll |
| + if poll <= 0 { |
| + poll = time.Second |
| + } |
| + slice := instancectl.SliceName(rt.Host) |
| + var last error |
| + for { |
| + active, _, err := state(ctx, unit) |
| + switch { |
| + case err != nil: |
| + last = fmt.Errorf("unit state: %w", err) |
| + case active != "inactive" && active != "failed": |
| + last = fmt.Errorf("unit is %s", active) |
| + default: |
| + exists, populated, eerr := events(slice, unit) |
| + switch { |
| + case eerr != nil: |
| + last = fmt.Errorf("cgroup.events: %w", eerr) |
| + case exists && populated: |
| + last = errors.New("processes remain in its cgroup") |
| + default: |
| + return nil |
| + } |
| + } |
| + select { |
| + case <-ctx.Done(): |
| + return fmt.Errorf("%s not proven stopped: %v", unit, last) |
| + case <-time.After(poll): |
| + } |
| + } |
| +} |
| + |
| // killFailedGen ends a generation whose start failed or answered no |
| // invocation. The start may have succeeded with its answer lost, so the |
| // generation is killed and proven empty (StopGen with kill) on a budget the |
diff --git a/internal/activator/backend_legacy_test.go b/internal/activator/backend_legacy_test.go| new file mode 100644 |
| index 000000000..20a98f426 |
| --- /dev/null |
| +++ b/internal/activator/backend_legacy_test.go |
| @@ -0,0 +1,108 @@ |
| +package activator |
| + |
| +import ( |
| + "context" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/instancectl" |
| +) |
| + |
| +// queuingControl is root's channel during an edge restart (#292): a start |
| +// whose job systemd has queued blocks until the client gives up, and the |
| +// job outlives that. Only a stop of the unit replaces it. |
| +type queuingControl struct { |
| + *fakeControl |
| + mu sync.Mutex |
| + queued bool |
| +} |
| + |
| +func (q *queuingControl) Start(ctx context.Context, host string, role instancectl.Role, max int64) (string, error) { |
| + q.rec("start " + host + " " + string(role)) |
| + q.mu.Lock() |
| + q.queued = true |
| + q.mu.Unlock() |
| + <-ctx.Done() |
| + return "", ctx.Err() |
| +} |
| + |
| +func (q *queuingControl) Stop(ctx context.Context, host string, role instancectl.Role) error { |
| + q.mu.Lock() |
| + q.queued = false |
| + q.mu.Unlock() |
| + return q.fakeControl.Stop(ctx, host, role) |
| +} |
| + |
| +func (q *queuingControl) stillQueued() bool { |
| + q.mu.Lock() |
| + defer q.mu.Unlock() |
| + return q.queued |
| +} |
| + |
| +// A legacy start that is cancelled while root's start job is queued comes |
| +// back marked, so the wake's cleanup knows the job may still run; the |
| +// proof stops the unit, which clears the job, and only then passes. |
| +func TestACancelledLegacyStartIsMarkedAndItsStopClearsTheQueuedJob(t *testing.T) { |
| + q := &queuingControl{fakeControl: &fakeControl{}} |
| + unit := &fakeUnit{} |
| + unit.set("inactive", "", false, false) |
| + b, rt := nestedBackend(t, &fakeControl{}, unit) |
| + b.Control = q |
| + |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + go func() { |
| + for !q.stillQueued() { |
| + time.Sleep(5 * time.Millisecond) |
| + } |
| + cancel() |
| + }() |
| + p, _, err := b.Start(ctx, rt) |
| + if err == nil || p != nil { |
| + t.Fatalf("Start = %v, %v; want no process and an error", p, err) |
| + } |
| + if !legacyStartMayRun(err) { |
| + t.Fatalf("a cancelled legacy start is not marked as one whose job may run: %v", err) |
| + } |
| + if !q.stillQueued() { |
| + t.Fatal("the fake lost its queued job on cancellation; it must model a job that survives") |
| + } |
| + |
| + pctx, pcancel := context.WithTimeout(context.Background(), 2*time.Second) |
| + defer pcancel() |
| + if err := b.LegacyRunGone(pctx, rt); err != nil { |
| + t.Fatalf("LegacyRunGone: %v", err) |
| + } |
| + if q.stillQueued() { |
| + t.Fatal("the proof passed with the start job still queued") |
| + } |
| + if !strings.Contains(q.joined(), "stop "+rt.Host+" run") { |
| + t.Fatalf("the run unit was not stopped through instance-control; calls:\n%s", q.joined()) |
| + } |
| +} |
| + |
| +// The proof fails, and so keeps the lease, while the unit is not stopped |
| +// or its cgroup still holds a process. |
| +func TestLegacyRunGoneRefusesWhileTheUnitRuns(t *testing.T) { |
| + for _, tc := range []struct { |
| + name string |
| + active string |
| + populated bool |
| + }{ |
| + {"active", "active", true}, |
| + {"deactivating", "deactivating", true}, |
| + {"inactive with processes left", "inactive", true}, |
| + } { |
| + t.Run(tc.name, func(t *testing.T) { |
| + unit := &fakeUnit{} |
| + unit.set(tc.active, "inv", true, tc.populated) |
| + b, rt := nestedBackend(t, &fakeControl{}, unit) |
| + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) |
| + defer cancel() |
| + if err := b.LegacyRunGone(ctx, rt); err == nil { |
| + t.Fatal("LegacyRunGone proved a unit that still runs") |
| + } |
| + }) |
| + } |
| +} |
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go| index 8c905f2a4..51daf227d 100644 |
| --- a/internal/activator/backend_unit.go |
| +++ b/internal/activator/backend_unit.go |
| @@ -362,6 +362,21 @@ func (b *SplitBackend) KillGen(ctx context.Context, rt registry.Route, n uint64) |
| return k.KillGen(ctx, rt, n) |
| } |
| +// LegacyRunGone is legacyRunProver for the composed backend (#292): an |
| +// exec row's nested run unit is the exec side's to stop and prove. A |
| +// unit-backed row never starts a carlos-instance@ unit, so it refuses |
| +// rather than claiming a proof. |
| +func (b *SplitBackend) LegacyRunGone(ctx context.Context, rt registry.Route) error { |
| + if rt.Backing == registry.BackingUnit { |
| + return fmt.Errorf("%s: a unit-backed row has no instance run unit: %w", rt.Host, errNoLegacyProof) |
| + } |
| + p, ok := b.Exec.(legacyRunProver) |
| + if !ok { |
| + return fmt.Errorf("%s: %w", rt.Host, errNoLegacyProof) |
| + } |
| + return p.LegacyRunGone(ctx, rt) |
| +} |
| + |
| // Layout implements Classifier for the composed backend: a unit-backed row |
| // is always today's flat unit shape (the exec-only layouts do not apply to |
| // it), and an exec row defers to the exec side's own classification. |
diff --git a/internal/activator/containment.go b/internal/activator/containment.go| index 9e907366d..5ba74d117 100644 |
| --- a/internal/activator/containment.go |
| +++ b/internal/activator/containment.go |
| @@ -37,6 +37,57 @@ func (e startUncertain) Unwrap() error { return e.err } |
| // markStartUncertain wraps err as startUncertain. |
| func markStartUncertain(err error) error { return startUncertain{err: err} } |
| +// legacyStartUncertain marks a failed start of a nested host's original |
| +// run unit (generation 0, carlos-instance@) that root was asked for and may |
| +// still carry out (#292): cancelling the client's /start does not cancel a |
| +// start job systemd has queued, and that job can launch a writer later. |
| +// The failed wake's cleanup must stop the unit and prove it gone |
| +// (legacyRunProver) before the lease goes; unproven, the lease is kept. |
| +// Like startUncertain it changes neither the text nor errors.Is, so the |
| +// member's wording is unchanged. |
| +type legacyStartUncertain struct{ err error } |
| + |
| +func (e legacyStartUncertain) Error() string { return e.err.Error() } |
| +func (e legacyStartUncertain) Unwrap() error { return e.err } |
| + |
| +// markLegacyStartUncertain wraps err as legacyStartUncertain. |
| +func markLegacyStartUncertain(err error) error { return legacyStartUncertain{err: err} } |
| + |
| +// legacyStartMayRun reports whether err is a legacy run unit's start whose |
| +// queued job may still run (legacyStartUncertain). |
| +func legacyStartMayRun(err error) bool { |
| + var l legacyStartUncertain |
| + return errors.As(err, &l) |
| +} |
| + |
| +// legacyRunProver is the backend half of that proof: stop the host's |
| +// original run unit through instance-control and prove it gone (stop |
| +// answered, unit inactive, its cgroup empty). nil is the proof. |
| +// ExecBackend has it; SplitBackend passes it to its exec side. |
| +type legacyRunProver interface { |
| + LegacyRunGone(ctx context.Context, rt registry.Route) error |
| +} |
| + |
| +// errNoLegacyProof: the backend cannot stop a legacy run unit, so nothing |
| +// proves an aborted start gone and its lease is kept. |
| +var errNoLegacyProof = errors.New("activator: this backend cannot prove a run unit stopped") |
| + |
| +// legacyStopBudget bounds one legacy proof: the stop call and the wait for |
| +// the unit to read stopped, on a context of its own (a cancelled wake's |
| +// context is often why the start was aborted). |
| +const legacyStopBudget = instancectl.DeadlineControl |
| + |
| +// legacyRunGone runs the backend's proof for rt under legacyStopBudget. |
| +func (a *Service) legacyRunGone(rt registry.Route) error { |
| + p, ok := a.backend.(legacyRunProver) |
| + if !ok { |
| + return errNoLegacyProof |
| + } |
| + ctx, cancel := context.WithTimeout(context.Background(), legacyStopBudget) |
| + defer cancel() |
| + return p.LegacyRunGone(ctx, rt) |
| +} |
| + |
| // errStartedThenExited is the one uncertain start whose outcome is known to |
| // be a crash at the start: root started the unit and it had exited before |
| // root read its invocation (Astra R3 ruling 206, zdd-readiness Task 6). The |
diff --git a/internal/activator/maintenance.go b/internal/activator/maintenance.go| index 533f615de..1b6cafe14 100644 |
| --- a/internal/activator/maintenance.go |
| +++ b/internal/activator/maintenance.go |
| @@ -332,7 +332,32 @@ func (a *Service) cleanupFailedWake(ctx context.Context, rt registry.Route, w *w |
| err := w.proc.Stop(stopCtx) |
| w.stopOutcome.mainStop, w.stopOutcome.mainExited = err == nil, w.proc.Exited() |
| } |
| - if (a.gens || w.genUnproven || w.proc != nil) && !a.failedWakeStopped(rt, w.gen, w.proc, w.launchLayout) { |
| + // An aborted or failed start of the host's original run unit that root |
| + // may still carry out (#292, Astra P1): cancelling the client's /start |
| + // leaves a start job systemd queued, and released over it, the lease |
| + // would let that job launch a writer no lease covers. The unit is |
| + // stopped (a stop job replaces the queued start) and proven gone on a |
| + // budget of its own first. Unproven, the lease is KEPT: retained below |
| + // and retried by the reaper, and at worst left to its TTL. A stall is |
| + // the safe outcome; two writers are not. |
| + legacy := w.proc == nil && w.legacyUncertain && !w.genUnproven |
| + if legacy { |
| + if err := a.legacyRunGone(rt); err != nil { |
| + log.Printf("activator: %s: the aborted start's unit is not proven stopped (%v); keeping the lease until it is", rt.Host, err) |
| + w.stopOutcome.mainExited = false |
| + w.stopOutcome.uncertain = true |
| + a.mu.Lock() |
| + if a.unprovenWakes == nil { |
| + a.unprovenWakes = make(map[string]*unprovenWake) |
| + } |
| + a.unprovenWakes[rt.Host] = &unprovenWake{rt: rt, gen: w.gen, lease: lease, rep: w.rep, build: w.spawnIdentity, |
| + layout: w.launchLayout, legacy: true} |
| + a.mu.Unlock() |
| + return |
| + } |
| + w.stopOutcome.mainStop, w.stopOutcome.mainExited = true, true |
| + } |
| + if !legacy && (a.gens || w.genUnproven || w.proc != nil) && !a.failedWakeStopped(rt, w.gen, w.proc, w.launchLayout) { |
| // Its leader did not exit (ErrStillRunning, a failed kill) or its |
| // child is not proven empty. |
| w.stopOutcome.mainExited = false |
| @@ -359,5 +384,5 @@ func (a *Service) cleanupFailedWake(ctx context.Context, rt registry.Route, w *w |
| w.stopOutcome.leaseReleased = w.stopOutcome.replicaClosed |
| return |
| } |
| - w.stopOutcome.leaseReleased = releaseLease(ctx, lease) == nil |
| + w.stopOutcome.leaseReleased = a.giveBack(ctx, rt.Host, lease) |
| } |
diff --git a/internal/activator/opset.go b/internal/activator/opset.go| index a14dc1514..3a502923c 100644 |
| --- a/internal/activator/opset.go |
| +++ b/internal/activator/opset.go |
| @@ -749,6 +749,9 @@ type unprovenWake struct { |
| // (wokenInstance.launchLayout), which every retry of the proof decides |
| // by (finding 6). |
| layout Layout |
| + // legacy: an aborted start of the host's original run unit whose |
| + // queued job may still run (#292, Astra P1). Proven by legacyRunGone. |
| + legacy bool |
| } |
| // errWakeUnproven refuses a wake while the host's last failed wake is not |
| @@ -828,10 +831,13 @@ func (a *Service) reapUnprovenWakesLocked() { |
| // |
| // A failed close or release is logged and recorded in the stop history, and |
| // the wake is still forgotten: the generation, which is what retention |
| -// guards, is proven stopped, and a lease handle is spent by its first |
| -// release attempt (lease.Held.Release answers ErrSpent ever after), so a |
| -// retry could never succeed and would refuse the host's wakes forever. The |
| -// lease's TTL bounds a failed release, as for finishStop. |
| +// guards, is proven stopped. A failed close is bounded by the lease's TTL, |
| +// as for finishStop. A failed raw release is kept as an obligation that is |
| +// retried with the same conditional delete (giveBack, #292). |
| +// |
| +// A retained legacy start (#292, Astra P1) has no process and no unit |
| +// generation: it is proven by stopping its original run unit again |
| +// (legacyRunGone), never by the generation proof. |
| func (a *Service) reapUnprovenWake(host string, u *unprovenWake) { |
| if u.proc != nil && !u.proc.Exited() { |
| ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| @@ -841,7 +847,13 @@ func (a *Service) reapUnprovenWake(host string, u *unprovenWake) { |
| log.Printf("activator: %s: retry stop of failed wake's generation %d: %v", host, u.gen.N, err) |
| } |
| } |
| - if !a.failedWakeStopped(u.rt, u.gen, u.proc, u.layout) { |
| + proven := false |
| + if u.legacy { |
| + proven = a.legacyRunGone(u.rt) == nil |
| + } else { |
| + proven = a.failedWakeStopped(u.rt, u.gen, u.proc, u.layout) |
| + } |
| + if !proven { |
| a.mu.Lock() |
| u.busy = false |
| a.mu.Unlock() |
| @@ -855,11 +867,14 @@ func (a *Service) reapUnprovenWake(host string, u *unprovenWake) { |
| outcome.replicaClosed = err == nil |
| } else if u.lease != nil { |
| err = releaseLease(ctx, u.lease) |
| + if err != nil { |
| + a.oweRelease(host, u.lease, err) |
| + } |
| } |
| cancel() |
| outcome.leaseReleased = err == nil |
| if err != nil { |
| - log.Printf("activator: %s: failed wake's generation %d is proven stopped, but giving its lease back failed: %v (the lease's TTL frees it)", host, u.gen.N, err) |
| + log.Printf("activator: %s: failed wake's generation %d is proven stopped, but giving its lease back failed: %v", host, u.gen.N, err) |
| } else { |
| log.Printf("activator: %s: failed wake's generation %d is proven stopped; lease given back", host, u.gen.N) |
| } |
diff --git a/internal/activator/shutdown_lease_test.go b/internal/activator/shutdown_lease_test.go| index e319ad7a2..2bf35fd86 100644 |
| --- a/internal/activator/shutdown_lease_test.go |
| +++ b/internal/activator/shutdown_lease_test.go |
| @@ -2,6 +2,12 @@ package activator |
| import ( |
| "context" |
| + "errors" |
| + "log" |
| + "os" |
| + "slices" |
| + "strings" |
| + "sync" |
| "testing" |
| "time" |
| ) |
| @@ -16,9 +22,63 @@ import ( |
| // is still in flight when the SIGTERM arrives. Shutdown has to abort it |
| // and give its lease back before the process exits; otherwise the next |
| // agent refuses the host for the lease's whole TTL. |
| +// |
| +// Aborting the client's start does not abort the start job systemd queued |
| +// (Astra, review of 9a8e295c, P1): released over that job, the lease would |
| +// let the job launch a writer nobody holds a lease for. So the abort stops |
| +// the unit and proves it gone first, and only then gives the lease back. |
| func TestShutdownAbortsAWakeAndReleasesItsLease(t *testing.T) { |
| + a, b, reqErr := arrangeQueuedRewake(t, "restart.example.com") |
| + |
| + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer cancel() |
| + a.Shutdown(ctx) |
| + if ctx.Err() != nil { |
| + t.Fatal("Shutdown ran out of time instead of aborting the wake") |
| + } |
| + |
| + // Every lease this agent took is given back by the time Shutdown |
| + // returns: the first wake's through its replica's Close, the |
| + // re-wake's through Release, and never over a start job still queued. |
| + acq, _, _ := b.counts() |
| + if got := b.releaseCount(); got != acq { |
| + t.Fatalf("lease releases = %d after Shutdown, want %d (one per acquire)", got, acq) |
| + } |
| + b.mu.Lock() |
| + queuedAt, queuedNow, stops := slices.Clone(b.queuedAtRelease), b.queuedJobs, b.legacyStops |
| + b.mu.Unlock() |
| + if stops == 0 || queuedNow != 0 { |
| + t.Fatalf("the aborted start's queued job was never stopped (stops %d, queued %d)", stops, queuedNow) |
| + } |
| + for _, q := range queuedAt { |
| + if q != 0 { |
| + t.Fatalf("a lease was released with %d start job(s) still queued: %v", q, queuedAt) |
| + } |
| + } |
| + for _, err := range b.releaseCtxErrors() { |
| + if err != nil { |
| + t.Fatalf("a lease was released under a dead context (%v)", err) |
| + } |
| + } |
| + if n := a.entryCount(); n != 0 { |
| + t.Fatalf("Shutdown left %d entr(ies) behind", n) |
| + } |
| + select { |
| + case err := <-reqErr: |
| + if err == nil { |
| + t.Fatal("the request was served by a wake Shutdown aborted") |
| + } |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("the queued request never returned") |
| + } |
| +} |
| + |
| +// arrangeQueuedRewake is #292's arrangement: host awake, its unit stopped |
| +// by systemd, and a request's re-wake inside a start that systemd has |
| +// queued behind the edge restart. It returns the request's outcome. |
| +func arrangeQueuedRewake(t *testing.T, host string) (*Service, *fakeBackend, <-chan error) { |
| + t.Helper() |
| a, b, _, _ := testActivator(t, time.Hour) |
| - host := "restart.example.com" |
| arrangeAwake(t, a, testRoute(host)) |
| // systemd stops the bound unit, well before the agent's SIGTERM. |
| @@ -41,39 +101,168 @@ func TestShutdownAbortsAWakeAndReleasesItsLease(t *testing.T) { |
| if acq, _, _ := b.counts(); acq != 2 { |
| t.Fatalf("acquires = %d, want 2 (the first wake and the re-wake)", acq) |
| } |
| + return a, b, reqErr |
| +} |
| + |
| +// P1's other half: an aborted legacy start whose unit cannot be proven |
| +// stopped KEEPS its lease. A lease left to its TTL is a 15-minute stall; a |
| +// lease given back over a queued start job is two writers. |
| +func TestAnAbortedStartNotProvenStoppedKeepsItsLease(t *testing.T) { |
| + a, b, _ := arrangeQueuedRewake(t, "unproven.example.com") |
| + b.mu.Lock() |
| + b.legacyStopErr = errors.New("stop carlos-instance@unproven: instance control socket: connection refused") |
| + b.mu.Unlock() |
| + |
| + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| + defer cancel() |
| + a.Shutdown(ctx) |
| + |
| + acq, _, _ := b.counts() |
| + if got := b.releaseCount(); got != acq-1 { |
| + t.Fatalf("lease releases = %d, want %d: the first wake's, and NOT the aborted start's", got, acq-1) |
| + } |
| + b.mu.Lock() |
| + stops := b.legacyStops |
| + b.mu.Unlock() |
| + if stops == 0 { |
| + t.Fatal("the aborted start's unit was never asked to stop") |
| + } |
| + a.mu.Lock() |
| + _, retained := a.unprovenWakes["unproven.example.com"] |
| + a.mu.Unlock() |
| + if !retained { |
| + t.Fatal("the unproven start was not retained with its lease") |
| + } |
| +} |
| + |
| +// P2: a wake that ignores its cancellation holds Shutdown's join, and an |
| +// entry whose stop was left pending must still be finished meanwhile: |
| +// its process exits, and its lease goes back then, not once every wake |
| +// has joined. |
| +func TestShutdownFinishesPendingStopsWhileJoiningWakes(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + stuckHost, wakingHost := "stuck.example.com", "waking.example.com" |
| + b.procStuck = true |
| + arrangeAwake(t, a, testRoute(stuckHost)) |
| + stuck := b.lastProc() |
| + stuckRep := b.lastRep() |
| + |
| + b.mu.Lock() |
| + b.procStuck = false |
| + gate := make(chan struct{}) |
| + b.startGate = gate |
| + b.mu.Unlock() |
| + go func() { |
| + if release, err := a.beginRequest(context.Background(), testRoute(wakingHost)); err == nil { |
| + release() |
| + } |
| + }() |
| + waitFor(t, "the wake to reach Start", func() bool { return b.startCount() == 2 }) |
| + |
| + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) |
| + defer cancel() |
| + done := make(chan struct{}) |
| + go func() { a.Shutdown(ctx); close(done) }() |
| + waitFor(t, "the stuck stop to be left pending", func() bool { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + e := a.entries[stuckHost] |
| + return e != nil && e.pendingStop |
| + }) |
| + stuck.exit() |
| + waitFor(t, "the pending stop's lease to go back while a wake is still joining", stuckRep.isClosed) |
| + select { |
| + case <-done: |
| + t.Fatal("Shutdown returned while a wake was still spawning") |
| + default: |
| + } |
| + close(gate) |
| + <-done |
| +} |
| + |
| +// P3: a failed release is an obligation, not a forgotten handle. A wake |
| +// that fails gives its lease back, the store refuses the DELETE, and |
| +// Shutdown retries it (conditionally) inside its own budget. |
| +func TestShutdownRetriesAFailedLeaseRelease(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + b.startErr = errors.New("start failed") |
| + b.releaseErrs = []error{errors.New("delete lease: context deadline exceeded")} |
| + if _, err := a.beginRequest(context.Background(), testRoute("refused.example.com")); err == nil { |
| + t.Fatal("the wake succeeded") |
| + } |
| + if got := b.releaseCount(); got != 0 { |
| + t.Fatalf("lease releases = %d before Shutdown, want 0 (the DELETE failed)", got) |
| + } |
| ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| defer cancel() |
| a.Shutdown(ctx) |
| if ctx.Err() != nil { |
| - t.Fatal("Shutdown ran out of time instead of aborting the wake") |
| + t.Fatal("Shutdown ran out of time") |
| + } |
| + b.mu.Lock() |
| + retries := b.retries |
| + b.mu.Unlock() |
| + if got := b.releaseCount(); got != 1 || retries == 0 { |
| + t.Fatalf("lease releases = %d after %d retries, want the failed release retried and landed", got, retries) |
| } |
| +} |
| - // Every lease this agent took is given back by the time Shutdown |
| - // returns: the first wake's through its replica's Close, the |
| - // re-wake's through Release. |
| - acq, _, _ := b.counts() |
| - if got := b.releaseCount(); got != acq { |
| - t.Fatalf("lease releases = %d after Shutdown, want %d (one per acquire)", got, acq) |
| +// P3's other half: a release still unresolved when Shutdown's budget ends |
| +// is logged by host, not dropped silently. |
| +func TestShutdownLogsALeaseReleaseItCouldNotResolve(t *testing.T) { |
| + logs := lockedLog(t) |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + b.startErr = errors.New("start failed") |
| + b.releaseErrs = []error{errors.New("delete lease: timeout")} |
| + for i := 0; i < 1000; i++ { |
| + b.retryErrs = append(b.retryErrs, errors.New("delete lease: timeout")) |
| } |
| - for _, err := range b.releaseCtxErrors() { |
| - if err != nil { |
| - t.Fatalf("a lease was released under a dead context (%v)", err) |
| - } |
| + if _, err := a.beginRequest(context.Background(), testRoute("dark.example.com")); err == nil { |
| + t.Fatal("the wake succeeded") |
| } |
| - if n := a.entryCount(); n != 0 { |
| - t.Fatalf("Shutdown left %d entr(ies) behind", n) |
| + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) |
| + defer cancel() |
| + a.Shutdown(ctx) |
| + if got := b.releaseCount(); got != 0 { |
| + t.Fatalf("lease releases = %d, want 0", got) |
| } |
| - select { |
| - case err := <-reqErr: |
| - if err == nil { |
| - t.Fatal("the request was served by a wake Shutdown aborted") |
| - } |
| - case <-time.After(5 * time.Second): |
| - t.Fatal("the queued request never returned") |
| + if out := logs.String(); !strings.Contains(out, "unresolved") || !strings.Contains(out, "dark.example.com") { |
| + t.Fatalf("no unresolved-release line naming the host at exit; log:\n%s", out) |
| } |
| } |
| +// lockedBuf is a log sink safe against the activator's own goroutines. |
| +type lockedBuf struct { |
| + mu sync.Mutex |
| + buf strings.Builder |
| +} |
| + |
| +func (l *lockedBuf) Write(p []byte) (int, error) { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + return l.buf.Write(p) |
| +} |
| + |
| +func (l *lockedBuf) String() string { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + return l.buf.String() |
| +} |
| + |
| +func lockedLog(t *testing.T) *lockedBuf { |
| + t.Helper() |
| + l := &lockedBuf{} |
| + flags := log.Flags() |
| + log.SetOutput(l) |
| + log.SetFlags(0) |
| + t.Cleanup(func() { |
| + log.SetOutput(os.Stderr) |
| + log.SetFlags(flags) |
| + }) |
| + return l |
| +} |
| + |
| // Once Shutdown has begun, nothing wakes. A request queued on a stop must |
| // not wake the host afresh when that stop finishes after Shutdown began: |
| // the wake would take a lease once Shutdown had stopped looking. Startup |
diff --git a/internal/lease/lease.go b/internal/lease/lease.go| index 61788e219..e221468d8 100644 |
| --- a/internal/lease/lease.go |
| +++ b/internal/lease/lease.go |
| @@ -142,6 +142,11 @@ type Held struct { |
| poison error |
| replicaClient litestream.ReplicaClient // private local-replica test seam |
| generationPrefix string |
| + // owed is the lease record a failed Release could not delete (#292). |
| + // Release spends the handle either way; RetryRelease re-issues the |
| + // same conditional delete until it lands or the lease is no longer |
| + // ours. Guarded by mu. |
| + owed *litestream.Lease |
| } |
| // Acquire claims the single-writer lease at s3://bucket/prefix/lock.json |
| @@ -276,7 +281,53 @@ func (h *Held) Release(ctx context.Context) error { |
| if lease == nil { |
| return nil |
| } |
| - return h.leaser.ReleaseLease(ctx, lease) |
| + err := h.leaser.ReleaseLease(ctx, lease) |
| + if err != nil && !releaseSettled(err) { |
| + // The handle is spent, but the lease record may still stand: keep |
| + // it so RetryRelease can delete it, on the same condition (#292). |
| + h.mu.Lock() |
| + h.owed = lease |
| + h.mu.Unlock() |
| + } |
| + return err |
| +} |
| + |
| +// releaseSettled reports whether a release error still leaves nothing of |
| +// this lease to delete: the record is already gone, or its ETag no longer |
| +// matches (the lease lapsed and is someone else's now). |
| +func releaseSettled(err error) bool { |
| + return errors.Is(err, lss3.ErrLeaseAlreadyReleased) || errors.Is(err, litestream.ErrLeaseNotHeld) |
| +} |
| + |
| +// ReleaseOwed reports whether a failed Release left the lease record |
| +// standing for RetryRelease. |
| +func (h *Held) ReleaseOwed() bool { |
| + h.mu.Lock() |
| + defer h.mu.Unlock() |
| + return h.owed != nil |
| +} |
| + |
| +// RetryRelease discharges what a failed Release left owed (#292): the same |
| +// conditional delete (If-Match on the lease's ETag), so it can never delete |
| +// a lease someone else has taken since. nil once nothing is owed, including |
| +// when the record is already gone or is no longer ours. Callers do not run |
| +// it concurrently with itself. |
| +func (h *Held) RetryRelease(ctx context.Context) error { |
| + h.mu.Lock() |
| + lease := h.owed |
| + h.mu.Unlock() |
| + if lease == nil { |
| + return nil |
| + } |
| + if err := h.leaser.ReleaseLease(ctx, lease); err != nil && !releaseSettled(err) { |
| + return err |
| + } |
| + h.mu.Lock() |
| + if h.owed == lease { |
| + h.owed = nil |
| + } |
| + h.mu.Unlock() |
| + return nil |
| } |
| // Replicate trades the held lease for running replication of dbPath to the |
diff --git a/internal/lease/release_retry_test.go b/internal/lease/release_retry_test.go| new file mode 100644 |
| index 000000000..e4f5696d7 |
| --- /dev/null |
| +++ b/internal/lease/release_retry_test.go |
| @@ -0,0 +1,140 @@ |
| +package lease |
| + |
| +import ( |
| + "context" |
| + "net/http" |
| + "net/http/httptest" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/aws/aws-sdk-go-v2/aws" |
| + "github.com/aws/aws-sdk-go-v2/credentials" |
| + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| + "github.com/benbjohnson/litestream" |
| + lss3 "github.com/benbjohnson/litestream/s3" |
| +) |
| + |
| +// deleteStub answers the lease's DELETE with each status in turn and |
| +// records the If-Match header it was sent. |
| +type deleteStub struct { |
| + mu sync.Mutex |
| + statuses []int |
| + ifMatch []string |
| +} |
| + |
| +func (s *deleteStub) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| + if r.Method != http.MethodDelete { |
| + http.Error(w, "unexpected "+r.Method, http.StatusMethodNotAllowed) |
| + return |
| + } |
| + s.mu.Lock() |
| + s.ifMatch = append(s.ifMatch, r.Header.Get("If-Match")) |
| + status := http.StatusNoContent |
| + if len(s.statuses) > 0 { |
| + status, s.statuses = s.statuses[0], s.statuses[1:] |
| + } |
| + s.mu.Unlock() |
| + switch status { |
| + case http.StatusNoContent: |
| + w.WriteHeader(status) |
| + case http.StatusPreconditionFailed: |
| + w.Header().Set("Content-Type", "application/xml") |
| + w.WriteHeader(status) |
| + _, _ = w.Write([]byte("<Error><Code>PreconditionFailed</Code><Message>no</Message></Error>")) |
| + default: |
| + w.Header().Set("Content-Type", "application/xml") |
| + w.WriteHeader(status) |
| + _, _ = w.Write([]byte("<Error><Code>AccessDenied</Code><Message>no</Message></Error>")) |
| + } |
| +} |
| + |
| +func (s *deleteStub) sent() []string { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + return append([]string(nil), s.ifMatch...) |
| +} |
| + |
| +// heldAgainst builds a Held whose lease record has ETag etag, with its |
| +// leaser pointed at srv. The TTL is long, so the renewer never renews |
| +// inside the test. |
| +func heldAgainst(t *testing.T, srv *httptest.Server, etag string) *Held { |
| + t.Helper() |
| + client := awss3.New(awss3.Options{ |
| + Region: "us-east-1", |
| + BaseEndpoint: aws.String(srv.URL), |
| + UsePathStyle: true, |
| + Credentials: credentials.NewStaticCredentialsProvider("k", "s", ""), |
| + RetryMaxAttempts: 1, |
| + }) |
| + leaser := lss3.NewLeaser() |
| + leaser.SetClient(client) |
| + leaser.Bucket = "b" |
| + leaser.Path = "p" |
| + leaser.TTL = time.Hour |
| + rec := &litestream.Lease{Generation: 1, ETag: etag, Owner: "test", ExpiresAt: time.Now().Add(time.Hour)} |
| + return &Held{leaser: leaser, rn: startRenewer(leaser, rec, time.Now(), marginForTTL(time.Hour))} |
| +} |
| + |
| +// A failed release leaves a conditional obligation that RetryRelease |
| +// discharges with the same ETag (#292): Release spends the handle, so |
| +// without it the lease stays held for its whole TTL. |
| +func TestRetryReleaseDischargesAFailedRelease(t *testing.T) { |
| + stub := &deleteStub{statuses: []int{http.StatusForbidden, http.StatusNoContent}} |
| + srv := httptest.NewServer(stub) |
| + defer srv.Close() |
| + h := heldAgainst(t, srv, `"e1"`) |
| + ctx := context.Background() |
| + |
| + if err := h.Release(ctx); err == nil { |
| + t.Fatal("Release succeeded against a refusing store") |
| + } |
| + if !h.ReleaseOwed() { |
| + t.Fatal("a failed release left no obligation") |
| + } |
| + if err := h.RetryRelease(ctx); err != nil { |
| + t.Fatalf("RetryRelease: %v", err) |
| + } |
| + if h.ReleaseOwed() { |
| + t.Fatal("the obligation outlived a successful retry") |
| + } |
| + if err := h.RetryRelease(ctx); err != nil { |
| + t.Fatalf("RetryRelease with nothing owed: %v", err) |
| + } |
| + got := stub.sent() |
| + if len(got) != 2 || got[0] != `"e1"` || got[1] != `"e1"` { |
| + t.Fatalf("If-Match sent = %q, want the lease's ETag on both deletes, and no third", got) |
| + } |
| +} |
| + |
| +// A retry whose condition fails (the lease expired and someone else holds |
| +// it now) has nothing of ours left to release: the obligation is resolved, |
| +// and nothing else's lease is deleted. |
| +func TestRetryReleaseResolvesWhenTheLeaseIsNoLongerOurs(t *testing.T) { |
| + stub := &deleteStub{statuses: []int{http.StatusForbidden, http.StatusPreconditionFailed}} |
| + srv := httptest.NewServer(stub) |
| + defer srv.Close() |
| + h := heldAgainst(t, srv, `"e1"`) |
| + ctx := context.Background() |
| + _ = h.Release(ctx) |
| + if err := h.RetryRelease(ctx); err != nil { |
| + t.Fatalf("RetryRelease after a failed condition: %v", err) |
| + } |
| + if h.ReleaseOwed() { |
| + t.Fatal("a lease that is no longer ours is still owed") |
| + } |
| +} |
| + |
| +// A successful release owes nothing. |
| +func TestASuccessfulReleaseOwesNothing(t *testing.T) { |
| + stub := &deleteStub{} |
| + srv := httptest.NewServer(stub) |
| + defer srv.Close() |
| + h := heldAgainst(t, srv, `"e1"`) |
| + if err := h.Release(context.Background()); err != nil { |
| + t.Fatalf("Release: %v", err) |
| + } |
| + if h.ReleaseOwed() { |
| + t.Fatal("a successful release left an obligation") |
| + } |
| +} |