Pace engine reads under the proxy's rate limit

The proxy allows 10 requests a second per address and answers the rest
503. Limiting how many reads run at once still sent 40-50 a second, since
each answers in about 100 ms. Reads now share one pace of 8 a second,
actions skip the queue, and a 503 pauses every read for 2 seconds.
Results and Schedule share each schedule's production status, and an
archived schedule is read once until an action.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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)
+  })
+})

src/api/client.ts

@@ -54,6 +54,9 @@ export async function request<T>(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<T>(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<void> {
+      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

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)
+  })
+})

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<number>[] = []
+
+/** 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<T>(fn: (s: Schedule) => Promise<T[]>, pause = 3000) {
   const last = new Map<number, T[]>()
+  const settled = new Set<number>()
+  settledReads.push(settled)
   return async (s: Schedule): Promise<T[]> => {
+    // 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<T>(fn: (s: Schedule) => Promise<T[]>, 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)

src/express/schedules.ts

@@ -350,7 +350,19 @@ export function planProgress(plan: ContentPlan) {
   }
 }
 
-export const productionStatus = (id: number) => request<ProductionStatus>(`${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<number, { at: number; read: Promise<ProductionStatus> }>()
+export function productionStatus(id: number): Promise<ProductionStatus> {
+  const kept = productionReads.get(id)
+  if (kept && Date.now() - kept.at < PRODUCTION_FRESH) return kept.read
+  const read = request<ProductionStatus>(`${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.

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({
-  mutationCache: new MutationCache({ onSettled: () => queryClient.invalidateQueries({ queryKey: ['credits'] }) }),
+    // 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: {