amadan

carlos / platform Public

activator: every stop records why (run spec §5.3)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 18d98eb633195c8b7b2d158c9d06b7ec2a963b3b parent 0d91bf5
7 files changed, +87 −30
  • approved_run_copy_test.go +1 −1
  • internal/activator/activator.go +37 −23
  • internal/activator/activator_test.go +1 −1
  • internal/activator/cmd.go +15 −0
  • internal/activator/cmd_test.go +28 −0
  • internal/activator/maintenance.go +1 −1
  • internal/activator/maintenance_test.go +4 −4

CI passed — run details

diff --git a/approved_run_copy_test.go b/approved_run_copy_test.go
index d6a0ce00f..5a534e988 100644
--- a/approved_run_copy_test.go
+++ b/approved_run_copy_test.go
@@ -33,7 +33,7 @@ var runCopyIDs = func() []string {
// runCopyBuilt is every deck id whose text the code or docs now carry. The
// task that writes a row's text adds its id here in the same commit.
-var runCopyBuilt = []string{"RUN-C9", "RUN-C10", "RUN-C11", "RUN-C12", "RUN-C13", "RUN-C14"}
+var runCopyBuilt = []string{"RUN-C1", "RUN-C2", "RUN-C3", "RUN-C4", "RUN-C5", "RUN-C6", "RUN-C7", "RUN-C8", "RUN-C9", "RUN-C10", "RUN-C11", "RUN-C12", "RUN-C13", "RUN-C14"}
func mustParseRunRecord(t *testing.T) []copyRow {
t.Helper()
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index f71e93bfd..e86ac422e 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -354,11 +354,16 @@ type actEntry struct {
route registry.Route
state actState
- ready chan struct{} // waking: closed when the wake resolves; then check err
- err error // waking: the wake's outcome
- stopped chan struct{} // stopping: closed when the stop completes
- wakeCancel context.CancelFunc
- stopOutcome *stopObservation
+ ready chan struct{} // waking: closed when the wake resolves; then check err
+ err error // waking: the wake's outcome
+ stopped chan struct{} // stopping: closed when the stop completes
+ wakeCancel context.CancelFunc
+ stopOutcome *stopObservation
+
+ // stopWhy is why this entry is stopping (beginStopLocked), written under
+ // a.mu; finishStop hands it to a run it interrupts (run spec §5.3).
+ stopWhy string
+
sidecarUncertain bool
// Activity tracking: an instance is idle only when nothing is in flight
@@ -757,12 +762,14 @@ func (a *Service) begin(ctx context.Context, rt registry.Route, stamp bool) (rel
// request waits for it then re-wakes — recovery is the routine
// wake path.
if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() {
+ why := StopExited
if repErr != nil {
log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
+ why = StopLeaseLost
} else {
log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
}
- a.beginStopLocked(e)
+ a.beginStopLocked(e, why)
a.mu.Unlock()
continue
}
@@ -842,12 +849,14 @@ func (a *Service) BeginSidecarRequest(ctx context.Context, rt registry.Route) er
// (or absent) entry and turns this request away instead of
// chaining into a re-wake.
if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() {
+ why := StopExited
if repErr != nil {
log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
+ why = StopLeaseLost
} else {
log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
}
- a.beginStopLocked(e)
+ a.beginStopLocked(e, why)
a.mu.Unlock()
continue
}
@@ -994,7 +1003,7 @@ func (a *Service) wake(e *actEntry) {
e.spawnVersion = w.spawnVersion
e.staleSince = time.Time{}
if a.fences[e.host] != 0 {
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopFenced)
} else if e.bounceAfterWake {
// A config delivery or `carlos restart` asked for while this wake
// was under way (Bounce). Same spot and same reason as the fence
@@ -1369,9 +1378,12 @@ const actStopTimeout = 20 * time.Second
// idle sweep (instance has been idle past cfg.Idle), a dead child, a lost
// lease (open requests on either of the last two are already doomed), and
// Bounce (config changed — the instance may be mid-request, but the drain
-// still gives it actStopTimeout to finish before SIGKILL).
-func (a *Service) beginStopLocked(e *actEntry) {
+// still gives it actStopTimeout to finish before SIGKILL). Every caller says
+// why (the Stop* reasons in cmd.go): a run the stop interrupts tells the
+// member (run spec §5.3).
+func (a *Service) beginStopLocked(e *actEntry, why string) {
e.state = actStopping
+ e.stopWhy = why
e.stopped = make(chan struct{})
go a.stop(e)
}
@@ -1450,7 +1462,7 @@ func (a *Service) bounceLocked(e *actEntry) bool {
msg = "sidecar restarting (carlos restart)"
}
emitEvent(a.Logs, e.route, "info", msg)
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopRestart)
return true
}
@@ -1563,7 +1575,7 @@ func (a *Service) ConvergeAdopted(host, ver string) bool {
return false
}
log.Printf("activator: %s serving %s but %s is adopted — cycling onto the new build", host, e.spawnVersion, ver)
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopRelease)
return true
}
@@ -1713,12 +1725,14 @@ func (a *Service) WakeWorker(rt registry.Route) error {
// dead child: its proc is mid-replacement, so leave it be.
if !e.workerRespawning {
if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() {
+ why := StopExited
if repErr != nil {
log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
+ why = StopLeaseLost
} else {
log.Printf("activator: worker %s exited — recycling", e.host)
}
- a.beginStopLocked(e)
+ a.beginStopLocked(e, why)
a.mu.Unlock()
continue
}
@@ -1954,7 +1968,7 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string) {
// lease, so a keyed sibling can take it. Nothing re-wakes it here
// until the box can provide the config (the config gate refuses
// every wake before then).
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopRestart)
a.mu.Unlock()
log.Printf("activator: worker %s NOT respawned (%s): %v — standing down and releasing its lease", host, why, refused)
case err != nil:
@@ -2069,7 +2083,7 @@ func (a *Service) SleepWorker(host string) {
return
}
emitEvent(a.Logs, e.route, "info", "sidecar reported idle — stopping")
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopIdle)
}
// WorkerToken reports the live per-incarnation token for host, or "" — the
@@ -2462,7 +2476,7 @@ func (a *Service) sweepOnce() {
}
if err := e.rep.Err(); err != nil {
log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, err)
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopLeaseLost)
continue
}
if e.proc.Exited() {
@@ -2504,7 +2518,7 @@ func (a *Service) sweepOnce() {
} else {
log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
}
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopExited)
continue
}
// The idle predicate is shared verbatim; only its WINDOW can differ.
@@ -2544,7 +2558,7 @@ func (a *Service) sweepOnce() {
heldForTick := ok && time.Unix(next, 0).Sub(now) <= idleWindow
if !heldForTick {
log.Printf("activator: %s idle %s — hibernating", e.host, now.Sub(e.lastActive).Truncate(time.Second))
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopIdle)
continue
}
// Fall through to the convergence check below: a host held awake
@@ -2571,7 +2585,7 @@ func (a *Service) sweepOnce() {
continue
}
log.Printf("activator: %s hit its max_run of %s — stopping", e.host, e.workerMaxRun)
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopIdle)
continue
}
// Converge onto a shipped build. A running child never re-reads
@@ -2608,7 +2622,7 @@ func (a *Service) sweepOnce() {
continue
}
log.Printf("activator: %s serving %s but live symlink is now %s — cycling onto the new build", e.host, e.spawnVersion, cur)
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopRelease)
cycled++
continue
}
@@ -2838,7 +2852,7 @@ func (a *Service) RetireExcept(keep map[string]bool) {
switch e.state {
case actAwake:
log.Printf("activator: %s is no longer hibernate-managed — releasing", e.host)
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopRetired)
case actWaking:
// A wake can't be stopped mid-flight; finish the retire once it
// resolves (below, off-lock).
@@ -2858,7 +2872,7 @@ func (a *Service) RetireExcept(keep map[string]bool) {
// same host means the route was re-flagged in the meantime.
if a.entries[e.host] == e && e.state == actAwake {
log.Printf("activator: %s is no longer hibernate-managed — releasing (wake finished first)", e.host)
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopRetired)
}
}(e)
}
@@ -2922,7 +2936,7 @@ func (a *Service) Shutdown(ctx context.Context) {
for _, e := range a.entries {
switch e.state {
case actAwake:
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopShutdown)
waits = append(waits, e.stopped)
case actStopping:
waits = append(waits, e.stopped)
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 8edb0a379..e9823ca7b 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -3910,7 +3910,7 @@ func TestStopScheduledAfterTheInterlockClearStillWaitsForTheShed(t *testing.T) {
time.Sleep(300 * time.Millisecond) // exec completes; that goroutine queues on a.mu
e := a.entries["bot.x.internal"]
if e != nil {
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopIdle)
}
a.mu.Unlock()
if e == nil {
diff --git a/internal/activator/cmd.go b/internal/activator/cmd.go
new file mode 100644
index 000000000..6c08edd09
--- /dev/null
+++ b/internal/activator/cmd.go
@@ -0,0 +1,15 @@
+package activator
+
+// Why an instance stop interrupted a run (run spec §5.3). The text is what
+// the member reads after "the run was interrupted:" (RUN-C1 to RUN-C8);
+// beginStopLocked records one on every stop.
+const (
+ StopIdle = "the instance hibernated"
+ StopRestart = "the instance was restarted"
+ StopRelease = "the instance moved to a new release"
+ StopExited = "the instance's process exited"
+ StopLeaseLost = "the instance lost its database lease"
+ StopFenced = "the instance was stopped for a move or a layout change"
+ StopRetired = "the instance was taken off this box"
+ StopShutdown = "the box's agent stopped"
+)
diff --git a/internal/activator/cmd_test.go b/internal/activator/cmd_test.go
new file mode 100644
index 000000000..04048ab17
--- /dev/null
+++ b/internal/activator/cmd_test.go
@@ -0,0 +1,28 @@
+package activator
+
+import (
+ "testing"
+ "time"
+)
+
+func (a *Service) stopWhyOf(host string) string {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if e := a.entries[host]; e != nil {
+ return e.stopWhy
+ }
+ return ""
+}
+
+func TestStopsRecordTheirReason(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Second)
+ rt := testRoute("why.example.com")
+ arrangeAwake(t, a, rt)
+ gate := make(chan struct{})
+ b.lastProc().setStopGate(gate) // hold the entry in stopping
+ defer close(gate)
+ a.Bounce(rt.Host)
+ if got := a.stopWhyOf(rt.Host); got != StopRestart {
+ t.Fatalf("Bounce: stopWhy %q, want %q", got, StopRestart)
+ }
+}
diff --git a/internal/activator/maintenance.go b/internal/activator/maintenance.go
index e1c9b2f36..093fd3a36 100644
--- a/internal/activator/maintenance.go
+++ b/internal/activator/maintenance.go
@@ -130,7 +130,7 @@ func (a *Service) fenceAndStop(ctx context.Context, host string, generation uint
}
wait = e.ready
case actAwake:
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopFenced)
wait = e.stopped
case actStopping:
wait = e.stopped
diff --git a/internal/activator/maintenance_test.go b/internal/activator/maintenance_test.go
index cd6458d06..0ecf5d35c 100644
--- a/internal/activator/maintenance_test.go
+++ b/internal/activator/maintenance_test.go
@@ -215,7 +215,7 @@ func TestMaintenanceOrdinaryStopFailureRemainsObservable(t *testing.T) {
a, e, main, _, _ := maintenanceAwake(t)
main.err = errors.New("stop refused")
a.mu.Lock()
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopIdle)
done := e.stopped
a.mu.Unlock()
<-done
@@ -228,14 +228,14 @@ func TestMaintenanceLaterOrdinaryStopCannotEraseUnresolvedCleanup(t *testing.T)
a, e, main, _, _ := maintenanceAwake(t)
main.err = errors.New("earlier process may still run")
a.mu.Lock()
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopIdle)
done := e.stopped
a.mu.Unlock()
<-done
replacement := &actEntry{host: e.host, route: e.route, state: actAwake, proc: &maintenanceProc{}, sidecar: &maintenanceProc{}, rep: &maintenanceRep{}}
a.mu.Lock()
a.entries[e.host] = replacement
- a.beginStopLocked(replacement)
+ a.beginStopLocked(replacement, StopIdle)
done = replacement.stopped
a.mu.Unlock()
<-done
@@ -693,7 +693,7 @@ func TestStopUncertainCoversLiveEntriesAndIncompleteHistory(t *testing.T) {
// entry but leaves an incomplete observation in stopHistory.
main.err = errors.New("stop refused")
a.mu.Lock()
- a.beginStopLocked(e)
+ a.beginStopLocked(e, StopIdle)
done := e.stopped
a.mu.Unlock()
<-done