trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns16g rEventsFlows

machine_sample keeps 48 h raw, then 5-minute rows for 30 days, then hourly rows #12028

Merged masoncrew opened this

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

mason/forge:mason/12007-machine-sample-keeps-48-h-raw-then-5-min → main 6 commits fork

  • Merged. mason/12007-machine-sample-keeps-48-h-raw-then-5-min was merged into main as 88a6bb7.

Review

Sign in to review this change.
9 files changed +381 −28
src/boards/ring.ts +5 −10
@@ -1,9 +1,8 @@
11 /**
2 * The machine's history on disk (`machine_sample`): every 15-s point for a
3 * day, and every 5-min point — the mean and the max of the 15-s points in
4 * it — for 30 days. Each write rolls up the 5-min windows that have closed
5 * and drops what has left its window, so the table never grows past
6 * 24 h / 15 s + 30 d / 5 min = 5,760 + 8,640 rows per target.
2 * The machine's history on disk (`machine_sample`): every 15-s point, and every
3 * 5-min point — the mean and the max of the 15-s points in it. Each write rolls
4 * up the 5-min windows that have closed. How long each is kept, and the hourly
5 * rows after that, is `system/retention_samples.ts`, which prunes in batches.
76 */
87 import type { Database } from "bun:sqlite";
98 import { withoutWaiting } from "../kernel/db.ts";
@@ -11,8 +10,6 @@ import { quietly } from "../core/write_clock.ts";
1110
1211 export const FINE_MS = 15_000;
1312 export const COARSE_MS = 300_000;
14 export const FINE_KEEP_MS = 24 * 3_600_000;
15 export const COARSE_KEEP_MS = 30 * 24 * 3_600_000;
1613
1714 /** One target at one instant. A null is a series this machine cannot read. */
1815 export type Point = {
@@ -95,7 +92,7 @@ type FineRow = {
9592 };
9693
9794 /**
98 * Keep `points` (taken at `now`), roll up the windows that closed, drop what is past its window. One transaction,
95 * Keep `points` (taken at `now`) and roll up the windows that closed. One transaction,
9996 * never waiting on another process's write lock: behind one it throws at once and the sampler's next tick tries again (#6957).
10097 */
10198 export function record(db: Database, points: readonly Point[], now: number): void {
@@ -136,7 +133,5 @@ export function record(db: Database, points: readonly Point[], now: number): voi
136133 r.celsius, r.celsiusMax, r.lagMs, r.lagMsMax, r.checks, r.agents,
137134 );
138135 }
139 db.query("DELETE FROM machine_sample WHERE step = 15 AND at < ?").run(iso(now - FINE_KEEP_MS));
140 db.query("DELETE FROM machine_sample WHERE step = 300 AND at < ?").run(iso(now - COARSE_KEEP_MS));
141136 })));
142137 }
src/boards/sampler.ts +8
@@ -31,6 +31,7 @@ import { readHwmon, readTopology, temperatures } from "./thermal.ts";
3131 import { during } from "../perf/jobs.ts";
3232 import { withoutWaiting } from "../kernel/db.ts";
3333 import { pruneEvents, TICK_ROWS } from "../system/retention_events.ts";
34 import { drainSamples } from "../system/retention_samples.ts";
3435
3536 export type Paths = { cgroup: string; proc: string; sys: string };
3637 export const SYSTEM: Paths = { cgroup: "/sys/fs/cgroup", proc: "/proc", sys: "/sys" };
@@ -161,6 +162,7 @@ export function startSampler(
161162 const cells = options.cells ?? cellCounters;
162163 const lag = options.lag ?? startLoopLag();
163164 let prev: Counters | undefined;
165 let pruning = false;
164166 const missing = gaps(paths);
165167 if (missing.length > 0) app.log.info("machine series unreadable here", { gaps: missing.join("; ") });
166168 const tick = (now = Date.now()) => {
@@ -178,6 +180,12 @@ export function startSampler(
178180 } catch (error) {
179181 app.log.warn("event prune failed", { error: String(error) });
180182 }
183 // What has left its window goes a bounded batch at a time (#12007); one drain at once.
184 if (pruning) return;
185 pruning = true;
186 void drainSamples(app.db, () => now)
187 .catch((error) => app.log.warn("machine sample prune failed", { error: String(error) }))
188 .finally(() => { pruning = false; });
181189 };
182190 during("sampler", tick);
183191 const timer = setInterval(() => during("sampler", tick), options.everyMs ?? FINE_MS);
src/boards/sources.ts +8 −7
@@ -202,15 +202,16 @@ export const SOURCES: Readonly<Record<string, Source>> = {
202202 value: "bytes received (or CPU-seconds, or share of installs that hit the cache) in the hour up to the reading",
203203 },
204204 },
205 // What the machine used (`boards/sampler.ts`, kept by `boards/ring.ts`):
206 // 15-s points for the last day, 5-min points before that, never both
207 // for one moment.
205 // What the machine used (`boards/sampler.ts`, kept by `system/retention_samples.ts`):
206 // 15-s points for the last 48 h, 5-min points before that, hourly points
207 // past 30 days, never two for one moment.
208208 machine: {
209209 table: "machine_sample",
210210 owner: "admin",
211211 where:
212 "(step = 15 OR at <= COALESCE((SELECT strftime('%Y-%m-%dT%H:%M:%fZ', MIN(at), '-300 seconds') " +
213 "FROM machine_sample WHERE step = 15), '9999'))",
212 "(step = 15 OR (step = 300 AND at <= COALESCE((SELECT strftime('%Y-%m-%dT%H:%M:%fZ', MIN(at), '-300 seconds') " +
213 "FROM machine_sample WHERE step = 15), '9999')) OR (step = 3600 AND at <= COALESCE((SELECT " +
214 "strftime('%Y-%m-%dT%H:%M:%fZ', MIN(at), '-3600 seconds') FROM machine_sample WHERE step = 300), '9999')))",
214215 time: "at",
215216 columns: {
216217 target: "text", at: "time", step: "number",
@@ -223,8 +224,8 @@ export const SOURCES: Readonly<Record<string, Source>> = {
223224 target:
224225 "forge (the pod), check:<run id>, app:<slug>/<env>, node, cpu:<nnn> (one logical CPU, zero-padded), " +
225226 "package:<id> or sensor:<chip>/<label> (a thermometer)",
226 at: "when the point was read (15 s) or its window began (5 min), UTC",
227 step: "seconds the point covers: 15, or 300 for a 5-min roll-up",
227 at: "when the point was read (15 s) or its window began (5 min, 1 h), UTC",
228 step: "seconds the point covers: 15, 300 for a 5-min roll-up, or 3600 for an hourly one",
228229 cpu_cores: "cores in use, mean over the step (CPU seconds per second); at most 1 for a cpu:<nnn>",
229230 cpu_cores_max: "cores, the highest 15-s point in the step",
230231 memory_bytes: "bytes resident, mean over the step",
src/core/migration-pins/20261114010000-machine-sample-hourly.pin added +1
@@ -0,0 +1,1 @@
1 72599f9b2be8c5e50182654eacc44bf7477cd1abea5614ebe3628b242cd56844
src/core/migrations/20261114010000-machine-sample-hourly.ts added +31
@@ -0,0 +1,31 @@
1 // What the machine used, kept longer and smaller (#12007, system/retention_samples.ts):
2 // 15-s points for 48 h, 5-min roll-ups for 30 days, hourly roll-ups after. SQLite cannot
3 // widen a CHECK in place, so the table is copied into one whose step also allows 3600.
4 // Telemetry, not state.
5 export default `
6 CREATE TABLE machine_sample_next (
7 at TEXT NOT NULL,
8 step INTEGER NOT NULL CHECK (step IN (15, 300, 3600)),
9 target TEXT NOT NULL,
10 cpu_cores REAL,
11 cpu_cores_max REAL,
12 memory_bytes REAL,
13 memory_bytes_max REAL,
14 load REAL,
15 celsius REAL,
16 celsius_max REAL,
17 lag_ms REAL,
18 lag_ms_max REAL,
19 checks REAL,
20 agents REAL,
21 PRIMARY KEY (step, at, target)
22 ) WITHOUT ROWID;
23 INSERT INTO machine_sample_next
24 (at, step, target, cpu_cores, cpu_cores_max, memory_bytes, memory_bytes_max, load,
25 celsius, celsius_max, lag_ms, lag_ms_max, checks, agents)
26 SELECT at, step, target, cpu_cores, cpu_cores_max, memory_bytes, memory_bytes_max, load,
27 celsius, celsius_max, lag_ms, lag_ms_max, checks, agents
28 FROM machine_sample;
29 DROP TABLE machine_sample;
30 ALTER TABLE machine_sample_next RENAME TO machine_sample;
31 `;
src/system/retention_samples.ts added +121
@@ -0,0 +1,121 @@
1 /**
2 * How long the machine's history (`machine_sample`) is kept, said once (#12007):
3 * every 15-s point for 48 h, the 5-min roll-ups (`boards/ring.ts` writes them as
4 * each window closes) for 30 days, and after that one row an hour per target —
5 * the mean of the hour's 5-min means, the max of their maxes.
6 *
7 * `pruneSamples` does one bounded batch: it never touches more than about
8 * `rowCap` rows, so it holds the write lock for milliseconds, and says whether
9 * there is `more` for the next call. `drainSamples` calls it a batch per turn of
10 * the event loop until there is none.
11 */
12 import type { Database } from "bun:sqlite";
13 import { withoutWaiting } from "../kernel/db.ts";
14 import { quietly } from "../core/write_clock.ts";
15
16 /** 15-s points are kept this long. */
17 export const RAW_KEEP_MS = 48 * 3_600_000;
18 /** 5-min roll-ups are kept this long, then folded into hourly rows. */
19 export const FIVE_MIN_KEEP_MS = 30 * 24 * 3_600_000;
20 /** An hourly row covers this long, `at` its start. */
21 export const HOUR_MS = 3_600_000;
22 /** Rows one batch touches at most (a single hour of many targets may go a little past it). */
23 export const ROW_CAP = 5_000;
24
25 export type PruneBatch = {
26 /** Hourly rows written. */
27 rolled: number;
28 /** 15-s and 5-min rows removed. */
29 deleted: number;
30 /** Whether rows past their window remain for another batch. */
31 more: boolean;
32 };
33
34 const iso = (ms: number) => new Date(ms).toISOString();
35 const hourOf = (ms: number) => Math.floor(ms / HOUR_MS) * HOUR_MS;
36
37 /** The `at` of the row `offset` places into `step`'s rows before `before`, oldest first. */
38 function nth(db: Database, step: number, before: string, offset: number): string | undefined {
39 return db
40 .query<{ at: string }, [number, string, number]>(
41 "SELECT at FROM machine_sample WHERE step = ?1 AND at < ?2 ORDER BY at LIMIT 1 OFFSET ?3",
42 )
43 .get(step, before, offset)?.at;
44 }
45
46 /** Drop up to `cap` 15-s points older than 48 h whose 5-min window is rolled up already. */
47 function dropRaw(db: Database, now: number, cap: number): { deleted: number; more: boolean } {
48 const rolledTo = db.query<{ at: string | null }, []>("SELECT MAX(at) AS at FROM machine_sample WHERE step = 300").get()?.at;
49 if (rolledTo == null) return { deleted: 0, more: false };
50 const keepFrom = Math.min(now - RAW_KEEP_MS, Date.parse(rolledTo) + 300_000);
51 const cut = iso(keepFrom);
52 const edge = nth(db, 15, cut, cap);
53 const del = db.query<unknown, [string]>("DELETE FROM machine_sample WHERE step = 15 AND at < ?1");
54 if (edge === undefined) return { deleted: del.run(cut).changes, more: false };
55 let deleted = del.run(edge).changes;
56 // More than `cap` points share the oldest instant: take that instant whole.
57 if (deleted === 0) deleted = db.query<unknown, [string]>("DELETE FROM machine_sample WHERE step = 15 AND at = ?1").run(edge).changes;
58 return { deleted, more: true };
59 }
60
61 /** Fold whole hours of 5-min rows past 30 days into hourly rows, about `cap` 5-min rows at most. */
62 function foldHours(db: Database, now: number, cap: number): PruneBatch {
63 const cut = iso(hourOf(now - FIVE_MIN_KEEP_MS));
64 const first = nth(db, 300, cut, 0);
65 if (first === undefined) return { rolled: 0, deleted: 0, more: false };
66 const from = hourOf(Date.parse(first));
67 const edge = nth(db, 300, cut, cap);
68 // Stop at the hour the cap falls in, so an hour is never folded in two halves; at least one hour.
69 const to = edge === undefined ? cut : iso(Math.max(hourOf(Date.parse(edge)), from + HOUR_MS));
70 const range = [iso(from), to] as [string, string];
71 const rolled = db
72 .query<unknown, [string, string]>(
73 `INSERT OR REPLACE INTO machine_sample
74 (at, step, target, cpu_cores, cpu_cores_max, memory_bytes, memory_bytes_max, load,
75 celsius, celsius_max, lag_ms, lag_ms_max, checks, agents)
76 SELECT substr(at, 1, 13) || ':00:00.000Z', 3600, target,
77 AVG(cpu_cores), MAX(cpu_cores_max), AVG(memory_bytes), MAX(memory_bytes_max), AVG(load),
78 AVG(celsius), MAX(celsius_max), AVG(lag_ms), MAX(lag_ms_max), AVG(checks), AVG(agents)
79 FROM machine_sample WHERE step = 300 AND at >= ?1 AND at < ?2
80 GROUP BY substr(at, 1, 13), target`,
81 )
82 .run(...range).changes;
83 const deleted = db.query<unknown, [string, string]>("DELETE FROM machine_sample WHERE step = 300 AND at >= ?1 AND at < ?2").run(...range).changes;
84 return { rolled, deleted, more: edge !== undefined };
85 }
86
87 /**
88 * One batch of the rule at `now`: 15-s points past 48 h go (their 5-min windows
89 * hold them), then 5-min rows past 30 days become hourly rows, at most about
90 * `rowCap` rows in all. One transaction that never waits on another writer:
91 * behind one it throws, and the next call tries again.
92 */
93 export function pruneSamples(db: Database, now: number, rowCap: number = ROW_CAP): PruneBatch {
94 const cap = Math.max(1, Math.floor(rowCap));
95 return withoutWaiting(db, db.transaction(() => quietly(db, () => {
96 const raw = dropRaw(db, now, cap);
97 if (raw.more) return { rolled: 0, deleted: raw.deleted, more: true };
98 const fold = foldHours(db, now, Math.max(1, cap - raw.deleted));
99 return { rolled: fold.rolled, deleted: raw.deleted + fold.deleted, more: fold.more };
100 })));
101 }
102
103 /**
104 * Batches until none is left, yielding the event loop between them so no
105 * request waits behind more than one; the totals, or what stopped it.
106 */
107 export async function drainSamples(
108 db: Database,
109 now: () => number = Date.now,
110 rowCap: number = ROW_CAP,
111 ): Promise<{ rolled: number; deleted: number; batches: number }> {
112 const sum = { rolled: 0, deleted: 0, batches: 0 };
113 for (;;) {
114 const b = pruneSamples(db, now(), rowCap);
115 sum.rolled += b.rolled;
116 sum.deleted += b.deleted;
117 sum.batches++;
118 if (!b.more) return sum;
119 await new Promise((resolve) => setImmediate(resolve));
120 }
121 }
test/boards-sampler.test.ts +38 −11
@@ -1,8 +1,8 @@
11 /**
22 * The machine sampler and its ring: a fake cgroup and /proc tree yields
33 * hand-calculated rates, a missing file is an absent series (never a
4 * zero), the 5-min roll-up of known points is exact, and the ring drops
5 * what has left its window. The `machine` source reads it like any other.
4 * zero), the 5-min roll-up of known points is exact, and the retention rule
5 * (`system/retention_samples.ts`) drops what has left its window. The `machine` source reads it like any other.
66 */
77 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
88 import type { Database } from "bun:sqlite";
@@ -10,7 +10,8 @@ import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
1010 import { tmpdir } from "node:os";
1111 import { join } from "node:path";
1212 import { runQuery } from "../src/boards/query.ts";
13 import { COARSE_KEEP_MS, COARSE_MS, FINE_KEEP_MS, FINE_MS, record, rollUp, type Point } from "../src/boards/ring.ts";
13 import { COARSE_MS, FINE_MS, record, rollUp, type Point } from "../src/boards/ring.ts";
14 import { FIVE_MIN_KEEP_MS as COARSE_KEEP_MS, pruneSamples, RAW_KEEP_MS as FINE_KEEP_MS } from "../src/system/retention_samples.ts";
1415 import { gaps, pointsOf, readCounters, startSampler, type Paths } from "../src/boards/sampler.ts";
1516 import { SOURCES } from "../src/boards/sources.ts";
1617 import type { CellCounter } from "../src/apps/contain.ts";
@@ -164,28 +165,32 @@ describe("the ring", () => {
164165 put.run(iso(now - COARSE_KEEP_MS - 1), 300);
165166 put.run(iso(now - COARSE_KEEP_MS), 300);
166167 record(db, [], now);
168 while (pruneSamples(db, now).more);
167169 const at = (step: number) =>
168170 db.query<{ at: string }, [number]>("SELECT at FROM machine_sample WHERE step = ? ORDER BY at").all(step).map((r) => r.at);
169171 expect(at(15)).toEqual([iso(now - FINE_KEEP_MS)]);
170 // The kept 15-s point was rolled up into its window as well; the old 5-min point is gone, the edge one stays.
172 // The kept 15-s point was rolled up into its window as well; the old 5-min point
173 // became an hourly one, the edge one stays.
171174 expect(at(300)).toContain(iso(now - COARSE_KEEP_MS));
172175 expect(at(300)).not.toContain(iso(now - COARSE_KEEP_MS - 1));
176 expect(at(3600)).toEqual([iso(Math.floor((now - COARSE_KEEP_MS - 1) / 3_600_000) * 3_600_000)]);
173177 });
174178
175 test("a day of 15-s samples stays bounded at a day", () => {
179 test("15-s samples stay bounded at 48 h", () => {
176180 const db = fresh();
177 // Two days, one point a minute (×4 fewer than real, same bound).
181 // Twice the window, one point a minute (×4 fewer than real, same bound).
178182 for (let t = 0; t <= 2 * FINE_KEEP_MS; t += 60_000) {
179183 record(db, [{ at: T0 + t, target: "forge", cpuCores: 1, memoryBytes: 1, load: null }], T0 + t);
184 while (pruneSamples(db, T0 + t).more);
180185 }
181186 const fine = db.query<{ n: number }, []>("SELECT COUNT(*) AS n FROM machine_sample WHERE step = 15").get()!.n;
182187 expect(fine).toBe(FINE_KEEP_MS / 60_000 + 1);
183 // Every closed 5-min window of the two days was rolled up, and kept.
188 // Every closed 5-min window was rolled up, and kept.
184189 const coarse = db.query<{ n: number }, []>("SELECT COUNT(*) AS n FROM machine_sample WHERE step = 300").get()!.n;
185190 expect(coarse).toBe((2 * FINE_KEEP_MS) / 300_000);
186191 });
187192
188 test("31 days of 15-s points keep at most 30 days of 5-min roll-ups, none past COARSE_KEEP_MS, within 5 s scaled to this machine", () => {
193 test("31 days of 15-s points keep at most 30 days of 5-min roll-ups, hourly past COARSE_KEEP_MS, within 5 s scaled to this machine", () => {
189194 const db = fresh();
190195 const days = 31;
191196 const end = T0 + days * 24 * 3_600_000;
@@ -210,19 +215,23 @@ describe("the ring", () => {
210215 batch.push({ at: t, target: "forge", cpuCores: 1, memoryBytes: 1, load: null });
211216 }
212217 record(db, batch, w + COARSE_MS - FINE_MS);
218 while (pruneSamples(db, w + COARSE_MS - FINE_MS).more);
213219 }
214220 const now = end - FINE_MS;
215221 expect(performance.now() - started).toBeLessThan(budget);
216222
217223 const count = (step: number) =>
218224 db.query<{ n: number }, [number]>("SELECT COUNT(*) AS n FROM machine_sample WHERE step = ? AND target = 'forge'").get(step)!.n;
219 // 30 × 288 closed windows, plus the one the edge of the window falls in.
220 expect(count(300)).toBeLessThanOrEqual(30 * 288 + 1);
225 // 30 × 288 closed windows, plus those in the hour the edge of the window falls in.
226 expect(count(300)).toBeLessThanOrEqual(30 * 288 + 12);
221227 expect(count(300)).toBeGreaterThan(29 * 288);
222228 expect(count(15)).toBeLessThanOrEqual(FINE_KEEP_MS / FINE_MS + 1);
223229 expect(count(15)).toBeGreaterThan(FINE_KEEP_MS / FINE_MS - 20);
224230 const oldest = db.query<{ at: string }, []>("SELECT MIN(at) AS at FROM machine_sample WHERE step = 300").get()!.at;
225 expect(Date.parse(oldest)).toBeGreaterThanOrEqual(now - COARSE_KEEP_MS);
231 expect(Date.parse(oldest)).toBeGreaterThanOrEqual(Math.floor((now - COARSE_KEEP_MS) / 3_600_000) * 3_600_000);
232 // The day before that is hourly rows, about 24 of them.
233 expect(count(3600)).toBeGreaterThanOrEqual(23);
234 expect(count(3600)).toBeLessThanOrEqual(25);
226235 });
227236 });
228237
@@ -263,6 +272,24 @@ describe("the machine source", () => {
263272 });
264273
265274 describe("startSampler", () => {
275 test("a tick prunes what has left its window, a batch at a time", () => {
276 const db = fresh();
277 const app = { db, log: { warn: () => {}, info: () => {} } } as unknown as App;
278 const put = db.prepare("INSERT INTO machine_sample (at, step, target) VALUES (?, ?, 'node')");
279 const old = T0 - FINE_KEEP_MS - 60_000;
280 put.run(new Date(old).toISOString(), 15);
281 put.run(new Date(Math.floor(old / COARSE_MS) * COARSE_MS).toISOString(), 300);
282 put.run(new Date(T0).toISOString(), 300);
283 const s = startSampler(app, { paths, everyMs: 3_600_000, cells: () => [] });
284 try {
285 // The drain's first batch runs inside the tick; only further batches wait a turn.
286 s.tick(T0);
287 } finally {
288 s.stop();
289 }
290 expect(db.query("SELECT COUNT(*) AS n FROM machine_sample WHERE step = 15 AND at = ?").get(new Date(old).toISOString())).toEqual({ n: 0 });
291 });
292
266293 test("ticks write the forge, its cells and the node into the ring", () => {
267294 const db = fresh();
268295 const warnings: string[] = [];
test/retention-samples.test.ts added +168
@@ -0,0 +1,168 @@
1 /**
2 * The machine history's retention rule (#12007): 15-s points for 48 h, 5-min
3 * roll-ups for 30 days, hourly after; one bounded batch at a time, never a
4 * 15-s point its 5-min window does not hold yet, never an hour folded in two
5 * halves, and the readers answer the same before and after.
6 */
7 import { describe, expect, test } from "bun:test";
8 import type { Database } from "bun:sqlite";
9 import type { App } from "../src/app.ts";
10 import { runQuery } from "../src/boards/query.ts";
11 import { record, rollUp, type Point } from "../src/boards/ring.ts";
12 import { migrations } from "../src/core/schema.ts";
13 import { cpuSecondsLastHour } from "../src/efficiency/readings.ts";
14 import { openDb } from "../src/kernel/db.ts";
15 import { historyOver } from "../src/system/telemetry.ts";
16 import { once } from "./helpers/statements.ts";
17 import { expectWork } from "./helpers/work.ts";
18 import {
19 drainSamples,
20 FIVE_MIN_KEEP_MS,
21 HOUR_MS,
22 pruneSamples,
23 RAW_KEEP_MS,
24 ROW_CAP,
25 } from "../src/system/retention_samples.ts";
26
27 const fresh = () => openDb(":memory:", migrations);
28 const iso = (ms: number) => new Date(ms).toISOString();
29 const NOW = Date.parse("2026-11-13T12:00:07.000Z");
30 const count = (db: Database, step: number) =>
31 db.query<{ n: number }, [number]>("SELECT COUNT(*) AS n FROM machine_sample WHERE step = ?").get(step)!.n;
32
33 /** 15-s points from `from` to `to` for `targets`, and their 5-min roll-ups, written straight in. */
34 function seed(db: Database, from: number, to: number, targets: readonly string[], value: (t: number, i: number) => number) {
35 const points: Point[] = [];
36 for (let t = from; t < to; t += 15_000) {
37 targets.forEach((target, i) => points.push({ at: t, target, cpuCores: value(t, i), memoryBytes: 1_000 + i, load: i }));
38 }
39 const put = db.prepare(
40 "INSERT OR REPLACE INTO machine_sample (at, step, target, cpu_cores, cpu_cores_max, memory_bytes, memory_bytes_max, load) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
41 );
42 db.transaction(() => {
43 for (const p of points) put.run(iso(p.at), 15, p.target, p.cpuCores, p.cpuCores, p.memoryBytes, p.memoryBytes, p.load);
44 for (const r of rollUp(points)) put.run(iso(r.at), 300, r.target, r.cpuCores, r.cpuCoresMax, r.memoryBytes, r.memoryBytesMax, r.load);
45 })();
46 }
47
48 describe("pruneSamples", () => {
49 test("one batch over 50k old 15-s points touches at most the cap in a few statements", async () => {
50 const db = fresh();
51 const targets = Array.from({ length: 20 }, (_, i) => `cpu:${String(i).padStart(3, "0")}`);
52 // 50,400 points (20 targets × 2,520 instants), all past 48 h, their windows rolled up.
53 seed(db, NOW - RAW_KEEP_MS - 2_520 * 15_000, NOW - RAW_KEEP_MS, targets, () => 1);
54 expect(count(db, 15)).toBeGreaterThanOrEqual(50_000);
55 record(db, [], NOW);
56 pruneSamples(db, NOW, 1); // warm the statements
57 let batch = { rolled: 0, deleted: 0, more: false };
58 const ran = await once(db, () => { batch = pruneSamples(db, NOW); });
59 expectWork(ran, { max: 6 });
60 expect(batch.deleted).toBeGreaterThan(0);
61 expect(batch.deleted).toBeLessThanOrEqual(ROW_CAP);
62 expect(batch.more).toBe(true);
63 });
64
65 test("a batch per call until `more` is false leaves no 15-s point past 48 h", async () => {
66 const db = fresh();
67 seed(db, NOW - RAW_KEEP_MS - 6 * HOUR_MS, NOW, ["node", "forge"], () => 1);
68 const sum = await drainSamples(db, () => NOW, 500);
69 expect(sum.batches).toBeGreaterThan(1);
70 const oldest = db.query<{ at: string }, []>("SELECT MIN(at) AS at FROM machine_sample WHERE step = 15").get()!.at;
71 expect(Date.parse(oldest)).toBeGreaterThanOrEqual(NOW - RAW_KEEP_MS);
72 expect(pruneSamples(db, NOW)).toEqual({ rolled: 0, deleted: 0, more: false });
73 });
74
75 test("a 15-s point whose 5-min window is not rolled up yet stays", () => {
76 const db = fresh();
77 db.query("INSERT INTO machine_sample (at, step, target, cpu_cores) VALUES (?, 15, 'node', 1)").run(iso(NOW - 3 * RAW_KEEP_MS));
78 expect(pruneSamples(db, NOW)).toEqual({ rolled: 0, deleted: 0, more: false });
79 expect(count(db, 15)).toBe(1);
80 });
81
82 test("5-min rows past 30 days become hourly means and maxes, whole hours whatever the cap", () => {
83 const build = () => {
84 const db = fresh();
85 const put = db.prepare(
86 "INSERT INTO machine_sample (at, step, target, cpu_cores, cpu_cores_max, memory_bytes, load, celsius) VALUES (?, 300, ?, ?, ?, ?, ?, ?)",
87 );
88 const start = Math.floor((NOW - FIVE_MIN_KEEP_MS) / HOUR_MS) * HOUR_MS - 3 * HOUR_MS;
89 // Three past hours plus the hour the edge falls in, three targets.
90 for (let t = start; t < start + 4 * HOUR_MS + 30 * 60_000; t += 300_000) {
91 for (const [i, target] of ["a", "b", "c"].entries()) {
92 const k = (t - start) / 300_000;
93 put.run(iso(t), target, k + i, 2 * k + i, 100, null, target === "b" ? null : 40 + (k % 12));
94 }
95 }
96 return { db, start };
97 };
98 const big = build();
99 const small = build();
100 while (pruneSamples(big.db, NOW).more);
101 let batches = 0;
102 for (;;) {
103 batches++;
104 if (!pruneSamples(small.db, NOW, 7).more) break;
105 }
106 expect(batches).toBeGreaterThan(3);
107 const rows = (db: Database) => db.query("SELECT * FROM machine_sample ORDER BY step, at, target").all();
108 expect(rows(small.db)).toEqual(rows(big.db));
109
110 const hourly = big.db
111 .query<{ at: string; target: string; cpu_cores: number; cpu_cores_max: number; memory_bytes: number; load: number | null; celsius: number | null }, []>(
112 "SELECT at, target, cpu_cores, cpu_cores_max, memory_bytes, load, celsius FROM machine_sample WHERE step = 3600 ORDER BY at, target",
113 )
114 .all();
115 expect(hourly).toHaveLength(9);
116 // Hour 0 of target a: k = 0..11, mean 5.5, max of maxes 22.
117 expect(hourly[0]).toEqual({ at: iso(big.start), target: "a", cpu_cores: 5.5, cpu_cores_max: 22, memory_bytes: 100, load: null, celsius: 45.5 });
118 expect(hourly[1]!.celsius).toBeNull();
119 // The hour the 30-day edge falls in stays 5-min, whole.
120 const edgeHour = Math.floor((NOW - FIVE_MIN_KEEP_MS) / HOUR_MS) * HOUR_MS;
121 const oldest5 = big.db.query<{ at: string }, []>("SELECT MIN(at) AS at FROM machine_sample WHERE step = 300").get()!.at;
122 expect(oldest5).toBe(iso(edgeHour));
123 });
124
125 test("the table still refuses a step it does not know", () => {
126 const db = fresh();
127 expect(() => db.query("INSERT INTO machine_sample (at, step, target) VALUES ('x', 60, 'node')").run()).toThrow();
128 db.query("INSERT INTO machine_sample (at, step, target) VALUES ('x', 3600, 'node')").run();
129 });
130 });
131
132 describe("the readers answer the same for the ages they show", () => {
133 test("3 days of samples: telemetry history and the efficiency reading before and after pruning", async () => {
134 const db = fresh();
135 const app = { db } as unknown as App;
136 const now = Date.now();
137 const from = Math.floor((now - 3 * 24 * HOUR_MS) / 300_000) * 300_000;
138 seed(db, from, now - 15_000, ["node", "forge"], (t, i) => ((t / 15_000) % 7) + i);
139 const read = () => ({
140 h1: historyOver(app, 1),
141 h24: historyOver(app, 24),
142 h72: historyOver(app, 72),
143 cpu: cpuSecondsLastHour(db, now),
144 });
145 const before = read();
146 expect(before.h72.length).toBeGreaterThan(0);
147 expect(before.cpu).toBeGreaterThan(0);
148 const fine = count(db, 15);
149 const sum = await drainSamples(db, () => now);
150 expect(sum.deleted).toBeGreaterThan(0);
151 expect(count(db, 15)).toBeLessThan(fine);
152 expect(read()).toEqual(before);
153 });
154 });
155
156 describe("the machine source", () => {
157 test("reads hourly rows before the 5-min ones, never two for one moment", () => {
158 const db = fresh();
159 const put = db.prepare("INSERT INTO machine_sample (at, step, target, cpu_cores) VALUES (?, ?, 'forge', ?)");
160 put.run("2026-10-01T10:00:00.000Z", 3600, 1);
161 put.run("2026-10-01T11:00:00.000Z", 3600, 2);
162 put.run("2026-10-01T11:00:00.000Z", 300, 4); // overlaps the 11:00 hour: the hourly one is left out
163 put.run("2026-10-01T11:05:00.000Z", 300, 8);
164 put.run("2026-10-01T11:10:00.000Z", 15, 16);
165 const sum = runQuery(db, { source: "machine", aggregate: "sum", column: "cpu_cores" }, { repoIds: "all" });
166 expect(sum).toEqual({ ok: true, rows: [{ group: null, value: 1 + 4 + 8 + 16 }] });
167 });
168 });
test/schema-additive.test.ts +1
@@ -163,6 +163,7 @@ const REBUILT: Record<string, string> = {
163163 "20261104100000-conversation-run-stopped-state": "conversation_run widens its state CHECK to admit stopped (#8972); same columns, same rows, the stopped column kept",
164164 "20261114000000-estimate-done": "issue_estimate widens its event CHECK to admit done (#11955); same columns, same rows, the indexes made again; the release before never writes or reads a done row",
165165 "20261110170000-fork-follow-dont-follow": "fork_follow widens its mode CHECK to admit dont-follow (#608); same columns, same rows; the release before reads dont-follow as not take-updates",
166 "20261114010000-machine-sample-hourly": "machine_sample widens its step CHECK to admit 3600 (#12007); same columns, same rows; the release before reads by step 15 or 300 and deletes past 30 days what an hourly row would hold",
166167 };
167168
168169 const src = join(import.meta.dir, "../src");

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs