diff --git a/src/api/client.test.ts b/src/api/client.test.ts index a984cde..54683df 100644 --- a/src/api/client.test.ts +++ b/src/api/client.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest' -import { limiter } from './client' +import { limiter, pacer } from './client' describe('limiter', () => { it('runs at most n calls at once, all of them, failures included', async () => { @@ -17,3 +17,23 @@ describe('limiter', () => { expect(out.map(r => (r.status === 'fulfilled' ? r.value : 'x'))).toEqual([1, 'x', 3, 4, 5]) }) }) + +describe('pacer', () => { + it('lets a burst through, then spaces the rest out in order', async () => { + const pace = pacer(100, 2) + const t0 = Date.now() + const order: number[] = [] + await Promise.all([1, 2, 3, 4, 5, 6].map(n => pace.wait().then(() => order.push(n)))) + // Two at once, then four more at 10 ms each. + expect(Date.now() - t0).toBeGreaterThanOrEqual(35) + expect(order).toEqual([1, 2, 3, 4, 5, 6]) + }) + + it('pauses everything after the proxy turns one away', async () => { + const pace = pacer(100, 5) + pace.pause(50) + const t0 = Date.now() + await pace.wait() + expect(Date.now() - t0).toBeGreaterThanOrEqual(45) + }) +}) diff --git a/src/api/client.ts b/src/api/client.ts index 72f3a6d..5a5424c 100644 --- a/src/api/client.ts +++ b/src/api/client.ts @@ -54,6 +54,9 @@ export async function request(path: string, { json, headers, as = 'json', ... body = JSON.stringify(json) } + if (!opts.method || opts.method === 'GET') await pace.wait() + else pace.take() + let res: Response try { res = await fetch(BASE + path, { ...opts, headers: h, body }) @@ -66,7 +69,10 @@ export async function request(path: string, { json, headers, as = 'json', ... if (!res.ok) { if (res.status === 502 || res.status === 504) throw new EngineError(t('err.unreachable'), res.status) // The proxy's rate limit answers 503 (or 429) with an HTML page, not the engine's JSON. - if ((res.status === 503 || res.status === 429) && !data) throw new EngineError(t('err.busy'), res.status) + if ((res.status === 503 || res.status === 429) && !data) { + pace.pause(BUSY_PAUSE) + throw new EngineError(t('err.busy'), res.status) + } const err = new EngineError(detail(data?.detail), res.status, data && typeof data === 'object' ? data : null) if (res.status === 402) err.message = creditMessage(err) throw err @@ -79,6 +85,47 @@ export const opKey = (prefix: string) => `lle-${prefix}-${Date.now()}-${Math.ran export const sleep = (ms: number) => new Promise(r => setTimeout(r, ms)) +/** Spaces requests out to at most `perSecond` after a first `burst`. A limit on how many run at once + * does not do this: the engine answers in about 100 ms, so four at a time is still 40 a second. */ +export function pacer(perSecond: number, burst: number) { + let tokens = burst + let last = Date.now() + let queue = Promise.resolve() + const refill = () => { + const now = Date.now() + tokens = Math.min(burst, tokens + ((now - last) * perSecond) / 1000) + last = now + } + return { + /** Waits for a turn; callers are served in the order they asked. */ + wait(): Promise { + const turn = queue.then(async () => { + for (refill(); tokens < 1; refill()) await sleep(((1 - tokens) * 1000) / perSecond) + tokens -= 1 + }) + queue = turn + return turn + }, + /** Takes a turn at once, ahead of the waiting ones, and the queue pays it back. */ + take() { + refill() + tokens -= 1 + }, + /** Sends nothing more for `ms`. */ + pause(ms: number) { + refill() + tokens = Math.min(tokens, (-ms * perSecond) / 1000) + }, + } +} + +// The engine's proxy allows 10 requests a second per address after a burst of 20, and answers the +// rest 503. Everything the app reads shares one pace below that; an action the user takes (any +// non-GET) is sent at once rather than queued behind a page's polling. When the proxy turns one away +// anyway (e.g. another tab sharing the address), every request waits for it to recover. +const pace = pacer(8, 10) +const BUSY_PAUSE = 2000 + /** Runs the calls it is given at most `n` at a time, for requests a page makes one per row. */ export function limiter(n: number) { let running = 0 diff --git a/src/express/results.test.ts b/src/express/results.test.ts index dd3d99b..919654f 100644 --- a/src/express/results.test.ts +++ b/src/express/results.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it, vi } from 'vitest' import { EngineError } from '../api/client' -import { correctIndex, keepSettled, questionsOf, remembering, settleLimited, toMarkdown, withPlanItem, type Material } from './results' +import { correctIndex, forgetReads, keepSettled, questionsOf, remembering, settleLimited, toMarkdown, withPlanItem, type Material } from './results' import { sentencesOf, scriptLang } from '../components/AudioPlayer' import type { Schedule } from './schedules' @@ -129,3 +129,18 @@ describe('keepSettled', () => { expect(keepSettled([])).toEqual([]) }) }) + +describe('remembering an archived schedule', () => { + const archived = { agent: { id: 7, status: 'archived' } } as Schedule + + it('reads it once until an action', async () => { + const fn = vi.fn(async () => [1]) + const read = remembering(fn, 0) + await read(archived) + expect(await read(archived)).toEqual([1]) + expect(fn).toHaveBeenCalledTimes(1) + forgetReads() + await read(archived) + expect(fn).toHaveBeenCalledTimes(2) + }) +}) diff --git a/src/express/results.ts b/src/express/results.ts index 995fa61..06a9358 100644 --- a/src/express/results.ts +++ b/src/express/results.ts @@ -5,7 +5,7 @@ import { EngineError, request, sleep } from '../api/client' import { t } from '../i18n' import { kindLabel, type ContentKind, type Rule } from './parse' -import { expressAgents, MAX_ATTEMPTS, PLAN_ACTIVE, productionStatus, retryItem, runNow, type PlanItem, type Schedule } from './schedules' +import { expressAgents, forgetProductionReads, MAX_ATTEMPTS, PLAN_ACTIVE, productionStatus, retryItem, runNow, type PlanItem, type Schedule } from './schedules' const AGENTS = '/organizations/current/agents' @@ -206,13 +206,27 @@ const PER_SCHEDULE_LIMIT = 4 // The proxy's rate limit (503/429), a 5xx or no answer at all passes; a 4xx (e.g. a deleted agent) does not. const passing = (e: unknown) => e instanceof EngineError && (e.status === 0 || e.status === 429 || e.status >= 500) +const settledReads: Set[] = [] + +/** After any action: an archived schedule's results (a decision on one) and production plans + * (a started or retried piece) may have changed, so the next poll reads them again. */ +export function forgetReads() { + for (const s of settledReads) s.clear() + forgetProductionReads() +} + /** Wraps a per-schedule read so a passing failure does not change what the page shows: a busy * engine is asked once more after a pause, and if it still fails, the schedule's results from the * last poll that read them are used. Dropping them instead made every count on the page dip for * one poll and come back on the next. */ export function remembering(fn: (s: Schedule) => Promise, pause = 3000) { const last = new Map() + const settled = new Set() + settledReads.push(settled) return async (s: Schedule): Promise => { + // An archived schedule makes nothing new, so after one read it costs nothing until an action. + const archived = s.agent.status === 'archived' + if (archived && settled.has(s.agent.id)) return last.get(s.agent.id)! try { const read = await fn(s).catch(async e => { if (!(e instanceof EngineError && (e.status === 503 || e.status === 429))) throw e @@ -220,6 +234,7 @@ export function remembering(fn: (s: Schedule) => Promise, pause = 3000) return fn(s) }) last.set(s.agent.id, read) + if (archived) settled.add(s.agent.id) return read } catch (e) { const kept = last.get(s.agent.id) diff --git a/src/express/schedules.ts b/src/express/schedules.ts index 07c26e0..a474b29 100644 --- a/src/express/schedules.ts +++ b/src/express/schedules.ts @@ -350,7 +350,19 @@ export function planProgress(plan: ContentPlan) { } } -export const productionStatus = (id: number) => request(`${AGENTS}/${id}/content`) +// The Schedule page (one query per row) and Results (one read per schedule) both poll this; a read +// from the last half minute, or one still on its way, answers both. forgetReads drops them after an action. +const PRODUCTION_FRESH = 30_000 +const productionReads = new Map }>() +export function productionStatus(id: number): Promise { + const kept = productionReads.get(id) + if (kept && Date.now() - kept.at < PRODUCTION_FRESH) return kept.read + const read = request(`${AGENTS}/${id}/content`) + productionReads.set(id, { at: Date.now(), read }) + read.catch(() => { if (productionReads.get(id)?.read === read) productionReads.delete(id) }) + return read +} +export const forgetProductionReads = () => productionReads.clear() // Produce now = the same plan the 09:00 tick would create, just earlier. The engine's tick then // approves and starts it within a minute because Express schedules run in automatic mode. diff --git a/src/main.tsx b/src/main.tsx index fdb412d..d8d6ae3 100644 --- a/src/main.tsx +++ b/src/main.tsx @@ -6,6 +6,7 @@ import { MutationCache, QueryClient, QueryClientProvider } from '@tanstack/react import { BrowserRouter } from 'react-router-dom' import { EngineError } from './api/client' import App from './App' +import { forgetReads } from './express/results' import { useLang } from './i18n' import './styles/mockup.css' import './styles/app.css' @@ -17,8 +18,12 @@ if (!clerkPubKey) { } const queryClient: QueryClient = new QueryClient({ - // Any action may have spent workspace credits, so the topbar balance is read again after each one. - mutationCache: new MutationCache({ onSettled: () => queryClient.invalidateQueries({ queryKey: ['credits'] }) }), + mutationCache: new MutationCache({ + // Runs before the action's own onSuccess, whose invalidations must not get a remembered read. + onSuccess: forgetReads, + // Any action may have spent workspace credits, so the topbar balance is read again after each one. + onSettled: () => queryClient.invalidateQueries({ queryKey: ['credits'] }), + }), defaultOptions: { // Retrying a 4xx (auth, credits, validation) only repeats the same answer. queries: {