feat: serializing queue with min-interval pacing
This commit is contained in:
27
server/src/queue.ts
Normal file
27
server/src/queue.ts
Normal file
@@ -0,0 +1,27 @@
|
||||
export interface SerialQueueOptions {
|
||||
/** Minimum delay between task starts. 0 = unpaced. */
|
||||
minIntervalMs?: number
|
||||
}
|
||||
|
||||
export class SerialQueue {
|
||||
private tail: Promise<unknown> = Promise.resolve()
|
||||
private lastStart = 0
|
||||
private minIntervalMs: number
|
||||
|
||||
constructor(opts: SerialQueueOptions = {}) {
|
||||
this.minIntervalMs = opts.minIntervalMs ?? 0
|
||||
}
|
||||
|
||||
run<T>(fn: () => Promise<T>): Promise<T> {
|
||||
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
|
||||
}
|
||||
}
|
||||
42
server/test/queue.test.ts
Normal file
42
server/test/queue.test.ts
Normal file
@@ -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()
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user