mirror of
https://github.com/colbymchenry/codegraph.git
synced 2026-10-18 05:41:18 +08:00
Colby Mchenry
a192df40f9
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>
439 lines
20 KiB
TypeScript
439 lines
20 KiB
TypeScript
/**
|
|
* 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 });
|
|
});
|
|
});
|
|
});
|