amadan
server: due-time index — hibernating instances wake for scheduled deliveries (#550 Part 4 v2) (#653)
The activator's biggest caveat was the named non-goal: a sleeping instance's future-messages/reminders didn't fire until something else woke it. Now they fire on time: - Instances report their earliest pending delivery at GET /api/next-due (admin-gated like every data endpoint; MIN(deliver_at) over the scheduled_messages table, 0 = nothing scheduled). - The activator's stop path probes it over the socket before killing the child and records the answer in the registry (routes.next_due, additive guarded-ALTER migration) — persisted, so due times survive router restarts; the registry is fleet-backed-up. Probe failure records 0 and the sleep proceeds: degrade to "sleeps until a request", never crash. - The sweeper reads the index each tick and wakes any sleeping host whose time has arrived — a request-less wake through the same singleflight state machine. Once up, the instance's own 5s scheduler delivers; the normal idle window puts it back to sleep, which re-records the NEXT due time — the cycle is self-maintaining. - A past-due time at sleep is recorded as-is and woken next tick; no minimum-sleep guard is needed because an awake instance only sleeps after a full idle window, so wake/sleep cycles can't thrash. Unit tests cover the full cycle, the no-ping-pong property, probe-failure fallback, restart survival, and the registry migration/filtering; the live S3 test now also exercises the real probe + record path.
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
10 files changed,
+558
−30
activator.go+114 −6activator_live_test.go+20 −1activator_test.go+205 −13control_test.go+70 −0db.go+11 −0main.go+3 −0miniapps.go+20 −0miniapps_test.go+60 −0registry.go+51 −9router.go+4 −1
diff --git a/activator.go b/activator.go| index 84c01f7..f91e509 100644 |
| --- a/activator.go |
| +++ b/activator.go |
| @@ -15,18 +15,28 @@ package main |
| // 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). |
| +// |
| +// The due-time index (a former non-goal, now in): scheduled future-messages/ |
| +// reminders fire on time even while their instance sleeps. At sleep the stop |
| +// path asks the instance for its earliest pending delivery (GET /api/next-due |
| +// over the socket, admin-gated) and records it in the registry (routes.next_due |
| +// — persisted so it survives router restarts); the sweeper wakes any sleeping |
| +// host whose time has arrived, with no client request. Once up, the instance's |
| +// own scheduler (miniapps.go runScheduler, a 5s tick) delivers, and the normal |
| +// idle window puts it back to sleep — which re-records the NEXT due time, so |
| +// the cycle maintains itself. Worst-case lateness ≈ one activator sweep |
| +// interval (≤30s) + spawn + one scheduler tick (≤5s). |
| import ( |
| "context" |
| + "encoding/json" |
| "errors" |
| "fmt" |
| + "io" |
| "log" |
| "net/http" |
| "os" |
| @@ -96,6 +106,18 @@ type actBackend interface { |
| 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) |
| + // nextDue asks a running instance for its earliest pending scheduled |
| + // delivery (unix seconds; 0 = nothing scheduled). Queried at sleep time, |
| + // while the socket still answers, to feed the due-time index. |
| + nextDue(ctx context.Context, rt Route) (int64, error) |
| +} |
| + |
| +// actDueStore persists the due-time index — the registry in production |
| +// (routes.next_due), so a sleeping host's timer survives router restarts |
| +// instead of dying with the process. |
| +type actDueStore interface { |
| + setNextDue(host string, due int64) error |
| + dueInstances(now int64) ([]Route, error) |
| } |
| // actLease is a held-but-not-yet-replicating lease (wake in progress). |
| @@ -186,6 +208,7 @@ var errWakeTimeout = errors.New("timed out waiting for instance wake") |
| type activatorSvc struct { |
| cfg hibernateConfig |
| backend actBackend |
| + due actDueStore |
| now func() time.Time // a seam for the unit tests' clock |
| mu sync.Mutex |
| @@ -195,10 +218,11 @@ type activatorSvc struct { |
| sweepDone chan struct{} |
| } |
| -func newActivator(cfg hibernateConfig, backend actBackend) *activatorSvc { |
| +func newActivator(cfg hibernateConfig, backend actBackend, due actDueStore) *activatorSvc { |
| return &activatorSvc{ |
| cfg: cfg, |
| backend: backend, |
| + due: due, |
| now: time.Now, |
| entries: make(map[string]*actEntry), |
| } |
| @@ -382,10 +406,33 @@ func (a *activatorSvc) beginStopLocked(e *actEntry) { |
| // 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) { |
| + // The due-time index: ask the instance for its earliest pending scheduled |
| + // delivery BEFORE stopping it — the socket dies with the process. (When the |
| + // stop was triggered by a dead child, the query just fails fast.) Any |
| + // failure records 0, i.e. "sleeps until a request": a missed probe must |
| + // degrade to pre-index behavior, never block or crash the sleep. A due time |
| + // already in the past is recorded as-is and the sweeper wakes the host on |
| + // its next tick — there is deliberately no minimum-sleep guard beyond that, |
| + // because thrash is already impossible: an awake instance only sleeps after |
| + // a full idle window, so wake/sleep cycles can never be spaced closer than |
| + // cfg.Idle. |
| + dueCtx, cancelDue := context.WithTimeout(context.Background(), 5*time.Second) |
| + due, err := a.backend.nextDue(dueCtx, e.route) |
| + cancelDue() |
| + if err != nil { |
| + log.Printf("activator: next-due probe %s: %v — recording none (it sleeps until a request)", e.host, err) |
| + due = 0 |
| + } |
| + if err := a.due.setNextDue(e.host, due); err != nil { |
| + log.Printf("activator: record next-due %s: %v", e.host, err) |
| + } else if due > 0 { |
| + log.Printf("activator: %s next scheduled delivery at %s — will wake for it", e.host, time.Unix(due, 0).UTC().Format(time.RFC3339)) |
| + } |
| + |
| // 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) |
| + err = e.proc.stop(procCtx) |
| cancelProc() |
| if err != nil { |
| log.Printf("activator: stop %s: %v", e.host, err) |
| @@ -435,9 +482,17 @@ func (a *activatorSvc) start() { |
| } |
| // sweepOnce examines every awake instance once: idle ones hibernate; a lost |
| -// lease or dead child stands down immediately regardless of activity. |
| +// lease or dead child stands down immediately regardless of activity. It then |
| +// checks the due-time index and wakes any sleeping host whose scheduled |
| +// delivery has come due. |
| func (a *activatorSvc) sweepOnce() { |
| now := a.now() |
| + // The due query runs before the lock — it's a registry read, and nothing |
| + // under a.mu should wait on SQLite. |
| + dueRoutes, err := a.due.dueInstances(now.Unix()) |
| + if err != nil { |
| + log.Printf("activator: due-time sweep: %v", err) |
| + } |
| a.mu.Lock() |
| for _, e := range a.entries { |
| if e.state != actAwake { |
| @@ -458,6 +513,24 @@ func (a *activatorSvc) sweepOnce() { |
| a.beginStopLocked(e) |
| } |
| } |
| + // Due-time wakes: a request-less wake through the same singleflight state |
| + // machine, so a client request arriving mid-wake just coalesces. Only hosts |
| + // with NO live entry are woken — an awake host is already serving its |
| + // schedule; a waking one is on its way; a stopping one will re-record its |
| + // next_due when the stop completes and be picked up on a later tick. The |
| + // registry's stale past value while the host is up is harmless for the same |
| + // reason. A failed due wake (e.g. a lease still held by a crashed |
| + // incarnation) removes the entry, and the index — only rewritten at the |
| + // next successful sleep — naturally retries it every tick until it works. |
| + for _, rt := range dueRoutes { |
| + if _, live := a.entries[rt.Host]; live { |
| + continue |
| + } |
| + log.Printf("activator: %s has a scheduled delivery due — waking it", rt.Host) |
| + e := &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})} |
| + a.entries[rt.Host] = e |
| + go a.wake(e) |
| + } |
| a.mu.Unlock() |
| } |
| @@ -585,6 +658,41 @@ func (b *s3Backend) restore(ctx context.Context, host, dbPath string) error { |
| return nil |
| } |
| +// nextDue asks the (still-running) instance for its earliest pending scheduled |
| +// delivery over its socket. The endpoint is admin-gated like every instance |
| +// data endpoint, so present the instance's own admin token — read from the same |
| +// env file the spawn path hands the child (the activator holds it either way). |
| +func (b *s3Backend) nextDue(ctx context.Context, rt Route) (int64, error) { |
| + fileEnv, err := readEnvFile(filepath.Join(instanceDataDir, rt.Host+".env")) |
| + if err != nil { |
| + return 0, fmt.Errorf("read env file: %w", err) |
| + } |
| + token := fileEnv["LCHAT_ADMIN_TOKEN"] |
| + if token == "" { |
| + return 0, errors.New("no LCHAT_ADMIN_TOKEN in env file") |
| + } |
| + req, err := http.NewRequestWithContext(ctx, "GET", "http://lchat/api/next-due", nil) |
| + if err != nil { |
| + return 0, err |
| + } |
| + req.Header.Set("Authorization", "Bearer "+token) |
| + resp, err := socketClient(rt.Socket).Do(req) |
| + if err != nil { |
| + return 0, err |
| + } |
| + defer resp.Body.Close() |
| + if resp.StatusCode != http.StatusOK { |
| + return 0, fmt.Errorf("GET /api/next-due: HTTP %d", resp.StatusCode) |
| + } |
| + var body struct { |
| + Due int64 `json:"due"` |
| + } |
| + if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<12)).Decode(&body); err != nil { |
| + return 0, fmt.Errorf("decode /api/next-due: %w", err) |
| + } |
| + return body.Due, 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 |
diff --git a/activator_live_test.go b/activator_live_test.go| index 218d824..d7efbd0 100644 |
| --- a/activator_live_test.go |
| +++ b/activator_live_test.go |
| @@ -101,7 +101,7 @@ func TestActivatorLiveHibernation(t *testing.T) { |
| } |
| cfg := hibernateConfig{Bucket: bucket, Region: region, Prefix: prefix, Idle: 3 * time.Second} |
| - act := newActivator(cfg, &s3Backend{cfg: cfg}) |
| + act := newActivator(cfg, &s3Backend{cfg: cfg}, reg) |
| act.start() |
| t.Cleanup(func() { |
| ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) |
| @@ -155,8 +155,27 @@ func TestActivatorLiveHibernation(t *testing.T) { |
| resp.Body.Close() |
| t.Logf("wrote group name %q through the admin API", groupName) |
| + // The due-time probe answers on the real instance (admin-gated; nothing |
| + // scheduled → 0). The sleep below re-runs the same call over the socket. |
| + nreq, _ := http.NewRequest("GET", front.URL+"/api/next-due", nil) |
| + nreq.Host = host |
| + nreq.Header.Set("Authorization", "Bearer "+adminToken) |
| + nresp, err := http.DefaultClient.Do(nreq) |
| + if err != nil { |
| + t.Fatalf("GET /api/next-due: %v", err) |
| + } |
| + nbody, _ := io.ReadAll(nresp.Body) |
| + nresp.Body.Close() |
| + if nresp.StatusCode != 200 || !strings.Contains(string(nbody), `"due":0`) { |
| + t.Fatalf("next-due probe: %d %s, want 200 with due 0", nresp.StatusCode, nbody) |
| + } |
| + |
| // --- 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 }) |
| + // The sleep recorded the (empty) due-time index in the registry. |
| + if rt, err := reg.get(host); err != nil || rt == nil || rt.NextDue != 0 { |
| + t.Fatalf("registry next_due after sleep = %+v, %v; want 0", rt, err) |
| + } |
| if _, err := net.DialTimeout("unix", route.Socket, 300*time.Millisecond); err == nil { |
| t.Fatal("instance socket still accepting after hibernation") |
| } |
diff --git a/activator_test.go b/activator_test.go| index 04a0fcf..3973c79 100644 |
| --- a/activator_test.go |
| +++ b/activator_test.go |
| @@ -96,6 +96,8 @@ type fakeBackend struct { |
| startGate chan struct{} // when non-nil, start() blocks until closed |
| reps []*fakeRep |
| procs []*fakeProc |
| + dueVal int64 // what the instance reports at the sleep-time next-due probe |
| + dueErr error // when non-nil, the probe fails (dead socket, missing token, …) |
| } |
| func (b *fakeBackend) acquire(context.Context, string) (actLease, error) { |
| @@ -131,6 +133,21 @@ func (b *fakeBackend) start(context.Context, Route) (actProc, error) { |
| return proc, nil |
| } |
| +func (b *fakeBackend) nextDue(context.Context, Route) (int64, error) { |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + if b.dueErr != nil { |
| + return 0, b.dueErr |
| + } |
| + return b.dueVal, nil |
| +} |
| + |
| +func (b *fakeBackend) setDue(v int64) { |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + b.dueVal = v |
| +} |
| + |
| func (b *fakeBackend) counts() (acquires, restores, starts int) { |
| b.mu.Lock() |
| defer b.mu.Unlock() |
| @@ -166,13 +183,19 @@ func (c *fakeClock) advance(d time.Duration) { |
| c.t = c.t.Add(d) |
| } |
| -func testActivator(t *testing.T, idle time.Duration) (*activatorSvc, *fakeBackend, *fakeClock) { |
| +// testActivator wires the fakes to a REAL (temp-dir) registry as the due store, |
| +// so the due-time tests exercise the actual routes.next_due SQL, not a fake of it. |
| +func testActivator(t *testing.T, idle time.Duration) (*activatorSvc, *fakeBackend, *fakeClock, *Registry) { |
| t.Helper() |
| b := &fakeBackend{} |
| clk := &fakeClock{t: time.Unix(1_700_000_000, 0)} |
| - a := newActivator(hibernateConfig{Bucket: "test-bucket", Idle: idle}, b) |
| + reg, err := openRegistry(filepath.Join(t.TempDir(), "registry.db")) |
| + if err != nil { |
| + t.Fatalf("open registry: %v", err) |
| + } |
| + a := newActivator(hibernateConfig{Bucket: "test-bucket", Idle: idle}, b, reg) |
| a.now = clk.now |
| - return a, b, clk |
| + return a, b, clk, reg |
| } |
| func testRoute(host string) Route { |
| @@ -195,7 +218,7 @@ func (a *activatorSvc) entryCount() int { |
| // 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) |
| + a, b, _, _ := testActivator(t, time.Minute) |
| b.startGate = make(chan struct{}) |
| const n = 12 |
| @@ -233,7 +256,7 @@ func TestActivatorWakeCoalesces(t *testing.T) { |
| // 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) |
| + a, b, clk, _ := testActivator(t, 10*time.Second) |
| release, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| if err != nil { |
| @@ -272,7 +295,7 @@ func TestActivatorIdleSleep(t *testing.T) { |
| // 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) |
| + a, b, clk, _ := testActivator(t, 10*time.Second) |
| release, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| if err != nil { |
| @@ -298,7 +321,7 @@ func TestActivatorInflightKeepsAwake(t *testing.T) { |
| // 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) |
| + a, b, _, _ := testActivator(t, time.Hour) |
| release, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| if err != nil { |
| @@ -320,7 +343,7 @@ func TestActivatorLeaseLostStandsDown(t *testing.T) { |
| // 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) |
| + a, b, _, _ := testActivator(t, time.Hour) |
| release, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| if err != nil { |
| @@ -345,7 +368,7 @@ func TestActivatorDeadChildRewakesOnRequest(t *testing.T) { |
| // 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) |
| + a, b, clk, _ := testActivator(t, 10*time.Second) |
| release, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| if err != nil { |
| @@ -384,7 +407,7 @@ func TestActivatorRequestDuringStopRewakes(t *testing.T) { |
| // 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) |
| + a, b, _, _ := testActivator(t, time.Minute) |
| old := wakeWaitBound |
| wakeWaitBound = 100 * time.Millisecond |
| t.Cleanup(func() { wakeWaitBound = old }) |
| @@ -402,7 +425,7 @@ func TestActivatorWakeTimeout(t *testing.T) { |
| // 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) |
| + 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")) |
| @@ -451,7 +474,7 @@ func TestActivatorServiceRoutesBypass(t *testing.T) { |
| go backend.Serve(ln) |
| t.Cleanup(func() { backend.Close() }) |
| - a, b, _ := testActivator(t, time.Minute) |
| + a, b, _, _ := testActivator(t, time.Minute) |
| reg, err := openRegistry(filepath.Join(t.TempDir(), "registry.db")) |
| if err != nil { |
| t.Fatalf("open registry: %v", err) |
| @@ -484,7 +507,7 @@ func TestActivatorServiceRoutesBypass(t *testing.T) { |
| // shutdown checkpoints and stops every awake child before the router exits. |
| func TestActivatorShutdownStopsChildren(t *testing.T) { |
| - a, b, _ := testActivator(t, time.Hour) |
| + 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 { |
| @@ -514,6 +537,175 @@ func TestActivatorShutdownStopsChildren(t *testing.T) { |
| } |
| } |
| +// --- due-time index -------------------------------------------------------------- |
| + |
| +// entryAwake reports whether host has a live entry in the awake state — how the |
| +// due tests observe an async due-time wake resolving. |
| +func (a *activatorSvc) entryAwake(host string) bool { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + e := a.entries[host] |
| + return e != nil && e.state == actAwake |
| +} |
| + |
| +// The full self-maintaining cycle: sleep records the instance's next due time |
| +// in the registry; the sweeper wakes the host (no request) when it arrives; the |
| +// following sleep re-records — and once nothing is scheduled, no more due wakes. |
| +func TestActivatorDueTimeWakeCycle(t *testing.T) { |
| + a, b, clk, reg := testActivator(t, 10*time.Second) |
| + route := testRoute("due.example.com") |
| + if err := reg.upsert(route); err != nil { |
| + t.Fatalf("upsert: %v", err) |
| + } |
| + |
| + // Wake by request; the instance has a delivery scheduled 30s out. |
| + dueAt := clk.now().Add(30 * time.Second).Unix() |
| + b.setDue(dueAt) |
| + release, err := a.beginRequest(context.Background(), route) |
| + if err != nil { |
| + t.Fatalf("beginRequest: %v", err) |
| + } |
| + release() |
| + |
| + // Idle out: the sleep probes the instance and records its due time. |
| + clk.advance(11 * time.Second) |
| + a.sweepOnce() |
| + waitFor(t, "hibernation", func() bool { return a.entryCount() == 0 }) |
| + if rt, err := reg.get(route.Host); err != nil || rt == nil || rt.NextDue != dueAt { |
| + t.Fatalf("registry next_due = %+v, %v; want %d", rt, err, dueAt) |
| + } |
| + |
| + // Before the due time: sweeps leave it asleep. |
| + a.sweepOnce() |
| + if _, _, starts := b.counts(); starts != 1 { |
| + t.Fatalf("woke before its due time (starts=%d)", starts) |
| + } |
| + |
| + // At the due time: the sweeper wakes it with no client request. |
| + clk.advance(20 * time.Second) // now = wake+31s ≥ due |
| + a.sweepOnce() |
| + waitFor(t, "due-time wake", func() bool { return a.entryAwake(route.Host) }) |
| + if _, _, starts := b.counts(); starts != 2 { |
| + t.Fatalf("starts=%d after due wake, want 2", starts) |
| + } |
| + |
| + // The woken instance stays up for the normal idle window, then sleeps and |
| + // re-records — nothing scheduled now, so the index clears and stays quiet. |
| + b.setDue(0) |
| + clk.advance(11 * time.Second) |
| + a.sweepOnce() |
| + waitFor(t, "re-hibernation", func() bool { return a.entryCount() == 0 }) |
| + if rt, _ := reg.get(route.Host); rt.NextDue != 0 { |
| + t.Fatalf("next_due = %d after empty-schedule sleep, want 0", rt.NextDue) |
| + } |
| + clk.advance(time.Hour) |
| + a.sweepOnce() |
| + if _, _, starts := b.counts(); starts != 2 { |
| + t.Fatalf("starts=%d with nothing scheduled, want 2 (no phantom due wake)", starts) |
| + } |
| +} |
| + |
| +// A due time already in the past at sleep time (a delivery that raced the stop) |
| +// is recorded as-is: the sweeper wakes the host on its next tick — once — and |
| +// there is no ping-pong, because a woken instance can't sleep again until a |
| +// full idle window passes. |
| +func TestActivatorDuePastWakesOnceNoPingPong(t *testing.T) { |
| + a, b, clk, reg := testActivator(t, 10*time.Second) |
| + route := testRoute("late.example.com") |
| + if err := reg.upsert(route); err != nil { |
| + t.Fatalf("upsert: %v", err) |
| + } |
| + b.setDue(clk.now().Add(-5 * time.Second).Unix()) // overdue before it even sleeps |
| + |
| + release, err := a.beginRequest(context.Background(), route) |
| + if err != nil { |
| + t.Fatalf("beginRequest: %v", err) |
| + } |
| + release() |
| + clk.advance(11 * time.Second) |
| + a.sweepOnce() |
| + waitFor(t, "hibernation", func() bool { return a.entryCount() == 0 }) |
| + |
| + // Next tick: woken for the overdue delivery. |
| + a.sweepOnce() |
| + waitFor(t, "overdue wake", func() bool { return a.entryAwake(route.Host) }) |
| + |
| + // No thrash: with the wake fresh (clock unmoved), repeated sweeps neither |
| + // re-sleep it nor double-wake it, even though next_due is still in the past. |
| + for i := 0; i < 5; i++ { |
| + a.sweepOnce() |
| + } |
| + if a.entryCount() != 1 || !a.entryAwake(route.Host) { |
| + t.Fatal("due-woken instance did not stay awake through its idle window") |
| + } |
| + if _, _, starts := b.counts(); starts != 2 { |
| + t.Fatalf("starts=%d, want 2 (request wake + one due wake)", starts) |
| + } |
| +} |
| + |
| +// A failed next-due probe (dead socket, missing admin token, …) records 0 and |
| +// the sleep proceeds — the instance degrades to "sleeps until a request", |
| +// never a crash and never a stuck stop. |
| +func TestActivatorDueProbeFailureRecordsZero(t *testing.T) { |
| + a, b, clk, reg := testActivator(t, 10*time.Second) |
| + route := testRoute("deaf.example.com") |
| + if err := reg.upsert(route); err != nil { |
| + t.Fatalf("upsert: %v", err) |
| + } |
| + b.dueErr = errors.New("dial unix: no such file") |
| + |
| + release, err := a.beginRequest(context.Background(), route) |
| + if err != nil { |
| + t.Fatalf("beginRequest: %v", err) |
| + } |
| + release() |
| + clk.advance(11 * time.Second) |
| + a.sweepOnce() |
| + waitFor(t, "hibernation despite probe failure", func() bool { return a.entryCount() == 0 }) |
| + |
| + if rt, _ := reg.get(route.Host); rt.NextDue != 0 { |
| + t.Fatalf("next_due = %d after failed probe, want 0", rt.NextDue) |
| + } |
| + clk.advance(time.Hour) |
| + a.sweepOnce() |
| + if _, _, starts := b.counts(); starts != 1 { |
| + t.Fatalf("starts=%d, want 1 (a failed probe must not invent a wake)", starts) |
| + } |
| +} |
| + |
| +// The index lives in the registry, so a due time recorded before a router |
| +// restart still fires: a brand-new activator over the same registry — no |
| +// in-memory state carried over — wakes the host when its time arrives. |
| +func TestActivatorDueSurvivesRestart(t *testing.T) { |
| + a1, b1, clk, reg := testActivator(t, 10*time.Second) |
| + route := testRoute("reborn.example.com") |
| + if err := reg.upsert(route); err != nil { |
| + t.Fatalf("upsert: %v", err) |
| + } |
| + b1.setDue(clk.now().Add(30 * time.Second).Unix()) |
| + |
| + release, err := a1.beginRequest(context.Background(), route) |
| + if err != nil { |
| + t.Fatalf("beginRequest: %v", err) |
| + } |
| + release() |
| + clk.advance(11 * time.Second) |
| + a1.sweepOnce() |
| + waitFor(t, "hibernation", func() bool { return a1.entryCount() == 0 }) |
| + |
| + // "Restart": a fresh activator, same registry. |
| + b2 := &fakeBackend{} |
| + a2 := newActivator(hibernateConfig{Bucket: "test-bucket", Idle: 10 * time.Second}, b2, reg) |
| + a2.now = clk.now |
| + |
| + clk.advance(25 * time.Second) // past the due time |
| + a2.sweepOnce() |
| + waitFor(t, "post-restart due wake", func() bool { return a2.entryAwake(route.Host) }) |
| + if _, _, starts := b2.counts(); starts != 1 { |
| + t.Fatalf("starts=%d on the new activator, want 1", starts) |
| + } |
| +} |
| + |
| func TestHibernateConfigFromEnv(t *testing.T) { |
| t.Setenv("LCHAT_HIBERNATE_BUCKET", "") |
| if cfg, err := hibernateConfigFromEnv(); err != nil || cfg != nil { |
diff --git a/control_test.go b/control_test.go| index b9ac305..1408efa 100644 |
| --- a/control_test.go |
| +++ b/control_test.go |
| @@ -253,6 +253,10 @@ INSERT INTO routes (host, kind, socket, created_at) VALUES ('old.example.com', ' |
| if rt.HomeRegion != "" || rt.Provider != "" { |
| t.Errorf("pre-existing row: home_region=%q provider=%q, want the implicit \"\"/\"\"", rt.HomeRegion, rt.Provider) |
| } |
| + // The pre-seam schema predates next_due too — the same open must add it. |
| + if rt.NextDue != 0 { |
| + t.Errorf("pre-existing row: next_due=%d, want 0", rt.NextDue) |
| + } |
| // New values round-trip through the migrated table. |
| rt.HomeRegion, rt.Provider = "us-east-1", "aws" |
| @@ -269,3 +273,69 @@ INSERT INTO routes (host, kind, socket, created_at) VALUES ('old.example.com', ' |
| t.Errorf("reopen after migration: %v", err) |
| } |
| } |
| + |
| +// The due-time index's registry half (activator.go's actDueStore): setNextDue |
| +// is a narrow write, dueInstances returns exactly the ready instances whose |
| +// time has arrived, and a route rewrite (deploy's read-modify-write upsert) |
| +// must not clobber a sleeping host's due time. |
| +func TestRegistryDueTimeIndex(t *testing.T) { |
| + reg, err := openRegistry(filepath.Join(t.TempDir(), "reg.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + seed := []Route{ |
| + {Host: "sleepy.example.com", Kind: "instance", Socket: "/run/a.sock"}, |
| + {Host: "quiet.example.com", Kind: "instance", Socket: "/run/b.sock"}, |
| + {Host: "svc.example.com", Kind: "service", Socket: "/run/c.sock"}, |
| + {Host: "broken.example.com", Kind: "instance", Socket: "/run/d.sock", Status: "failed"}, |
| + } |
| + for _, rt := range seed { |
| + if err := reg.upsert(rt); err != nil { |
| + t.Fatalf("upsert %s: %v", rt.Host, err) |
| + } |
| + } |
| + const now = int64(1_700_000_000) |
| + for host, due := range map[string]int64{ |
| + "sleepy.example.com": now - 5, // due |
| + "svc.example.com": now - 5, // due but a service — never woken |
| + "broken.example.com": now - 5, // due but not ready — never woken |
| + // quiet: next_due stays 0 — nothing scheduled, never due-woken |
| + } { |
| + if err := reg.setNextDue(host, due); err != nil { |
| + t.Fatalf("setNextDue %s: %v", host, err) |
| + } |
| + } |
| + |
| + due, err := reg.dueInstances(now) |
| + if err != nil { |
| + t.Fatalf("dueInstances: %v", err) |
| + } |
| + if len(due) != 1 || due[0].Host != "sleepy.example.com" { |
| + t.Fatalf("dueInstances = %+v, want just sleepy.example.com", due) |
| + } |
| + // Not yet due: a future timestamp stays out of the set. |
| + if err := reg.setNextDue("sleepy.example.com", now+60); err != nil { |
| + t.Fatal(err) |
| + } |
| + if due, _ := reg.dueInstances(now); len(due) != 0 { |
| + t.Fatalf("future due time returned early: %+v", due) |
| + } |
| + |
| + // Read-modify-write (how deploy/rollback update a route) carries next_due. |
| + rt, err := reg.get("sleepy.example.com") |
| + if err != nil || rt == nil { |
| + t.Fatalf("get: %+v, %v", rt, err) |
| + } |
| + rt.Version = "v2" |
| + if err := reg.upsert(*rt); err != nil { |
| + t.Fatalf("upsert rewrite: %v", err) |
| + } |
| + if got, _ := reg.get("sleepy.example.com"); got.NextDue != now+60 { |
| + t.Fatalf("route rewrite clobbered next_due: %d, want %d", got.NextDue, now+60) |
| + } |
| + |
| + // An unknown host is a quiet no-op — it can't be woken anyway. |
| + if err := reg.setNextDue("ghost.example.com", now); err != nil { |
| + t.Fatalf("setNextDue unknown host: %v", err) |
| + } |
| +} |
diff --git a/db.go b/db.go| index 1be5674..b07767d 100644 |
| --- a/db.go |
| +++ b/db.go |
| @@ -2570,6 +2570,17 @@ func (db *DB) countScheduledMessages(threadID string) int { |
| return n |
| } |
| +// nextScheduledDeliverAt returns the earliest pending deliver_at (unix |
| +// seconds), or 0 when nothing is scheduled. Delivered schedules are deleted |
| +// (deliverScheduledMessage), so MIN over the whole table IS the next due time — |
| +// this feeds the router's due-time index so a hibernating instance is woken in |
| +// time to fire it (see getNextDue). |
| +func (db *DB) nextScheduledDeliverAt() (int64, error) { |
| + var due int64 |
| + err := db.sql.QueryRow("SELECT COALESCE(MIN(deliver_at), 0) FROM scheduled_messages").Scan(&due) |
| + return due, err |
| +} |
| + |
| // dueScheduledMessages returns schedules whose time has come, oldest first. |
| func (db *DB) dueScheduledMessages(now int64, limit int) ([]ScheduledMessage, error) { |
| rows, err := db.sql.Query( |
diff --git a/main.go b/main.go| index da7e4df..57e88ae 100644 |
| --- a/main.go |
| +++ b/main.go |
| @@ -395,6 +395,9 @@ func (s *Server) handler() http.Handler { |
| mux.HandleFunc("POST /api/schedule", s.postSchedule) |
| mux.HandleFunc("GET /api/schedule", s.getSchedules) |
| mux.HandleFunc("DELETE /api/schedule/{id}", s.deleteSchedule) |
| + // The activator's due-time probe (admin-gated; queried over the instance |
| + // socket at sleep time so hibernated schedules still fire on time). |
| + mux.HandleFunc("GET /api/next-due", s.getNextDue) |
| mux.HandleFunc("GET /api/threads/{id}/state", s.getThreadStateHandler) |
| mux.HandleFunc("PUT /api/threads/{id}/state", s.putThreadStateHandler) |
diff --git a/miniapps.go b/miniapps.go| index 139ac25..354840a 100644 |
| --- a/miniapps.go |
| +++ b/miniapps.go |
| @@ -140,6 +140,26 @@ func (s *Server) deleteSchedule(w http.ResponseWriter, r *http.Request) { |
| writeJSON(w, http.StatusOK, map[string]any{"ok": true}) |
| } |
| +// getNextDue reports the instance's earliest pending scheduled delivery: |
| +// {due: <unix seconds>}, 0 when nothing is scheduled. It exists for the |
| +// router's activator: at sleep time it records this so a hibernating instance |
| +// is woken when the delivery comes due (activator.go's due-time index). |
| +// Admin-gated — the payload is a single timestamp the server already stores in |
| +// the clear (the schedule's (thread, time) metadata), but every data endpoint |
| +// carries auth, and the activator already holds the instance's admin token via |
| +// the env file it spawns instances with. |
| +func (s *Server) getNextDue(w http.ResponseWriter, r *http.Request) { |
| + if !s.requireAdmin(w, r) { |
| + return |
| + } |
| + due, err := s.db.nextScheduledDeliverAt() |
| + if err != nil { |
| + httpError(w, http.StatusInternalServerError, "db error") |
| + return |
| + } |
| + writeJSON(w, http.StatusOK, map[string]any{"due": due}) |
| +} |
| + |
| // getThreadStateHandler returns a slot's snapshot: {blob, version}. version 0 + |
| // empty blob means nothing saved yet (the client starts a fresh fold). |
| func (s *Server) getThreadStateHandler(w http.ResponseWriter, r *http.Request) { |
diff --git a/miniapps_test.go b/miniapps_test.go| index a590dfe..9c53cb9 100644 |
| --- a/miniapps_test.go |
| +++ b/miniapps_test.go |
| @@ -142,6 +142,66 @@ func TestScheduleDelivery(t *testing.T) { |
| } |
| } |
| +// The activator's due-time probe: /api/next-due is admin-gated and reports the |
| +// earliest pending deliver_at (0 when nothing is scheduled) so the router can |
| +// wake a hibernating instance in time to fire it. |
| +func TestNextDue(t *testing.T) { |
| + ts, srv := newTestServer(t) |
| + tok, _, tid := joinedMember(t, srv, "Sam") |
| + |
| + getDue := func(bearer string) (int, int64) { |
| + t.Helper() |
| + req, _ := http.NewRequest("GET", ts.URL+"/api/next-due", nil) |
| + if bearer != "" { |
| + req.Header.Set("Authorization", "Bearer "+bearer) |
| + } |
| + resp, err := http.DefaultClient.Do(req) |
| + if err != nil { |
| + t.Fatalf("GET /api/next-due: %v", err) |
| + } |
| + if resp.StatusCode != http.StatusOK { |
| + resp.Body.Close() |
| + return resp.StatusCode, 0 |
| + } |
| + var body struct { |
| + Due int64 `json:"due"` |
| + } |
| + decodeJSON(t, resp, &body) |
| + return resp.StatusCode, body.Due |
| + } |
| + |
| + // No auth → 401; a plain member's device token isn't admin either. |
| + if code, _ := getDue(""); code != http.StatusUnauthorized { |
| + t.Fatalf("unauthenticated: want 401, got %d", code) |
| + } |
| + if code, _ := getDue(tok); code != http.StatusUnauthorized { |
| + t.Fatalf("member token: want 401, got %d", code) |
| + } |
| + |
| + // Nothing scheduled → 0. |
| + if code, due := getDue("admintok"); code != 200 || due != 0 { |
| + t.Fatalf("empty schedule: want 200/0, got %d/%d", code, due) |
| + } |
| + |
| + // The earliest of several pending schedules wins, regardless of post order. |
| + now := time.Now().Unix() |
| + later, sooner := now+7200, now+3600 |
| + postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"later"}`, tok, tid, later)) |
| + postJSON(t, ts.URL+"/api/schedule", fmt.Sprintf(`{"token":"%s","thread_id":"%s","deliver_at":%d,"blob":"sooner"}`, tok, tid, sooner)) |
| + if _, due := getDue("admintok"); due != sooner { |
| + t.Fatalf("due = %d, want the earlier %d", due, sooner) |
| + } |
| + |
| + // Delivery consumes the row, so the next due time advances. |
| + pending, _ := srv.db.listScheduledMessages(tid) |
| + if err := srv.db.deleteScheduledMessage(pending[0].ID); err != nil { // pending is deliver_at-ordered: [0] = sooner |
| + t.Fatalf("delete: %v", err) |
| + } |
| + if _, due := getDue("admintok"); due != later { |
| + t.Fatalf("after consuming the sooner schedule: due = %d, want %d", due, later) |
| + } |
| +} |
| + |
| // Cancellation is member-gated; cancelling an already-fired/unknown id is quiet. |
| func TestScheduleCancel(t *testing.T) { |
| ts, srv := newTestServer(t) |
diff --git a/registry.go b/registry.go| index 330717b..c2333b3 100644 |
| --- a/registry.go |
| +++ b/registry.go |
| @@ -34,7 +34,13 @@ type Route struct { |
| // them yet; register() just records what a caller sets. |
| HomeRegion string |
| Provider string |
| - CreatedAt int64 |
| + // NextDue is the due-time index (#550 Part 4 v2): the earliest pending |
| + // scheduled delivery (unix seconds; 0 = none) recorded by the activator when |
| + // the instance hibernates, so the sweeper can wake it in time to fire. In |
| + // the registry — not activator memory — so due times survive router restarts |
| + // (the registry is fleet-backed-up). Written via setNextDue only. |
| + NextDue int64 |
| + CreatedAt int64 |
| } |
| func openRegistry(path string) (*Registry, error) { |
| @@ -79,6 +85,7 @@ CREATE TABLE IF NOT EXISTS routes ( |
| push_relay_token TEXT NOT NULL DEFAULT '', |
| home_region TEXT NOT NULL DEFAULT '', |
| provider TEXT NOT NULL DEFAULT '', |
| + next_due INTEGER NOT NULL DEFAULT 0, |
| created_at INTEGER NOT NULL |
| );`); err != nil { |
| return err |
| @@ -90,6 +97,7 @@ CREATE TABLE IF NOT EXISTS routes ( |
| `ALTER TABLE routes ADD COLUMN prev_version TEXT NOT NULL DEFAULT ''`, |
| `ALTER TABLE routes ADD COLUMN home_region TEXT NOT NULL DEFAULT ''`, |
| `ALTER TABLE routes ADD COLUMN provider TEXT NOT NULL DEFAULT ''`, |
| + `ALTER TABLE routes ADD COLUMN next_due INTEGER NOT NULL DEFAULT 0`, |
| } { |
| if _, err := r.sql.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") { |
| return err |
| @@ -103,7 +111,7 @@ func (r *Registry) list() ([]Route, error) { |
| rows, err := r.sql.Query(` |
| SELECT host, kind, socket, status, last_error, db_path, version, prev_version, |
| app_store_url, home_origin, push_relay_url, push_relay_token, |
| - home_region, provider, created_at |
| + home_region, provider, next_due, created_at |
| FROM routes ORDER BY host`) |
| if err != nil { |
| return nil, err |
| @@ -114,7 +122,7 @@ FROM routes ORDER BY host`) |
| var t Route |
| if err := rows.Scan(&t.Host, &t.Kind, &t.Socket, &t.Status, &t.LastError, |
| &t.DBPath, &t.Version, &t.PrevVersion, &t.AppStoreURL, &t.HomeOrigin, |
| - &t.PushRelayURL, &t.PushRelayToken, &t.HomeRegion, &t.Provider, &t.CreatedAt); err != nil { |
| + &t.PushRelayURL, &t.PushRelayToken, &t.HomeRegion, &t.Provider, &t.NextDue, &t.CreatedAt); err != nil { |
| return nil, err |
| } |
| out = append(out, t) |
| @@ -128,10 +136,10 @@ func (r *Registry) get(host string) (*Route, error) { |
| err := r.sql.QueryRow(` |
| SELECT host, kind, socket, status, last_error, db_path, version, prev_version, |
| app_store_url, home_origin, push_relay_url, push_relay_token, |
| - home_region, provider, created_at |
| + home_region, provider, next_due, created_at |
| FROM routes WHERE host = ?`, host).Scan(&t.Host, &t.Kind, &t.Socket, &t.Status, |
| &t.LastError, &t.DBPath, &t.Version, &t.PrevVersion, &t.AppStoreURL, &t.HomeOrigin, |
| - &t.PushRelayURL, &t.PushRelayToken, &t.HomeRegion, &t.Provider, &t.CreatedAt) |
| + &t.PushRelayURL, &t.PushRelayToken, &t.HomeRegion, &t.Provider, &t.NextDue, &t.CreatedAt) |
| if err == sql.ErrNoRows { |
| return nil, nil |
| } |
| @@ -164,17 +172,51 @@ func (r *Registry) upsert(t Route) error { |
| _, err := r.sql.Exec(` |
| INSERT INTO routes (host, kind, socket, status, last_error, db_path, version, prev_version, |
| app_store_url, home_origin, push_relay_url, push_relay_token, |
| - home_region, provider, created_at) |
| -VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) |
| + home_region, provider, next_due, created_at) |
| +VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) |
| ON CONFLICT(host) DO UPDATE SET |
| kind=excluded.kind, socket=excluded.socket, status=excluded.status, |
| last_error=excluded.last_error, db_path=excluded.db_path, version=excluded.version, |
| prev_version=excluded.prev_version, |
| app_store_url=excluded.app_store_url, home_origin=excluded.home_origin, |
| push_relay_url=excluded.push_relay_url, push_relay_token=excluded.push_relay_token, |
| - home_region=excluded.home_region, provider=excluded.provider`, |
| + home_region=excluded.home_region, provider=excluded.provider, |
| + next_due=excluded.next_due`, |
| t.Host, t.Kind, t.Socket, t.Status, t.LastError, t.DBPath, t.Version, t.PrevVersion, |
| t.AppStoreURL, t.HomeOrigin, t.PushRelayURL, t.PushRelayToken, |
| - t.HomeRegion, t.Provider, t.CreatedAt) |
| + t.HomeRegion, t.Provider, t.NextDue, t.CreatedAt) |
| return err |
| } |
| + |
| +// ---- due-time index (#550 Part 4 v2 follow-up) -------------------------------- |
| +// |
| +// The activator's persistence for scheduled deliveries on hibernating hosts: |
| +// setNextDue records at sleep time, dueInstances answers the sweeper's "who |
| +// needs waking". Together they implement actDueStore (activator.go). |
| + |
| +// setNextDue records host's next scheduled-delivery time (unix seconds; 0 = |
| +// nothing scheduled) without touching the rest of the route. A host not in the |
| +// registry is a quiet no-op — it can't be woken anyway. |
| +func (r *Registry) setNextDue(host string, due int64) error { |
| + _, err := r.sql.Exec(`UPDATE routes SET next_due = ? WHERE host = ?`, due, host) |
| + return err |
| +} |
| + |
| +// dueInstances returns the ready instance routes whose recorded next_due has |
| +// arrived (0 < next_due <= now) — the sweeper's due-time wake set. Reading the |
| +// registry (not activator memory) each sweep is what makes the index survive |
| +// router restarts and track route removals for free; the fleet is tens of |
| +// routes, so a filtered list per tick is nothing. |
| +func (r *Registry) dueInstances(now int64) ([]Route, error) { |
| + routes, err := r.list() |
| + if err != nil { |
| + return nil, err |
| + } |
| + var due []Route |
| + for _, t := range routes { |
| + if t.Kind == "instance" && t.Status == "ready" && t.NextDue > 0 && t.NextDue <= now { |
| + due = append(due, t) |
| + } |
| + } |
| + return due, nil |
| +} |
diff --git a/router.go b/router.go| index bedb186..78e94ae 100644 |
| --- a/router.go |
| +++ b/router.go |
| @@ -212,7 +212,10 @@ func runRouter(args []string) { |
| if hc, err := hibernateConfigFromEnv(); err != nil { |
| log.Fatalf("router: %v", err) |
| } else if hc != nil { |
| - act = newActivator(*hc, &s3Backend{cfg: *hc}) |
| + // The registry doubles as the due-time index's store (routes.next_due), |
| + // so scheduled deliveries survive router restarts and fire on time even |
| + // for sleeping instances. |
| + act = newActivator(*hc, &s3Backend{cfg: *hc}, reg) |
| act.start() |
| log.Printf("router: hibernation active (bucket %s, prefix %q, idle %s) — instances wake on demand", |
| hc.Bucket, hc.Prefix, hc.Idle) |