openvibers/openvibe.ai Commits

Run queue caps: a run that would wait for a slot is refused 429 queue.full (problem+json, Retry-After) once its caller already has AI_MAX_QUEUED_RUNS_PER_CALLER runs waiting (default 50) or the queue holds AI_MAX_QUEUED_RUNS (default 200), checked before quota so a refused run leaves no row and spends nothing; one caller can no longer fill the queue for every other

OP OpenVibers committed 23 Sept 2026 · bdca064a82 · 6 files, +96 −6


Changes

ddiff --git a/.env.example b/.env.example
iindex ebcd2fd..779d4d6 100644
---- a/.env.example
++++ b/.env.example
@@@ -81,6 +81,9 @@ AI_MAX_COST_USD_PER_DAY=0
# ── Runs, cache, logs ──
#AI_MAX_WAIT_MS=60000
#AI_MAX_CONCURRENT_RUNS=4
++# Runs waiting for a slot; past either cap a new run is refused (429 queue.full, Retry-After)
++#AI_MAX_QUEUED_RUNS=200
++#AI_MAX_QUEUED_RUNS_PER_CALLER=50
#AI_CACHE_TTL_SEC=604800
#AI_MAX_INPUT_BYTES=6291456
#AI_RUN_RETENTION_DAYS=30
ddiff --git a/README.md b/README.md
iindex d45bd7d..2874a90 100644
---- a/README.md
++++ b/README.md
@@@ -62,7 +62,8 @@ idempotent replay):
Errors are `application/problem+json` (openvibe-contracts `http.problem`): `input.invalid` (with the
schema errors), `workflow.not_found`, `idempotency.conflict`, `quota.exceeded` (429 +
--`Retry-After`), `capability.denied`, `capability.namespace_denied`, `token.*`. A run that cannot
++`Retry-After`), `queue.full` (429 + `Retry-After`: the run would wait behind `AI_MAX_QUEUED_RUNS`
++runs, or `AI_MAX_QUEUED_RUNS_PER_CALLER` of this caller's), `capability.denied`, `capability.namespace_denied`, `token.*`. A run that cannot
produce a real answer ends `failed` with an explicit code — `provider.unavailable`,
`route.unavailable`, `fetch.refused`, `source.unavailable`, `input.insufficient`, `output.empty`,
`output.invalid`, `run.interrupted` — and never with filler content.
@@@ -140,6 +141,7 @@ stub joins every route as a last resort outside production only (`AI_STUB_FALLBA
| A degraded primary falls back and records it; breaker; skips; timeouts; explicit `provider.unavailable` | `test/fallback.test.js` |
| Quotas refuse with 429 + Retry-After **before** any provider call; per-service, attribution, cost caps | `test/quota.test.js` |
| The cache never crosses requester, actor, target or attribution scope | `test/cache.test.js` |
++| Queue caps: one caller cannot fill the run queue (per-caller and global 429 `queue.full` + Retry-After) | `test/queue.test.js` |
| Idempotency, cancel (running and queued), retry, async polling, audit rows, restart recovery, no raw prompts or inline images kept | `test/runs.test.js` |
| Media fetched only from allow-listed https OpenVibe hosts; DNS answers and every redirect hop re-checked (a public hop never reaches internal Media); size caps | `test/ssrf.test.js` |
| The shared compiled-schema cache (caller-supplied schemas) is an LRU bounded by count and bytes, in Ajv too | `test/schemas.test.js` |
ddiff --git a/server/api/runs.js b/server/api/runs.js
iindex a736828..825d6e7 100644
---- a/server/api/runs.js
++++ b/server/api/runs.js
@@@ -14,7 +14,7 @@
*
* Responses: 201 when a new run finished (or was served from cache) within the wait, 202 while it
* is still queued/running (poll GET /runs/:id), 200 for an idempotent replay. Errors are
-- * problem+json; a quota refusal is 429 with Retry-After.
++ * problem+json; a quota refusal or a full run queue (queue.full) is 429 with Retry-After.
*/
const express = require('express');
const { http } = require('openvibe-contracts');
ddiff --git a/server/config.js b/server/config.js
iindex c377b8e..d27c91d 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -136,6 +136,9 @@ function load(env = process.env) {
runs: {
maxWaitMs: int(env.AI_MAX_WAIT_MS, 60000),
maxConcurrent: Math.max(1, int(env.AI_MAX_CONCURRENT_RUNS, 4)),
++ // Runs waiting for a slot: past either cap a new run is refused 429 queue.full + Retry-After.
++ maxQueued: Math.max(1, int(env.AI_MAX_QUEUED_RUNS, 200)),
++ maxQueuedPerCaller: Math.max(1, int(env.AI_MAX_QUEUED_RUNS_PER_CALLER, 50)),
defaultCacheTtlSec: int(env.AI_CACHE_TTL_SEC, 7 * 24 * 3600),
maxInputBytes: int(env.AI_MAX_INPUT_BYTES, 6 * 1024 * 1024),
retentionDays: int(env.AI_RUN_RETENTION_DAYS, 30),
ddiff --git a/server/runs.js b/server/runs.js
iindex 95b5961..9cba4ca 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -9,7 +9,10 @@
* 3. cache: a hit in THIS request's scope becomes a run with status 'cached' (no provider call)
* 4. quota: every applicable quota is checked and counted — a refusal is a 429 before any
* provider is touched
-- * 5. the run is stored 'queued' and executed (inline up to ?wait=ms, else by the worker)
++ * 5. queue caps: a run that would have to wait is refused (429 queue.full + Retry-After) when the
++ * queue already holds AI_MAX_QUEUED_RUNS runs, or AI_MAX_QUEUED_RUNS_PER_CALLER of this
++ * requester's, so one caller cannot fill the queue for everyone — also before any provider call
++ * 6. the run is stored 'queued' and executed (inline up to ?wait=ms, else by the worker)
* Every create, cancel and retry writes an audit row. Output is attributable to workflow + version
* + template version + route version + provider/model + run id — never to a person.
*/
@@@ -18,6 +21,7 @@ const { AiError, sha256, stableStringify, parseJson, iso } = require('./util');
const schemas = require('./schemas');
const TERMINAL = new Set(['succeeded', 'failed', 'cancelled', 'cached']);
++const QUEUE_RETRY_AFTER_S = 5;
const newRunId = () => `run_${ids.ulid()}`;
function entityKey(ref) { return ref ? `${ref.service}:${ref.type}:${ref.id}` : null; }
@@@ -53,7 +57,7 @@ function retainedInput(input, wf, debugRaw) {
}
function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console }) {
-- const inflight = new Map(); // run id -> { controller, promise }
++ const inflight = new Map(); // run id -> { controller, promise, release, caller }
const queue = []; // run ids waiting for the worker
let active = 0;
let shuttingDown = false;
@@@ -200,7 +204,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
}
}
-- // Quota, before any provider call.
++ // Queue caps, then quota: both before any provider call (a refused run is not counted).
++ const caller = `${requester.type}:${requester.id}`;
++ admit(caller);
const reserved = quotas.reserve(ctx);
db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, template_key, template_version, route_key, route_version, status, requester_type, requester_id, on_behalf_of, attribution,
@@@ -214,12 +220,22 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
let release;
const gate = new Promise((r) => { release = r; });
const promise = gate.then(() => execute(id, { controller, reserved, scope, ctx, input })).catch((err) => { log.error(`[runs] ${id}: ${err.stack || err}`); });
-- inflight.set(id, { controller, promise, release });
++ inflight.set(id, { controller, promise, release, caller });
queue.push(id);
pump();
return { run: get(id), created: true, promise };
}
++ /** Refuse a run that would wait in a full queue, globally or for this caller. */
++ function admit(caller) {
++ if (active < config.runs.maxConcurrent && !queue.length) return; // starts at once
++ const full = (detail, scope) => new AiError(429, 'queue.full', detail, { retry_after_seconds: QUEUE_RETRY_AFTER_S, queue: { scope, queued: queue.length } });
++ if (queue.length >= config.runs.maxQueued) throw full(`the run queue is full (${config.runs.maxQueued} waiting); retry shortly`, 'global');
++ let mine = 0;
++ for (const qid of queue) { const h = inflight.get(qid); if (h && h.caller === caller) mine++; }
++ if (mine >= config.runs.maxQueuedPerCaller) throw full(`${caller} already has ${mine} runs waiting (the per-caller limit); retry shortly`, 'caller');
++ }
++
function pump() {
while (active < config.runs.maxConcurrent && queue.length) {
const id = queue.shift();
ddiff --git a/test/queue.test.js b/test/queue.test.js
nnew file mode 100644
iindex 0000000..b0866b5
---- /dev/null
++++ b/test/queue.test.js
@@@ -0,0 +1,66 @@
++'use strict';
++// Run queue caps: a run that would have to wait is refused 429 queue.full (problem+json with
++// Retry-After) once the caller already has AI_MAX_QUEUED_RUNS_PER_CALLER runs waiting, or the queue
++// holds AI_MAX_QUEUED_RUNS; one caller cannot fill the queue for everyone. A refused run leaves no
++// row and spends no quota; a slot freed by cancel or completion admits the next run.
++
++const assert = require('assert');
++const { boot, request, token, suite, ALL } = require('./helpers');
++
++const t = suite('queue');
++const admin = token('ops', ALL);
++const a = token('live', ['ai.run.create', 'ai.run.read']);
++const b = token('tools', ['ai.run.create', 'ai.run.read']);
++let h;
++
++const sleep = (ms) => new Promise(r => setTimeout(r, ms));
++const create = (tok, i) => request(h.base, 'POST', '/api/v1/runs', { tok, body: { workflow: 'ai.generate', input: { prompt: `queued ${i}` }, options: { cache: false } } });
++const count = (who) => h.db.prepare('SELECT COUNT(*) AS n FROM runs WHERE requester_id = ?').get(who).n;
++
++function assertFull(r, scope) {
++ assert.strictEqual(r.status, 429, r.text);
++ assert.strictEqual(r.body.code, 'queue.full');
++ assert.match(r.headers.get('content-type'), /application\/problem\+json/);
++ assert.ok(Number(r.headers.get('retry-after')) >= 1, 'Retry-After');
++ assert.strictEqual(r.body.queue.scope, scope);
++}
++
++t.test('boot with one slot, a slow provider and small queue caps', async () => {
++ h = await boot({ env: { AI_MAX_CONCURRENT_RUNS: '1', AI_MAX_QUEUED_RUNS: '5', AI_MAX_QUEUED_RUNS_PER_CALLER: '3', AI_STUB_FALLBACK: 'false' } });
++ const p = await request(h.base, 'POST', '/api/v1/providers', { tok: admin, body: { key: 'slowstub', kind: 'stub', auth_mode: 'none', capabilities: ['chat', 'generate', 'summarize', 'json'], metadata: { delay_ms: 1500 } } });
++ assert.strictEqual(p.status, 201, p.text);
++ await request(h.base, 'POST', '/api/v1/routes/default.chat/versions', { tok: admin, body: { primary: { provider: 'slowstub' }, fallbacks: [] } });
++});
++
++t.test('one caller is held to its per-caller cap; others still get in', async () => {
++ const mine = [];
++ for (let i = 0; i < 4; i++) { // 1 running + 3 waiting
++ const r = await create(a, i);
++ assert.strictEqual(r.status, 202, r.text);
++ mine.push(r.body.run.id);
++ }
++ const before = count('live');
++ assertFull(await create(a, 'over'), 'caller');
++ assert.strictEqual(count('live'), before, 'a refused run leaves no row');
++ const other = await create(b, 0);
++ assert.strictEqual(other.status, 202, 'another caller still queues');
++ h.mine = mine;
++});
++
++t.test('the global cap holds across callers', async () => {
++ assert.strictEqual((await create(b, 1)).status, 202); // 5 waiting now
++ assertFull(await create(b, 2), 'global');
++ assertFull(await create(token('games', ['ai.run.create']), 0), 'global');
++});
++
++t.test('a freed slot admits the next run', async () => {
++ const waiting = h.mine[3];
++ const c = await request(h.base, 'POST', `/api/v1/runs/${waiting}/cancel`, { tok: a });
++ assert.strictEqual(c.body.run.status, 'cancelled');
++ assert.strictEqual((await create(a, 'again')).status, 202, 'room again after a cancel');
++ assertFull(await create(a, 'again 2'), 'global');
++});
++
++t.test('shutdown', async () => { await h.stop(); await sleep(10); });
++
++t.run();