src/inference-queue.js

@@ -0,0 +1,130 @@
+export class OverloadError extends Error {
+  constructor(message = 'Timmy is busy right now. Your journal still works — try again shortly or continue manually.') {
+    super(message);
+    this.name = 'OverloadError';
+  }
+}
+
+function normalizeReason(reason) {
+  if (reason instanceof Error) return reason;
+  const text = String(reason ?? '').trim().slice(0, 200);
+  return new Error(text || 'Request cancelled.');
+}
+
+function expiredMessage() {
+  return 'Timmy is busy right now and your request timed out waiting. Your journal still works — try again shortly or continue manually.';
+}
+
+export function createInferenceQueue({
+  concurrency = 1,
+  maxQueueDepth = 0,
+  requestTimeoutMs = 60_000,
+  now = () => Date.now(),
+} = {}) {
+  let active = 0;
+  const waiting = [];
+  const handles = new Map();
+
+  function expireStale(nowMs) {
+    while (waiting.length) {
+      if (nowMs - waiting[0].enqueuedAt <= requestTimeoutMs) break;
+      const expired = waiting.shift();
+      expired.settled = true;
+      expired.abort(new OverloadError(expiredMessage()));
+    }
+  }
+
+  function pump(nowMs = now()) {
+    expireStale(nowMs);
+    while (active < concurrency && waiting.length) {
+      const entry = waiting.shift();
+      if (entry.settled) continue;
+      active += 1;
+      entry.grant();
+    }
+  }
+
+  function release() {
+    active -= 1;
+    pump();
+  }
+
+  async function admit(record) {
+    const { controller } = record;
+    if (controller.signal.aborted) throw normalizeReason(controller.signal.reason);
+    expireStale(now());
+    if (active < concurrency) {
+      active += 1;
+      return;
+    }
+    if (waiting.length >= maxQueueDepth) throw new OverloadError();
+    await new Promise((resolve, reject) => {
+      const entry = {
+        enqueuedAt: now(),
+        settled: false,
+        grant: resolve,
+        abort: reject,
+      };
+      record.entry = entry;
+      waiting.push(entry);
+    });
+    record.entry = null;
+  }
+
+  function run(task) {
+    const controller = new AbortController();
+    const record = { controller, entry: null };
+    let resolveOutcome;
+    let rejectOutcome;
+    // Deliberately not an async function: the caller must receive the very
+    // promise registered in `handles`, or cancellation lookups would target
+    // a different object than the one they hold.
+    const promise = new Promise((resolve, reject) => {
+      resolveOutcome = resolve;
+      rejectOutcome = reject;
+    });
+    handles.set(promise, record);
+    (async () => {
+      let holdsSlot = false;
+      try {
+        await admit(record);
+        holdsSlot = true;
+        // The task body may not have started even though a slot is held; an
+        // external cancel that raced ahead must still stop it here.
+        if (controller.signal.aborted) throw normalizeReason(controller.signal.reason);
+        resolveOutcome(await task(controller.signal));
+      } catch (error) {
+        rejectOutcome(error instanceof Error ? error : normalizeReason(error));
+      } finally {
+        if (holdsSlot) release();
+      }
+      // Drop the handle one microtask after the caller-visible promise
+      // settles, so a same-tick external cancel still finds it.
+      promise.then(
+        () => handles.delete(promise),
+        () => handles.delete(promise),
+      );
+    })();
+    promise.catch(() => {});
+    return promise;
+  }
+
+  function cancel(handle, reason) {
+    const record = handles.get(handle);
+    if (!record) return false;
+    handles.delete(handle);
+    const normalized = normalizeReason(reason);
+    const entry = record.entry;
+    if (entry && !entry.settled) {
+      entry.settled = true;
+      const index = waiting.indexOf(entry);
+      if (index >= 0) waiting.splice(index, 1);
+      entry.abort(normalized);
+    }
+    record.controller.abort(normalized);
+    return true;
+  }
+
+  pump();
+  return { run, cancel };
+}

