Bound and deduplicate server image rendering per account

src/rendering/queue.test.ts

@@ -0,0 +1,27 @@
+import { expect, test } from 'bun:test';
+import { createRenderQueue, RenderBusyError } from './queue';
+
+const result = new Uint8Array([1, 2, 3]);
+test('render queue bounds global concurrency and shares only the same owner and content', async () => {
+  const queue = createRenderQueue({ concurrency: 2, maxPending: 4, maxPerUser: 2 });
+  let active = 0, peak = 0, calls = 0;
+  const run = async () => { calls++; active++; peak = Math.max(peak, active); await Bun.sleep(5); active--; return result; };
+  const first = queue('alice', 'one', run);
+  expect(queue('alice', 'one', run)).toBe(first);
+  const all = [first, queue('alice', 'two', run), queue('bob', 'one', run), queue('carol', 'one', run)];
+  await Promise.all(all); expect(calls).toBe(4); expect(peak).toBe(2);
+  expect(await queue('alice', 'one', run)).toBe(result); expect(calls).toBe(5);
+});
+
+test('render admission enforces per-user and global capacity and recovers after failures', async () => {
+  const queue = createRenderQueue({ concurrency: 1, maxPending: 2, maxPerUser: 1 });
+  let release!: () => void;
+  const first = queue('alice', 'a', async () => { await new Promise<void>(resolve => { release = resolve; }); throw new Error('render failed'); });
+  const failure = first.catch(error => error);
+  await expect(queue('alice', 'b', async () => result)).rejects.toBeInstanceOf(RenderBusyError);
+  const second = queue('bob', 'b', async () => result);
+  await expect(queue('carol', 'c', async () => result)).rejects.toBeInstanceOf(RenderBusyError);
+  while (!release) await Bun.sleep(1);
+  release(); expect((await failure).message).toBe('render failed'); await second;
+  expect(await queue('alice', 'a', async () => result)).toBe(result);
+});

src/rendering/queue.ts

@@ -0,0 +1,42 @@
+import { createHash } from 'node:crypto';
+import { ApiError } from '../habits/service';
+
+export class RenderBusyError extends ApiError {
+  constructor() { super(429, 'Image rendering is busy. Try again shortly.'); }
+}
+type Job = { run: () => Promise<Uint8Array>; resolve: (image: Uint8Array) => void; reject: (reason: unknown) => void };
+/** One process-wide admission boundary for every server canvas renderer. */
+export function createRenderQueue({ concurrency = 2, maxPending = 16, maxPerUser = 2 } = {}) {
+  const pending = new Map<string, Promise<Uint8Array>>();
+  const owners = new Map<string, number>();
+  const waiting: Job[] = [];
+  let active = 0;
+  function pump() {
+    while (active < concurrency && waiting.length) {
+      const job = waiting.shift()!; active++;
+      // Yield before drawing so a burst of callers cannot monopolize this turn.
+      setTimeout(() => {
+        void Promise.resolve().then(job.run).then(job.resolve, job.reject).finally(() => { active--; pump(); });
+      }, 0);
+    }
+  }
+  return (userId: string, key: string, run: Job['run']): Promise<Uint8Array> => {
+    const scopedKey = JSON.stringify([userId, key]);
+    const existing = pending.get(scopedKey);
+    if (existing) return existing;
+    if (pending.size >= maxPending || (owners.get(userId) ?? 0) >= maxPerUser) return Promise.reject(new RenderBusyError());
+    owners.set(userId, (owners.get(userId) ?? 0) + 1);
+    const work = new Promise<Uint8Array>((resolve, reject) => { waiting.push({ run, resolve, reject }); }).finally(() => {
+      pending.delete(scopedKey);
+      const count = owners.get(userId)! - 1;
+      if (count) owners.set(userId, count); else owners.delete(userId);
+    });
+    pending.set(scopedKey, work); pump();
+    return work;
+  };
+}
+const schedule = createRenderQueue();
+export function renderImage(kind: string, userId: string, data: unknown, theme: string, run: Job['run']) {
+  const hash = createHash('sha256').update(JSON.stringify([kind, data, theme])).digest('hex');
+  return schedule(userId, hash, run);
+}