From e615ca4df058ecbeeece41ea3438a800ac9ac9fa Mon Sep 17 00:00:00 2001 From: Jose Date: Tue, 14 Jul 2026 13:54:40 +0100 Subject: [PATCH] fix(ai-builder): Prioritize pinned scenario waiters over relocatable builds (no-changelog) Co-Authored-By: Claude Fable 5 --- .../__tests__/lane-allocator.test.ts | 46 +++++++++++++++++ .../@n8n/instance-ai/evaluations/cli/index.ts | 6 ++- .../evaluations/cli/lane-allocator.ts | 49 ++++++++++++++----- 3 files changed, 88 insertions(+), 13 deletions(-) diff --git a/packages/@n8n/instance-ai/evaluations/__tests__/lane-allocator.test.ts b/packages/@n8n/instance-ai/evaluations/__tests__/lane-allocator.test.ts index 409d08469cc..81c2b7c7d7d 100644 --- a/packages/@n8n/instance-ai/evaluations/__tests__/lane-allocator.test.ts +++ b/packages/@n8n/instance-ai/evaluations/__tests__/lane-allocator.test.ts @@ -93,6 +93,10 @@ describe('LaneAllocator', () => { }); describe('pinned work slots (scenario executions)', () => { + afterEach(() => { + vi.useRealTimers(); + }); + it('counts pinned work toward the lane cap so a backlogged lane stops winning builds', async () => { const lanes = newLanes(2); const a = new LaneAllocator(lanes, 2); @@ -159,6 +163,48 @@ describe('LaneAllocator', () => { vi.useRealTimers(); }); + it('serves a queued pinned waiter before an earlier-queued build waiter', async () => { + const lanes = newLanes(1); + const a = new LaneAllocator(lanes, 1); + await a.acquire('p1'); // fills the lane + const order: string[] = []; + void a.acquire('p2').then(() => order.push('build')); + void a.acquireOn(lanes[0]).then(() => order.push('pinned')); + a.release(lanes[0], 'p1'); + await new Promise((r) => setImmediate(r)); + // The build queued first, but it could run on any lane — the pinned + // scenario can only run here, so it wins the freed slot. + expect(order).toEqual(['pinned']); + a.release(lanes[0]); + await new Promise((r) => setImmediate(r)); + expect(order).toEqual(['pinned', 'build']); + }); + + it('hands a re-admitted lane to its pinned backlog before queued builds', async () => { + vi.useFakeTimers(); + const lanes = newLanes(1); + let healthy = false; + const a = new LaneAllocator(lanes, 2, { + probe: async () => await Promise.resolve(healthy), + probeIntervalMs: 1000, + quarantineThreshold: 1, + allQuarantinedGraceMs: 60_000, + }); + a.reportOutcome(lanes[0], 'transient-failure'); + const order: string[] = []; + void a.acquire('p1').then(() => order.push('build')); + void a.acquireOn(lanes[0]).then(() => order.push('pinned-1')); + void a.acquireOn(lanes[0]).then(() => order.push('pinned-2')); + healthy = true; + await vi.advanceTimersByTimeAsync(1000); + // Two slots on re-admission: the pinned backlog (which waited through + // the outage) fills them; the build — queued first — stays behind. + expect(order).toEqual(['pinned-1', 'pinned-2']); + a.release(lanes[0]); + await vi.advanceTimersByTimeAsync(0); + expect(order).toEqual(['pinned-1', 'pinned-2', 'build']); + }); + it('does not hand a freed slot on another lane to a pinned waiter', async () => { const lanes = newLanes(2); const a = new LaneAllocator(lanes, 1); diff --git a/packages/@n8n/instance-ai/evaluations/cli/index.ts b/packages/@n8n/instance-ai/evaluations/cli/index.ts index 8d4ea8ea337..fb28977e4a3 100644 --- a/packages/@n8n/instance-ai/evaluations/cli/index.ts +++ b/packages/@n8n/instance-ai/evaluations/cli/index.ts @@ -1092,8 +1092,10 @@ async function runWithLangSmith(config: RunConfig): Promise<{ // so they take one of its work slots — that's what makes scenario // backlog visible to the build allocator. A quarantined lane keeps // the slot request queued until re-admission (the workflow survives - // the container restart), bounded by the scenario's own budget. - await allocator.acquireOn(builtOnLane, { deadlineMs: scenarioTimeoutMs }); + // the container restart). Waiting burns no compute, so the deadline + // is 2× the scenario budget: enough to outlast in-flight builds + // ahead of it, while still bounding a truly wedged lane. + await allocator.acquireOn(builtOnLane, { deadlineMs: scenarioTimeoutMs * 2 }); try { result = await builtOnLane.tracedExecute({ workflowId: build.workflowId, diff --git a/packages/@n8n/instance-ai/evaluations/cli/lane-allocator.ts b/packages/@n8n/instance-ai/evaluations/cli/lane-allocator.ts index fcb7afeedf2..21e9eb45360 100644 --- a/packages/@n8n/instance-ai/evaluations/cli/lane-allocator.ts +++ b/packages/@n8n/instance-ai/evaluations/cli/lane-allocator.ts @@ -2,9 +2,11 @@ // scenario executions alike — capped at `laneConcurrency`, and a lane never // runs the same build key twice concurrently. Builds may run on any lane // (least-loaded wins); scenario executions are pinned to the lane holding -// their workflow and wait for a slot there via `acquireOn`. Counting both -// kinds of work is what stops a lane that falls behind on scenario -// executions from continuing to win new builds while its backlog grows. +// their workflow and wait for a slot there via `acquireOn`. Two rules keep +// the pinned work from starving behind builds: a freed slot serves pinned +// waiters before build waiters (builds have other lanes to go to), and build +// placement counts a lane's queued pinned work as load, steering new builds +// away from a backlog instead of piling more work in front of it. // // A dead lane fails builds in milliseconds, making it permanently the // least-loaded lane — it would swallow the whole remaining queue. Consecutive @@ -48,6 +50,11 @@ const DEFAULT_ALL_QUARANTINED_GRACE_MS = 5 * 60_000; export class LaneAllocator { private readonly waiters: Array> = []; + /** Queued pinned waiters per lane. Build placement counts these as load so + * new builds route away from a lane whose scenario backlog is growing — + * builds can run anywhere, the queued scenarios can't. */ + private readonly pinnedQueued = new Map(); + private readonly consecutiveFailures = new Map(); private readonly quarantined = new Set(); @@ -94,7 +101,10 @@ export class LaneAllocator { const deadlineMs = opts.deadlineMs; waiter.deadline = setTimeout(() => { const i = this.waiters.indexOf(waiter); - if (i !== -1) this.waiters.splice(i, 1); + if (i !== -1) { + this.waiters.splice(i, 1); + this.trackPinned(lane, -1); + } reject( new Error( `no lane slot within ${String(deadlineMs)}ms (${this.quarantined.has(lane) ? 'lane quarantined' : 'lane at capacity'})`, @@ -104,6 +114,7 @@ export class LaneAllocator { waiter.deadline.unref?.(); } this.waiters.push(waiter); + this.trackPinned(lane, +1); }); } @@ -199,15 +210,28 @@ export class LaneAllocator { if (w.deadline) clearTimeout(w.deadline); w.reject(error); } + this.pinnedQueued.clear(); + } + + private trackPinned(lane: L, delta: number): void { + const next = (this.pinnedQueued.get(lane) ?? 0) + delta; + if (next > 0) this.pinnedQueued.set(lane, next); + else this.pinnedQueued.delete(lane); } private findFree(key: string, not?: L): L | undefined { // Least-loaded policy: spread builds evenly across lanes rather than - // filling lane 0 to cap before touching lane 1. Avoids hot-spotting. + // filling lane 0 to cap before touching lane 1. Queued pinned scenario + // work counts as load so builds route away from a growing backlog. let best: L | undefined; + let bestLoad = Number.POSITIVE_INFINITY; for (const lane of this.lanes) { if (lane === not || !this.canServe(lane, key)) continue; - if (best === undefined || lane.activeWork < best.activeWork) best = lane; + const load = lane.activeWork + (this.pinnedQueued.get(lane) ?? 0); + if (load < bestLoad) { + best = lane; + bestLoad = load; + } } return best; } @@ -226,13 +250,16 @@ export class LaneAllocator { } private wakeNext(lane: L): boolean { - // Wake the first waiter this lane can now serve. FIFO ordering. - for (let i = 0; i < this.waiters.length; i++) { - const w = this.waiters[i]; - if (w.lane !== undefined && w.lane !== lane) continue; - if (this.canServe(lane, w.key)) { + // Pinned waiters win a freed slot over builds — a queued build can run + // on any lane, the pinned scenario only here. FIFO within each class. + for (const pinnedPass of [true, false]) { + for (let i = 0; i < this.waiters.length; i++) { + const w = this.waiters[i]; + if (pinnedPass ? w.lane !== lane : w.lane !== undefined) continue; + if (!this.canServe(lane, w.key)) continue; this.waiters.splice(i, 1); if (w.deadline) clearTimeout(w.deadline); + if (w.lane !== undefined) this.trackPinned(w.lane, -1); this.markBusy(lane, w.key); w.resolve(lane); return true;