Timmy
38a59c8ee9
Shared primitive for Timmy #17: concurrency slots, FIFO max-depth overload rejection with sanitized manual-fallback copy, queue-deadline expiry, per-request AbortSignal delivery, immediate slot release on external cancel.
73 lines
2.9 KiB
JavaScript
73 lines
2.9 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
|
|
import { OverloadError, createInferenceQueue } from '../src/inference-queue.js';
|
|
|
|
function deferred() {
|
|
let resolve;
|
|
let reject;
|
|
const promise = new Promise((res, rej) => { resolve = res; reject = rej; });
|
|
return { promise, resolve, reject };
|
|
}
|
|
|
|
test('vision analysis runs with bounded concurrency and strict FIFO ordering', async () => {
|
|
let clock = 0;
|
|
const gates = new Map();
|
|
const started = [];
|
|
const queue = createInferenceQueue({ concurrency: 2, maxQueueDepth: 4, requestTimeoutMs: 10_000, now: () => clock });
|
|
|
|
const run = label => queue.run(signal => {
|
|
started.push(label);
|
|
const gate = deferred();
|
|
gates.set(label, gate);
|
|
gate.promise.catch(() => {});
|
|
return gate.promise;
|
|
});
|
|
|
|
const first = run('first');
|
|
const second = run('second');
|
|
const third = run('third');
|
|
const fourth = run('fourth');
|
|
while (started.length < 2) await Promise.resolve();
|
|
assert.deepEqual(started, ['first', 'second'], 'only concurrency slots start immediately');
|
|
|
|
gates.get('first').resolve({ ok: 'first' });
|
|
assert.equal((await first).ok, 'first');
|
|
|
|
// The slot freed by `first` must be granted to `third` (head of the queue).
|
|
while (!gates.has('third')) await Promise.resolve();
|
|
gates.get('third').resolve({ ok: 'third' });
|
|
|
|
// Only after `third` releases its slot may `fourth` start.
|
|
while (!gates.has('fourth')) await Promise.resolve();
|
|
gates.get('second').resolve({ ok: 'second' });
|
|
gates.get('fourth').resolve({ ok: 'fourth' });
|
|
|
|
assert.deepEqual(await Promise.all([second, third, fourth]), [{ ok: 'second' }, { ok: 'third' }, { ok: 'fourth' }]);
|
|
assert.equal(started.length, 4, 'every admitted request eventually started');
|
|
assert.deepEqual([...new Set(started)], ['first', 'second', 'third', 'fourth'], 'each queued request started exactly once');
|
|
assert.equal(started.indexOf('third') < started.indexOf('fourth'), true, 'queued requests were granted in FIFO order');
|
|
});
|
|
|
|
test('overloaded submissions fail fast with a stable sanitized manual-fallback error', async () => {
|
|
let clock = 0;
|
|
const gates = [];
|
|
const queue = createInferenceQueue({ concurrency: 1, maxQueueDepth: 1, requestTimeoutMs: 10_000, now: () => clock });
|
|
|
|
const hold = queue.run(() => { const gate = deferred(); gates.push(gate); gate.promise.catch(() => {}); return gate.promise; });
|
|
await Promise.resolve();
|
|
const held = queue.run(() => new Promise(() => {}));
|
|
await Promise.resolve();
|
|
|
|
await assert.rejects(() => queue.run(() => new Promise(() => {})), error => {
|
|
assert.ok(error instanceof OverloadError);
|
|
assert.match(error.message, /busy|full/i);
|
|
assert.doesNotMatch(error.message, /stack|internal|fetch|127\.0\.0\.1/i);
|
|
return true;
|
|
}, 'excess request beyond maxQueueDepth is rejected immediately');
|
|
|
|
// Active and already-queued work continues unaffected by the rejected submission.
|
|
gates[0].resolve('done');
|
|
assert.equal(await hold, 'done');
|
|
});
|