amadan

eleven / messenger Public

server: instance hibernation on one box — the activator (#550 Part 4 v2, Phase 2) (#641)

Feature-gated on LCHAT_HIBERNATE_BUCKET (+ _REGION/_PREFIX/_IDLE) on the
router process; unset keeps today's behavior byte-identical. With the gate
on, instance routes stop being always-on systemd backends and become the
router's own children, woken on demand:

- Wake: request for a host with no live process → claim the per-host S3
  single-writer lease, restore the DB via litestream's restore API if the
  local copy is missing (wiping stale -wal/-shm/-litestream sidecar state
  first), spawn the same versioned binary the systemd unit would exec
  (live symlink + EnvironmentFile), wait for the socket, proxy. Concurrent
  requests coalesce into one wake (singleflight); queued waits are bounded
  (30s → 503).
- Sleep: per-host in-flight counting + last-activity stamping (an open
  WebSocket holds an instance awake); after the idle window the child is
  SIGTERMed, final-checkpointed to S3, and the lease released. The local
  DB stays as a warm cache; S3 is truth under the lease.
- Lease safety: contention answers 503 + Retry-After and logs loudly (on
  one box it means a previous incarnation hasn't released); a lost lease
  or dead child stands down immediately, and recovery is the routine wake
  path. Router shutdown checkpoints and stops every child.

internal/activator grows the split primitives the wake path needs:
AcquireLease/HeldLease (restore must happen under the lock, before
replication can start) and Restore (nil-DB litestream URL-restore;
"no backups" = boot fresh). ReplicateWithLease is now the composition of
the two; its API and live tests are unchanged.

Tests: state-machine unit tests with fakes (coalescing, idle→sleep,
inflight keep-awake, lease-lost stand-down, contention 503, wake timeout,
crash recycling, shutdown) — no AWS needed; plus a live end-to-end money
test gated on LCHAT_TEST_S3_BUCKET that boots a real instance behind the
real router, writes through the admin API, hibernates, deletes the local
DB, and proves the wake restores from S3 with the data intact and the
lease cleanly released.

Non-goals held for follow-ups (commented in code): due-time index (a
hibernated instance's future-messages/reminders don't fire while asleep —
why the feature stays opt-in), gossip/HRW/multi-box, bot sidecar
co-spawn, migrating existing fleet instances.
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 3d3ab0a4ea080494c49d14c524e0374825f6c704 parent 5c26ce0
6 files changed, +1745 −83
  • activator.go +710 −0
  • activator_live_test.go +254 −0
  • activator_test.go +547 −0
  • internal/activator/replica.go +169 −69
  • router.go +60 −9
  • router_test.go +5 −5
diff --git a/activator.go b/activator.go
new file mode 100644
index 0000000..84c01f7
--- /dev/null
+++ b/activator.go
@@ -0,0 +1,710 @@
+package main
+
+// The activator (issue #550 Part 4 v2, Phase 2): instances hibernate. When the
+// router is given an S3 bucket (LCHAT_HIBERNATE_BUCKET) it stops relying on
+// systemd to keep instance processes alive and instead runs them as its own
+// children, on demand — "units are for the router, instances are its
+// children". A request for a host with no live process claims the per-host S3
+// single-writer lease, restores the DB from S3 if the local copy is missing,
+// spawns the instance, and proxies; after an idle window the instance is
+// checkpointed back to S3, the lease released, and the process stopped. State
+// parks in S3 as ciphertext (an instance DB is opaque blobs plus routing
+// metadata the server already sees — the E2EE rule makes the bucket
+// threat-model-free); processes exist only while someone's talking.
+//
+// Scope (this phase): ONE box. No gossip/HRW placement, no peer proxying —
+// membership of one, hash picks self, leases uncontested except against a
+// previous incarnation of ourselves. Explicit non-goals held for follow-ups:
+// - No due-time index yet: scheduled future-messages/reminders of a
+// HIBERNATED instance don't fire while it sleeps (they deliver late, on
+// the next wake). That gap is why the feature stays opt-in.
+// - No bot sidecar co-spawn: a hibernating instance's sidecar isn't managed.
+// - No migration of existing fleet instances: setting the env on a box whose
+// instances run under systemd is unsupported (the units would fight the
+// activator over sockets and DBs).
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "net/http"
+ "os"
+ "os/exec"
+ "path"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "sync"
+ "syscall"
+ "time"
+
+ "github.com/benbjohnson/litestream"
+
+ "lchat/internal/activator"
+)
+
+// hibernateConfig is the router-process opt-in, read from env. Nil (bucket
+// unset) keeps today's behavior byte-identical: instances run under systemd,
+// the router only proxies.
+type hibernateConfig struct {
+ Bucket string // LCHAT_HIBERNATE_BUCKET (required to enable)
+ Region string // LCHAT_HIBERNATE_REGION (optional; AWS default chain resolves "")
+ Prefix string // LCHAT_HIBERNATE_PREFIX (optional key prefix inside the bucket)
+ Idle time.Duration // LCHAT_HIBERNATE_IDLE (default 15m; seconds-granular, e.g. "90s")
+}
+
+func hibernateConfigFromEnv() (*hibernateConfig, error) {
+ bucket := strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_BUCKET"))
+ if bucket == "" {
+ return nil, nil
+ }
+ cfg := &hibernateConfig{
+ Bucket: bucket,
+ Region: strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_REGION")),
+ Prefix: strings.Trim(strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_PREFIX")), "/"),
+ Idle: 15 * time.Minute,
+ }
+ if v := strings.TrimSpace(os.Getenv("LCHAT_HIBERNATE_IDLE")); v != "" {
+ d, err := time.ParseDuration(v)
+ if err != nil || d <= 0 {
+ return nil, fmt.Errorf("LCHAT_HIBERNATE_IDLE %q: want a positive duration like 15m or 90s", v)
+ }
+ cfg.Idle = d
+ }
+ return cfg, nil
+}
+
+// hibernatePrefix is the per-instance S3 key prefix: <LCHAT_HIBERNATE_PREFIX>/<host>/
+// (the lease's lock.json is colocated under it by the activator package).
+func (c *hibernateConfig) hibernatePrefix(host string) string {
+ return path.Join(c.Prefix, host)
+}
+
+// --- backend seam ------------------------------------------------------------
+//
+// The state machine's side effects (S3 lease, restore, process spawn) sit
+// behind these interfaces so the machine itself is unit-testable with fakes —
+// wake coalescing, idle→sleep, and lease-lost stand-down need no AWS.
+
+type actBackend interface {
+ // acquire claims the per-host single-writer lease. Contention surfaces as
+ // errLeaseHeld so the router can answer 503 + Retry-After.
+ acquire(ctx context.Context, host string) (actLease, error)
+ // restore materializes the local DB from S3 when it's absent (a no-op when
+ // the warm-cache copy is still on disk).
+ restore(ctx context.Context, host, dbPath string) error
+ // start spawns the instance process and waits until its socket accepts.
+ start(ctx context.Context, rt Route) (actProc, error)
+}
+
+// actLease is a held-but-not-yet-replicating lease (wake in progress).
+type actLease interface {
+ // replicate trades the lease for running replication of the (now present)
+ // local DB; the lease handle is spent on success.
+ replicate(ctx context.Context, dbPath string) (actRep, error)
+ // release gives the lease up — the bail-out when a wake fails partway.
+ release(ctx context.Context) error
+}
+
+// actRep is running replication coupled to the lease.
+type actRep interface {
+ // Err reports lost leadership (a lease renewal failed).
+ Err() error
+ // Close does the final checkpoint/sync and releases the lease.
+ Close(ctx context.Context) error
+}
+
+// actProc is a running instance child process.
+type actProc interface {
+ // stop terminates gracefully (SIGTERM, bounded wait, then SIGKILL).
+ stop(ctx context.Context) error
+ // exited reports whether the process has died on its own.
+ exited() bool
+}
+
+// --- state machine -----------------------------------------------------------
+
+// Per-host lifecycle. An absent entry is the fourth state: asleep.
+//
+// asleep --request--------------------> waking (one wake; concurrent
+// waking --wake ok--------------------> awake requests coalesce on it)
+// waking --wake failed----------------> asleep (error served; next request retries)
+// awake --idle timeout / lease lost /
+// child died----------------> stopping (final checkpoint, release lease)
+// stopping --stop done----------------> asleep (local DB left as a warm cache)
+//
+// A request that lands on waking waits (bounded) for the wake; one that lands
+// on stopping waits for the stop, then re-enters as a fresh wake.
+type actState int
+
+const (
+ actWaking actState = iota
+ actAwake
+ actStopping
+)
+
+type actEntry struct {
+ host string
+ route Route
+ state actState
+
+ ready chan struct{} // waking: closed when the wake resolves; then check err
+ err error // waking: the wake's outcome
+ stopped chan struct{} // stopping: closed when the stop completes
+
+ // Activity tracking: an instance is idle only when nothing is in flight
+ // AND the idle window has passed since the last request finished. In-flight
+ // counting (not just timestamps) is what keeps an instance with one open
+ // WebSocket awake — the proxy's ServeHTTP blocks for the tunnel's life.
+ inflight int
+ lastActive time.Time
+
+ rep actRep
+ proc actProc
+}
+
+// wakeWaitBound caps how long a request queues on someone else's wake or stop
+// before giving up with 503 — a wedged wake must not pile up connections
+// forever. A var so tests can shrink it.
+var wakeWaitBound = 30 * time.Second
+
+// errLeaseHeld: the per-host lease is held elsewhere. On one box this means a
+// previous incarnation hasn't released (crashed router; lease expires within
+// ~LeaseTTL) — the router answers 503 with Retry-After.
+type errLeaseHeld struct {
+ holder string
+ retryAfter time.Duration
+}
+
+func (e *errLeaseHeld) Error() string {
+ return fmt.Sprintf("instance lease held by %q (retry in %s)", e.holder, e.retryAfter)
+}
+
+var errWakeTimeout = errors.New("timed out waiting for instance wake")
+
+type activatorSvc struct {
+ cfg hibernateConfig
+ backend actBackend
+ now func() time.Time // a seam for the unit tests' clock
+
+ mu sync.Mutex
+ entries map[string]*actEntry
+
+ sweepStop chan struct{}
+ sweepDone chan struct{}
+}
+
+func newActivator(cfg hibernateConfig, backend actBackend) *activatorSvc {
+ return &activatorSvc{
+ cfg: cfg,
+ backend: backend,
+ now: time.Now,
+ entries: make(map[string]*actEntry),
+ }
+}
+
+// beginRequest ensures the host's instance is awake and registers one in-flight
+// request against it. The returned release must be called when the request
+// finishes (it stamps last-activity, which starts the idle clock).
+func (a *activatorSvc) beginRequest(ctx context.Context, rt Route) (release func(), err error) {
+ deadline := time.NewTimer(wakeWaitBound)
+ defer deadline.Stop()
+ for {
+ a.mu.Lock()
+ e := a.entries[rt.Host]
+ if e == nil {
+ // Asleep → kick off ONE wake; this request and any latecomers all
+ // coalesce on e.ready rather than racing a second lease claim
+ // (singleflight). The wake runs detached so even its initiator is
+ // held to the same bounded wait as everyone queued behind it.
+ e = &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})}
+ a.entries[rt.Host] = e
+ a.mu.Unlock()
+ go a.wake(e)
+ continue
+ }
+
+ switch e.state {
+ case actAwake:
+ // Stand down synchronously-detected failure: a dead child or a
+ // lost lease found at request time turns into a stop, and the
+ // request waits for it then re-wakes — recovery is the routine
+ // wake path.
+ if repErr := e.rep.Err(); repErr != nil || e.proc.exited() {
+ if repErr != nil {
+ log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
+ } else {
+ log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
+ }
+ a.beginStopLocked(e)
+ a.mu.Unlock()
+ continue
+ }
+ e.inflight++
+ a.mu.Unlock()
+ return a.releaseFunc(e), nil
+
+ case actWaking:
+ ready := e.ready
+ a.mu.Unlock()
+ select {
+ case <-ready:
+ // A failed wake reports to every coalesced waiter (retrying
+ // here would turn one bad wake into a storm); a successful
+ // one loops to register against the now-awake entry.
+ if e.err != nil {
+ return nil, e.err
+ }
+ continue
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case <-deadline.C:
+ return nil, errWakeTimeout
+ }
+
+ case actStopping:
+ stopped := e.stopped
+ a.mu.Unlock()
+ select {
+ case <-stopped:
+ continue // entry is gone; loop wakes it fresh
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case <-deadline.C:
+ return nil, errWakeTimeout
+ }
+ }
+ }
+}
+
+func (a *activatorSvc) releaseFunc(e *actEntry) func() {
+ var once sync.Once
+ return func() {
+ once.Do(func() {
+ a.mu.Lock()
+ e.inflight--
+ e.lastActive = a.now()
+ a.mu.Unlock()
+ })
+ }
+}
+
+// wake drives asleep→awake: lease, restore-if-missing, spawn, replicate. It
+// runs on the first requester's goroutine but under its own generous timeout,
+// detached from that request's context — queued requests still need the wake
+// to finish even if the requester who started it hangs up.
+func (a *activatorSvc) wake(e *actEntry) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+ defer cancel()
+ rep, proc, err := a.doWake(ctx, e.route)
+
+ a.mu.Lock()
+ if err != nil {
+ e.err = err
+ delete(a.entries, e.host) // next request retries from scratch
+ } else {
+ e.state = actAwake
+ e.rep, e.proc = rep, proc
+ e.lastActive = a.now()
+ }
+ a.mu.Unlock()
+ close(e.ready)
+}
+
+func (a *activatorSvc) doWake(ctx context.Context, rt Route) (actRep, actProc, error) {
+ host := rt.Host
+ dbPath := rt.DBPath
+ if dbPath == "" {
+ dbPath = filepath.Join(instanceDataDir, host+".db")
+ }
+
+ // 1. The lease first: everything after this line is done as the prefix's
+ // single writer. Contention is not retried here — on one box it means a
+ // previous incarnation hasn't released; tell the client to come back after
+ // the TTL instead of stalling its request.
+ lease, err := a.backend.acquire(ctx, host)
+ if err != nil {
+ var held *errLeaseHeld
+ if errors.As(err, &held) {
+ log.Printf("activator: LEASE HELD for %s by %q — a previous incarnation hasn't released; refusing to wake (retry after %s)",
+ host, held.holder, held.retryAfter)
+ }
+ return nil, nil, err
+ }
+
+ // 2. Restore the DB from S3 when the local copy is gone. When it's still
+ // on disk it is the warm cache: on one box, with the lease held, the local
+ // copy can only be what we ourselves checkpointed at the last sleep.
+ if err := a.backend.restore(ctx, host, dbPath); err != nil {
+ _ = lease.release(ctx)
+ return nil, nil, fmt.Errorf("restore %s: %w", host, err)
+ }
+
+ // 3. Spawn before replication starts: a fresh instance creates its own DB
+ // on boot, so waiting for the socket guarantees the file exists for
+ // litestream to open. Writes that land in the gap are picked up by the
+ // replica's first sync.
+ proc, err := a.backend.start(ctx, rt)
+ if err != nil {
+ _ = lease.release(ctx)
+ return nil, nil, fmt.Errorf("start %s: %w", host, err)
+ }
+
+ // 4. Trade the held lease for running replication.
+ rep, err := lease.replicate(ctx, dbPath)
+ if err != nil {
+ stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
+ defer cancel()
+ _ = proc.stop(stopCtx)
+ _ = lease.release(ctx)
+ return nil, nil, fmt.Errorf("replicate %s: %w", host, err)
+ }
+ return rep, proc, nil
+}
+
+// actStopTimeout bounds a graceful instance stop (the instance's own SIGTERM
+// drain is 10s; leave headroom before the SIGKILL).
+const actStopTimeout = 20 * time.Second
+
+// beginStopLocked flips an awake entry to stopping and kicks the stop off; the
+// caller holds a.mu. In-flight requests are not waited for — a stop is only
+// initiated when the instance is idle, dead, or has lost its lease, and in the
+// latter two cases open requests are already doomed.
+func (a *activatorSvc) beginStopLocked(e *actEntry) {
+ e.state = actStopping
+ e.stopped = make(chan struct{})
+ go a.stop(e)
+}
+
+// stop drives awake→asleep: stop the child (its SIGTERM drain is the final
+// write), checkpoint what's on disk to S3 and release the lease, then forget
+// the entry. The local DB stays in place — it's the warm cache for the next
+// wake; S3 is the durable truth under the lease.
+func (a *activatorSvc) stop(e *actEntry) {
+ // Separate budgets: a SIGTERM drain that runs long must not eat the final
+ // checkpoint's time — the S3 sync is the part that makes sleep durable.
+ procCtx, cancelProc := context.WithTimeout(context.Background(), actStopTimeout)
+ err := e.proc.stop(procCtx)
+ cancelProc()
+ if err != nil {
+ log.Printf("activator: stop %s: %v", e.host, err)
+ }
+ // Close = final sync + lease release. After a lost lease the release part
+ // fails harmlessly (someone else owns the lock); the sync still preserves
+ // a local checkpoint.
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer cancel()
+ if err := e.rep.Close(ctx); err != nil {
+ log.Printf("activator: final checkpoint %s: %v", e.host, err)
+ }
+ a.mu.Lock()
+ delete(a.entries, e.host)
+ a.mu.Unlock()
+ close(e.stopped)
+ log.Printf("activator: %s hibernated", e.host)
+}
+
+// start launches the background sweeper that puts idle instances to sleep and
+// stands down lost leases/dead children between requests.
+func (a *activatorSvc) start() {
+ a.sweepStop = make(chan struct{})
+ a.sweepDone = make(chan struct{})
+ // Sweep well inside the idle window so a sleep happens promptly after it
+ // elapses; clamp so second-granular test idles still sweep sanely.
+ interval := a.cfg.Idle / 4
+ if interval < 100*time.Millisecond {
+ interval = 100 * time.Millisecond
+ }
+ if interval > 30*time.Second {
+ interval = 30 * time.Second
+ }
+ go func() {
+ defer close(a.sweepDone)
+ t := time.NewTicker(interval)
+ defer t.Stop()
+ for {
+ select {
+ case <-a.sweepStop:
+ return
+ case <-t.C:
+ a.sweepOnce()
+ }
+ }
+ }()
+}
+
+// sweepOnce examines every awake instance once: idle ones hibernate; a lost
+// lease or dead child stands down immediately regardless of activity.
+func (a *activatorSvc) sweepOnce() {
+ now := a.now()
+ a.mu.Lock()
+ for _, e := range a.entries {
+ if e.state != actAwake {
+ continue
+ }
+ if err := e.rep.Err(); err != nil {
+ log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, err)
+ a.beginStopLocked(e)
+ continue
+ }
+ if e.proc.exited() {
+ log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
+ a.beginStopLocked(e)
+ continue
+ }
+ if e.inflight == 0 && now.Sub(e.lastActive) >= a.cfg.Idle {
+ log.Printf("activator: %s idle %s — hibernating", e.host, now.Sub(e.lastActive).Truncate(time.Second))
+ a.beginStopLocked(e)
+ }
+ }
+ a.mu.Unlock()
+}
+
+// shutdown checkpoints and stops every child — the router going down takes its
+// instances with it (they're its children; leaving them orphaned would strand
+// the leases until TTL expiry and leave the sockets bound against the next
+// incarnation's spawns). Blocks until all stops complete or ctx expires.
+func (a *activatorSvc) shutdown(ctx context.Context) {
+ if a.sweepStop != nil {
+ close(a.sweepStop)
+ <-a.sweepDone
+ }
+ a.mu.Lock()
+ var waits []chan struct{}
+ for _, e := range a.entries {
+ switch e.state {
+ case actAwake:
+ a.beginStopLocked(e)
+ waits = append(waits, e.stopped)
+ case actStopping:
+ waits = append(waits, e.stopped)
+ case actWaking:
+ // A mid-flight wake resolves on its own; its instance will be
+ // orphaned only if it wins the race with process exit — acceptable
+ // for the rare shutdown-during-wake, and the lease TTL still
+ // bounds the damage.
+ }
+ }
+ a.mu.Unlock()
+ for _, w := range waits {
+ select {
+ case <-w:
+ case <-ctx.Done():
+ return
+ }
+ }
+}
+
+// serveHTTP answers one proxied request for a hibernating instance: wake (or
+// join a wake), proxy, stamp activity. Lease contention and wake timeouts are
+// 503s — "sleeping spaces are slow to wake", never a 5xx cascade of retries.
+func (a *activatorSvc) serveHTTP(w http.ResponseWriter, r *http.Request, rt Route, proxy http.Handler) {
+ release, err := a.beginRequest(r.Context(), rt)
+ if err != nil {
+ var held *errLeaseHeld
+ switch {
+ case errors.As(err, &held):
+ retry := int(held.retryAfter/time.Second) + 1
+ w.Header().Set("Retry-After", strconv.Itoa(retry))
+ http.Error(w, "instance is waking elsewhere; retry shortly", http.StatusServiceUnavailable)
+ case errors.Is(err, errWakeTimeout):
+ w.Header().Set("Retry-After", "5")
+ http.Error(w, "instance is waking; retry shortly", http.StatusServiceUnavailable)
+ case errors.Is(err, context.Canceled):
+ // The client hung up while queued; nothing to write.
+ default:
+ log.Printf("activator: wake %s: %v", rt.Host, err)
+ http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
+ }
+ return
+ }
+ defer release()
+ proxy.ServeHTTP(w, r)
+}
+
+// --- the real backend ---------------------------------------------------------
+
+// s3Backend is the production actBackend: leases + restore via the
+// internal/activator litestream primitives, processes via exec of the same
+// versioned binary the systemd unit would run.
+type s3Backend struct {
+ cfg hibernateConfig
+}
+
+func (b *s3Backend) acquire(ctx context.Context, host string) (actLease, error) {
+ held, err := activator.AcquireLease(ctx, b.cfg.Bucket, b.cfg.hibernatePrefix(host), b.cfg.Region)
+ if err != nil {
+ var lx *litestream.LeaseExistsError
+ if errors.As(err, &lx) {
+ retry := time.Until(lx.ExpiresAt)
+ if retry < time.Second {
+ retry = time.Second
+ }
+ return nil, &errLeaseHeld{holder: lx.Owner, retryAfter: retry}
+ }
+ return nil, err
+ }
+ return &s3Lease{held: held}, nil
+}
+
+type s3Lease struct{ held *activator.HeldLease }
+
+func (l *s3Lease) replicate(ctx context.Context, dbPath string) (actRep, error) {
+ return l.held.Replicate(ctx, dbPath)
+}
+func (l *s3Lease) release(ctx context.Context) error { return l.held.Release(ctx) }
+
+func (b *s3Backend) restore(ctx context.Context, host, dbPath string) error {
+ if _, err := os.Stat(dbPath); err == nil {
+ return nil // warm cache — see doWake
+ } else if !os.IsNotExist(err) {
+ return err
+ }
+ // The DB is gone, so any leftover litestream sidecar state (the
+ // <db>-litestream meta dir, WAL/SHM) describes a database that no longer
+ // exists — litestream's own recovery guidance is to delete it. Clear it so
+ // the restored copy starts a clean lineage.
+ for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} {
+ _ = os.Remove(p)
+ }
+ _ = os.RemoveAll(dbPath + litestream.MetaDirSuffix)
+
+ restored, err := activator.Restore(ctx, dbPath, b.cfg.Bucket, b.cfg.hibernatePrefix(host), b.cfg.Region)
+ if err != nil {
+ return err
+ }
+ if restored {
+ log.Printf("activator: restored %s from s3://%s/%s", host, b.cfg.Bucket, b.cfg.hibernatePrefix(host))
+ } else {
+ // Never replicated: a brand-new space boots fresh and creates its own
+ // DB — this is the future register-only create path (provisioning as a
+ // metadata write) landing on its first wake.
+ log.Printf("activator: no replica for %s in s3://%s/%s — booting fresh", host, b.cfg.Bucket, b.cfg.hibernatePrefix(host))
+ }
+ return nil
+}
+
+// spawnHealthTimeout bounds how long a spawned instance may take to bind its
+// socket and answer. A var so tests can shrink the failure path.
+var spawnHealthTimeout = 30 * time.Second
+
+func (b *s3Backend) start(ctx context.Context, rt Route) (actProc, error) {
+ host := rt.Host
+ dbPath := rt.DBPath
+ if dbPath == "" {
+ dbPath = filepath.Join(instanceDataDir, host+".db")
+ }
+
+ // Exec exactly what the systemd unit would: the per-instance live symlink
+ // (/opt/lchat/live/<host> → a staged version), same --socket/--db flags,
+ // same EnvironmentFile. If the host was registered but never materialized
+ // (the future create-is-a-row flow), stage the running binary and point
+ // the link, exactly as materializeInstance would have.
+ bin := liveLink(host)
+ if _, err := os.Stat(bin); err != nil {
+ ver, err := ensureSelfStaged()
+ if err != nil {
+ return nil, fmt.Errorf("stage binary: %w", err)
+ }
+ if err := pointLive(host, ver); err != nil {
+ return nil, fmt.Errorf("point live symlink: %w", err)
+ }
+ }
+
+ // Environment: the unit gives an instance a clean env plus its
+ // EnvironmentFile. Approximate that from inside the router: inherit the
+ // process env but strip every LCHAT_* var first, so the router's own
+ // config (hibernation, registry, wildcard tokens) can never leak into a
+ // child; then layer the instance's env file (admin token, per-instance
+ // config) on top. The env file is optional, like the unit's "-" prefix.
+ env := make([]string, 0, 32)
+ for _, kv := range os.Environ() {
+ if strings.HasPrefix(kv, "LCHAT_") {
+ continue
+ }
+ env = append(env, kv)
+ }
+ if fileEnv, err := readEnvFile(filepath.Join(instanceDataDir, host+".env")); err == nil {
+ for k, v := range fileEnv {
+ env = append(env, k+"="+v)
+ }
+ } else if !os.IsNotExist(err) {
+ return nil, fmt.Errorf("read env file: %w", err)
+ }
+
+ cmd := exec.Command(bin, "--socket", rt.Socket, "--db", dbPath)
+ cmd.Env = env
+ // The child logs through the router's own stderr/stdout (journald under
+ // systemd); there's no per-instance unit to own a log stream anymore.
+ cmd.Stdout = os.Stdout
+ cmd.Stderr = os.Stderr
+ if err := cmd.Start(); err != nil {
+ return nil, err
+ }
+
+ p := &childProc{cmd: cmd, done: make(chan struct{})}
+ go func() {
+ p.err = cmd.Wait()
+ close(p.done)
+ }()
+
+ if err := waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout); err != nil {
+ stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
+ defer cancel()
+ _ = p.stop(stopCtx)
+ return nil, err
+ }
+ return p, nil
+}
+
+// waitHealthyCtx is waitHealthy bounded by ctx as well as its own timeout, so
+// an aborted wake doesn't keep probing a socket nobody wants anymore.
+func waitHealthyCtx(ctx context.Context, socket string, timeout time.Duration) error {
+ done := make(chan error, 1)
+ go func() { done <- waitHealthy(socket, timeout) }()
+ select {
+ case err := <-done:
+ return err
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+}
+
+// childProc is a spawned instance: SIGTERM starts the instance's own graceful
+// drain (main.go's shutdown path); SIGKILL only if it overstays.
+type childProc struct {
+ cmd *exec.Cmd
+ done chan struct{}
+ err error
+}
+
+func (p *childProc) exited() bool {
+ select {
+ case <-p.done:
+ return true
+ default:
+ return false
+ }
+}
+
+func (p *childProc) stop(ctx context.Context) error {
+ if p.exited() {
+ return nil
+ }
+ if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil {
+ if errors.Is(err, os.ErrProcessDone) { // lost the race with its own exit
+ <-p.done
+ return nil
+ }
+ return err
+ }
+ select {
+ case <-p.done:
+ return nil
+ case <-ctx.Done():
+ _ = p.cmd.Process.Kill()
+ <-p.done
+ return fmt.Errorf("instance did not exit on SIGTERM; killed")
+ }
+}
diff --git a/activator_live_test.go b/activator_live_test.go
new file mode 100644
index 0000000..218d824
--- /dev/null
+++ b/activator_live_test.go
@@ -0,0 +1,254 @@
+package main
+
+// 🤖 The live end-to-end hibernation test ("the money test", #550 Part 4 v2
+// Phase 2): a REAL instance process behind the REAL router, replicating to
+// REAL S3. It proves the whole loop on one box — wake fresh, write data
+// through the public API, idle out (final checkpoint + lease release), DELETE
+// the local DB, wake again (restore from S3), and read the data back.
+//
+// Gated on LCHAT_TEST_S3_BUCKET like the activator package's spike tests, so
+// `go test ./...` stays green without AWS credentials. Run it with:
+//
+// eval "$(aws configure export-credentials --profile eleven-dev --format env)"
+// LCHAT_TEST_S3_BUCKET=<bucket> go test -run TestActivatorLiveHibernation -v .
+
+import (
+ "bytes"
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awsconfig "github.com/aws/aws-sdk-go-v2/config"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+ s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
+
+ "lchat/internal/activator"
+)
+
+func TestActivatorLiveHibernation(t *testing.T) {
+ bucket := os.Getenv("LCHAT_TEST_S3_BUCKET")
+ if bucket == "" {
+ t.Skip("LCHAT_TEST_S3_BUCKET not set; skipping live S3 hibernation test")
+ }
+ region := os.Getenv("LCHAT_TEST_S3_REGION")
+ if region == "" {
+ region = "eu-west-1"
+ }
+ rb := make([]byte, 4)
+ _, _ = rand.Read(rb)
+ prefix := fmt.Sprintf("activator-live/%d-%s", time.Now().Unix(), hex.EncodeToString(rb))
+ t.Cleanup(func() { cleanupS3Prefix(t, bucket, prefix, region) })
+
+ const host = "hib-live.test"
+
+ // Sandbox the box-layout globals (control.go / versions.go) in temp dirs.
+ // The socket dir must stay short — sun_path is ~104 bytes on macOS.
+ dataDir := t.TempDir()
+ sockDir := fmt.Sprintf("/tmp/lchat-act-%d", os.Getpid())
+ if err := os.MkdirAll(sockDir, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(sockDir) })
+ optTmp := t.TempDir()
+ oldData, oldSock, oldOpt := instanceDataDir, instanceSocketDir, optDir
+ instanceDataDir, instanceSocketDir, optDir = dataDir, sockDir, optTmp
+ t.Cleanup(func() { instanceDataDir, instanceSocketDir, optDir = oldData, oldSock, oldOpt })
+
+ // The activator execs the same live-symlinked binary a systemd unit would,
+ // so build the real lchat binary (the test binary must never be spawned).
+ bin := filepath.Join(optTmp, "lchat-built")
+ build := exec.Command("go", "build", "-o", bin, ".")
+ build.Env = append(os.Environ(), "GOTOOLCHAIN=auto")
+ if out, err := build.CombinedOutput(); err != nil {
+ t.Fatalf("go build: %v\n%s", err, out)
+ }
+ if err := stageVersion(bin, "livetest"); err != nil {
+ t.Fatalf("stage: %v", err)
+ }
+ if err := pointLive(host, "livetest"); err != nil {
+ t.Fatalf("point live: %v", err)
+ }
+
+ // The instance's EnvironmentFile, as add-instance would have written it.
+ adminToken := "livetest-" + hex.EncodeToString(rb)
+ if err := writeEnvFile(filepath.Join(dataDir, host+".env"),
+ map[string]string{"LCHAT_ADMIN_TOKEN": adminToken}); err != nil {
+ t.Fatalf("env file: %v", err)
+ }
+
+ reg, err := openRegistry(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatalf("open registry: %v", err)
+ }
+ route := Route{Host: host, Kind: "instance",
+ Socket: filepath.Join(sockDir, host+".sock"),
+ DBPath: filepath.Join(dataDir, host+".db")}
+ if err := reg.upsert(route); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+
+ cfg := hibernateConfig{Bucket: bucket, Region: region, Prefix: prefix, Idle: 3 * time.Second}
+ act := newActivator(cfg, &s3Backend{cfg: cfg})
+ act.start()
+ t.Cleanup(func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+ defer cancel()
+ act.shutdown(ctx)
+ })
+
+ rt, err := newRouter(reg, act)
+ if err != nil {
+ t.Fatalf("new router: %v", err)
+ }
+ front := httptest.NewServer(rt)
+ t.Cleanup(front.Close)
+
+ get := func(path string) (int, string) {
+ t.Helper()
+ req, _ := http.NewRequest("GET", front.URL+path, nil)
+ req.Host = host
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("GET %s: %v", path, err)
+ }
+ defer resp.Body.Close()
+ body, _ := io.ReadAll(resp.Body)
+ return resp.StatusCode, string(body)
+ }
+
+ // --- Act 1: first wake is a FRESH boot (nothing in S3, no local DB). ---
+ start := time.Now()
+ if code, _ := get("/api/group"); code != 200 {
+ t.Fatalf("first wake: GET /api/group = %d, want 200", code)
+ }
+ t.Logf("wake #1 (fresh boot, no replica) served in %s", time.Since(start).Truncate(time.Millisecond))
+
+ // Write real data through the public path: the admin names the space.
+ groupName := "Hibernation survives " + hex.EncodeToString(rb)
+ body, _ := json.Marshal(map[string]string{"name": groupName})
+ req, _ := http.NewRequest("POST", front.URL+"/api/admin/group", bytes.NewReader(body))
+ req.Host = host
+ req.Header.Set("Authorization", "Bearer "+adminToken)
+ req.Header.Set("Content-Type", "application/json")
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("set group name: %v", err)
+ }
+ if resp.StatusCode != 200 {
+ b, _ := io.ReadAll(resp.Body)
+ resp.Body.Close()
+ t.Fatalf("set group name: %d %s", resp.StatusCode, b)
+ }
+ resp.Body.Close()
+ t.Logf("wrote group name %q through the admin API", groupName)
+
+ // --- Act 2: idle out. Final checkpoint to S3, lease released, process gone. ---
+ waitForLive(t, "hibernation after idle", 90*time.Second, func() bool { return act.entryCount() == 0 })
+ if _, err := net.DialTimeout("unix", route.Socket, 300*time.Millisecond); err == nil {
+ t.Fatal("instance socket still accepting after hibernation")
+ }
+ if _, err := os.Stat(route.DBPath); err != nil {
+ t.Fatalf("local DB should remain as a warm cache after sleep: %v", err)
+ }
+ t.Logf("instance hibernated (idle %s); local DB left as warm cache", cfg.Idle)
+
+ // --- Act 3: destroy the local state; the wake path must restore from S3. ---
+ matches, _ := filepath.Glob(route.DBPath + "*")
+ for _, m := range matches {
+ if err := os.RemoveAll(m); err != nil {
+ t.Fatalf("remove %s: %v", m, err)
+ }
+ }
+ t.Logf("deleted local state: %v", matches)
+
+ start = time.Now()
+ code, groupJSON := get("/api/group")
+ if code != 200 {
+ t.Fatalf("wake from S3: GET /api/group = %d, want 200", code)
+ }
+ if !strings.Contains(groupJSON, groupName) {
+ t.Fatalf("restored instance lost the data: %s does not contain %q", groupJSON, groupName)
+ }
+ t.Logf("wake #2 (restored from s3://%s/%s) served in %s; data survived: %s",
+ bucket, prefix, time.Since(start).Truncate(time.Millisecond), groupJSON)
+
+ // --- Act 4: idle out again and prove the lease was released cleanly. ---
+ waitForLive(t, "second hibernation", 90*time.Second, func() bool { return act.entryCount() == 0 })
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer cancel()
+ held, err := activator.AcquireLease(ctx, bucket, cfg.hibernatePrefix(host), region)
+ if err != nil {
+ t.Fatalf("lease not cleanly released after sleep: %v", err)
+ }
+ if err := held.Release(ctx); err != nil {
+ t.Fatalf("release probe lease: %v", err)
+ }
+ t.Log("second sleep released the lease cleanly (probe acquire+release succeeded)")
+}
+
+// waitForLive is waitFor with a caller-chosen deadline — S3 round trips make
+// the live test's transitions slower than the fakes'.
+func waitForLive(t *testing.T, what string, timeout time.Duration, cond func() bool) {
+ t.Helper()
+ deadline := time.Now().Add(timeout)
+ for time.Now().Before(deadline) {
+ if cond() {
+ return
+ }
+ time.Sleep(100 * time.Millisecond)
+ }
+ t.Fatalf("timed out waiting for %s", what)
+}
+
+// cleanupS3Prefix deletes every object AND version under the test prefix so
+// live runs leave nothing behind (mirrors the activator package's helper).
+func cleanupS3Prefix(t *testing.T, bucket, prefix, region string) {
+ t.Helper()
+ ctx := context.Background()
+ cfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region))
+ if err != nil {
+ t.Logf("cleanup aws config: %v", err)
+ return
+ }
+ client := awss3.NewFromConfig(cfg)
+ p := awss3.NewListObjectVersionsPaginator(client, &awss3.ListObjectVersionsInput{
+ Bucket: aws.String(bucket),
+ Prefix: aws.String(prefix),
+ })
+ for p.HasMorePages() {
+ out, err := p.NextPage(ctx)
+ if err != nil {
+ t.Logf("cleanup list versions: %v", err)
+ return
+ }
+ var objs []s3types.ObjectIdentifier
+ for _, v := range out.Versions {
+ objs = append(objs, s3types.ObjectIdentifier{Key: v.Key, VersionId: v.VersionId})
+ }
+ for _, m := range out.DeleteMarkers {
+ objs = append(objs, s3types.ObjectIdentifier{Key: m.Key, VersionId: m.VersionId})
+ }
+ if len(objs) == 0 {
+ continue
+ }
+ if _, err := client.DeleteObjects(ctx, &awss3.DeleteObjectsInput{
+ Bucket: aws.String(bucket),
+ Delete: &s3types.Delete{Objects: objs},
+ }); err != nil {
+ t.Logf("cleanup delete: %v", err)
+ }
+ }
+}
diff --git a/activator_test.go b/activator_test.go
new file mode 100644
index 0000000..04a0fcf
--- /dev/null
+++ b/activator_test.go
@@ -0,0 +1,547 @@
+package main
+
+// Unit tests for the activator state machine (activator.go) with a fake
+// backend — wake coalescing, idle→sleep, lease-lost stand-down, contention
+// 503s. No AWS, no processes: the live end-to-end test (the wake/restore/
+// sleep money path against real S3 + a real spawned instance) is
+// TestActivatorLiveHibernation in activator_live_test.go.
+
+import (
+ "context"
+ "errors"
+ "io"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+// --- fakes -------------------------------------------------------------------
+
+type fakeRep struct {
+ mu sync.Mutex
+ err error
+ closed bool
+}
+
+func (r *fakeRep) Err() error {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return r.err
+}
+func (r *fakeRep) setErr(err error) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ r.err = err
+}
+func (r *fakeRep) Close(context.Context) error {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ r.closed = true
+ return nil
+}
+func (r *fakeRep) isClosed() bool {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return r.closed
+}
+
+type fakeProc struct {
+ mu sync.Mutex
+ stopped bool
+ dead bool
+ // stopGate, when non-nil, blocks stop() until closed — holds an entry in
+ // the stopping state so a test can poke at it.
+ stopGate chan struct{}
+}
+
+func (p *fakeProc) stop(context.Context) error {
+ p.mu.Lock()
+ gate := p.stopGate
+ p.mu.Unlock()
+ if gate != nil {
+ <-gate
+ }
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ p.stopped = true
+ return nil
+}
+func (p *fakeProc) exited() bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.dead
+}
+func (p *fakeProc) isStopped() bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return p.stopped
+}
+
+type fakeLease struct{ rep *fakeRep }
+
+func (l *fakeLease) replicate(context.Context, string) (actRep, error) { return l.rep, nil }
+func (l *fakeLease) release(context.Context) error { return nil }
+
+type fakeBackend struct {
+ mu sync.Mutex
+ acquires int
+ restores int
+ starts int
+ acquireErr error
+ startGate chan struct{} // when non-nil, start() blocks until closed
+ reps []*fakeRep
+ procs []*fakeProc
+}
+
+func (b *fakeBackend) acquire(context.Context, string) (actLease, error) {
+ b.mu.Lock()
+ b.acquires++
+ err := b.acquireErr
+ rep := &fakeRep{}
+ b.reps = append(b.reps, rep)
+ b.mu.Unlock()
+ if err != nil {
+ return nil, err
+ }
+ return &fakeLease{rep: rep}, nil
+}
+
+func (b *fakeBackend) restore(context.Context, string, string) error {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ b.restores++
+ return nil
+}
+
+func (b *fakeBackend) start(context.Context, Route) (actProc, error) {
+ b.mu.Lock()
+ b.starts++
+ gate := b.startGate
+ proc := &fakeProc{}
+ b.procs = append(b.procs, proc)
+ b.mu.Unlock()
+ if gate != nil {
+ <-gate
+ }
+ return proc, nil
+}
+
+func (b *fakeBackend) counts() (acquires, restores, starts int) {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.acquires, b.restores, b.starts
+}
+
+func (b *fakeBackend) lastProc() *fakeProc {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.procs[len(b.procs)-1]
+}
+
+func (b *fakeBackend) lastRep() *fakeRep {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.reps[len(b.reps)-1]
+}
+
+// fakeClock is the activator's `now` seam.
+type fakeClock struct {
+ mu sync.Mutex
+ t time.Time
+}
+
+func (c *fakeClock) now() time.Time {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return c.t
+}
+func (c *fakeClock) advance(d time.Duration) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.t = c.t.Add(d)
+}
+
+func testActivator(t *testing.T, idle time.Duration) (*activatorSvc, *fakeBackend, *fakeClock) {
+ t.Helper()
+ b := &fakeBackend{}
+ clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
+ a := newActivator(hibernateConfig{Bucket: "test-bucket", Idle: idle}, b)
+ a.now = clk.now
+ return a, b, clk
+}
+
+func testRoute(host string) Route {
+ return Route{Host: host, Kind: "instance", Socket: "/tmp/nope.sock",
+ DBPath: filepath.Join("/tmp", host+".db")}
+}
+
+// (waitFor — poll until a condition holds — is shared from bot_run_test.go;
+// the stop path runs on its own goroutine, so tests observe its completion
+// rather than assuming it.)
+
+func (a *activatorSvc) entryCount() int {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ return len(a.entries)
+}
+
+// --- tests ---------------------------------------------------------------------
+
+// Many concurrent requests during a wake coalesce into exactly one wake: one
+// lease claim, one restore, one spawn.
+func TestActivatorWakeCoalesces(t *testing.T) {
+ a, b, _ := testActivator(t, time.Minute)
+ b.startGate = make(chan struct{})
+
+ const n = 12
+ var ok atomic.Int32
+ var wg sync.WaitGroup
+ for i := 0; i < n; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Errorf("beginRequest: %v", err)
+ return
+ }
+ ok.Add(1)
+ release()
+ }()
+ }
+ // Let the requests pile up on the gated wake, then open it.
+ waitFor(t, "the wake to begin", func() bool { _, _, s := b.counts(); return s == 1 })
+ time.Sleep(50 * time.Millisecond)
+ close(b.startGate)
+ wg.Wait()
+
+ if got := ok.Load(); got != n {
+ t.Fatalf("%d/%d requests succeeded", got, n)
+ }
+ acquires, restores, starts := b.counts()
+ if acquires != 1 || restores != 1 || starts != 1 {
+ t.Fatalf("wake did not coalesce: acquires=%d restores=%d starts=%d (want 1 each)", acquires, restores, starts)
+ }
+}
+
+// After the idle window with nothing in flight, the sweeper hibernates the
+// instance: process stopped, replication closed (final checkpoint + lease
+// release), entry forgotten; the next request wakes it afresh.
+func TestActivatorIdleSleep(t *testing.T) {
+ a, b, clk := testActivator(t, 10*time.Second)
+
+ release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+
+ // Not yet idle: nothing happens.
+ clk.advance(9 * time.Second)
+ a.sweepOnce()
+ if a.entryCount() != 1 {
+ t.Fatal("instance hibernated before its idle window elapsed")
+ }
+
+ clk.advance(2 * time.Second)
+ a.sweepOnce()
+ waitFor(t, "hibernation", func() bool { return a.entryCount() == 0 })
+ if !b.lastProc().isStopped() {
+ t.Fatal("idle instance's process was not stopped")
+ }
+ if !b.lastRep().isClosed() {
+ t.Fatal("idle instance's replication was not closed (no final checkpoint/lease release)")
+ }
+
+ // A new request wakes it again — a second full wake.
+ release, err = a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatalf("beginRequest after sleep: %v", err)
+ }
+ release()
+ if _, _, starts := b.counts(); starts != 2 {
+ t.Fatalf("starts=%d after sleep+rewake, want 2", starts)
+ }
+}
+
+// An in-flight request (an open WebSocket, in production) holds the instance
+// awake past the idle window; only after it finishes does the idle clock run.
+func TestActivatorInflightKeepsAwake(t *testing.T) {
+ a, b, clk := testActivator(t, 10*time.Second)
+
+ release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+
+ clk.advance(time.Hour)
+ a.sweepOnce()
+ if a.entryCount() != 1 || b.lastProc().isStopped() {
+ t.Fatal("instance with an in-flight request was hibernated")
+ }
+
+ release() // stamps lastActive = now
+ a.sweepOnce()
+ if a.entryCount() != 1 {
+ t.Fatal("instance hibernated immediately after its last request finished")
+ }
+ clk.advance(11 * time.Second)
+ a.sweepOnce()
+ waitFor(t, "hibernation", func() bool { return a.entryCount() == 0 })
+}
+
+// A lost lease (failed renewal) stands the instance down immediately — no
+// waiting for idle: someone else may own the S3 prefix now.
+func TestActivatorLeaseLostStandsDown(t *testing.T) {
+ a, b, _ := testActivator(t, time.Hour)
+
+ release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+
+ b.lastRep().setErr(errors.New("lease renewal failed"))
+ a.sweepOnce() // idle window is an hour away; the lost lease alone must trigger the stop
+ waitFor(t, "stand-down", func() bool { return a.entryCount() == 0 })
+ if !b.lastProc().isStopped() {
+ t.Fatal("lease-lost instance's process was not stopped")
+ }
+ if !b.lastRep().isClosed() {
+ t.Fatal("lease-lost instance's replication was not closed")
+ }
+}
+
+// A dead child (crashed instance) is recycled by the next request: the stale
+// entry is stopped and a fresh wake happens in the same beginRequest call.
+func TestActivatorDeadChildRewakesOnRequest(t *testing.T) {
+ a, b, _ := testActivator(t, time.Hour)
+
+ release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+
+ p := b.lastProc()
+ p.mu.Lock()
+ p.dead = true
+ p.mu.Unlock()
+
+ release, err = a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatalf("beginRequest after crash: %v", err)
+ }
+ release()
+ if _, _, starts := b.counts(); starts != 2 {
+ t.Fatalf("starts=%d after crash+request, want 2 (recycle + rewake)", starts)
+ }
+}
+
+// A request that arrives mid-stop waits for the stop, then wakes fresh.
+func TestActivatorRequestDuringStopRewakes(t *testing.T) {
+ a, b, clk := testActivator(t, 10*time.Second)
+
+ release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+
+ // Hold the stop open so the entry sits in stopping.
+ gate := make(chan struct{})
+ p := b.lastProc()
+ p.mu.Lock()
+ p.stopGate = gate
+ p.mu.Unlock()
+
+ clk.advance(time.Minute)
+ a.sweepOnce()
+
+ got := make(chan error, 1)
+ go func() {
+ release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err == nil {
+ release()
+ }
+ got <- err
+ }()
+ time.Sleep(50 * time.Millisecond) // the request should now be parked on e.stopped
+ close(gate)
+ if err := <-got; err != nil {
+ t.Fatalf("request during stop: %v", err)
+ }
+ if _, _, starts := b.counts(); starts != 2 {
+ t.Fatalf("starts=%d, want 2 (sleep then rewake)", starts)
+ }
+}
+
+// A wake that exceeds the bound turns queued requests away with errWakeTimeout
+// (503 at the router) rather than piling connections up behind it.
+func TestActivatorWakeTimeout(t *testing.T) {
+ a, b, _ := testActivator(t, time.Minute)
+ old := wakeWaitBound
+ wakeWaitBound = 100 * time.Millisecond
+ t.Cleanup(func() { wakeWaitBound = old })
+
+ gate := make(chan struct{})
+ b.startGate = gate
+ t.Cleanup(func() { close(gate) }) // let the stuck wake finish at test end
+
+ _, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if !errors.Is(err, errWakeTimeout) {
+ t.Fatalf("got %v, want errWakeTimeout", err)
+ }
+}
+
+// Lease contention surfaces through the router as 503 + Retry-After, and a
+// failed wake leaves no entry behind (the next request retries the claim).
+func TestActivatorLeaseContention503(t *testing.T) {
+ a, b, _ := testActivator(t, time.Minute)
+ b.acquireErr = &errLeaseHeld{holder: "old-incarnation", retryAfter: 7 * time.Second}
+
+ reg, err := openRegistry(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatalf("open registry: %v", err)
+ }
+ if err := reg.upsert(Route{Host: "held.example.com", Kind: "instance", Socket: "/tmp/held.sock"}); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ rt, err := newRouter(reg, a)
+ if err != nil {
+ t.Fatalf("new router: %v", err)
+ }
+ front := httptest.NewServer(rt)
+ t.Cleanup(front.Close)
+
+ req, _ := http.NewRequest("GET", front.URL+"/", nil)
+ req.Host = "held.example.com"
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("request: %v", err)
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusServiceUnavailable {
+ t.Fatalf("status %d, want 503", resp.StatusCode)
+ }
+ if ra := resp.Header.Get("Retry-After"); ra != "8" { // 7s truncated + 1
+ t.Fatalf("Retry-After %q, want 8", ra)
+ }
+ if a.entryCount() != 0 {
+ t.Fatal("failed wake left an entry behind")
+ }
+}
+
+// With hibernation on, sibling services (kind "service") are untouched: the
+// router proxies them directly, no wake, no activity tracking.
+func TestActivatorServiceRoutesBypass(t *testing.T) {
+ sock := tmpSock(t)
+ ln, err := net.Listen("unix", sock)
+ if err != nil {
+ t.Fatalf("listen unix: %v", err)
+ }
+ backend := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ io.WriteString(w, "service ok")
+ })}
+ go backend.Serve(ln)
+ t.Cleanup(func() { backend.Close() })
+
+ a, b, _ := testActivator(t, time.Minute)
+ reg, err := openRegistry(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatalf("open registry: %v", err)
+ }
+ if err := reg.upsert(Route{Host: "svc.example.com", Kind: "service", Socket: sock}); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ rt, err := newRouter(reg, a)
+ if err != nil {
+ t.Fatalf("new router: %v", err)
+ }
+ front := httptest.NewServer(rt)
+ t.Cleanup(front.Close)
+
+ req, _ := http.NewRequest("GET", front.URL+"/", nil)
+ req.Host = "svc.example.com"
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("request: %v", err)
+ }
+ body, _ := io.ReadAll(resp.Body)
+ resp.Body.Close()
+ if resp.StatusCode != 200 || string(body) != "service ok" {
+ t.Fatalf("got %d %q, want 200 service ok", resp.StatusCode, body)
+ }
+ if acquires, _, _ := b.counts(); acquires != 0 {
+ t.Fatal("a service route touched the activator")
+ }
+}
+
+// shutdown checkpoints and stops every awake child before the router exits.
+func TestActivatorShutdownStopsChildren(t *testing.T) {
+ a, b, _ := testActivator(t, time.Hour)
+ for _, h := range []string{"one.example.com", "two.example.com"} {
+ release, err := a.beginRequest(context.Background(), testRoute(h))
+ if err != nil {
+ t.Fatalf("beginRequest %s: %v", h, err)
+ }
+ release()
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ a.shutdown(ctx)
+
+ if a.entryCount() != 0 {
+ t.Fatal("shutdown left entries behind")
+ }
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ for i, p := range b.procs {
+ if !p.stopped {
+ t.Fatalf("proc %d not stopped on shutdown", i)
+ }
+ }
+ for i, r := range b.reps {
+ if !r.closed {
+ t.Fatalf("replication %d not closed on shutdown", i)
+ }
+ }
+}
+
+func TestHibernateConfigFromEnv(t *testing.T) {
+ t.Setenv("LCHAT_HIBERNATE_BUCKET", "")
+ if cfg, err := hibernateConfigFromEnv(); err != nil || cfg != nil {
+ t.Fatalf("unset bucket: got %+v, %v; want nil, nil (feature off)", cfg, err)
+ }
+
+ t.Setenv("LCHAT_HIBERNATE_BUCKET", "b")
+ t.Setenv("LCHAT_HIBERNATE_PREFIX", "/instances/")
+ t.Setenv("LCHAT_HIBERNATE_IDLE", "90s")
+ cfg, err := hibernateConfigFromEnv()
+ if err != nil {
+ t.Fatalf("config: %v", err)
+ }
+ if cfg.Bucket != "b" || cfg.Prefix != "instances" || cfg.Idle != 90*time.Second {
+ t.Fatalf("got %+v", cfg)
+ }
+ if got := cfg.hibernatePrefix("x.example.com"); got != "instances/x.example.com" {
+ t.Fatalf("prefix: %q", got)
+ }
+
+ t.Setenv("LCHAT_HIBERNATE_IDLE", "")
+ cfg, err = hibernateConfigFromEnv()
+ if err != nil || cfg.Idle != 15*time.Minute {
+ t.Fatalf("default idle: %+v, %v", cfg, err)
+ }
+
+ t.Setenv("LCHAT_HIBERNATE_IDLE", "banana")
+ if _, err := hibernateConfigFromEnv(); err == nil {
+ t.Fatal("bad idle duration should error, not default silently")
+ }
+}
diff --git a/internal/activator/replica.go b/internal/activator/replica.go
index bfc37b3..aba972f 100644
--- a/internal/activator/replica.go
+++ b/internal/activator/replica.go
@@ -1,16 +1,17 @@
-// 🤖 Package activator holds the replication primitive for instance
+// 🤖 Package activator holds the replication primitives for instance
// hibernation (issue #550 Part 4 v2): litestream embedded as a library — pure
// modernc.org/sqlite, no cgo — with its S3 lease (lock.json via conditional
// writes) wrapped around replication as a single-writer lock. The lease is
// the truth of ownership: a second claimant fails loudly with a typed
// *litestream.LeaseExistsError (the CLI, by contrast, silently absorbs a
// second replicator — proven on #550), and a failed renewal means leadership
-// is lost and the caller must stand down. Nothing imports this package yet;
-// the Phase 2 activator (wake/sleep) builds on it.
+// is lost and the caller must stand down. The Phase 2 activator (wake/sleep,
+// activator.go in the main package) builds on these primitives.
package activator
import (
"context"
+ "errors"
"fmt"
"sync"
"time"
@@ -26,32 +27,94 @@ import (
// ~30s of its last renewal — the failover window of the whole design.
const LeaseTTL = 30 * time.Second
-// LeasedReplicator couples an exclusive S3 lease with a running litestream
-// replication of one SQLite database to the same S3 prefix. The lease is
-// acquired *before* replication starts and renewed on a ticker; Close stops
-// replication and releases the lease.
-type LeasedReplicator struct {
+// renewer keeps a held lease alive on a ticker. A failed renewal is recorded
+// and renewal stops — leadership is lost and the owner must stand down;
+// retrying could fence with a new claimant who took over after expiry.
+type renewer struct {
leaser *lss3.Leaser
- store *litestream.Store
- db *litestream.DB
mu sync.Mutex
lease *litestream.Lease
+ err error // set if a renewal ever fails (leadership lost)
- cancelRenew context.CancelFunc
- renewDone chan struct{}
- renewErr error // set if a renewal ever fails (leadership lost)
+ cancel context.CancelFunc
+ done chan struct{}
}
-// ReplicateWithLease acquires the single-writer lease at
-// s3://bucket/prefix/lock.json and, only if that succeeds, starts litestream
-// replication of dbPath to s3://bucket/prefix. If another process holds the
-// lease it fails LOUDLY with a *litestream.LeaseExistsError (check with
-// errors.As) — this is the signal the activator reacts to.
-func ReplicateWithLease(ctx context.Context, dbPath, bucket, prefix, region string) (*LeasedReplicator, error) {
+func startRenewer(leaser *lss3.Leaser, lease *litestream.Lease) *renewer {
+ ctx, cancel := context.WithCancel(context.Background())
+ rn := &renewer{leaser: leaser, lease: lease, cancel: cancel, done: make(chan struct{})}
+ go func() {
+ defer close(rn.done)
+ t := time.NewTicker(LeaseTTL / 3)
+ defer t.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-t.C:
+ rn.mu.Lock()
+ cur := rn.lease
+ rn.mu.Unlock()
+ renewed, err := rn.leaser.RenewLease(ctx, cur)
+ rn.mu.Lock()
+ if err != nil {
+ rn.err = err
+ } else {
+ rn.lease = renewed
+ }
+ rn.mu.Unlock()
+ if err != nil {
+ return
+ }
+ }
+ }
+ }()
+ return rn
+}
+
+// Err reports whether a renewal has failed (leadership lost).
+func (rn *renewer) Err() error {
+ rn.mu.Lock()
+ defer rn.mu.Unlock()
+ return rn.err
+}
+
+// stop halts renewal and returns the most recent lease (for release, or to
+// hand renewal duty to a successor renewer).
+func (rn *renewer) stop() *litestream.Lease {
+ rn.cancel()
+ <-rn.done
+ rn.mu.Lock()
+ defer rn.mu.Unlock()
+ return rn.lease
+}
+
+// HeldLease is an acquired single-writer lease that is not yet coupled to
+// replication. The wake path needs this split: the DB restore must happen
+// *under* the lock (restoring without it could later shadow newer S3 state
+// written by the true holder), but replication can only start once the DB
+// file exists. Renewal runs from acquisition, so a slow restore can't silently
+// outlive the TTL.
+type HeldLease struct {
+ leaser *lss3.Leaser
+ region string // the leaser's client keeps its region private; carry it for Replicate
+ rn *renewer
+}
+
+// AcquireLease claims the single-writer lease at s3://bucket/prefix/lock.json
+// and starts renewing it. If another process holds the lease it fails LOUDLY
+// with a *litestream.LeaseExistsError (check with errors.As) — this is the
+// signal the activator reacts to. Region may be empty; the AWS default chain
+// (env/profile) resolves it.
+func AcquireLease(ctx context.Context, bucket, prefix, region string) (*HeldLease, error) {
// The leaser wants a ready-made S3 client (unlike the replica client,
// which builds its own). Default AWS credential chain.
- cfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region))
+ var opts []func(*awsconfig.LoadOptions) error
+ if region != "" {
+ opts = append(opts, awsconfig.WithRegion(region))
+ }
+ cfg, err := awsconfig.LoadDefaultConfig(ctx, opts...)
if err != nil {
return nil, fmt.Errorf("load aws config: %w", err)
}
@@ -67,56 +130,75 @@ func ReplicateWithLease(ctx context.Context, dbPath, bucket, prefix, region stri
// Includes *litestream.LeaseExistsError when someone else holds it.
return nil, fmt.Errorf("acquire lease: %w", err)
}
+ return &HeldLease{leaser: leaser, region: region, rn: startRenewer(leaser, lease)}, nil
+}
+
+// Err reports whether lease renewal has failed (leadership lost).
+func (h *HeldLease) Err() error { return h.rn.Err() }
+
+// Release stops renewal and releases the lease — the bail-out when a wake
+// fails between acquisition and replication.
+func (h *HeldLease) Release(ctx context.Context) error {
+ lease := h.rn.stop()
+ if lease == nil {
+ return nil
+ }
+ return h.leaser.ReleaseLease(ctx, lease)
+}
+
+// Replicate trades the held lease for running replication of dbPath to the
+// lease's S3 prefix. On success the HeldLease is spent — renewal duty moves to
+// the returned LeasedReplicator (Close it to release). On failure the lease is
+// STILL HELD; the caller decides whether to retry or Release.
+func (h *HeldLease) Replicate(ctx context.Context, dbPath string) (*LeasedReplicator, error) {
+ if err := h.rn.Err(); err != nil {
+ return nil, fmt.Errorf("lease already lost: %w", err)
+ }
- // Lease held — start replication.
db := litestream.NewDB(dbPath)
client := lss3.NewReplicaClient()
- client.Bucket = bucket
- client.Path = prefix
- client.Region = region
+ client.Bucket = h.leaser.Bucket
+ client.Path = h.leaser.Path
+ client.Region = h.region
db.Replica = litestream.NewReplicaWithClient(db, client)
store := litestream.NewStore([]*litestream.DB{db}, litestream.DefaultCompactionLevels)
if err := store.Open(ctx); err != nil {
- _ = leaser.ReleaseLease(ctx, lease)
return nil, fmt.Errorf("open litestream store: %w", err)
}
- r := &LeasedReplicator{leaser: leaser, store: store, db: db, lease: lease}
+ // Hand renewal to the replicator. The handoff gap between stopping one
+ // renewer and starting the next is microseconds against a 30s TTL.
+ lease := h.rn.stop()
+ return &LeasedReplicator{leaser: h.leaser, store: store, db: db, rn: startRenewer(h.leaser, lease)}, nil
+}
- // Renew inside the TTL; a failed renewal means we lost the lease
- // (someone else may take over after expiry) — record it so the caller
- // can notice and stand down.
- renewCtx, cancel := context.WithCancel(context.Background())
- r.cancelRenew = cancel
- r.renewDone = make(chan struct{})
- go func() {
- defer close(r.renewDone)
- t := time.NewTicker(LeaseTTL / 3)
- defer t.Stop()
- for {
- select {
- case <-renewCtx.Done():
- return
- case <-t.C:
- r.mu.Lock()
- cur := r.lease
- r.mu.Unlock()
- renewed, err := r.leaser.RenewLease(renewCtx, cur)
- r.mu.Lock()
- if err != nil {
- r.renewErr = err
- } else {
- r.lease = renewed
- }
- r.mu.Unlock()
- if err != nil {
- return
- }
- }
- }
- }()
+// LeasedReplicator couples an exclusive S3 lease with a running litestream
+// replication of one SQLite database to the same S3 prefix. The lease is
+// acquired *before* replication starts and renewed on a ticker; Close stops
+// replication and releases the lease.
+type LeasedReplicator struct {
+ leaser *lss3.Leaser
+ store *litestream.Store
+ db *litestream.DB
+ rn *renewer
+}
+// ReplicateWithLease acquires the single-writer lease at
+// s3://bucket/prefix/lock.json and, only if that succeeds, starts litestream
+// replication of dbPath to s3://bucket/prefix. If another process holds the
+// lease it fails LOUDLY with a *litestream.LeaseExistsError (check with
+// errors.As) — this is the signal the activator reacts to.
+func ReplicateWithLease(ctx context.Context, dbPath, bucket, prefix, region string) (*LeasedReplicator, error) {
+ held, err := AcquireLease(ctx, bucket, prefix, region)
+ if err != nil {
+ return nil, err
+ }
+ r, err := held.Replicate(ctx, dbPath)
+ if err != nil {
+ _ = held.Release(ctx)
+ return nil, err
+ }
return r, nil
}
@@ -128,23 +210,14 @@ func (r *LeasedReplicator) SyncAndWait(ctx context.Context) error {
}
// Err reports whether lease renewal has failed (leadership lost).
-func (r *LeasedReplicator) Err() error {
- r.mu.Lock()
- defer r.mu.Unlock()
- return r.renewErr
-}
+func (r *LeasedReplicator) Err() error { return r.rn.Err() }
// Close stops replication (final sync included via litestream's DB.Close)
// and releases the lease so the next claimant can acquire immediately.
func (r *LeasedReplicator) Close(ctx context.Context) error {
- r.cancelRenew()
- <-r.renewDone
-
err := r.store.Close(ctx)
- r.mu.Lock()
- lease := r.lease
- r.mu.Unlock()
+ lease := r.rn.stop()
if lease != nil {
if e := r.leaser.ReleaseLease(ctx, lease); e != nil && err == nil {
err = e
@@ -152,3 +225,30 @@ func (r *LeasedReplicator) Close(ctx context.Context) error {
}
return err
}
+
+// Restore pulls the newest replicated state of s3://bucket/prefix into
+// dbPath via litestream's restore API — the cold-wake path when the local DB
+// is missing. Returns restored=false with a nil error when the prefix holds
+// no backup at all (a brand-new instance that has never replicated boots
+// fresh instead). The caller must hold the prefix's lease: restoring outside
+// it could resurrect a stale copy that later shadows the true holder's newer
+// state.
+func Restore(ctx context.Context, dbPath, bucket, prefix, region string) (bool, error) {
+ client := lss3.NewReplicaClient()
+ client.Bucket = bucket
+ client.Path = prefix
+ client.Region = region
+
+ // A nil *DB is litestream's own pattern for URL restores (the CLI does the
+ // same): Restore only needs the replica client and the output path.
+ r := litestream.NewReplicaWithClient(nil, client)
+ opt := litestream.NewRestoreOptions()
+ opt.OutputPath = dbPath
+ if err := r.Restore(ctx, opt); err != nil {
+ if errors.Is(err, litestream.ErrTxNotAvailable) || errors.Is(err, litestream.ErrNoSnapshots) {
+ return false, nil
+ }
+ return false, err
+ }
+ return true, nil
+}
diff --git a/router.go b/router.go
index 5d4b536..bedb186 100644
--- a/router.go
+++ b/router.go
@@ -25,7 +25,12 @@ import (
// never drop a live connection. See docs/instance-router.md.
type router struct {
reg *Registry
- upstreams atomic.Pointer[map[string]*httputil.ReverseProxy]
+ upstreams atomic.Pointer[map[string]upstream]
+ // act, when non-nil, is the hibernation activator (activator.go): instance
+ // routes stop being always-on systemd backends and become on-demand child
+ // processes woken per request. Nil (LCHAT_HIBERNATE_BUCKET unset) keeps
+ // today's proxy-only behavior.
+ act *activatorSvc
// fallbackURL, when set, is where requests for unknown hosts are redirected
// (e.g. the apex marketing site) instead of a bare 404 — a wildcard DNS
// record then makes typo'd or retired subdomains land somewhere useful. Note
@@ -36,8 +41,19 @@ type router struct {
fallbackURL string
}
-func newRouter(reg *Registry) (*router, error) {
- rt := &router{reg: reg}
+// upstream is one served host: the proxy to its backend, plus — when the
+// activator manages it — the route itself, so a request can trigger the wake.
+type upstream struct {
+ proxy *httputil.ReverseProxy
+ route Route
+ // hibernating marks an activator-managed instance: the router itself
+ // knows whether its process is live (it's the router's child), so "dead
+ // backend" detection is the activator's state table, not a failed dial.
+ hibernating bool
+}
+
+func newRouter(reg *Registry, act *activatorSvc) (*router, error) {
+ rt := &router{reg: reg, act: act}
if err := rt.reload(); err != nil {
return nil, err
}
@@ -51,20 +67,26 @@ func (rt *router) reload() error {
if err != nil {
return err
}
- m := make(map[string]*httputil.ReverseProxy, len(routes))
+ m := make(map[string]upstream, len(routes))
for _, t := range routes {
// Only serve ready routes: a 'creating'/'failed' entry has no working
// backend, so proxying it (and minting a cert for it) would just stall → 502.
if t.Socket == "" || t.Status != "ready" {
continue
}
- m[t.Host] = unixProxy(t.Socket)
+ // With hibernation on, every instance is activator-managed; sibling
+ // services (home, push) are externally run and stay proxy-only.
+ m[t.Host] = upstream{
+ proxy: unixProxy(t.Socket),
+ route: t,
+ hibernating: rt.act != nil && t.Kind == "instance",
+ }
}
rt.upstreams.Store(&m)
return nil
}
-func (rt *router) current() map[string]*httputil.ReverseProxy {
+func (rt *router) current() map[string]upstream {
if p := rt.upstreams.Load(); p != nil {
return *p
}
@@ -83,12 +105,19 @@ func (rt *router) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if h, _, err := net.SplitHostPort(host); err == nil {
host = h
}
- proxy, ok := rt.current()[host]
+ up, ok := rt.current()[host]
if !ok {
rt.unknownHost(w, r)
return
}
- proxy.ServeHTTP(w, r)
+ if up.hibernating {
+ // Wake (or join an in-progress wake) before proxying; the activator
+ // also stamps this request's activity for the idle clock. Note a bare
+ // TLS handshake never wakes anything — only a proxied request does.
+ rt.act.serveHTTP(w, r, up.route, up.proxy)
+ return
+ }
+ up.proxy.ServeHTTP(w, r)
}
// unknownHost answers a request for a host we don't serve: a redirect to the
@@ -175,7 +204,21 @@ func runRouter(args []string) {
if err != nil {
log.Fatalf("router: open registry %s: %v", *registryPath, err)
}
- rt, err := newRouter(reg)
+
+ // Instance hibernation (activator.go): opt-in via LCHAT_HIBERNATE_BUCKET.
+ // Unset keeps the classic shape — instances under systemd, router as a
+ // pure proxy — so self-hosters and the current fleet are untouched.
+ var act *activatorSvc
+ if hc, err := hibernateConfigFromEnv(); err != nil {
+ log.Fatalf("router: %v", err)
+ } else if hc != nil {
+ act = newActivator(*hc, &s3Backend{cfg: *hc})
+ act.start()
+ log.Printf("router: hibernation active (bucket %s, prefix %q, idle %s) — instances wake on demand",
+ hc.Bucket, hc.Prefix, hc.Idle)
+ }
+
+ rt, err := newRouter(reg, act)
if err != nil {
log.Fatalf("router: load routes: %v", err)
}
@@ -280,4 +323,12 @@ func runRouter(args []string) {
defer cancel()
_ = httpsSrv.Shutdown(ctx)
_ = httpSrv.Shutdown(ctx)
+ if act != nil {
+ // Hibernated instances are the router's children: checkpoint each to
+ // S3 and release its lease before exiting, so the next incarnation
+ // wakes them cleanly instead of finding bound sockets and held leases.
+ sctx, scancel := context.WithTimeout(context.Background(), 60*time.Second)
+ defer scancel()
+ act.shutdown(sctx)
+ }
}
diff --git a/router_test.go b/router_test.go
index fce6a7b..3902b5f 100644
--- a/router_test.go
+++ b/router_test.go
@@ -36,7 +36,7 @@ func TestRouterProxiesByHost(t *testing.T) {
if err := reg.upsert(Route{Host: "chat.example.com", Kind: "instance", Socket: sock}); err != nil {
t.Fatalf("upsert: %v", err)
}
- rt, err := newRouter(reg)
+ rt, err := newRouter(reg, nil)
if err != nil {
t.Fatalf("new router: %v", err)
}
@@ -101,7 +101,7 @@ func TestRouterProxiesWebSocket(t *testing.T) {
if err := reg.upsert(Route{Host: "127.0.0.1", Kind: "instance", Socket: sock}); err != nil {
t.Fatalf("upsert: %v", err)
}
- rt, err := newRouter(reg)
+ rt, err := newRouter(reg, nil)
if err != nil {
t.Fatalf("new router: %v", err)
}
@@ -149,7 +149,7 @@ func TestRouterRetriesAcrossRestart(t *testing.T) {
if err := reg.upsert(Route{Host: "chat.example.com", Kind: "instance", Socket: sock}); err != nil {
t.Fatal(err)
}
- rt, err := newRouter(reg)
+ rt, err := newRouter(reg, nil)
if err != nil {
t.Fatal(err)
}
@@ -211,7 +211,7 @@ func TestReloadOnlyServesReady(t *testing.T) {
t.Fatal(err)
}
}
- rt, err := newRouter(reg)
+ rt, err := newRouter(reg, nil)
if err != nil {
t.Fatal(err)
}
@@ -279,7 +279,7 @@ func TestRouterUnknownHostFallback(t *testing.T) {
if err != nil {
t.Fatalf("open registry: %v", err)
}
- rt, err := newRouter(reg)
+ rt, err := newRouter(reg, nil)
if err != nil {
t.Fatalf("new router: %v", err)
}