amadan

carlos / platform Public

host: make the housekeeper stand-down reachable and bind duties to the holding

Review round 1, two Important findings, one change.

1. The graceful release was unreachable: startHousekeeper ran its loop on a
   context.Background() nothing cancelled, so every systemctl restart
   stranded control/housekeeper-lease/lock.json for up to its 15m TTL (plus
   a retry) — single-box deployments included, and from Part 2 that is ~20
   minutes with no cert issuer after each roll. start() now returns a
   stand-down that cancels the loop, waits for it to release and exit
   (bounded by housekeeperStopWait so a wedged S3 call cannot hold a restart
   open), and closes the gate; RunAgent calls it in the same post-drain block
   that shuts the activator's children down, on the same don't-strand-leases
   principle.

2. Duties had no cancellation on stand-down: they ran under the agent's
   background context, so a prune (a full list+delete sweep) or Part 2's ACME
   order could keep running against a lease this box had already lost. The
   holding now owns a context — one atomic.Pointer replacing the held flag,
   so "held" and "the duty context" can never disagree — begun at
   acquisition, ended on BOTH exits, gate closed first and context cancelled
   second. runHousekeeperDuties takes that gate; an interrupted duty stood
   down rather than failed, so it is not logged as an error and the rest of
   the tick is skipped.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev aadd18e487023203a5f61add57bb8a16b7b97847 parent 45a2ea6
3 files changed, +288 −57
  • internal/host/agent.go +15 −3
  • internal/host/housekeeper.go +138 −28
  • internal/host/housekeeper_test.go +135 −26
