amadan
activator: seed stop before cleanup, stale socket cleared, observer stands down after a requested stop (plan 3, task 4 fix round 1)
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
4 files changed,
+425
−58
internal/activator/backend_exec.go+82 −29internal/activator/prepare.go+15 −10internal/activator/unitproc.go+75 −12internal/activator/unitproc_test.go+253 −7
CI passed — run details
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index 7313baace..1287c7576 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -570,8 +570,8 @@ func (c *capBuffer) String() string { |
| // seedArgv and withoutVars calls seedChild uses, so the two cannot drift. |
| func (b *ExecBackend) seedUnit(ctx context.Context, rt registry.Route) error { |
| host := rt.Host |
| - if b.Control == nil { |
| - return fmt.Errorf("%s: instance control channel not configured", host) |
| + if err := b.nestedGate(rt); err != nil { |
| + return err |
| } |
| if err := b.ensureLiveLink(ctx, host); err != nil { |
| return err |
| @@ -600,33 +600,58 @@ func (b *ExecBackend) seedUnit(ctx context.Context, rt registry.Route) error { |
| _, statErr := os.Stat(dbPath) |
| seedOwnsDB := seedCreatedTheDB(statErr) |
| - sctx, cancel := context.WithTimeout(ctx, seedTimeout) |
| - defer cancel() |
| log.Printf("activator: seeding %s: %s", host, strings.Join(argv, " ")) |
| + // The client already bounds a seed call at instancectl.DeadlineSeed — |
| + // wrapping ctx any tighter here (seedTimeout, at 2m, is shorter) would |
| + // guarantee the failure path on every seed that legitimately runs |
| + // longer than that. Wrapping at the SAME deadline is redundant with the |
| + // client's own timeout but harmless, and keeps this call visibly |
| + // bounded without second-guessing how long a seed may take. |
| + sctx, cancel := context.WithTimeout(ctx, instancectl.DeadlineSeed) |
| res, err := b.Control.Seed(sctx, host, b.tier(rt).MemoryMax) |
| + cancel() |
| if err != nil { |
| - if seedOwnsDB { |
| - removeSeedDB(dbPath) |
| - } |
| - return fmt.Errorf("seed for %s: %w", host, err) |
| - } |
| - // §6.2: the result the control service reports is only trustworthy once |
| - // the seed's cgroup is provably empty — systemd reporting the unit |
| - // inactive is not the same fact as no process from it still running and |
| - // possibly still writing the database this result is about to judge. |
| - if err := b.waitSeedCgroupEmpty(ctx, rt); err != nil { |
| - log.Printf("activator: %s: seed cgroup did not empty: %v", host, err) |
| + return b.seedUnitFailed(ctx, rt, dbPath, seedOwnsDB, fmt.Errorf("seed for %s: %w", host, err)) |
| } |
| if res.Result != "success" { |
| - if seedOwnsDB { |
| - removeSeedDB(dbPath) |
| - } |
| - return fmt.Errorf("seed for %s failed: exit %s (%s)", host, res.Status, res.Result) |
| + return b.seedUnitFailed(ctx, rt, dbPath, seedOwnsDB, fmt.Errorf("seed for %s failed: exit %s (%s)", host, res.Status, res.Result)) |
| } |
| log.Printf("activator: seeded %s", host) |
| return nil |
| } |
| +// seedUnitFailed is seedUnit's shared failure tail (controller ruling, fix |
| +// round 1): on EITHER failure shape — Control.Seed itself errored, or it |
| +// answered with a non-success result — the seed unit is stopped and its |
| +// cgroup is proven empty BEFORE any database cleanup runs, never after. |
| +// The earlier ordering ran the cleanup on the bare result, which could race |
| +// a seed process SIGKILL hadn't actually reaped yet — exactly the class of |
| +// bug Stop's own ErrStillRunning discipline exists to rule out elsewhere. |
| +// removeSeedDB runs only when seedOwnsDB AND the wait proved the cgroup |
| +// empty; when it did not, nothing is removed and the returned error says |
| +// so, because a database is not safe to delete out from under a process |
| +// that might still be writing it. |
| +func (b *ExecBackend) seedUnitFailed(ctx context.Context, rt registry.Route, dbPath string, seedOwnsDB bool, cause error) error { |
| + host := rt.Host |
| + // Detached from ctx's own cancellation (it may already be why the seed |
| + // failed) but still bounded, exactly like Stop's own stop call. |
| + sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl) |
| + if err := b.Control.Stop(sctx, host, instancectl.RoleSeed); err != nil { |
| + log.Printf("activator: %s: stop seed unit after failure: %v", host, err) |
| + } |
| + cancel() |
| + wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout) |
| + emptyErr := b.waitSeedCgroupEmpty(wctx, rt) |
| + wcancel() |
| + if emptyErr != nil { |
| + return fmt.Errorf("%w (the seed's processes remain after stopping it; its database was left in place: %v)", cause, emptyErr) |
| + } |
| + if seedOwnsDB { |
| + removeSeedDB(dbPath) |
| + } |
| + return cause |
| +} |
| + |
| // waitSeedCgroupEmpty polls the seed unit's cgroup.events until it reports |
| // empty or actStopTimeout elapses, honouring ctx and b.ObservePoll like the |
| // exit observer (unitproc.go). A timeout or a read error is returned to the |
| @@ -978,13 +1003,8 @@ func (b *ExecBackend) normalizeSocketMode(rt registry.Route, proc Proc) { |
| // than exec'd directly — the agent has no privilege to enter the tenant's |
| // namespace, only the control service does. |
| func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, string, error) { |
| - if b.Control == nil { |
| - return nil, "", fmt.Errorf("%s: instance control channel not configured", rt.Host) |
| - } |
| - if b.NestedOK != nil { |
| - if err := b.NestedOK(); err != nil { |
| - return nil, "", fmt.Errorf("%s: refusing to wake an isolated instance: %w", rt.Host, err) |
| - } |
| + if err := b.nestedGate(rt); err != nil { |
| + return nil, "", err |
| } |
| if err := b.ensureLiveLink(ctx, rt.Host); err != nil { |
| return nil, "", err |
| @@ -1029,15 +1049,48 @@ func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, s |
| } |
| } |
| if err := waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout); err != nil { |
| + // Set before the Stop call below, closing the same race Stop's own |
| + // internal set leaves open by a few instructions: observe runs |
| + // concurrently from the moment newUnitProc started it, and setting |
| + // this here rather than waiting for it to happen inside up.Stop |
| + // means the observer can never see a foreign invocation appear and |
| + // fire its own redundant stop for a unit we already know we are |
| + // about to stop (controller ruling, fix round 1). |
| + up.stopRequested.Store(true) |
| stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| defer cancel() |
| - _ = up.Stop(stopCtx) |
| + if serr := up.Stop(stopCtx); serr != nil { |
| + // The health error, not this one, is the wake's failure — the |
| + // observer stands down on its own once the stop actually |
| + // lands (stopRequested above), so an ErrStillRunning here |
| + // 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) |
| + } |
| return nil, "", err |
| } |
| b.normalizeSocketMode(rt, up) |
| return up, spawnVer, nil |
| } |
| +// nestedGate is the pre-flight check every nested spawn path runs first |
| +// (controller ruling, fix round 1): no root instance-control channel |
| +// configured, or the box itself is not fit to run an isolated instance |
| +// (NestedOK's own mounts and /ready check), refuses before anything else |
| +// runs. Shared so seedUnit and startSidecarUnit gate on NestedOK exactly |
| +// like startUnit does — before this ruling, only startUnit checked it. |
| +func (b *ExecBackend) nestedGate(rt registry.Route) error { |
| + if b.Control == nil { |
| + return fmt.Errorf("%s: instance control channel not configured", rt.Host) |
| + } |
| + if b.NestedOK != nil { |
| + if err := b.NestedOK(); err != nil { |
| + return fmt.Errorf("%s: refusing to wake an isolated instance: %w", rt.Host, err) |
| + } |
| + } |
| + return nil |
| +} |
| + |
| // instanceEnv composes the host's two EnvironmentFile layers, both optional |
| // like a unit's "-" prefix: the sync-owned bucket config (<host>.config.env) |
| // underneath, the instance's own file (<host>.env) on top — it wins conflicts, |
| @@ -1220,8 +1273,8 @@ func (b *ExecBackend) startSidecarChild(_ context.Context, rt registry.Route) (P |
| // single layer, read straight from its own file, never merged with the |
| // instance's two-layer config. |
| func (b *ExecBackend) startSidecarUnit(ctx context.Context, rt registry.Route) (Proc, error) { |
| - if b.Control == nil { |
| - return nil, fmt.Errorf("%s: instance control channel not configured", rt.Host) |
| + if err := b.nestedGate(rt); err != nil { |
| + return nil, err |
| } |
| p, _, logs, fileEnv, err := b.prepare(rt, instancectl.RoleSidecar) |
| if err != nil { |
diff --git a/internal/activator/prepare.go b/internal/activator/prepare.go| index 5322e0653..25cbad8f9 100644 |
| --- a/internal/activator/prepare.go |
| +++ b/internal/activator/prepare.go |
| @@ -71,16 +71,21 @@ func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instanc |
| } |
| var ctl *edge.ControlListener |
| if role == instancectl.RoleRun { |
| - // No pre-emptive removal of a stale rt.Socket here (deviation from |
| - // an earlier draft of this method): startChild does not do this |
| - // for a flat host either — a leftover socket file fails the |
| - // SUBSEQUENT bind with "address already in use", which |
| - // memberSafeWakeReason already recognises as the stale-socket wake |
| - // trap (#129) and reports as a diagnosis rather than a symptom. |
| - // Removing it here would silently paper over exactly the case #129 |
| - // exists to surface: two invocations of the same host actually |
| - // live at once. The two spawn paths stay consistent on this by |
| - // construction rather than by a second copy of the same rule. |
| + // Nothing is running: either this is the first invocation of this |
| + // role for this host, or the unit is not active (Start would not |
| + // have reached prepare otherwise) — so any file at rt.Socket is a |
| + // leftover from an earlier invocation, most likely one that was |
| + // SIGKILLed before its own bind was ever established. On the |
| + // nested path the agent itself never binds this socket (the app |
| + // does, inside its own namespace, once the trampoline execs it), |
| + // so unlike startChild's flat spawn there is no "our own live |
| + // process squatting on it" case this could ever race: removing it |
| + // here is safe by construction, and leaving it in place would fail |
| + // EVERY later wake's health check with nothing but a dial timeout — |
| + // no member-facing diagnosis at all, unlike the flat path's #129 |
| + // handling, which at least turns the bind failure into a named |
| + // error (memberSafeWakeReason). |
| + _ = os.Remove(rt.Socket) |
| ctl = b.listenControl(rt, p.envFile) |
| } |
| logs, err := listenLogs(b.logSocket(rt.Host, role), b.logEmitter(rt)) |
diff --git a/internal/activator/unitproc.go b/internal/activator/unitproc.go| index 05f81d613..47ce5de61 100644 |
| --- a/internal/activator/unitproc.go |
| +++ b/internal/activator/unitproc.go |
| @@ -6,6 +6,8 @@ import ( |
| "errors" |
| "fmt" |
| "log" |
| + "os" |
| + "path/filepath" |
| "sync/atomic" |
| "time" |
| @@ -38,6 +40,12 @@ type unitProc struct { |
| exited atomic.Bool |
| done chan struct{} |
| closeOnce atomic.Bool |
| + // stopRequested is set before the FIRST Control.Stop this proc's owner |
| + // ever issues for it (fix round 1, controller ruling): once true, the |
| + // observer stops treating a differently-invoked or not-yet-invoked unit |
| + // as a FOREIGN restart to stop on its own — it is instead the proof |
| + // that our own invocation is already over (see observe). |
| + stopRequested atomic.Bool |
| } |
| func (b *ExecBackend) newUnitProc(rt registry.Route, role instancectl.Role, p instancePaths, inv string, ctl *edge.ControlListener, logs *logListener) *unitProc { |
| @@ -55,6 +63,10 @@ func (p *unitProc) Exited() bool { return p.exited.Load() } |
| // listener after the process is gone). |
| func (p *unitProc) Stop(ctx context.Context) error { |
| if !p.Exited() { |
| + // Set before the Control.Stop call, not after: observe polls |
| + // concurrently, and a stop it cannot yet see as "requested" would |
| + // still fire its own foreign-invocation stop for the same unit. |
| + p.stopRequested.Store(true) |
| sctx, cancel := context.WithTimeout(ctx, instancectl.DeadlineControl) |
| err := p.b.Control.Stop(sctx, p.rt.Host, p.role) |
| cancel() |
| @@ -101,6 +113,7 @@ func (b *ExecBackend) observe(p *unitProc) { |
| } |
| foreignStopped := false |
| lastRemain := time.Time{} |
| + lastEventsErr := time.Time{} |
| for { |
| ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| active, inv, err := state(ctx, p.unit) |
| @@ -117,7 +130,26 @@ func (b *ExecBackend) observe(p *unitProc) { |
| case active == "inactive" || active == "failed": |
| terminal = true |
| } |
| - if foreign && !foreignStopped { |
| + |
| + // A stop of OUR OWN invocation is outstanding (Stop was already |
| + // called): systemd starts a fresh invocation of the same unit only |
| + // once the previous one's stop job has fully completed, so seeing |
| + // anything that is not provably that same old invocation — a |
| + // different id, or none reported yet while the unit sits in a |
| + // non-terminal ActiveState (mid-cycle to a fresh one) — is itself |
| + // proof our invocation is over. No cgroup check is needed here: |
| + // there is no "our process might still be running" left to |
| + // disprove, because systemd's own state machine already disproved |
| + // it by allowing a new invocation to begin (controller ruling, fix |
| + // round 1). |
| + if p.stopRequested.Load() && err == nil && !terminal && (foreign || inv == "") { |
| + emitEvent(b.Logs, p.rt, "warn", "instance exited (status unknown: replaced after stop)") |
| + p.exited.Store(true) |
| + close(p.done) |
| + return |
| + } |
| + |
| + if foreign && !foreignStopped && !p.stopRequested.Load() { |
| log.Printf("activator: %s: unit restarted outside the activator (invocation %s, ours %s); stopping it", p.rt.Host, inv, p.invocation) |
| sctx, scancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl) |
| if err := b.Control.Stop(sctx, p.rt.Host, p.role); err != nil { |
| @@ -129,11 +161,31 @@ func (b *ExecBackend) observe(p *unitProc) { |
| if terminal || (foreign && foreignStopped) { |
| exists, populated, eerr := events(p.slice, p.unit) |
| if eerr != nil { |
| - log.Printf("activator: %s: cgroup.events: %v", p.rt.Host, eerr) |
| + // A read error is not evidence of anything — not exited, |
| + // not still running, just unknown. Treating it as "cgroup |
| + // empty" would let a still-running process's exit be |
| + // reported anyway, exactly what this proof scheme exists |
| + // to prevent (design §6.5); rate-limited the same way the |
| + // "processes remain" line below is, since a persistent |
| + // read failure would otherwise flood the log every poll. |
| + if time.Since(lastEventsErr) > 2*time.Second { |
| + log.Printf("activator: %s: cgroup.events: %v", p.rt.Host, eerr) |
| + lastEventsErr = time.Now() |
| + } |
| + time.Sleep(poll) |
| + continue |
| } |
| if !exists || !populated { |
| if terminal { |
| - emitEvent(b.Logs, p.rt, exitLevel(p), exitLine(p)) |
| + rec, ok := p.readRecord() |
| + emitEvent(b.Logs, p.rt, exitLevel(rec, ok), exitLine(rec, ok)) |
| + // The record is single-use (design §6.5): consumed |
| + // exactly once into the member-facing line above, then |
| + // cleared so it can never be misread as evidence for a |
| + // LATER invocation of this same role (prepare only |
| + // clears records left over from EARLIER invocations; |
| + // this one is this invocation's own to remove). |
| + p.removeRecord() |
| } else { |
| emitEvent(b.Logs, p.rt, "warn", "instance exited (status unknown: restarted outside the activator)") |
| } |
| @@ -153,13 +205,14 @@ func (b *ExecBackend) observe(p *unitProc) { |
| } |
| } |
| -// exitLine reads the advisory record for this invocation and words the |
| -// event line as childProc does: a non-zero status is a warn line, a clean |
| -// exit is info, and no record is unknown. The record is tenant-writable |
| -// text (design §6.5, final review of plan 2), so every field that reaches |
| -// the line passes through recordToken first. |
| -func exitLine(p *unitProc) string { |
| - rec, ok := p.readRecord() |
| +// exitLine words the event line from the advisory record observe already |
| +// read (fix round 1: read once, pass to both exitLine and exitLevel, so the |
| +// two can never disagree from reading the file twice with something |
| +// changing in between), the same way childProc does: a non-zero status is a |
| +// warn line, a clean exit is info, and no record is unknown. The record is |
| +// tenant-writable text (design §6.5, final review of plan 2), so every |
| +// field that reaches the line passes through recordToken first. |
| +func exitLine(rec instancectl.ExitRecord, ok bool) string { |
| switch { |
| case !ok: |
| return "instance exited (status unknown: no exit record)" |
| @@ -188,8 +241,9 @@ func recordToken(s string) string { |
| return s |
| } |
| -func exitLevel(p *unitProc) string { |
| - if rec, ok := p.readRecord(); ok && (rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0")) { |
| +// exitLevel is exitLine's companion, from the same read (see exitLine). |
| +func exitLevel(rec instancectl.ExitRecord, ok bool) string { |
| + if ok && (rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0")) { |
| return "info" |
| } |
| return "warn" |
| @@ -207,6 +261,15 @@ func (p *unitProc) readRecord() (instancectl.ExitRecord, bool) { |
| return rec, true |
| } |
| +// removeRecord deletes this invocation's own advisory exit record (design |
| +// §6.5) after observe has consumed it into the member-facing event line. |
| +// Best-effort: a record that is already gone, or a filesystem that refuses |
| +// the remove, leaves nothing worse than what prepare's own stale-record |
| +// sweep would have cleared on the NEXT invocation anyway. |
| +func (p *unitProc) removeRecord() { |
| + _ = os.Remove(filepath.Join(p.runDir, instancectl.ExitRecordName(p.role, p.invocation))) |
| +} |
| + |
| // cgroupRoot is CgroupRoot with the nil default applied. |
| func (b *ExecBackend) cgroupRoot() string { |
| if b.CgroupRoot != "" { |
diff --git a/internal/activator/unitproc_test.go b/internal/activator/unitproc_test.go| index 755bcb0e0..05374a3ee 100644 |
| --- a/internal/activator/unitproc_test.go |
| +++ b/internal/activator/unitproc_test.go |
| @@ -27,15 +27,31 @@ type fakeControl struct { |
| startErr error |
| stopHook func(host string, role instancectl.Role) |
| readyErr error |
| + // onStart runs inside Start, in the fake control's own goroutine — the |
| + // same place the real control service's trampoline would exec the app |
| + // and the app would bind its socket (fix round 1: prepare's stale-socket |
| + // removal races a test's OWN pre-bound fixture socket when that bind |
| + // happens before Start is even called, so tests move it here instead). |
| + onStart func() |
| + // onSeed runs inside Seed, standing in for whatever a real seed process |
| + // might have written to disk before it failed — a test uses it to give |
| + // seedOwnsDB cleanup something real to (not) remove. |
| + onSeed func() |
| } |
| func (f *fakeControl) rec(s string) { f.mu.Lock(); f.calls = append(f.calls, s); f.mu.Unlock() } |
| func (f *fakeControl) Start(_ context.Context, host string, role instancectl.Role, max int64) (string, error) { |
| f.rec("start " + host + " " + string(role)) |
| + if f.onStart != nil { |
| + f.onStart() |
| + } |
| return f.inv, f.startErr |
| } |
| func (f *fakeControl) Seed(_ context.Context, host string, max int64) (instancectl.SeedResponse, error) { |
| f.rec("seed " + host) |
| + if f.onSeed != nil { |
| + f.onSeed() |
| + } |
| return f.seed, nil |
| } |
| func (f *fakeControl) Stop(_ context.Context, host string, role instancectl.Role) error { |
| @@ -117,9 +133,17 @@ func TestStartUnitWritesTheLaunchFileAndReturnsAUnitProc(t *testing.T) { |
| unit := &fakeUnit{} |
| b, rt := nestedBackend(t, ctl, unit) |
| // The fake control does not start anything, so health must be faked: |
| - // bind a listener at rt.Socket that answers 200. |
| - stopServing := serveOK(t, rt.Socket) |
| - defer stopServing() |
| + // bind a listener at rt.Socket that answers 200, from inside Start |
| + // itself — where the app's own bind would actually happen — so |
| + // prepare's stale-socket removal (which runs before Start ever calls |
| + // the fake control) cannot race it. |
| + var stopServing func() |
| + ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) } |
| + defer func() { |
| + if stopServing != nil { |
| + stopServing() |
| + } |
| + }() |
| unit.set("active", "inv1", true, true) |
| p, ver, err := b.Start(context.Background(), rt) |
| if err != nil { |
| @@ -173,8 +197,13 @@ func TestUnitProcDoesNotExitOnTerminalStateWhileTheCgroupIsPopulated(t *testing. |
| ctl := &fakeControl{inv: "inv1"} |
| unit := &fakeUnit{} |
| b, rt := nestedBackend(t, ctl, unit) |
| - stopServing := serveOK(t, rt.Socket) |
| - defer stopServing() |
| + var stopServing func() |
| + ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) } |
| + defer func() { |
| + if stopServing != nil { |
| + stopServing() |
| + } |
| + }() |
| unit.set("active", "inv1", true, true) |
| p, _, err := b.Start(context.Background(), rt) |
| if err != nil { |
| @@ -198,8 +227,13 @@ func TestUnitProcStopsAForeignInvocation(t *testing.T) { |
| ctl := &fakeControl{inv: "inv1"} |
| unit := &fakeUnit{} |
| b, rt := nestedBackend(t, ctl, unit) |
| - stopServing := serveOK(t, rt.Socket) |
| - defer stopServing() |
| + var stopServing func() |
| + ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) } |
| + defer func() { |
| + if stopServing != nil { |
| + stopServing() |
| + } |
| + }() |
| unit.set("active", "inv1", true, true) |
| p, _, err := b.Start(context.Background(), rt) |
| if err != nil { |
| @@ -260,6 +294,218 @@ func TestSeedUnitCleansUpOnlyItsOwnDBOnFailure(t *testing.T) { |
| } |
| } |
| +// capturedEvent is one call eventCapture recorded. |
| +type capturedEvent struct{ level, msg string } |
| + |
| +// eventCapture is a minimal EventSink that records Event calls for |
| +// assertions; App is unused by these tests but required by the interface. |
| +type eventCapture struct { |
| + mu sync.Mutex |
| + events []capturedEvent |
| +} |
| + |
| +func (e *eventCapture) App(acct, app, host, line string) {} |
| +func (e *eventCapture) Event(acct, app, host, level, msg string) { |
| + e.mu.Lock() |
| + e.events = append(e.events, capturedEvent{level, msg}) |
| + e.mu.Unlock() |
| +} |
| +func (e *eventCapture) last() (capturedEvent, bool) { |
| + e.mu.Lock() |
| + defer e.mu.Unlock() |
| + if len(e.events) == 0 { |
| + return capturedEvent{}, false |
| + } |
| + return e.events[len(e.events)-1], true |
| +} |
| + |
| +// TestUnitProcExitLineBoundsAForgedExitRecord covers fix round 1's |
| +// controller ruling on the exit record: it is tenant-writable text, so a |
| +// record with fields no real trampoline would ever write must still |
| +// produce a bounded, single-line, member-safe event — and the record must |
| +// be consumed (removed) exactly once. |
| +func TestUnitProcExitLineBoundsAForgedExitRecord(t *testing.T) { |
| + ctl := &fakeControl{inv: "inv1"} |
| + unit := &fakeUnit{} |
| + b, rt := nestedBackend(t, ctl, unit) |
| + sink := &eventCapture{} |
| + b.Logs = sink |
| + var stopServing func() |
| + ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) } |
| + defer func() { |
| + if stopServing != nil { |
| + stopServing() |
| + } |
| + }() |
| + unit.set("active", "inv1", true, true) |
| + p, _, err := b.Start(context.Background(), rt) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + up, ok := p.(*unitProc) |
| + if !ok { |
| + t.Fatalf("proc = %T", p) |
| + } |
| + rec := instancectl.ExitRecord{ |
| + Invocation: up.invocation, |
| + Result: "boom\nboom", // a newline: must never reach the event line raw |
| + Status: strings.Repeat("9", 200), // far past recordToken's 32-byte bound |
| + Code: "exi\x00ted", // a control character |
| + } |
| + raw, err := json.Marshal(rec) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + recordPath := filepath.Join(up.runDir, instancectl.ExitRecordName(instancectl.RoleRun, up.invocation)) |
| + if err := os.WriteFile(recordPath, raw, 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + unit.set("inactive", "inv1", false, false) |
| + waitFor(t, "the unit to exit", func() bool { return p.Exited() }) |
| + ev, ok := sink.last() |
| + if !ok { |
| + t.Fatal("no event emitted") |
| + } |
| + if ev.level != "warn" { |
| + t.Errorf("level = %q, want warn", ev.level) |
| + } |
| + if ev.msg != "instance exited: exit ? (?)" { |
| + t.Errorf("msg = %q, want %q", ev.msg, "instance exited: exit ? (?)") |
| + } |
| + if strings.Contains(ev.msg, "\n") { |
| + t.Errorf("event line contains a newline: %q", ev.msg) |
| + } |
| + if _, err := os.Stat(recordPath); !os.IsNotExist(err) { |
| + t.Errorf("exit record was not removed after being consumed: %v", err) |
| + } |
| +} |
| + |
| +// TestStartUnitWithEmptyInvocationFailsCleanly covers the case the control |
| +// service reports no invocation at all: the unit already exited before its |
| +// start was observed (plan 2 final review). The wake must fail with a |
| +// diagnosis naming that, and — because nothing is running — no unitProc |
| +// is ever created to observe, so Control must never see a stop. |
| +func TestStartUnitWithEmptyInvocationFailsCleanly(t *testing.T) { |
| + ctl := &fakeControl{inv: ""} |
| + unit := &fakeUnit{} |
| + b, rt := nestedBackend(t, ctl, unit) |
| + if _, _, err := b.Start(context.Background(), rt); err == nil || !strings.Contains(err.Error(), "exited before it was observed") { |
| + t.Fatalf("Start = %v", err) |
| + } |
| + if strings.Contains(ctl.joined(), "stop") { |
| + t.Errorf("nothing was ever running; nothing should have been stopped: %s", ctl.joined()) |
| + } |
| +} |
| + |
| +// TestStopKeepsListenersOpenUntilTerminationIsProven covers Stop's |
| +// ordering contract: an unproven stop (ErrStillRunning) must leave the |
| +// control and log listeners open — a writer that may still exist must |
| +// still be able to reach them — and only a stop that observes done may |
| +// close them. |
| +func TestStopKeepsListenersOpenUntilTerminationIsProven(t *testing.T) { |
| + ctl := &fakeControl{inv: "inv1"} |
| + unit := &fakeUnit{} |
| + b, rt := nestedBackend(t, ctl, unit) |
| + var stopServing func() |
| + ctl.onStart = func() { stopServing = serveOK(t, rt.Socket) } |
| + defer func() { |
| + if stopServing != nil { |
| + stopServing() |
| + } |
| + }() |
| + unit.set("active", "inv1", true, true) |
| + p, _, err := b.Start(context.Background(), rt) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + logSocket := b.logSocket(rt.Host, instancectl.RoleRun) |
| + |
| + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) |
| + defer cancel() |
| + if err := p.Stop(ctx); !errors.Is(err, ErrStillRunning) { |
| + t.Fatalf("Stop = %v, want ErrStillRunning", err) |
| + } |
| + c, err := net.Dial("unix", logSocket) |
| + if err != nil { |
| + t.Fatalf("log socket refused a dial after an unproven stop: %v", err) |
| + } |
| + c.Close() |
| + |
| + unit.set("inactive", "inv1", false, false) |
| + waitFor(t, "the unit to exit", func() bool { return p.Exited() }) |
| + ctx2, cancel2 := context.WithTimeout(context.Background(), time.Second) |
| + defer cancel2() |
| + if err := p.Stop(ctx2); err != nil { |
| + t.Errorf("second Stop after exit: %v", err) |
| + } |
| + if c2, err := net.Dial("unix", logSocket); err == nil { |
| + c2.Close() |
| + t.Error("log socket still accepts a dial after Stop closed the listeners") |
| + } |
| +} |
| + |
| +// TestSeedUnitFailureStopsBeforeCleanupAndKeepsTheDBWhenProcessesRemain |
| +// covers the controller ruling's ordering: Control.Stop for the seed unit |
| +// runs, and is proven by an empty cgroup, BEFORE any database cleanup — |
| +// and when the cgroup never reports empty, nothing is removed at all. The |
| +// wait is bounded by actStopTimeout, a const this package does not let a |
| +// test shrink, so this test genuinely waits out that bound. |
| +func TestSeedUnitFailureStopsBeforeCleanupAndKeepsTheDBWhenProcessesRemain(t *testing.T) { |
| + ctl := &fakeControl{seed: instancectl.SeedResponse{InvocationID: "s1", Result: "exit-code", Status: "3", Code: "exited"}} |
| + unit := &fakeUnit{} |
| + b, rt := nestedBackend(t, ctl, unit) |
| + rt.Seed = `["seed","--fast"]` |
| + // Stand in for a real seed process that wrote its half-built database |
| + // before dying — created only once Seed actually runs, so the PRE-seed |
| + // stat that decides seedOwnsDB still sees it absent (seedOwnsDB=true). |
| + ctl.onSeed = func() { |
| + if err := os.WriteFile(rt.DBPath, []byte("half-built"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + // The cgroup stays populated for the whole wait: a process from the |
| + // seed survived SIGKILL and is still there when actStopTimeout elapses. |
| + unit.set("active", "s1", true, true) |
| + err := b.Seed(context.Background(), rt) |
| + if err == nil || !strings.Contains(err.Error(), "processes remain") { |
| + t.Fatalf("Seed = %v, want a processes-remain error", err) |
| + } |
| + if !strings.Contains(ctl.joined(), "stop h.example.com seed") { |
| + t.Errorf("seed unit was never stopped before cleanup was attempted: %s", ctl.joined()) |
| + } |
| + if _, err := os.Stat(rt.DBPath); err != nil { |
| + t.Errorf("the database was removed despite its processes remaining: %v", err) |
| + } |
| +} |
| + |
| +// TestPrepareRemovesAStaleSocketFile covers the controller ruling that |
| +// restored prepare's stale-socket removal (fix round 1, item 2): a file |
| +// left behind at rt.Socket by an earlier, SIGKILLed invocation must not |
| +// survive into the next one, or every later wake's health check would fail |
| +// with nothing but a dial timeout. |
| +func TestPrepareRemovesAStaleSocketFile(t *testing.T) { |
| + ctl := &fakeControl{inv: "inv1"} |
| + unit := &fakeUnit{} |
| + b, rt := nestedBackend(t, ctl, unit) |
| + if err := os.MkdirAll(filepath.Dir(rt.Socket), 0o750); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(rt.Socket, []byte("stale"), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + _, ctl2, logs, _, err := b.prepare(rt, instancectl.RoleRun) |
| + if err != nil { |
| + t.Fatalf("prepare: %v", err) |
| + } |
| + defer logs.Close() |
| + if ctl2 != nil { |
| + defer ctl2.Close() |
| + } |
| + if _, err := os.Stat(rt.Socket); !os.IsNotExist(err) { |
| + t.Errorf("stale socket file was not removed by prepare: %v", err) |
| + } |
| +} |
| + |
| // serveOK binds a unix socket at path and answers 200 to every request, so |
| // waitHealthyCtx passes without a real unit. |
| func serveOK(t *testing.T, path string) func() { |