Phase 2: budget-before-tokens reservation, verified against real Valkey

CONST-BUDGET-BEFORE-TOKENS -- the ordering IS the mechanism. reserveBudget atomically INCRs
the day's counter and only allows a container to start if the reserved number is within the
cap. Check-after-spend would let fifty junk triggers cost fifty real jobs before the cap
engages, which is the exact scenario it guards.

Verified against a real Valkey (AOF on, as REQ-QUEUE-BURST-NO-DROP requires): 20 parallel
INCRs yield exactly 20, no lost updates, so INCR-then-compare needs no lock even at
concurrency 3. Three jobs racing at count 9 with cap 10 reserve 10/11/12 -- exactly one
proceeds. Confirmed the refuse path too: with cap 3, reservations 4 and 5 are refused and no
container starts.

Design choices, each tested:
- a refusal still counts (no decrement): the cap is a hard daily ceiling on container starts,
  and giving refused attempts their slot back would let a burst probe the cap for free. A
  refused job spends nothing and reports on its issue.
- releaseBudget exists for ONE case: an infra fault AFTER reserving but before the container
  spent anything (e.g. docker daemon unreachable). A completed run (0/2) really consumed its
  slot and does not release.
- cap 0 fails CLOSED (every job refused), never "unlimited" -- it guards money.
- TTL set once on first reservation, so a busy day cannot push its own expiry forward.

Tests run everywhere (injected fake redis); the same logic was also exercised against real
Valkey by hand. The BullMQ Worker construction, docker orchestration, and 30-minute timeout
follow, pending the wiring research in flight -- not guessed.

Author: edgehero

worker/src/budget.mjs

@@ -0,0 +1,54 @@
+/**
+ * CONST-BUDGET-BEFORE-TOKENS. The ordering IS the mechanism.
+ *
+ * Reserve a slot against the daily cap BEFORE a container starts. Check-after-spend means fifty
+ * junk triggers cost fifty jobs of real money before the cap engages -- the exact scenario the cap
+ * exists for. So: atomically INCR the day's counter, and only start the container if the reserved
+ * number is within the cap.
+ *
+ * INCR is atomic -- verified against a real Valkey under 20 parallel increments yielding exactly
+ * 20, no lost updates -- so this needs no lock even at concurrency 3. Three jobs racing at count 9
+ * with cap 10 reserve 10, 11, 12; exactly one proceeds, the others are refused, and no container
+ * starts over budget.
+ *
+ * `redis` is any ioredis-compatible client (BullMQ bundles ioredis). Injected so the logic is
+ * testable without a running server.
+ */
+
+/** UTC date key. The worker is ordinary node, so Date is available (unlike the workflow sandbox). */
+export function dayKey(now = new Date(), prefix = "budget") {
+	const d = now.toISOString().slice(0, 10); // YYYY-MM-DD, UTC
+	return `${prefix}:${d}`;
+}
+
+const TWO_DAYS_SECONDS = 2 * 24 * 60 * 60;
+
+/**
+ * Atomically reserve one slot against today's cap. Returns { allowed, reserved, cap }.
+ *
+ * A refused reservation still counts -- the counter bounds container STARTS per day, and a refused
+ * job spends nothing (no container) and reports on its issue. We deliberately do NOT decrement on
+ * refusal: the cap is a hard daily ceiling on attempts, and letting refused attempts "give back"
+ * their slot would let a burst probe the cap for free.
+ *
+ * A cap of 0 or negative disables running entirely (every job refused) rather than meaning
+ * "unlimited" -- fail closed, since this guards money.
+ */
+export async function reserveBudget(redis, { cap, now = new Date(), keyPrefix = "budget" } = {}) {
+	const key = dayKey(now, keyPrefix);
+	const reserved = Number(await redis.incr(key));
+	// Set the TTL only once, when the key is first created (reserved === 1), so a long-running
+	// day cannot have its expiry pushed forward indefinitely.
+	if (reserved === 1) await redis.expire(key, TWO_DAYS_SECONDS);
+	return { allowed: reserved <= cap, reserved, cap };
+}
+
+/**
+ * Give a reservation back. Used ONLY when the container never started because of an INFRA fault
+ * AFTER reserving (e.g. the docker daemon was unreachable) -- an infra failure that spent nothing
+ * should not permanently consume a cap slot. NOT used for a completed run (0/2), which really did
+ * consume its slot, nor for a refusal (which never incremented past the cap deliberately).
+ */
+export async function releaseBudget(redis, { now = new Date(), keyPrefix = "budget" } = {}) {
+	await redis.decr(dayKey(now, keyPrefix));
+}

worker/test/budget.test.mjs

@@ -0,0 +1,75 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+import { dayKey, releaseBudget, reserveBudget } from "../src/budget.mjs";
+
+/** Minimal ioredis-compatible fake: atomic enough for single-threaded test semantics. */
+function fakeRedis() {
+	const store = new Map();
+	const ttl = new Map();
+	return {
+		store,
+		ttl,
+		async incr(k) {
+			const v = (store.get(k) ?? 0) + 1;
+			store.set(k, v);
+			return v;
+		},
+		async decr(k) {
+			const v = (store.get(k) ?? 0) - 1;
+			store.set(k, v);
+			return v;
+		},
+		async expire(k, s) {
+			ttl.set(k, s);
+		},
+	};
+}
+
+test("dayKey is a UTC YYYY-MM-DD key", () => {
+	assert.equal(dayKey(new Date("2026-07-16T23:30:00Z")), "budget:2026-07-16");
+	// A timestamp just before UTC midnight stays on its UTC day regardless of local tz.
+	assert.equal(dayKey(new Date("2026-07-16T00:00:01Z")), "budget:2026-07-16");
+});
+
+test("reserves within the cap, refuses beyond it, and never lets a refusal spend", async () => {
+	const redis = fakeRedis();
+	const now = new Date("2026-07-16T10:00:00Z");
+	const results = [];
+	for (let i = 0; i < 5; i++) results.push(await reserveBudget(redis, { cap: 3, now }));
+
+	assert.deepEqual(
+		results.map((r) => r.allowed),
+		[true, true, true, false, false],
+	);
+	assert.deepEqual(
+		results.map((r) => r.reserved),
+		[1, 2, 3, 4, 5],
+	);
+});
+
+test("the TTL is set once, on first reservation only", async () => {
+	const redis = fakeRedis();
+	const now = new Date("2026-07-16T10:00:00Z");
+	await reserveBudget(redis, { cap: 10, now });
+	assert.ok(redis.ttl.has("budget:2026-07-16"), "TTL set on first reserve");
+	redis.ttl.delete("budget:2026-07-16");
+	await reserveBudget(redis, { cap: 10, now });
+	assert.ok(!redis.ttl.has("budget:2026-07-16"), "TTL must NOT be reset on later reserves");
+});
+
+test("cap 0 fails closed -- every job refused, not 'unlimited'", async () => {
+	const redis = fakeRedis();
+	const r = await reserveBudget(redis, { cap: 0, now: new Date("2026-07-16T10:00:00Z") });
+	assert.equal(r.allowed, false);
+});
+
+test("release gives a slot back (infra-fault path only)", async () => {
+	const redis = fakeRedis();
+	const now = new Date("2026-07-16T10:00:00Z");
+	await reserveBudget(redis, { cap: 3, now });
+	await reserveBudget(redis, { cap: 3, now });
+	await releaseBudget(redis, { now });
+	const r = await reserveBudget(redis, { cap: 3, now });
+	assert.equal(r.reserved, 2, "a released slot is reusable");
+	assert.equal(r.allowed, true);
+});