diff --git a/src/express/results.test.ts b/src/express/results.test.ts index 52a0e81..e37f551 100644 --- a/src/express/results.test.ts +++ b/src/express/results.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest' -import { correctIndex, keepSettled, questionsOf, toMarkdown, withPlanItem, type Material } from './results' +import { correctIndex, keepSettled, questionsOf, settleLimited, toMarkdown, withPlanItem, type Material } from './results' import { sentencesOf, scriptLang } from '../components/AudioPlayer' describe('correctIndex', () => { @@ -66,6 +66,21 @@ describe('withPlanItem', () => { }) }) +describe('settleLimited', () => { + it('runs at most the limit at once and keeps order and failures', async () => { + let running = 0, peak = 0 + const out = await settleLimited([1, 2, 3, 4, 5, 6], 2, async n => { + peak = Math.max(peak, ++running) + await new Promise(r => setTimeout(r, 5)) + running-- + if (n === 3) throw new Error('503') + return n * 10 + }) + expect(peak).toBe(2) + expect(out.map(r => (r.status === 'fulfilled' ? r.value : 'x'))).toEqual([10, 20, 'x', 40, 50, 60]) + }) +}) + describe('keepSettled', () => { const ok = (v: number[]): PromiseSettledResult => ({ status: 'fulfilled', value: v }) const bad = (message: string): PromiseSettledResult => ({ status: 'rejected', reason: new Error(message) }) diff --git a/src/express/results.ts b/src/express/results.ts index c9540d9..56e45bc 100644 --- a/src/express/results.ts +++ b/src/express/results.ts @@ -182,12 +182,32 @@ export function keepSettled(results: PromiseSettledResult[]): T[] { return kept } +// Promise.allSettled over a function, running at most `limit` at a time. Results are loaded one +// request per schedule and polled; firing them all at once floods the engine, which then answers +// some with 503 and drops those schedules' results until the next poll. +export async function settleLimited(items: T[], limit: number, fn: (item: T) => Promise): Promise[]> { + const out: PromiseSettledResult[] = new Array(items.length) + let next = 0 + const worker = async () => { + while (next < items.length) { + const i = next++ + try { out[i] = { status: 'fulfilled', value: await fn(items[i]) } } + catch (reason) { out[i] = { status: 'rejected', reason } } + } + } + await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker)) + return out +} + +const PER_SCHEDULE_LIMIT = 4 + export async function listResults(): Promise<{ materials: Material[]; sources: FoundSource[] }> { const schedules = await expressAgents({ archived: true }) - const [materials, sources] = await Promise.all([ - Promise.allSettled(schedules.filter(s => s.rule.type === 'content').map(materialsOf)), - Promise.allSettled(schedules.filter(s => s.rule.type === 'research').map(sourcesOf)), - ]) + const content = schedules.filter(s => s.rule.type === 'content') + const research = schedules.filter(s => s.rule.type === 'research') + // One shared budget: research runs after content rather than alongside it. + const materials = await settleLimited(content, PER_SCHEDULE_LIMIT, materialsOf) + const sources = await settleLimited(research, PER_SCHEDULE_LIMIT, sourcesOf) const newest = (a: T, b: T) => b.createdAt.localeCompare(a.createdAt) return { materials: keepSettled(materials).sort(newest), sources: keepSettled(sources).sort(newest) } }