cbcdf39 · 549655e
@@ -0,0 +1,175 @@
/**
 * Usage counts: requests by the hour, counted in memory and written in bulk.
 *
 * `count` is one map increment and runs no query. `flush` writes every
 * pending key in one transaction as an upsert that adds to `n`. It runs by
 * itself about a second after the first pending count, once 500 keys are
 * pending (deferred, never inside `count`) and on `stop()`. The timer is
 * `unref`'d so it never holds a process or a test open.
 *
 * A failed write loses nothing: the counts go back and the timer retries.
 * Only an explicit `flush()` or `stop()` sees the error; the timer swallows it.
 * `count` throws RangeError on a malformed event (a programming error), so a
 * caller on the request path must wrap it in try/catch.
 */
import type { Database } from "bun:sqlite";
export const ROLES = ["anon", "user", "admin", "agent"] as const;
export const SURFACES = ["api", "mcp", "web"] as const;
export const METHODS = [
  "GET",
  "HEAD",
  "POST",
  "PUT",
  "PATCH",
  "DELETE",
  "OPTIONS",
] as const;
export type UsageRole = (typeof ROLES)[number];
export type UsageSurface = (typeof SURFACES)[number];
export interface UsageEvent {
  readonly pattern: string;
  readonly method: string;
  readonly status: number;
  readonly role: string;
  readonly surface: string;
}
export interface UsageRow {
  readonly hour: string;
  readonly pattern: string;
  readonly method: string;
  readonly status: string;
  readonly role: string;
  readonly surface: string;
  readonly n: number;
}
export interface UsageCounter {
  count(event: UsageEvent): void;
  flush(): void;
  stop(): void;
}
const FLUSH_MS = 1000;
const FLUSH_KEYS = 500;
const MAX_PATTERN = 200;
const SEP = "\u0000";
/** `2026-10-16T13`: the hour a moment falls in, UTC. */
export function hourOf(at: Date): string {
  return at.toISOString().slice(0, 13);
}
function refuse(event: UsageEvent): string | null {
  if (!(ROLES as readonly string[]).includes(event.role)) return "role";
  if (!(SURFACES as readonly string[]).includes(event.surface))
    return "surface";
  if (!(METHODS as readonly string[]).includes(event.method)) return "method";
  if (
    !Number.isInteger(event.status) ||
    event.status < 200 ||
    event.status > 599
  )
    return "status";
  if (
    typeof event.pattern !== "string" ||
    event.pattern.length === 0 ||
    event.pattern.length > MAX_PATTERN
  )
    return "pattern";
  return null;
}
export function createUsageCounter(
  db: Database,
  now: () => Date = () => new Date(),
  flushMs: number = FLUSH_MS,
): UsageCounter {
  const pending = new Map<string, number>();
  let timer: ReturnType<typeof setTimeout> | null = null;
  let stopped = false;
  function clear(): void {
    if (timer) clearTimeout(timer);
    timer = null;
  }
  function schedule(delay: number): void {
    clear();
    timer = setTimeout(() => {
      try {
        flush();
      } catch {
        // flush already put the counts back and rearmed the retry.
      }
    }, delay);
    timer.unref();
  }
  function flush(): void {
    clear();
    if (pending.size === 0) return;
    const batch = [...pending];
    pending.clear();
    try {
      const upsert = db.query(
        `INSERT INTO usage_count (hour, pattern, method, status, role, surface, n)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
         ON CONFLICT (hour, pattern, method, status, role, surface)
         DO UPDATE SET n = n + excluded.n`,
      );
      db.transaction(() => {
        for (const [key, n] of batch) {
      
        junipercrew
        29d ago
        
      
      A permanently failing write retries every second forever with no signal; when this is wired in, surface it once (a log line) rather than swallowing every attempt.
    
          const [hour, pattern, method, status, role, surface] = key.split(
            SEP,
          ) as [string, string, string, string, string, string];
          upsert.run(hour, pattern, method, status, role, surface, n);
        }
      })();
    } catch (err) {
      // Put the counts back so a failed write loses nothing.
      for (const [key, n] of batch)
        pending.set(key, (pending.get(key) ?? 0) + n);
      if (!stopped) schedule(Math.max(flushMs, 1));
      throw err;
    }
  }
  function count(event: UsageEvent): void {
    const bad = refuse(event);
    if (bad) throw new RangeError(`usage count refused: bad ${bad}`);
    if (stopped) return;
    const key = [
      hourOf(now()),
      event.pattern,
      event.method,
      `${Math.floor(event.status / 100)}xx`,
      event.role,
      event.surface,
    ].join(SEP);
    pending.set(key, (pending.get(key) ?? 0) + 1);
    if (pending.size >= FLUSH_KEYS) schedule(0);
    else if (timer === null) schedule(flushMs);
  }
  function stop(): void {
    stopped = true;
    flush();
  }
  return { count, flush, stop };
}
/** Every row from `sinceHour` (`2026-10-16T13`) on, oldest hour first. */
export function readUsage(db: Database, sinceHour: string): UsageRow[] {
  return db
    .query(
      `SELECT hour, pattern, method, status, role, surface, n FROM usage_count
       WHERE hour >= ?1 ORDER BY hour, pattern, method, status, role, surface`,
    )
    .all(sinceHour) as UsageRow[];
}
@@ -0,0 +1,102 @@
import { expect, test } from "bun:test";
import { Database } from "bun:sqlite";
import migration from "../src/core/migrations/20261016130000-usage-count.ts";
import { createUsageCounter, readUsage } from "../src/core/usage_count.ts";
const ev = {
  pattern: "/api/issues",
  method: "GET",
  status: 200,
  role: "user",
  surface: "api",
};
function fresh() {
  const db = new Database(":memory:");
  db.run(migration);
  return db;
}
test("three counts of one key are one row of 3 after a flush, none before", () => {
  const db = fresh();
  const c = createUsageCounter(db);
  c.count(ev);
  c.count({ ...ev, status: 204 });
  c.count(ev);
  expect(readUsage(db, "")).toEqual([]);
  c.flush();
  const rows = readUsage(db, "");
  expect(rows).toHaveLength(1);
  expect(rows[0]).toMatchObject({ n: 3, status: "2xx" });
  c.count(ev);
  c.stop();
  expect(readUsage(db, "")[0]!.n).toBe(4);
});
test("two hours make two rows", () => {
  const db = fresh();
  let t = new Date("2026-10-16T10:59:00Z");
  const c = createUsageCounter(db, () => t);
  c.count(ev);
  t = new Date("2026-10-16T11:00:00Z");
  c.count(ev);
  c.flush();
  expect(readUsage(db, "").map((r) => r.hour)).toEqual([
    "2026-10-16T10",
    "2026-10-16T11",
  ]);
  expect(readUsage(db, "2026-10-16T11")).toHaveLength(1);
});
test("a refused event writes nothing", () => {
  const db = fresh();
  const c = createUsageCounter(db);
  for (const bad of [
    { role: "root" },
    { surface: "ssh" },
    { method: "BREW" },
    { status: 199 },
    { status: 600 },
    { pattern: "" },
    { pattern: "x".repeat(201) },
  ])
    expect(() => c.count({ ...ev, ...bad })).toThrow();
  c.flush();
  expect(readUsage(db, "")).toEqual([]);
});
test("it flushes by itself after its delay", async () => {
  const db = fresh();
  const c = createUsageCounter(db, () => new Date(), 30);
  c.count(ev);
  expect(readUsage(db, "")).toEqual([]);
  await Bun.sleep(80);
  expect(readUsage(db, "")[0]!.n).toBe(1);
  c.stop();
});
test("500 pending keys flush without count running a query", async () => {
  const db = fresh();
  const c = createUsageCounter(db);
  for (let i = 0; i < 500; i++) c.count({ ...ev, pattern: `/p/${i}` });
  expect(readUsage(db, "")).toEqual([]);
  await Bun.sleep(20);
  expect(readUsage(db, "")).toHaveLength(500);
  c.stop();
});
test("a failed write keeps the counts, retries by itself and never throws from the timer", async () => {
  const db = fresh();
  const c = createUsageCounter(db, () => new Date(), 20);
  db.run("DROP TABLE usage_count");
  c.count(ev);
  c.count(ev);
  await Bun.sleep(60); // timer fires and fails, then retries; no uncaught throw
  expect(() => c.flush()).toThrow();
  db.run(migration);
  await Bun.sleep(60); // the rearmed timer writes the kept counts
  expect(readUsage(db, "")[0]!.n).toBe(2);
  c.count(ev);
  c.stop();
  expect(readUsage(db, "")[0]!.n).toBe(3);
});
		@@ -0,0 +1,175 @@
	1	/**
	2	* Usage counts: requests by the hour, counted in memory and written in bulk.
	3	*
	4	* `count` is one map increment and runs no query. `flush` writes every
	5	* pending key in one transaction as an upsert that adds to `n`. It runs by
	6	* itself about a second after the first pending count, once 500 keys are
	7	* pending (deferred, never inside `count`) and on `stop()`. The timer is
	8	* `unref`'d so it never holds a process or a test open.
	9	*
	10	* A failed write loses nothing: the counts go back and the timer retries.
	11	* Only an explicit `flush()` or `stop()` sees the error; the timer swallows it.
	12	* `count` throws RangeError on a malformed event (a programming error), so a
	13	* caller on the request path must wrap it in try/catch.
	14	*/
	15
	16	import type { Database } from "bun:sqlite";
	17
	18	export const ROLES = ["anon", "user", "admin", "agent"] as const;
	19	export const SURFACES = ["api", "mcp", "web"] as const;
	20	export const METHODS = [
	21	"GET",
	22	"HEAD",
	23	"POST",
	24	"PUT",
	25	"PATCH",
	26	"DELETE",
	27	"OPTIONS",
	28	] as const;
	29
	30	export type UsageRole = (typeof ROLES)[number];
	31	export type UsageSurface = (typeof SURFACES)[number];
	32
	33	export interface UsageEvent {
	34	readonly pattern: string;
	35	readonly method: string;
	36	readonly status: number;
	37	readonly role: string;
	38	readonly surface: string;
	39	}
	40
	41	export interface UsageRow {
	42	readonly hour: string;
	43	readonly pattern: string;
	44	readonly method: string;
	45	readonly status: string;
	46	readonly role: string;
	47	readonly surface: string;
	48	readonly n: number;
	49	}
	50
	51	export interface UsageCounter {
	52	count(event: UsageEvent): void;
	53	flush(): void;
	54	stop(): void;
	55	}
	56
	57	const FLUSH_MS = 1000;
	58	const FLUSH_KEYS = 500;
	59	const MAX_PATTERN = 200;
	60	const SEP = "\u0000";
	61
	62	/** `2026-10-16T13`: the hour a moment falls in, UTC. */
	63	export function hourOf(at: Date): string {
	64	return at.toISOString().slice(0, 13);
	65	}
	66
	67	function refuse(event: UsageEvent): string \| null {
	68	if (!(ROLES as readonly string[]).includes(event.role)) return "role";
	69	if (!(SURFACES as readonly string[]).includes(event.surface))
	70	return "surface";
	71	if (!(METHODS as readonly string[]).includes(event.method)) return "method";
	72	if (
	73	!Number.isInteger(event.status) \|\|
	74	event.status < 200 \|\|
	75	event.status > 599
	76	)
	77	return "status";
	78	if (
	79	typeof event.pattern !== "string" \|\|
	80	event.pattern.length === 0 \|\|
	81	event.pattern.length > MAX_PATTERN
	82	)
	83	return "pattern";
	84	return null;
	85	}
	86
	87	export function createUsageCounter(
	88	db: Database,
	89	now: () => Date = () => new Date(),
	90	flushMs: number = FLUSH_MS,
	91	): UsageCounter {
	92	const pending = new Map<string, number>();
	93	let timer: ReturnType<typeof setTimeout> \| null = null;
	94	let stopped = false;
	95
	96	function clear(): void {
	97	if (timer) clearTimeout(timer);
	98	timer = null;
	99	}
	100
	101	function schedule(delay: number): void {
	102	clear();
	103	timer = setTimeout(() => {
	104	try {
	105	flush();
	106	} catch {
	107	// flush already put the counts back and rearmed the retry.
	108	}
	109	}, delay);
	110	timer.unref();
	111	}
	112
	113	function flush(): void {
	114	clear();
	115	if (pending.size === 0) return;
	116	const batch = [...pending];
	117	pending.clear();
	118	try {
	119	const upsert = db.query(
	120	`INSERT INTO usage_count (hour, pattern, method, status, role, surface, n)
	121	VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
	122	ON CONFLICT (hour, pattern, method, status, role, surface)
	123	DO UPDATE SET n = n + excluded.n`,
	124	);
	125	db.transaction(() => {
	126	for (const [key, n] of batch) {
		junipercrew 29d ago A permanently failing write retries every second forever with no signal; when this is wired in, surface it once (a log line) rather than swallowing every attempt.
	127	const [hour, pattern, method, status, role, surface] = key.split(
	128	SEP,
	129	) as [string, string, string, string, string, string];
	130	upsert.run(hour, pattern, method, status, role, surface, n);
	131	}
	132	})();
	133	} catch (err) {
	134	// Put the counts back so a failed write loses nothing.
	135	for (const [key, n] of batch)
	136	pending.set(key, (pending.get(key) ?? 0) + n);
	137	if (!stopped) schedule(Math.max(flushMs, 1));
	138	throw err;
	139	}
	140	}
	141
	142	function count(event: UsageEvent): void {
	143	const bad = refuse(event);
	144	if (bad) throw new RangeError(`usage count refused: bad ${bad}`);
	145	if (stopped) return;
	146	const key = [
	147	hourOf(now()),
	148	event.pattern,
	149	event.method,
	150	`${Math.floor(event.status / 100)}xx`,
	151	event.role,
	152	event.surface,
	153	].join(SEP);
	154	pending.set(key, (pending.get(key) ?? 0) + 1);
	155	if (pending.size >= FLUSH_KEYS) schedule(0);
	156	else if (timer === null) schedule(flushMs);
	157	}
	158
	159	function stop(): void {
	160	stopped = true;
	161	flush();
	162	}
	163
	164	return { count, flush, stop };
	165	}
	166
	167	/** Every row from `sinceHour` (`2026-10-16T13`) on, oldest hour first. */
	168	export function readUsage(db: Database, sinceHour: string): UsageRow[] {
	169	return db
	170	.query(
	171	`SELECT hour, pattern, method, status, role, surface, n FROM usage_count
	172	WHERE hour >= ?1 ORDER BY hour, pattern, method, status, role, surface`,
	173	)
	174	.all(sinceHour) as UsageRow[];
	175	}
		@@ -0,0 +1,102 @@
	1	import { expect, test } from "bun:test";
	2	import { Database } from "bun:sqlite";
	3	import migration from "../src/core/migrations/20261016130000-usage-count.ts";
	4	import { createUsageCounter, readUsage } from "../src/core/usage_count.ts";
	5
	6	const ev = {
	7	pattern: "/api/issues",
	8	method: "GET",
	9	status: 200,
	10	role: "user",
	11	surface: "api",
	12	};
	13
	14	function fresh() {
	15	const db = new Database(":memory:");
	16	db.run(migration);
	17	return db;
	18	}
	19
	20	test("three counts of one key are one row of 3 after a flush, none before", () => {
	21	const db = fresh();
	22	const c = createUsageCounter(db);
	23	c.count(ev);
	24	c.count({ ...ev, status: 204 });
	25	c.count(ev);
	26	expect(readUsage(db, "")).toEqual([]);
	27	c.flush();
	28	const rows = readUsage(db, "");
	29	expect(rows).toHaveLength(1);
	30	expect(rows[0]).toMatchObject({ n: 3, status: "2xx" });
	31	c.count(ev);
	32	c.stop();
	33	expect(readUsage(db, "")[0]!.n).toBe(4);
	34	});
	35
	36	test("two hours make two rows", () => {
	37	const db = fresh();
	38	let t = new Date("2026-10-16T10:59:00Z");
	39	const c = createUsageCounter(db, () => t);
	40	c.count(ev);
	41	t = new Date("2026-10-16T11:00:00Z");
	42	c.count(ev);
	43	c.flush();
	44	expect(readUsage(db, "").map((r) => r.hour)).toEqual([
	45	"2026-10-16T10",
	46	"2026-10-16T11",
	47	]);
	48	expect(readUsage(db, "2026-10-16T11")).toHaveLength(1);
	49	});
	50
	51	test("a refused event writes nothing", () => {
	52	const db = fresh();
	53	const c = createUsageCounter(db);
	54	for (const bad of [
	55	{ role: "root" },
	56	{ surface: "ssh" },
	57	{ method: "BREW" },
	58	{ status: 199 },
	59	{ status: 600 },
	60	{ pattern: "" },
	61	{ pattern: "x".repeat(201) },
	62	])
	63	expect(() => c.count({ ...ev, ...bad })).toThrow();
	64	c.flush();
	65	expect(readUsage(db, "")).toEqual([]);
	66	});
	67
	68	test("it flushes by itself after its delay", async () => {
	69	const db = fresh();
	70	const c = createUsageCounter(db, () => new Date(), 30);
	71	c.count(ev);
	72	expect(readUsage(db, "")).toEqual([]);
	73	await Bun.sleep(80);
	74	expect(readUsage(db, "")[0]!.n).toBe(1);
	75	c.stop();
	76	});
	77
	78	test("500 pending keys flush without count running a query", async () => {
	79	const db = fresh();
	80	const c = createUsageCounter(db);
	81	for (let i = 0; i < 500; i++) c.count({ ...ev, pattern: `/p/${i}` });
	82	expect(readUsage(db, "")).toEqual([]);
	83	await Bun.sleep(20);
	84	expect(readUsage(db, "")).toHaveLength(500);
	85	c.stop();
	86	});
	87
	88	test("a failed write keeps the counts, retries by itself and never throws from the timer", async () => {
	89	const db = fresh();
	90	const c = createUsageCounter(db, () => new Date(), 20);
	91	db.run("DROP TABLE usage_count");
	92	c.count(ev);
	93	c.count(ev);
	94	await Bun.sleep(60); // timer fires and fails, then retries; no uncaught throw
	95	expect(() => c.flush()).toThrow();
	96	db.run(migration);
	97	await Bun.sleep(60); // the rearmed timer writes the kept counts
	98	expect(readUsage(db, "")[0]!.n).toBe(2);
	99	c.count(ev);
	100	c.stop();
	101	expect(readUsage(db, "")[0]!.n).toBe(3);
	102	});