diff --git a/src/engine.ts b/src/engine.ts index 5104d3e..87ffe32 100644 --- a/src/engine.ts +++ b/src/engine.ts @@ -496,7 +496,7 @@ export class Engine implements DecisionEngine { state: JsonValue, seed: number, ): Promise { - if (this.waiting >= this.settings.maxQueue) { + if (this.waiting >= this.settings.maxInflight + this.settings.maxQueue) { throw new OverloadedError("LocalJev is at capacity. Retry shortly."); } this.waiting += 1; diff --git a/test/engine.test.ts b/test/engine.test.ts index 15bf1ec..1f3fc46 100644 --- a/test/engine.test.ts +++ b/test/engine.test.ts @@ -3,6 +3,7 @@ import { describe, expect, test } from "bun:test"; import { loadSettings } from "../src/config"; import { Engine, + OverloadedError, buildSystemPrompt, confidence, decodeAnswers, @@ -142,3 +143,41 @@ test("engine retries malformed output and sends upstream authentication", async expect(secondBody.messages.at(-1).role).toBe("user"); expect(secondBody.response_format.type).toBe("json_schema"); }); + +test("allows maxQueue waiting decisions in addition to maxInflight calls", async () => { + let unblockFirst!: () => void; + let signalFirstStarted!: () => void; + const firstBlocked = new Promise((resolve) => { + unblockFirst = resolve; + }); + const firstStarted = new Promise((resolve) => { + signalFirstStarted = resolve; + }); + let calls = 0; + const engine = new Engine( + loadSettings({ maxInflight: 1, maxQueue: 1, malformedRetries: 0 }), + async () => { + calls += 1; + if (calls === 1) { + signalFirstStarted(); + await firstBlocked; + } + return Response.json({ + choices: [{ message: { content: JSON.stringify({ answers: { q1: 0.5 } }) } }], + }); + }, + ); + const oneQuestion: Record = { + urgent: { type: "noul", instructions: "Is it urgent?", criteria: null }, + }; + + const active = engine.decide(oneQuestion, "message", 1); + await firstStarted; + const queued = engine.decide(oneQuestion, "message", 2); + const overCapacity = engine.decide(oneQuestion, "message", 3); + + await expect(overCapacity).rejects.toBeInstanceOf(OverloadedError); + unblockFirst(); + await expect(Promise.all([active, queued])).resolves.toHaveLength(2); + expect(calls).toBe(2); +});