From 0eb96a78e0bed5f7e58dec965611ec5679d4ec03 Mon Sep 17 00:00:00 2001 From: Spencer Fuller Date: Thu, 30 Jul 2026 21:15:03 -0500 Subject: [PATCH] fix(queue): make rejected publication a no-op; always tear down the group on hooks-off MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-5 implementation review (fiducian-spencer-001, CHANGES REQUESTED) found one real atomicity bug and two test gaps. Both bugs are fixed and both are now mutation-guarded. BLOCKER — rejected publishLaneConfiguration left lane maxima mutated. Phase 1 widened lanes, then setCommandLaneGroup could throw (e.g. sum(reservations) > budget) with no rollback. No drain ran, so the old test's activeCount === 0 assertion passed — but the lane sat at the new width governed by NO group, and the next unrelated drain trigger would dispatch its preserved queue ungoverned. The function comment promised exactly what the code did not do. Validation is now a distinct phase 0 over every group spec before anything is mutated; validateCommandLaneGroupSpec/installCommandLaneGroup split out so setCommandLaneGroup and the transaction share one validation path. BUG — hooks-off skipped group teardown when the grouped lane was suspended. applyGatewayLaneConcurrency published only when the lane map was non-empty. With hooks off, cron-nested is the only lane that can enter it, so a suspended cron-nested left the map empty and clearGroups was never published. A previously installed cron-hooks group survived, and the member resumed still paying a reservation for a hook lane receiving no work. Now publishes whenever hooks are disabled, regardless of the lane map. Also (review item 4): a lane may now belong to at most one group. installCommandLaneGroup removes it from any prior owner's members, which otherwise kept counting its active tasks toward a budget it had left. Not reachable with the single default group, but this is a public API. Tests: 3 new. Mutation-verified — - validating during install instead of before: 'expected 8 to be +0' - restoring the non-empty-lane-map guard: 'expected cron-hooks to be undefined' (this one initially SURVIVED; the first version of the teardown test never simulated suspension, so it could not see the bug. Now seeds a cleared lane resume and publishes via the gatewayStart path.) - hooks-off now proves it DRAINS the work its teardown releases, not just that membership was deleted. Test teardown fixed: resetAllLanes preserves queued entries by design, so work on a lane that never opened never settles. clearCommandLane rejects it instead. Typecheck clean (tsgo core + core test, exit 0). Refs: openclaw#98813 Co-Authored-By: Claude Opus 5 (1M context) --- src/gateway/server-lanes.hook-group.test.ts | 61 +++++++++++++++++++ src/gateway/server-lanes.ts | 7 ++- .../command-queue.publish-transaction.test.ts | 52 +++++++++++++++- src/process/command-queue.ts | 47 ++++++++++---- 4 files changed, 152 insertions(+), 15 deletions(-) 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