From b79a72a81b0766423d4c66efbd9e64cbd45d64f3 Mon Sep 17 00:00:00 2001 From: Samu Date: Sat, 29 Aug 2026 16:05:29 +0200 Subject: [PATCH] feat: serializing queue with min-interval pacing --- server/src/queue.ts | 27 +++++++++++++++++++++++++ server/test/queue.test.ts | 42 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+) create mode 100644 server/src/queue.ts create mode 100644 server/test/queue.test.ts diff --git a/server/src/queue.ts b/server/src/queue.ts new file mode 100644 index 0000000..927686e --- /dev/null +++ b/server/src/queue.ts @@ -0,0 +1,27 @@ +export interface SerialQueueOptions { + /** Minimum delay between task starts. 0 = unpaced. */ + minIntervalMs?: number +} + +export class SerialQueue { + private tail: Promise = Promise.resolve() + private lastStart = 0 + private minIntervalMs: number + + constructor(opts: SerialQueueOptions = {}) { + this.minIntervalMs = opts.minIntervalMs ?? 0 + } + + run(fn: () => Promise): Promise { + const result = this.tail.then(async () => { + if (this.minIntervalMs > 0) { + const wait = this.lastStart + this.minIntervalMs - Date.now() + if (wait > 0) await new Promise((r) => setTimeout(r, wait)) + } + this.lastStart = Date.now() + return fn() + }) + this.tail = result.catch(() => {}) + return result + } +} diff --git a/server/test/queue.test.ts b/server/test/queue.test.ts new file mode 100644 index 0000000..3f7dadf --- /dev/null +++ b/server/test/queue.test.ts @@ -0,0 +1,42 @@ +import { describe, it, expect, vi } from 'vitest' +import { SerialQueue } from '../src/queue.js' + +describe('SerialQueue', () => { + it('runs tasks strictly in order even when they resolve out of order', async () => { + const q = new SerialQueue() + const order: number[] = [] + const slow = () => new Promise((r) => setTimeout(r, 20)).then(() => order.push(1)) + const fast = () => Promise.resolve().then(() => order.push(2)) + await Promise.all([q.run(slow), q.run(fast)]) + expect(order).toEqual([1, 2]) + }) + + it('continues after a failing task', async () => { + const q = new SerialQueue() + await expect( + q.run(async () => { + throw new Error('boom') + }) + ).rejects.toThrow('boom') + const result = await q.run(async () => 'ok') + expect(result).toBe('ok') + }) + + it('paces call starts at least minIntervalMs apart', async () => { + vi.useFakeTimers() + const q = new SerialQueue({ minIntervalMs: 1000 }) + let t1 = 0 + let t2 = 0 + const p1 = q.run(async () => { + t1 = Date.now() + }) + const p2 = q.run(async () => { + t2 = Date.now() + }) + await vi.advanceTimersByTimeAsync(1000) + await vi.advanceTimersByTimeAsync(1000) + await Promise.all([p1, p2]) + expect(t2 - t1).toBeGreaterThanOrEqual(1000) + vi.useRealTimers() + }) +})