From e66954583348a394ccc32882135bf31d6f9de1b3 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Mon, 9 Nov 2026 22:24:13 -0800 Subject: [PATCH 1/2] chore(quest): claim quest/m1/gate-fifo-admission.md Quest-Claim: c303c1a339dd80d9 From cc14fa7381773c0d3c53be4ad595febcf00040f1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 10 Nov 2026 00:28:39 -0800 Subject: [PATCH 2/2] feat(gate): admit waiting commands first come, first served Each waiting caller records its place in the admission ledger and counts every caller queued ahead of it as a recent admission, so a later caller starts first only while the host has room for all of them, and only the head of the queue waits out the spacing. Priority and host callers go ahead of the queue; a quick caller still starts at its deadline. Co-Authored-By: Claude Opus 5.5 --- README.md | 23 ++++-- quest/m1/gate-fifo-admission.md | 44 ----------- src/cli/gate.ts | 130 +++++++++++++++++++++++--------- test/gate.test.ts | 102 ++++++++++++++++++++++++- 4 files changed, 215 insertions(+), 84 deletions(-) delete mode 100644 quest/m1/gate-fifo-admission.md diff --git a/README.md b/README.md index 1ef9d14c..c4c8da22 100644 --- a/README.md +++ b/README.md @@ -1114,6 +1114,14 @@ host has room even counting those, commands start at once: a burst of gated test runs on an idle host starts together. Once only the raw load and memory have room, commands start one at a time, 5 seconds after the previous admission. +Waiting commands start first come, first served. Each records its place in a +queue when it starts waiting, and counts every command queued ahead of it as +one more recent admission: it starts before them only while the host has room +for all of them, and only the first in line waits out the 5-second spacing. A +command's wait is therefore bounded by the work queued ahead of it, never by +commands that started waiting later. A place is freed once its command starts, +its gate is cancelled, or its gate process exits. + The load limit applies only while a gated command is still running. Otherwise the load comes from work the gate cannot wait out, so commands start one at a time as if the raw load had room. A freshly booted hosted macOS runner is one @@ -1128,11 +1136,13 @@ an account missing from it, as in arbitrary-UID containers, uses its UID can write it. Two kinds of command go ahead of the rest. `quest gate --quick ` waits -at most a minute, then starts the command whatever the host's load and memory; +at most a minute, then starts the command whatever the host's load, memory, and +place in the queue; pass it only for cheap commands, such as a formatter or `quest check`, that cost less to run than to hold back. A check that lands PRs goes first: with `QUEST_GATE_PRIORITY=1` in its environment, every gate inside the check waits -for memory but never for load, whether or not other gated commands run. Both +for memory but never for load or the queue, whether or not other gated commands +run. Both record their admissions, so the other callers count them against the host's headroom. Set the variable where the check starts, such as a merge queue build's workflow or the land daemon's sandbox, rather than relying on a slot in @@ -1153,7 +1163,10 @@ where a reused process ID could pass for a holder (see below), refuses Each admission records the gate's process ID and, except on Windows, when the gate started, and counts as running until that process exits; a reused process ID counts as running only on Windows, where it makes a command wait for the -load again. The gate holds no lock while the command runs, and it owns the +load again. A place in the queue records the same, except that on Windows the +waiting gate refreshes it, and it frees a minute after the last refresh, so a +reused process ID cannot hold the queue. The gate holds no lock while the +command runs, and it owns the command's descendants. Once the command exits, any descendant still running is killed, so a build or test run leaves nothing behind. @@ -1215,8 +1228,8 @@ first one as Just's copy, not as a second signal: a SIGTERM to Just's process group reaches the gate both directly and through Just. A repeated SIGTERM still kills, though when sent to Just or the gate alone, or when the two copies arrive together, it takes a third; a repeated Ctrl-C, which Just does not -forward, kills at the second. There is no fairness guarantee, and no timeout except -`--quick`'s: any other command waits as long as the host stays busy. +forward, kills at the second. There is no timeout except `--quick`'s: any other +command waits its turn for as long as the host stays busy. ## Route work by size diff --git a/quest/m1/gate-fifo-admission.md b/quest/m1/gate-fifo-admission.md deleted file mode 100644 index 701d92f0..00000000 --- a/quest/m1/gate-fifo-admission.md +++ /dev/null @@ -1,44 +0,0 @@ -# [M] Gate admission is first come, first served - -## Goal - -`quest gate` admits waiting commands in arrival order, so no caller is passed -by commands that started waiting after it, and its wait is bounded by the work -queued ahead of it. The host's load and memory still decide when the head of -the queue starts. `quick`, `priority`, and `host` keep their meanings. - -Today admission has no queue: every waiter polls the 1-minute load average and -available memory, and whichever sees room first starts. With about 12 agents -gating on one desktop, a diff-scoped `just check` that ran in 84 s waited 77 -minutes to be admitted -([moq-dev/moq.pro#3751](https://github.com/moq-dev/moq.pro/pull/3751)). - -## Plan - -Decided 2026-11-09 (maintainer, `/quest-plan` of moq-dev/moq.pro#3751's -follow-up): - -- Plan it: ✅ in m1, beside [gate-cpu-weight](/quest/m1/gate-cpu-weight.md) - (rejected: m0; drop). -- Mechanism: ✅ FIFO admission: waiters start in arrival order, with no cost - estimate (rejected: shortest job first from measured costs, which needs a - cost model and can starve full gates; a slot reserved for small commands; - leaving the choice to the implementer). - -The admission ledger (`src/cli/gate.ts`) already records each admission under -an atomic versioned publish; the queue can live in the same ledger, recording -each waiter when it starts waiting. A waiter whose gate process died leaves -the queue the way a dead admission leaves the ledger today. `quick` callers -still start after `QUICK_MS` whatever their place, and a `host` reservation -still holds everyone back. A `priority` caller, which today ignores load so it -goes first while the host is loaded, keeps going ahead of the queue rather than -taking a place in it. - -`admit` drops a ledger entry once it is older than `RECENT_MS` and its process -no longer runs; a queued waiter's entry must stay while its gate lives, however -long it waits, and must not count as an admission against anyone's headroom. -`admit`'s doc comment, which today promises no fairness, states the new order. - -## Related - -- [Gate scopes yield the CPU to the interactive session](/quest/m1/gate-cpu-weight.md) - changes where admitted commands run; this changes when they are admitted diff --git a/src/cli/gate.ts b/src/cli/gate.ts index 7020d382..17b8d2c3 100644 --- a/src/cli/gate.ts +++ b/src/cli/gate.ts @@ -16,17 +16,20 @@ import { delegated, nest, questScope, startTime, systemdScopes } from "./supervi * One admitted command: a unique id, the wall-clock time it was admitted, and the process ID of * the gate that runs it, which holds a slot until that command ends, with when that gate started * (see `processStart`), so a reused process ID holds nothing. `host` marks a command that holds the - * whole host (see `Precedence`), from when it starts waiting until it ends. + * whole host (see `Precedence`), from when it starts waiting until it ends. `queued` marks a caller + * still waiting for its turn instead, as of `at`: it holds a place in the queue, the ledger's order + * of queued entries, and no slot. */ export interface Admission { id: string; at: number; pid?: number; host?: boolean; + queued?: boolean; start?: string; } -/** One version of the host's admission ledger. Versions only grow; 0 is the empty ledger. */ +/** One version of the host's admission ledger and queue. Versions only grow; 0 is the empty ledger. */ export interface Ledger { version: number; admissions: Admission[]; @@ -59,7 +62,7 @@ export interface Limits { memory: number; } -export type Waiting = "load" | "memory" | "host" | "commands"; +export type Waiting = "load" | "memory" | "host" | "commands" | "queue"; /** * Admissions that only fit the host's raw load and memory, not its headroom, @@ -256,6 +259,7 @@ function parse(text: string): Admission[] { Number.isFinite(entry?.at) && (entry.pid === undefined || (Number.isSafeInteger(entry.pid) && entry.pid > 0)) && (entry.host === undefined || entry.host === true) && + (entry.queued === undefined || entry.queued === true) && (entry.start === undefined || typeof entry.start === "string"), ); return valid ? (value as Admission[]) : []; @@ -409,14 +413,23 @@ export function fileHost(directory: string, now = Date.now): Host { * one, the load is work the gate cannot wait out, as on a freshly booted macOS * runner, so the raw load counts as having room. * + * Callers are admitted first come, first served: each records its place in the + * queue when it starts waiting, and counts every caller queued ahead of it as + * one more recent admission, so it starts before them only while the host has + * room for them all. Only the head of the queue waits out the spacing. A place + * lasts until its caller is admitted, cancelled, or its gate ends; one whose + * gate's start time is unknown, as on Windows, could pass for a reused process + * ID, so its caller refreshes it, and it frees a minute after the last refresh. + * * A running command that holds the host keeps everyone else waiting; one that - * waits to hold it ignores load and memory, and waits only for the gated - * commands still running to end (see `Precedence`). + * waits to hold it ignores load, memory, and the queue, and waits only for the + * gated commands still running to end. A priority caller goes ahead of the + * queue, and a quick one starts at its deadline whatever its place (see + * `Precedence`). * - * Each caller decides against the newest ledger and publishes its admission - * atomically, so concurrent callers never exceed the rule and one always wins - * while it has room. No fairness promise, and no admission timeout except a - * quick caller's (see `Precedence`). + * Each caller decides against the newest ledger and publishes atomically, so + * concurrent callers never exceed the rule and one always wins while it has + * room. No admission timeout except a quick caller's. */ export async function admit( limits: Limits, @@ -426,18 +439,30 @@ export async function admit( precedence: Precedence = {}, ) { const id = randomUUID(); + // A place in the queue has an id of its own, so publishing the admission adds a new entry. + const place = randomUUID(); // Every admission names its gate's start, so a reused process ID holds no slot, as for a host holder. const self = { pid: process.pid, start: processStart(process.pid) }; const deadline = precedence.quick ? host.now() + QUICK_MS : Infinity; - for (;;) { - signal.throwIfAborted(); + const queues = !precedence.priority && !precedence.host; + + /** One decision against the newest ledger: true once admitted. */ + const turn = async (): Promise => { const ledger = host.ledger(); // Sampled after the read, so only a backward clock step dates an admission in the future. const now = host.now(); - // A running admission stays in the ledger past the minute, so later callers still see its slot. - const running = new Set(ledger.admissions.filter((admission) => host.running(admission))); - const kept = ledger.admissions.filter((admission) => admission.at > now - RECENT_MS || running.has(admission)); - const recent = kept.filter((admission) => admission.at > now - RECENT_MS); + const live = new Set(ledger.admissions.filter((admission) => host.running(admission))); + const kept = ledger.admissions.filter((admission) => + admission.queued + ? // A place whose gate cannot be told from a reused process ID lasts only while refreshed. + admission.id === place || + (live.has(admission) && (admission.start !== undefined || admission.at > now - RECENT_MS)) + : // A running admission stays in the ledger past the minute, so later callers still see its slot. + admission.at > now - RECENT_MS || live.has(admission), + ); + const admissions = kept.filter((admission) => !admission.queued); + const running = new Set(admissions.filter((admission) => live.has(admission))); + const recent = admissions.filter((admission) => admission.at > now - RECENT_MS); if (kept.some((admission) => admission.at > now)) { // Date those now, so they still count and space admissions but expire a minute after // the step. A forward step can only expire admissions up to a minute early. @@ -445,30 +470,42 @@ export async function admit( ledger, kept.map((admission) => ({ ...admission, at: Math.min(admission.at, now) })), ); - continue; + return false; } + const position = kept.findIndex((admission) => admission.id === place); + const mine = kept[position]; + if (queues && (mine === undefined || (mine.start === undefined && mine.at <= now - RECENT_MS / 2))) { + // Take a place at the back, or refresh one that would otherwise expire where it stands. + const entry = { id: place, at: now, ...self, queued: true }; + host.publish( + ledger, + mine ? kept.map((admission) => (admission === mine ? entry : admission)) : [...kept, entry], + ); + return false; + } + const ahead = kept.slice(0, Math.max(position, 0)).filter((admission) => admission.queued).length; + // An admission takes the place of this caller's entry in the queue. + const admitted = (at: number) => + host.publish(ledger, [...kept.filter((admission) => admission.id !== place), { id, at, ...self }]); // A command holding the host keeps every other caller out until it ends, quick or not. if (kept.some((admission) => admission.host && admission.id !== id && running.has(admission))) { waiting?.("host"); await host.sleep(SETTLE_MS, signal); - continue; + return false; } if (precedence.host) { // Reserve the host first, so no other caller is admitted while this one waits for the rest. if (!kept.some((admission) => admission.id === id)) host.publish(ledger, [...kept, { id, at: now, ...self, host: true }]); - else if (!kept.some((admission) => admission.id !== id && running.has(admission))) return; + else if (!kept.some((admission) => admission.id !== id && running.has(admission))) return true; else { waiting?.("commands"); await host.sleep(SETTLE_MS, signal); } - continue; - } - if (now >= deadline) { - // Recorded like any admission, so it still counts against the others' headroom. - if (host.publish(ledger, [...kept, { id, at: now, ...self }])) return; - continue; + return false; } + // Recorded like any admission, so it still counts against the others' headroom. + if (now >= deadline) return admitted(now); // Priority ignores load, whether or not a gated command runs; memory still applies. const load = precedence.priority ? 0 : running.size > 0 ? host.load() : Math.min(host.load(), limits.load); const memory = host.memory(); @@ -476,19 +513,43 @@ export async function admit( if (load > limits.load) waiting?.("load"); else if (memory < limits.memory) waiting?.("memory"); else { - const headroom = - load + recent.length < limits.load && memory - recent.length * limits.memory >= limits.memory; - const last = Math.max(-Infinity, ...recent.map((admission) => admission.at)); - wait = headroom ? 0 : last + SETTLE_MS - now; - if (wait <= 0) { - // A caller descheduled since sampling would record a stale time, which lets the - // next one skip its spacing; decide again once the stall reaches that interval. - const at = host.now(); - if (at - now < SETTLE_MS && host.publish(ledger, [...kept, { id, at, ...self }])) return; - continue; + // Every caller queued ahead takes a slot before this one, so it starts first only while + // the host has room for them all. + const taken = recent.length + ahead; + const headroom = load + taken < limits.load && memory - taken * limits.memory >= limits.memory; + if (!headroom && ahead > 0) waiting?.("queue"); + else { + const last = Math.max(-Infinity, ...recent.map((admission) => admission.at)); + wait = headroom ? 0 : last + SETTLE_MS - now; + if (wait <= 0) { + // A caller descheduled since sampling would record a stale time, which lets the + // next one skip its spacing; decide again once the stall reaches that interval. + const at = host.now(); + return at - now < SETTLE_MS && admitted(at); + } } } await host.sleep(Math.min(wait, deadline - now), signal); + return false; + }; + + try { + for (;;) { + signal.throwIfAborted(); + if (await turn()) return; + } + } finally { + // A cancelled caller leaves the queue at once, rather than once its gate ends. + for (let ledger = host.ledger(); ledger.admissions.some((admission) => admission.id === place);) { + if ( + host.publish( + ledger, + ledger.admissions.filter((admission) => admission.id !== place), + ) + ) + break; + ledger = host.ledger(); + } } } @@ -666,6 +727,7 @@ const notices: Record = { memory: "quest gate: waiting for free memory", host: "quest gate: waiting for a command holding the whole host to end", commands: "quest gate: holding the host, waiting for gated commands to end", + queue: "quest gate: waiting for commands queued earlier to start", }; /** Run from the CLI. */ diff --git a/test/gate.test.ts b/test/gate.test.ts index 7aa7def6..bedc1099 100644 --- a/test/gate.test.ts +++ b/test/gate.test.ts @@ -334,6 +334,106 @@ test("a caller holding the host reserves it, waits out gated commands, and holds expect(waited).toEqual([]); }); +test("waiting callers start in arrival order, a later one first only while the host has room for both", async () => { + const limits = { load: 4, memory: GiB }; + // A build still runs, and an earlier caller waits for its turn. + const build = { id: "build", at: -10 * RECENT_MS, ...me }; + const earlier = { id: "earlier", at: 0, ...me, queued: true }; + const directory = tempDir(); + const state = { now: 0, load: 3, memory: Infinity }; + const events: string[] = []; + const host = testHost(directory, state, events); + expect(host.publish(host.ledger(), [build, earlier])).toBe(true); + const sleep = host.sleep; + host.sleep = async (ms, signal) => { + await sleep(ms, signal); + // The earlier caller takes its turn. + const ledger = host.ledger(); + const rest = ledger.admissions.filter(({ id }) => id !== "earlier"); + if (rest.length < ledger.admissions.length) + expect(host.publish(ledger, [...rest, { id: "started", at: state.now, ...me }])).toBe(true); + }; + // Alone, this caller would start at once; behind the earlier one, it waits, then spaces itself. + await admit(limits, host, signal, (reason) => events.push(reason)); + expect(events).toEqual(["queue", `sleep ${SETTLE_MS}`, `sleep ${SETTLE_MS}`]); + expect(host.ledger().admissions).toEqual([ + build, + { id: "started", at: SETTLE_MS, ...me }, + { id: expect.any(String), at: 2 * SETTLE_MS, ...me }, + ]); + // With room for both, it starts at once and leaves the earlier caller's place. + const roomy = testHost(tempDir(), { now: 0, load: 2, memory: Infinity }, events); + expect(roomy.publish(roomy.ledger(), [build, earlier])).toBe(true); + events.length = 0; + await admit(limits, roomy, signal, (reason) => events.push(reason)); + expect(events).toEqual([]); + expect(roomy.ledger().admissions).toEqual([build, earlier, { id: expect.any(String), at: 0, ...me }]); +}); + +test("a place frees once its gate ends or its caller cancels, or a minute after its last refresh without a start", async () => { + const limits = { load: 4, memory: GiB }; + const build = { id: "build", at: -10 * RECENT_MS, ...me }; + const exited = spawnSync(process.execPath, ["-e", ""]).pid; + const directory = tempDir(); + const state = { now: 0, load: 3, memory: Infinity }; + const host = testHost(directory, state); + expect( + host.publish(host.ledger(), [ + build, + { id: "ended", at: 0, pid: exited, queued: true }, + { id: "reused", at: 0, pid: process.pid, start: "an earlier process", queued: true }, + { id: "unrefreshed", at: -RECENT_MS, pid: process.pid, queued: true }, + ]), + ).toBe(true); + const events: string[] = []; + await admit(limits, testHost(directory, state, events), signal, (reason) => events.push(reason)); + expect(events).toEqual([]); + expect(host.ledger().admissions).toEqual([build, { id: expect.any(String), at: 0, ...me }]); + // A place refreshed within the minute still holds, and a cancelled caller behind it leaves the queue. + const fresh = { id: "fresh", at: -RECENT_MS / 2, pid: process.pid, queued: true }; + const queued = testHost(tempDir(), state); + expect(queued.publish(queued.ledger(), [build, fresh])).toBe(true); + const controller = new AbortController(); + const waited: string[] = []; + queued.sleep = async () => { + expect(queued.ledger().admissions).toEqual([build, fresh, expect.objectContaining({ queued: true, ...me })]); + controller.abort(new Error("cancelled")); + }; + await expect(admit(limits, queued, controller.signal, (reason) => waited.push(reason))).rejects.toThrow( + "cancelled", + ); + expect(waited).toEqual(["queue"]); + expect(queued.ledger().admissions).toEqual([build, fresh]); +}); + +test("priority and host callers go ahead of the queue, and a quick one starts at its deadline whatever its place", async () => { + const limits = { load: 4, memory: GiB }; + const build = { id: "build", at: -10 * RECENT_MS, ...me }; + const earlier = { id: "earlier", at: 0, ...me, queued: true }; + const directory = tempDir(); + const state = { now: 0, load: 3, memory: Infinity }; + const host = testHost(directory, state); + expect(host.publish(host.ledger(), [build, earlier])).toBe(true); + const events: string[] = []; + await admit(limits, testHost(directory, state, events), signal, undefined, { priority: true }); + expect(events).toEqual([]); + await admit(limits, testHost(directory, state, events), signal, undefined, { quick: true }); + expect(events).toEqual(Array.from({ length: QUICK_MS / SETTLE_MS }, () => `sleep ${SETTLE_MS}`)); + expect(host.ledger().admissions).toEqual([ + build, + earlier, + { id: expect.any(String), at: 0, ...me }, + { id: expect.any(String), at: QUICK_MS, ...me }, + ]); + // A waiting caller runs no command, so a host holder starts at once ahead of it. + const held = testHost(tempDir(), { now: 0, load: 99, memory: 0 }, events); + expect(held.publish(held.ledger(), [earlier])).toBe(true); + events.length = 0; + await admit(limits, held, signal, undefined, { host: true }); + expect(events).toEqual([]); + expect(held.ledger().admissions).toEqual([earlier, { id: expect.any(String), at: 0, ...me, host: true }]); +}); + test("a lone quick or priority caller starts at once on load from outside the gate", async () => { // No gated command runs, so the load is work the gate cannot wait out, whatever the precedence. const state = { now: 0, load: 12, memory: Infinity }; @@ -421,7 +521,7 @@ test("a caller that dies mid-admission blocks no one past a minute", async () => await admit({ load: 1, memory: 1 }, testHost(directory, state, events), signal); expect(events).toEqual([]); // The stale draft is gone; one still being written stays. - expect(readdirSync(directory).sort()).toEqual(["2", "live.tmp"]); + expect(readdirSync(directory).sort()).toEqual(["3", "live.tmp"]); // A version cut short, as by power loss, reads as empty rather than wedging every caller. writeFileSync(join(directory, "100"), ""); expect(host.ledger()).toEqual({ version: 100, admissions: [] });