tests/inference-queue-cancellation.test.js

@@ -0,0 +1,96 @@
+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('a queued request past its deadline is expired deterministically and its place is reusable', async () => {
+  let clock = 0;
+  const queue = createInferenceQueue({ concurrency: 1, maxQueueDepth: 1, requestTimeoutMs: 100, now: () => clock });
+
+  const wedgeGate = deferred();
+  const active = queue.run(signal => {
+    return new Promise((resolve, reject) => {
+      signal.addEventListener('abort', () => reject(new Error('upstream aborted')));
+      wedgeGate.promise.then(resolve, reject);
+    });
+  });
+  active.catch(() => {});
+  await Promise.resolve();
+
+  let started = false;
+  const queued = queue.run(() => { started = true; return 'never'; });
+  await Promise.resolve();
+
+  clock += 150; // deadline (submitted at t=0, limit 100) has elapsed
+  const probe = queue.run(() => 'ok'); // queue activity triggers the deadline sweep
+
+  await assert.rejects(() => queued, error => {
+    assert.ok(error instanceof OverloadError);
+    assert.match(error.message, /timed out|busy/i);
+    return true;
+  });
+  assert.equal(started, false, 'the expired request must never reach the provider');
+
+  // Capacity is intact: cancelling the wedged holder lets the surviving request through.
+  queue.cancel(active, 'cleanup');
+  assert.equal(await Promise.race([probe, new Promise((_, reject) => setTimeout(() => reject(new Error('probe starved')), 200))]), 'ok');
+});
+
+test('client disconnect cancels a queued request before it reaches the provider', async () => {
+  let clock = 0;
+  const queue = createInferenceQueue({ concurrency: 1, maxQueueDepth: 2, requestTimeoutMs: 60_000, now: () => clock });
+
+  const wedgeGate = deferred();
+  const active = queue.run(() => wedgeGate.promise);
+  active.catch(() => {});
+  await Promise.resolve();
+
+  let sawStart = false;
+  const queued = queue.run(() => { sawStart = true; return 'too late'; });
+  await Promise.resolve();
+
+  queue.cancel(queued, 'client disconnected');
+  await assert.rejects(() => queued, error => error.message === 'client disconnected');
+  assert.equal(sawStart, false, 'cancelled request never invoked its task');
+
+  const admitted = queue.run(() => 'admitted-after-cancel');
+  wedgeGate.resolve('wedge-done');
+  assert.equal(await admitted, 'admitted-after-cancel');
+  assert.equal(await active, 'wedge-done');
+});
+
+test('cancelling an already-running task aborts its signal so buffers can be released', async () => {
+  let clock = 0;
+  const queue = createInferenceQueue({ concurrency: 1, maxQueueDepth: 1, requestTimeoutMs: 60_000, now: () => clock });
+
+  const observed = [];
+  const running = queue.run(async signal => {
+    observed.push(signal);
+    signal.addEventListener('abort', () => observed.push(`aborted:${signal.reason?.message ?? signal.reason}`));
+    await new Promise((resolve, reject) => {
+      signal.addEventListener('abort', () => reject(new Error('task stopped')));
+      setTimeout(resolve, 5_000);
+    });
+  });
+
+  // Let the wrapper admit the request and actually invoke the task body.
+  await Promise.resolve();
+  await Promise.resolve();
+  assert.equal(observed.length, 1, 'task is genuinely running before cancellation');
+
+  queue.cancel(running, 'stop');
+  await assert.rejects(() => running, /task stopped/);
+  assert.ok(observed[0] instanceof AbortSignal, 'task received a real AbortSignal');
+  assert.match(String(observed[1]), /stop/, 'running task observed the abort with its reason');
+
+  // The slot was released by the cancelled task, so the next request starts immediately.
+  const next = queue.run(() => 'next-runs');
+  assert.equal(await next, 'next-runs');
+});

tests/inference-queue.test.js

@@ -0,0 +1,72 @@
+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');
+});