amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1// Unit-backed tenants: hibernation for whole-app instances whose process
2// manager stays systemd (carlos-app@<host>.service — DynamicUser,
3// StateDirectory) rather than the agent. The activator still owns WHEN the
4// process runs — stopped after the idle window, started when a request
5// arrives, the edge holding that request while the unit boots and its
6// upstream turns healthy — but the HOW is systemctl, behind the Systemctl
7// seam so the state-machine tests need no systemd.
8//
9// The lease/replication seams are satisfied by inert implementations, on
10// purpose — this is a deliberate division of labor, not a shortcut:
11//
12// - A unit tenant's data lives in its DynamicUser-owned state dir
13// (/var/lib/carlos-app/<host>, 0700) which the agent, as user carlos,
14// cannot read — so the agent can neither restore it nor run litestream
15// over it. Its replication belongs to the box-wide carlos-litestream
16// unit (root), which streams changes continuously whether the tenant is
17// awake or asleep. That IS the snapshot-on-sleep story here: a stop only
18// happens after a full idle window, so streaming replication shipped the
19// final writes long before the unit goes down.
20//
21// WHAT THIS ACTUALLY REQUIRES from the operator: box-wide litestream
22// (ops.FleetDBs) has no platform-wide convention for a unit tenant's
23// database filename inside its state dir — that filename is the APP's
24// own choice, unlike the dataDir/<host>.db convention exec children
25// get for free — so it can only replicate a unit tenant whose route
26// was added with an explicit `carlos add -backing unit -db <path>`.
27// `-hibernate` unit routes REQUIRE -db (cli.go's runAdd enforces it:
28// going unreplicated there is a genuine data-loss risk, the moment the
29// unit is ever stopped for idleness, not just a missed nicety).
30// Non-hibernating unit routes do NOT enforce it — -db is strongly
31// recommended but stays the operator's unchecked call, and a route
32// added without it is silently replicated by nobody. See
33// ops.FleetDBs's own doc comment for the mechanics.
34//
35// - There is no lease to trade: the unit exists on exactly one box (its
36// state dir is that box's disk) and the box replicator never hands
37// ownership off, so wake/sleep is not a writer handoff. When unit
38// tenants go multi-box, a real lease slots in through this same seam.
39//
40// Pairing with carlos-app@.socket (see internal/host/units): with the
41// tenant's socket unit enabled, systemd holds the listener bound across
42// stops, so the edge's unix dial never hits the socket-absent window — and a
43// connection while the service is stopped starts it (wake-on-connect). The
44// activator's wake path here remains the primary one because it is
45// health-gated: a broken boot becomes a clean 503, not a connect that hangs
46// in the listen backlog.
47package activator
48
49import (
50 "context"
51 "errors"
52 "fmt"
53 "net/http"
54 "os/exec"
55 "strings"
56 "sync"
57 "time"
58
59 "github.com/carlosframework/platform/internal/cgroup"
60 "github.com/carlosframework/platform/internal/registry"
61 "github.com/carlosframework/platform/internal/release"
62)
63
64// UnitName is the systemd unit backing a unit-backed tenant's route —
65// carlos-app@<host>.service, the same instance name the adopt pass
66// try-restarts for non-hibernating tenants.
67func UnitName(host string) string { return "carlos-app@" + host + ".service" }
68
69// ExecName is the metrics sample key for an exec-backed tenant's children —
70// the delegated cgroup directory the activator spawns them into. Defined by
71// internal/cgroup (which creates the directories); surfaced here next to
72// UnitName so the console learns both of a host's legal metric keys from
73// the same place.
74func ExecName(host string) string { return cgroup.ExecName(host) }
75
76// UnitBackend is the Backend for routes with Backing == registry.BackingUnit.
77type UnitBackend struct {
78 // Systemctl runs one systemctl verb against a unit — the Proc-style seam
79 // wrapping systemctl that keeps hibernation tests systemd-free. Nil = the
80 // real systemctl binary.
81 Systemctl func(ctx context.Context, verb, unit string) error
82 // OptDir is the on-box install root (the live/<host> symlinks) — same
83 // layout as ExecBackend's, read for spawnVersion so build convergence
84 // cycles awake unit tenants exactly like exec children.
85 OptDir string
86 // WaitHealthy blocks until the tenant answers HTTP on its upstream
87 // (rt.Socket, else rt.Addr). Nil = the package's socket/TCP prober with
88 // the standard spawnHealthTimeout. A seam so unit tests need no listener.
89 WaitHealthy func(ctx context.Context, rt registry.Route) error
90}
91
92var _ Backend = (*UnitBackend)(nil)
93
94func (b *UnitBackend) systemctl(ctx context.Context, verb, unit string) error {
95 if b.Systemctl != nil {
96 return b.Systemctl(ctx, verb, unit)
97 }
98 out, err := exec.CommandContext(ctx, "systemctl", verb, unit).CombinedOutput()
99 if err != nil {
100 return fmt.Errorf("systemctl %s %s: %v: %s", verb, unit, err, strings.TrimSpace(string(out)))
101 }
102 return nil
103}
104
105// LiveVersion mirrors ExecBackend.LiveVersion — the unit's ExecStart is the
106// same live/<host> symlink the adopt pass repoints, so convergence reads the
107// same layout, through the same one definition of which path segment is the
108// version (see ExecBackend.LiveVersion for why that matters now).
109func (b *UnitBackend) LiveVersion(host string) (string, error) {
110 return release.LiveVersion(b.OptDir, host)
111}
112
113// Acquire returns the inert lease (see the package comment: replication
114// ownership stays with the root carlos-litestream unit, and a single box's
115// unit has no writer handoff to guard).
116func (b *UnitBackend) Acquire(context.Context, registry.Route) (Lease, error) {
117 return unitLease{}, nil
118}
119
120// Restore is a no-op, and NEVER fresh: the tenant's state dir persists on the
121// box and the agent can't even read it (a DynamicUser's 0700 dir), so it
122// cannot know whether a database exists in there — and "I can't tell" must
123// read as "not a first boot", because the alternative is running an app's
124// seed over its live data. Disaster recovery here is a litestream restore run
125// by the operator as root. A unit tenant that wants first-boot initialization
126// does it in its own unit (ExecStartPre), where the state dir is readable.
127func (b *UnitBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil }
128
129// Seed is unreachable for unit tenants (Restore never reports fresh) and is a
130// no-op for the same reason it has to be: the agent can neither read the
131// tenant's state dir nor run a process as its DynamicUser.
132func (b *UnitBackend) Seed(context.Context, registry.Route) error { return nil }
133
134// Start brings the unit up and holds until its upstream answers — the
135// health gate that makes the edge-mediated wake strictly better than bare
136// socket activation (a broken boot becomes a clean 503, not a connect
137// parked in the listen backlog). A start against an already-running unit is
138// systemd's no-op, which is what lets a fresh agent re-adopt tenants left
139// awake by its previous incarnation. No live-symlink precheck: if the
140// binary is missing the unit's exec fails, the health wait times out, and
141// the stop below puts the unit back down.
142func (b *UnitBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
143 unit := UnitName(rt.Host)
144 // Read the symlink before the start, same fail-safe direction as
145 // ExecBackend: a concurrent repoint at worst records the older version
146 // (one extra convergence cycle), never the newer (a stale child the
147 // sweep would think converged).
148 spawnVer, _ := b.LiveVersion(rt.Host)
149 if err := b.systemctl(ctx, "start", unit); err != nil {
150 return nil, "", err
151 }
152 if err := b.waitHealthy(ctx, rt); err != nil {
153 stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
154 defer cancel()
155 _ = b.systemctl(stopCtx, "stop", unit)
156 return nil, "", err
157 }
158 return &svcUnitProc{b: b, unit: unit}, spawnVer, nil
159}
160
161func (b *UnitBackend) waitHealthy(ctx context.Context, rt registry.Route) error {
162 if b.WaitHealthy != nil {
163 return b.WaitHealthy(ctx, rt)
164 }
165 switch {
166 case rt.Socket != "":
167 return waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout)
168 case rt.Addr != "":
169 return waitHealthyAddrCtx(ctx, rt.Addr, spawnHealthTimeout)
170 }
171 return fmt.Errorf("route %s has neither socket nor addr to health-check", rt.Host)
172}
173
174// NextDue reports "nothing scheduled": the due-time probe is admin-gated by
175// the instance's CARLOS_ADMIN_TOKEN, and a unit tenant's env file
176// (/etc/carlos/apps/<host>.env) is root-0600 — unreadable to the agent by
177// design. Unit tenants therefore sleep until a request; the due-time index
178// stays exec-only until the contract grows an agent-readable channel.
179func (b *UnitBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }
180
181// HasSidecar: sidecar co-spawn is exec-child machinery (the sidecar is the
182// agent's own child, CARLOS_*-scrubbed); a unit tenant that wants a sidecar
183// runs it inside its own unit.
184func (b *UnitBackend) HasSidecar(registry.Route) bool { return false }
185
186func (b *UnitBackend) StartSidecar(context.Context, registry.Route) (Proc, error) {
187 return nil, errors.New("unit-backed tenants have no agent-managed sidecars")
188}
189
190// StartWorker: worker spawn is exec-child machinery exactly like sidecar
191// co-spawn — a unit tenant that wants a worker runs it inside its own unit.
192func (b *UnitBackend) StartWorker(context.Context, registry.Route, map[string]string) (Proc, string, error) {
193 return nil, "", errors.New("unit-backed tenants have no agent-managed workers")
194}
195
196// svcUnitProc adapts a UnitBackend-managed systemd unit (carlos-app@<host>.service,
197// a whole-app tenant whose process manager stays systemd) to the Proc seam.
198// Named apart from the isolated-exec unitProc in unitproc.go, which is a
199// different thing entirely: this one wraps a tenant unit the operator
200// authored, that one is a nested exec instance the activator itself spawns
201// through the root instance-control channel.
202type svcUnitProc struct {
203 b *UnitBackend
204 unit string
205
206 mu sync.Mutex
207 stopped bool
208}
209
210// Stop is `systemctl stop` — systemd runs the unit's own SIGTERM drain and
211// the ctx bounds how long we wait for the job.
212func (p *svcUnitProc) Stop(ctx context.Context) error {
213 err := p.b.systemctl(ctx, "stop", p.unit)
214 p.mu.Lock()
215 p.stopped = true
216 p.mu.Unlock()
217 return err
218}
219
220// Exited reports false until Stop: crash supervision belongs to systemd
221// (Restart=always on the unit), so a crashed tenant restarts under its unit
222// without the activator noticing — deliberately, because Exited is polled
223// under a.mu on every request and sweep tick, and an exec'd `systemctl
224// is-active` there would put a fork on the hot path. The trade-off: a unit
225// wedged past its start rate limit looks awake until its idle window stops
226// it (the edge's dials 502 meanwhile) — operator territory, exactly as it
227// is for a non-hibernating unit today.
228func (p *svcUnitProc) Exited() bool {
229 p.mu.Lock()
230 defer p.mu.Unlock()
231 return p.stopped
232}
233
234// unitLease / unitRep: the inert lease (rationale in the package comment).
235type unitLease struct{}
236
237func (unitLease) Replicate(context.Context, string, time.Duration) (Rep, error) {
238 return unitRep{}, nil
239}
240func (unitLease) Release(context.Context) error { return nil }
241
242type unitRep struct{}
243
244func (unitRep) Err() error { return nil }
245func (unitRep) Close(context.Context) error { return nil }
246
247// waitHealthyAddr is waitHealthy for TCP upstreams (rt.Addr — unit tenants
248// that can't accept a unix listener yet; see the carlos-app@.socket unit's
249// fallback note).
250func waitHealthyAddr(addr string, timeout time.Duration) error {
251 deadline := time.Now().Add(timeout)
252 client := &http.Client{Timeout: 3 * time.Second}
253 var last error
254 var poll time.Duration
255 for time.Now().Before(deadline) {
256 resp, err := client.Get("http://" + addr + "/")
257 if err == nil {
258 resp.Body.Close()
259 if resp.StatusCode < 500 {
260 return nil
261 }
262 last = fmt.Errorf("unhealthy: HTTP %d", resp.StatusCode)
263 } else {
264 last = err
265 }
266 poll = nextHealthPoll(poll)
267 time.Sleep(poll)
268 }
269 return fmt.Errorf("instance not healthy within %s: %v", timeout, last)
270}
271
272func waitHealthyAddrCtx(ctx context.Context, addr string, timeout time.Duration) error {
273 done := make(chan error, 1)
274 go func() { done <- waitHealthyAddr(addr, timeout) }()
275 select {
276 case err := <-done:
277 return err
278 case <-ctx.Done():
279 return ctx.Err()
280 }
281}
282
283// SplitBackend composes the two production backends behind the one Backend
284// seam the state machine sees: each call dispatches on the route's Backing,
285// so exec children and unit tenants share the singleflight wakes, idle
286// clock, due index, and convergence sweep without forking any of it.
287type SplitBackend struct {
288 Exec Backend // rt.Backing == "" — agent-spawned children (ExecBackend)
289 Unit Backend // rt.Backing == registry.BackingUnit (UnitBackend)
290}
291
292var _ Backend = (*SplitBackend)(nil)
293
294func (b *SplitBackend) pick(rt registry.Route) Backend {
295 if rt.Backing == registry.BackingUnit {
296 return b.Unit
297 }
298 return b.Exec
299}
300
301func (b *SplitBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) {
302 return b.pick(rt).Acquire(ctx, rt)
303}
304
305// AdmitConfig dispatches like every other call: on Backing alone. A unit
306// tenant's config is the root adopt pass's, so only the exec half is gated.
307func (b *SplitBackend) AdmitConfig(ctx context.Context, rt registry.Route) error {
308 if g, ok := b.pick(rt).(configAdmitter); ok {
309 return g.AdmitConfig(ctx, rt)
310 }
311 return nil
312}
313
314// SeedPlaced dispatches on Backing, like every other call. A unit tenant
315// never carries a seed.
316func (b *SplitBackend) SeedPlaced(ctx context.Context, rt registry.Route, rep Rep) error {
317 if p, ok := b.pick(rt).(seedPlacer); ok {
318 return p.SeedPlaced(ctx, rt, rep)
319 }
320 return nil
321}
322func (b *SplitBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) {
323 return b.pick(rt).Restore(ctx, rt)
324}
325func (b *SplitBackend) Seed(ctx context.Context, rt registry.Route) error {
326 return b.pick(rt).Seed(ctx, rt)
327}
328func (b *SplitBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
329 return b.pick(rt).Start(ctx, rt)
330}
331func (b *SplitBackend) NextDue(ctx context.Context, rt registry.Route) (int64, error) {
332 return b.pick(rt).NextDue(ctx, rt)
333}
334func (b *SplitBackend) HasSidecar(rt registry.Route) bool { return b.pick(rt).HasSidecar(rt) }
335
336// StartGen dispatches like Start. A side without generation support (the
337// unit side: systemd binds a unit tenant's canonical socket) is started
338// with Start and gets the canonical handle.
339func (b *SplitBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) {
340 if g, ok := b.pick(rt).(GenBackend); ok {
341 return g.StartGen(ctx, rt, n, perGen)
342 }
343 proc, ver, err := b.pick(rt).Start(ctx, rt)
344 // The handle goes back with an error too, as ExecBackend.StartGen's does
345 // (P2-13).
346 return proc, canonicalGen(rt, n, ver), err
347}
348
349// NextDueVia dispatches like NextDue; a side without generation support
350// answers on rt's own socket.
351func (b *SplitBackend) NextDueVia(ctx context.Context, rt registry.Route, socket string) (int64, error) {
352 if g, ok := b.pick(rt).(GenBackend); ok {
353 return g.NextDueVia(ctx, rt, socket)
354 }
355 return b.pick(rt).NextDue(ctx, rt)
356}
357
358// KillGen dispatches like Start, for an exec row only (zero-downtime spec
359// §8.2 step 1). A unit-backed row has no generation child to kill: systemd
360// owns the unit tenant's one process tree. An exec half without KillGen (a
361// test double) cannot prove a kill either. Both refuse, wrapping
362// errNoGenContainment, so a caller never believes a kill it could not make.
363func (b *SplitBackend) KillGen(ctx context.Context, rt registry.Route, n uint64) error {
364 if rt.Backing == registry.BackingUnit {
365 return fmt.Errorf("%s: a unit-backed row has no generation %d to kill: %w", rt.Host, n, errNoGenContainment)
366 }
367 k, ok := b.Exec.(interface {
368 KillGen(ctx context.Context, rt registry.Route, n uint64) error
369 })
370 if !ok {
371 return fmt.Errorf("%s: this backend cannot kill generation %d: %w", rt.Host, n, errNoGenContainment)
372 }
373 return k.KillGen(ctx, rt, n)
374}
375
376// LegacyRunGone is legacyRunProver for the composed backend (#292): an
377// exec row's nested run unit is the exec side's to stop and prove. A
378// unit-backed row never starts a carlos-instance@ unit, so it refuses
379// rather than claiming a proof.
380func (b *SplitBackend) LegacyRunGone(ctx context.Context, rt registry.Route) error {
381 if rt.Backing == registry.BackingUnit {
382 return fmt.Errorf("%s: a unit-backed row has no instance run unit: %w", rt.Host, errNoLegacyProof)
383 }
384 p, ok := b.Exec.(legacyRunProver)
385 if !ok {
386 return fmt.Errorf("%s: %w", rt.Host, errNoLegacyProof)
387 }
388 return p.LegacyRunGone(ctx, rt)
389}
390
391// Layout implements Classifier for the composed backend: a unit-backed row
392// is always today's flat unit shape (the exec-only layouts do not apply to
393// it), and an exec row defers to the exec side's own classification.
394func (b *SplitBackend) Layout(rt registry.Route) (Layout, string) {
395 if rt.Backing == registry.BackingUnit {
396 return LayoutFlat, ""
397 }
398 if c, ok := b.Exec.(Classifier); ok {
399 return c.Layout(rt)
400 }
401 return LayoutFlat, ""
402}
403
404func (b *SplitBackend) StartSidecar(ctx context.Context, rt registry.Route) (Proc, error) {
405 return b.pick(rt).StartSidecar(ctx, rt)
406}
407func (b *SplitBackend) StartWorker(ctx context.Context, rt registry.Route, env map[string]string) (Proc, string, error) {
408 return b.pick(rt).StartWorker(ctx, rt, env)
409}
410
411// StartCmd dispatches like every other call; a unit-backed row has no
412// nested instance to run a command in.
413func (b *SplitBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) {
414 if c, ok := b.pick(rt).(CmdBackend); ok {
415 return c.StartCmd(ctx, rt, spec)
416 }
417 return nil, fmt.Errorf("%w: %s is unit-backed", ErrCmdNotNested, rt.Host)
418}
419
420// StartWorkerStaged forwards spec §7.9's staged spawn to the backend the
421// route picks. One without the capability (UnitBackend, which refuses
422// workers anyway) gets today's StartWorker, and its identity is the version
423// alone, which no recycle can be verified against. That one cannot ask ctx's
424// exec gate at its start, so it is asked here, immediately before
425// StartWorker, as startWorker asks it for a backend without the capability
426// (Astra R3 §5 item 4).
427func (b *SplitBackend) StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) {
428 if s, ok := b.pick(rt).(stagedWorkerStarter); ok {
429 return s.StartWorkerStaged(ctx, rt, env, onto)
430 }
431 if err := runExecGate(ctx); err != nil {
432 return nil, release.StagedIdentity{}, err
433 }
434 p, ver, err := b.pick(rt).StartWorker(ctx, rt, env)
435 return p, release.StagedIdentity{Version: ver}, err
436}
437
438// LiveVersion forwards to the exec side's reader when it has one — both
439// backings share the live/<host> symlink layout, so one reader serves the
440// convergence sweep for every host. (New picks this up via the optional
441// interface, exactly as it does for a bare ExecBackend.)
442func (b *SplitBackend) LiveVersion(host string) (string, error) {
443 if lv, ok := b.Exec.(interface {
444 LiveVersion(host string) (string, error)
445 }); ok {
446 return lv.LiveVersion(host)
447 }
448 return "", nil
449}
450