fix(telemetry): count late index uploads, treat usage as live ingest, purge before catch-up (#2333) (#2356)

Three bugs in the self-hosted telemetry services (telemetry-worker/ and
telemetry-dashboard/), reported against 6560052a and all still on main.

1. An index run that uploads late never counted toward activation. The
   dashboard's funnel reads machine_first_seen.first_index_day, which only the
   nightly rollup wrote, and the rollup re-rolls just the last three days plus
   days with no daily_machines row. Ingest accepts timestamps up to 30 days old
   and the client keeps an event's original timestamp when it re-queues it, so
   an index event 4+ days late landed on a day nothing revisited and
   first_index_day stayed NULL. The ingest upsert of machine_first_seen now
   lowers first_index_day the way it lowers first_day:
   coalesce(min(old, new), old, new), because SQLite's min() is NULL if either
   side is. It is the same statement and row, with no new index and no
   migration (the column exists since 0002). The rollup still re-derives it
   from raw events, which covers events stored before this change.

2. ingest_stalled read only max(day) FROM events, but usage counters have
   lived in usage_daily since 0003, so a day with usage and no lifecycle events
   read as "The ingest worker ... is not storing anything". /api/meta now also
   reads max(day) FROM usage_daily (a primary-key lookup) and returns
   latest_usage_day and latest_ingest_day. Ingest counts as stalled only when
   there is no lifecycle event from yesterday or later and no usage counter
   from the day before yesterday or later, because clients upload a day's
   counters only after that day ends. The banner names latest_ingest_day.

3. runNightly rolled up the 3 regular days, then up to 31 missed days, and ran
   purgeOldEvents last. Each day first folds its legacy usage_rollup rows,
   which takes minutes on a heavy day, so a backlog could run past Cloudflare's
   15-minute Cron Trigger limit and skip the purge and the summary line night
   after night. The purge now runs first: it is bounded and touches only days
   past the window, which no rollup reads. After that, rollups stop starting
   new days or fold chunks 10 minutes in (NIGHTLY_BUDGET_MS). A day stopped
   mid-fold gets no rollup at all, so it stays a missed day and the next night
   continues from its last committed chunk. The summary line counts those days
   as `deferred`, and `caught_up` now counts missed days actually rolled up.

Verification: __tests__/telemetry-services.test.ts (new; the root vitest
config picks it up) runs the worker's fetch/cron code and the dashboard API
against the checked-in migrations in in-memory node:sqlite through a small
D1-shaped adapter. With the source fix reversed, all 5 tests fail (activation
0, first_index_day null, ingest_stalled true, and "nightly rollup incomplete:
4 day(s) failed, purge failed" once statements pass the 15-minute mark). With
the fix, all 5 pass. `npm run check` is clean in both packages. Against
wrangler dev, run locally only: smoke:rollup 71/71, smoke:cutover 62/62,
smoke:api 122/122 and render-check 87/87. smoke (ingest) is 46/47; the one
failure is the 70 KB oversized-body curl argument hitting the Windows
command-line limit, and main fails it the same way. The new smoke assertions
fail on main.

Takes effect only once both workers are redeployed; no migration.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

__tests__/telemetry-services.test.ts

@@ -0,0 +1,438 @@
+/**
+ * The self-hosted telemetry services — `telemetry-worker/` (ingest + nightly rollup)
+ * and `telemetry-dashboard/` (the read API) — run against the checked-in D1
+ * migrations in an in-memory node:sqlite database (#2333).
+ *
+ * D1 is SQLite, so every statement here is the SQL production runs. The one stand-in
+ * is a thin adapter giving node:sqlite D1's prepare/bind/first/all/run/batch shape,
+ * with `batch()` as one transaction the way D1 runs it. No wrangler, no workerd, no
+ * network: the end-to-end versions of these checks are the packages' own smoke
+ * suites (`npm run smoke`, `smoke:rollup`, `smoke:cutover`, `smoke:api`), which boot
+ * `wrangler dev`.
+ *
+ *   npx vitest run __tests__/telemetry-services.test.ts
+ */
+
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
+import * as fs from 'fs';
+import * as path from 'path';
+import worker from '../telemetry-worker/src/index';
+import { runNightly } from '../telemetry-worker/src/rollup';
+import { handleApi } from '../telemetry-dashboard/src/api';
+
+let nodeSqlite: typeof import('node:sqlite') | null = null;
+try {
+  nodeSqlite = require('node:sqlite') as typeof import('node:sqlite');
+} catch {
+  /* Node < 22.5 — skipped below */
+}
+
+const MIGRATIONS = path.join(__dirname, '..', 'telemetry-worker', 'migrations');
+const MINUTE = 60_000;
+/** Cloudflare ends a Cron Trigger invocation after 15 minutes of wall-clock time. */
+const CRON_LIMIT_MS = 15 * MINUTE;
+
+type Row = Record<string, unknown>;
+
+// ---------------------------------------------------------------------------
+// D1 over node:sqlite
+// ---------------------------------------------------------------------------
+
+/** Binds integral numbers as INTEGER, as D1 does (node:sqlite would bind them as REAL). */
+const bindable = (v: unknown): unknown => (typeof v === 'number' && Number.isInteger(v) ? BigInt(v) : v);
+
+class SqliteD1Statement {
+  constructor(
+    private readonly d1: SqliteD1,
+    readonly sql: string,
+    private readonly params: unknown[] = [],
+  ) {}
+
+  bind(...params: unknown[]): SqliteD1Statement {
+    return new SqliteD1Statement(this.d1, this.sql, params);
+  }
+
+  execute(): { success: true; results: Row[]; meta: { changes: number } } {
+    this.d1.onStatement?.(this.sql);
+    const stmt = this.d1.db.prepare(this.sql);
+    const params = this.params.map(bindable) as never[];
+    if (/^\s*(SELECT|WITH)\b/i.test(this.sql)) {
+      return { success: true, results: stmt.all(...params) as Row[], meta: { changes: 0 } };
+    }
+    const info = stmt.run(...params);
+    return { success: true, results: [], meta: { changes: Number(info.changes) } };
+  }
+
+  async first<T>(): Promise<T | null> {
+    return (this.execute().results[0] as T | undefined) ?? null;
+  }
+
+  async all(): Promise<ReturnType<SqliteD1Statement['execute']>> {
+    return this.execute();
+  }
+
+  async run(): Promise<ReturnType<SqliteD1Statement['execute']>> {
+    return this.execute();
+  }
+}
+
+class SqliteD1 {
+  /** Runs before every statement — the cron tests use it to model elapsed wall-clock time. */
+  onStatement: ((sql: string) => void) | null = null;
+
+  constructor(readonly db: import('node:sqlite').DatabaseSync) {}
+
+  prepare(sql: string): SqliteD1Statement {
+    return new SqliteD1Statement(this, sql);
+  }
+
+  async batch(statements: SqliteD1Statement[]): Promise<ReturnType<SqliteD1Statement['execute']>[]> {
+    this.db.exec('BEGIN');
+    try {
+      const results = statements.map((s) => s.execute());
+      this.db.exec('COMMIT');
+      return results;
+    } catch (err) {
+      this.db.exec('ROLLBACK');
+      throw err;
+    }
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Harness
+// ---------------------------------------------------------------------------
+
+describe.skipIf(!nodeSqlite)('telemetry services against the D1 schema (#2333)', () => {
+  let db: import('node:sqlite').DatabaseSync;
+  let d1: SqliteD1;
+  let env: never;
+  let logged: Row[];
+
+  const at = (iso: string): void => {
+    vi.setSystemTime(new Date(iso));
+  };
+
+  const one = (sql: string, ...params: unknown[]): Row | undefined =>
+    db.prepare(sql).get(...(params.map(bindable) as never[])) as Row | undefined;
+
+  /** POST /v1/events the way a client does, then wait for the off-response-path write. */
+  async function post(machineId: string, events: Row[]): Promise<number> {
+    const body = JSON.stringify({
+      machine_id: machineId,
+      codegraph_version: '1.6.2',
+      os: 'linux',
+      arch: 'x64',
+      node_major: 22,
+      ci: false,
+      schema_version: 2,
+      events,
+    });
+    const pending: Promise<unknown>[] = [];
+    const request = new Request('https://telemetry.test/v1/events', {
+      method: 'POST',
+      headers: { 'content-type': 'application/json', 'content-length': String(Buffer.byteLength(body)) },
+      body,
+    });
+    const ctx = { waitUntil: (p: Promise<unknown>) => void pending.push(p), passThroughOnException: () => {} };
+    const response = await worker.fetch(request as never, env, ctx as never);
+    await Promise.all(pending);
+    return response.status;
+  }
+
+  /** GET one dashboard endpoint, as the signed-in page does. */
+  async function api(pathAndQuery: string): Promise<Row> {
+    const result = await handleApi(env, new URL(`https://stats.test${pathAndQuery}`));
+    expect(result.status ?? 200).toBe(200);
+    return result.body as Row;
+  }
+
+  /**
+   * One cron invocation at `iso`. `chunkMs` is the wall-clock time each legacy-usage
+   * fold chunk is charged; a statement that starts after Cloudflare's 15-minute limit
+   * throws, the way the runtime ends the invocation there.
+   */
+  async function nightly(iso: string, chunkMs = 0): Promise<{ latestStartMs: number; summary: Row | undefined }> {
+    at(iso);
+    const started = Date.now();
+    let latestStartMs = 0;
+    d1.onStatement = (sql) => {
+      const elapsed = Date.now() - started;
+      if (elapsed > CRON_LIMIT_MS) {
+        throw new Error(`statement started ${elapsed / MINUTE} min into the cron run — past the 15-minute limit`);
+      }
+      latestStartMs = Math.max(latestStartMs, elapsed);
+      if (/INSERT INTO usage_daily/.test(sql) && /FROM events/.test(sql)) vi.setSystemTime(Date.now() + chunkMs);
+    };
+    const before = logged.length;
+    try {
+      await runNightly(env, Date.parse(iso));
+    } finally {
+      d1.onStatement = null;
+    }
+    const summary = logged.slice(before).find((line) => line.msg === 'nightly rollup');
+    return { latestStartMs, summary };
+  }
+
+  beforeEach(() => {
+    vi.useFakeTimers({ toFake: ['Date'] });
+    db = new nodeSqlite!.DatabaseSync(':memory:');
+    for (const file of fs.readdirSync(MIGRATIONS).filter((f) => f.endsWith('.sql')).sort()) {
+      db.exec(fs.readFileSync(path.join(MIGRATIONS, file), 'utf8'));
+    }
+    d1 = new SqliteD1(db);
+    env = {
+      DB: d1,
+      RETENTION_DAYS: 90,
+      MACHINE_RATE_LIMITER: { limit: async () => ({ success: true }) },
+      ADMIN_RATE_LIMITER: { limit: async () => ({ success: true }) },
+    } as never;
+    logged = [];
+    const capture = (line: unknown): void => {
+      try {
+        logged.push(JSON.parse(String(line)) as Row);
+      } catch {
+        /* not one of the worker's JSON lines */
+      }
+    };
+    vi.spyOn(console, 'log').mockImplementation(capture);
+    vi.spyOn(console, 'error').mockImplementation(capture);
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+    vi.useRealTimers();
+    db.close();
+  });
+
+  const M1 = '00000000-0000-4000-8000-000000000001';
+  const M2 = '00000000-0000-4000-8000-000000000002';
+  const M3 = '00000000-0000-4000-8000-000000000003';
+  const install = (ts: string): Row => ({ event: 'install', ts, props: { scope: 'local', kind: 'fresh' } });
+  const index = (ts: string): Row => ({ event: 'index', ts, props: { languages: ['typescript'] } });
+  const usage = (day: string, count: number): Row => ({
+    event: 'usage_rollup',
+    ts: `${day}T12:00:00.000Z`,
+    props: { kind: 'mcp_tool', name: 'codegraph_explore', count, client_name: 'Claude Code' },
+  });
+
+  // -------------------------------------------------------------------------
+  // 1. Activation counts an index run that uploads late
+  // -------------------------------------------------------------------------
+
+  describe('a late index upload', () => {
+    it('counts toward activation even when it arrives after the nightly run stopped revisiting its day', async () => {
+      // Sep 25: the machine installs. Its index run fails to upload and stays in the
+      // client's queue, keeping its original timestamp.
+      at('2026-09-25T15:00:00Z');
+      expect(await post(M1, [install('2026-09-25T14:00:00Z')])).toBe(204);
+      await nightly('2026-09-26T00:30:00Z');
+
+      // Oct 3: the queued index run finally goes out — eight days late, inside the
+      // 30 days ingest accepts, but long past the three days each nightly run re-rolls.
+      at('2026-10-03T10:00:00Z');
+      expect(await post(M1, [index('2026-09-25T14:05:00Z')])).toBe(204);
+      // Someone else's activity, so the rollup's coverage moves past the cohort's window.
+      expect(await post(M2, [install('2026-10-03T09:00:00Z')])).toBe(204);
+      await nightly('2026-10-04T00:30:00Z');
+
+      at('2026-10-04T08:00:00Z');
+      const funnel = await api('/api/activation?from=2026-09-25&to=2026-09-25&window=7');
+      expect(funnel).toMatchObject({ installs: 1, activated: 1, dropped: 0, covered_through: '2026-10-03' });
+      expect(one('SELECT first_day, first_index_day FROM machine_first_seen WHERE machine_id = ?', M1)).toEqual({
+        first_day: '2026-09-25',
+        first_index_day: '2026-09-25',
+      });
+    });
+
+    it('is applied as it is stored, and only ever moves the first index day earlier', async () => {
+      const firstSeen = (machineId: string): Row | undefined =>
+        one('SELECT first_day, first_index_day FROM machine_first_seen WHERE machine_id = ?', machineId);
+
+      at('2026-10-03T10:00:00Z');
+      // A batch without an index run says nothing about indexing.
+      expect(await post(M1, [install('2026-10-02T10:00:00Z')])).toBe(204);
+      expect(await post(M1, [usage('2026-10-01', 4)])).toBe(204);
+      expect(firstSeen(M1)).toEqual({ first_day: '2026-10-01', first_index_day: null });
+
+      // The earliest index run in a batch wins, with no rollup involved.
+      expect(await post(M1, [index('2026-10-03T09:00:00Z'), index('2026-10-02T11:00:00Z')])).toBe(204);
+      expect(firstSeen(M1)).toEqual({ first_day: '2026-10-01', first_index_day: '2026-10-02' });
+
+      // A later index run leaves it alone; a backdated one lowers it, and first_day with it.
+      expect(await post(M1, [index('2026-10-03T09:30:00Z')])).toBe(204);
+      expect(firstSeen(M1)).toEqual({ first_day: '2026-10-01', first_index_day: '2026-10-02' });
+      expect(await post(M1, [index('2026-09-20T08:00:00Z')])).toBe(204);
+      expect(firstSeen(M1)).toEqual({ first_day: '2026-09-20', first_index_day: '2026-09-20' });
+
+      // A machine whose very first batch carries an index run.
+      expect(await post(M2, [install('2026-10-03T08:00:00Z'), index('2026-10-03T08:01:00Z')])).toBe(204);
+      expect(firstSeen(M2)).toEqual({ first_day: '2026-10-03', first_index_day: '2026-10-03' });
+
+      // The nightly rollup agrees with what ingest already wrote.
+      await nightly('2026-10-04T00:30:00Z');
+      expect(firstSeen(M1)).toEqual({ first_day: '2026-09-20', first_index_day: '2026-09-20' });
+      expect(firstSeen(M2)).toEqual({ first_day: '2026-10-03', first_index_day: '2026-10-03' });
+    });
+  });
+
+  // -------------------------------------------------------------------------
+  // 2. Usage-only days are not a stalled ingest
+  // -------------------------------------------------------------------------
+
+  describe('the stalled-ingest flag', () => {
+    it('counts usage counters as ingest that is still storing', async () => {
+      // The last lifecycle event is Sep 25; after that only usage counters arrive.
+      at('2026-09-25T15:00:00Z');
+      expect(await post(M1, [install('2026-09-25T14:00:00Z')])).toBe(204);
+      // Clients upload a day's counters only once it is over: Oct 3's arrive on Oct 4.
+      at('2026-10-04T06:00:00Z');
+      expect(await post(M2, [usage('2026-10-03', 12)])).toBe(204);
+
+      at('2026-10-04T08:00:00Z');
+      const meta = await api('/api/meta');
+      expect(meta).toMatchObject({ latest_raw_day: '2026-09-25', latest_ingest_day: '2026-10-03', ingest_stalled: false });
+
+      // Just past midnight nobody has sent Oct 4's counters yet; that is not a stall.
+      at('2026-10-05T00:10:00Z');
+      expect(await api('/api/meta')).toMatchObject({ ingest_stalled: false });
+
+      // A whole day later with nothing stored at all, it is — and the banner date is
+      // the last day anything was stored, not the last lifecycle event.
+      at('2026-10-06T08:00:00Z');
+      expect(await api('/api/meta')).toMatchObject({ latest_ingest_day: '2026-10-03', ingest_stalled: true });
+    });
+
+    it('still reports a stall when nothing at all has arrived, and not on an empty database', async () => {
+      at('2026-10-04T08:00:00Z');
+      expect(await api('/api/meta')).toMatchObject({ latest_ingest_day: null, ingest_stalled: false });
+
+      at('2026-09-25T15:00:00Z');
+      expect(await post(M1, [install('2026-09-25T14:00:00Z')])).toBe(204);
+      at('2026-10-04T08:00:00Z');
+      expect(await api('/api/meta')).toMatchObject({ latest_ingest_day: '2026-09-25', ingest_stalled: true });
+
+      // A lifecycle event from yesterday is fresh, as before.
+      expect(await post(M1, [index('2026-10-03T22:00:00Z')])).toBe(204);
+      expect(await api('/api/meta')).toMatchObject({ latest_ingest_day: '2026-10-03', ingest_stalled: false });
+    });
+  });
+
+  // -------------------------------------------------------------------------
+  // 3. A long catch-up stays inside the cron's wall-clock limit, and the purge runs
+  // -------------------------------------------------------------------------
+
+  describe('the nightly run under a catch-up backlog', () => {
+    /** The rollup folds legacy usage rows 50,000 per transaction (rollup.ts, LEGACY_CHUNK_ROWS). */
+    const CHUNK_ROWS = 50_000;
+    /**
+     * Each fold chunk is charged 4 minutes, so a backlog of legacy-heavy days costs
+     * several minutes apiece, as it did in production. The six missed days below hold
+     * seven chunks: 28 minutes of folding, against a 15-minute limit.
+     */
+    const CHUNK_MS = 4 * MINUTE;
+    const NIGHT_1 = '2026-10-05T00:30:00Z';
+    /** Missed days, newest first — the order catch-up takes them in. Sep 21 needs two chunks. */
+    const MISSED = ['2026-09-23', '2026-09-22', '2026-09-21', '2026-09-20', '2026-09-19', '2026-09-18'];
+    const LEGACY_ROWS: Record<string, number> = Object.fromEntries(
+      MISSED.map((day) => [day, day === '2026-09-21' ? CHUNK_ROWS + 3 : 3]),
+    );
+    const PAST_WINDOW = '2026-06-01';
+
+    /** Legacy rows as they were stored before migrations/0003: one `events` row per upload, count 1. */
+    function seedLegacyDay(day: string, rows: number): void {
+      db.prepare(
+        `WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ?1)
+         INSERT INTO events (received_at, ts, day, event, machine_id, codegraph_version, os, arch,
+                             node_major, ci, schema_version, props)
+         SELECT ?2 || 'T13:00:00.000Z', ?2 || 'T12:00:00.000Z', ?2, 'usage_rollup',
+                '00000000-0000-4000-8000-00000000010' || ((i - 1) % 5), '1.5.0', 'linux', 'x64', 22, 0, 1,
+                '{"kind":"mcp_tool","name":"codegraph_explore","count":1,"error_count":0,"client_name":"Claude Code"}'
+           FROM n`,
+      ).run(BigInt(rows), day);
+      for (let m = 0; m < Math.min(rows, 5); m++) {
+        const machine = `00000000-0000-4000-8000-00000000010${m}`;
+        db.prepare('INSERT INTO machine_days (machine_id, day, prod) VALUES (?, ?, 1)').run(machine, day);
+        db.prepare(
+          `INSERT INTO machine_first_seen (machine_id, first_day) VALUES (?, ?)
+             ON CONFLICT (machine_id) DO UPDATE SET first_day = min(first_day, excluded.first_day)`,
+        ).run(machine, day);
+      }
+    }
+
+    /** Where each seeded usage count sits right now, and whether the day has been rolled up. */
+    function ledger(day: string) {
+      const folded = Number(one('SELECT coalesce(sum(count), 0) AS n FROM usage_daily WHERE day = ?', day)?.n);
+      const unfolded = Number(
+        one(
+          `SELECT coalesce(sum(json_extract(props, '$.count')), 0) AS n FROM events
+            WHERE day = ? AND event = 'usage_rollup'`,
+          day,
+        )?.n,
+      );
+      const rolledUsage = one(`SELECT count FROM daily_event_counts WHERE day = ? AND event = 'usage_rollup'`, day);
+      const rolledUp = one('SELECT machines FROM daily_machines WHERE day = ?', day) !== undefined;
+      return { folded, unfolded, rolledUsage: rolledUsage ? Number(rolledUsage.count) : null, rolledUp };
+    }
+
+    function expectConsistent(): void {
+      for (const day of MISSED) {
+        const l = ledger(day);
+        // Nothing lost and nothing counted twice, however far the fold got.
+        expect(l.folded + l.unfolded, day).toBe(LEGACY_ROWS[day]);
+        if (l.rolledUp) {
+          expect(l, day).toMatchObject({ unfolded: 0, rolledUsage: LEGACY_ROWS[day] });
+        } else {
+          // A day whose fold did not finish gets no rollup at all, so it stays "missed".
+          expect(l.rolledUsage, day).toBeNull();
+        }
+      }
+    }
+
+    beforeEach(() => {
+      for (const day of MISSED) seedLegacyDay(day, LEGACY_ROWS[day]!);
+      // Past the 90-day window: the purge must take these.
+      db.prepare(
+        `INSERT INTO events (received_at, ts, day, event, machine_id, props)
+         VALUES ('2026-06-01T10:00:00.000Z', NULL, ?1, 'install', ?2, '{}')`,
+      ).run(PAST_WINDOW, M3);
+      db.prepare(
+        `INSERT INTO usage_daily (day, machine_id, kind, name, count) VALUES (?, ?, 'cli_command', 'index', 2)`,
+      ).run(PAST_WINDOW, M3);
+    });
+
+    it('purges on schedule, stops before the limit, and resumes the backlog the next nights', async () => {
+      // The three days the run re-rolls every night have ordinary, current traffic.
+      at('2026-10-04T12:00:00Z');
+      expect(await post(M1, [install('2026-10-04T11:00:00Z'), usage('2026-10-03', 5)])).toBe(204);
+      expect(await post(M2, [index('2026-10-02T11:00:00Z')])).toBe(204);
+
+      const first = await nightly(NIGHT_1, CHUNK_MS);
+      expect(first.latestStartMs).toBeLessThanOrEqual(CRON_LIMIT_MS);
+      // The purge ran, and the summary line was written.
+      expect(one('SELECT count(*) AS n FROM events WHERE day < ?', '2026-07-07')?.n).toBe(0);
+      expect(one('SELECT count(*) AS n FROM usage_daily WHERE day < ?', '2026-07-07')?.n).toBe(0);
+      expect(first.summary).toMatchObject({ purged: 1, usage_purged: 1, failed: 0 });
+      // Tonight's regular days were rolled up first.
+      for (const day of ['2026-10-04', '2026-10-03', '2026-10-02']) {
+        expect(one('SELECT machines FROM daily_machines WHERE day = ?', day), day).toEqual({ machines: 1 });
+      }
+      // The backlog did not fit: Sep 21 was left part-folded and un-rolled, the rest untouched.
+      expect(ledger('2026-09-21')).toMatchObject({ folded: CHUNK_ROWS, unfolded: 3, rolledUp: false });
+      expect(MISSED.filter((day) => !ledger(day).rolledUp)).toEqual(MISSED.slice(2));
+      expect(first.summary).toMatchObject({ caught_up: 2, deferred: 4 });
+      expectConsistent();
+
+      // The next nights pick up exactly where this one stopped.
+      let nights = 1;
+      for (let day = 6; MISSED.some((d) => !ledger(d).rolledUp) && nights < 5; day++, nights++) {
+        const night = await nightly(`2026-10-0${day}T00:30:00Z`, CHUNK_MS);
+        expect(night.latestStartMs).toBeLessThanOrEqual(CRON_LIMIT_MS);
+        expect(night.summary).toMatchObject({ failed: 0 });
+        expectConsistent();
+      }
+      expect(nights).toBe(3);
+      for (const day of MISSED) expect(ledger(day), day).toMatchObject({ unfolded: 0, rolledUp: true });
+    });
+  });
+});

docs/design/telemetry.md

@@ -232,9 +232,9 @@ Full documentation is [`telemetry-dashboard/README.md`](../../telemetry-dashboar
   `telemetry-worker/scripts/smoke-cutover.sh` exists to cover — a mismatch there is silent,
   showing up as a panel that reads zero forever rather than as an error.
 - **Reads rollups, not raw events**, so a chart stays correct for days whose raw rows have
-  been purged. `/api/activation` is the one exception — "did this machine ever run an index"
-  is not a daily aggregate — so it reads raw `events` and is bounded by the retention window,
-  which it reports as `raw_events_from`.
+  been purged. The activation funnel's "did this machine ever run an index" is not a daily
+  aggregate, so it reads `machine_first_seen.first_index_day` instead, which the ingest Worker
+  lowers as each index event is stored and the nightly cron re-derives from raw `events`.
 - **Auth is a shared password and a signed cookie**, sized for exactly two people:
   `ADMIN_PASSWORD` + `SESSION_SECRET` as Worker secrets, constant-time compare, HMAC-signed
   cookie with no session store, everything except `/login` and `robots.txt` gated. Rotating

telemetry-dashboard/README.md

@@ -40,7 +40,7 @@ renders. Bad input is a `400` with a message, never a guess. Chart data carries
 
 | Endpoint | Answers |
 |---|---|
-| `/api/meta` | How current the data is: `today`, the rollup's last day (`latest_rollup_day`), the last day an event arrived, machines active yesterday, and two flags — `ingest_stalled` and `rollup_behind` — that put a warning above the panels when either writer stops. Cached 60 s. |
+| `/api/meta` | How current the data is: `today`, the rollup's last day (`latest_rollup_day`), the last day ingest stored anything (`latest_ingest_day` — a lifecycle event or a usage counter), machines active yesterday, and two flags — `ingest_stalled` and `rollup_behind` — that put a warning above the panels when either writer stops. Cached 60 s. |
 | `/api/summary` | Big numbers: production users, active machines, new machines, installs, uninstalls, indexing runs, tool calls. |
 | `/api/timeseries?metric=` | `installs_uninstalls`, `new_installs`, `production_users`, `indexing_activity`, `tool_calls`, `duration_buckets`. One dense point per day — a day with nothing is a zero, not a gap. Days after the rollup's last day are `null` ("not counted yet") and `covered_through` says where that is; `new_installs` is written live and runs through today. |
 | `/api/breakdown?dim=` | `os`, `arch`, `codegraph_version`, `node_major`, `language`, `file_count_bucket`, `duration_bucket`, `target`, `scope`, `kind`, `name`, `client_name`, `name_error`. Optional `&event=`, `&metric=count\|machines`, `&limit=`. |
@@ -53,7 +53,8 @@ kept forever, so a chart stays correct for days whose raw events have been purge
 panel reads raw `events`.** D1 runs one query at a time per database: the activation funnel
 used to join every cohort machine against `events`, which took ~55 s for one week of
 cohorts in production and failed every panel queued behind it. It now reads
-`machine_first_seen.first_index_day`, which the nightly rollup maintains.
+`machine_first_seen.first_index_day`, which the ingest worker lowers as each index event is
+stored and the nightly rollup re-derives from raw events.
 
 The presets end on **today** (UTC — every event and rollup is keyed on the UTC day). Live
 numbers — production users, new machines, retention — run through today; rolled-up ones
@@ -138,6 +139,10 @@ npm run smoke:render             # the panels, in a browser (79 assertions)
 Each suite starts its own throwaway `wrangler dev` on its own port and cleans up after
 itself, so they can be run in any order (`DASH_PORT` overrides the port).
 
+The repo's own test suite also runs `src/api.ts`, with the ingest worker feeding it, against
+the writer's migrations in an in-memory SQLite database — no wrangler needed. From the repo
+root (it is part of `npm test` there): `npx vitest run __tests__/telemetry-services.test.ts`.
+
 **`smoke-auth.sh`** is the regression net for the gate: unauthenticated requests reach
 nothing (pages, API *and* static assets), the cookie is persistent and correctly flagged,
 flipped/truncated/forged cookies are all rejected, brute force is capped, and rotating the

telemetry-dashboard/public/app.js

@@ -408,7 +408,8 @@ function drawDataWarning() {
 
   if (meta?.ingest_stalled) {
     lines.push([
-      `No new events since ${shortDay(meta.latest_raw_day)}.`,
+      // The last day anything was stored, lifecycle event or usage counter.
+      `No new events since ${shortDay(meta.latest_ingest_day ?? meta.latest_raw_day)}.`,
       ' The ingest worker at telemetry.getcodegraph.com is not storing anything; its logs and the D1 database are where to look.',
     ]);
   }

telemetry-dashboard/scripts/fixture.sql

@@ -36,6 +36,7 @@ DELETE FROM daily_machines;
 DELETE FROM machine_days;
 DELETE FROM machine_first_seen;
 DELETE FROM events;
+DELETE FROM usage_daily;
 
 -- ---------------------------------------------------------------------------
 -- install — 12, one per machine on its first day

telemetry-dashboard/scripts/smoke-api.sh

@@ -105,6 +105,7 @@ field "retention window"     retention_days   14         "$META"
 # The fixture stops on 07-10, months ago: nothing has arrived since, but the
 # rollup did cover every day that saw activity, so it is not behind.
 field "ingest stalled (no events since 07-10)" ingest_stalled true "$META"
+field "…named by the last day anything was stored" latest_ingest_day 2026-07-10 "$META"
 field "rollup not behind (it covered every active day)" rollup_behind false "$META"
 field "nobody active yesterday" machines_yesterday 0 "$META"
 
@@ -307,6 +308,19 @@ field "…and yesterday's machine is counted"     machines_yesterday 1 "$META"
 npx wrangler d1 execute codegraph-telemetry --local \
   --command "DELETE FROM machine_days WHERE machine_id = '$STALE_ID'" >/dev/null 2>&1
 
+echo
+echo "Usage counters alone are not a stalled ingest"
+# Usage lands in usage_daily, not events, so a day of tool calls with no install or
+# index run must not read as "not storing anything". Removed again straight after.
+npx wrangler d1 execute codegraph-telemetry --local \
+  --command "INSERT INTO usage_daily (day, machine_id, kind, name, count) VALUES ('$YESTERDAY', '$STALE_ID', 'mcp_tool', 'codegraph_explore', 3)" >/dev/null 2>&1
+META="$(get "/api/meta")"
+field "not stalled while usage arrives"           ingest_stalled    false        "$META"
+field "…the usage day is the last day stored"     latest_ingest_day "$YESTERDAY" "$META"
+field "…though the last lifecycle event is 07-10" latest_raw_day    2026-07-10   "$META"
+npx wrangler d1 execute codegraph-telemetry --local \
+  --command "DELETE FROM usage_daily WHERE machine_id = '$STALE_ID'" >/dev/null 2>&1
+
 echo
 printf '%d passed, %d failed\n' "$PASS" "$FAIL"
 [[ "$FAIL" -eq 0 ]]

telemetry-dashboard/src/api.ts

@@ -6,8 +6,8 @@
  * - **Rollups only.** Every panel is answered from `daily_*`, `machine_days` and
  *   `machine_first_seen`, which are kept forever. No panel reads raw `events`:
  *   D1 runs one query at a time per database, so one slow scan there fails every
- *   panel queued behind it. (/api/meta reads the table's first and last day, one
- *   indexed lookup each.)
+ *   panel queued behind it. (/api/meta reads the table's first and last day, and
+ *   `usage_daily`'s last, one indexed lookup each.)
  * - **Today is in range; uncounted days are not zeros.** Rolled-up numbers stop
  *   at the nightly rollup's last day and come back null after it, so a chart
  *   ending today draws a gap where the count has not happened yet, not a cliff.
@@ -214,6 +214,7 @@ interface MetaRow {
   earliest_active_day: string | null;
   earliest_raw_day: string | null;
   latest_raw_day: string | null;
+  latest_usage_day: string | null;
   machines_yesterday: number | null;
 }
 
@@ -246,6 +247,7 @@ async function meta(env: Env): Promise<ApiResult> {
             (SELECT min(day) FROM machine_days)       AS earliest_active_day,
             (SELECT min(day) FROM events)             AS earliest_raw_day,
             (SELECT max(day) FROM events)             AS latest_raw_day,
+            (SELECT max(day) FROM usage_daily)        AS latest_usage_day,
             (SELECT count(*) FROM machine_days WHERE day = ?) AS machines_yesterday`,
   )
     .bind(yesterday)
@@ -253,11 +255,22 @@ async function meta(env: Env): Promise<ApiResult> {
 
   const latestRollup = row?.latest_rollup_day ?? null;
   const latestRaw = row?.latest_raw_day ?? null;
+  const latestUsage = row?.latest_usage_day ?? null;
   const latestActive = row?.latest_active_day ?? null;
+  // Ingest stores lifecycle events in `events` and usage counters in `usage_daily`, so
+  // either one arriving means it is working — a day of usage and no installs or index
+  // runs is quiet, not stalled.
+  const latestIngest =
+    latestUsage !== null && (latestRaw === null || latestUsage > latestRaw) ? latestUsage : latestRaw;
 
-  // Nothing at all since before yesterday. (Client clocks may run a few minutes
-  // ahead, so the latest day can be tomorrow — that is fresh, not stale.)
-  const ingestStalled = latestRaw !== null && latestRaw < yesterday;
+  // Stalled: nothing stored since before yesterday. (Client clocks may run a few
+  // minutes ahead, so the latest day can be tomorrow — that is fresh, not stale.) Usage
+  // counters get a day more, because a client uploads a day's counters only once that
+  // day is over: just after midnight UTC the newest one can be the day before yesterday.
+  const ingestStalled =
+    latestIngest !== null &&
+    (latestRaw === null || latestRaw < yesterday) &&
+    (latestUsage === null || latestUsage < addDays(today, -2));
   // The 00:30 UTC run rolls up yesterday, so the day before that must always be in
   // by now. Only "behind" if there was activity after the last rolled-up day — a day
   // nobody used codegraph would be a silent rollup, not a missed one.
@@ -274,7 +287,12 @@ async function meta(env: Env): Promise<ApiResult> {
       latest_rollup_day: latestRollup,
       latest_active_day: latestActive,
       earliest_raw_day: row?.earliest_raw_day ?? null,
+      /** The last day of a stored lifecycle event (install, index, uninstall). */
       latest_raw_day: latestRaw,
+      /** The last day of a stored usage counter. */
+      latest_usage_day: latestUsage,
+      /** The later of the two: the last day ingest stored anything, which the stall banner names. */
+      latest_ingest_day: latestIngest,
       machines_yesterday: row?.machines_yesterday ?? 0,
       rollup_behind: rollupBehind,
       ingest_stalled: ingestStalled,
@@ -670,17 +688,18 @@ interface ActivationRow {
  * reinstalls does not re-enter the funnel, which is what makes this a
  * conversion rate rather than an install-event ratio.
  *
- * "Ran an index" is `machine_first_seen.first_index_day`, which the nightly rollup
- * keeps at the earliest day each machine indexed. That makes this a range read over
- * one small table. It used to be a join against raw `events` — on production volume
- * ~55 s per week of cohorts, which held D1's single query lane long enough to fail
- * every other panel waiting behind it. A first index day is never before the first
- * day (the ingest path keeps first_day at the machine's earliest event), so "within
- * the window" is just `first_index_day <= first_day + window`.
+ * "Ran an index" is `machine_first_seen.first_index_day`, the earliest day each
+ * machine indexed. The ingest worker lowers it as each index event is stored — so a
+ * run that uploads days late still counts — and the nightly rollup re-derives it from
+ * raw events. That makes this a range read over one small table. It used to be a
+ * join against raw `events` — on production volume ~55 s per week of cohorts, which
+ * held D1's single query lane long enough to fail every other panel waiting behind
+ * it. A first index day is never before the first day (the ingest path keeps
+ * first_day at the machine's earliest event), so "within the window" is just
+ * `first_index_day <= first_day + window`.
  *
- * Because first_index_day is rolled up, cohorts after the rollup's last day have no
- * conversions counted yet. They are left out of the totals and drawn as gaps —
- * counting them would show a drop in conversion that is only a lag.
+ * Cohorts after the rollup's last day are left out of the totals and drawn as gaps,
+ * like every other rolled-up number on the page.
  */
 async function activation(env: Env, url: URL, range: Range): Promise<ApiResult> {
   const rawWindow = url.searchParams.get('window');

telemetry-worker/README.md

@@ -80,7 +80,8 @@ nothing looks broken from the outside. That ran from 2026-08-11 to October 2026.
 
 ## Rollups & retention (nightly cron)
 
-`src/rollup.ts` runs on a Cron Trigger at **00:30 UTC** and does two things.
+`src/rollup.ts` runs on a Cron Trigger at **00:30 UTC** and does two things, the purge
+first.
 
 **Rolls up** the day that just ended into `daily_machines`, `daily_event_counts`,
 `daily_dim_counts` and `machine_first_seen.first_index_day`, then re-runs the two days before
@@ -88,7 +89,12 @@ it — offline clients ship completed-day rollups late, so a day keeps growing a
 It then **catches up** on any earlier day that saw activity but never got a rollup (a
 `machine_days` day with no `daily_machines` row — a night the run failed or the database
 refused writes), newest first, up to 31 a night. An outage heals on the first good night
-instead of leaving a hole someone has to notice. The aggregation is one
+instead of leaving a hole someone has to notice. Cloudflare ends a Cron Trigger after 15
+minutes of wall-clock time, and a day that still holds millions of legacy usage rows (below)
+takes minutes, so the run starts no new day and no new fold chunk once 10 minutes have
+passed (`NIGHTLY_BUDGET_MS`). A day it stops partway gets no rollup at all, so it is still a
+missed day the next night, which carries on from the last folded chunk; the summary line
+counts those days as `deferred`. The aggregation is one
 `INSERT … SELECT … ON CONFLICT DO UPDATE` per table or dimension, so it happens inside D1 and
 no event row crosses the wire. Every write overwrites the recomputed value rather than adding
 to it: **re-running a day is a no-op, never a double count.** Two things the SQL is careful
@@ -100,7 +106,9 @@ rollup over old days is the whole migration. Adding a breakdown is a line in `RO
 never a migration — that is what the generic `(dim, value)` shape buys.
 
 **Purges** raw `events` and `usage_daily` rows older than `RETENTION_DAYS` (90, a var in
-`wrangler.jsonc`) in bounded `DELETE` batches, and logs one line of counts. `machine_days` and `machine_first_seen` are
+`wrangler.jsonc`) in bounded `DELETE` batches, before any rollup: behind a long catch-up it
+would be the part the 15-minute limit cuts off, night after night, while the database
+grows. The run ends with one line of counts. `machine_days` and `machine_first_seen` are
 never purged — retention cohorts need the full history and they are two orders of magnitude
 smaller. Rollups are kept forever, so shortening the window costs ad-hoc drill-back, never a
 chart.
@@ -122,10 +130,13 @@ each day is a full scan of that day's events, and the request has a wall-clock b
 
 ### Backfilling first_index_day
 
-`first_index_day` (migration `0002`) is what the dashboard's activation funnel reads, and only
-the rollup writes it. Days rolled up before the migration left it NULL, and days stored before
+`first_index_day` (migration `0002`) is what the dashboard's activation funnel reads. The
+ingest worker lowers it as each index event is stored, so an index run that uploads days late
+still counts, and the rollup re-derives it from the raw events of every day it rolls up. Days
+rolled up before the migration left it NULL, an index event stored before the ingest worker
+began setting it (#2333) counts only once its day is rolled up again, and days stored before
 `0003` still hold their usage as one `events` row per upload. Re-running the rollup over every
-day that still has raw events fixes both — it folds that day's legacy usage rows into
+day that still has raw events fixes all three — it folds that day's legacy usage rows into
 `usage_daily`, sets `first_index_day`, and recomputes the day's rollups. It is idempotent, so
 overlapping or repeating a range is harmless. A day with millions of legacy rows takes several
 minutes, so go a day at a time:
@@ -300,6 +311,15 @@ To drive the cron body by hand, run `wrangler dev --test-scheduled` and hit
 `.dev.vars.example` to `.dev.vars` — without an `ADMIN_TOKEN` the route 404s, exactly as a
 deploy that never set the secret does.
 
+The repo's own test suite also covers this worker and the dashboard API without wrangler:
+it runs their source against these migrations in an in-memory SQLite database, including
+late uploads and a nightly run whose catch-up outlasts the cron's time limit. From the repo
+root (it is part of `npm test` there):
+
+```bash
+npx vitest run __tests__/telemetry-services.test.ts
+```
+
 ## Changing the schema
 
 The allowlist in `src/index.ts` mirrors `docs/design/telemetry.md` (and the user-facing

telemetry-worker/scripts/smoke-cutover.sh

@@ -235,11 +235,14 @@ ts "tool calls (sums the prop)" tool_calls       '[20]' '[2]'
 MET=$(api "meta")
 is "meta reports the rolled-up day"     "$DAY" "$(jget "$MET" latest_day)"
 is "meta reports the rollup ran"       "$DAY" "$(jget "$MET" latest_rollup_day)"
+# Usage lands in usage_daily rather than events; the stalled-ingest check reads both.
+is "meta reports the last day ingest stored" "$DAY" "$(jget "$MET" latest_ingest_day)"
 
-# The funnel reads machine_first_seen.first_index_day, which only the nightly
-# rollup writes (from raw `index` events). This is the seam that pins it: ingest
-# writes the events, the rollup sets the column, the dashboard reads it. If the
-# rollup stopped setting it, "activated" here would read 0.
+# The funnel reads machine_first_seen.first_index_day. The ingest worker sets it as
+# each index event is stored, and the nightly rollup re-derives it from raw `index`
+# events (smoke-rollup.sh pins that half, on events it seeds past the ingest path).
+# This is the seam: ingest writes the column, the dashboard reads it. If neither
+# writer set it, "activated" here would read 0.
 #
 # Its denominator is FIRST-SEEN MACHINES, not `install` events (api.ts: "a machine
 # that reinstalls does not re-enter the funnel"). m3 is the discriminator: it never

telemetry-worker/scripts/smoke-ingest.sh

@@ -214,6 +214,14 @@ is "machine_days: each backdated batch gets its own day" "$BACK_DAY,$RECENT_DAY"
 is "machine_first_seen recorded" "$RECENT_DAY" "$(q "select first_day from machine_first_seen where machine_id='$M_OK'")"
 is "machine_first_seen only moves earlier" "$BACK_DAY" \
    "$(q "select first_day from machine_first_seen where machine_id='$M_BACK'")"
+# The activation funnel's input is kept here too, not only by the nightly rollup: an
+# index run that uploads after the cron stopped re-rolling its day must still count.
+is "first_index_day recorded as the index run is stored" "$RECENT_DAY" \
+   "$(q "select first_index_day from machine_first_seen where machine_id='$M_OK'")"
+is "first_index_day only moves earlier, like first_day" "$BACK_DAY" \
+   "$(q "select first_index_day from machine_first_seen where machine_id='$M_BACK'")"
+is "a machine that never sent an index run has none" "null" \
+   "$(q "select coalesce(first_index_day, 'null') from machine_first_seen where machine_id='$M_USE'")"
 
 echo
 echo "usage counters add up instead of piling up"

telemetry-worker/scripts/smoke-rollup.sh

@@ -16,7 +16,8 @@
 #   * the purge deletes only rows past the window, and leaves machine_days /
 #     machine_first_seen alone
 #   * each machine's first_index_day is its earliest index, and survives the purge
-#   * the cron catches up on a day that saw activity but was never rolled up
+#   * the cron catches up on a day that saw activity but was never rolled up, and
+#     logs its one summary line
 #   * usage comes from usage_daily; a legacy usage row still in `events` is folded
 #     into it (counts added, row deleted) before the day is rolled up
 #   * /admin/rollup does not exist without ADMIN_TOKEN, and rejects a wrong one
@@ -196,6 +197,12 @@ is "rollup $DAY_RESET with reset → 200" 200 "$(roll "day=$DAY_RESET&reset=1")"
 # rolled up), and purges everything past the window.
 is "cron trigger → 200" 200 "$(curl -s -o /dev/null -w '%{http_code}' "$BASE/__scheduled?cron=30+0+*+*+*")"
 sleep 2
+# Its one summary line: written once the purge and every rollup are done, and with
+# nothing this small left for the next night by the time budget.
+is "the cron logs its summary, nothing failed or deferred" "0/0" \
+   "$(grep -ao '{"msg":"nightly rollup".*}' "$LOG" | tail -1 |
+      node -e 'let s="";process.stdin.on("data",d=>s+=d).on("end",()=>{
+        const r=JSON.parse(s||"{}");console.log(`${r.failed}/${r.deferred}`);})')"
 
 # Rolling a purged day with reset=1 must NOT blank the rollups it already has: past
 # the window the reset is ignored, so the delete-then-rebuild can't find zero events.

telemetry-worker/src/index.ts

@@ -200,9 +200,18 @@ const INSERT_EVENT = `INSERT INTO events (
 const UPSERT_MACHINE_DAY = `INSERT INTO machine_days (machine_id, day, prod) VALUES (?, ?, ?)
   ON CONFLICT (machine_id, day) DO UPDATE SET prod = max(machine_days.prod, excluded.prod)`;
 
-// A late-arriving offline buffer can move a machine's first day earlier, never later.
-const UPSERT_FIRST_SEEN = `INSERT INTO machine_first_seen (machine_id, first_day) VALUES (?, ?)
-  ON CONFLICT (machine_id) DO UPDATE SET first_day = min(machine_first_seen.first_day, excluded.first_day)`;
+// A late-arriving offline buffer can move a machine's first day earlier, never later —
+// and its first index day (the activation funnel's input) the same way. That one is set
+// here as well as by the nightly rollup because the rollup only re-rolls the last three
+// days, while this path accepts events up to 30 days old: an index run that uploads a
+// week late would otherwise never count. SQLite's multi-argument min() is NULL when
+// either side is, hence the coalesce: a batch with no index run leaves the stored day
+// alone, and the first index run on record sets it.
+const UPSERT_FIRST_SEEN = `INSERT INTO machine_first_seen (machine_id, first_day, first_index_day) VALUES (?, ?, ?)
+  ON CONFLICT (machine_id) DO UPDATE SET
+    first_day = min(machine_first_seen.first_day, excluded.first_day),
+    first_index_day = coalesce(min(machine_first_seen.first_index_day, excluded.first_index_day),
+                               machine_first_seen.first_index_day, excluded.first_index_day)`;
 
 // usage_rollup counters ADD into one row per machine × day × tool (migrations/0003):
 // clients upload the same counter many times over — once per process — so storing
@@ -305,10 +314,13 @@ async function writeToD1(
       }
     }
 
+    // The earliest day this batch indexed on, if it did — see UPSERT_FIRST_SEEN.
+    let firstIndexDay: string | null = null;
     for (const e of batch) {
       if (e.event === 'usage_rollup') continue;
       const day = (e.ts ?? receivedAt).slice(0, 10);
       days.add(day);
+      if (e.event === 'index' && (firstIndexDay === null || day < firstIndexDay)) firstIndexDay = day;
       stmts.push(
         insertEvent.bind(
           receivedAt,
@@ -328,7 +340,7 @@ async function writeToD1(
 
     const firstDay = [...days].sort()[0];
     if (firstDay !== undefined) {
-      stmts.push(env.DB.prepare(UPSERT_FIRST_SEEN).bind(machineId, firstDay));
+      stmts.push(env.DB.prepare(UPSERT_FIRST_SEEN).bind(machineId, firstDay, firstIndexDay));
     }
 
     await env.DB.batch(stmts);

telemetry-worker/src/rollup.ts

@@ -9,12 +9,13 @@
  * 1. ROLL UP the just-completed UTC day into `daily_machines`, `daily_event_counts`,
  *    `daily_dim_counts` and `machine_first_seen.first_index_day` — plus the two days
  *    before it, because clients buffer offline and ship completed-day rollups late, so
- *    a day keeps growing after it ends, plus any earlier day a failed run missed. Every
- *    write is an upsert that OVERWRITES the recomputed value rather than adding to it
- *    (or, for first_index_day, only ever lowers it), so re-running a day is a no-op
- *    and never double-counts.
+ *    a day keeps growing after it ends, plus any earlier day a failed run missed, as
+ *    many as fit in the run's time budget. Every write is an upsert that OVERWRITES
+ *    the recomputed value rather than adding to it (or, for first_index_day, only ever
+ *    lowers it), so re-running a day is a no-op and never double-counts.
  *
- * 2. PURGE raw `events` past the retention window, in bounded batches. Rollups are
+ * 2. PURGE raw `events` past the retention window, in bounded batches — first, so a
+ *    long catch-up can never push it past the cron's wall-clock limit. Rollups are
  *    kept forever, so only ad-hoc drill-down has a horizon; `machine_days` and
  *    `machine_first_seen` are never purged, because retention cohorts need the full
  *    history and they are two orders of magnitude smaller than the raw rows.
@@ -32,6 +33,15 @@ export const ROLLUP_LOOKBACK_DAYS = 3;
 export const MAX_MANUAL_DAYS = 31;
 /** Most missed days one nightly run catches up on, newest first; the next night takes the rest. */
 export const MAX_CATCHUP_DAYS = 31;
+/**
+ * How long into a nightly run it still starts rollup work. Cloudflare ends a Cron
+ * Trigger after 15 minutes of wall-clock time, and catching up a day that still holds
+ * millions of legacy usage rows takes minutes, so a backlog can run straight into that
+ * limit. Past this point the run starts no new day and no new fold chunk; the 5
+ * minutes left cover the one statement already in flight and the summary line. What
+ * did not fit stays a missed day, so the next night carries on from there.
+ */
+export const NIGHTLY_BUDGET_MS = 10 * 60_000;
 
 /** Rows per purge DELETE — bounded so one statement stays well inside D1's limits. */
 const PURGE_BATCH_ROWS = 5_000;
@@ -148,6 +158,10 @@ const DAILY_MACHINES = `INSERT INTO daily_machines (day, machines, prod_machines
  * it straight off `machine_first_seen`, which keeps the funnel off raw `events` (a
  * cohort join there took most of a minute per week of cohorts) and past the purge.
  *
+ * The ingest path lowers it the same way as each index event is stored (src/index.ts),
+ * because a run that uploads late lands on a day this cron has stopped re-rolling. This
+ * statement is what fills it for events stored before that, and what a backfill re-runs.
+ *
  * Only rows that actually move are written, so re-running a day is free. `?1` is the
  * day — still the one bound parameter, used three times.
  */
@@ -221,6 +235,11 @@ export interface DayResult {
   rows: number;
   /** Day is past the retention window — a `reset` on it is ignored (see below). */
   pastRetention: boolean;
+  /**
+   * The deadline passed before the day's legacy usage rows were all folded, so nothing
+   * was rolled up: the next run finishes the fold and rolls the day up then.
+   */
+  deferred: boolean;
 }
 
 /**
@@ -236,14 +255,20 @@ export interface DayResult {
  * exists would otherwise linger. It is IGNORED past the retention window, where it
  * would delete rows and then find no events to rebuild them from: silently blanking a
  * real day is the one irreversible thing this file could do.
+ *
+ * `deadline` (epoch ms) bounds the legacy fold that runs first. If it stops the fold
+ * partway, the day is not rolled up at all — usage counted from a half-folded day would
+ * be short, and its `daily_machines` row would stop the nightly run from ever coming
+ * back to it. The chunks already folded are committed, so the next call carries on.
  */
 export async function rollupDay(
   env: Env,
   day: string,
-  opts: { cutoff: string; reset?: boolean },
+  opts: { cutoff: string; reset?: boolean; deadline?: number },
 ): Promise<DayResult> {
   const pastRetention = day < opts.cutoff;
-  await foldLegacyUsage(env, day);
+  const fold = await foldLegacyUsage(env, day, opts.deadline);
+  if (!fold.complete) return { day, rows: 0, pastRetention, deferred: true };
   const statements: D1PreparedStatement[] = [];
 
   if (opts.reset && !pastRetention) {
@@ -257,7 +282,7 @@ export async function rollupDay(
 
   const results = await env.DB.batch(statements);
   const rows = results.reduce((total, r) => total + (r.meta?.changes ?? 0), 0);
-  return { day, rows, pastRetention };
+  return { day, rows, pastRetention, deferred: false };
 }
 
 // ---------------------------------------------------------------------------
@@ -298,15 +323,32 @@ const LEGACY_FOLD = `INSERT INTO usage_daily (
 
 const LEGACY_DELETE = `DELETE FROM events WHERE day = ?1 AND event = 'usage_rollup' AND id < ?2`;
 
-/** Folds any legacy usage rows for `day` into usage_daily. Returns how many rows it moved. */
-export async function foldLegacyUsage(env: Env, day: string): Promise<number> {
+export interface FoldResult {
+  /** Legacy rows moved into usage_daily by this call. */
+  moved: number;
+  /** False when the deadline stopped it with rows still to fold; the next call carries on. */
+  complete: boolean;
+}
+
+/**
+ * Folds any legacy usage rows for `day` into usage_daily, a chunk per transaction.
+ * With a `deadline` (epoch ms) it starts no chunk once that has passed: each chunk
+ * commits on its own and always takes the oldest rows left, so stopping between two
+ * loses nothing and the next call simply continues.
+ */
+export async function foldLegacyUsage(
+  env: Env,
+  day: string,
+  deadline = Number.POSITIVE_INFINITY,
+): Promise<FoldResult> {
   const any = await env.DB.prepare(`SELECT 1 AS hit FROM events WHERE day = ? AND event = 'usage_rollup' LIMIT 1`)
     .bind(day)
     .first<{ hit: number }>();
-  if (!any) return 0;
+  if (!any) return { moved: 0, complete: true };
 
   let moved = 0;
   for (let chunk = 0; chunk < LEGACY_MAX_CHUNKS; chunk++) {
+    if (Date.now() >= deadline) return { moved, complete: false };
     // Always the oldest remaining rows: everything below the next chunk's first id.
     // Rows folded by earlier chunks are gone, so there is no cursor to carry.
     const next = await env.DB.prepare(
@@ -320,7 +362,7 @@ export async function foldLegacyUsage(env: Env, day: string): Promise<number> {
       env.DB.prepare(LEGACY_DELETE).bind(day, below),
     ]);
     moved += deleted?.meta?.changes ?? 0;
-    if (!next) return moved;
+    if (!next) return { moved, complete: true };
   }
   throw new Error(`legacy usage fold for ${day} did not finish within ${LEGACY_MAX_CHUNKS} chunks`);
 }
@@ -399,44 +441,30 @@ async function missedDays(env: Env, cutoff: string, before: string): Promise<str
 }
 
 /**
- * The cron body: roll up the completed day and the two before it, catch up any
- * earlier day a past run missed, then purge.
+ * The cron body: purge, roll up the completed day and the two before it, then catch
+ * up any earlier day a past run missed, for as long as the time budget lasts.
+ *
+ * The purge goes first. It is bounded, it deletes only days past the window (which
+ * nothing below reads), and it is the job that must not slip: a database that stops
+ * purging grows until D1's size cap refuses every write. Run last, it was what
+ * Cloudflare's 15-minute limit cut off whenever a catch-up ran long, night after night.
+ *
+ * Rollups then start until NIGHTLY_BUDGET_MS has passed. A day the budget stops —
+ * before it starts, or partway through its legacy fold — gets no rollup at all, so it
+ * is still a missed day tomorrow and picks up where it stopped (each fold chunk is
+ * already committed).
  *
  * Logs one line of counts — never a day's contents, never a machine id. Throws if
  * anything failed so the invocation is marked failed (and retried) rather than
- * quietly skipping a day; every write here is idempotent, so a retry is safe.
+ * quietly skipping a day; every write here is idempotent, so a retry is safe. A day
+ * left for tomorrow is not a failure.
  */
 export async function runNightly(env: Env, atMs: number): Promise<void> {
   const started = Date.now();
+  const deadline = started + NIGHTLY_BUDGET_MS;
   const keepDays = retentionDays(env);
   const cutoff = retentionCutoff(atMs, keepDays);
 
-  const days: string[] = [];
-  for (let back = 1; back <= ROLLUP_LOOKBACK_DAYS; back++) days.push(utcDay(atMs - back * DAY_MS));
-
-  // A failure to find the missed days must not cost tonight's regular rollup.
-  let caughtUp = 0;
-  try {
-    const missed = await missedDays(env, cutoff, days[days.length - 1] ?? utcDay(atMs));
-    days.push(...missed);
-    caughtUp = missed.length;
-  } catch (err) {
-    console.error(JSON.stringify({ msg: 'missed-day scan failed', err: String(err) }));
-  }
-
-  const rolled: string[] = [];
-  const failed: string[] = [];
-  let rows = 0;
-  for (const day of days) {
-    try {
-      rows += (await rollupDay(env, day, { cutoff })).rows;
-      rolled.push(day);
-    } catch (err) {
-      failed.push(day);
-      console.error(JSON.stringify({ msg: 'rollup day failed', day, err: String(err) }));
-    }
-  }
-
   let purge: PurgeResult | null = null;
   try {
     purge = await purgeOldEvents(env, cutoff);
@@ -444,11 +472,50 @@ export async function runNightly(env: Env, atMs: number): Promise<void> {
     console.error(JSON.stringify({ msg: 'purge failed', cutoff, err: String(err) }));
   }
 
+  const regular: string[] = [];
+  for (let back = 1; back <= ROLLUP_LOOKBACK_DAYS; back++) regular.push(utcDay(atMs - back * DAY_MS));
+
+  // A failure to find the missed days must not cost tonight's regular rollup.
+  let missed: string[] = [];
+  try {
+    missed = await missedDays(env, cutoff, regular[regular.length - 1] ?? utcDay(atMs));
+  } catch (err) {
+    console.error(JSON.stringify({ msg: 'missed-day scan failed', err: String(err) }));
+  }
+
+  const rolled: string[] = [];
+  const failed: string[] = [];
+  const deferred: string[] = [];
+  let caughtUp = 0;
+  let rows = 0;
+  for (const day of [...regular, ...missed]) {
+    if (Date.now() >= deadline) {
+      deferred.push(day);
+      continue;
+    }
+    try {
+      const result = await rollupDay(env, day, { cutoff, deadline });
+      if (result.deferred) {
+        deferred.push(day);
+        continue;
+      }
+      rows += result.rows;
+      rolled.push(day);
+      if (!regular.includes(day)) caughtUp++;
+    } catch (err) {
+      failed.push(day);
+      console.error(JSON.stringify({ msg: 'rollup day failed', day, err: String(err) }));
+    }
+  }
+
   console.log(
     JSON.stringify({
       msg: 'nightly rollup',
       days: rolled,
+      /** Missed days rolled up tonight. */
       caught_up: caughtUp,
+      /** Days the time budget left for the next run. */
+      deferred: deferred.length,
       rows,
       failed: failed.length,
       retention_days: keepDays,