trevato/forge

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

More
Forks1g fMapg aRuns11g rEventsFlows

Meter awake time and storage: per app and environment, the two quantities a forge charges for #527

Merged torquecrew opened this

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

ConversationFiles
Merged

torque/forge:torque/505-meter-awake-time-and-storage-per-app-and → main 4 commits fork

  • Merged. torque/505-meter-awake-time-and-storage-per-app-and was merged into main as ffc20f7.

Review

Sign in to review this change.
10 files changed +764 −4
docs/APPS.md +16
@@ -778,6 +778,22 @@ it was kept awake. Busy buys one crew run's length past the idle window
778778 (`FORGE_CREW_TIMEOUT_MIN`, 45 minutes by default) and no more, so an app
779779 that keeps a child forever still rests.
780780
781 ### The meter
782
783 Awake and asleep are also what an app costs, so the runtime writes them
784 down (`core/meter.ts`). Every stretch an environment serves is one row in
785 `meter` — repository, environment, placement (`owned` or `cloud`), started,
786 ended — opened when a deploy or a wake starts serving (a redeploy keeps the
787 same row open) and closed when it sleeps, stops, crashes, is torn down or
788 the forge shuts down. Each sweep is a heartbeat on the open rows; a row the
789 last forge left open is closed at boot at its last heartbeat, never left
790 running. Once an hour the runtime samples each app's bytes — every
791 environment's `FORGE_APP_DATA` and its bare repository, files only, links
792 not followed (`apps/bytes.ts`) — into `storage`. `usageBetween` reads both
793 for a `[from, to)` range: awake seconds per app (by placement too) and
794 GB-hours, bytes × hours / 1e9, each sample standing for the hour after it
795 or until the next one.
796
781797 ### Subscriptions
782798
783799 One token in the environment is one account for the whole forge. An admin
forge.json +1 −1
@@ -6,7 +6,7 @@
66 "ready": 90
77 },
88 "check": {
9 "run": "bun x tsc --noEmit && bun scripts/test-shards.ts --timeout 60000 test/crew.test.ts test/crew-wrapup.test.ts test/crew-limits.test.ts test/crew-sandbox.test.ts test/steer.test.ts test/landing.test.ts test/decide.test.ts test/ledger.test.ts test/checks.test.ts test/dependencies.test.ts test/autonomy.test.ts test/studio.test.ts test/studio-words.test.ts test/forks.test.ts test/fork-flow.test.ts test/cross-repo-pull.test.ts test/pulls.test.ts test/apps.test.ts test/apps-adversarial.test.ts test/apps-busy.test.ts test/apps-token.test.ts test/web.test.ts test/git-trust.test.ts test/helpers.test.ts test/shards.test.ts test/migrate.test.ts test/vitals.test.ts test/asks.test.ts test/feed.test.ts test/feed-page.test.ts test/offline.test.ts test/share.test.ts test/cloud.test.ts test/run-secrets.test.ts test/forge-cloudflare.test.ts test/usage.test.ts test/standing.test.ts test/board-windows.test.ts test/spawn-plan.test.ts test/spawn.test.ts test/web-apps.test.ts test/fleet-ops.test.ts",
9 "run": "bun x tsc --noEmit && bun scripts/test-shards.ts --timeout 60000 -- crew crew-wrapup crew-limits crew-sandbox steer landing decide ledger checks dependencies autonomy studio studio-words forks fork-flow cross-repo-pull pulls apps apps-adversarial apps-busy apps-token web git-trust helpers shards migrate vitals asks feed feed-page offline share cloud run-secrets forge-cloudflare usage standing board-windows spawn-plan spawn web-apps fleet-ops meter",
1010 "timeout": 1800
1111 }
1212 }
scripts/shards.ts +28
@@ -79,6 +79,34 @@ export function splitFiles(
7979 /** How a test file is named: the same rule `scripts/test-shards.ts` uses. */
8080 export const TEST_FILE = /\.test\.[cm]?[jt]sx?$/;
8181
82 /** A test's short name after `--`: `crew-wrapup` is `test/crew-wrapup.test.ts`. */
83 const TEST_NAME = /^[a-z0-9][a-z0-9-]{0,99}$/;
84
85 /**
86 * Splits `scripts/test-shards.ts`'s arguments into `bun test` flags and
87 * test files. Before `--`, an argument naming a test file is a file and
88 * anything else a flag; after it, every argument is a short name under
89 * `test/` (`crew` is `test/crew.test.ts`), which keeps `forge.json`'s
90 * `check.run` under the manifest's 1000-character bound.
91 */
92 export function splitArgs(
93 args: readonly string[],
94 ):
95 | { ok: true; flags: string[]; files: string[] }
96 | { ok: false; error: string } {
97 const at = args.indexOf("--");
98 const head = at === -1 ? args : args.slice(0, at);
99 const names = at === -1 ? [] : args.slice(at + 1);
100 const flags = head.filter((arg) => !TEST_FILE.test(arg));
101 const files = head.filter((arg) => TEST_FILE.test(arg));
102 for (const name of names) {
103 if (!TEST_NAME.test(name))
104 return { ok: false, error: `${JSON.stringify(name)} is not a test name` };
105 files.push(`test/${name}.test.ts`);
106 }
107 return { ok: true, flags, files };
108 }
109
82110 /** A sane bound on one entry of the times table: a day, in seconds. */
83111 const TIME_MAX = 86_400;
84112
scripts/test-shards.ts +8 −3
@@ -3,6 +3,7 @@
33 *
44 * bun scripts/test-shards.ts [bun test flags…] test/a.test.ts test/b.test.ts …
55 * bun scripts/test-shards.ts --record [bun test flags…] test/a.test.ts …
6 * bun scripts/test-shards.ts [bun test flags…] -- a b … (test/a.test.ts …)
67 *
78 * The files come from the command line, so `forge.json`'s `check.run`
89 * stays the one place that names them. Each shard is its own `bun test`
@@ -21,10 +22,10 @@
2122 import { availableParallelism } from "node:os";
2223 import { readFileSync, statSync, writeFileSync } from "node:fs";
2324 import {
24 TEST_FILE,
2525 formatTimes,
2626 parseTimes,
2727 shardCount,
28 splitArgs,
2829 splitFiles,
2930 tally,
3031 weigher,
@@ -36,8 +37,12 @@ const TIMES = new URL("./test-times.json", import.meta.url).pathname;
3637 const argv = process.argv.slice(2);
3738 const record = argv.includes("--record");
3839 const args = argv.filter((arg) => arg !== "--record");
39 const files = args.filter((arg) => TEST_FILE.test(arg));
40 const flags = args.filter((arg) => !TEST_FILE.test(arg));
40 const split = splitArgs(args);
41 if (!split.ok) {
42 console.error(`test-shards: ${split.error}`);
43 process.exit(2);
44 }
45 const { files, flags } = split;
4146
4247 if (files.length === 0) {
4348 console.error("test-shards: name the test files to run");
src/apps/bytes.ts added +40
@@ -0,0 +1,40 @@
1 /**
2 * How many bytes a directory holds: the apparent size of every regular
3 * file under it, summed. Symlinks are counted as themselves and never
4 * followed, so a link out of the tree cannot bill for what it points at;
5 * a path that is missing (an environment never deployed) holds 0 bytes.
6 * Asynchronous, so walking a large repository does not hold up a request.
7 */
8
9 import { lstat, readdir } from "node:fs/promises";
10 import { join } from "node:path";
11
12 export async function dirBytes(path: string): Promise<number> {
13 let total = 0;
14 const pending = [path];
15 while (pending.length > 0) {
16 const dir = pending.pop()!;
17 let names: string[];
18 try {
19 const info = await lstat(dir);
20 if (!info.isDirectory()) {
21 if (info.isFile()) total += info.size;
22 continue;
23 }
24 names = await readdir(dir);
25 } catch {
26 continue; // gone between the listing and the look: nothing to count
27 }
28 for (const name of names) {
29 const child = join(dir, name);
30 try {
31 const info = await lstat(child);
32 if (info.isDirectory()) pending.push(child);
33 else if (info.isFile()) total += info.size;
34 } catch {
35 // removed while walking
36 }
37 }
38 }
39 return total;
40 }
src/apps/runtime.ts +110
@@ -45,6 +45,18 @@ import type { Result } from "../kernel/result.ts";
4545 import { holdLease, STALE_AFTER_MS } from "../runs/lease.ts";
4646 import { appendRunLog } from "../runs/log.ts";
4747 import { primaryDomain } from "../core/domains.ts";
48 import {
49 beatAwake,
50 closeAllAwake,
51 closeAwake,
52 closeOrphans,
53 lastStorageAt,
54 openAwake,
55 recordStorage,
56 SAMPLE_SPAN_MS,
57 type Placement,
58 } from "../core/meter.ts";
59 import { dirBytes } from "./bytes.ts";
4860 import { BUSY_WORDS, busyReason, type BusyReason } from "./busy.ts";
4961 import { exportTree, readFileAt } from "./checkout.ts";
5062 import type { DomainResolver } from "./domains.ts";
@@ -110,6 +122,10 @@ export interface RuntimeOptions {
110122 * its own, so proving `app.example.com` needs no network in a test.
111123 */
112124 readonly dns?: DomainResolver | undefined;
125 /** Where this forge's processes run, as the meter records it. @default "owned" */
126 readonly placement?: Placement | undefined;
127 /** How often storage is sampled for the meter. @default one hour */
128 readonly storageEveryMs?: number | undefined;
113129 }
114130
115131 export type EnvState =
@@ -522,6 +538,9 @@ export class AppsRuntime {
522538 failAbandonedRuns(this.app.db, STALE_AFTER_MS, ["deploy"]) +
523539 requeueAbandonedChecks(this.app, STALE_AFTER_MS);
524540 if (abandoned > 0) log.info("settled abandoned runs", { count: abandoned });
541 // An awake interval the last forge left open ends at its last heartbeat.
542 const orphans = this.metered(() => closeOrphans(this.app.db));
543 if (orphans) log.info("closed open meter intervals", { count: orphans });
525544 this.unsubscribe = this.app.bus.subscribe((event) => {
526545 // Never let an event handler's rejection reach the bus.
527546 this.onEvent(event).catch((error) =>
@@ -539,6 +558,7 @@ export class AppsRuntime {
539558 await this.reconcileApp(record, repo, "boot");
540559 }
541560 this.pruneDirs(known);
561 await this.sampleStorage();
542562 this.sweeper = setInterval(
543563 () => void this.sweep(),
544564 this.options.sweepEveryMs ?? SWEEP_EVERY_MS,
@@ -566,6 +586,7 @@ export class AppsRuntime {
566586 stops.push(state.chain.catch(() => {}));
567587 }
568588 await Promise.all(stops);
589 this.metered(() => closeAllAwake(this.app.db, new Date().toISOString()));
569590 runtimes.delete(this.app);
570591 }
571592
@@ -921,6 +942,7 @@ export class AppsRuntime {
921942 deployStartedAt,
922943 };
923944 state.live = live;
945 this.awake(state, servingAt);
924946 state.asleep = undefined;
925947 state.lastRequestAt = Date.now();
926948 state.lastError = undefined;
@@ -1153,6 +1175,7 @@ export class AppsRuntime {
11531175 .then((code) => {
11541176 if (this.stopped || state.live !== live) return; // replaced or stopped on purpose
11551177 state.live = undefined;
1178 this.resting(state);
11561179 revokeAppToken(this.app.db, live.tokenId);
11571180 state.lastError = `exited with code ${code}`;
11581181 appendRunLog(
@@ -1286,6 +1309,7 @@ export class AppsRuntime {
12861309 };
12871310 state.asleep = undefined;
12881311 state.live = live;
1312 this.awake(state, now);
12891313 state.lastRequestAt = Date.now();
12901314 this.watch(state, repo, live);
12911315 write("awake");
@@ -1451,6 +1475,7 @@ export class AppsRuntime {
14511475 const after = this.options.sleepAfterMs ?? SLEEP_AFTER_MS;
14521476 const grace = this.options.busyGraceMs ?? BUSY_GRACE_MS;
14531477 const now = Date.now();
1478 this.metered(() => beatAwake(this.app.db, new Date(now).toISOString()));
14541479 for (const state of this.envs.values()) {
14551480 const live = state.live;
14561481 // Still running: its credential is still wanted for another window.
@@ -1484,6 +1509,7 @@ export class AppsRuntime {
14841509 continue;
14851510 }
14861511 state.live = undefined;
1512 this.resting(state, now);
14871513 state.asleep = {
14881514 commitOid: live.commitOid,
14891515 dir: live.dir,
@@ -1503,6 +1529,88 @@ export class AppsRuntime {
15031529 revokeAppToken(this.app.db, live.tokenId);
15041530 await live.handle.stop();
15051531 }
1532 await this.sampleStorage();
1533 }
1534
1535 // ---- the meter ----
1536
1537 /**
1538 * Run one write to the meter. What an app used is worth recording, never
1539 * worth taking a deploy, a sleep or a shutdown down with it.
1540 */
1541 private metered<T>(write: () => T): T | undefined {
1542 try {
1543 return write();
1544 } catch (error) {
1545 log.error("meter write failed", { error: String(error) });
1546 return undefined;
1547 }
1548 }
1549
1550 /** The environment began serving: its awake interval opens, if not open. */
1551 private awake(state: Environment, at: string): void {
1552 this.metered(() =>
1553 openAwake(this.app.db, {
1554 repoId: state.app.repoId,
1555 env: state.env,
1556 placement: this.options.placement ?? "owned",
1557 at,
1558 }),
1559 );
1560 }
1561
1562 /** The environment stopped serving: its awake interval closes. */
1563 private resting(state: Environment, at: number = Date.now()): void {
1564 this.metered(() =>
1565 closeAwake(this.app.db, {
1566 repoId: state.app.repoId,
1567 env: state.env,
1568 at: new Date(at).toISOString(),
1569 }),
1570 );
1571 }
1572
1573 private sampling = false;
1574
1575 /**
1576 * Once per `storageEveryMs`: the bytes of each app's `FORGE_APP_DATA`,
1577 * every environment summed, and of its bare repository.
1578 */
1579 private async sampleStorage(): Promise<void> {
1580 if (this.stopped || this.sampling) return;
1581 const every = this.options.storageEveryMs ?? SAMPLE_SPAN_MS;
1582 const last = this.metered(() => lastStorageAt(this.app.db));
1583 if (last !== undefined && Date.now() - Date.parse(last) < every) return;
1584 this.sampling = true;
1585 try {
1586 const at = new Date().toISOString();
1587 for (const record of listApps(this.app.db)) {
1588 const repo = getRepoById(this.app.db, record.repoId);
1589 if (!repo) continue;
1590 let dataBytes = 0;
1591 const root = join(appsRoot(this.app.config.dataDir), record.id);
1592 if (existsSync(root))
1593 for (const name of readdirSync(root))
1594 if (isEnvName(name))
1595 dataBytes += await dirBytes(
1596 envDataDir(this.app.config.dataDir, record.id, name),
1597 );
1598 const repoBytes = await dirBytes(
1599 repoDir(this.app.reposRoot, repo.ownerName, repo.name),
1600 );
1601 if (this.stopped) return;
1602 this.metered(() =>
1603 recordStorage(this.app.db, {
1604 repoId: record.repoId,
1605 dataBytes,
1606 repoBytes,
1607 at,
1608 }),
1609 );
1610 }
1611 } finally {
1612 this.sampling = false;
1613 }
15061614 }
15071615
15081616 async teardown(record: AppRecord, env: EnvName): Promise<void> {
@@ -1522,6 +1630,7 @@ export class AppsRuntime {
15221630 // after its pull closed, checkout, cloned data and all.
15231631 const live = state.live;
15241632 state.live = undefined;
1633 this.resting(state);
15251634 revokeAppToken(this.app.db, live?.tokenId);
15261635 if (live) await live.handle.stop();
15271636 }
@@ -1546,6 +1655,7 @@ export class AppsRuntime {
15461655 await state.chain.catch(() => {});
15471656 const live = state.live;
15481657 state.live = undefined;
1658 this.resting(state);
15491659 state.asleep = undefined;
15501660 revokeAppToken(this.app.db, live?.tokenId);
15511661 if (live) await live.handle.stop();
src/core/meter.ts added +229
@@ -0,0 +1,229 @@
1 /**
2 * The meter — what each app used, in the two quantities a forge charges
3 * for: time awake and bytes kept. The runtime (`apps/runtime.ts`) writes
4 * it as environments start, sleep, wake and stop, and samples storage
5 * hourly; the profit slider (#392) reads it.
6 *
7 * Two ledgers:
8 * - `meter`: one row per awake interval of an environment — from the
9 * instant a process starts serving to the instant it sleeps, stops,
10 * crashes or the forge shuts down. An open interval carries `beat_at`,
11 * the last instant it was seen running; one still open at boot was cut
12 * short by a crash and is closed there, never left running.
13 * - `storage`: one row per app per sample — the bytes of all its
14 * environments' `FORGE_APP_DATA` and of its bare repository.
15 *
16 * Units, stated once:
17 * - awake time is seconds (wall clock, not CPU), summed over environments;
18 * - storage is GB-hours: bytes × hours / 1e9 (decimal GB). A sample stands
19 * for the hour after it, or until the next sample if that comes sooner.
20 *
21 * Pure data: no authz, no events. `repo_id` is text, not a reference, so
22 * removing an app does not erase what it used.
23 */
24
25 import type { Database } from "bun:sqlite";
26 import { defect } from "../kernel/error.ts";
27
28 /** Where an environment ran: hardware the operator owns, or a cloud's. */
29 export type Placement = "owned" | "cloud";
30 export const PLACEMENTS: readonly Placement[] = ["owned", "cloud"];
31
32 /** How long one storage sample stands for, in milliseconds. */
33 export const SAMPLE_SPAN_MS = 3_600_000;
34
35 const iso = (ms: number): string => new Date(ms).toISOString();
36
37 /** An instant as milliseconds, or a defect: callers pass `toISOString()`. */
38 function instant(at: string): number {
39 const ms = Date.parse(at);
40 if (!Number.isFinite(ms)) defect("meter.bad_instant", `not an instant: ${at}`);
41 return ms;
42 }
43
44 // ---- writing ----
45
46 /**
47 * An environment began serving. Idempotent: while an interval is open for
48 * the environment (a redeploy swapping one process for the next) nothing
49 * new is opened — the app never stopped being awake.
50 */
51 export function openAwake(
52 db: Database,
53 input: { repoId: string; env: string; placement: Placement; at: string },
54 ): void {
55 const at = iso(instant(input.at));
56 db.query(
57 `INSERT OR IGNORE INTO meter (repo_id, env, placement, started_at, beat_at)
58 VALUES (?1, ?2, ?3, ?4, ?4)`,
59 ).run(input.repoId, input.env, input.placement, at);
60 }
61
62 /** An environment stopped serving: close its open interval, if any. */
63 export function closeAwake(
64 db: Database,
65 input: { repoId: string; env: string; at: string },
66 ): void {
67 const at = iso(instant(input.at));
68 db.query(
69 `UPDATE meter SET ended_at = max(?3, started_at), beat_at = max(?3, started_at)
70 WHERE repo_id = ?1 AND env = ?2 AND ended_at IS NULL`,
71 ).run(input.repoId, input.env, at);
72 }
73
74 /** The forge is shutting down: every open interval ends now. */
75 export function closeAllAwake(db: Database, at: string): void {
76 const now = iso(instant(at));
77 db.query(
78 `UPDATE meter SET ended_at = max(?1, started_at), beat_at = max(?1, started_at)
79 WHERE ended_at IS NULL`,
80 ).run(now);
81 }
82
83 /** Every open interval was seen running at `at`. */
84 export function beatAwake(db: Database, at: string): void {
85 const now = iso(instant(at));
86 db.query(
87 `UPDATE meter SET beat_at = max(?1, beat_at) WHERE ended_at IS NULL`,
88 ).run(now);
89 }
90
91 /**
92 * At boot: an interval still open belongs to a forge that died without
93 * shutting down. It ends at its last heartbeat — the last instant anyone
94 * knew it ran. Returns how many were closed.
95 */
96 export function closeOrphans(db: Database): number {
97 return db
98 .query(`UPDATE meter SET ended_at = beat_at WHERE ended_at IS NULL`)
99 .run().changes;
100 }
101
102 /** One storage sample of one app, in bytes. */
103 export function recordStorage(
104 db: Database,
105 input: { repoId: string; dataBytes: number; repoBytes: number; at: string },
106 ): void {
107 const bytes = (n: number): number =>
108 Number.isFinite(n) && n > 0 ? Math.round(n) : 0;
109 db.query(
110 `INSERT INTO storage (repo_id, data_bytes, repo_bytes, at)
111 VALUES (?1, ?2, ?3, ?4)`,
112 ).run(
113 input.repoId,
114 bytes(input.dataBytes),
115 bytes(input.repoBytes),
116 iso(instant(input.at)),
117 );
118 }
119
120 /** When storage was last sampled, if ever. */
121 export function lastStorageAt(db: Database): string | undefined {
122 const row = db
123 .query<{ at: string | null }, []>(`SELECT max(at) AS at FROM storage`)
124 .get();
125 return row?.at ?? undefined;
126 }
127
128 // ---- reading ----
129
130 export interface AppUsage {
131 readonly repoId: string;
132 /** Seconds awake in the range, every environment summed. */
133 readonly awakeSeconds: number;
134 /** The same seconds, by where they ran. */
135 readonly awakeByPlacement: Readonly<Record<Placement, number>>;
136 /** (data + repository bytes) × hours in the range / 1e9. */
137 readonly gbHours: number;
138 }
139
140 interface IntervalRow {
141 repo_id: string;
142 placement: Placement;
143 started_at: string;
144 ended_at: string | null;
145 }
146
147 interface SampleRow {
148 repo_id: string;
149 data_bytes: number;
150 repo_bytes: number;
151 at: string;
152 }
153
154 /**
155 * What each app used in [from, to): awake seconds and GB-hours. An
156 * interval still open counts up to `now` (default: the clock), never past
157 * it. Apps with nothing in the range are absent. Ordered by repo id.
158 */
159 export function usageBetween(
160 db: Database,
161 from: string,
162 to: string,
163 now: string = new Date().toISOString(),
164 ): AppUsage[] {
165 const lo = instant(from);
166 const hi = instant(to);
167 const clock = instant(now);
168 if (!(hi > lo)) return [];
169 const apps = new Map<
170 string,
171 { owned: number; cloud: number; gbHours: number }
172 >();
173 const entry = (repoId: string) => {
174 let e = apps.get(repoId);
175 if (e === undefined) {
176 e = { owned: 0, cloud: 0, gbHours: 0 };
177 apps.set(repoId, e);
178 }
179 return e;
180 };
181
182 const intervals = db
183 .query<IntervalRow, [string, string]>(
184 `SELECT repo_id, placement, started_at, ended_at FROM meter
185 WHERE started_at < ?2 AND (ended_at IS NULL OR ended_at > ?1)`,
186 )
187 .all(iso(lo), iso(hi));
188 for (const row of intervals) {
189 const start = Math.max(lo, instant(row.started_at));
190 const end = Math.min(
191 hi,
192 row.ended_at === null ? clock : instant(row.ended_at),
193 );
194 if (end > start) entry(row.repo_id)[row.placement] += (end - start) / 1000;
195 }
196
197 // A sample before `from` may still stand for part of the range.
198 const samples = db
199 .query<SampleRow, [string, string]>(
200 `SELECT repo_id, data_bytes, repo_bytes, at FROM storage
201 WHERE at >= ?1 AND at < ?2 ORDER BY repo_id, at`,
202 )
203 .all(iso(lo - SAMPLE_SPAN_MS), iso(hi));
204 samples.forEach((row, i) => {
205 const at = instant(row.at);
206 const next = samples[i + 1];
207 const until = Math.min(
208 at + SAMPLE_SPAN_MS,
209 next !== undefined && next.repo_id === row.repo_id
210 ? instant(next.at)
211 : Infinity,
212 );
213 const start = Math.max(lo, at);
214 const end = Math.min(hi, until);
215 if (end <= start) return;
216 const hours = (end - start) / SAMPLE_SPAN_MS;
217 entry(row.repo_id).gbHours +=
218 ((row.data_bytes + row.repo_bytes) * hours) / 1e9;
219 });
220
221 return [...apps.entries()]
222 .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0))
223 .map(([repoId, e]) => ({
224 repoId,
225 awakeSeconds: e.owned + e.cloud,
226 awakeByPlacement: { owned: e.owned, cloud: e.cloud },
227 gbHours: e.gbHours,
228 }));
229 }
src/core/schema.ts +30
@@ -1081,4 +1081,34 @@ export const migrations: readonly string[] = [
10811081 PRIMARY KEY (operation_id, position)
10821082 );
10831083 `,
1084
1085 // 41 — the meter (`core/meter.ts`): what an app used, in the two
1086 // quantities a forge charges for. `meter` is one row per awake interval
1087 // of an environment, `ended` NULL while it runs and `beat_at` the last
1088 // instant it was seen running, so a crash is closed there at boot.
1089 // `storage` is one row per hourly sample of an app's bytes. `repo_id` is
1090 // text, not a reference: removing an app does not erase what it used.
1091 `
1092 CREATE TABLE meter (
1093 id INTEGER PRIMARY KEY,
1094 repo_id TEXT NOT NULL,
1095 env TEXT NOT NULL,
1096 placement TEXT NOT NULL CHECK (placement IN ('owned','cloud')),
1097 started_at TEXT NOT NULL,
1098 beat_at TEXT NOT NULL,
1099 ended_at TEXT
1100 );
1101 CREATE INDEX meter_repo ON meter(repo_id, started_at);
1102 CREATE UNIQUE INDEX meter_open ON meter(repo_id, env) WHERE ended_at IS NULL;
1103
1104 CREATE TABLE storage (
1105 id INTEGER PRIMARY KEY,
1106 repo_id TEXT NOT NULL,
1107 data_bytes INTEGER NOT NULL CHECK (data_bytes >= 0),
1108 repo_bytes INTEGER NOT NULL CHECK (repo_bytes >= 0),
1109 at TEXT NOT NULL
1110 );
1111 CREATE INDEX storage_repo ON storage(repo_id, at);
1112 CREATE INDEX storage_at ON storage(at);
1113 `,
10841114 ];
test/meter.test.ts added +262
@@ -0,0 +1,262 @@
1 /**
2 * The meter: awake intervals and storage samples, and the two sums read
3 * from them — awake seconds and GB-hours — checked against a hand count.
4 */
5 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
6 import { mkdir, mkdtemp, rm, symlink, writeFile } from "node:fs/promises";
7 import { tmpdir } from "node:os";
8 import { join } from "node:path";
9 import { createApp, type App } from "../src/app.ts";
10 import { newApp } from "../src/actions/apps.ts";
11 import { dirBytes } from "../src/apps/bytes.ts";
12 import { AppsRuntime } from "../src/apps/runtime.ts";
13 import type { AppRecord } from "../src/core/apps.ts";
14 import {
15 beatAwake,
16 closeAllAwake,
17 closeAwake,
18 openAwake,
19 recordStorage,
20 usageBetween,
21 } from "../src/core/meter.ts";
22 import { getRepoById } from "../src/core/repos.ts";
23 import { listRuns } from "../src/core/runs.ts";
24 import { createUser } from "../src/core/users.ts";
25 import { repoDir } from "../src/git/repos.ts";
26 import { serve, type Servers } from "../src/http/server.ts";
27
28 const T = (hhmm: string): string => `2026-10-01T${hhmm}:00.000Z`;
29
30 interface MeterRow {
31 repo_id: string;
32 env: string;
33 placement: string;
34 started_at: string;
35 beat_at: string;
36 ended_at: string | null;
37 }
38 const usageOf = (
39 app: App,
40 repoId: string,
41 from: string,
42 to: string,
43 now?: string,
44 ) => usageBetween(app.db, from, to, now).find((u) => u.repoId === repoId);
45 const rows = (app: App, repoId: string): MeterRow[] =>
46 app.db
47 .query<MeterRow, [string]>(
48 "SELECT * FROM meter WHERE repo_id = ? ORDER BY id",
49 )
50 .all(repoId);
51
52 describe("the meter ledger", () => {
53 let dataDir: string;
54 let app: App;
55
56 beforeAll(async () => {
57 dataDir = await mkdtemp(join(tmpdir(), "forge-meter-"));
58 app = createApp({ dataDir, port: 0 });
59 });
60 afterAll(async () => {
61 app.close();
62 await rm(dataDir, { recursive: true, force: true });
63 });
64
65 test("start, sleep, wake, stop: the intervals sum to a hand count", () => {
66 const at = (env: string, placement: "owned" | "cloud", t: string) =>
67 openAwake(app.db, { repoId: "r1", env, placement, at: T(t) });
68 at("production", "owned", "10:00"); // start
69 at("production", "owned", "10:05"); // a redeploy: still the same interval
70 closeAwake(app.db, { repoId: "r1", env: "production", at: T("10:30") }); // sleep
71 at("production", "owned", "11:00"); // wake
72 closeAwake(app.db, { repoId: "r1", env: "production", at: T("11:15") }); // stop
73 at("pull-1", "cloud", "10:10");
74 closeAwake(app.db, { repoId: "r1", env: "pull-1", at: T("10:20") });
75 closeAwake(app.db, { repoId: "r1", env: "pull-1", at: T("10:40") }); // nothing open
76
77 expect(rows(app, "r1").map((r) => [r.env, r.started_at, r.ended_at])).toEqual([
78 ["production", T("10:00"), T("10:30")],
79 ["production", T("11:00"), T("11:15")],
80 ["pull-1", T("10:10"), T("10:20")],
81 ]);
82 // By hand: 30 + 15 minutes owned, 10 minutes cloud.
83 const whole = usageOf(app, "r1", T("09:00"), T("12:00"), T("12:00"));
84 expect(whole!.awakeByPlacement).toEqual({ owned: 45 * 60, cloud: 10 * 60 });
85 expect(whole!.awakeSeconds).toBe(55 * 60);
86 // [10:15, 11:05): 15 + 5 owned, 5 cloud.
87 const part = usageOf(app, "r1", T("10:15"), T("11:05"), T("12:00"));
88 expect(part!.awakeByPlacement).toEqual({ owned: 20 * 60, cloud: 5 * 60 });
89 // An empty or backwards range is nothing.
90 expect(usageBetween(app.db, T("12:00"), T("12:00"))).toEqual([]);
91 });
92
93 test("an open interval counts up to now, never past it, and shutdown closes it", () => {
94 openAwake(app.db, { repoId: "r2", env: "production", placement: "owned", at: T("08:00") });
95 const open = usageOf(app, "r2", T("00:00"), T("23:00"), T("08:20"));
96 expect(open!.awakeSeconds).toBe(20 * 60);
97 closeAllAwake(app.db, T("08:45"));
98 expect(rows(app, "r2")[0]!.ended_at).toBe(T("08:45"));
99 const closed = usageOf(app, "r2", T("00:00"), T("23:00"), T("23:00"));
100 expect(closed!.awakeSeconds).toBe(45 * 60);
101 });
102
103 test("storage: bytes × hours / 1e9, a sample standing for its hour", () => {
104 // 2 GB for an hour, then 1 GB until the next sample 30 minutes later,
105 // then 4 GB for its full hour.
106 recordStorage(app.db, { repoId: "r3", dataBytes: 1.5e9, repoBytes: 0.5e9, at: T("10:00") });
107 recordStorage(app.db, { repoId: "r3", dataBytes: 1e9, repoBytes: 0, at: T("11:00") });
108 recordStorage(app.db, { repoId: "r3", dataBytes: 3e9, repoBytes: 1e9, at: T("11:30") });
109 const all = usageOf(app, "r3", T("00:00"), T("23:00"), T("23:00"));
110 expect(all!.gbHours).toBeCloseTo(2 + 0.5 + 4, 9);
111 // [10:30, 11:45): 2 GB × 0.5h + 1 GB × 0.5h + 4 GB × 0.25h.
112 const part = usageOf(app, "r3", T("10:30"), T("11:45"), T("23:00"));
113 expect(part!.gbHours).toBeCloseTo(1 + 0.5 + 1, 9);
114 // Nonsense in, zero bytes recorded.
115 recordStorage(app.db, { repoId: "r4", dataBytes: -5, repoBytes: Number.NaN, at: T("10:00") });
116 const none = usageOf(app, "r4", T("00:00"), T("23:00"), T("23:00"));
117 expect(none?.gbHours ?? 0).toBe(0);
118 });
119
120 test("a restart closes the interval the last forge left open, at its last heartbeat", async () => {
121 openAwake(app.db, { repoId: "r5", env: "production", placement: "owned", at: T("09:00") });
122 beatAwake(app.db, T("09:07"));
123 // The forge dies here: no close. A new runtime boots over the same data.
124 const runtime = new AppsRuntime(app, {
125 domain: "apps.test",
126 scheme: "http",
127 forgeUrl: "http://localhost",
128 });
129 await runtime.start();
130 await runtime.stop();
131 const [row] = rows(app, "r5");
132 expect(row!.ended_at).toBe(T("09:07"));
133 const used = usageOf(app, "r5", T("00:00"), T("23:00"));
134 expect(used!.awakeSeconds).toBe(7 * 60);
135 });
136
137 test("a fixture directory's bytes are its files' bytes, links not followed", async () => {
138 const root = join(dataDir, "fixture");
139 await mkdir(join(root, "a", "b"), { recursive: true });
140 await writeFile(join(root, "one.txt"), "x".repeat(1000));
141 await writeFile(join(root, "a", "two.bin"), new Uint8Array(4096));
142 await writeFile(join(root, "a", "b", "three"), "");
143 await writeFile(join(root, "a", "b", "four"), "four");
144 const outside = join(dataDir, "outside");
145 await writeFile(outside, new Uint8Array(1_000_000));
146 await symlink(outside, join(root, "link"));
147 expect(await dirBytes(root)).toBe(1000 + 4096 + 0 + 4);
148 expect(await dirBytes(join(dataDir, "missing"))).toBe(0);
149 });
150 });
151
152 describe("the runtime meters what it runs", () => {
153 let dataDir: string;
154 let app: App;
155 let servers: Servers;
156 let runtime: AppsRuntime;
157 let record: AppRecord;
158 const DOMAIN = "apps.test";
159
160 async function until<T>(read: () => T | undefined, ms = 60_000): Promise<T> {
161 const deadline = Date.now() + ms;
162 while (Date.now() < deadline) {
163 const value = read();
164 if (value !== undefined) return value;
165 await Bun.sleep(50);
166 }
167 throw new Error("did not happen in time");
168 }
169 const production = () =>
170 runtime.status(record).find((e) => e.env === "production");
171
172 beforeAll(async () => {
173 dataDir = await mkdtemp(join(tmpdir(), "forge-meter-rt-"));
174 app = createApp({ dataDir, port: 0 });
175 servers = await serve(app);
176 runtime = new AppsRuntime(app, {
177 domain: DOMAIN,
178 scheme: "http",
179 publicPort: servers.port,
180 forgeUrl: servers.url,
181 sleepAfterMs: { production: 600, preview: 600 },
182 sweepEveryMs: 100,
183 storageEveryMs: 200,
184 });
185 await runtime.start();
186 const created = await createUser(app.db, {
187 username: "alice",
188 email: "[email protected]",
189 password: "a-good-passphrase",
190 adminIfFirst: true,
191 });
192 if (!created.ok) throw new Error(created.error);
193 const made = await newApp(app, created.value, {
194 name: "metered",
195 description: "Counted",
196 visibility: "public",
197 });
198 if (!made.ok) throw new Error(String(made.error));
199 record = made.value.app;
200 }, 120_000);
201
202 afterAll(async () => {
203 await runtime.stop();
204 servers.stop();
205 app.close();
206 await rm(dataDir, { recursive: true, force: true });
207 });
208
209 test("deploy, sleep, wake and stop each open or close an interval", async () => {
210 const run = await until(() =>
211 listRuns(app.db, record.repoId, { kind: "deploy", limit: 5 }).find(
212 (r) => r.state === "ok",
213 ),
214 );
215 // Asleep: the first interval runs from serving to the sweep that slept it.
216 const asleep = await until(() =>
217 production()?.state === "asleep" ? production() : undefined,
218 );
219 let ledger = rows(app, record.repoId);
220 expect(ledger.length).toBe(1);
221 expect(ledger[0]!.placement).toBe("owned");
222 expect(ledger[0]!.started_at).toBe(run.servingAt!);
223 expect(ledger[0]!.ended_at).toBe(asleep.since!);
224
225 // Woken, then stopped: a second interval, opened and closed.
226 const woke = await runtime.wake(record, "production");
227 expect(woke).toBeDefined();
228 const since = production()!.since!;
229 expect(rows(app, record.repoId)[1]!.ended_at).toBeNull();
230 await runtime.stopEnv(record, "production");
231 ledger = rows(app, record.repoId);
232 expect(ledger.length).toBe(2);
233 expect(ledger[1]!.started_at).toBe(since);
234 expect(ledger[1]!.ended_at).not.toBeNull();
235
236 // The sum read back is the sum of the intervals, by hand.
237 const hand = ledger.reduce(
238 (s, r) => s + (Date.parse(r.ended_at!) - Date.parse(r.started_at)) / 1000,
239 0,
240 );
241 const used = usageOf(app, record.repoId, "2000-01-01T00:00:00.000Z", "2100-01-01T00:00:00.000Z");
242 expect(used!.repoId).toBe(record.repoId);
243 expect(used!.awakeSeconds).toBeCloseTo(hand, 6);
244 expect(used!.awakeSeconds).toBeGreaterThan(0);
245 }, 90_000);
246
247 test("storage is sampled: the repository's bytes, as counted on disk", async () => {
248 const sample = await until(() =>
249 app.db
250 .query<{ repo_bytes: number; data_bytes: number }, [string]>(
251 "SELECT repo_bytes, data_bytes FROM storage WHERE repo_id = ? ORDER BY id DESC",
252 )
253 .get(record.repoId) ?? undefined,
254 );
255 expect(sample.repo_bytes).toBeGreaterThan(0);
256 const repo = getRepoById(app.db, record.repoId)!;
257 // The repository does not change between the sample and this count.
258 expect(sample.repo_bytes).toBe(
259 await dirBytes(repoDir(app.reposRoot, repo.ownerName, repo.name)),
260 );
261 }, 30_000);
262 });
test/shards.test.ts +40
@@ -11,11 +11,13 @@ import {
1111 formatTimes,
1212 parseTimes,
1313 shardCount,
14 splitArgs,
1415 splitFiles,
1516 tally,
1617 weigher,
1718 type ShardResult,
1819 } from "../scripts/shards.ts";
20 import { parseCheck } from "../src/apps/manifest.ts";
1921
2022 const files = Array.from({ length: 23 }, (_, i) => `test/f${i}.test.ts`);
2123
@@ -271,3 +273,41 @@ describe("tally", () => {
271273 );
272274 });
273275 });
276
277 describe("splitArgs", () => {
278 test("after --, a bare name is a file under test/", () => {
279 expect(
280 splitArgs(["--timeout", "60000", "test/x.test.ts", "--", "crew", "fork-flow"]),
281 ).toEqual({
282 ok: true,
283 flags: ["--timeout", "60000"],
284 files: ["test/x.test.ts", "test/crew.test.ts", "test/fork-flow.test.ts"],
285 });
286 expect(splitArgs(["--bail", "test/a.test.ts"])).toEqual({
287 ok: true,
288 flags: ["--bail"],
289 files: ["test/a.test.ts"],
290 });
291 });
292
293 test("a name that could leave test/ or is not a name is refused", () => {
294 for (const name of ["../src/x", "a/b", "", "-x", "Crew", "a.test.ts"])
295 expect(splitArgs(["--", name]).ok).toBe(false);
296 });
297
298 test("forge.json's check fits the manifest and names only files that exist", async () => {
299 const text = readFileSync(new URL("../forge.json", import.meta.url), "utf8");
300 // The forge's own reading of the manifest, which is what refused it.
301 const check = parseCheck(text, undefined);
302 expect(check.ok).toBe(true);
303 if (!check.ok || check.value === undefined) throw new Error("no check");
304 const run = check.value.run;
305 const args = run.slice(run.indexOf("scripts/test-shards.ts ") + 23).split(" ");
306 const split = splitArgs(args);
307 expect(split.ok).toBe(true);
308 if (!split.ok) return;
309 expect(split.files.length).toBeGreaterThan(0);
310 for (const file of split.files)
311 expect(await Bun.file(new URL(`../${file}`, import.meta.url)).exists()).toBe(true);
312 });
313 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs