DeepSeek Harness Team
208 lines
10 KiB
TypeScript
208 lines
10 KiB
TypeScript
/** The durable effect queue: coalescing, per-kind limits, retries with backoff, health and shutdown. */
|
|
import { mkdtempSync, rmSync } from 'node:fs'
|
|
import { tmpdir } from 'node:os'
|
|
import { join } from 'node:path'
|
|
import { afterEach, beforeEach, expect, it, vi } from 'vitest'
|
|
import { Effects, Later } from '../src/app/effects.ts'
|
|
import { Store } from '../src/app/store.ts'
|
|
import { HealthBoard } from '../src/infra/health.ts'
|
|
import { deferred } from './fixtures/async.ts'
|
|
|
|
let directory: string
|
|
let store: Store
|
|
let clock: number
|
|
let board: HealthBoard
|
|
let health: Readonly<Record<string, unknown>>
|
|
let effects: Effects
|
|
beforeEach(() => {
|
|
directory = mkdtempSync(join(tmpdir(), 'effects-'))
|
|
store = new Store(join(directory, 'db.sqlite'))
|
|
clock = Date.parse('2026-09-09T00:00:00Z')
|
|
board = new HealthBoard()
|
|
health = board.all
|
|
effects = new Effects(store, board, () => clock)
|
|
effects.start()
|
|
})
|
|
afterEach(() => { store.close(); rmSync(directory, { recursive: true, force: true }) })
|
|
const relay = (text: string) => ({ text })
|
|
|
|
it('runs an effect queued inside a transaction once it commits, and none from a transaction that rolled back', async () => {
|
|
const run = vi.fn(async () => {})
|
|
effects.handle('relay', run)
|
|
store.atomic(() => { effects.enqueue('relay', 'kept', relay('a')) })
|
|
expect(() => store.atomic(() => { effects.enqueue('relay', 'dropped', relay('b')); throw new Error('rolled back') })).toThrow('rolled back')
|
|
await effects.idle()
|
|
expect(run.mock.calls).toEqual([[relay('a'), 'kept']])
|
|
expect(effects.pending('relay')).toBe(false)
|
|
})
|
|
|
|
it('keeps one effect per key with the latest payload, and runs it again when a new payload arrives while it runs', async () => {
|
|
const hold = deferred()
|
|
const run = vi.fn(async (payload: { text: string }) => { if (payload.text === 'first') await hold.promise })
|
|
effects.handle('relay', run)
|
|
effects.enqueue('relay', 'k', relay('first'))
|
|
await vi.waitFor(() => expect(run).toHaveBeenCalledTimes(1))
|
|
effects.enqueue('relay', 'k', relay('second')); effects.enqueue('relay', 'k', relay('third'))
|
|
hold.resolve()
|
|
await effects.idle()
|
|
expect(run.mock.calls.map(([payload]) => payload.text)).toEqual(['first', 'third'])
|
|
})
|
|
|
|
it('runs a kind oldest first, as many at once as its limit allows, and holds it while the limit is 0', async () => {
|
|
const holds = new Map<string, ReturnType<typeof deferred<void>>>()
|
|
let limit = 2
|
|
const run = vi.fn(async (_payload: unknown, key: string) => { const hold = deferred(); holds.set(key, hold); await hold.promise })
|
|
effects.handle('launch', run, { limit: () => limit })
|
|
for (const key of ['a', 'b', 'c']) effects.enqueue('launch', key, { ticketId: key, assignmentId: key })
|
|
await vi.waitFor(() => expect(run.mock.calls.map(([, key]) => key)).toEqual(['a', 'b']))
|
|
limit = 0
|
|
holds.get('a')!.resolve()
|
|
await vi.waitFor(() => expect(effects.pending('launch', 'a')).toBe(false))
|
|
effects.kick(); await Promise.resolve()
|
|
expect(run).toHaveBeenCalledTimes(2)
|
|
limit = 2
|
|
holds.get('b')!.resolve()
|
|
await vi.waitFor(() => expect(run.mock.calls.map(([, key]) => key)).toEqual(['a', 'b', 'c']))
|
|
holds.get('c')!.resolve()
|
|
await effects.idle()
|
|
})
|
|
|
|
it('keeps an effect\'s place in its queue when it is queued again with a new payload', async () => {
|
|
const run = vi.fn(async (_payload: { ticketId: string; assignmentId: string }, _key: string) => {})
|
|
let room = 0
|
|
effects.handle('launch', run, { limit: () => room })
|
|
store.atomic(() => {
|
|
effects.enqueue('launch', 'first', { ticketId: 't1', assignmentId: 'a1' })
|
|
effects.enqueue('launch', 'second', { ticketId: 't2', assignmentId: 'a2' })
|
|
effects.enqueue('launch', 'first', { ticketId: 't1', assignmentId: 'a1' })
|
|
})
|
|
await effects.idle()
|
|
room = 1
|
|
await effects.idle()
|
|
expect(run.mock.calls.map(([, key]) => key)).toEqual(['first', 'second'])
|
|
})
|
|
|
|
it('retries a failed effect after a delay that doubles up to its maximum, and reports the kind in error after three failures', async () => {
|
|
let fail = true
|
|
const run = vi.fn(async () => { if (fail) throw new Error('session busy') })
|
|
effects.handle('notice', run, { retry: [1000, 3000] })
|
|
effects.enqueue('notice', 'prompts', { value: { prompt_config_saved: 2 }, summary: 'Prompt configuration saved' })
|
|
await effects.idle()
|
|
expect(health.notice).toMatchObject({ state: 'ready', detail: '1 waiting since 2026-09-09T00:00:00.000Z' })
|
|
for (const wait of [1000, 2000, 3000, 3000]) {
|
|
clock += wait - 1; await effects.idle()
|
|
const calls = run.mock.calls.length
|
|
clock += 1; await effects.idle()
|
|
expect(run).toHaveBeenCalledTimes(calls + 1)
|
|
if (wait === 2000) expect(health.notice).toMatchObject({ state: 'error', detail: expect.stringContaining('1 waiting since 2026-09-09T00:00:00.000Z; the last failure: Error: session busy') })
|
|
}
|
|
fail = false
|
|
clock += 3000; await effects.idle()
|
|
expect(health.notice).toMatchObject({ state: 'ready', detail: 'Nothing waiting' })
|
|
expect(effects.pending('notice')).toBe(false)
|
|
})
|
|
|
|
it('starts nothing once drained, keeps what waits and what shutdown cut short, and runs it after the next start', async () => {
|
|
const hold = deferred()
|
|
effects.handle('relay', async () => { await hold.promise; throw new Error('aborted') })
|
|
effects.enqueue('relay', 'cut', relay('a'))
|
|
await vi.waitFor(() => expect(effects.pending('relay', 'cut')).toBe(true))
|
|
await Promise.resolve()
|
|
const drained = effects.drain()
|
|
effects.enqueue('relay', 'later', relay('b'))
|
|
hold.resolve(); await drained
|
|
const next = new Effects(store, new HealthBoard(), () => clock)
|
|
next.start()
|
|
const run = vi.fn(async (_payload: { text: string }, _key: string) => {})
|
|
next.handle('relay', run)
|
|
await next.idle()
|
|
expect(run.mock.calls.map(([, key]) => key)).toEqual(['cut', 'later'])
|
|
})
|
|
|
|
it('lets a handler defer an effect without a failure, until a later dispatch after its first delay', async () => {
|
|
let later = true
|
|
const run = vi.fn(async () => { if (later) throw new Later('paused') })
|
|
effects.handle('launch', run, { retry: [5000, 60_000] })
|
|
effects.enqueue('launch', 'a', { ticketId: 't', assignmentId: 'a' })
|
|
await effects.idle()
|
|
expect(health.launch).toMatchObject({ state: 'ready', detail: '1 waiting since 2026-09-09T00:00:00.000Z' })
|
|
later = false
|
|
clock += 4999; await effects.idle()
|
|
expect(run).toHaveBeenCalledOnce()
|
|
clock += 1; await effects.idle()
|
|
expect(run).toHaveBeenCalledTimes(2)
|
|
expect(effects.pending('launch')).toBe(false)
|
|
})
|
|
|
|
it('runs an ordered kind one at a time in the order queued, and holds the later effects while an earlier one fails', async () => {
|
|
const failing = new Set(['first'])
|
|
const run = vi.fn(async (_payload: { text: string }, key: string) => { if (failing.has(key)) throw new Error('offline') })
|
|
effects.handle('relay', run, { ordered: true, limit: () => 4, retry: [1000, 1000] })
|
|
effects.enqueue('relay', 'first', relay('a')); effects.enqueue('relay', 'second', relay('b'))
|
|
await effects.idle()
|
|
clock += 1000; await effects.idle()
|
|
expect(run.mock.calls.map(([, key]) => key)).toEqual(['first', 'first'])
|
|
failing.clear()
|
|
clock += 1000; await effects.idle()
|
|
expect(run.mock.calls.map(([, key]) => key)).toEqual(['first', 'first', 'first', 'second'])
|
|
})
|
|
|
|
it('keeps a quiet kind\'s failures out of its health, and asks no limit while nothing is due', async () => {
|
|
const limit = vi.fn(() => 4)
|
|
effects.handle('delivery', async () => { throw new Error('Session gone') }, { quiet: true, limit, retry: [1000, 1000] })
|
|
effects.kick(); await effects.idle()
|
|
expect(limit).not.toHaveBeenCalled()
|
|
effects.enqueue('delivery', 't', { ticketId: 't' })
|
|
for (let attempt = 0; attempt < 4; attempt++) { await effects.idle(); clock += 1000 }
|
|
expect(health.delivery).toMatchObject({ state: 'ready', detail: '1 waiting since 2026-09-09T00:00:00.000Z' })
|
|
})
|
|
|
|
it('keeps an effect cancelled and queued again while its first run goes on, which the first run does not end', async () => {
|
|
const hold = deferred<void>()
|
|
const run = vi.fn(async (_payload: { text: string }, _key: string) => { if (run.mock.calls.length === 1) await hold.promise })
|
|
effects.handle('relay', run)
|
|
effects.enqueue('relay', 'k', relay('first'))
|
|
await vi.waitFor(() => expect(run).toHaveBeenCalledOnce())
|
|
effects.cancel('relay', 'k')
|
|
effects.enqueue('relay', 'k', relay('second'))
|
|
hold.resolve()
|
|
await effects.idle()
|
|
expect(run.mock.calls.map(([payload]) => payload.text)).toEqual(['first', 'second'])
|
|
expect(effects.pending('relay')).toBe(false)
|
|
})
|
|
|
|
it('queues an effect after a restart behind the ones that waited, for a kind that keeps its order', async () => {
|
|
const hold = deferred<void>()
|
|
effects.handle('relay', async () => { await hold.promise; throw new Error('aborted') }, { ordered: true })
|
|
effects.enqueue('relay', 'first', relay('a'))
|
|
effects.enqueue('relay', 'second', relay('b'))
|
|
await vi.waitFor(() => expect(effects.pending('relay', 'first')).toBe(true))
|
|
const drained = effects.drain()
|
|
hold.resolve(); await drained
|
|
// Sequence numbers continue across restarts so newly queued effects run after existing pending ones.
|
|
const next = new Effects(store, new HealthBoard(), () => clock)
|
|
next.enqueue('relay', 'third', relay('c'))
|
|
next.start()
|
|
const run = vi.fn(async (_payload: { text: string }, _key: string) => {})
|
|
next.handle('relay', run, { ordered: true })
|
|
await next.idle()
|
|
expect(run.mock.calls.map(([, key]) => key)).toEqual(['first', 'second', 'third'])
|
|
})
|
|
|
|
it('reports a failure of the queue itself under its kind, until a dispatch of that kind succeeds again', async () => {
|
|
let broken = true
|
|
effects.handle('relay', async () => {}, { limit: () => { if (broken) throw new Error('database is locked'); return 1 } })
|
|
effects.enqueue('relay', 'k', relay('a'))
|
|
await effects.idle()
|
|
expect(health.relay).toMatchObject({ state: 'error', detail: 'The effect queue failed: Error: database is locked' })
|
|
broken = false
|
|
effects.kick()
|
|
await effects.idle()
|
|
expect(health.relay).toMatchObject({ state: 'ready', detail: 'Nothing waiting' })
|
|
})
|
|
|
|
it('refuses a second handler for a kind', () => {
|
|
effects.handle('relay', async () => {})
|
|
expect(() => effects.handle('relay', async () => {})).toThrow('The relay effects already have a handler')
|
|
})
|