diff --git a/src/gateway/server-lanes.hook-group.test.ts b/src/gateway/server-lanes.hook-group.test.ts index efba7e7f8a47..b6e6db3ad1b6 100644 --- a/src/gateway/server-lanes.hook-group.test.ts +++ b/src/gateway/server-lanes.hook-group.test.ts @@ -119,6 +119,67 @@ describe("cron+hook capacity group", () => { await Promise.all(runs); }); + it("hooks-off immediately drains cron work released by the teardown", async () => { + // Teardown must WAKE the lanes it frees, not merely delete membership. + // Asserting only `group === undefined` on an idle lane would pass even if + // clearGroups forgot to add its former members to the commit-drain set, + // leaving released work stuck until some unrelated enqueue pokes the lane. + publish(HOOKS_ON); + + const gates = Array.from({ length: DEFAULT_CRON_MAX_CONCURRENT_RUNS }, () => gate()); + const runs = gates.map((g) => + enqueueCommandInLane(CommandLane.CronNested, async () => await g.promise, { + warnAfterMs: 10_000, + }), + ); + await settle(); + + // One short of the budget, with the last entry queued behind the hook's + // reservation rather than running. + expect(getCommandLaneSnapshot(CommandLane.CronNested).activeCount).toBe( + DEFAULT_CRON_MAX_CONCURRENT_RUNS - 1, + ); + expect(getCommandLaneSnapshot(CommandLane.CronNested).queuedCount).toBe(1); + expect(getCommandLaneSnapshot(CommandLane.CronNested).blockedBy).toBe("sibling-reservation"); + + // Turning hooks off returns the reserved slot to cron. The queued entry + // must start on the publish itself. + publish(HOOKS_OFF); + await settle(); + + expect(getCommandLaneSnapshot(CommandLane.CronNested).group).toBeUndefined(); + expect(getCommandLaneSnapshot(CommandLane.CronNested).activeCount).toBe( + DEFAULT_CRON_MAX_CONCURRENT_RUNS, + ); + expect(getCommandLaneSnapshot(CommandLane.CronNested).queuedCount).toBe(0); + + for (const g of gates) g.release(); + await Promise.all(runs); + }); + + it("clears the group on hooks-off even when the grouped lane is suspended", async () => { + // The teardown path publishes only lanes that are NOT suspended. With hooks + // off, `cron-nested` is the only lane that can be published, so if it is + // suspended the lane map is empty — and a guard that skips publication on + // an empty map would skip the group teardown with it. The stale group + // survives, and the suspended member resumes still paying a reservation for + // a hook lane that no longer receives work. + publish(HOOKS_ON); + expect(getCommandLaneSnapshot(CommandLane.CronNested).group).toBe("cron-hooks"); + + const { seedClearedLaneResumeForTest } = + await import("../agents/session-suspension.test-support.js"); + seedClearedLaneResumeForTest(CommandLane.CronNested, { + resumeConcurrency: DEFAULT_CRON_MAX_CONCURRENT_RUNS, + resumeAtMs: Date.now() + 60_000, + }); + + // gatewayStart consults the cleared-resume map for the suspended set. + applyGatewayLaneConcurrency(resolveGatewayLaneConcurrency(HOOKS_OFF), { gatewayStart: true }); + + expect(getCommandLaneSnapshot(CommandLane.CronNested).group).toBeUndefined(); + }); + it("removes the group when hooks are turned off by a config reload", async () => { publish(HOOKS_ON); expect(getCommandLaneSnapshot(CommandLane.CronNested).group).toBe("cron-hooks"); diff --git a/src/gateway/server-lanes.ts b/src/gateway/server-lanes.ts index 3ec31e6b314e..c20f77b0da85 100644 --- a/src/gateway/server-lanes.ts +++ b/src/gateway/server-lanes.ts @@ -85,7 +85,12 @@ export function applyGatewayLaneConcurrency( // saturation, not that hooks run concurrently with each other. grouped[CommandLane.HookDispatch] = concurrency.hookDispatch; } - if (Object.keys(grouped).length > 0) { + // Publish even when `grouped` is empty. With hooks off, `cron-nested` is the + // only lane that can enter `grouped`, so if it happens to be suspended the + // guard would skip publication entirely — leaving a previously installed + // `cron-hooks` group alive. The suspended member would then resume still + // paying a reservation for a hook lane that no longer receives work. + if (Object.keys(grouped).length > 0 || !hooksEnabled) { publishLaneConfiguration({ lanes: grouped, // Opt-in. With hooks disabled there is no hook work to protect, so no diff --git a/src/process/command-queue.publish-transaction.test.ts b/src/process/command-queue.publish-transaction.test.ts index fc88cce8033d..3302a3844abd 100644 --- a/src/process/command-queue.publish-transaction.test.ts +++ b/src/process/command-queue.publish-transaction.test.ts @@ -12,6 +12,7 @@ */ import { afterEach, beforeEach, describe, expect, test } from "vitest"; import { + clearCommandLane, clearCommandLaneGroup, enqueueCommandInLane, getCommandLaneSnapshot, @@ -126,11 +127,56 @@ describe("publishLaneConfiguration", () => { expect(getCommandLaneSnapshot(CRON).activeCount).toBe(0); for (const g of gates) g.release(); - // The lane never opened, so these stay queued; clear them for teardown. - resetAllLanes(); + // The lane never opened, so this work is still queued. resetAllLanes + // PRESERVES queued entries by design, so it would never settle these — + // clearCommandLane rejects them instead. + clearCommandLane(CRON); await Promise.allSettled(runs); }); + test("a rejected configuration does not leave lane maxima mutated", async () => { + // Stronger than asserting activeCount === 0 after the throw: that only + // proves no commit-time drain ran, not that the lane was left alone. If + // phase 1 widens a lane and group validation then throws, the lane sits at + // the new width governed by NO group, and the next unrelated drain trigger + // dispatches the preserved queue ungoverned. + setCommandLaneConcurrency(CRON, 0); + const gates = Array.from({ length: 4 }, () => gate()); + const runs = gates.map((g) => enqueueCommandInLane(CRON, async () => await g.promise)); + await settle(); + expect(getCommandLaneSnapshot(CRON).maxConcurrent).toBe(0); + + expect(() => + publishLaneConfiguration({ + lanes: { [CRON]: 8 }, + groups: { + [GROUP]: { + budget: 2, + members: [CRON, HOOK], + reservations: { [CRON]: 2, [HOOK]: 1 }, + }, + }, + }), + ).toThrow(/reserves 3 slots but its budget is 2/); + await settle(); + + // The lane must be exactly as it was before the rejected publish. + expect(getCommandLaneSnapshot(CRON).maxConcurrent).toBe(0); + expect(getCommandLaneSnapshot(CRON).group).toBeUndefined(); + + // And a later drain trigger must not dispatch the queue that was preserved + // across the failed publish. + const extra = gate(); + const extraRun = enqueueCommandInLane(CRON, async () => await extra.promise); + await settle(); + expect(getCommandLaneSnapshot(CRON).activeCount).toBe(0); + + for (const g of gates) g.release(); + extra.release(); + clearCommandLane(CRON); + await Promise.allSettled([...runs, extraRun]); + }); + test("republishing a narrower budget does not admit beyond the new cap", async () => { publishLaneConfiguration({ lanes: { [CRON]: 8, [HOOK]: 1 }, @@ -161,7 +207,7 @@ describe("publishLaneConfiguration", () => { for (const g of gates) g.release(); extra.release(); - resetAllLanes(); + clearCommandLane(CRON); await Promise.allSettled([...runs, blocked]); }); }); diff --git a/src/process/command-queue.ts b/src/process/command-queue.ts index 6852c199b7d1..9637084af404 100644 --- a/src/process/command-queue.ts +++ b/src/process/command-queue.ts @@ -647,7 +647,7 @@ function drainGroupSiblings(lane: string): void { * from its group, or session suspend/resume would silently restore a member to * ungoverned concurrency. */ -export function setCommandLaneGroup(group: string, spec: CommandLaneGroupSpec): void { +function validateCommandLaneGroupSpec(group: string, spec: CommandLaneGroupSpec): LaneGroupState { const members = spec.members.map((member) => normalizeLane(member)); for (const member of members) { assertGroupEligibleLane(member); @@ -671,17 +671,35 @@ export function setCommandLaneGroup(group: string, spec: CommandLaneGroupSpec): `command lane group "${group}" reserves ${reservedTotal} slots but its budget is ${budget}`, ); } + return { group, budget, members: new Set(members), reservations }; +} + +/** Install a validated group, detaching its members from any previous owner. */ +function installCommandLaneGroup(next: LaneGroupState): void { const { groups, groupByLane } = getGroupRegistry(); - const previous = groups.get(group); + const previous = groups.get(next.group); if (previous) { for (const member of previous.members) { groupByLane.delete(member); } } - groups.set(group, { group, budget, members: new Set(members), reservations }); - for (const member of members) { - groupByLane.set(member, group); + for (const member of next.members) { + // A lane may belong to at most one group. Without this, the old owner's + // `members` would still contain the lane and would keep counting its active + // tasks toward a budget it no longer participates in. + const owner = groupByLane.get(member); + if (owner && owner !== next.group) { + groups.get(owner)?.members.delete(member); + } } + groups.set(next.group, next); + for (const member of next.members) { + groupByLane.set(member, next.group); + } +} + +export function setCommandLaneGroup(group: string, spec: CommandLaneGroupSpec): void { + installCommandLaneGroup(validateCommandLaneGroupSpec(group, spec)); } /** Remove a group and release its members back to lane-local admission. */ @@ -836,6 +854,15 @@ export function publishLaneConfiguration(config: { /** Groups to remove as part of the same transaction. */ clearGroups?: readonly string[]; }): void { + // Phase 0 — validate EVERYTHING before mutating anything. Validating inside + // the install loop would leave already-widened lanes behind on a throw: + // governed by no group, and dispatching their preserved queue on the next + // unrelated drain trigger. Rejection must be a no-op, not a partial apply. + const validated: LaneGroupState[] = []; + for (const [group, spec] of Object.entries(config.groups ?? {})) { + validated.push(validateCommandLaneGroupSpec(group, spec)); + } + const touched = new Set(); // Phase 1 — install state with dispatch suppressed. Nothing may start here. for (const [rawLane, maxConcurrent] of Object.entries(config.lanes ?? {})) { @@ -856,12 +883,10 @@ export function publishLaneConfiguration(config: { groups.delete(group); } } - for (const [group, spec] of Object.entries(config.groups ?? {})) { - // Validation throws BEFORE any drain, so a rejected configuration cannot - // leave lanes widened and dispatching under no group at all. - setCommandLaneGroup(group, spec); - for (const member of spec.members) { - touched.add(normalizeLane(member)); + for (const next of validated) { + installCommandLaneGroup(next); + for (const member of next.members) { + touched.add(member); } } // Phase 2 — commit. Group membership and budgets are now final, so every