diff --git a/internal/host/agent.go b/internal/host/agent.go
index ec245302d..24ee06d0b 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -316,8 +316,9 @@ func RunAgent(args []string) {
// The deployment-global election, armed BEFORE the loops it gates so a
// single-box deployment is holding well before their first tick (a
// minute away) — see housekeeper.go. Not a `go` of its own: it publishes
- // the process's housekeeper and spawns its loop itself.
- startHousekeeper()
+ // the process's housekeeper and spawns its loop itself. Its stand-down
+ // runs in the post-drain block at the end of this function.
+ stopHousekeeper := startHousekeeper()
go retentionLoop(reg)
go registrySnapshotLoop(reg)
// The sweep reads the registry only through this closure — the stamped
@@ -353,6 +354,14 @@ func RunAgent(args []string) {
defer cancel()
svc.Shutdown(ctx)
}
+ // Same principle one level up: this process may hold the DEPLOYMENT's
+ // housekeeper lease, and a restart that walked away from it would leave
+ // control/housekeeper-lease/lock.json standing for up to its 15-minute
+ // TTL — no pruner, and from Part 2 no cert issuer, across an ordinary
+ // roll of a single-box deployment. Stopping cancels every duty in flight
+ // and releases the lock, so the next box (or this one, a second later)
+ // takes over immediately.
+ stopHousekeeper()
}
// peekRegistryPath pre-parses the agent args for -registry (default
@@ -1602,6 +1611,9 @@ func retentionLoop(reg *registry.Registry) {
return
}
log.Printf("agent: retention loop armed (5min + jitter) — prunes only while this box holds the housekeeper lease")
+ // The pass runs under the HOLDING's context, not this loop's: a sweep
+ // that is halfway through its deletes when renewal fails must stop, not
+ // finish against a lock another box now owns (housekeeper.go).
prune := housekeeperDuty{"retention pass", func(ctx context.Context) error {
return retentionPass(ctx, store, pub, reg)
}}
@@ -1609,7 +1621,7 @@ func retentionLoop(reg *registry.Registry) {
for {
time.Sleep(delay)
delay = 5*time.Minute + rand.N(time.Minute)
- runHousekeeperDuties(ctx, housekeeperHeld, prune)
+ runHousekeeperDuties(housekeeperGate, prune)
}
}
diff --git a/internal/host/housekeeper.go b/internal/host/housekeeper.go
index 9ff5cf7ec..aea344a68 100644
--- a/internal/host/housekeeper.go
+++ b/internal/host/housekeeper.go
@@ -33,8 +33,25 @@
// housekeeper strands the lease until expiry, so handover can stall up to
// ~15m. Nothing here cares. The prune is idempotent and not urgent (a policy
// written now is promised "within ~10 minutes", and the loop is a 5-minute
-// tick anyway), and cert renewals run ~30 days early. A GRACEFUL stand-down
-// releases immediately.
+// tick anyway), and cert renewals run ~30 days early.
+//
+// A GRACEFUL stand-down releases immediately, and that path has to be
+// REACHABLE, not merely written: a `systemctl restart carlos-agent` is not a
+// crash, and if it left the lock behind, every ordinary platform roll would
+// cost the deployment up to ~15m (+ a retry) with no housekeeper — including
+// a single-box deployment, which would be inert against its own lock, and
+// from Part 2 that gap is "no cert issuer" after every deploy. So
+// startHousekeeper returns a stop func and RunAgent calls it in the same
+// post-drain block that shuts the activator's children down, beside the
+// comment about not orphaning leases until TTL. That is the same principle.
+//
+// The hold owns a context, not just a flag. Every duty runs under a context
+// derived from the holding, cancelled the instant the hold ends — a prune is
+// a full list+delete sweep and Part 2's ACME order takes seconds to minutes,
+// so a duty that kept running after a lost renewal would stretch the window
+// where two boxes act on deployment-global state from housekeeperPoll to
+// housekeeperPoll plus however long the duty runs. Duties must therefore
+// honour their context; the ones here already do (every store call takes it).
package host
import (
@@ -64,6 +81,12 @@ const housekeeperLeasePrefix = "control/housekeeper-lease"
// 15m TTL plus one retry.
const housekeeperRetry = 5 * time.Minute
+// housekeeperStopWait bounds how long a graceful stand-down may take before
+// the agent stops waiting for it. The release itself gets 5s
+// (releaseHousekeeperLease) and a holder's poll tick is instant, so this is
+// generous; it exists only so a wedged S3 call cannot hold a restart open.
+const housekeeperStopWait = 10 * time.Second
+
// housekeeperPoll is how often a HOLDER checks that renewal is still winning.
// Renewal itself runs inside internal/lease (TTL/3 = 5m); this only decides
// how quickly a box that has LOST the lease stops believing it holds it —
@@ -122,8 +145,17 @@ const (
housekeeperElect
)
+// holding is one continuous stretch of this box being the housekeeper. Its
+// context is what every duty runs under, so ending the holding ends the
+// duties: there is no state where this box believes it does not hold the
+// lease while work it started as holder is still running.
+type holding struct {
+ ctx context.Context
+ cancel context.CancelFunc
+}
+
// housekeeper is the long-lived acquire/renew/stand-down lifecycle plus the
-// flag every gated caller reads.
+// state every gated caller reads.
type housekeeper struct {
mode housekeeperMode
reason string // the one line run() logs when it arms
@@ -132,14 +164,15 @@ type housekeeper struct {
retry time.Duration
poll time.Duration
- flag atomic.Bool
+ // cur is the current holding, or nil when this box is not the
+ // housekeeper. ONE atomic, deliberately: a separate "held" flag beside
+ // it would have an instant where the two disagree, and both the duty
+ // gate and Task 4's cert gate read this on live paths.
+ cur atomic.Pointer[holding]
}
// newHousekeeper decides the mode from the environment. Nothing is acquired
-// here — run() does the work — EXCEPT the sole-housekeeper case, which is a
-// pure fact about the deployment shape and is therefore true from the moment
-// the box is configured, not from whenever a goroutine happens to be
-// scheduled.
+// and no holding is begun here — start() does both.
func newHousekeeper(getenv func(string) string, acquire leaseAcquirer) *housekeeper {
hk := &housekeeper{acquire: acquire, retry: housekeeperRetry, poll: housekeeperPoll}
switch {
@@ -157,7 +190,6 @@ func newHousekeeper(getenv func(string) string, acquire leaseAcquirer) *housekee
case getenv("CARLOS_DEPLOYMENT_DIR") != "":
hk.mode = housekeeperSole
hk.reason = "directory-backed deployment — this box is its own housekeeper (no bucket to elect in)"
- hk.flag.Store(true)
default:
hk.mode = housekeeperNever
hk.reason = "no deployment store configured — never contending"
@@ -169,10 +201,67 @@ func newHousekeeper(getenv func(string) string, acquire leaseAcquirer) *housekee
// non-blocking by contract: it is read on a duty tick and, from Part 2, on
// the edge's certificate path.
func (hk *housekeeper) held() bool {
+ _, ok := hk.gate()
+ return ok
+}
+
+// gate is held() plus the context the caller's work must run under: valid
+// only while the holding lasts, cancelled the moment it ends. A duty that
+// takes this context cannot outlive the authority it started under.
+func (hk *housekeeper) gate() (context.Context, bool) {
if hk == nil {
- return false
+ return nil, false
+ }
+ h := hk.cur.Load()
+ if h == nil {
+ return nil, false
+ }
+ return h.ctx, true
+}
+
+// beginHolding marks this box the housekeeper for as long as parent lives.
+func (hk *housekeeper) beginHolding(parent context.Context) {
+ ctx, cancel := context.WithCancel(parent)
+ hk.cur.Store(&holding{ctx: ctx, cancel: cancel})
+}
+
+// endHolding stops this box being the housekeeper: the gate closes FIRST (no
+// new duty can start) and only then is the context cancelled (whatever is
+// already running unwinds). Idempotent — the stop func calls it after the
+// loop has usually done it already.
+func (hk *housekeeper) endHolding() {
+ if h := hk.cur.Swap(nil); h != nil {
+ h.cancel()
+ }
+}
+
+// start arms this housekeeper and returns the stand-down: cancel the loop,
+// wait for it to release and exit, and close the gate. Calling it is what
+// makes the graceful release reachable — see the file header.
+func (hk *housekeeper) start(parent context.Context) (stop func()) {
+ ctx, cancel := context.WithCancel(parent)
+ // A lone box is the housekeeper from the instant it is configured, not
+ // from whenever a goroutine is scheduled: no acquisition is involved, so
+ // there is nothing to wait for.
+ if hk.mode == housekeeperSole {
+ hk.beginHolding(ctx)
+ }
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ hk.run(ctx)
+ }()
+ return func() {
+ cancel()
+ select {
+ case <-done:
+ case <-time.After(housekeeperStopWait):
+ // Never wedge a shutdown on this: the lock expires on its own
+ // TTL, which is exactly the crash behaviour we already accept.
+ log.Printf("agent: housekeeper: stand-down did not complete within %s — the lease will expire on its TTL instead", housekeeperStopWait)
+ }
+ hk.endHolding()
}
- return hk.flag.Load()
}
// run is the election loop; it returns only when ctx ends (or immediately, on
@@ -203,14 +292,15 @@ func (hk *housekeeper) run(ctx context.Context) {
continue
}
log.Printf("agent: housekeeper: lease acquired — this box runs the deployment's global duties")
- hk.flag.Store(true)
+ hk.beginHolding(ctx)
hk.hold(ctx, held)
}
}
-// hold blocks while this box is the housekeeper, and clears the flag BEFORE
-// it gives the lease up either way — a duty must never start against a lease
-// this box has already decided to let go of.
+// hold blocks while this box is the housekeeper, and ends the holding BEFORE
+// it gives the lease up either way — closing the gate so no duty starts
+// against a lease this box has decided to let go of, and cancelling any duty
+// already in flight so none outlives the authority it began under.
//
// A failed renewal means leadership is lost (internal/lease stops renewing on
// the first failure, precisely so a re-try can't fence a claimant who took
@@ -221,12 +311,12 @@ func (hk *housekeeper) run(ctx context.Context) {
func (hk *housekeeper) hold(ctx context.Context, held heldLease) {
for {
if !sleepCtx(ctx, hk.poll) {
- hk.flag.Store(false)
+ hk.endHolding()
releaseHousekeeperLease(held)
return
}
if err := held.Err(); err != nil {
- hk.flag.Store(false)
+ hk.endHolding()
log.Printf("agent: housekeeper: lease renewal failed — standing down: %v", err)
return
}
@@ -265,16 +355,24 @@ func sleepCtx(ctx context.Context, d time.Duration) bool {
var theHousekeeper atomic.Pointer[housekeeper]
// housekeeperHeld reports whether THIS box currently holds the deployment's
-// housekeeper lease — the gate every deployment-global duty consults. Cheap,
-// non-blocking, and safe before (or without) startHousekeeper: false.
+// housekeeper lease. Cheap, non-blocking, and safe before (or without)
+// startHousekeeper: false. This is the read for callers that only need the
+// answer — Part 2's edge cert gate. Anything that then goes and DOES work
+// wants housekeeperGate instead, so the work is bound to the holding.
func housekeeperHeld() bool { return theHousekeeper.Load().held() }
-// startHousekeeper arms the election for this process. Called once from
-// RunAgent, before the loops whose duties it gates.
-func startHousekeeper() {
+// housekeeperGate is housekeeperHeld plus the context the duty must run
+// under — the gate every deployment-global duty consults.
+func housekeeperGate() (context.Context, bool) { return theHousekeeper.Load().gate() }
+
+// startHousekeeper arms the election for this process and returns its
+// stand-down. Called once from RunAgent, before the loops whose duties it
+// gates; the returned func MUST be called on the way out (RunAgent's
+// post-drain block) or a restart strands the lock until its TTL.
+func startHousekeeper() (stop func()) {
hk := newHousekeeper(os.Getenv, acquireDeploymentLease)
theHousekeeper.Store(hk)
- go hk.run(context.Background())
+ return hk.start(context.Background())
}
// housekeeperDuty is one deployment-global job: a name for the log and the
@@ -286,17 +384,29 @@ type housekeeperDuty struct {
}
// runHousekeeperDuties runs one tick of the given duties — but only on the
-// box holding the lease. A non-holder does nothing and says nothing: that is
-// the normal state of every box but one, on every tick, forever.
+// box holding the lease, and only under the context of THAT holding. A
+// non-holder does nothing and says nothing: that is the normal state of every
+// box but one, on every tick, forever.
+//
+// The gate is consulted once per tick, not once per duty: within a tick the
+// context is what carries a mid-flight stand-down, and re-reading the gate
+// between duties would just replace one cancellation mechanism with two.
//
// Per-duty isolation, the shape retentionPass already uses for per-app
// errors: one duty's failure is logged and the next still runs.
-func runHousekeeperDuties(ctx context.Context, held func() bool, duties ...housekeeperDuty) {
- if !held() {
+func runHousekeeperDuties(gate func() (context.Context, bool), duties ...housekeeperDuty) {
+ ctx, ok := gate()
+ if !ok {
return
}
for _, d := range duties {
- if err := d.run(ctx); err != nil {
+ // The holding can end mid-tick. What is left is not this box's to
+ // run, and the duty that was interrupted stood down — it did not
+ // fail, so it gets no error line either.
+ if ctx.Err() != nil {
+ return
+ }
+ if err := d.run(ctx); err != nil && ctx.Err() == nil {
log.Printf("agent: housekeeper: %s: %v", d.name, err)
}
}
diff --git a/internal/host/housekeeper_test.go b/internal/host/housekeeper_test.go
index 69aa4176b..13ec86c3e 100644
--- a/internal/host/housekeeper_test.go
+++ b/internal/host/housekeeper_test.go
@@ -148,28 +148,31 @@ func waitFor(t *testing.T, what string, cond func() bool) {
t.Fatalf("timed out waiting for %s", what)
}
-// runElection starts hk.run in the background with fast timings and returns a
-// stop func that cancels it and waits for the loop to exit.
+// runElection arms hk the way RunAgent does and registers its stand-down, so
+// every test exercises the same start/stop pair production uses. The returned
+// func is safe to call twice (the cleanup calls it again).
func runElection(t *testing.T, hk *housekeeper) func() {
t.Helper()
- ctx, cancel := context.WithCancel(context.Background())
- done := make(chan struct{})
- go func() {
- defer close(done)
- hk.run(ctx)
- }()
- stop := func() {
- cancel()
- select {
- case <-done:
- case <-time.After(3 * time.Second):
- t.Error("election loop did not exit after cancel")
+ stop := hk.start(context.Background())
+ stopped := false
+ once := func() {
+ if stopped {
+ return
}
+ stopped = true
+ stop()
}
- t.Cleanup(stop)
- return stop
+ t.Cleanup(once)
+ return once
}
+// heldGate is the gate a duty sees on a holder, for the tests that are about
+// duty dispatch rather than the election.
+func heldGate() (context.Context, bool) { return context.Background(), true }
+
+// noGate is the gate every non-holding box sees.
+func noGate() (context.Context, bool) { return nil, false }
+
func fastHousekeeper(getenv func(string) string, a *acquirer) *housekeeper {
hk := newHousekeeper(getenv, a.acquire)
hk.retry = time.Millisecond
@@ -182,7 +185,7 @@ func fastHousekeeper(getenv func(string) string, a *acquirer) *housekeeper {
func TestRunHousekeeperDutiesRunsEachDutyOncePerTickWhenHeld(t *testing.T) {
quietLog(t)
a, b := 0, 0
- runHousekeeperDuties(context.Background(), func() bool { return true },
+ runHousekeeperDuties(heldGate,
housekeeperDuty{"a", func(context.Context) error { a++; return nil }},
housekeeperDuty{"b", func(context.Context) error { b++; return nil }},
)
@@ -194,7 +197,7 @@ func TestRunHousekeeperDutiesRunsEachDutyOncePerTickWhenHeld(t *testing.T) {
func TestRunHousekeeperDutiesSkipsQuietlyWhenNotHeld(t *testing.T) {
buf := captureRacyLog(t)
ran := 0
- runHousekeeperDuties(context.Background(), func() bool { return false },
+ runHousekeeperDuties(noGate,
housekeeperDuty{"retention pass", func(context.Context) error { ran++; return nil }},
)
if ran != 0 {
@@ -210,7 +213,7 @@ func TestRunHousekeeperDutiesSkipsQuietlyWhenNotHeld(t *testing.T) {
func TestRunHousekeeperDutiesLogsOneDutysErrorAndRunsTheRest(t *testing.T) {
buf := captureRacyLog(t)
later := 0
- runHousekeeperDuties(context.Background(), func() bool { return true },
+ runHousekeeperDuties(heldGate,
housekeeperDuty{"retention pass", func(context.Context) error { return errors.New("boom") }},
housekeeperDuty{"later duty", func(context.Context) error { later++; return nil }},
)
@@ -263,14 +266,22 @@ func TestHousekeeperDirDeploymentIsItsOwnHousekeeper(t *testing.T) {
quietLog(t)
a := &acquirer{answer: func(int) (heldLease, error) { return &fakeLease{}, nil }}
hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_DIR": t.TempDir()}), a)
+ stop := runElection(t, hk)
- hk.run(context.Background())
-
+ // No goroutine to wait on: a lone box holds from the instant it is armed,
+ // because there is nothing to acquire.
+ if !hk.held() {
+ t.Fatal("dir deployment does not hold: a directory-backed deployment is one box by construction and must keep pruning")
+ }
if a.count() != 0 {
t.Fatalf("dir deployment attempted acquisition %d times, want 0 — there is no bucket to elect in", a.count())
}
- if !hk.held() {
- t.Fatal("dir deployment does not hold: a directory-backed deployment is one box by construction and must keep pruning")
+
+ // And it stands down with the process, so nothing believes it is the
+ // housekeeper after the agent has gone.
+ stop()
+ if hk.held() {
+ t.Fatal("dir deployment still reports holding after stand-down")
}
}
@@ -377,7 +388,11 @@ func TestHousekeeperStandsDownWhenRenewalFails(t *testing.T) {
}
}
-func TestHousekeeperReleasesTheLeaseOnShutdown(t *testing.T) {
+// The stand-down is the path an ordinary `systemctl restart` takes, so it has
+// to release the lock AND come back — a stop that hung would wedge the
+// restart, and one that returned before the loop had released would strand
+// control/housekeeper-lease/lock.json for its 15-minute TTL.
+func TestHousekeeperStopReleasesTheLeaseAndReturns(t *testing.T) {
quietLog(t)
held := &fakeLease{}
a := &acquirer{answer: func(int) (heldLease, error) { return held, nil }}
@@ -385,7 +400,17 @@ func TestHousekeeperReleasesTheLeaseOnShutdown(t *testing.T) {
stop := runElection(t, hk)
waitFor(t, "the lease to be held", hk.held)
- stop()
+
+ returned := make(chan struct{})
+ go func() {
+ stop()
+ close(returned)
+ }()
+ select {
+ case <-returned:
+ case <-time.After(3 * time.Second):
+ t.Fatal("stop() did not return — a shutdown would hang here")
+ }
if hk.held() {
t.Fatal("still reports holding the lease after shutdown")
@@ -393,12 +418,96 @@ func TestHousekeeperReleasesTheLeaseOnShutdown(t *testing.T) {
if held.releases() != 1 {
t.Fatalf("releases = %d, want 1 — a graceful stand-down hands the lease over immediately", held.releases())
}
+ // The election goroutine is gone, not merely told to go: stop() waits on
+ // it, so a leaked loop would still be acquiring after this point.
+ before := a.count()
+ time.Sleep(20 * time.Millisecond) // many retries at the 1ms test cadence
+ if after := a.count(); after != before {
+ t.Fatalf("acquisitions after stop: %d → %d, want no further attempts (the loop leaked)", before, after)
+ }
+}
+
+// ---- duties are bound to the holding ---------------------------------------
+
+// A duty must not outlive the authority it started under: when renewal fails
+// mid-prune, the sweep's context is cancelled rather than left to finish its
+// deletes against a lock another box now owns.
+func TestHousekeeperCancelsDutiesWhenTheHoldEnds(t *testing.T) {
+ quietLog(t)
+ held := &fakeLease{}
+ a := &acquirer{answer: func(n int) (heldLease, error) {
+ if n == 1 {
+ return held, nil
+ }
+ return nil, contended()
+ }}
+ hk := fastHousekeeper(envFunc(map[string]string{"CARLOS_DEPLOYMENT_BUCKET": "dep-bucket"}), a)
+ runElection(t, hk)
+ waitFor(t, "the lease to be held", hk.held)
+
+ started := make(chan struct{})
+ observed := make(chan error, 1)
+ go func() {
+ runHousekeeperDuties(hk.gate, housekeeperDuty{"slow duty", func(ctx context.Context) error {
+ close(started)
+ select {
+ case <-ctx.Done():
+ observed <- ctx.Err()
+ case <-time.After(3 * time.Second):
+ observed <- nil
+ }
+ return ctx.Err()
+ }})
+ }()
+
+ <-started
+ held.fail(errors.New("renew: precondition failed"))
+
+ select {
+ case err := <-observed:
+ if err == nil {
+ t.Fatal("the duty ran to completion after the hold ended — it must be cancelled")
+ }
+ case <-time.After(3 * time.Second):
+ t.Fatal("timed out waiting for the duty to observe the stand-down")
+ }
+
+ // And the gate is shut for the next tick, so nothing new starts either.
+ if _, ok := hk.gate(); ok {
+ t.Fatal("the gate is still open after the hold ended")
+ }
}
-// ---- the process-wide flag -------------------------------------------------
+// A duty that starts on a holder and is cancelled mid-flight stood down; it
+// did not fail, so it must not be logged as a failure — and the duties behind
+// it in the same tick are no longer this box's to run.
+func TestRunHousekeeperDutiesStopQuietlyWhenTheHoldEndsMidTick(t *testing.T) {
+ buf := captureRacyLog(t)
+ ctx, cancel := context.WithCancel(context.Background())
+ later := 0
+ runHousekeeperDuties(
+ func() (context.Context, bool) { return ctx, true },
+ housekeeperDuty{"retention pass", func(ctx context.Context) error {
+ cancel() // the hold ends while this duty is running
+ return ctx.Err()
+ }},
+ housekeeperDuty{"later duty", func(context.Context) error { later++; return nil }},
+ )
+ if later != 0 {
+ t.Fatalf("later duty ran %d times after the hold ended, want 0", later)
+ }
+ if out := buf.String(); out != "" {
+ t.Fatalf("stand-down logged as a failure: %q", out)
+ }
+}
+
+// ---- the process-wide gate -------------------------------------------------
func TestHousekeeperHeldIsFalseBeforeAnyElectionIsArmed(t *testing.T) {
if housekeeperHeld() {
t.Fatal("housekeeperHeld() is true with no housekeeper armed")
}
+ if _, ok := housekeeperGate(); ok {
+ t.Fatal("housekeeperGate() reports held with no housekeeper armed")
+ }
}