diff --git a/__tests__/telemetry-services.test.ts b/__tests__/telemetry-services.test.ts new file mode 100644 index 000000000..657681cc7 --- /dev/null +++ b/__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; + +// --------------------------------------------------------------------------- +// 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(): Promise { + return (this.execute().results[0] as T | undefined) ?? null; + } + + async all(): Promise> { + return this.execute(); + } + + async run(): Promise> { + 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[]> { + 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 { + 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[] = []; + 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) => 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 { + 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 = 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 }); + }); + }); +}); diff --git a/docs/design/telemetry.md b/docs/design/telemetry.md index 530600d93..bf5deed9c 100644 --- a/docs/design/telemetry.md +++ b/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 diff --git a/telemetry-dashboard/README.md b/telemetry-dashboard/README.md index 26feaac2d..10c1fb3fc 100644 --- a/telemetry-dashboard/README.md +++ b/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 diff --git a/telemetry-dashboard/public/app.js b/telemetry-dashboard/public/app.js index b29506aad..abb2f46ef 100644 --- a/telemetry-dashboard/public/app.js +++ b/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.', ]); } diff --git a/telemetry-dashboard/scripts/fixture.sql b/telemetry-dashboard/scripts/fixture.sql index 9e2ddef02..e9ca7c151 100644 --- a/telemetry-dashboard/scripts/fixture.sql +++ b/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 diff --git a/telemetry-dashboard/scripts/smoke-api.sh b/telemetry-dashboard/scripts/smoke-api.sh index 67008c47b..8dda68403 100755 --- a/telemetry-dashboard/scripts/smoke-api.sh +++ b/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 ]] diff --git a/telemetry-dashboard/src/api.ts b/telemetry-dashboard/src/api.ts index 63b77556c..24deb3352 100644 --- a/telemetry-dashboard/src/api.ts +++ b/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 { (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 { 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 { 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 { const rawWindow = url.searchParams.get('window'); diff --git a/telemetry-worker/README.md b/telemetry-worker/README.md index 296ec2251..077d0812f 100644 --- a/telemetry-worker/README.md +++ b/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 diff --git a/telemetry-worker/scripts/smoke-cutover.sh b/telemetry-worker/scripts/smoke-cutover.sh index f0dff1244..6d55f8c79 100755 --- a/telemetry-worker/scripts/smoke-cutover.sh +++ b/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 diff --git a/telemetry-worker/scripts/smoke-ingest.sh b/telemetry-worker/scripts/smoke-ingest.sh index 458123443..9fd2b87fe 100755 --- a/telemetry-worker/scripts/smoke-ingest.sh +++ b/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" diff --git a/telemetry-worker/scripts/smoke-rollup.sh b/telemetry-worker/scripts/smoke-rollup.sh index 294b4378f..2befab4eb 100755 --- a/telemetry-worker/scripts/smoke-rollup.sh +++ b/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. diff --git a/telemetry-worker/src/index.ts b/telemetry-worker/src/index.ts index 9a6efa339..3c4ac7cde 100644 --- a/telemetry-worker/src/index.ts +++ b/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); diff --git a/telemetry-worker/src/rollup.ts b/telemetry-worker/src/rollup.ts index 87072a59a..950221098 100644 --- a/telemetry-worker/src/rollup.ts +++ b/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 { 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 { +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 { 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 { 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 { 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 { 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,