carlos / platform Public

Sidecar isolation Task 6 (2/6): unitProc.Crashed reads systemd's Result for the proc's own invocation (D6)

The observer's read gains the unit's Result (UnitFacts: ActiveState,
InvocationID and Result in one systemctl show; UnitState stays as the
fake-able state read with the result unknown). Once the two termination
facts hold, a Result other than success for the proc's own invocation
marks the unit crashed; a collected invocation and the tenant's advisory
exit record never do. Crashed is procCrasher for a nested worker, so a
hibernating worker that died of SIGTERM or SIGPIPE on its own (143, 141)
restarts instead of parking.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

claude 17 hours ago dcd04fb4d1169556c31d6af78a9f0bec02e9e529 parent f1533ce

6 files changed, +293 −20

internal/activator/backend_exec.go +8 −6
internal/activator/control.go +36 −6
internal/activator/unitproc.go +33 −5
internal/activator/unitproc_gen_test.go +2 −2
internal/activator/unitproc_test.go +19 −1
internal/activator/worker_nested_test.go +195 −0

diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go

index 4ed763f7b..7e5c1f3b4 100644

--- a/internal/activator/backend_exec.go

+++ b/internal/activator/backend_exec.go

@@ -197,8 +197,13 @@ type ExecBackend struct {

// worker's prepare: a worker with no channel back is not started.

AgentSock *agentsock.Server

// UnitState answers `systemctl show` for the observer (design §6.5).

- // Nil: systemctlUnitState, the real systemctl binary.

+ // Nil: the real systemctl binary. UnitFacts is the same read with the

+ // unit's Result, which unitProc.Crashed needs (sidecar isolation D6);

+ // when set it is the observer's read, and UnitState is not consulted.

+ // Nil with UnitState set: UnitState, with the result unknown (never a

+ // crash). Both nil: systemctlUnitFacts.

UnitState UnitState

+ UnitFacts UnitFacts

// CgroupEvents reads a unit's cgroup.events for the observer and for a

// seed's cgroup-empty wait. Nil: cgroupEventsUnder("/sys/fs/cgroup"),

// or CgroupRoot when set.

@@ -2619,10 +2624,7 @@ func (b *ExecBackend) LegacyRunGone(ctx context.Context, rt registry.Route) erro

if err := b.Control.Stop(ctx, rt.Host, instancectl.RoleRun); err != nil {

return fmt.Errorf("stop %s: %w", unit, err)

}

- state := b.UnitState

- if state == nil {

- state = systemctlUnitState

- }

+ facts := b.unitFacts()

events := b.CgroupEvents

if events == nil {

events = cgroupEventsUnder(b.cgroupRoot())

@@ -2634,7 +2636,7 @@ func (b *ExecBackend) LegacyRunGone(ctx context.Context, rt registry.Route) erro

slice := instancectl.SliceName(rt.Host)

var last error

for {

- active, _, err := state(ctx, unit)

+ active, _, _, err := facts(ctx, unit)

switch {

case err != nil:

last = fmt.Errorf("unit state: %w", err)

diff --git a/internal/activator/control.go b/internal/activator/control.go

index bdac8ae1c..9cbc4a2d3 100644

--- a/internal/activator/control.go

+++ b/internal/activator/control.go

@@ -63,15 +63,27 @@ var _ StagedExeControl = (*instancectl.Client)(nil)

// UnitState answers `systemctl show -p ActiveState,InvocationID <unit>`,

// which needs no privilege. An unloaded unit answers inactive with an

-// empty invocation.

+// empty invocation. Kept for the callers and tests that fake it; the

+// observer reads UnitFacts, which an ExecBackend derives from it (unitFacts)

+// with the result unknown.

type UnitState func(ctx context.Context, unit string) (active, invocation string, err error)

-func systemctlUnitState(ctx context.Context, unit string) (string, string, error) {

- out, err := exec.CommandContext(ctx, "systemctl", "show", "-p", "ActiveState,InvocationID", unit).Output()

+// UnitFacts is UnitState with systemd's Result for the unit: "success",

+// "exit-code", "signal", "core-dump", "timeout" or another of its words, in

+// one `systemctl show` read with the state and the invocation, so the three

+// are of the same moment. unitProc.Crashed reads it (sidecar isolation D6):

+// a terminal unit whose invocation is the proc's own and whose Result is

+// not "success" crashed. An unloaded unit answers inactive, no invocation

+// and (systemd's default) success. "" is a result the reader does not know,

+// which never reads as a crash.

+type UnitFacts func(ctx context.Context, unit string) (active, invocation, result string, err error)

+func systemctlUnitFacts(ctx context.Context, unit string) (string, string, string, error) {

+ out, err := exec.CommandContext(ctx, "systemctl", "show", "-p", "ActiveState,InvocationID,Result", unit).Output()

if err != nil {

- return "", "", err

+ return "", "", "", err

}

- var active, inv string

+ var active, inv, result string

for _, line := range strings.Split(string(out), "\n") {

if k, v, ok := strings.Cut(line, "="); ok {

switch k {

@@ -79,10 +91,28 @@ func systemctlUnitState(ctx context.Context, unit string) (string, string, error

active = strings.TrimSpace(v)

case "InvocationID":

inv = strings.TrimSpace(v)

+ case "Result":

+ result = strings.TrimSpace(v)

}

- return active, inv, nil

+ return active, inv, result, nil

+// unitFacts is the observer's read: UnitFacts when set; otherwise UnitState

+// (a fake, in tests) with the result unknown; otherwise the real systemctl.

+func (b *ExecBackend) unitFacts() UnitFacts {

+ if b.UnitFacts != nil {

+ return b.UnitFacts

+ }

+ if b.UnitState != nil {

+ state := b.UnitState

+ return func(ctx context.Context, unit string) (string, string, string, error) {

+ active, inv, err := state(ctx, unit)

+ return active, inv, "", err

+ }

+ return systemctlUnitFacts

}

// CgroupEvents reads the unit's cgroup.events under the host slice. cgroup

diff --git a/internal/activator/unitproc.go b/internal/activator/unitproc.go

index 8e877b142..581520afc 100644

--- a/internal/activator/unitproc.go

+++ b/internal/activator/unitproc.go

@@ -78,8 +78,30 @@ type unitProc struct {

// D3), nil for every other role. closeListeners closes it with ctl and

// logs, after done.

agent *agentsock.Pinned

+ // crashed is set by observe, before done closes, when the two

+ // termination facts held AND systemd reported this proc's own

+ // invocation with a Result other than success (sidecar isolation D6).

+ // Read by Crashed. The advisory exit record never sets it.

+ crashed atomic.Bool

}

+// Crashed is procCrasher for a nested unit (sidecar isolation D6): true

+// only once termination is proven (Exited) and systemd reported the proc's

+// own invocation with Result other than "success". The worker template has

+// no SuccessExitStatus, so a worker that died of SIGTERM or SIGPIPE on its

+// own (the trampoline's 143 or 141) reads as exit-code, a crash, as a flat

+// child's does. An inactive unit whose invocation was collected reports no

+// invocation of ours and reads as "not a crash", the safe default

+// procCrashed documents; a failed unit keeps its Result until it is reset or

+// started again, so a real crash is still readable when the observer gets

+// there. The tenant's exit record is its claim and decides nothing here.

+//

+// A stop the agent asked for may leave a worker unit failed too (the

+// trampoline answers systemd's SIGTERM with 143). That is told apart by the

+// agent: no procCrashed reader is reached for an entry the agent is

+// stopping (beginStopLocked), so this answer is never read for one.

+func (p *unitProc) Crashed() bool { return p.exited.Load() && p.crashed.Load() }

// launchEnvProc is a generation handle that kept the file environment its

// generation launched with (merge ruling M3).

type launchEnvProc interface {

@@ -239,10 +261,7 @@ func (b *ExecBackend) observe(p *unitProc) {

if poll <= 0 {

poll = time.Second

}

- state := b.UnitState

- if state == nil {

- state = systemctlUnitState

- }

+ facts := b.unitFacts()

events := b.CgroupEvents

if events == nil {

events = cgroupEventsUnder(b.cgroupRoot())

@@ -252,7 +271,7 @@ func (b *ExecBackend) observe(p *unitProc) {

lastEventsErr := time.Time{}

for {

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

- active, inv, err := state(ctx, p.unit)

+ active, inv, result, err := facts(ctx, p.unit)

cancel()

if err != nil {

log.Printf("activator: %s: unit state: %v", p.rt.Host, err)

@@ -312,6 +331,15 @@ func (b *ExecBackend) observe(p *unitProc) {

}

if !exists || !populated {

if terminal {

+ // D6: the crash fact, from the same read as the

+ // terminal state. The invocation is ours (a foreign one

+ // is never terminal here) or collected (inv ""), and

+ // only a Result systemd reported as other than success

+ // for OUR invocation is a crash. Set before done closes,

+ // so a reader that sees Exited sees this too.

+ if inv == p.invocation && result != "" && result != "success" {

+ p.crashed.Store(true)

+ }

rec, ok := p.readRecord()

emitEvent(b.Logs, p.rt, exitLevel(rec, ok), exitLine(rec, ok))

// The record is single-use (design §6.5): consumed

diff --git a/internal/activator/unitproc_gen_test.go b/internal/activator/unitproc_gen_test.go

index 1d8394c42..b2b9300e0 100644

--- a/internal/activator/unitproc_gen_test.go

+++ b/internal/activator/unitproc_gen_test.go

@@ -69,11 +69,11 @@ func TestGenerationUnitProcStopsAndWatchesItsOwnUnit(t *testing.T) {

b.Control = fakeGenControl{ctl}

var mu sync.Mutex

var seen []string

- b.UnitState = func(ctx context.Context, name string) (string, string, error) {

+ b.UnitFacts = func(ctx context.Context, name string) (string, string, string, error) {

mu.Lock()

seen = append(seen, name)

mu.Unlock()

- return unit.state(ctx, name)

+ return unit.facts(ctx, name)

}

unit.set("active", "inv7", true, true)

ctl.stopHook = func(string, instancectl.Role) { unit.set("inactive", "inv7", false, false) }

diff --git a/internal/activator/unitproc_test.go b/internal/activator/unitproc_test.go

index f002c37d6..a225b8a2e 100644

--- a/internal/activator/unitproc_test.go

+++ b/internal/activator/unitproc_test.go

@@ -100,6 +100,10 @@ type fakeUnit struct {

inv string

exists bool

populated bool

+ // result is systemd's Result for the unit ("success", "exit-code",

+ // "signal", ...), read with the state (unitProc.Crashed, sidecar

+ // isolation D6). "" is unknown, which never reads as a crash.

+ result string

}

func (u *fakeUnit) set(active, inv string, exists, populated bool) {

@@ -107,11 +111,25 @@ func (u *fakeUnit) set(active, inv string, exists, populated bool) {

u.active, u.inv, u.exists, u.populated = active, inv, exists, populated

u.mu.Unlock()

}

+// setResult sets what systemd reports as the unit's Result.

+func (u *fakeUnit) setResult(result string) {

+ u.mu.Lock()

+ u.result = result

+ u.mu.Unlock()

func (u *fakeUnit) state(context.Context, string) (string, string, error) {

u.mu.Lock()

defer u.mu.Unlock()

return u.active, u.inv, nil

}

+// facts is state with the Result (the observer's UnitFacts seam).

+func (u *fakeUnit) facts(context.Context, string) (string, string, string, error) {

+ u.mu.Lock()

+ defer u.mu.Unlock()

+ return u.active, u.inv, u.result, nil

func (u *fakeUnit) events(string, string) (bool, bool, error) {

u.mu.Lock()

defer u.mu.Unlock()

@@ -134,7 +152,7 @@ func nestedBackend(t *testing.T, ctl *fakeControl, unit *fakeUnit) (*ExecBackend

b := testExecBackend(t)

b.DataDir = filepath.Join(root, "data")

b.Control = ctl

- b.UnitState = unit.state

+ b.UnitFacts = unit.facts

b.CgroupEvents = unit.events

b.ObservePoll = 20 * time.Millisecond

b.RunDirOverride = filepath.Join(root, "run")

diff --git a/internal/activator/worker_nested_test.go b/internal/activator/worker_nested_test.go

index 7e58d7c71..f31990362 100644

--- a/internal/activator/worker_nested_test.go

+++ b/internal/activator/worker_nested_test.go

@@ -3,8 +3,16 @@

package activator

import (

+ "encoding/json"

+ "os"

+ "path/filepath"

+ "slices"

+ "sync"

+ "sync/atomic"

"testing"

"time"

+ "github.com/carlosframework/platform/internal/instancectl"

)

// Sidecar isolation D4: every worker, flat or nested, is told where its

@@ -30,3 +38,190 @@ func TestFlatWorkerGetsSidecarDB(t *testing.T) {

t.Fatalf("restart env CARLOS_SIDECAR_DB = %q, want %q", env["CARLOS_SIDECAR_DB"], rt.DBPath)

}

+// workerSink is an EventSink safe for the log listener's goroutines.

+type workerSink struct {

+ mu sync.Mutex

+ apps []string

+ events []eventRec

+func (s *workerSink) App(_, _, _, line string) {

+ s.mu.Lock()

+ s.apps = append(s.apps, line)

+ s.mu.Unlock()

+func (s *workerSink) Event(acct, app, host, level, msg string) {

+ s.mu.Lock()

+ s.events = append(s.events, eventRec{acct, app, host, level, msg})

+ s.mu.Unlock()

+func (s *workerSink) eventLines() []string {

+ s.mu.Lock()

+ defer s.mu.Unlock()

+ out := make([]string, 0, len(s.events))

+ for _, e := range s.events {

+ out = append(out, e.level+": "+e.msg)

+ }

+ return out

+func (s *workerSink) appLines() []string {

+ s.mu.Lock()

+ defer s.mu.Unlock()

+ return append([]string(nil), s.apps...)

+// exitedWorkerUnit builds a nested worker's unitProc over fakes whose unit

+// has already reached the terminal state given (active, result) for

+// invocation inv, with its cgroup gone, and waits for the observer to prove

+// termination. record, when set, is written as the invocation's advisory

+// exit record first, the tenant's claim about its own exit.

+func exitedWorkerUnit(t *testing.T, active, inv, result string, record *instancectl.ExitRecord) (*unitProc, *workerSink) {

+ t.Helper()

+ ctl := &fakeControl{inv: "inv1"}

+ unit := &fakeUnit{}

+ unit.set("active", "inv1", true, true)

+ b, rt, _ := nestedWorkerBackend(t, ctl, unit)

+ sink := &workerSink{}

+ b.Logs = sink

+ p, logs, agent, _, _, err := b.prepareWorker(rt)

+ if err != nil {

+ t.Fatalf("prepareWorker: %v", err)

+ }

+ if record != nil {

+ raw, _ := json.Marshal(record)

+ if err := os.WriteFile(filepath.Join(p.runDir, instancectl.ExitRecordName(instancectl.RoleWorker, "inv1")), raw, 0o600); err != nil {

+ t.Fatal(err)

+ }

+ up := b.newWorkerUnitProc(rt, p, "inv1", logs, agent)

+ t.Cleanup(func() { up.closeListeners() })

+ if up.Crashed() {

+ t.Fatal("Crashed before the unit exited")

+ }

+ unit.setResult(result)

+ unit.set(active, inv, false, false)

+ select {

+ case <-up.done:

+ case <-time.After(5 * time.Second):

+ t.Fatal("the observer never proved termination")

+ }

+ return up, sink

+// D6: a worker that died of SIGTERM (143) or SIGPIPE (141) on its own is a

+// crash. The worker template has no SuccessExitStatus, so systemd reports

+// the invocation failed with exit-code, and that fact, with the two

+// termination facts, is what Crashed reads.

+func TestUnitProcCrashedOnExit143(t *testing.T) {

+ t.Parallel()

+ up, sink := exitedWorkerUnit(t, "failed", "inv1", "exit-code",

+ &instancectl.ExitRecord{Invocation: "inv1", Result: "exit-code", Code: "exited", Status: "143"})

+ if !up.Crashed() {

+ t.Fatalf("Crashed = false for a failed invocation (exit 143); events %v", sink.eventLines())

+ }

+ if !slices.Contains(sink.eventLines(), "warn: instance exited: exit 143 (exit-code)") {

+ t.Fatalf("events %v, want the exit line naming 143", sink.eventLines())

+ }

+func TestUnitProcCrashedOnExit141(t *testing.T) {

+ t.Parallel()

+ up, _ := exitedWorkerUnit(t, "failed", "inv1", "exit-code",

+ &instancectl.ExitRecord{Invocation: "inv1", Result: "exit-code", Code: "exited", Status: "141"})

+ if !up.Crashed() {

+ t.Fatal("Crashed = false for a failed invocation (exit 141)")

+ }

+// D6: exit 0 is the only success, and it is not a crash.

+func TestUnitProcNotCrashedOnExit0(t *testing.T) {

+ t.Parallel()

+ up, sink := exitedWorkerUnit(t, "inactive", "inv1", "success",

+ &instancectl.ExitRecord{Invocation: "inv1", Result: "success", Code: "exited", Status: "0"})

+ if up.Crashed() {

+ t.Fatal("Crashed = true for a successful invocation")

+ }

+ if !slices.Contains(sink.eventLines(), "info: instance exited (exit 0)") {

+ t.Fatalf("events %v, want the clean exit line", sink.eventLines())

+ }

+// D6: the advisory exit record is the tenant's claim and never decides a

+// crash. A record claiming a crash over a successful invocation is ignored,

+// and one claiming success over a failed invocation is ignored too; the

+// record words the event line and nothing else.

+func TestUnitProcIgnoresForgedExitRecord(t *testing.T) {

+ t.Parallel()

+ t.Run("a forged crash over a successful unit", func(t *testing.T) {

+ up, _ := exitedWorkerUnit(t, "inactive", "inv1", "success",

+ &instancectl.ExitRecord{Invocation: "inv1", Result: "exit-code", Code: "exited", Status: "1"})

+ if up.Crashed() {

+ t.Fatal("a forged exit record made a successful invocation read as a crash")

+ }

+ })

+ t.Run("a forged success over a failed unit", func(t *testing.T) {

+ up, _ := exitedWorkerUnit(t, "failed", "inv1", "exit-code",

+ &instancectl.ExitRecord{Invocation: "inv1", Result: "success", Code: "exited", Status: "0"})

+ if !up.Crashed() {

+ t.Fatal("a forged exit record hid a failed invocation's crash")

+ }

+ })

+// D6: an inactive unit whose invocation systemd has collected reports no

+// invocation and no result of ours; that reads as "not a crash", the safe

+// default procCrashed documents.

+func TestUnitProcCollectedInvocationNotCrash(t *testing.T) {

+ t.Parallel()

+ up, _ := exitedWorkerUnit(t, "inactive", "", "", nil)

+ if up.Crashed() {

+ t.Fatal("a collected invocation read as a crash")

+ }

+// crashProbeProc wraps a worker's fake child and counts every Crashed

+// question. It answers true, as a requested stop that left the unit failed

+// would, so a reader that asked would misread the stop as a crash.

+type crashProbeProc struct {

+ *fakeProc

+ asked atomic.Int32

+func (p *crashProbeProc) Crashed() bool {

+ p.asked.Add(1)

+ return true

+// D6: a stop the agent asked for is told apart by the agent, not by the exit

+// status. Every procCrashed reader reads a child that exited on its own;

+// none of them is reached for an entry the agent is stopping, so a requested

+// stop that leaves the unit failed (the trampoline's 143) is never misread

+// as a crash to restart.

+func TestRequestedStopNeverReachesProcCrashed(t *testing.T) {

+ t.Parallel()

+ a, b, clk, _ := testActivator(t, time.Hour)

+ rt := alwaysOnWorkerRoute("bot.x.internal") // any exit of an always-on worker restarts

+ if err := a.WakeWorker(rt); err != nil {

+ t.Fatalf("WakeWorker: %v", err)

+ }

+ a.mu.Lock()

+ e := a.entries[rt.Host]

+ probe := &crashProbeProc{fakeProc: e.proc.(*fakeProc)}

+ e.proc = probe

+ a.mu.Unlock()

+ if !a.Bounce(rt.Host) {

+ t.Fatal("Bounce did not stop the worker")

+ }

+ for i := 0; i < 5; i++ {

+ clk.advance(time.Second)

+ a.sweepOnce()

+ }

+ waitFor(t, "the entry to go", func() bool { return !a.Owns(rt.Host) })

+ if n := probe.asked.Load(); n != 0 {

+ t.Fatalf("Crashed was asked %d time(s) during a requested stop; the stop is told apart by the agent, not by the exit status", n)

+ }

+ if b.workerStarts() != 1 {

+ t.Fatalf("worker starts = %d, want 1: a requested stop was restarted as a crash", b.workerStarts())

+ }