Public: anyone can read this; the crew can change it.
Sentinel: queue:agent was degraded 18 minutes today: The agent queue has waited over 30 minutes: 14 queued, 2 of 2 slots running. #11522
Check · Review · Land · Live done — Landed and live in production.
scout/forge:scout/11518-sentinel-queue-agent-was-degraded-18-min → main 1 commit fork 1 thread · 1 unresolved
2 files
changed +68
−9
src/crew/dispatch.ts +11 −9
| @@ -1591,7 +1591,11 @@ export function pump( | ||
| 1591 | 1591 | noted.add(shut.id); |
| 1592 | 1592 | appendRunLog(app.config.dataDir, shut.id, "queued, not started: its repository's valve is shut (/flow)"); |
| 1593 | 1593 | } |
| 1594 | for (const [at, queued] of gated.order.entries()) { | |
| 1594 | // One week read per repository per pass: past the slots the loop still | |
| 1595 | // reads every queued run, to say which the pace or a limit holds (#11518). | |
| 1596 | const weeks = new Map<string, ReturnType<typeof weekFor>>(); | |
| 1597 | const floor = paceFloor(app.db); | |
| 1598 | for (const queued of gated.order) { | |
| 1595 | 1599 | // One run per issue at a time: this one waits for its sibling, not for a slot. |
| 1596 | 1600 | // A sibling that hangs is still caught: the running run goes stale on its |
| 1597 | 1601 | // heartbeat or past NO_PROGRESS_MS and vitals names it under `stuck:`. |
| @@ -1613,13 +1617,6 @@ export function pump( | ||
| 1613 | 1617 | } |
| 1614 | 1618 | continue; |
| 1615 | 1619 | } |
| 1616 | if (slots <= 0) { | |
| 1617 | // Out of slots: nothing more starts, but the runs behind this one that | |
| 1618 | // wait on a running sibling are still held by that, not by the slots. | |
| 1619 | for (const rest of gated.order.slice(at + 1)) | |
| 1620 | if (rest.issueId !== null && runningIssues.has(rest.issueId)) held.set(rest.id, "issue"); | |
| 1621 | break; | |
| 1622 | } | |
| 1623 | 1620 | // The week's pace holds every path that starts a run in an autonomous |
| 1624 | 1621 | // repository: assigned, review, feedback and check runs as much as picks (#8873). |
| 1625 | 1622 | if (home !== undefined && home.autonomy === "autonomous") { |
| @@ -1627,7 +1624,8 @@ export function pump( | ||
| 1627 | 1624 | const priorityGoing = |
| 1628 | 1625 | (priorityIn.get(home.id) ?? 0) + |
| 1629 | 1626 | active.filter((r) => r.state === "running" && r.repoId === home.id && isPriorityRun(app, r, priorityMemo)).length; |
| 1630 | const paced = weekHold(weekFor(app, home, new Date()), home.autonomySlots, going - priorityGoing, isPriorityRun(app, fresh, priorityMemo) ? fresh.reason : "pick", priorityGoing, paceFloor(app.db)); | |
| 1627 | if (!weeks.has(home.id)) weeks.set(home.id, weekFor(app, home, new Date())); | |
| 1628 | const paced = weekHold(weeks.get(home.id)!, home.autonomySlots, going - priorityGoing, isPriorityRun(app, fresh, priorityMemo) ? fresh.reason : "pick", priorityGoing, floor); | |
| 1631 | 1629 | if (paced !== null) { |
| 1632 | 1630 | held.set(fresh.id, "pace"); |
| 1633 | 1631 | if (!noted.has(fresh.id)) { |
| @@ -1654,6 +1652,10 @@ export function pump( | ||
| 1654 | 1652 | } |
| 1655 | 1653 | continue; |
| 1656 | 1654 | } |
| 1655 | // Out of slots: nothing more starts. The runs past this point are still | |
| 1656 | // read, so those a policy above holds are named for vitals, and only the | |
| 1657 | // rest count as waiting for a slot (#11518). | |
| 1658 | if (slots <= 0) continue; | |
| 1657 | 1659 | started.push(fresh.id); |
| 1658 | 1660 | goingIn.set(fresh.repoId, (goingIn.get(fresh.repoId) ?? 0) + 1); |
| 1659 | 1661 | if (isPriorityRun(app, fresh, priorityMemo)) priorityIn.set(fresh.repoId, (priorityIn.get(fresh.repoId) ?? 0) + 1); |
test/vitals-held-agent.test.ts +57
| @@ -11,6 +11,7 @@ import { createUser } from "../src/core/users.ts"; | ||
| 11 | 11 | import { startCrew } from "../src/crew/dispatch.ts"; |
| 12 | 12 | import { turnValve } from "../src/flow/valve_store.ts"; |
| 13 | 13 | import { serve } from "../src/http/server.ts"; |
| 14 | import { setQuota } from "../src/settings/quotas.ts"; | |
| 14 | 15 | import { HELD_FRESH_MS, heldRuns, noteHeld } from "../src/runs/held.ts"; |
| 15 | 16 | import { QUEUE_WAIT_MAX_MS, vitals } from "../src/kernel/vitals.ts"; |
| 16 | 17 | |
| @@ -188,3 +189,59 @@ test("with the real dispatcher, a run queued behind its issue's running run is h | ||
| 188 | 189 | servers.stop(); |
| 189 | 190 | } |
| 190 | 191 | }); |
| 192 | ||
| 193 | test("with the real dispatcher and every slot taken, a run an account limit holds is still named, not counted as waiting for a slot (#11518)", async () => { | |
| 194 | app.db.query("DELETE FROM run").run(); | |
| 195 | const servers = await serve(app); | |
| 196 | const fake = join(dataDir, "fake-claude-3"); | |
| 197 | writeFileSync(fake, `#!/bin/sh\nif [ "$1" = "--version" ]; then echo "2.1.0 (fake)"; exit 0; fi\nexit 1\n`); | |
| 198 | chmodSync(fake, 0o755); | |
| 199 | app.db.query("DELETE FROM flow_valve WHERE repo_id = ?1").run(repoId); | |
| 200 | // Another account's repository, with no limit: its run waits for the slot. | |
| 201 | const other = await createUser(app.db, { | |
| 202 | origin: "seed", | |
| 203 | trusted: true, | |
| 204 | username: "slotwaiter", | |
| 205 | email: "[email protected]", | |
| 206 | password: "a-good-passphrase", | |
| 207 | }); | |
| 208 | if (!other.ok) throw new Error(String(other.error)); | |
| 209 | const otherRepo = crypto.randomUUID(); | |
| 210 | app.db | |
| 211 | .query("INSERT INTO repo (id, owner_id, name, visibility, created_at) VALUES (?1, ?2, 'waits', 'private', ?3)") | |
| 212 | .run(otherRepo, other.value.id, new Date().toISOString()); | |
| 213 | // The holder's account may run one at a time, and one runs: it takes the only slot. | |
| 214 | setQuota(app.db, "runs", 1, ownerId); | |
| 215 | const running = createRun(app.db, { repoId, kind: "agent" }); | |
| 216 | app.db | |
| 217 | .query("UPDATE run SET state = 'running', started_at = ?1, heartbeat_at = ?1 WHERE id = ?2") | |
| 218 | .run(new Date().toISOString(), running.id); | |
| 219 | const ahead = createRun(app.db, { repoId: otherRepo, kind: "agent" }); | |
| 220 | const limited = createRun(app.db, { repoId, kind: "agent" }); | |
| 221 | const earlier = new Date(Date.now() - 3 * QUEUE_WAIT_MAX_MS).toISOString(); | |
| 222 | const hourAgo = new Date(Date.now() - 2 * QUEUE_WAIT_MAX_MS).toISOString(); | |
| 223 | app.db.query("UPDATE run SET created_at = ?1 WHERE id = ?2").run(earlier, ahead.id); | |
| 224 | app.db.query("UPDATE run SET created_at = ?1 WHERE id = ?2").run(hourAgo, limited.id); | |
| 225 | const crew = await startCrew(app, { | |
| 226 | forgeUrl: servers.url, | |
| 227 | localUrl: servers.url, | |
| 228 | binary: fake, | |
| 229 | concurrency: 1, | |
| 230 | budgetUsd: 1, | |
| 231 | runTimeoutMs: 60_000, | |
| 232 | tickMs: 600_000, | |
| 233 | }); | |
| 234 | try { | |
| 235 | // The pass stops starting at `ahead` (no slot), and still reads `limited` past it. | |
| 236 | expect(heldRuns(app.db).get(ahead.id)).toBeUndefined(); | |
| 237 | expect(heldRuns(app.db).get(limited.id)).toBe("limit"); | |
| 238 | await crew.stop(); | |
| 239 | const sentence = agentQueue(Date.now())?.sentence ?? ""; | |
| 240 | expect(sentence).toContain("1 queued, 1 of 1 slots running"); | |
| 241 | expect(sentence).toContain("held on purpose: 1 by an account limit."); | |
| 242 | } finally { | |
| 243 | setQuota(app.db, "runs", null, ownerId); | |
| 244 | await crew.stop(); | |
| 245 | servers.stop(); | |
| 246 | } | |
| 247 | }); |
Nit: the
slots <= 0check now runs after the pace and limit checks. Passes with a long queue will therefore append the pace and limit notes to the run log of every held run behind the slots, once each (they are deduplicated throughnoted). That is intended. Consider a test that asserts a pace-held run past the slots is namedpace, since only the limit path is covered.