| 1 | package main |
| 2 | |
| 3 | // The activator (issue #550 Part 4 v2, Phase 2): instances hibernate. When the |
| 4 | // router is given an S3 bucket (LCHAT_HIBERNATE_BUCKET) it stops relying on |
| 5 | // systemd to keep instance processes alive and instead runs them as its own |
| 6 | // children, on demand — "units are for the router, instances are its |
| 7 | // children". A request for a host with no live process claims the per-host S3 |
| 8 | // single-writer lease, restores the DB from S3 if the local copy is missing, |
| 9 | // spawns the instance, and proxies; after an idle window the instance is |
| 10 | // checkpointed back to S3, the lease released, and the process stopped. State |
| 11 | // parks in S3 as ciphertext (an instance DB is opaque blobs plus routing |
| 12 | // metadata the server already sees — the E2EE rule makes the bucket |
| 13 | // threat-model-free); processes exist only while someone's talking. |
| 14 | // |
| 15 | // Scope (this phase): ONE box. No gossip/HRW placement, no peer proxying — |
| 16 | // membership of one, hash picks self, leases uncontested except against a |
| 17 | // previous incarnation of ourselves. |
| 18 | // |
| 19 | // Bot sidecar co-spawn (a former non-goal, now in): a host with a sidecar |
| 20 | // staged (/data/<host>-bot.env — the same file lchat-bot@.service execs |
| 21 | // against) gets its `bot run` child spawned right after the instance turns |
| 22 | // healthy, and SIGTERMed first at sleep — the sidecar exists exactly while its |
| 23 | // instance is awake, so its lifecycle solves itself. Its own connections flow |
| 24 | // back through this router (the box pins its hosted names to loopback), so the |
| 25 | // sidecar marks them (bot.SidecarHeader) and the activator excludes them from |
| 26 | // the idle clock — otherwise the pair could never sleep, and an orphaned |
| 27 | // sidecar's reconnect loop would wake the instance right back (see serveHTTP). |
| 28 | // Member-bot charters (docs/member-bots.md) run IN the sidecar, and their |
| 29 | // schedules live in blobs the server can't read — the former honest gap. It's |
| 30 | // closed by charter-wake registration: the sidecar reports its earliest |
| 31 | // charter run to the instance (POST /api/charter-next-due), the instance folds |
| 32 | // it into GET /api/next-due (combineNextDue, miniapps.go), and the due-time |
| 33 | // index below wakes for it like any scheduled delivery — no activator changes. |
| 34 | // The hardened dedicated-sidecar tier (bot keys off-box) stays a non-goal here. |
| 35 | // |
| 36 | // Opt-in is PER HOST (routes.hibernate, registry.go): with the bucket env set, |
| 37 | // only flagged instances are activator-managed; the rest stay on their systemd |
| 38 | // units and are proxied straight. That's what makes a production canary safe — |
| 39 | // the env ships fleet-wide as a no-op, then `lchat hibernate <host> --on` |
| 40 | // (control.go) migrates exactly one instance: its unit stops, the flag flips, |
| 41 | // and the first request wakes it here. |
| 42 | // |
| 43 | // The due-time index (a former non-goal, now in): scheduled future-messages/ |
| 44 | // reminders fire on time even while their instance sleeps. At sleep the stop |
| 45 | // path asks the instance for its earliest pending delivery (GET /api/next-due |
| 46 | // over the socket, admin-gated) and records it in the registry (routes.next_due |
| 47 | // — persisted so it survives router restarts); the sweeper wakes any sleeping |
| 48 | // host whose time has arrived, with no client request. Once up, the instance's |
| 49 | // own scheduler (miniapps.go runScheduler, a 5s tick) delivers, and the normal |
| 50 | // idle window puts it back to sleep — which re-records the NEXT due time, so |
| 51 | // the cycle maintains itself. Worst-case lateness ≈ one activator sweep |
| 52 | // interval (≤30s) + spawn + one scheduler tick (≤5s). |
| 53 | |
| 54 | import ( |
| 55 | "context" |
| 56 | "encoding/json" |
| 57 | "errors" |
| 58 | "fmt" |
| 59 | "io" |
| 60 | "log" |
| 61 | "net/http" |
| 62 | "os" |
| 63 | "os/exec" |
| 64 | "path" |
| 65 | "path/filepath" |
| 66 | "strconv" |
| 67 | "strings" |
| 68 | "sync" |
| 69 | "syscall" |
| 70 | "time" |
| 71 | |
| 72 | "github.com/benbjohnson/litestream" |
| 73 | |
| 74 | "lchat/internal/activator" |
| 75 | "lchat/internal/bot" |
| 76 | ) |
| 77 | |
| 78 | // hibernateConfig is the router-process opt-in, read from env. Nil (bucket |
| 79 | // unset) keeps today's behavior byte-identical: instances run under systemd, |
| 80 | // the router only proxies. |
| 81 | type hibernateConfig struct { |
| 82 | Bucket string // LCHAT_HIBERNATE_BUCKET (required to enable) |
| 83 | Region string // LCHAT_HIBERNATE_REGION (optional; AWS default chain resolves "") |
| 84 | Prefix string // LCHAT_HIBERNATE_PREFIX (optional key prefix inside the bucket) |
| 85 | Idle time.Duration // LCHAT_HIBERNATE_IDLE (default 15m; seconds-granular, e.g. "90s") |
| 86 | } |
| 87 | |
| 88 | func hibernateConfigFromEnv() (*hibernateConfig, error) { |
| 89 | bucket := strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_BUCKET")) |
| 90 | if bucket == "" { |
| 91 | return nil, nil |
| 92 | } |
| 93 | cfg := &hibernateConfig{ |
| 94 | Bucket: bucket, |
| 95 | Region: strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_REGION")), |
| 96 | Prefix: strings.Trim(strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_PREFIX")), "/"), |
| 97 | Idle: 15 * time.Minute, |
| 98 | } |
| 99 | if v := strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_IDLE")); v != "" { |
| 100 | d, err := time.ParseDuration(v) |
| 101 | if err != nil || d <= 0 { |
| 102 | return nil, fmt.Errorf("LCHAT_HIBERNATE_IDLE %q: want a positive duration like 15m or 90s", v) |
| 103 | } |
| 104 | cfg.Idle = d |
| 105 | } |
| 106 | return cfg, nil |
| 107 | } |
| 108 | |
| 109 | // hibernatePrefix is the per-instance S3 key prefix: <LCHAT_HIBERNATE_PREFIX>/<host>/ |
| 110 | // (the lease's lock.json is colocated under it by the activator package). |
| 111 | func (c *hibernateConfig) hibernatePrefix(host string) string { |
| 112 | return path.Join(c.Prefix, host) |
| 113 | } |
| 114 | |
| 115 | // meshPrefix is where the mesh's membership heartbeats live, beside the |
| 116 | // per-instance prefixes — the bucket is the mesh's only shared substrate. |
| 117 | // It can't collide with hibernatePrefix: instance prefixes are public FQDNs |
| 118 | // (they contain a dot), and the bare label "mesh" never is one. |
| 119 | func (c *hibernateConfig) meshPrefix() string { |
| 120 | return path.Join(c.Prefix, "mesh/members") |
| 121 | } |
| 122 | |
| 123 | // --- backend seam ------------------------------------------------------------ |
| 124 | // |
| 125 | // The state machine's side effects (S3 lease, restore, process spawn) sit |
| 126 | // behind these interfaces so the machine itself is unit-testable with fakes — |
| 127 | // wake coalescing, idle→sleep, and lease-lost stand-down need no AWS. |
| 128 | |
| 129 | type actBackend interface { |
| 130 | // acquire claims the per-host single-writer lease. Contention surfaces as |
| 131 | // errLeaseHeld so the router can answer 503 + Retry-After. |
| 132 | acquire(ctx context.Context, host string) (actLease, error) |
| 133 | // restore materializes the local DB from S3 when it's absent (a no-op when |
| 134 | // the warm-cache copy is still on disk). |
| 135 | restore(ctx context.Context, host, dbPath string) error |
| 136 | // start spawns the instance process and waits until its socket accepts. It |
| 137 | // also returns the build it execed (the live/<host> symlink target read at |
| 138 | // spawn), so the sweep can later spot a repoint under a still-running child |
| 139 | // and converge it. Empty when unknown — the sweep treats that as "don't |
| 140 | // cycle" (fail-safe). |
| 141 | start(ctx context.Context, rt Route) (actProc, string, error) |
| 142 | // nextDue asks a running instance for its earliest pending scheduled |
| 143 | // delivery (unix seconds; 0 = nothing scheduled). Queried at sleep time, |
| 144 | // while the socket still answers, to feed the due-time index. |
| 145 | nextDue(ctx context.Context, rt Route) (int64, error) |
| 146 | // hasSidecar reports whether host has a bot sidecar staged on this box. |
| 147 | hasSidecar(host string) bool |
| 148 | // startSidecar spawns the host's bot sidecar. Fire-and-forget: a sidecar |
| 149 | // is a pure outbound client with no socket, so there's nothing to |
| 150 | // health-check — its own connect/backoff loop recovers from a slow start. |
| 151 | startSidecar(ctx context.Context, rt Route) (actProc, error) |
| 152 | } |
| 153 | |
| 154 | // actDueStore persists the due-time index — the registry in production |
| 155 | // (routes.next_due), so a sleeping host's timer survives router restarts |
| 156 | // instead of dying with the process. |
| 157 | type actDueStore interface { |
| 158 | setNextDue(host string, due int64) error |
| 159 | dueInstances(now int64) ([]Route, error) |
| 160 | } |
| 161 | |
| 162 | // actLease is a held-but-not-yet-replicating lease (wake in progress). |
| 163 | type actLease interface { |
| 164 | // replicate trades the lease for running replication of the (now present) |
| 165 | // local DB; the lease handle is spent on success. syncInterval is the |
| 166 | // route's replication-class tick (replicationSyncInterval; 0 = stream). |
| 167 | replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (actRep, error) |
| 168 | // release gives the lease up — the bail-out when a wake fails partway. |
| 169 | release(ctx context.Context) error |
| 170 | } |
| 171 | |
| 172 | // actRep is running replication coupled to the lease. |
| 173 | type actRep interface { |
| 174 | // Err reports lost leadership (a lease renewal failed). |
| 175 | Err() error |
| 176 | // Close does the final checkpoint/sync and releases the lease. |
| 177 | Close(ctx context.Context) error |
| 178 | } |
| 179 | |
| 180 | // actProc is a running instance child process. |
| 181 | type actProc interface { |
| 182 | // stop terminates gracefully (SIGTERM, bounded wait, then SIGKILL). |
| 183 | stop(ctx context.Context) error |
| 184 | // exited reports whether the process has died on its own. |
| 185 | exited() bool |
| 186 | } |
| 187 | |
| 188 | // --- state machine ----------------------------------------------------------- |
| 189 | |
| 190 | // Per-host lifecycle. An absent entry is the fourth state: asleep. |
| 191 | // |
| 192 | // asleep --request--------------------> waking (one wake; concurrent |
| 193 | // waking --wake ok--------------------> awake requests coalesce on it) |
| 194 | // waking --wake failed----------------> asleep (error served; next request retries) |
| 195 | // awake --idle timeout / lease lost / |
| 196 | // child died----------------> stopping (final checkpoint, release lease) |
| 197 | // stopping --stop done----------------> asleep (local DB left as a warm cache) |
| 198 | // |
| 199 | // A request that lands on waking waits (bounded) for the wake; one that lands |
| 200 | // on stopping waits for the stop, then re-enters as a fresh wake. |
| 201 | type actState int |
| 202 | |
| 203 | const ( |
| 204 | actWaking actState = iota |
| 205 | actAwake |
| 206 | actStopping |
| 207 | ) |
| 208 | |
| 209 | type actEntry struct { |
| 210 | host string |
| 211 | route Route |
| 212 | state actState |
| 213 | |
| 214 | ready chan struct{} // waking: closed when the wake resolves; then check err |
| 215 | err error // waking: the wake's outcome |
| 216 | stopped chan struct{} // stopping: closed when the stop completes |
| 217 | |
| 218 | // Activity tracking: an instance is idle only when nothing is in flight |
| 219 | // AND the idle window has passed since the last request finished. In-flight |
| 220 | // counting (not just timestamps) is what keeps an instance with one open |
| 221 | // WebSocket awake — the proxy's ServeHTTP blocks for the tunnel's life. |
| 222 | // (Sidecar-marked requests register in neither — see beginSidecarRequest.) |
| 223 | inflight int |
| 224 | lastActive time.Time |
| 225 | |
| 226 | rep actRep |
| 227 | proc actProc |
| 228 | |
| 229 | // Convergence: the `live/<host>` symlink target this child was spawned |
| 230 | // against. A running process never re-reads the symlink, so once a ship |
| 231 | // (or `deploy --version`) repoints it, an awake child keeps serving the old |
| 232 | // build until it's cycled. spawnVersion lets the sweep notice the drift; |
| 233 | // staleSince is when it first did, so a brief transient (a symlink moved |
| 234 | // then rolled straight back) doesn't trigger a needless cycle. Both empty/ |
| 235 | // zero until the wake resolves. |
| 236 | spawnVersion string |
| 237 | staleSince time.Time |
| 238 | |
| 239 | // Bot sidecar co-spawn: the sidecar exists exactly while its instance is |
| 240 | // awake. All sidecar fields are written under a.mu in the awake state only |
| 241 | // (wake's attach, the sweeper's restart); a stop flips the state under the |
| 242 | // same lock first, so stop() owns them uncontended from then on. |
| 243 | sidecar actProc // the co-spawned `bot run` child; nil = none running |
| 244 | sidecarWanted bool // the host had a sidecar staged at wake time |
| 245 | sidecarStarted time.Time // when the current sidecar spawned (backoff reset) |
| 246 | sidecarBackoff time.Duration // last restart delay (doubles, capped) |
| 247 | sidecarRetryAt time.Time // earliest next restart attempt |
| 248 | } |
| 249 | |
| 250 | // Sidecar restart pacing — Restart=always semantics without a supervisor: the |
| 251 | // sweeper redoes a dead sidecar with exponential backoff from the systemd |
| 252 | // unit's RestartSec, capped, and a run that survived sidecarHealthyRun earns a |
| 253 | // fresh backoff (so one crash a day never escalates to the cap). |
| 254 | const ( |
| 255 | sidecarRestartBase = 2 * time.Second |
| 256 | sidecarRestartMax = time.Minute |
| 257 | sidecarHealthyRun = time.Minute |
| 258 | ) |
| 259 | |
| 260 | // sidecarStopTimeout bounds the sidecar's stop at sleep time. `bot run` has no |
| 261 | // drain handler, so the SIGTERM is effectively immediate; the SIGKILL fallback |
| 262 | // behind this timeout is insurance. |
| 263 | const sidecarStopTimeout = 5 * time.Second |
| 264 | |
| 265 | // Build convergence: how long an awake child may serve a symlink-stale binary |
| 266 | // before the sweep gracefully cycles it, and how many to cycle per tick. |
| 267 | // - convergeGrace comfortably exceeds one sweep interval (≤30s) and the |
| 268 | // sidecar's ~32s reconnect, so a genuinely-persistent drift is required — |
| 269 | // a symlink that moved then rolled straight back never trips it, and a |
| 270 | // briefly-busy space gets a chance to idle-hibernate (which converges for |
| 271 | // free) first. |
| 272 | // - one per tick staggers a fleet-wide ship across sweeps instead of blipping |
| 273 | // every space's connections at once. |
| 274 | // |
| 275 | // Convergence is a latency/observability fix, not urgent (the supply-chain bake |
| 276 | // window already holds releases 72h), so a leisurely, gentle cadence is right. |
| 277 | const ( |
| 278 | convergeGrace = 90 * time.Second |
| 279 | maxConvergeCyclesPerSweep = 1 |
| 280 | ) |
| 281 | |
| 282 | // convergeDecision is the pure state machine behind build convergence: given the |
| 283 | // symlink target read this tick (cur/readErr), the build the child execed |
| 284 | // (spawnVer), and when the drift was first seen (staleSince), it says whether to |
| 285 | // cycle now and returns the staleSince to carry forward. Split out so the grace/ |
| 286 | // reset logic is unit-testable without a live activator. Three cases: |
| 287 | // - read error: unknown — neither cycle nor clear the grace; retry next tick. |
| 288 | // - converged / rolled-back / nothing to compare (cur=="" or spawnVer=="" or |
| 289 | // cur==spawnVer): clear the grace clock, don't cycle. |
| 290 | // - stale (symlink moved since spawn): start the grace clock on first sight, |
| 291 | // cycle once the drift has persisted past grace. |
| 292 | func convergeDecision(cur string, readErr error, spawnVer string, staleSince, now time.Time, grace time.Duration) (cycle bool, nextStale time.Time) { |
| 293 | switch { |
| 294 | case readErr != nil: |
| 295 | return false, staleSince |
| 296 | case cur == "" || spawnVer == "" || cur == spawnVer: |
| 297 | return false, time.Time{} |
| 298 | default: |
| 299 | if staleSince.IsZero() { |
| 300 | staleSince = now |
| 301 | } |
| 302 | return now.Sub(staleSince) >= grace, staleSince |
| 303 | } |
| 304 | } |
| 305 | |
| 306 | // wakeWaitBound caps how long a request queues on someone else's wake or stop |
| 307 | // before giving up with 503 — a wedged wake must not pile up connections |
| 308 | // forever. A var so tests can shrink it. |
| 309 | var wakeWaitBound = 30 * time.Second |
| 310 | |
| 311 | // errLeaseHeld: the per-host lease is held elsewhere. On one box this means a |
| 312 | // previous incarnation hasn't released (crashed router; lease expires within |
| 313 | // ~LeaseTTL) — the router answers 503 with Retry-After. |
| 314 | type errLeaseHeld struct { |
| 315 | holder string |
| 316 | retryAfter time.Duration |
| 317 | } |
| 318 | |
| 319 | func (e *errLeaseHeld) Error() string { |
| 320 | return fmt.Sprintf("instance lease held by %q (retry in %s)", e.holder, e.retryAfter) |
| 321 | } |
| 322 | |
| 323 | var errWakeTimeout = errors.New("timed out waiting for instance wake") |
| 324 | |
| 325 | type activatorSvc struct { |
| 326 | cfg hibernateConfig |
| 327 | backend actBackend |
| 328 | due actDueStore |
| 329 | now func() time.Time // a seam for the unit tests' clock |
| 330 | liveVer func(string) (string, error) // live/<host> symlink target; a seam for tests |
| 331 | |
| 332 | // mesh, when non-nil, is the multi-box placement layer (mesh.go): requests |
| 333 | // for hosts whose HRW owner is another live box are proxied there, and the |
| 334 | // due-time sweeper only wakes hosts this box owns. Nil (LCHAT_MESH_SELF |
| 335 | // unset) is a single box — hashing would pick self everywhere, so no mesh |
| 336 | // codepath is touched at all. |
| 337 | mesh *meshSvc |
| 338 | |
| 339 | mu sync.Mutex |
| 340 | entries map[string]*actEntry |
| 341 | |
| 342 | sweepStop chan struct{} |
| 343 | sweepDone chan struct{} |
| 344 | } |
| 345 | |
| 346 | func newActivator(cfg hibernateConfig, backend actBackend, due actDueStore) *activatorSvc { |
| 347 | return &activatorSvc{ |
| 348 | cfg: cfg, |
| 349 | backend: backend, |
| 350 | due: due, |
| 351 | now: time.Now, |
| 352 | liveVer: liveVersion, |
| 353 | entries: make(map[string]*actEntry), |
| 354 | } |
| 355 | } |
| 356 | |
| 357 | // beginRequest ensures the host's instance is awake and registers one in-flight |
| 358 | // request against it. The returned release must be called when the request |
| 359 | // finishes (it stamps last-activity, which starts the idle clock). |
| 360 | func (a *activatorSvc) beginRequest(ctx context.Context, rt Route) (release func(), err error) { |
| 361 | deadline := time.NewTimer(wakeWaitBound) |
| 362 | defer deadline.Stop() |
| 363 | for { |
| 364 | a.mu.Lock() |
| 365 | e := a.entries[rt.Host] |
| 366 | if e == nil { |
| 367 | // Asleep → kick off ONE wake; this request and any latecomers all |
| 368 | // coalesce on e.ready rather than racing a second lease claim |
| 369 | // (singleflight). The wake runs detached so even its initiator is |
| 370 | // held to the same bounded wait as everyone queued behind it. |
| 371 | e = &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})} |
| 372 | a.entries[rt.Host] = e |
| 373 | a.mu.Unlock() |
| 374 | go a.wake(e) |
| 375 | continue |
| 376 | } |
| 377 | |
| 378 | switch e.state { |
| 379 | case actAwake: |
| 380 | // Stand down synchronously-detected failure: a dead child or a |
| 381 | // lost lease found at request time turns into a stop, and the |
| 382 | // request waits for it then re-wakes — recovery is the routine |
| 383 | // wake path. |
| 384 | if repErr := e.rep.Err(); repErr != nil || e.proc.exited() { |
| 385 | if repErr != nil { |
| 386 | log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr) |
| 387 | } else { |
| 388 | log.Printf("activator: instance %s exited unexpectedly — recycling", e.host) |
| 389 | } |
| 390 | a.beginStopLocked(e) |
| 391 | a.mu.Unlock() |
| 392 | continue |
| 393 | } |
| 394 | e.inflight++ |
| 395 | a.mu.Unlock() |
| 396 | return a.releaseFunc(e), nil |
| 397 | |
| 398 | case actWaking: |
| 399 | ready := e.ready |
| 400 | a.mu.Unlock() |
| 401 | select { |
| 402 | case <-ready: |
| 403 | // A failed wake reports to every coalesced waiter (retrying |
| 404 | // here would turn one bad wake into a storm); a successful |
| 405 | // one loops to register against the now-awake entry. |
| 406 | if e.err != nil { |
| 407 | return nil, e.err |
| 408 | } |
| 409 | continue |
| 410 | case <-ctx.Done(): |
| 411 | return nil, ctx.Err() |
| 412 | case <-deadline.C: |
| 413 | return nil, errWakeTimeout |
| 414 | } |
| 415 | |
| 416 | case actStopping: |
| 417 | stopped := e.stopped |
| 418 | a.mu.Unlock() |
| 419 | select { |
| 420 | case <-stopped: |
| 421 | continue // entry is gone; loop wakes it fresh |
| 422 | case <-ctx.Done(): |
| 423 | return nil, ctx.Err() |
| 424 | case <-deadline.C: |
| 425 | return nil, errWakeTimeout |
| 426 | } |
| 427 | } |
| 428 | } |
| 429 | } |
| 430 | |
| 431 | // errSidecarAsleep: a sidecar-marked request arrived for a host with no awake |
| 432 | // instance. Sidecar traffic never wakes anything — the activator itself spawns |
| 433 | // the sidecar as part of a wake, so a marked request for a sleeping host is |
| 434 | // either a stray reconnect racing a stop or an outsider spoofing the marker; |
| 435 | // both get a 503 and change nothing. |
| 436 | var errSidecarAsleep = errors.New("instance asleep; sidecar requests do not wake it") |
| 437 | |
| 438 | // beginSidecarRequest is beginRequest for sidecar-marked traffic: pass it |
| 439 | // through to an awake instance WITHOUT registering activity — no in-flight |
| 440 | // count, no lastActive stamp (the sidecar's own long-lived WebSocket must not |
| 441 | // hold its instance awake; that non-count is the whole reason the pair can |
| 442 | // sleep) — and refuse, rather than wake, when the instance is asleep or |
| 443 | // stopping. |
| 444 | func (a *activatorSvc) beginSidecarRequest(ctx context.Context, rt Route) error { |
| 445 | deadline := time.NewTimer(wakeWaitBound) |
| 446 | defer deadline.Stop() |
| 447 | for { |
| 448 | a.mu.Lock() |
| 449 | e := a.entries[rt.Host] |
| 450 | if e == nil { |
| 451 | a.mu.Unlock() |
| 452 | return errSidecarAsleep |
| 453 | } |
| 454 | switch e.state { |
| 455 | case actAwake: |
| 456 | // Same dead-child / lost-lease detection as a member request, but |
| 457 | // the recovery stops at the stop: the loop then finds a stopping |
| 458 | // (or absent) entry and turns this request away instead of |
| 459 | // chaining into a re-wake. |
| 460 | if repErr := e.rep.Err(); repErr != nil || e.proc.exited() { |
| 461 | if repErr != nil { |
| 462 | log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr) |
| 463 | } else { |
| 464 | log.Printf("activator: instance %s exited unexpectedly — recycling", e.host) |
| 465 | } |
| 466 | a.beginStopLocked(e) |
| 467 | a.mu.Unlock() |
| 468 | continue |
| 469 | } |
| 470 | a.mu.Unlock() |
| 471 | return nil |
| 472 | |
| 473 | case actWaking: |
| 474 | // Joining a wake someone else started is fine — the co-spawned |
| 475 | // sidecar boots moments before its entry flips awake, so its first |
| 476 | // dial can land in that sliver; joining can't initiate anything. |
| 477 | // A failed wake just reads as asleep (the member path already |
| 478 | // logged it). |
| 479 | ready := e.ready |
| 480 | a.mu.Unlock() |
| 481 | select { |
| 482 | case <-ready: |
| 483 | if e.err != nil { |
| 484 | return errSidecarAsleep |
| 485 | } |
| 486 | continue |
| 487 | case <-ctx.Done(): |
| 488 | return ctx.Err() |
| 489 | case <-deadline.C: |
| 490 | return errWakeTimeout |
| 491 | } |
| 492 | |
| 493 | default: // actStopping — the stop exists to shed exactly this connection |
| 494 | a.mu.Unlock() |
| 495 | return errSidecarAsleep |
| 496 | } |
| 497 | } |
| 498 | } |
| 499 | |
| 500 | // hasLiveEntry reports whether host has an entry in any state — the mesh's |
| 501 | // "already living here" check (waking and stopping both count: a waking host |
| 502 | // is committed to this box, and a stopping one still holds the lease, so |
| 503 | // proxying it away would only trade a short wait for a loud lease race). |
| 504 | func (a *activatorSvc) hasLiveEntry(host string) bool { |
| 505 | a.mu.Lock() |
| 506 | defer a.mu.Unlock() |
| 507 | _, ok := a.entries[host] |
| 508 | return ok |
| 509 | } |
| 510 | |
| 511 | // awakeCount is this box's own count of genuinely serving hibernating |
| 512 | // children — actAwake only, not actWaking/actStopping transients — for the |
| 513 | // Fleet status page's per-box awake/asleep split (meshstats.go). Unlike |
| 514 | // deploy/status-publish.sh's own `ss`-probing of the LOCAL box (an external |
| 515 | // process with no other view into the router), this is the activator's |
| 516 | // authoritative in-memory state — no socket probing needed. |
| 517 | func (a *activatorSvc) awakeCount() int { |
| 518 | a.mu.Lock() |
| 519 | defer a.mu.Unlock() |
| 520 | n := 0 |
| 521 | for _, e := range a.entries { |
| 522 | if e.state == actAwake { |
| 523 | n++ |
| 524 | } |
| 525 | } |
| 526 | return n |
| 527 | } |
| 528 | |
| 529 | func (a *activatorSvc) releaseFunc(e *actEntry) func() { |
| 530 | var once sync.Once |
| 531 | return func() { |
| 532 | once.Do(func() { |
| 533 | a.mu.Lock() |
| 534 | e.inflight-- |
| 535 | e.lastActive = a.now() |
| 536 | a.mu.Unlock() |
| 537 | }) |
| 538 | } |
| 539 | } |
| 540 | |
| 541 | // wake drives asleep→awake: lease, restore-if-missing, spawn, replicate, |
| 542 | // co-spawn the sidecar. It runs on the first requester's goroutine but under |
| 543 | // its own generous timeout, detached from that request's context — queued |
| 544 | // requests still need the wake to finish even if the requester who started it |
| 545 | // hangs up. |
| 546 | func (a *activatorSvc) wake(e *actEntry) { |
| 547 | ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) |
| 548 | defer cancel() |
| 549 | w, err := a.doWake(ctx, e.route) |
| 550 | |
| 551 | a.mu.Lock() |
| 552 | if err != nil { |
| 553 | e.err = err |
| 554 | delete(a.entries, e.host) // next request retries from scratch |
| 555 | } else { |
| 556 | e.state = actAwake |
| 557 | e.rep, e.proc = w.rep, w.proc |
| 558 | e.sidecar, e.sidecarWanted = w.sidecar, w.sidecarWanted |
| 559 | e.sidecarStarted = a.now() |
| 560 | e.sidecarBackoff = 0 |
| 561 | e.sidecarRetryAt = time.Time{} // a wake-time spawn failure retries on the next sweep |
| 562 | e.lastActive = a.now() |
| 563 | // The build this child actually execed, captured at Start() inside the |
| 564 | // backend (not re-read here — a ship during the wake window would make a |
| 565 | // re-read name the new target while the process runs the old binary, and |
| 566 | // the sweep would never converge it). Empty = unknown → the sweep won't |
| 567 | // cycle. Fresh wake, so it can't be stale yet. |
| 568 | e.spawnVersion = w.spawnVersion |
| 569 | e.staleSince = time.Time{} |
| 570 | } |
| 571 | a.mu.Unlock() |
| 572 | close(e.ready) |
| 573 | } |
| 574 | |
| 575 | // wokenInstance is a successful wake's yield. sidecar is nil when the host has |
| 576 | // none staged — or when the spawn failed, which is deliberately not an error: |
| 577 | // members > bots, so a space whose bots won't start still serves (the sweeper |
| 578 | // retries via sidecarWanted). |
| 579 | type wokenInstance struct { |
| 580 | rep actRep |
| 581 | proc actProc |
| 582 | sidecar actProc |
| 583 | sidecarWanted bool |
| 584 | spawnVersion string // live/<host> target read at exec — for build convergence |
| 585 | } |
| 586 | |
| 587 | func (a *activatorSvc) doWake(ctx context.Context, rt Route) (*wokenInstance, error) { |
| 588 | host := rt.Host |
| 589 | dbPath := rt.DBPath |
| 590 | if dbPath == "" { |
| 591 | dbPath = filepath.Join(instanceDataDir, host+".db") |
| 592 | } |
| 593 | |
| 594 | // 1. The lease first: everything after this line is done as the prefix's |
| 595 | // single writer. Contention is not retried here — on one box it means a |
| 596 | // previous incarnation hasn't released; tell the client to come back after |
| 597 | // the TTL instead of stalling its request. |
| 598 | lease, err := a.backend.acquire(ctx, host) |
| 599 | if err != nil { |
| 600 | var held *errLeaseHeld |
| 601 | if errors.As(err, &held) { |
| 602 | log.Printf("activator: LEASE HELD for %s by %q — a previous incarnation hasn't released; refusing to wake (retry after %s)", |
| 603 | host, held.holder, held.retryAfter) |
| 604 | } |
| 605 | return nil, err |
| 606 | } |
| 607 | |
| 608 | // 2. Restore the DB from S3 when the local copy is gone. When it's still |
| 609 | // on disk it is the warm cache: on one box, with the lease held, the local |
| 610 | // copy can only be what we ourselves checkpointed at the last sleep. |
| 611 | if err := a.backend.restore(ctx, host, dbPath); err != nil { |
| 612 | _ = lease.release(ctx) |
| 613 | return nil, fmt.Errorf("restore %s: %w", host, err) |
| 614 | } |
| 615 | |
| 616 | // 3. Spawn before replication starts: a fresh instance creates its own DB |
| 617 | // on boot, so waiting for the socket guarantees the file exists for |
| 618 | // litestream to open. Writes that land in the gap are picked up by the |
| 619 | // replica's first sync. |
| 620 | proc, spawnVer, err := a.backend.start(ctx, rt) |
| 621 | if err != nil { |
| 622 | _ = lease.release(ctx) |
| 623 | return nil, fmt.Errorf("start %s: %w", host, err) |
| 624 | } |
| 625 | |
| 626 | // 4. Trade the held lease for running replication, at the route's |
| 627 | // replication-class sync interval (the per-plan durability ladder). |
| 628 | rep, err := lease.replicate(ctx, dbPath, replicationSyncInterval(rt.Replication)) |
| 629 | if err != nil { |
| 630 | stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| 631 | defer cancel() |
| 632 | _ = proc.stop(stopCtx) |
| 633 | _ = lease.release(ctx) |
| 634 | return nil, fmt.Errorf("replicate %s: %w", host, err) |
| 635 | } |
| 636 | |
| 637 | // 5. Co-spawn the bot sidecar, last — only once the instance is healthy |
| 638 | // and replicating, so the sidecar always boots against a live instance |
| 639 | // (its marked requests pass through the router while we're awake). A |
| 640 | // spawn failure is loud but never fails the wake: members come before |
| 641 | // bots, and the sweeper retries with backoff. |
| 642 | w := &wokenInstance{rep: rep, proc: proc, spawnVersion: spawnVer} |
| 643 | if a.backend.hasSidecar(host) { |
| 644 | w.sidecarWanted = true |
| 645 | if sc, err := a.backend.startSidecar(ctx, rt); err != nil { |
| 646 | log.Printf("activator: SIDECAR for %s failed to start: %v — serving without bots (the sweep retries)", host, err) |
| 647 | } else { |
| 648 | w.sidecar = sc |
| 649 | } |
| 650 | } |
| 651 | return w, nil |
| 652 | } |
| 653 | |
| 654 | // actStopTimeout bounds a graceful instance stop (the instance's own SIGTERM |
| 655 | // drain is 10s; leave headroom before the SIGKILL). |
| 656 | const actStopTimeout = 20 * time.Second |
| 657 | |
| 658 | // beginStopLocked flips an awake entry to stopping and kicks the stop off; the |
| 659 | // caller holds a.mu. In-flight requests are not waited for — a stop is only |
| 660 | // initiated when the instance is idle, dead, or has lost its lease, and in the |
| 661 | // latter two cases open requests are already doomed. |
| 662 | func (a *activatorSvc) beginStopLocked(e *actEntry) { |
| 663 | e.state = actStopping |
| 664 | e.stopped = make(chan struct{}) |
| 665 | go a.stop(e) |
| 666 | } |
| 667 | |
| 668 | // stop drives awake→asleep: stop the child (its SIGTERM drain is the final |
| 669 | // write), checkpoint what's on disk to S3 and release the lease, then forget |
| 670 | // the entry. The local DB stays in place — it's the warm cache for the next |
| 671 | // wake; S3 is the durable truth under the lease. |
| 672 | func (a *activatorSvc) stop(e *actEntry) { |
| 673 | // Co-stop, sidecar FIRST — before the next-due probe and the instance's |
| 674 | // own SIGTERM. Two reasons: the sidecar must never outlive its instance |
| 675 | // (an orphan's reconnect loop would be a standing wake source — the |
| 676 | // ping-pong this feature exists to prevent — and its member-bot charters |
| 677 | // would keep firing into a host we're about to stop), and with the bots |
| 678 | // gone first, the schedule the next-due probe reads can't shift under it. |
| 679 | if e.sidecar != nil { |
| 680 | scCtx, scCancel := context.WithTimeout(context.Background(), sidecarStopTimeout) |
| 681 | if err := e.sidecar.stop(scCtx); err != nil { |
| 682 | log.Printf("activator: stop sidecar %s: %v", e.host, err) |
| 683 | } |
| 684 | scCancel() |
| 685 | } |
| 686 | |
| 687 | // The due-time index: ask the instance for its earliest pending scheduled |
| 688 | // delivery BEFORE stopping it — the socket dies with the process. (When the |
| 689 | // stop was triggered by a dead child, the query just fails fast.) Any |
| 690 | // failure records 0, i.e. "sleeps until a request": a missed probe must |
| 691 | // degrade to pre-index behavior, never block or crash the sleep. A due time |
| 692 | // already in the past is recorded as-is and the sweeper wakes the host on |
| 693 | // its next tick — there is deliberately no minimum-sleep guard beyond that, |
| 694 | // because thrash is already impossible: an awake instance only sleeps after |
| 695 | // a full idle window, so wake/sleep cycles can never be spaced closer than |
| 696 | // cfg.Idle. |
| 697 | dueCtx, cancelDue := context.WithTimeout(context.Background(), 5*time.Second) |
| 698 | due, err := a.backend.nextDue(dueCtx, e.route) |
| 699 | cancelDue() |
| 700 | if err != nil { |
| 701 | log.Printf("activator: next-due probe %s: %v — recording none (it sleeps until a request)", e.host, err) |
| 702 | due = 0 |
| 703 | } |
| 704 | if err := a.due.setNextDue(e.host, due); err != nil { |
| 705 | log.Printf("activator: record next-due %s: %v", e.host, err) |
| 706 | } else if due > 0 { |
| 707 | log.Printf("activator: %s next scheduled delivery at %s — will wake for it", e.host, time.Unix(due, 0).UTC().Format(time.RFC3339)) |
| 708 | } |
| 709 | |
| 710 | // Separate budgets: a SIGTERM drain that runs long must not eat the final |
| 711 | // checkpoint's time — the S3 sync is the part that makes sleep durable. |
| 712 | procCtx, cancelProc := context.WithTimeout(context.Background(), actStopTimeout) |
| 713 | err = e.proc.stop(procCtx) |
| 714 | cancelProc() |
| 715 | if err != nil { |
| 716 | log.Printf("activator: stop %s: %v", e.host, err) |
| 717 | } |
| 718 | // Close = final sync + lease release. After a lost lease the release part |
| 719 | // fails harmlessly (someone else owns the lock); the sync still preserves |
| 720 | // a local checkpoint. |
| 721 | ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| 722 | defer cancel() |
| 723 | if err := e.rep.Close(ctx); err != nil { |
| 724 | log.Printf("activator: final checkpoint %s: %v", e.host, err) |
| 725 | } |
| 726 | a.mu.Lock() |
| 727 | delete(a.entries, e.host) |
| 728 | a.mu.Unlock() |
| 729 | close(e.stopped) |
| 730 | log.Printf("activator: %s hibernated", e.host) |
| 731 | } |
| 732 | |
| 733 | // start launches the background sweeper that puts idle instances to sleep and |
| 734 | // stands down lost leases/dead children between requests. |
| 735 | func (a *activatorSvc) start() { |
| 736 | a.sweepStop = make(chan struct{}) |
| 737 | a.sweepDone = make(chan struct{}) |
| 738 | // Sweep well inside the idle window so a sleep happens promptly after it |
| 739 | // elapses; clamp so second-granular test idles still sweep sanely. |
| 740 | interval := a.cfg.Idle / 4 |
| 741 | if interval < 100*time.Millisecond { |
| 742 | interval = 100 * time.Millisecond |
| 743 | } |
| 744 | if interval > 30*time.Second { |
| 745 | interval = 30 * time.Second |
| 746 | } |
| 747 | go func() { |
| 748 | defer close(a.sweepDone) |
| 749 | t := time.NewTicker(interval) |
| 750 | defer t.Stop() |
| 751 | for { |
| 752 | select { |
| 753 | case <-a.sweepStop: |
| 754 | return |
| 755 | case <-t.C: |
| 756 | a.sweepOnce() |
| 757 | } |
| 758 | } |
| 759 | }() |
| 760 | } |
| 761 | |
| 762 | // sweepOnce examines every awake instance once: idle ones hibernate; a lost |
| 763 | // lease or dead child stands down immediately regardless of activity. It then |
| 764 | // checks the due-time index and wakes any sleeping host whose scheduled |
| 765 | // delivery has come due. |
| 766 | func (a *activatorSvc) sweepOnce() { |
| 767 | now := a.now() |
| 768 | // The due query runs before the lock — it's a registry read, and nothing |
| 769 | // under a.mu should wait on SQLite. |
| 770 | dueRoutes, err := a.due.dueInstances(now.Unix()) |
| 771 | if err != nil { |
| 772 | log.Printf("activator: due-time sweep: %v", err) |
| 773 | } |
| 774 | a.mu.Lock() |
| 775 | cycled := 0 // build-convergence cycles started this tick (rate-limited) |
| 776 | for _, e := range a.entries { |
| 777 | if e.state != actAwake { |
| 778 | continue |
| 779 | } |
| 780 | if err := e.rep.Err(); err != nil { |
| 781 | log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, err) |
| 782 | a.beginStopLocked(e) |
| 783 | continue |
| 784 | } |
| 785 | if e.proc.exited() { |
| 786 | log.Printf("activator: instance %s exited unexpectedly — recycling", e.host) |
| 787 | a.beginStopLocked(e) |
| 788 | continue |
| 789 | } |
| 790 | if e.inflight == 0 && now.Sub(e.lastActive) >= a.cfg.Idle { |
| 791 | log.Printf("activator: %s idle %s — hibernating", e.host, now.Sub(e.lastActive).Truncate(time.Second)) |
| 792 | a.beginStopLocked(e) |
| 793 | continue |
| 794 | } |
| 795 | // Converge onto a shipped build. A running child never re-reads |
| 796 | // live/<host>, so once a ship or `deploy --version` repoints the symlink |
| 797 | // this awake child is serving a stale binary. Idle spaces converge for |
| 798 | // free on their next wake; this cycle is for the always-awake ones — |
| 799 | // notably every bot-sidecar host, which never sleeps — that otherwise |
| 800 | // serve the old build indefinitely. Gracefully cycle: beginStopLocked |
| 801 | // drains via SIGTERM, takes the final checkpoint, releases the lease, and |
| 802 | // leaves the host asleep; a member request (or the due-time index for |
| 803 | // scheduled charters) then wakes it onto the new binary — the same clean |
| 804 | // cycle `systemctl restart` gives a unit-managed host. Unlike idle |
| 805 | // hibernation this can fire with a request in flight; the stop still |
| 806 | // drains HTTP gracefully and reconnecting clients coalesce onto the fresh |
| 807 | // wake, so a long upload past the ~10s drain is cut exactly as a restart |
| 808 | // would cut it. A pinned canary never trips this: the updater leaves its |
| 809 | // symlink alone, so spawnVersion still matches. See |
| 810 | // docs/awake-instance-convergence.md. |
| 811 | cur, readErr := a.liveVer(e.host) |
| 812 | cycle, next := convergeDecision(cur, readErr, e.spawnVersion, e.staleSince, now, convergeGrace) |
| 813 | e.staleSince = next |
| 814 | if cycle && cycled < maxConvergeCyclesPerSweep { |
| 815 | log.Printf("activator: %s serving %s but live symlink is now %s — cycling onto the new build", e.host, e.spawnVersion, cur) |
| 816 | a.beginStopLocked(e) |
| 817 | cycled++ |
| 818 | continue |
| 819 | } |
| 820 | // Sidecar supervision — Restart=always, sweep-driven, no supervisor |
| 821 | // goroutine: a sidecar that died (or failed to spawn at wake) while |
| 822 | // its instance stays awake is respawned with doubling, capped backoff; |
| 823 | // a run past sidecarHealthyRun resets it. Spawned under a.mu |
| 824 | // deliberately: an exec is quick, and the lock is what makes a restart |
| 825 | // impossible to race with a stop (sidecar fields only change in the |
| 826 | // awake state, and every stop flips the state under this same lock |
| 827 | // first). |
| 828 | // |
| 829 | // A bot env staged MID-FLIGHT (attach-bot on an awake child — the AI |
| 830 | // opt-in, #858) is ADOPTED here rather than waiting for the next |
| 831 | // wake, so opting in takes effect in seconds. Safe against |
| 832 | // double-running: attach-bot never enables the systemd unit for a |
| 833 | // hibernating host, and only hibernating hosts are activator |
| 834 | // children — the staged sidecar has exactly one possible runner. |
| 835 | if !e.sidecarWanted && a.backend.hasSidecar(e.host) { |
| 836 | e.sidecarWanted = true |
| 837 | log.Printf("activator: sidecar staged for awake %s — adopting (AI opt-in)", e.host) |
| 838 | } |
| 839 | if e.sidecarWanted && (e.sidecar == nil || e.sidecar.exited()) && !now.Before(e.sidecarRetryAt) { |
| 840 | if e.sidecar != nil && now.Sub(e.sidecarStarted) >= sidecarHealthyRun { |
| 841 | e.sidecarBackoff = 0 |
| 842 | } |
| 843 | if e.sidecarBackoff == 0 { |
| 844 | e.sidecarBackoff = sidecarRestartBase |
| 845 | } else { |
| 846 | e.sidecarBackoff = min(e.sidecarBackoff*2, sidecarRestartMax) |
| 847 | } |
| 848 | e.sidecarRetryAt = now.Add(e.sidecarBackoff) |
| 849 | sc, err := a.backend.startSidecar(context.Background(), e.route) |
| 850 | if err != nil { |
| 851 | log.Printf("activator: SIDECAR restart for %s failed: %v (next attempt in %s)", e.host, err, e.sidecarBackoff) |
| 852 | e.sidecar = nil |
| 853 | continue |
| 854 | } |
| 855 | log.Printf("activator: sidecar for %s was down — restarted (backoff now %s)", e.host, e.sidecarBackoff) |
| 856 | e.sidecar = sc |
| 857 | e.sidecarStarted = now |
| 858 | } |
| 859 | } |
| 860 | // Due-time wakes: a request-less wake through the same singleflight state |
| 861 | // machine, so a client request arriving mid-wake just coalesces. Only hosts |
| 862 | // with NO live entry are woken — an awake host is already serving its |
| 863 | // schedule; a waking one is on its way; a stopping one will re-record its |
| 864 | // next_due when the stop completes and be picked up on a later tick. The |
| 865 | // registry's stale past value while the host is up is harmless for the same |
| 866 | // reason. A failed due wake (e.g. a lease still held by a crashed |
| 867 | // incarnation) removes the entry, and the index — only rewritten at the |
| 868 | // next successful sleep — naturally retries it every tick until it works. |
| 869 | for _, rt := range dueRoutes { |
| 870 | if _, live := a.entries[rt.Host]; live { |
| 871 | continue |
| 872 | } |
| 873 | // Mesh: only the HRW owner due-wakes a host — otherwise every box |
| 874 | // sweeping a due entry would race a wake per tick (the lease would let |
| 875 | // exactly one win, but loudly, forever). If the owner box is dead its |
| 876 | // heartbeat goes stale and the next tick's membership makes someone |
| 877 | // else the owner — eventual, bounded by staleAfter, and a scheduled |
| 878 | // delivery is late by at most that. (Today each box records next_due |
| 879 | // in its own registry, so the recorder and the owner are usually the |
| 880 | // same box; a host whose ownership moved while asleep waits for the |
| 881 | // registry-sync work — or a member request — to fire. Known gap, |
| 882 | // deliberately deferred with the shared-registry question.) |
| 883 | if a.mesh != nil && !a.mesh.owns(rt.Host) { |
| 884 | continue |
| 885 | } |
| 886 | log.Printf("activator: %s has a scheduled delivery due — waking it", rt.Host) |
| 887 | e := &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})} |
| 888 | a.entries[rt.Host] = e |
| 889 | go a.wake(e) |
| 890 | } |
| 891 | a.mu.Unlock() |
| 892 | } |
| 893 | |
| 894 | // retireExcept stops (final checkpoint + lease release) every live entry whose |
| 895 | // host is not in keep — the reload-time reconcile behind `lchat hibernate |
| 896 | // <host> --off` (and, as a bonus, route removal while awake). The flip clears |
| 897 | // the route's flag and SIGHUPs the router; the map swap stops routing requests |
| 898 | // through the activator, and this puts the child down and gives the S3 lease |
| 899 | // back so the host's systemd unit can bind the socket. It reuses the ordinary |
| 900 | // stop path — the narrowest hook that exists — so the release still records |
| 901 | // next_due and checkpoints to S3 like any sleep (the flip-off's caller waits |
| 902 | // for the socket to go quiet, not for the checkpoint; see hibernateOff). |
| 903 | // In-flight requests on a retired child are cut off as a deploy restart would |
| 904 | // cut them — the flip is a deliberate operator action. |
| 905 | func (a *activatorSvc) retireExcept(keep map[string]bool) { |
| 906 | a.mu.Lock() |
| 907 | var waking []*actEntry |
| 908 | for _, e := range a.entries { |
| 909 | if keep[e.host] { |
| 910 | continue |
| 911 | } |
| 912 | switch e.state { |
| 913 | case actAwake: |
| 914 | log.Printf("activator: %s is no longer hibernate-managed — releasing", e.host) |
| 915 | a.beginStopLocked(e) |
| 916 | case actWaking: |
| 917 | // A wake can't be stopped mid-flight; finish the retire once it |
| 918 | // resolves (below, off-lock). |
| 919 | waking = append(waking, e) |
| 920 | case actStopping: |
| 921 | // Already on its way out. |
| 922 | } |
| 923 | } |
| 924 | a.mu.Unlock() |
| 925 | for _, e := range waking { |
| 926 | go func(e *actEntry) { |
| 927 | <-e.ready |
| 928 | a.mu.Lock() |
| 929 | defer a.mu.Unlock() |
| 930 | // Only if THIS entry is still the live one and its wake succeeded — |
| 931 | // a failed wake already removed itself, and a fresh entry for the |
| 932 | // same host means the route was re-flagged in the meantime. |
| 933 | if a.entries[e.host] == e && e.state == actAwake { |
| 934 | log.Printf("activator: %s is no longer hibernate-managed — releasing (wake finished first)", e.host) |
| 935 | a.beginStopLocked(e) |
| 936 | } |
| 937 | }(e) |
| 938 | } |
| 939 | } |
| 940 | |
| 941 | // shutdown checkpoints and stops every child — the router going down takes its |
| 942 | // instances with it (they're its children; leaving them orphaned would strand |
| 943 | // the leases until TTL expiry and leave the sockets bound against the next |
| 944 | // incarnation's spawns). Blocks until all stops complete or ctx expires. |
| 945 | func (a *activatorSvc) shutdown(ctx context.Context) { |
| 946 | // Drain step 1, order-pinned: leave the mesh membership FIRST — the |
| 947 | // heartbeat is deleted before any checkpoint or lease release below, so |
| 948 | // peers stop proxying here immediately and their wakes succeed the moment |
| 949 | // each lease releases. Deploys, box failure drills, and scale-down all |
| 950 | // ride this same path. |
| 951 | if a.mesh != nil { |
| 952 | a.mesh.stop(ctx) |
| 953 | } |
| 954 | if a.sweepStop != nil { |
| 955 | close(a.sweepStop) |
| 956 | <-a.sweepDone |
| 957 | } |
| 958 | a.mu.Lock() |
| 959 | var waits []chan struct{} |
| 960 | for _, e := range a.entries { |
| 961 | switch e.state { |
| 962 | case actAwake: |
| 963 | a.beginStopLocked(e) |
| 964 | waits = append(waits, e.stopped) |
| 965 | case actStopping: |
| 966 | waits = append(waits, e.stopped) |
| 967 | case actWaking: |
| 968 | // A mid-flight wake resolves on its own; its instance will be |
| 969 | // orphaned only if it wins the race with process exit — acceptable |
| 970 | // for the rare shutdown-during-wake, and the lease TTL still |
| 971 | // bounds the damage. |
| 972 | } |
| 973 | } |
| 974 | a.mu.Unlock() |
| 975 | for _, w := range waits { |
| 976 | select { |
| 977 | case <-w: |
| 978 | case <-ctx.Done(): |
| 979 | return |
| 980 | } |
| 981 | } |
| 982 | } |
| 983 | |
| 984 | // serveHTTP answers one proxied request for a hibernating instance: wake (or |
| 985 | // join a wake), proxy, stamp activity. Lease contention and wake timeouts are |
| 986 | // 503s — "sleeping spaces are slow to wake", never a 5xx cascade of retries. |
| 987 | func (a *activatorSvc) serveHTTP(w http.ResponseWriter, r *http.Request, rt Route, proxy http.Handler) { |
| 988 | // Sidecar-marked traffic (bot.SidecarHeader — the Go sidecar stamps every |
| 989 | // REST call and its WebSocket dial) is activity-invisible: proxied while |
| 990 | // the instance is awake, refused while it isn't, never a wake. The |
| 991 | // co-spawned sidecar dials its instance's public URL through this very |
| 992 | // router (the box pins its hosted names to loopback), so without this its |
| 993 | // keepalives would read as member activity — the instance could never |
| 994 | // idle — and after a sleep its reconnect loop would be the wake source: |
| 995 | // ping-pong forever. Spoofing the marker is harmless by construction: it |
| 996 | // only DOWNGRADES the sender's own requests (no activity credit; 503 while |
| 997 | // asleep). It cannot suppress other clients' activity stamps and cannot |
| 998 | // wake a sleeping host, so an outsider sending it gains nothing. |
| 999 | if r.Header.Get(bot.SidecarHeader) != "" { |
| 1000 | if err := a.beginSidecarRequest(r.Context(), rt); err != nil { |
| 1001 | if errors.Is(err, context.Canceled) { |
| 1002 | return // the sidecar hung up while queued on a wake |
| 1003 | } |
| 1004 | w.Header().Set("Retry-After", "5") |
| 1005 | http.Error(w, "instance is asleep; sidecar requests do not wake it", http.StatusServiceUnavailable) |
| 1006 | return |
| 1007 | } |
| 1008 | proxy.ServeHTTP(w, r) |
| 1009 | return |
| 1010 | } |
| 1011 | |
| 1012 | // Mesh placement (mesh.go): a member request for a host with no live local |
| 1013 | // entry and an HRW owner on another live box is proxied there instead of |
| 1014 | // woken here. A forwarded request is never re-proxied (one hop max — see |
| 1015 | // meshForwardedHeader). A host with ANY live local entry stays local: an |
| 1016 | // awake instance finishes its natural life where it is, and placement |
| 1017 | // converges at the next sleep boundary — its state is already in S3, so |
| 1018 | // "migrating" is just waking on the new owner. The check-then-proxy is |
| 1019 | // deliberately not atomic with the entry map: a race sends one request the |
| 1020 | // long way, or wakes a not-the-hint box, and the lease arbitrates exactly |
| 1021 | // as it would for real membership disagreement. (Sidecar-marked traffic |
| 1022 | // above never routes through the mesh — a co-spawned sidecar's instance is |
| 1023 | // by construction local, and a stray marked request for a host living |
| 1024 | // elsewhere is exactly the reconnect-racing-a-stop case the 503 exists for.) |
| 1025 | if a.mesh != nil && r.Header.Get(meshForwardedHeader) == "" && !a.hasLiveEntry(rt.Host) { |
| 1026 | if peer, remote := a.mesh.remoteOwner(rt.Host); remote { |
| 1027 | a.mesh.proxyTo(w, r, peer) |
| 1028 | return |
| 1029 | } |
| 1030 | } |
| 1031 | |
| 1032 | release, err := a.beginRequest(r.Context(), rt) |
| 1033 | if err != nil { |
| 1034 | var held *errLeaseHeld |
| 1035 | switch { |
| 1036 | case errors.As(err, &held): |
| 1037 | retry := int(held.retryAfter/time.Second) + 1 |
| 1038 | w.Header().Set("Retry-After", strconv.Itoa(retry)) |
| 1039 | http.Error(w, "instance is waking elsewhere; retry shortly", http.StatusServiceUnavailable) |
| 1040 | case errors.Is(err, errWakeTimeout): |
| 1041 | w.Header().Set("Retry-After", "5") |
| 1042 | http.Error(w, "instance is waking; retry shortly", http.StatusServiceUnavailable) |
| 1043 | case errors.Is(err, context.Canceled): |
| 1044 | // The client hung up while queued; nothing to write. |
| 1045 | default: |
| 1046 | log.Printf("activator: wake %s: %v", rt.Host, err) |
| 1047 | http.Error(w, "instance failed to wake", http.StatusServiceUnavailable) |
| 1048 | } |
| 1049 | return |
| 1050 | } |
| 1051 | defer release() |
| 1052 | proxy.ServeHTTP(w, r) |
| 1053 | } |
| 1054 | |
| 1055 | // --- the real backend --------------------------------------------------------- |
| 1056 | |
| 1057 | // s3Backend is the production actBackend: leases + restore via the |
| 1058 | // internal/activator litestream primitives, processes via exec of the same |
| 1059 | // versioned binary the systemd unit would run. |
| 1060 | type s3Backend struct { |
| 1061 | cfg hibernateConfig |
| 1062 | // materialize, when non-nil, stages the running binary and points a |
| 1063 | // host's live symlink at it via the control daemon (cloud.go's |
| 1064 | // controldMaterializeFunc) instead of doing it directly — needed because |
| 1065 | // this process runs sandboxed and can't write to /opt/lchat itself. nil |
| 1066 | // falls back to the direct calls (start's else branch), today's |
| 1067 | // long-standing behavior for any box that hasn't opted in. |
| 1068 | materialize func(ctx context.Context, host string) error |
| 1069 | } |
| 1070 | |
| 1071 | func (b *s3Backend) acquire(ctx context.Context, host string) (actLease, error) { |
| 1072 | held, err := activator.AcquireLease(ctx, b.cfg.Bucket, b.cfg.hibernatePrefix(host), b.cfg.Region) |
| 1073 | if err != nil { |
| 1074 | var lx *litestream.LeaseExistsError |
| 1075 | if errors.As(err, &lx) { |
| 1076 | retry := time.Until(lx.ExpiresAt) |
| 1077 | if retry < time.Second { |
| 1078 | retry = time.Second |
| 1079 | } |
| 1080 | return nil, &errLeaseHeld{holder: lx.Owner, retryAfter: retry} |
| 1081 | } |
| 1082 | return nil, err |
| 1083 | } |
| 1084 | return &s3Lease{held: held}, nil |
| 1085 | } |
| 1086 | |
| 1087 | type s3Lease struct{ held *activator.HeldLease } |
| 1088 | |
| 1089 | func (l *s3Lease) replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (actRep, error) { |
| 1090 | return l.held.Replicate(ctx, dbPath, syncInterval) |
| 1091 | } |
| 1092 | func (l *s3Lease) release(ctx context.Context) error { return l.held.Release(ctx) } |
| 1093 | |
| 1094 | func (b *s3Backend) restore(ctx context.Context, host, dbPath string) error { |
| 1095 | if _, err := os.Stat(dbPath); err == nil { |
| 1096 | return nil // warm cache — see doWake |
| 1097 | } else if !os.IsNotExist(err) { |
| 1098 | return err |
| 1099 | } |
| 1100 | // The DB is gone, so any leftover litestream sidecar state (the |
| 1101 | // <db>-litestream meta dir, WAL/SHM) describes a database that no longer |
| 1102 | // exists — litestream's own recovery guidance is to delete it. Clear it so |
| 1103 | // the restored copy starts a clean lineage. |
| 1104 | for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} { |
| 1105 | _ = os.Remove(p) |
| 1106 | } |
| 1107 | _ = os.RemoveAll(dbPath + litestream.MetaDirSuffix) |
| 1108 | |
| 1109 | restored, err := activator.Restore(ctx, dbPath, b.cfg.Bucket, b.cfg.hibernatePrefix(host), b.cfg.Region) |
| 1110 | if err != nil { |
| 1111 | return err |
| 1112 | } |
| 1113 | if restored { |
| 1114 | log.Printf("activator: restored %s from s3://%s/%s", host, b.cfg.Bucket, b.cfg.hibernatePrefix(host)) |
| 1115 | } else { |
| 1116 | // Never replicated: a brand-new space boots fresh and creates its own |
| 1117 | // DB — this is the future register-only create path (provisioning as a |
| 1118 | // metadata write) landing on its first wake. |
| 1119 | log.Printf("activator: no replica for %s in s3://%s/%s — booting fresh", host, b.cfg.Bucket, b.cfg.hibernatePrefix(host)) |
| 1120 | } |
| 1121 | return nil |
| 1122 | } |
| 1123 | |
| 1124 | // nextDue asks the (still-running) instance for its earliest pending scheduled |
| 1125 | // delivery over its socket. The endpoint is admin-gated like every instance |
| 1126 | // data endpoint, so present the instance's own admin token — read from the same |
| 1127 | // env file the spawn path hands the child (the activator holds it either way). |
| 1128 | func (b *s3Backend) nextDue(ctx context.Context, rt Route) (int64, error) { |
| 1129 | fileEnv, err := readEnvFile(filepath.Join(instanceDataDir, rt.Host+".env")) |
| 1130 | if err != nil { |
| 1131 | return 0, fmt.Errorf("read env file: %w", err) |
| 1132 | } |
| 1133 | token := fileEnv["LCHAT_ADMIN_TOKEN"] |
| 1134 | if token == "" { |
| 1135 | return 0, errors.New("no LCHAT_ADMIN_TOKEN in env file") |
| 1136 | } |
| 1137 | req, err := http.NewRequestWithContext(ctx, "GET", "http://lchat/api/next-due", nil) |
| 1138 | if err != nil { |
| 1139 | return 0, err |
| 1140 | } |
| 1141 | req.Header.Set("Authorization", "Bearer "+token) |
| 1142 | resp, err := socketClient(rt.Socket).Do(req) |
| 1143 | if err != nil { |
| 1144 | return 0, err |
| 1145 | } |
| 1146 | defer resp.Body.Close() |
| 1147 | if resp.StatusCode != http.StatusOK { |
| 1148 | return 0, fmt.Errorf("GET /api/next-due: HTTP %d", resp.StatusCode) |
| 1149 | } |
| 1150 | var body struct { |
| 1151 | Due int64 `json:"due"` |
| 1152 | } |
| 1153 | if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<12)).Decode(&body); err != nil { |
| 1154 | return 0, fmt.Errorf("decode /api/next-due: %w", err) |
| 1155 | } |
| 1156 | return body.Due, nil |
| 1157 | } |
| 1158 | |
| 1159 | // spawnHealthTimeout bounds how long a spawned instance may take to bind its |
| 1160 | // socket and answer. A var so tests can shrink the failure path. |
| 1161 | var spawnHealthTimeout = 30 * time.Second |
| 1162 | |
| 1163 | // ensureLiveLink materializes host's live symlink (/opt/lchat/live/<host> → |
| 1164 | // a staged version) if this box hasn't already — the future create-is-a-row |
| 1165 | // flow, or (found live, 2026-07-13) a mesh-replicated route, which only ever |
| 1166 | // inserts a registry row (meshreplicate.go), never a symlink. Split out of |
| 1167 | // s3Backend.start for focused testing: exercising this doesn't need a real |
| 1168 | // spawned process, just the filesystem. |
| 1169 | // |
| 1170 | // Prefers routing the write through controld (materialize, non-nil) over |
| 1171 | // doing it directly: this process runs sandboxed (ProtectSystem=strict) and |
| 1172 | // can't write to /opt/lchat itself — a direct call here only ever worked by |
| 1173 | // accident, when some other, unsandboxed path (add-instance) had already |
| 1174 | // materialized every host this box would ever wake. |
| 1175 | func ensureLiveLink(ctx context.Context, host string, materialize func(ctx context.Context, host string) error) error { |
| 1176 | bin := liveLink(host) |
| 1177 | if _, err := os.Stat(bin); err == nil { |
| 1178 | return nil |
| 1179 | } |
| 1180 | if materialize != nil { |
| 1181 | if err := materialize(ctx, host); err != nil { |
| 1182 | return fmt.Errorf("materialize via controld: %w", err) |
| 1183 | } |
| 1184 | return nil |
| 1185 | } |
| 1186 | ver, err := ensureSelfStaged() |
| 1187 | if err != nil { |
| 1188 | return fmt.Errorf("stage binary: %w", err) |
| 1189 | } |
| 1190 | if err := pointLive(host, ver); err != nil { |
| 1191 | return fmt.Errorf("point live symlink: %w", err) |
| 1192 | } |
| 1193 | return nil |
| 1194 | } |
| 1195 | |
| 1196 | func (b *s3Backend) start(ctx context.Context, rt Route) (actProc, string, error) { |
| 1197 | host := rt.Host |
| 1198 | dbPath := rt.DBPath |
| 1199 | if dbPath == "" { |
| 1200 | dbPath = filepath.Join(instanceDataDir, host+".db") |
| 1201 | } |
| 1202 | |
| 1203 | // Exec exactly what the systemd unit would: the per-instance live symlink |
| 1204 | // (/opt/lchat/live/<host> → a staged version), same --socket/--db flags, |
| 1205 | // same EnvironmentFile. ensureLiveLink materializes it first if this box |
| 1206 | // never has (a mesh-replicated route, or the future create-is-a-row flow). |
| 1207 | bin := liveLink(host) |
| 1208 | if err := ensureLiveLink(ctx, host, b.materialize); err != nil { |
| 1209 | return nil, "", err |
| 1210 | } |
| 1211 | |
| 1212 | // The env file is optional, like the unit's "-" prefix. |
| 1213 | fileEnv, err := readEnvFile(filepath.Join(instanceDataDir, host+".env")) |
| 1214 | if err != nil && !os.IsNotExist(err) { |
| 1215 | return nil, "", fmt.Errorf("read env file: %w", err) |
| 1216 | } |
| 1217 | |
| 1218 | cmd := exec.Command(bin, "--socket", rt.Socket, "--db", dbPath) |
| 1219 | cmd.Env = childEnv(fileEnv) |
| 1220 | // The child logs through the router's own stderr/stdout (journald under |
| 1221 | // systemd); there's no per-instance unit to own a log stream anymore. |
| 1222 | cmd.Stdout = os.Stdout |
| 1223 | cmd.Stderr = os.Stderr |
| 1224 | // Record the build we're about to exec — read the symlink immediately before |
| 1225 | // Start() so it names the version the kernel will resolve. Reading it here, |
| 1226 | // rather than back in wake() after the health wait, closes a real window: a |
| 1227 | // concurrent ship's pointLive during the up-to-30s wait would otherwise make |
| 1228 | // us record the NEW target while this child runs the OLD binary, and the |
| 1229 | // sweep would then think it's already converged and never cycle it. The |
| 1230 | // residual sub-exec race (a repoint between this read and Start) is in the |
| 1231 | // safe direction — we'd record the older version, so the sweep cycles once |
| 1232 | // too often rather than never. Empty on a read error: the sweep won't cycle. |
| 1233 | spawnVer, _ := liveVersion(host) |
| 1234 | if err := cmd.Start(); err != nil { |
| 1235 | return nil, "", err |
| 1236 | } |
| 1237 | |
| 1238 | p := &childProc{cmd: cmd, done: make(chan struct{})} |
| 1239 | go func() { |
| 1240 | p.err = cmd.Wait() |
| 1241 | close(p.done) |
| 1242 | }() |
| 1243 | |
| 1244 | if err := waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout); err != nil { |
| 1245 | stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| 1246 | defer cancel() |
| 1247 | _ = p.stop(stopCtx) |
| 1248 | return nil, "", err |
| 1249 | } |
| 1250 | return p, spawnVer, nil |
| 1251 | } |
| 1252 | |
| 1253 | // childEnv approximates a systemd unit's environment from inside the router: |
| 1254 | // inherit the process env but strip every LCHAT_* var first, so the router's |
| 1255 | // own config (hibernation, registry, wildcard tokens) can never leak into a |
| 1256 | // child; then layer the child's EnvironmentFile on top. A nil fileEnv (no env |
| 1257 | // file, like the unit's "-" prefix) yields just the cleaned inherit. |
| 1258 | func childEnv(fileEnv map[string]string) []string { |
| 1259 | env := make([]string, 0, 32) |
| 1260 | for _, kv := range os.Environ() { |
| 1261 | if strings.HasPrefix(kv, "LCHAT_") { |
| 1262 | continue |
| 1263 | } |
| 1264 | env = append(env, kv) |
| 1265 | } |
| 1266 | for k, v := range fileEnv { |
| 1267 | env = append(env, k+"="+v) |
| 1268 | } |
| 1269 | return env |
| 1270 | } |
| 1271 | |
| 1272 | // hasSidecar: a host has a bot sidecar iff its sidecar EnvironmentFile exists |
| 1273 | // — the same /data/<host>-bot.env convention add-bot writes and |
| 1274 | // lchat-bot@.service reads. Derived from the filesystem on purpose, the way |
| 1275 | // the units do it: no registry column to drift from what's actually staged. |
| 1276 | func (b *s3Backend) hasSidecar(host string) bool { |
| 1277 | _, err := os.Stat(botEnvPath(host)) |
| 1278 | return err == nil |
| 1279 | } |
| 1280 | |
| 1281 | // startSidecar mirrors lchat-bot@.service: exec the host's live symlink with |
| 1282 | // `bot run` under the sidecar's EnvironmentFile (LCHAT_BASE, session paths, |
| 1283 | // invites, optional ANTHROPIC_API_KEY), LCHAT_*-clean like the instance |
| 1284 | // spawn. No health wait — the sidecar is a pure outbound client with no |
| 1285 | // socket to probe; its own connect/backoff loop absorbs a slow instance. The |
| 1286 | // live symlink exists by now: the instance spawn staged it. The ctx is unused |
| 1287 | // (an exec doesn't block); it's here for the backend seam's symmetry. |
| 1288 | func (b *s3Backend) startSidecar(_ context.Context, rt Route) (actProc, error) { |
| 1289 | fileEnv, err := readEnvFile(botEnvPath(rt.Host)) |
| 1290 | if err != nil { |
| 1291 | return nil, fmt.Errorf("read bot env: %w", err) |
| 1292 | } |
| 1293 | cmd := exec.Command(liveLink(rt.Host), "bot", "run") |
| 1294 | cmd.Env = childEnv(fileEnv) |
| 1295 | // Like the instance child: logs through the router's own streams |
| 1296 | // (journald under systemd) — no per-host unit owns a log stream anymore. |
| 1297 | cmd.Stdout = os.Stdout |
| 1298 | cmd.Stderr = os.Stderr |
| 1299 | if err := cmd.Start(); err != nil { |
| 1300 | return nil, err |
| 1301 | } |
| 1302 | p := &childProc{cmd: cmd, done: make(chan struct{})} |
| 1303 | go func() { |
| 1304 | p.err = cmd.Wait() |
| 1305 | close(p.done) |
| 1306 | }() |
| 1307 | return p, nil |
| 1308 | } |
| 1309 | |
| 1310 | // waitHealthyCtx is waitHealthy bounded by ctx as well as its own timeout, so |
| 1311 | // an aborted wake doesn't keep probing a socket nobody wants anymore. |
| 1312 | func waitHealthyCtx(ctx context.Context, socket string, timeout time.Duration) error { |
| 1313 | done := make(chan error, 1) |
| 1314 | go func() { done <- waitHealthy(socket, timeout) }() |
| 1315 | select { |
| 1316 | case err := <-done: |
| 1317 | return err |
| 1318 | case <-ctx.Done(): |
| 1319 | return ctx.Err() |
| 1320 | } |
| 1321 | } |
| 1322 | |
| 1323 | // childProc is a spawned instance: SIGTERM starts the instance's own graceful |
| 1324 | // drain (main.go's shutdown path); SIGKILL only if it overstays. |
| 1325 | type childProc struct { |
| 1326 | cmd *exec.Cmd |
| 1327 | done chan struct{} |
| 1328 | err error |
| 1329 | } |
| 1330 | |
| 1331 | func (p *childProc) exited() bool { |
| 1332 | select { |
| 1333 | case <-p.done: |
| 1334 | return true |
| 1335 | default: |
| 1336 | return false |
| 1337 | } |
| 1338 | } |
| 1339 | |
| 1340 | func (p *childProc) stop(ctx context.Context) error { |
| 1341 | if p.exited() { |
| 1342 | return nil |
| 1343 | } |
| 1344 | if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil { |
| 1345 | if errors.Is(err, os.ErrProcessDone) { // lost the race with its own exit |
| 1346 | <-p.done |
| 1347 | return nil |
| 1348 | } |
| 1349 | return err |
| 1350 | } |
| 1351 | select { |
| 1352 | case <-p.done: |
| 1353 | return nil |
| 1354 | case <-ctx.Done(): |
| 1355 | _ = p.cmd.Process.Kill() |
| 1356 | <-p.done |
| 1357 | return fmt.Errorf("instance did not exit on SIGTERM; killed") |
| 1358 | } |
| 1359 | } |
| 1360 | |
amadan