amadan
zdd plan 3 task 2: the entry's operation set, the barrier in stop(), Shutdown joins waking entries, failed wakes keep an unproven generation's lease (P2-13)
Every generation other than the current one lives in the entry's operation set, and every terminal stop joins it before the final checkpoint and the lease release. A stop gives its lease up only through releasable: no member, no stray sidecar, and the current generation proven terminated by genTerminated (leader exited, child empty through KillGen, or a claims-on Stop that completed). With swaps on, Shutdown cancels and waits for waking entries. P2-13: StartGen hands back the generation number with its error, a start whose kill was unproven (errGenUnproven, instancectl.ErrGenUnproven) keeps its lease in unprovenWakes whatever a.gens says, the host's next wake is refused until the sweep proves the generation's child empty, and startUnitAt returns its unit when the stop after a failed health wait failed. Gen gains Worker (moved here from Task 11).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
9 files changed,
+1408
−31
internal/activator/activator.go+132 −20internal/activator/backend_exec.go+17 −2internal/activator/backend_unit.go+3 −4internal/activator/claimwake.go+13 −0internal/activator/generations.go+5 −0internal/activator/genservice.go+4 −4internal/activator/maintenance.go+28 −1internal/activator/opset.go+495 −0internal/activator/opset_test.go+711 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index 7dea2b1cf..d98cc7bcd 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -412,6 +412,16 @@ type actEntry struct { |
| stopRetrying bool |
| stopRetriedAt time.Time |
| + // ops is this entry's operation set (zero-downtime spec §7.4): every |
| + // generation other than the current one that is spawning, |
| + // health-checking or draining. Written under a.mu. Empty unless a swap |
| + // has run on this entry, and swaps need plan 1's generations switch. |
| + ops opSet |
| + // genReaping is true while reapCurrentGen is running for this pending |
| + // stop, so finishPendingStops does not start a second one. Written |
| + // under a.mu. |
| + genReaping bool |
| + |
| // Standalone sidecar ("worker") entries. A worker is this entry's own |
| // proc, not a passenger on someone else's: the co-spawn fields above stay |
| // zero for it, and it never becomes a co-spawn parent either. |
| @@ -673,6 +683,15 @@ type Service struct { |
| fences map[string]uint64 |
| tokenPreparations map[string]chan struct{} |
| stopHistory map[string]*stopObservation |
| + // straySidecars are legacy sidecars whose termination is unproven, per |
| + // host (zero-downtime spec §7.4). Task 7 tracks into it; a stop keeps |
| + // its lease while its host has one (releasable). Under a.mu. |
| + straySidecars map[string][]*straySidecar |
| + // unprovenWakes are failed wakes whose generation is not proven stopped |
| + // (spec §7.4), per host, made lazily. Each keeps its lease and refuses |
| + // the host's next wake until the sweep proves it |
| + // (reapUnprovenWakesLocked). Under a.mu. |
| + unprovenWakes map[string]*unprovenWake |
| sweepStop chan struct{} |
| sweepDone chan struct{} |
| @@ -991,7 +1010,7 @@ func (a *Service) wake(e *actEntry) { |
| defer cancel() |
| a.mu.Lock() |
| e.wakeCancel = cancel |
| - if a.fences[e.host] != 0 { |
| + if a.fences[e.host] != 0 || (a.gens && a.shuttingDown) { |
| cancel() |
| } |
| a.mu.Unlock() |
| @@ -1071,7 +1090,7 @@ func (a *Service) wake(e *actEntry) { |
| log.Printf("activator: %s: link %s to %s: %v (processes dialling the canonical path will not reach it)", e.host, e.route.Socket, e.gen.Socket, err) |
| } |
| } |
| - if a.fences[e.host] != 0 { |
| + if a.fences[e.host] != 0 || (a.gens && a.shuttingDown) { |
| a.beginStopLocked(e) |
| } else if e.bounceAfterWake { |
| // A config delivery or `carlos restart` asked for while this wake |
| @@ -1160,6 +1179,10 @@ type wokenInstance struct { |
| spawnVersion string // live/<host> target read at exec — for build convergence |
| spawnIdentity release.StagedIdentity // the build exec'd, as the backend captured it (workers; spec §7.9) |
| gen Gen // the spawned generation's handle (instances only; zero for a worker) |
| + // genUnproven: the spawn failed and its error says the generation's |
| + // kill was not proven (genStartUnproven). cleanupFailedWake keeps the |
| + // lease until the sweep proves it, whatever a.gens says. |
| + genUnproven bool |
| // spawnTook is how long Backend.Start took: exec to the socket accepting, |
| // which is the instance's OWN boot and nothing of the platform's. wakeTook |
| // is the whole wake around it, filled in by wake() rather than here |
| @@ -1231,6 +1254,17 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| } |
| } |
| + // 0.6. A failed wake whose generation is not proven stopped still holds |
| + // the host's lease, and its process may still run against the database |
| + // (zero-downtime spec §7.4). No new wake starts until the sweep proves |
| + // it (reapUnprovenWakesLocked). |
| + a.mu.Lock() |
| + _, unproven := a.unprovenWakes[host] |
| + a.mu.Unlock() |
| + if unproven { |
| + return w, fmt.Errorf("%s: refusing to wake: %w", host, errWakeUnproven) |
| + } |
| + |
| // 1. The lease first: everything after this line is done as the prefix's |
| // single writer. Contention is not retried here — on one box it means a |
| // previous incarnation hasn't released; tell the client to come back after |
| @@ -1382,8 +1416,12 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| spawnTook := a.now().Sub(spawnStart) |
| w.proc = proc |
| if err != nil { |
| - if proc != nil { |
| - a.cleanupFailedWake(ctx, w, lease) |
| + // A start whose kill was unproven may have left its generation |
| + // 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.cleanupFailedWake(ctx, rt, w, lease) |
| } else { |
| w.stopOutcome.leaseReleased = lease.Release(ctx) == nil |
| } |
| @@ -1402,13 +1440,13 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| // stopHistory blind to exactly the case it exists for. |
| if e.worker { |
| if serr := a.awaitWorkerSettle(ctx, e, proc); serr != nil { |
| - a.cleanupFailedWake(ctx, w, lease) |
| + a.cleanupFailedWake(ctx, rt, w, lease) |
| return w, fmt.Errorf("start %s: %w", host, serr) |
| } |
| } |
| if ctx.Err() != nil || !a.wakeAllowed(host) { |
| - a.cleanupFailedWake(ctx, w, lease) |
| + a.cleanupFailedWake(ctx, rt, w, lease) |
| return w, ErrFenced |
| } |
| // 5. Trade the held lease for running replication, at the route's |
| @@ -1419,7 +1457,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| rep, err := lease.Replicate(ctx, rt.DBPath, replicationSyncInterval(rt.Replication)) |
| w.rep = rep |
| if err != nil { |
| - a.cleanupFailedWake(ctx, w, lease) |
| + a.cleanupFailedWake(ctx, rt, w, lease) |
| return w, fmt.Errorf("replicate %s: %w", host, err) |
| } |
| @@ -2274,6 +2312,13 @@ func (a *Service) stop(e *actEntry) { |
| } |
| } |
| + // The barrier (zero-downtime spec §7.4). Every generation in this |
| + // entry's operation set is cancelled and joined before the final |
| + // checkpoint and the lease release below: a spawning or health-checking |
| + // successor, and a draining retiring generation. With no swap ever run |
| + // on this entry the set is empty and this returns at once. |
| + joined := a.joinOps(e) // swapsim:M1 |
| + |
| // Co-stop, sidecar FIRST — before the next-due probe and the instance's |
| // own SIGTERM. Two reasons: the sidecar must never outlive its instance |
| // (an orphan's reconnect loop would be a standing wake source — the |
| @@ -2346,15 +2391,14 @@ func (a *Service) stop(e *actEntry) { |
| // the rest of what this function would have done, the moment |
| // termination is actually proven. |
| log.Printf("activator: stop %s: %v; keeping the lease until termination is proven", e.host, err) |
| - a.mu.Lock() |
| - e.pendingStop = true |
| - e.stopOutcome = outcome |
| // The call just above is this entry's first stop attempt, and it |
| - // was already bounded by actStopTimeout. Recording it here is |
| - // what makes finishPendingStops' own cadence a true actStopTimeout |
| - // apart from THIS attempt, rather than firing a redundant retry |
| - // the moment pendingStop becomes visible to it. |
| - e.stopRetriedAt = a.now() |
| + // was already bounded by actStopTimeout. keepPendingLocked |
| + // records it as stopRetriedAt, which is what makes |
| + // finishPendingStops' own cadence a true actStopTimeout apart |
| + // from THIS attempt, rather than firing a redundant retry the |
| + // moment pendingStop becomes visible to it. |
| + a.mu.Lock() |
| + a.keepPendingLocked(e, outcome) |
| a.mu.Unlock() |
| return |
| } |
| @@ -2365,6 +2409,29 @@ func (a *Service) stop(e *actEntry) { |
| if err != nil { |
| log.Printf("activator: stop %s: %v", e.host, err) |
| } |
| + if !joined { |
| + // A generation's termination is unproven, so the lease stays held, |
| + // as for ErrStillRunning above. finishPendingStops reaps the member |
| + // and then finishes this stop. |
| + log.Printf("activator: stop %s: a generation's termination is unproven; keeping the lease until it is", e.host) |
| + a.mu.Lock() |
| + a.keepPendingLocked(e, outcome) |
| + a.mu.Unlock() |
| + return |
| + } |
| + if a.gens && !a.releasable(e) { // swapsim:M8 |
| + // THE retain-until-proven rule (spec §7.4): unproven, the stop stays |
| + // pending with the lease held, exactly as for ErrStillRunning, and |
| + // the sweep retries (reapCurrentGen). That covers a leader still |
| + // alive after a failed kill (a claims-on KillAndEmpty error is not |
| + // ErrStillRunning, so it reaches here), a child not proven empty, a |
| + // worker's host cgroup, and a stray sidecar. |
| + log.Printf("activator: stop %s: not every process of the entry is proven gone; keeping the lease until it is", e.host) |
| + a.mu.Lock() |
| + a.keepPendingLocked(e, outcome) |
| + a.mu.Unlock() |
| + return |
| + } |
| a.finishStop(e, outcome) |
| } |
| @@ -2399,8 +2466,32 @@ func (a *Service) finishPendingStop(e *actEntry, outcome *stopObservation) { |
| // that exits during shutdown would never get its final checkpoint. The |
| // caller holds a.mu. |
| 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() |
| for _, e := range a.entries { |
| - if !e.pendingStop || e.proc == nil { |
| + // Members first (zero-downtime spec §7.4): a generation whose stop |
| + // could not be proven is finished here once it exits. |
| + a.reapMembersLocked(e) |
| + if !e.pendingStop || len(e.ops.members) > 0 { |
| + continue |
| + } |
| + if a.gens && (e.proc == nil || e.proc.Exited()) { |
| + // Members gone and the main process (if any) exited: the stop |
| + // finishes once the retain-until-proven rule holds (releasable, |
| + // in reapCurrentGen). A process still alive falls through to |
| + // main's re-stop cadence below. |
| + if !e.genReaping { |
| + e.genReaping = true |
| + go a.reapCurrentGen(e) |
| + } |
| + continue |
| + } |
| + if e.proc == nil { |
| + // Not reached: only the barrier and releasable leave a stop |
| + // pending without a process, and both need generations, which |
| + // the arm above takes. Kept as the guard this loop always had, |
| + // so a broken invariant holds the lease rather than panicking. |
| continue |
| } |
| if e.proc.Exited() { |
| @@ -3039,6 +3130,7 @@ func (a *Service) Shutdown(ctx context.Context) { |
| // checkpoint instead of being abandoned mid-drain. |
| a.finishPendingStops() |
| var waits []chan struct{} |
| + var waking []*actEntry |
| for _, e := range a.entries { |
| switch e.state { |
| case actAwake: |
| @@ -3047,13 +3139,33 @@ func (a *Service) Shutdown(ctx context.Context) { |
| case actStopping: |
| waits = append(waits, e.stopped) |
| case actWaking: |
| - // A mid-flight wake resolves on its own; its instance will be |
| - // orphaned only if it wins the race with process exit — acceptable |
| - // for the rare shutdown-during-wake, and the lease TTL still |
| - // bounds the damage. |
| + // With swaps on, Shutdown covers waking entries too |
| + // (zero-downtime spec §7.4). The wake is cancelled, and once it |
| + // resolves, wake() itself stops the entry through the barrier. |
| + // With swaps off, today's rule holds: a mid-flight wake resolves |
| + // on its own, its instance is orphaned only if it wins the race |
| + // with process exit, and the lease TTL still bounds the damage. |
| + if a.gens { // swapsim:M6 |
| + if e.wakeCancel != nil { |
| + e.wakeCancel() |
| + } |
| + waking = append(waking, e) |
| + } |
| } |
| } |
| a.mu.Unlock() |
| + for _, e := range waking { |
| + select { |
| + case <-e.ready: |
| + case <-ctx.Done(): |
| + return |
| + } |
| + a.mu.Lock() |
| + if a.entries[e.host] == e && e.state == actStopping { |
| + waits = append(waits, e.stopped) |
| + } |
| + a.mu.Unlock() |
| + } |
| poll := time.NewTicker(100 * time.Millisecond) |
| defer poll.Stop() |
| for _, w := range waits { |
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index 99e2f193d..0a7c545be 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -1212,10 +1212,16 @@ func (b *ExecBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, |
| } |
| proc, ver, err = b.startChildAt(ctx, rt, sock, n) |
| } |
| + g := Gen{N: n, Socket: sock, Control: sock + ".control"} |
| if err != nil { |
| - return proc, Gen{}, err |
| + // The handle goes back with the error (P2-13): a spawn that failed |
| + // may have left generation n running, and its number is what the |
| + // caller proves it stopped by (KillGen). Refusals above spawned |
| + // nothing and answer no handle. |
| + return proc, g, err |
| } |
| - return proc, Gen{N: n, Socket: sock, Control: sock + ".control", Build: ver}, nil |
| + g.Build = ver |
| + return proc, g, nil |
| } |
| // flatGenSocket is GenSocketPath, under RunDirOverride/g in a test. |
| @@ -1620,6 +1626,15 @@ func (b *ExecBackend) startUnitAt(ctx context.Context, rt registry.Route, sock s |
| // just means it had not landed yet within actStopTimeout, not |
| // that the wake should be retried differently. |
| log.Printf("activator: %s: stop after failed health wait: %v", rt.Host, serr) |
| + // The unit may still run, so its process goes back with the |
| + // health error and the caller's failed-wake cleanup proves it |
| + // stopped before the lease goes (P2-13). Under GenContainment |
| + // the error also says so (errGenUnproven), so that cleanup |
| + // asks for the proof with generations off as well. |
| + if gen != 0 { |
| + return up, "", fmt.Errorf("%w; %w: stop after failed health wait: %v", err, errGenUnproven, serr) |
| + } |
| + return up, "", err |
| } |
| return nil, "", err |
| } |
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go| index 96495f493..57a073eb7 100644 |
| --- a/internal/activator/backend_unit.go |
| +++ b/internal/activator/backend_unit.go |
| @@ -330,10 +330,9 @@ func (b *SplitBackend) StartGen(ctx context.Context, rt registry.Route, n uint64 |
| return g.StartGen(ctx, rt, n, perGen) |
| } |
| proc, ver, err := b.pick(rt).Start(ctx, rt) |
| - if err != nil { |
| - return proc, Gen{}, err |
| - } |
| - return proc, canonicalGen(rt, n, ver), nil |
| + // The handle goes back with an error too, as ExecBackend.StartGen's does |
| + // (P2-13). |
| + return proc, canonicalGen(rt, n, ver), err |
| } |
| // NextDueVia dispatches like NextDue; a side without generation support |
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go| index f6d3aa125..c6e5ac36e 100644 |
| --- a/internal/activator/claimwake.go |
| +++ b/internal/activator/claimwake.go |
| @@ -381,6 +381,19 @@ func (p *claimProcess) Stop(ctx context.Context) error { |
| return nil |
| } |
| +// provenStopped is a claims-on generation's termination proof (zero-downtime |
| +// spec §7.4): its Stop has completed, which takes KillAndEmpty on its own |
| +// child, the exit, the settle and the close. Stop is idempotent once |
| +// closed, so this retries a Stop that failed before. |
| +func (p *claimProcess) provenStopped(ctx context.Context) bool { |
| + if err := p.Stop(ctx); err != nil { |
| + return false |
| + } |
| + p.mu.Lock() |
| + defer p.mu.Unlock() |
| + return p.closed |
| +} |
| + |
| func sameAdmission(a, b hostclaim.Observed) bool { |
| if a.Absent || b.Absent || a.Host != b.Host || a.Claim == nil || b.Claim == nil || a.Claim.Holder == nil || b.Claim.Holder == nil { |
| return false |
diff --git a/internal/activator/generations.go b/internal/activator/generations.go| index ceba2c8dd..6d499315c 100644 |
| --- a/internal/activator/generations.go |
| +++ b/internal/activator/generations.go |
| @@ -60,6 +60,11 @@ type Gen struct { |
| // §7.3), captured when an attempt is admitted (plan 3). Zero for a |
| // live-link wake until a swap captures it from Build. |
| Tier cgroup.Limit |
| + // Worker marks a worker's generation (zero-downtime spec §7.9): it has |
| + // no generation child, and its termination proof is its host cgroup, |
| + // with or without a process handle (genTerminated). Plan 3 Task 11 |
| + // sets it. |
| + Worker bool |
| } |
| // GenSocketDir is where flat generation sockets live: /run/carlos/g. |
diff --git a/internal/activator/genservice.go b/internal/activator/genservice.go| index 5b6cbeac6..ec5c9c73b 100644 |
| --- a/internal/activator/genservice.go |
| +++ b/internal/activator/genservice.go |
| @@ -25,10 +25,10 @@ func (a *Service) startGen(ctx context.Context, rt registry.Route) (Proc, Gen, e |
| return g.StartGen(ctx, rt, n, a.gens) |
| } |
| proc, ver, err := a.backend.Start(ctx, rt) |
| - if err != nil { |
| - return proc, Gen{}, err |
| - } |
| - return proc, canonicalGen(rt, n, ver), nil |
| + // The handle goes back with an error too (P2-13): a failed start may |
| + // have left its process running, and the failed-wake cleanup proves it |
| + // stopped by the generation it names. |
| + return proc, canonicalGen(rt, n, ver), err |
| } |
| // AppSocket names host's CURRENT generation's socket, for traffic: the edge |
diff --git a/internal/activator/maintenance.go b/internal/activator/maintenance.go| index e1c9b2f36..aecb05bdf 100644 |
| --- a/internal/activator/maintenance.go |
| +++ b/internal/activator/maintenance.go |
| @@ -3,11 +3,13 @@ package activator |
| import ( |
| "context" |
| "errors" |
| + "log" |
| "net/http" |
| "time" |
| "github.com/carlosframework/platform/internal/controlname" |
| "github.com/carlosframework/platform/internal/edge" |
| + "github.com/carlosframework/platform/internal/registry" |
| ) |
| var ( |
| @@ -249,13 +251,38 @@ func maintenanceResponse(w http.ResponseWriter, r *http.Request) { |
| http.Error(w, "instance failed to wake", http.StatusServiceUnavailable) |
| } |
| -func (a *Service) cleanupFailedWake(ctx context.Context, w *wokenInstance, lease Lease) { |
| +// cleanupFailedWake stops a failed wake's process and gives its lease back, |
| +// recording the outcome on w.stopOutcome. rt is the route the wake spawned |
| +// against, which the termination proof needs. |
| +// |
| +// The lease is given back only over a generation proven terminated (spec |
| +// §7.4) whenever that proof is asked for: with generations on, or when the |
| +// start's own error said its kill was unproven (w.genUnproven, which only |
| +// GenContainment produces and which can be on with generations off). |
| +// Unproven, ownership and the lease are kept in unprovenWakes, the sweep |
| +// retries the stop and the proof (reapUnprovenWakesLocked), and no new wake |
| +// of the host starts until it holds. |
| +func (a *Service) cleanupFailedWake(ctx context.Context, rt registry.Route, w *wokenInstance, lease Lease) { |
| stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| defer cancel() |
| if w.proc != nil { |
| err := w.proc.Stop(stopCtx) |
| w.stopOutcome.mainStop, w.stopOutcome.mainExited = err == nil, w.proc.Exited() |
| } |
| + if (a.gens || w.genUnproven) && !a.failedWakeStopped(rt, w.gen, w.proc) { |
| + // Its leader did not exit (ErrStillRunning, a failed kill) or its |
| + // child is not proven empty. |
| + w.stopOutcome.mainExited = false |
| + w.stopOutcome.uncertain = true |
| + log.Printf("activator: %s: failed wake's generation %d is not proven stopped; keeping the lease until it is", rt.Host, w.gen.N) |
| + a.mu.Lock() |
| + if a.unprovenWakes == nil { |
| + a.unprovenWakes = make(map[string]*unprovenWake) |
| + } |
| + a.unprovenWakes[rt.Host] = &unprovenWake{rt: rt, gen: w.gen, proc: w.proc, lease: lease, rep: w.rep} |
| + a.mu.Unlock() |
| + return |
| + } |
| // A backend failure may conceal resources beyond the returned handle. |
| // Successful cleanup of that handle does not erase the uncertainty. |
| if w.rep != nil { |
diff --git a/internal/activator/opset.go b/internal/activator/opset.go| new file mode 100644 |
| index 000000000..adad828ef |
| --- /dev/null |
| +++ b/internal/activator/opset.go |
| @@ -0,0 +1,495 @@ |
| +// opset.go is the entry's operation set and its barrier (zero-downtime |
| +// deploys spec §7.4). The current generation stays where it always was |
| +// (e.gen, e.proc). The set holds every OTHER generation of the entry that |
| +// is spawning, health-checking or draining, and each is tracked until its |
| +// termination is proven. Every terminal stop joins the set before its |
| +// final checkpoint and lease release. |
| +package activator |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "log" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/instancectl" |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +type genRole int |
| + |
| +const ( |
| + roleSpawning genRole = iota // registered; StartGen has not returned |
| + roleSuccessor // spawned; health, settle or switch pending |
| + roleRetiring // was current; SIGTERM sent or about to be |
| +) |
| + |
| +func (r genRole) String() string { |
| + switch r { |
| + case roleSpawning: |
| + return "spawning" |
| + case roleSuccessor: |
| + return "successor" |
| + case roleRetiring: |
| + return "retiring" |
| + } |
| + return "unknown" |
| +} |
| + |
| +// genMember is one generation in an entry's operation set. Every field is |
| +// written under a.mu. Its owner (the Swap that created it) stops it, and |
| +// leaveSet removes it once its termination is proven. done closes at that |
| +// moment, and that is what the barrier waits on. |
| +type genMember struct { |
| + gen Gen |
| + role genRole |
| + proc Proc // nil while spawning |
| + cancel context.CancelFunc // cancels the attempt (spawning, successor); nil for retiring |
| + done chan struct{} |
| + // unproven: the owner's stop could not prove termination. The member |
| + // stays in the set and reapMembersLocked finishes it. |
| + unproven bool |
| + reaping bool |
| + retriedAt time.Time |
| +} |
| + |
| +// straySidecar is a legacy sidecar whose termination is unproven, with the |
| +// route its proof needs (Task 7). reaping marks a re-proof in flight. |
| +type straySidecar struct { |
| + proc Proc |
| + rt registry.Route |
| + reaping bool |
| +} |
| + |
| +// opSet is an entry's operation set. raised records that the parent's |
| +// limit is at twice the tier (spec §7.3), so the last member to leave |
| +// restores it. |
| +type opSet struct { |
| + members map[uint64]*genMember |
| + raised bool |
| +} |
| + |
| +// opJoinBound bounds the barrier's wait. A member's owner needs at most a |
| +// cancelled spawn (bounded by the control channel's deadline) plus one |
| +// bounded stop. A var so tests can shrink it. |
| +var opJoinBound = instancectl.DeadlineControl + actStopTimeout |
| + |
| +// addMemberLocked registers m. The caller holds a.mu. |
| +func (a *Service) addMemberLocked(e *actEntry, m *genMember) { |
| + if e.ops.members == nil { |
| + e.ops.members = make(map[uint64]*genMember) |
| + } |
| + e.ops.members[m.gen.N] = m |
| +} |
| + |
| +// isMember reports whether m is still in e's set. |
| +func (a *Service) isMember(e *actEntry, m *genMember) bool { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + return e != nil && e.ops.members[m.gen.N] == m |
| +} |
| + |
| +// joinOps is the barrier (spec §7.4). It cancels every member that is still |
| +// an attempt (spawning or successor) and waits, bounded by opJoinBound, for |
| +// every member to leave the set. It reports false when a member's |
| +// termination was not proven in time. stop() then keeps the lease, exactly |
| +// as it does for ErrStillRunning. An empty set returns true at once. |
| +func (a *Service) joinOps(e *actEntry) bool { |
| + a.mu.Lock() |
| + waits := make([]chan struct{}, 0, len(e.ops.members)) |
| + for _, m := range e.ops.members { |
| + if m.cancel != nil { |
| + m.cancel() |
| + } |
| + waits = append(waits, m.done) |
| + } |
| + a.mu.Unlock() |
| + if len(waits) == 0 { |
| + return true |
| + } |
| + bound := time.NewTimer(opJoinBound) |
| + defer bound.Stop() |
| + for _, w := range waits { |
| + select { |
| + case <-w: |
| + case <-bound.C: |
| + return false |
| + } |
| + } |
| + return true |
| +} |
| + |
| +// genKiller is plan 2's generation kill (ExecBackend.KillGen and its |
| +// SplitBackend forward): it kills what is left of generation n's own child |
| +// cgroup and returns once that child is empty. Plan 3 owns the name; plan 5 |
| +// uses it and does not declare its own. |
| +type genKiller interface { |
| + KillGen(ctx context.Context, rt registry.Route, n uint64) error |
| +} |
| + |
| +// stopMember stops m's process, bounded by actStopTimeout: SIGTERM, the |
| +// generation's own graceful drain, then SIGKILL (spec §7.5 step 6). It |
| +// reports whether termination is proven, which takes two things: the |
| +// leader has exited, and the generation's child cgroup is empty (spec |
| +// §7.3, §7.4). The flat childProc signals only its leader, so a process |
| +// the app forked can outlive it in g<n>. |
| +// |
| +// A member with no process never got one back from its spawn. That is not |
| +// proof: a nested start whose kill was unproven answers no process |
| +// (startUnitAt), so its generation's child is proven empty here instead. |
| +// Only a member that never had a number is trivially proven. |
| +func (a *Service) stopMember(e *actEntry, m *genMember) bool { |
| + a.mu.Lock() |
| + rt := e.route |
| + a.mu.Unlock() |
| + if m.proc == nil { |
| + return a.handlelessGenEmpty(rt, m.gen) |
| + } |
| + ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| + defer cancel() |
| + err := m.proc.Stop(ctx) |
| + if errors.Is(err, ErrStillRunning) { |
| + return false |
| + } |
| + if err != nil { |
| + log.Printf("activator: stop generation %d: %v", m.gen.N, err) |
| + } |
| + return a.genTerminated(rt, m.gen, m.proc) // swapsim:M7 |
| +} |
| + |
| +// handlelessGenEmpty is the termination proof for a generation with no |
| +// process handle (a member that never spawned, a failed wake whose start |
| +// returned none): its own child, proven empty through KillGen. A number of |
| +// 0 names no child, so it is trivially proven for a member (nothing was |
| +// ever asked to start under it). |
| +func (a *Service) handlelessGenEmpty(rt registry.Route, g Gen) bool { |
| + if g.N == 0 { |
| + return true |
| + } |
| + return a.genChildEmpty(rt, &genMember{gen: g}) |
| +} |
| + |
| +// genTerminated is the ONE termination proof for a generation (spec §7.3, |
| +// §7.4): its leader has exited AND its child is proven empty. Every path |
| +// that would give up a generation's lease or its place in the operation |
| +// set asks it: a member's stop, the reaper, the current generation's stop, |
| +// a pending stop's finish, a failed wake and its retained retry. A nil |
| +// process never ran. |
| +func (a *Service) genTerminated(rt registry.Route, g Gen, p Proc) bool { |
| + if p == nil { |
| + if g.Worker { |
| + // A worker's host cgroup is proven whether or not a process |
| + // handle is left: a dead leader's descendants live there (Task 11). |
| + return a.genChildEmpty(rt, &genMember{gen: g}) |
| + } |
| + return true |
| + } |
| + if !p.Exited() { |
| + return false |
| + } |
| + return a.genChildEmpty(rt, &genMember{gen: g, proc: p}) |
| +} |
| + |
| +// genChildEmpty proves m's generation child empty, after its leader has |
| +// exited, through plan 2's KillGen: whatever is left in g<n> (or the |
| +// nested carlos-instance-gen@<id>:<n> unit) is killed and the child is |
| +// waited empty. That kill is the only one that can reach a descendant the |
| +// leader left behind, and it touches no other generation. |
| +// |
| +// A backend with no generation kill (a test double), or one that answers |
| +// errNoGenContainment, has no generation children: the generation ran in |
| +// the host cgroup, and the terminal stop's own stop evidence on that cgroup |
| +// is the proof (plan 2, "stop evidence on the parent"). Any other error is |
| +// an unproven termination. |
| +func (a *Service) genChildEmpty(rt registry.Route, m *genMember) bool { |
| + if cp, ok := m.proc.(*claimProcess); ok { |
| + // A claims-on generation (Task 10): its own Stop is the proof, and |
| + // only a Stop that completed (KillAndEmpty, the exit, the settle and |
| + // the close) counts. A leader that exited after KillAndEmpty failed |
| + // is not terminated. |
| + ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl) |
| + defer cancel() |
| + return cp.provenStopped(ctx) |
| + } |
| + k, ok := a.backend.(genKiller) |
| + if !ok || m.gen.N == 0 { |
| + return true |
| + } |
| + ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl) |
| + defer cancel() |
| + err := k.KillGen(ctx, rt, m.gen.N) |
| + if err == nil || errors.Is(err, errNoGenContainment) { |
| + return true |
| + } |
| + log.Printf("activator: %s: generation %d's child is not proven empty: %v", rt.Host, m.gen.N, err) |
| + return false |
| +} |
| + |
| +// dropMember stops m and takes it out of e's set once its termination is |
| +// proven. It reports whether termination was proven. An unproven member |
| +// stays in the set, marked, and the sweep's reaper finishes it. |
| +func (a *Service) dropMember(e *actEntry, m *genMember) bool { |
| + proven := a.stopMember(e, m) // swapsim:M4 |
| + if !proven { |
| + a.mu.Lock() |
| + m.unproven, m.cancel, m.retriedAt = true, nil, a.now() |
| + a.mu.Unlock() |
| + log.Printf("activator: generation %d of %s did not prove its exit; keeping it in the operation set", m.gen.N, e.host) |
| + return false |
| + } |
| + a.leaveSet(e, m) |
| + return true |
| +} |
| + |
| +// leaveSet removes a member whose termination is proven. When it is the |
| +// last member and the parent was raised, the parent is restored FIRST, |
| +// while the member still holds the set. That way no new attempt can raise |
| +// the parent again in between (spec §7.3: the restore happens after the |
| +// retiring child is empty). The restore is to the CURRENT generation's |
| +// captured tier (spec §7.3): the incumbent's after a failed successor, the |
| +// successor's after a switch. done closes last. |
| +func (a *Service) leaveSet(e *actEntry, m *genMember) { |
| + a.mu.Lock() |
| + restore := e.ops.raised && len(e.ops.members) == 1 && e.ops.members[m.gen.N] == m |
| + rt, tier := e.route, e.gen.Tier |
| + a.mu.Unlock() |
| + if restore { |
| + if gb, ok := a.backend.(SwapBackend); ok { |
| + rctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl) |
| + if err := gb.RestoreGenParent(rctx, rt, tier); err != nil { |
| + log.Printf("activator: restore parent limit for %s: %v", rt.Host, err) |
| + } |
| + cancel() |
| + } |
| + } |
| + a.mu.Lock() |
| + if restore { |
| + e.ops.raised = false |
| + } |
| + if e.ops.members[m.gen.N] == m { |
| + delete(e.ops.members, m.gen.N) |
| + } |
| + a.mu.Unlock() |
| + close(m.done) |
| +} |
| + |
| +// reapMembersLocked finishes members whose owner could not prove their stop. |
| +// Once a member's process has exited (or it never had one), the reaper |
| +// proves its generation child empty (reapExited) and only then takes it out |
| +// of the set. Until the leader exits it re-issues Stop every |
| +// actStopTimeout, the cadence finishPendingStops keeps for the main |
| +// process. The caller holds a.mu. |
| +func (a *Service) reapMembersLocked(e *actEntry) { |
| + for _, m := range e.ops.members { |
| + if !m.unproven || m.reaping { |
| + continue |
| + } |
| + if m.proc == nil || m.proc.Exited() { |
| + m.reaping = true |
| + go a.reapExited(e, m) |
| + continue |
| + } |
| + if a.now().Sub(m.retriedAt) < actStopTimeout { |
| + continue |
| + } |
| + m.reaping, m.retriedAt = true, a.now() |
| + go func(m *genMember) { |
| + ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| + err := m.proc.Stop(ctx) |
| + cancel() |
| + if err != nil && !errors.Is(err, ErrStillRunning) { |
| + log.Printf("activator: retry stop of generation %d of %s: %v", m.gen.N, e.host, err) |
| + } |
| + a.mu.Lock() |
| + m.reaping = false |
| + a.mu.Unlock() |
| + }(m) |
| + } |
| +} |
| + |
| +// reapExited finishes an unproven member whose leader has exited, or that |
| +// never had one: it leaves the set only once its generation child is |
| +// proven empty. A failed proof leaves it marked, and the next sweep tries |
| +// the kill again. |
| +func (a *Service) reapExited(e *actEntry, m *genMember) { |
| + a.mu.Lock() |
| + rt := e.route |
| + a.mu.Unlock() |
| + var proven bool |
| + if m.proc == nil { |
| + proven = a.handlelessGenEmpty(rt, m.gen) |
| + } else { |
| + proven = a.genTerminated(rt, m.gen, m.proc) |
| + } |
| + if proven { |
| + a.leaveSet(e, m) |
| + return |
| + } |
| + a.mu.Lock() |
| + m.reaping, m.retriedAt = false, a.now() |
| + a.mu.Unlock() |
| +} |
| + |
| +// The CURRENT generation is held to the same proof (spec §7.4: "each is |
| +// tracked until its termination is proven"). stop() and a pending stop's |
| +// finish prove its child empty before the final checkpoint and the lease |
| +// release, and a failed wake proves its generation's child empty before it |
| +// gives the lease back. |
| + |
| +// releasable is THE retain-until-proven rule for an entry's lease (spec |
| +// §7.4): a stop finishes, and its lease is released, only when every |
| +// member has left the operation set, no stray sidecar of the host is |
| +// tracked (Task 7), and the current generation has passed genTerminated |
| +// (its leader exited and its child, or a worker's host cgroup, empty). It |
| +// runs without a.mu held; genTerminated may take DeadlineControl. |
| +func (a *Service) releasable(e *actEntry) bool { |
| + a.mu.Lock() |
| + members, strays := len(e.ops.members), len(a.straySidecars[e.host]) |
| + rt, g, p := e.route, e.gen, e.proc |
| + a.mu.Unlock() |
| + if members > 0 || strays > 0 { |
| + return false |
| + } |
| + return a.genTerminated(rt, g, p) |
| +} |
| + |
| +// keepPendingLocked leaves a stop pending with its lease held: the entry |
| +// stays in stopping, and finishPendingStops finishes it once termination |
| +// is proven. stopRetriedAt is this attempt, so the sweep's re-stop cadence |
| +// runs a full actStopTimeout after it. The caller holds a.mu. |
| +func (a *Service) keepPendingLocked(e *actEntry, outcome *stopObservation) { |
| + e.pendingStop = true |
| + e.stopOutcome = outcome |
| + e.stopRetriedAt = a.now() |
| +} |
| + |
| +// reapCurrentGen finishes a pending stop whose main process has exited, once |
| +// the retain-until-proven rule holds (releasable). A failed proof leaves the |
| +// stop pending, and the next sweep tries again. |
| +func (a *Service) reapCurrentGen(e *actEntry) { |
| + if !a.releasable(e) { |
| + a.mu.Lock() |
| + e.genReaping = false |
| + a.mu.Unlock() |
| + return |
| + } |
| + a.mu.Lock() |
| + e.genReaping, e.pendingStop = false, false |
| + outcome, p := e.stopOutcome, e.proc |
| + a.mu.Unlock() |
| + if outcome == nil { |
| + outcome = &stopObservation{} |
| + } |
| + if p == nil { |
| + a.finishStop(e, outcome) // only the barrier left it pending |
| + return |
| + } |
| + a.finishPendingStop(e, outcome) |
| +} |
| + |
| +// unprovenWake is a failed wake whose generation is not proven stopped |
| +// (spec §7.4). Its lease (and replica, if it got one) is kept until the |
| +// sweep proves the generation stopped; until then no new wake of the host |
| +// starts. busy marks a retry in flight. |
| +type unprovenWake struct { |
| + rt registry.Route |
| + gen Gen |
| + proc Proc |
| + lease Lease |
| + rep Rep |
| + busy bool |
| +} |
| + |
| +// errWakeUnproven refuses a wake while the host's last failed wake is not |
| +// proven stopped. |
| +var errWakeUnproven = errors.New("the last failed wake's generation is not proven stopped") |
| + |
| +// genStartUnproven reports whether a generation start's error says the |
| +// start failed and its kill did not prove the generation gone: plan 2's |
| +// errGenUnproven (startUnitAt) or root's instancectl.ErrGenUnproven. The |
| +// generation may still be running, so whatever the wake holds for the host |
| +// is kept. Both exist only under GenContainment, which can be on while |
| +// a.gens is off, so callers never gate this on a.gens. |
| +func genStartUnproven(err error) bool { |
| + return errors.Is(err, errGenUnproven) || errors.Is(err, instancectl.ErrGenUnproven) |
| +} |
| + |
| +// failedWakeStopped is the termination proof for a failed wake's |
| +// generation. With a process it is genTerminated. Without one (the start |
| +// answered none, as startUnitAt does when its kill was unproven) it is the |
| +// generation's own child, proven empty through KillGen, and never the |
| +// missing handle: a retained wake with no number cannot be proven at all |
| +// and stays retained. |
| +func (a *Service) failedWakeStopped(rt registry.Route, g Gen, p Proc) bool { |
| + if p != nil { |
| + return a.genTerminated(rt, g, p) |
| + } |
| + if g.N == 0 { |
| + log.Printf("activator: %s: failed wake left no process handle and no generation number; it cannot be proven stopped, so its lease stays held", rt.Host) |
| + return false |
| + } |
| + return a.handlelessGenEmpty(rt, g) |
| +} |
| + |
| +// reapUnprovenWakesLocked retries each unproven failed wake's proof and, |
| +// once it holds, gives the lease back. The caller holds a.mu. |
| +func (a *Service) reapUnprovenWakesLocked() { |
| + for host, u := range a.unprovenWakes { |
| + if u.busy { |
| + continue |
| + } |
| + u.busy = true |
| + go a.reapUnprovenWake(host, u) |
| + } |
| +} |
| + |
| +// reapUnprovenWake is one retry of a retained failed wake. A leader that is |
| +// still alive is stopped again, bounded. Once the generation is proven |
| +// stopped, the replica is closed (or the lease released) and the retained |
| +// wake is forgotten, which reopens the host to wakes. |
| +// |
| +// 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. |
| +func (a *Service) reapUnprovenWake(host string, u *unprovenWake) { |
| + if u.proc != nil && !u.proc.Exited() { |
| + ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| + err := u.proc.Stop(ctx) |
| + cancel() |
| + if err != nil && !errors.Is(err, ErrStillRunning) { |
| + 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) { |
| + a.mu.Lock() |
| + u.busy = false |
| + a.mu.Unlock() |
| + return |
| + } |
| + outcome := &stopObservation{mainStop: true, mainExited: true, uncertain: true} |
| + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| + var err error |
| + if u.rep != nil { |
| + err = u.rep.Close(ctx) |
| + outcome.replicaClosed = err == nil |
| + } else if u.lease != nil { |
| + err = u.lease.Release(ctx) |
| + } |
| + 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) |
| + } else { |
| + log.Printf("activator: %s: failed wake's generation %d is proven stopped; lease given back", host, u.gen.N) |
| + } |
| + a.mu.Lock() |
| + a.rememberStopLocked(host, outcome) |
| + if a.unprovenWakes[host] == u { |
| + delete(a.unprovenWakes, host) |
| + } |
| + u.busy = false |
| + a.mu.Unlock() |
| +} |
diff --git a/internal/activator/opset_test.go b/internal/activator/opset_test.go| new file mode 100644 |
| index 000000000..9a13fa134 |
| --- /dev/null |
| +++ b/internal/activator/opset_test.go |
| @@ -0,0 +1,711 @@ |
| +package activator |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "os/exec" |
| + "slices" |
| + "sync" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/instancectl" |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +// plantMember puts a generation into host's operation set the way Swap does |
| +// (Task 4). A successor waits for the barrier's cancel and is then stopped by |
| +// its owner. A retiring generation is stopped by its owner at once. |
| +func plantMember(t *testing.T, a *Service, host string, role genRole, p *fakeProc, n uint64) *genMember { |
| + t.Helper() |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + t.Cleanup(cancel) |
| + m := &genMember{gen: Gen{N: n, Build: "v2"}, role: role, proc: p, done: make(chan struct{})} |
| + if role != roleRetiring { |
| + m.cancel = cancel |
| + } |
| + a.mu.Lock() |
| + e := a.entries[host] |
| + if e == nil { |
| + a.mu.Unlock() |
| + t.Fatalf("no entry for %s", host) |
| + } |
| + a.addMemberLocked(e, m) |
| + a.mu.Unlock() |
| + go func() { |
| + if role != roleRetiring { |
| + <-ctx.Done() |
| + } |
| + a.dropMember(e, m) |
| + }() |
| + return m |
| +} |
| + |
| +type barrierEnv struct { |
| + a *Service |
| + b *fakeBackend |
| + clk *fakeClock |
| + host string |
| +} |
| + |
| +func awakeForBarrier(t *testing.T, swap bool) barrierEnv { |
| + t.Helper() |
| + a, b, clk, _ := testActivator(t, time.Hour) |
| + a.gens = swap |
| + host := "barrier.example.com" |
| + arrangeAwake(t, a, testRoute(host)) |
| + return barrierEnv{a, b, clk, host} |
| +} |
| + |
| +// stopSignal is an EventSink that closes began when an entry's stop emits |
| +// "instance stopping": the stop goroutine has started, and from there it |
| +// reaches the barrier without blocking on anything else. |
| +type stopSignal struct { |
| + once sync.Once |
| + began chan struct{} |
| +} |
| + |
| +func (s *stopSignal) App(string, string, string, string) {} |
| +func (s *stopSignal) Event(_, _, _, _, msg string) { |
| + if msg == "instance stopping" { |
| + s.once.Do(func() { close(s.began) }) |
| + } |
| +} |
| + |
| +// Every terminal stop cancels and joins the whole operation set before the |
| +// final checkpoint and the lease release (spec §7.4). The member's stop is |
| +// held open by a gate; until it closes, neither the instance's own stop |
| +// nor the lease release may happen. |
| +func TestSwapWitnessBarrierJoinsBeforeTheLease(t *testing.T) { |
| + stops := []struct { |
| + name string |
| + stop func(x barrierEnv) |
| + rewake bool |
| + }{ |
| + {"hibernate", func(x barrierEnv) { x.clk.advance(2 * time.Hour); x.a.sweepOnce() }, true}, |
| + {"lease lost", func(x barrierEnv) { x.b.lastRep().setErr(errors.New("renewal failed")); x.a.sweepOnce() }, true}, |
| + {"restart", func(x barrierEnv) { x.a.Bounce(x.host) }, true}, |
| + {"maintenance fence", func(x barrierEnv) { |
| + go func() { |
| + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer cancel() |
| + x.a.FenceAndStop(ctx, x.host, 1) //nolint:errcheck |
| + }() |
| + }, false}, |
| + {"move or park retire", func(x barrierEnv) { x.a.RetireExcept(map[string]bool{}) }, true}, |
| + {"shutdown", func(x barrierEnv) { go x.a.Shutdown(context.Background()) }, false}, |
| + } |
| + for _, tc := range stops { |
| + for _, role := range []genRole{roleSuccessor, roleRetiring} { |
| + t.Run(tc.name+"/"+role.String(), func(t *testing.T) { |
| + x := awakeForBarrier(t, true) |
| + sig := &stopSignal{began: make(chan struct{})} |
| + x.a.Logs = sig |
| + gate := make(chan struct{}) |
| + member := &fakeProc{label: "gen:9", rec: x.b, stopGate: gate} |
| + plantMember(t, x.a, x.host, role, member, 9) |
| + incumbent, rep := x.b.lastProc(), x.b.lastRep() |
| + tc.stop(x) |
| + select { |
| + case <-sig.began: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("the stop never began") |
| + } |
| + // The stop has begun; nothing between its first line and the |
| + // barrier blocks. There is no signal at the barrier itself, so |
| + // this short wait covers those few instructions. |
| + time.Sleep(50 * time.Millisecond) |
| + if rep.isClosed() || incumbent.isStopped() { |
| + t.Fatal("the stop went past the barrier while a generation was still draining") |
| + } |
| + close(gate) |
| + waitFor(t, "the lease release", rep.isClosed) |
| + ev := x.b.events() |
| + if i, j := slices.Index(ev, "stop:gen:9"), slices.Index(ev, "stop:instance"); i < 0 || j < 0 || i > j { |
| + t.Fatalf("order %v: the member must stop before the instance", ev) |
| + } |
| + if tc.rewake { |
| + // #231: Owns stays true while an entry drains, so the stop is |
| + // proven by a fresh spawn instead. |
| + arrangeAwake(t, x.a, testRoute(x.host)) |
| + if got := x.b.startCount(); got != 2 { |
| + t.Fatalf("Start calls = %d after the stop, want 2", got) |
| + } |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +// A member whose termination cannot be proven (ErrStillRunning) keeps the |
| +// lease, exactly as an unproven main process does (spec §7.4). The reaper |
| +// finishes the member once it exits, and then the stop completes. |
| +func TestBarrierKeepsTheLeaseWhileAGenerationIsUnproven(t *testing.T) { |
| + x := awakeForBarrier(t, true) |
| + old := opJoinBound |
| + opJoinBound = 100 * time.Millisecond |
| + t.Cleanup(func() { opJoinBound = old }) |
| + member := &fakeProc{label: "gen:9", rec: x.b, stuck: true} |
| + plantMember(t, x.a, x.host, roleRetiring, member, 9) |
| + waitFor(t, "the member marked unproven", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + m := x.a.entries[x.host].ops.members[9] |
| + return m != nil && m.unproven |
| + }) |
| + rep := x.b.lastRep() |
| + x.a.Bounce(x.host) |
| + waitFor(t, "the stop to pend", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + e := x.a.entries[x.host] |
| + return e != nil && e.pendingStop |
| + }) |
| + if rep.isClosed() { |
| + t.Fatal("the lease was released over an unproven generation") |
| + } |
| + member.exit() |
| + x.a.sweepOnce() // reaps the member |
| + waitFor(t, "the member reaped", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + e := x.a.entries[x.host] |
| + return e == nil || len(e.ops.members) == 0 |
| + }) |
| + x.a.sweepOnce() // finishes the stop |
| + waitFor(t, "the lease release", rep.isClosed) |
| + arrangeAwake(t, x.a, testRoute(x.host)) |
| + if got := x.b.startCount(); got != 2 { |
| + t.Fatalf("Start calls = %d, want 2", got) |
| + } |
| +} |
| + |
| +// killGenBackend is fakeBackend plus plan 2's KillGen. err, while set, is |
| +// what every kill answers: a generation child that still holds a process. |
| +type killGenBackend struct { |
| + *fakeBackend |
| + kmu sync.Mutex |
| + err error |
| + kills []uint64 |
| +} |
| + |
| +func (k *killGenBackend) KillGen(_ context.Context, _ registry.Route, n uint64) error { |
| + k.kmu.Lock() |
| + defer k.kmu.Unlock() |
| + k.kills = append(k.kills, n) |
| + return k.err |
| +} |
| + |
| +func (k *killGenBackend) setErr(err error) { |
| + k.kmu.Lock() |
| + k.err = err |
| + k.kmu.Unlock() |
| +} |
| + |
| +func (k *killGenBackend) killed() []uint64 { |
| + k.kmu.Lock() |
| + defer k.kmu.Unlock() |
| + return slices.Clone(k.kills) |
| +} |
| + |
| +// A generation's leader can exit while a process it forked lives on in its |
| +// child cgroup. The leader's exit is not termination (spec §7.3, §7.4). |
| +// After the graceful drain the member's own child is killed and proven |
| +// empty through plan 2's KillGen. While that proof fails, the member stays |
| +// in the set and the lease stays held, exactly as for ErrStillRunning. The |
| +// reaper retries the kill and finishes the stop once the child is empty. |
| +// |
| +// Without the KillGen step, stopMember accepts Exited() alone: the member |
| +// leaves the set at once, "marked unproven" never happens, and the Bounce |
| +// releases the lease over the surviving descendant. |
| +func TestSwapWitnessMemberStopProvesTheGenerationChildEmpty(t *testing.T) { |
| + x := awakeForBarrier(t, true) |
| + old := opJoinBound |
| + opJoinBound = 100 * time.Millisecond |
| + t.Cleanup(func() { opJoinBound = old }) |
| + kb := &killGenBackend{fakeBackend: x.b, err: errors.New("g9 still holds a process")} |
| + x.a.backend = kb |
| + member := &fakeProc{label: "gen:9", rec: x.b} // its leader exits on SIGTERM |
| + plantMember(t, x.a, x.host, roleRetiring, member, 9) |
| + waitFor(t, "the member marked unproven", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + m := x.a.entries[x.host].ops.members[9] |
| + return m != nil && m.unproven |
| + }) |
| + if !member.isStopped() || !member.Exited() { |
| + t.Fatal("the graceful stop never ran, or the leader did not exit") |
| + } |
| + if got := kb.killed(); !slices.Equal(got, []uint64{9}) { |
| + t.Fatalf("KillGen calls = %v, want [9] (the member's own generation)", got) |
| + } |
| + rep := x.b.lastRep() |
| + x.a.Bounce(x.host) |
| + waitFor(t, "the stop to pend", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + e := x.a.entries[x.host] |
| + return e != nil && e.pendingStop |
| + }) |
| + if rep.isClosed() { |
| + t.Fatal("the lease was released while generation 9's child still held a process") |
| + } |
| + x.a.sweepOnce() // the reaper retries the kill, which still fails |
| + waitFor(t, "the reaper's retry to finish", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + m := x.a.entries[x.host].ops.members[9] |
| + return len(kb.killed()) >= 2 && m != nil && !m.reaping |
| + }) |
| + if rep.isClosed() { |
| + t.Fatal("the lease was released on a failed retry") |
| + } |
| + kb.setErr(nil) |
| + x.a.sweepOnce() // the kill proves the child empty now |
| + waitFor(t, "the member reaped", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + e := x.a.entries[x.host] |
| + return e == nil || len(e.ops.members) == 0 |
| + }) |
| + x.a.sweepOnce() // finishes the stop |
| + waitFor(t, "the lease release", rep.isClosed) |
| +} |
| + |
| +// A backend with no generation children (containment off answers |
| +// errNoGenContainment) has nothing more to prove than the leader's exit: |
| +// the generation shares the host cgroup, whose emptiness the terminal |
| +// stop's own evidence proves. |
| +func TestMemberStopWithoutContainmentAcceptsTheLeaderExit(t *testing.T) { |
| + x := awakeForBarrier(t, true) |
| + kb := &killGenBackend{fakeBackend: x.b, err: fmt.Errorf("wrapped: %w", errNoGenContainment)} |
| + x.a.backend = kb |
| + m := plantMember(t, x.a, x.host, roleRetiring, &fakeProc{label: "gen:9", rec: x.b}, 9) |
| + select { |
| + case <-m.done: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("the member never left the set") |
| + } |
| +} |
| + |
| +// A member that never got a process (its spawn failed with its kill |
| +// unproven, as startUnitAt answers) is not proven by the missing handle: |
| +// its generation's child is, through KillGen (P2-13 (f)). |
| +func TestMemberStopWithoutAProcessProvesItsChild(t *testing.T) { |
| + x := awakeForBarrier(t, true) |
| + kb := &killGenBackend{fakeBackend: x.b, err: errors.New("g12 still holds a process")} |
| + x.a.backend = kb |
| + m := &genMember{gen: Gen{N: 12, Build: "v2"}, role: roleSpawning, done: make(chan struct{})} |
| + x.a.mu.Lock() |
| + e := x.a.entries[x.host] |
| + x.a.addMemberLocked(e, m) |
| + x.a.mu.Unlock() |
| + if x.a.dropMember(e, m) { |
| + t.Fatal("a member with no process and an unproven child was taken as terminated") |
| + } |
| + if got := kb.killed(); !slices.Equal(got, []uint64{12}) { |
| + t.Fatalf("KillGen calls = %v, want [12]", got) |
| + } |
| + kb.setErr(nil) |
| + x.a.sweepOnce() |
| + select { |
| + case <-m.done: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("the member never left once its child was empty") |
| + } |
| +} |
| + |
| +// stubbornProc is a main process whose Stop fails with an error other than |
| +// ErrStillRunning (a claims-on KillAndEmpty error) and whose leader lives |
| +// on until exit. stops counts Stop calls. |
| +type stubbornProc struct { |
| + exited atomic.Bool |
| + stops atomic.Int32 |
| +} |
| + |
| +func (p *stubbornProc) Stop(context.Context) error { |
| + p.stops.Add(1) |
| + return errors.New("cgroup.kill: the child is not empty") |
| +} |
| +func (p *stubbornProc) Exited() bool { return p.exited.Load() } |
| +func (p *stubbornProc) Crashed() bool { return false } |
| + |
| +// THE retain-until-proven rule (spec §7.4): a stop whose main process is |
| +// still alive after a failed kill keeps its lease. Only ErrStillRunning |
| +// kept it before; any other Stop error, with the leader alive, skipped the |
| +// proof (it ran only for an exited leader) and released the lease. |
| +func TestAStopKeepsTheLeaseWhileItsLeaderLivesOn(t *testing.T) { |
| + x := awakeForBarrier(t, true) |
| + p := &stubbornProc{} |
| + x.a.mu.Lock() |
| + x.a.entries[x.host].proc = p |
| + x.a.mu.Unlock() |
| + rep := x.b.lastRep() |
| + x.a.Bounce(x.host) |
| + waitFor(t, "the stop to pend", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + e := x.a.entries[x.host] |
| + return e != nil && e.pendingStop |
| + }) |
| + // The leader is alive and the retry cadence has not come round (the |
| + // clock is fake), so this sweep starts nothing: the stop stays pending. |
| + x.a.sweepOnce() |
| + x.a.mu.Lock() |
| + e := x.a.entries[x.host] |
| + idle := e != nil && e.pendingStop && !e.genReaping && !e.stopRetrying |
| + x.a.mu.Unlock() |
| + if !idle || rep.isClosed() { |
| + t.Fatal("the lease was released while the leader still ran after a failed kill") |
| + } |
| + p.exited.Store(true) |
| + x.a.sweepOnce() |
| + waitFor(t, "the lease release", rep.isClosed) |
| +} |
| + |
| +// A stop keeps its lease while a stray sidecar of the host is tracked |
| +// (spec §7.4): its process may still run against the database. The lease |
| +// went with the stop before. |
| +func TestAStopKeepsTheLeaseWhileAStraySidecarRuns(t *testing.T) { |
| + x := awakeForBarrier(t, true) |
| + stray := &fakeProc{label: "sidecar", rec: x.b, stuck: true} |
| + x.a.mu.Lock() |
| + x.a.straySidecars = map[string][]*straySidecar{x.host: {{proc: stray, rt: testRoute(x.host)}}} |
| + x.a.mu.Unlock() |
| + rep := x.b.lastRep() |
| + x.a.Bounce(x.host) |
| + waitFor(t, "the stop to pend", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + e := x.a.entries[x.host] |
| + return e != nil && e.pendingStop |
| + }) |
| + if rep.isClosed() { |
| + t.Fatal("the lease was released while a stray sidecar ran") |
| + } |
| + x.a.mu.Lock() |
| + delete(x.a.straySidecars, x.host) // proven gone (Task 7's reaper does this) |
| + x.a.mu.Unlock() |
| + x.a.sweepOnce() |
| + waitFor(t, "the lease release", rep.isClosed) |
| +} |
| + |
| +// failingContainment is a claims-on child whose KillAndEmpty fails until |
| +// heal is called. |
| +type failingContainment struct { |
| + mu sync.Mutex |
| + bad bool |
| +} |
| + |
| +func (c *failingContainment) KillAndEmpty(context.Context) error { |
| + c.mu.Lock() |
| + defer c.mu.Unlock() |
| + if c.bad { |
| + return errors.New("cgroup.kill: the child is not empty") |
| + } |
| + return nil |
| +} |
| +func (c *failingContainment) Close() error { return nil } |
| +func (c *failingContainment) heal() { |
| + c.mu.Lock() |
| + c.bad = false |
| + c.mu.Unlock() |
| +} |
| + |
| +// A claims-on generation is terminated only when its own Stop completed |
| +// (spec §7.4): KillAndEmpty on its child included. Here the leader has |
| +// exited but KillAndEmpty fails, so the member stays in the set, unproven, |
| +// until the child is empty. Treating any claimProcess as proven accepted |
| +// the leader's exit alone. |
| +func TestAClaimsOnMemberNeedsItsChildEmptied(t *testing.T) { |
| + x := awakeForBarrier(t, true) |
| + cmd := exec.Command("true") |
| + if err := cmd.Run(); err != nil { |
| + t.Skipf("no true(1): %v", err) |
| + } |
| + done := make(chan struct{}) |
| + close(done) // the leader has exited |
| + cont := &failingContainment{bad: true} |
| + p := &claimProcess{cmd: cmd, prepared: cont, done: done} |
| + m := &genMember{gen: Gen{N: 11, Build: "v2"}, role: roleRetiring, proc: p, done: make(chan struct{})} |
| + x.a.mu.Lock() |
| + e := x.a.entries[x.host] |
| + x.a.addMemberLocked(e, m) |
| + x.a.mu.Unlock() |
| + if x.a.dropMember(e, m) { |
| + t.Fatal("a claims-on generation whose KillAndEmpty failed was taken as terminated") |
| + } |
| + cont.heal() |
| + x.a.sweepOnce() |
| + select { |
| + case <-m.done: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("the member never left once its child was empty") |
| + } |
| +} |
| + |
| +// The current generation is held to the same proof (spec §7.4): a terminal |
| +// stop whose main process has exited keeps the lease while that |
| +// generation's child is not proven empty, and finishes once it is. A stop |
| +// that accepts the leader's exit alone releases the lease over a surviving |
| +// descendant. This is the witness for mutation M8. |
| +func TestSwapWitnessStopProvesTheCurrentGenerationEmpty(t *testing.T) { |
| + x := awakeForBarrier(t, true) |
| + kb := &killGenBackend{fakeBackend: x.b, err: errors.New("the child still holds a process")} |
| + x.a.backend = kb |
| + x.a.mu.Lock() |
| + x.a.entries[x.host].gen.N = 1 << 41 // the woken generation, as plan 1's wake records it |
| + x.a.mu.Unlock() |
| + rep := x.b.lastRep() |
| + x.a.Bounce(x.host) |
| + waitFor(t, "the stop to pend", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + e := x.a.entries[x.host] |
| + return e != nil && e.pendingStop |
| + }) |
| + if !x.b.lastProc().isStopped() { |
| + t.Fatal("the graceful stop never ran") |
| + } |
| + kills := len(kb.killed()) |
| + x.a.sweepOnce() // retries the proof, which still fails |
| + waitFor(t, "the retried proof to finish", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + e := x.a.entries[x.host] |
| + return len(kb.killed()) > kills && e != nil && !e.genReaping |
| + }) |
| + if rep.isClosed() { |
| + t.Fatal("the lease was released while the current generation's child still held a process") |
| + } |
| + kb.setErr(nil) |
| + x.a.sweepOnce() |
| + waitFor(t, "the lease release", rep.isClosed) |
| + if k := kb.killed(); len(k) == 0 || k[len(k)-1] != 1<<41 { |
| + t.Fatalf("KillGen calls = %v, want the current generation's", k) |
| + } |
| +} |
| + |
| +// With swaps on, Shutdown covers a waking entry (spec §7.4): it cancels the |
| +// wake and waits for it. Its child is stopped and its lease given back |
| +// before Shutdown returns. |
| +func TestSwapWitnessShutdownJoinsAWakingEntry(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + a.gens = true |
| + gate := make(chan struct{}) |
| + b.startGate = gate |
| + host := "waking.example.com" |
| + go func() { |
| + if release, err := a.beginRequest(context.Background(), testRoute(host)); err == nil { |
| + release() |
| + } |
| + }() |
| + waitFor(t, "the wake to reach Start", func() bool { return b.startCount() == 1 }) |
| + done := make(chan struct{}) |
| + go func() { a.Shutdown(context.Background()); close(done) }() |
| + select { |
| + case <-done: |
| + t.Fatal("Shutdown returned while a wake was still spawning") |
| + case <-time.After(100 * time.Millisecond): |
| + } |
| + close(gate) |
| + select { |
| + case <-done: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("Shutdown never returned") |
| + } |
| + if !b.lastProc().isStopped() { |
| + t.Fatal("the woken child outlived Shutdown") |
| + } |
| + if got := b.releaseCount(); got != 1 { |
| + t.Fatalf("lease releases = %d, want 1", got) |
| + } |
| +} |
| + |
| +// Flag off: Shutdown keeps today's rule and does not wait for a waking entry. |
| +func TestShutdownLeavesAWakingEntryWithSwapOff(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + gate := make(chan struct{}) |
| + b.startGate = gate |
| + host := "waking.example.com" |
| + go func() { |
| + if release, err := a.beginRequest(context.Background(), testRoute(host)); err == nil { |
| + release() |
| + } |
| + }() |
| + waitFor(t, "the wake to reach Start", func() bool { return b.startCount() == 1 }) |
| + done := make(chan struct{}) |
| + go func() { a.Shutdown(context.Background()); close(done) }() |
| + select { |
| + case <-done: |
| + case <-time.After(time.Second): |
| + t.Fatal("Shutdown waited for a waking entry with swaps off") |
| + } |
| + close(gate) |
| + waitFor(t, "the wake to finish", func() bool { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + e := a.entries[host] |
| + return e != nil && e.state == actAwake |
| + }) |
| + a.Bounce(host) |
| + waitFor(t, "cleanup", func() bool { return a.entryCount() == 0 }) |
| +} |
| + |
| +// unprovenStartBackend is a GenBackend whose next StartGen fails the way a |
| +// generation start fails when its kill cannot be proven (plan 2, |
| +// startUnitAt): fail, while set, answers that one call. Every later call |
| +// spawns through the fake's Start. KillGen comes from killGenBackend. |
| +type unprovenStartBackend struct { |
| + *killGenBackend |
| + smu sync.Mutex |
| + fail func(n uint64) (Proc, error) |
| + gens []uint64 // every generation number StartGen was handed |
| +} |
| + |
| +func (u *unprovenStartBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, _ bool) (Proc, Gen, error) { |
| + u.smu.Lock() |
| + fail := u.fail |
| + u.fail = nil |
| + u.gens = append(u.gens, n) |
| + u.smu.Unlock() |
| + if fail != nil { |
| + p, err := fail(n) |
| + return p, Gen{N: n}, err |
| + } |
| + p, ver, err := u.fakeBackend.Start(ctx, rt) |
| + return p, canonicalGen(rt, n, ver), err |
| +} |
| + |
| +func (u *unprovenStartBackend) NextDueVia(ctx context.Context, rt registry.Route, _ string) (int64, error) { |
| + return u.fakeBackend.NextDue(ctx, rt) |
| +} |
| + |
| +func (u *unprovenStartBackend) firstGen() uint64 { |
| + u.smu.Lock() |
| + defer u.smu.Unlock() |
| + return u.gens[0] |
| +} |
| + |
| +var ( |
| + _ GenBackend = (*unprovenStartBackend)(nil) |
| + _ genKiller = (*unprovenStartBackend)(nil) |
| +) |
| + |
| +// P2-13 (plan 2's parked L1): a generation start that failed with its kill |
| +// unproven may still be running. The wake keeps the lease (spec §7.4, |
| +// "unproven, the lease is kept"), refuses the host's next wake, and gives |
| +// the lease back only once the sweep proves the generation's child empty |
| +// through KillGen. Before, a start that returned no process released the |
| +// lease at once, and one whose Stop failed released it anyway. |
| +// |
| +// The sentinel rows hold with generations off too: errGenUnproven and |
| +// instancectl.ErrGenUnproven exist only under plan 2's GenContainment, |
| +// which can be on while swaps are off. This is the witness for M9. |
| +func TestSwapWitnessUnprovenWakeKeepsTheLease(t *testing.T) { |
| + lost := errors.New("the unit exited before it was observed") |
| + rows := []struct { |
| + name string |
| + gens bool |
| + leader *stubbornProc // the row's process, alive until told to exit; nil: none |
| + err error |
| + }{ |
| + {"errGenUnproven with no process", true, nil, |
| + fmt.Errorf("start unit: %w; %w: kill timed out", lost, errGenUnproven)}, |
| + {"instancectl.ErrGenUnproven with no process", true, nil, |
| + fmt.Errorf("h: start unit: %w", fmt.Errorf("%w: %w", instancectl.ErrGenUnproven, lost))}, |
| + {"a failed stop while the leader lives", true, &stubbornProc{}, |
| + errors.New("health: no answer within 30s")}, |
| + {"generations off, sentinel present", false, nil, |
| + fmt.Errorf("start unit: %w; %w: kill timed out", lost, errGenUnproven)}, |
| + } |
| + for _, row := range rows { |
| + t.Run(row.name, func(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + a.gens = row.gens |
| + kb := &killGenBackend{fakeBackend: b, err: errors.New("the generation's child still holds a process")} |
| + ub := &unprovenStartBackend{killGenBackend: kb, fail: func(uint64) (Proc, error) { |
| + if row.leader != nil { |
| + return row.leader, row.err |
| + } |
| + return nil, row.err |
| + }} |
| + a.backend = ub |
| + host := "unproven.example.com" |
| + rt := testRoute(host) |
| + ctx := context.Background() |
| + retained := func() *unprovenWake { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + return a.unprovenWakes[host] |
| + } |
| + settled := func() bool { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + u := a.unprovenWakes[host] |
| + return u == nil || !u.busy |
| + } |
| + |
| + if _, err := a.beginRequest(ctx, rt); err == nil { |
| + t.Fatal("the wake should have failed") |
| + } |
| + if got := b.releaseCount(); got != 0 { |
| + t.Fatalf("lease releases = %d after a failed wake whose generation is unproven, want 0", got) |
| + } |
| + if retained() == nil { |
| + t.Fatal("the failed wake was not retained") |
| + } |
| + n := ub.firstGen() |
| + |
| + // The host's next wake is refused before it takes a lease. |
| + _, err := a.beginRequest(ctx, rt) |
| + if !errors.Is(err, errWakeUnproven) { |
| + t.Fatalf("second wake: err = %v, want errWakeUnproven", err) |
| + } |
| + if acq, _, _ := b.counts(); acq != 1 { |
| + t.Fatalf("Acquire calls = %d, want 1 (the refusal comes before the lease)", acq) |
| + } |
| + |
| + // A sweep whose proof still fails keeps the lease. |
| + a.sweepOnce() |
| + waitFor(t, "the retained wake's retry to finish", settled) |
| + if got := b.releaseCount(); got != 0 || retained() == nil { |
| + t.Fatalf("lease releases = %d after a failed proof, want 0 and the wake retained", got) |
| + } |
| + if row.leader != nil { |
| + if got := row.leader.stops.Load(); got < 2 { |
| + t.Fatalf("leader Stop calls = %d, want the sweep to have stopped it again", got) |
| + } |
| + // The leader exits, but its generation's child is still not |
| + // proven empty: the lease stays. |
| + row.leader.exited.Store(true) |
| + a.sweepOnce() |
| + waitFor(t, "the retained wake's retry to finish", settled) |
| + if got := b.releaseCount(); got != 0 || retained() == nil { |
| + t.Fatalf("lease releases = %d with the leader gone but the child unproven, want 0", got) |
| + } |
| + } |
| + |
| + kb.setErr(nil) |
| + a.sweepOnce() |
| + waitFor(t, "the lease release", func() bool { return b.releaseCount() == 1 }) |
| + waitFor(t, "the retained wake dropped", func() bool { return retained() == nil }) |
| + if k := kb.killed(); !slices.Contains(k, n) { |
| + t.Fatalf("KillGen calls = %v, want generation %d's", k, n) |
| + } |
| + |
| + release, err := a.beginRequest(ctx, rt) |
| + if err != nil { |
| + t.Fatalf("third wake: %v", err) |
| + } |
| + release() |
| + if got := b.startCount(); got != 1 { |
| + t.Fatalf("Start calls = %d, want 1 (the third wake)", got) |
| + } |
| + a.Bounce(host) |
| + waitFor(t, "cleanup", func() bool { return a.entryCount() == 0 }) |
| + }) |
| + } |
| +} |