diff --git a/docs/observability.md b/docs/observability.md index 84a80f21..86e71cb7 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -411,8 +411,12 @@ The daemon also serves two stores agents can use at runtime: namespaced key-values scoped `private` (the writing agent only) or `fleet` (every agent on the daemon). -Reporting is fire-and-forget: a down daemon drops and counts events, it -never blocks a run. Eval scores flow through the same pipe — +Reporting is fire-and-forget and never blocks a run: when the daemon is +down, the failed batch is requeued and retried with exponential backoff (1 s, +2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. Only +events evicted past the buffer cap (`maxBuffered`, default 5000, oldest +first) or still unsent after the exit flush's final attempt are dropped and +counted in the reporter's `dropped`. Eval scores flow through the same pipe — `lousho eval --observe` emits them even when `LOUSHO_OBSERVE` is unset. ## The built server diff --git a/llms-full.txt b/llms-full.txt index 2a367a48..e1acb526 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -14071,8 +14071,12 @@ The daemon also serves two stores agents can use at runtime: namespaced key-values scoped `private` (the writing agent only) or `fleet` (every agent on the daemon). -Reporting is fire-and-forget: a down daemon drops and counts events, it -never blocks a run. Eval scores flow through the same pipe — +Reporting is fire-and-forget and never blocks a run: when the daemon is +down, the failed batch is requeued and retried with exponential backoff (1 s, +2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. Only +events evicted past the buffer cap (`maxBuffered`, default 5000, oldest +first) or still unsent after the exit flush's final attempt are dropped and +counted in the reporter's `dropped`. Eval scores flow through the same pipe — `lousho eval --observe` emits them even when `LOUSHO_OBSERVE` is unset. ## The built server diff --git a/src/observe/reporter.test.ts b/src/observe/reporter.test.ts index f8dda214..281633d6 100644 --- a/src/observe/reporter.test.ts +++ b/src/observe/reporter.test.ts @@ -215,6 +215,188 @@ describe('observeExporter', () => { } }); + describe('requeue with backoff', () => { + /** A fetch whose n-th call answers `outcomes[n]` (the last outcome repeats); every attempt is recorded. */ + function scriptedFetch(outcomes: Array<'fail' | 'ok'>): { fetchFn: typeof fetch; posted: PostedBatch[] } { + const posted: PostedBatch[] = []; + const fetchFn = vi.fn(async (input: string | URL | Request, init?: RequestInit): Promise => { + const body = JSON.parse(String(init?.body)) as { batch: IngestEvent[] }; + const outcome = outcomes[Math.min(posted.length, outcomes.length - 1)]; + posted.push({ url: String(input), events: body.batch }); + if (outcome === 'fail') throw new TypeError('fetch failed'); + return new Response('{}', { status: 202 }); + }); + return { fetchFn: fetchFn as unknown as typeof fetch, posted }; + } + + const emit = (r: ObserveReporter, ids: string[]) => { + for (const id of ids) { + const span = makeSpan({ id }); + r.onSpanStart(span); + r.onSpanEnd(span); + } + }; + + const silenceWarn = () => vi.spyOn(console, 'warn').mockImplementation(() => undefined); + let warn: ReturnType; + beforeEach(() => { + vi.useFakeTimers(); + warn = silenceWarn(); + }); + afterEach(() => { + vi.useRealTimers(); + warn.mockRestore(); + }); + + it('requeues a failed batch and delivers it on the next flush', async () => { + const { fetchFn, posted } = scriptedFetch(['fail', 'ok']); + const { r } = reporter({ fetchFn, flushIntervalMs: 1000 }); + emit(r, ['a', 'b', 'c']); + await vi.advanceTimersByTimeAsync(1000); // first flush: rejected + expect(posted).toHaveLength(1); + await vi.advanceTimersByTimeAsync(1000); // second flush, after the 1s backoff + expect(posted).toHaveLength(2); + expect(posted[1]!.events).toHaveLength(3); + expect(posted[1]!.events.map((e) => e.type === 'observation' && e.data.obs_id)).toEqual(['a', 'b', 'c']); + expect(r.dropped).toBe(0); + expect(warn).toHaveBeenCalledTimes(1); + }); + + it('evicts oldest events past maxBuffered and counts them', async () => { + const { fetchFn, posted } = scriptedFetch(['fail']); + const { r } = reporter({ fetchFn, maxBuffered: 4, flushIntervalMs: 1000 }); + emit(r, ['s0', 's1', 's2', 's3', 's4', 's5']); + await vi.advanceTimersByTimeAsync(1000); // flush: rejected, requeued + expect(r.dropped).toBe(2); + await vi.advanceTimersByTimeAsync(1000); // the retry carries what is still pending + expect(posted).toHaveLength(2); + expect(posted[1]!.events.map((e) => e.type === 'observation' && e.data.obs_id)).toEqual(['s2', 's3', 's4', 's5']); + expect(r.dropped).toBe(2); + }); + + it('trims a requeued batch from the front when new events arrived during the failed POST', async () => { + const posted: PostedBatch[] = []; + let reject!: (error: Error) => void; + const fetchFn = vi.fn(async (input: string | URL | Request, init?: RequestInit): Promise => { + posted.push({ url: String(input), events: (JSON.parse(String(init?.body)) as { batch: IngestEvent[] }).batch }); + if (posted.length === 1) return new Promise((_, rej) => (reject = rej)); + return new Response('{}', { status: 202 }); + }) as unknown as typeof fetch; + const { r } = reporter({ fetchFn, maxBuffered: 4, flushSize: 3, flushIntervalMs: 1000 }); + emit(r, ['a', 'b', 'c']); // size flush: in flight + await vi.advanceTimersByTimeAsync(0); + emit(r, ['d', 'e']); // arrive while the POST hangs + reject(new TypeError('fetch failed')); + await vi.advanceTimersByTimeAsync(0); + expect(r.dropped).toBe(1); // a, b, c, d, e requeued into 4 slots: the oldest goes + await vi.advanceTimersByTimeAsync(1000); + expect(posted.at(-1)!.events.map((e) => e.type === 'observation' && e.data.obs_id)).toEqual(['b', 'c', 'd', 'e']); + expect(r.dropped).toBe(1); + }); + + it('backs off between failed attempts', async () => { + const { fetchFn, posted } = scriptedFetch(['fail']); + const { r } = reporter({ fetchFn, flushIntervalMs: 100, flushSize: 2 }); + emit(r, ['a']); + await vi.advanceTimersByTimeAsync(100); // attempt 1 + expect(posted).toHaveLength(1); + let failedAt = Date.now(); + // n = failures so far - 1: 1s, 2s, 4s, 8s, 16s, then the 30s cap + for (let n = 0; n < 7; n++) { + const retryAt = failedAt + Math.min(2 ** n * 1000, 30_000); + emit(r, [`x${n}`, `y${n}`]); // a size flush inside the window must not retry either + await vi.advanceTimersByTimeAsync(retryAt - 1 - Date.now()); + expect(posted).toHaveLength(n + 1); + await vi.advanceTimersByTimeAsync(1); + expect(posted).toHaveLength(n + 2); + failedAt = Date.now(); + } + expect(r.dropped).toBe(0); + }); + + it('close() makes one final attempt despite the backoff and counts leftovers as dropped', async () => { + const { fetchFn, posted } = scriptedFetch(['fail']); + const { r } = reporter({ fetchFn, flushIntervalMs: 1000 }); + emit(r, ['a', 'b']); + await vi.advanceTimersByTimeAsync(1000); // rejected; next attempt not before t=2000 + expect(r.dropped).toBe(0); + await r.close(); + expect(posted).toHaveLength(2); + expect(posted[1]!.events.map((e) => e.type)).toEqual(['observation', 'observation', 'agent.deregister']); + expect(r.dropped).toBe(3); + }); + + it('requeues a batch too large to spread into call arguments, and the flush chain survives', async () => { + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => unhandled.push(reason); + process.on('unhandledRejection', onUnhandled); + try { + // Count events without JSON.parse-ing a ~50MB body twice. + const sizes: number[] = []; + const fetchFn = vi.fn(async (_input: string | URL | Request, init?: RequestInit): Promise => { + sizes.push(String(init?.body).split('"type":').length - 1); + if (sizes.length === 1) throw new TypeError('fetch failed'); + return new Response('{}', { status: 202 }); + }) as unknown as typeof fetch; + const total = 500_000; // past the spread limit even on a worker thread's larger stack (200k fits there) + const { r } = reporter({ fetchFn, maxBuffered: total, flushSize: total + 1, flushIntervalMs: 1000 }); + for (let i = 0; i < total; i++) r.onSpanEnd({ id: `s${i}`, name: 'n', attributes: {}, startTime: 1, endTime: 2 }); + await vi.advanceTimersByTimeAsync(1000); // rejected: the whole batch is requeued + await vi.advanceTimersByTimeAsync(1000); // the retry still runs and carries all of it + expect(sizes).toEqual([total, total]); + expect(r.dropped).toBe(0); + await r.close(); + expect(sizes).toHaveLength(3); // the chain still runs close()'s flush (the deregister) + await vi.advanceTimersByTimeAsync(10); // let any stray rejection surface + expect(unhandled).toEqual([]); + } finally { + process.off('unhandledRejection', onUnhandled); + } + }, 60_000); + }); + + it('clamps maxBuffered to 1 so the deregister survives', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); + try { + const { r, posted } = reporter({ maxBuffered: 0 }); + const span = makeSpan(); + r.onSpanStart(span); + r.onSpanEnd(span); // fits in the single slot + await r.close(); // the deregister evicts it + expect(allEvents(posted).map((e) => e.type)).toEqual(['agent.deregister']); + expect(r.dropped).toBe(1); + } finally { + warn.mockRestore(); + } + }); + + it('close() aborts a regular flush in flight and sends its batch in the final attempt', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); + try { + const posted: PostedBatch[] = []; + const fetchFn = vi.fn(async (input: string | URL | Request, init?: RequestInit): Promise => { + posted.push({ url: String(input), events: (JSON.parse(String(init?.body)) as { batch: IngestEvent[] }).batch }); + if (posted.length > 1) return new Response('{}', { status: 202 }); + // The first POST hangs until its signal fires (a daemon that accepted the socket but never answers). + return new Promise((_, reject) => init?.signal?.addEventListener('abort', () => reject(init.signal?.reason))); + }) as unknown as typeof fetch; + const { r } = reporter({ fetchFn, flushSize: 1, requestTimeoutMs: 60_000 }); + const span = makeSpan({ id: 'hung' }); + r.onSpanStart(span); + r.onSpanEnd(span); // size flush: in flight, hanging + await vi.waitFor(() => expect(posted).toHaveLength(1)); + const started = Date.now(); + await r.close(); + expect(Date.now() - started).toBeLessThan(1000); + expect(posted).toHaveLength(2); + expect(posted[1]!.events.map((e) => (e.type === 'observation' ? e.data.obs_id : e.type))).toEqual(['hung', 'agent.deregister']); + expect(r.dropped).toBe(0); + expect(warn).not.toHaveBeenCalled(); // a healthy daemon: no outage warning for the aborted flush + } finally { + warn.mockRestore(); + } + }); + it('emits a heartbeat immediately when heartbeats are on', async () => { const { r, posted } = reporter({ heartbeatIntervalMs: 60_000 }); await r.close(); diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts index 30876090..eca5d4b2 100644 --- a/src/observe/reporter.ts +++ b/src/observe/reporter.ts @@ -12,10 +12,14 @@ * Rules it lives by (the tracing-insights spec): * - One event per FINISHED span (`start`+`end`); `obs_id` is the span's own * id (the `TraceLine.id`), so a resent batch upserts by key. - * - Flush on size (50), on interval (5s), and on process exit (~2s cap). + * - Flush on size (50), on interval (5s), and on process exit (~2s cap: + * close() aborts a regular flush still in flight and makes one final + * attempt with the 2s timeout). * Crash loss is accepted and bounded: nothing persists mid run. - * - The daemon being down, or a full buffer, drops events into a `dropped` - * counter and warns ONCE - the reporter never blocks or throws into a run. + * - The daemon being down requeues the failed batch and retries with + * exponential backoff (1s base, 30s cap); only events evicted past the + * buffer cap, or still unsent after close()'s final attempt, drop into a + * `dropped` counter. It warns ONCE and never blocks or throws into a run. * Timers are unref'd so observation never keeps a process alive. * - Its own HTTP calls are plain `fetch`, outside any span: instrumenting * them would recurse. @@ -57,9 +61,9 @@ export interface ObserveExporterOptions { flushIntervalMs?: number; /** Heartbeat cadence (default 60000ms); `0` disables heartbeats. */ heartbeatIntervalMs?: number; - /** Buffer cap; events past it drop into `dropped` (default 5000). */ + /** Buffer cap; past it the oldest events are evicted into `dropped` (default 5000, floor 1). */ maxBuffered?: number; - /** Per-request timeout (default 5000ms); the exit flush always caps at ~2000ms. */ + /** Per-request timeout (default 5000ms); close() aborts such a request and caps its final attempt at ~2000ms. */ requestTimeoutMs?: number; /** Project dir for `.lousho/identity/`; default `process.cwd()`. */ cwd?: string; @@ -81,7 +85,7 @@ export interface ObserveExporterOptions { } export interface ObserveReporter extends TraceExporter { - /** Events dropped because the daemon was unreachable or the buffer was full. */ + /** Events lost for good: evicted past the buffer cap, or still unsent after close()'s final attempt. */ readonly dropped: number; /** This reporter's persisted `agent_id`. */ readonly agentId: string; @@ -104,6 +108,57 @@ const DEFAULT_HEARTBEAT_MS = 60_000; const DEFAULT_MAX_BUFFERED = 5000; const DEFAULT_REQUEST_TIMEOUT_MS = 5000; const EXIT_FLUSH_TIMEOUT_MS = 2000; +const RETRY_BASE_MS = 1000; +const RETRY_CAP_MS = 30_000; + +/** Wait before the next send after `failures` consecutive failed ones (0-based): 1s, 2s, 4s ... capped at 30s. */ +const retryDelayMs = (failures: number): number => Math.min(RETRY_BASE_MS * 2 ** failures, RETRY_CAP_MS); + +/** + * The reporter's pending events: FIFO, capped at `cap` (the oldest are + * evicted first, reported through `onEvict`), and gated by a retry backoff + * after a failed send. Every send path - size, interval, exit - goes through + * `take` / `delivered` / `failed`, so a failed batch is never lost while + * there is room to keep it. + */ +function createOutbox(cap: number, now: () => number, onEvict: (count: number) => void) { + let items: IngestEvent[] = []; + let failures = 0; + let retryAt = 0; + const trim = (): void => { + const over = items.length - cap; + if (over <= 0) return; + items.splice(0, over); + onEvict(over); + }; + return { + get size(): number { + return items.length; + }, + /** False while backing off from a failed send. */ + ready(): boolean { + return now() >= retryAt; + }, + push(event: IngestEvent): void { + items.push(event); + trim(); + }, + /** Everything pending, for one send - or nothing while backing off, unless `force` (the final attempt). */ + take(force = false): IngestEvent[] { + return force || now() >= retryAt ? items.splice(0, items.length) : []; + }, + delivered(): void { + failures = 0; + retryAt = 0; + }, + /** Put a failed batch back in front of anything queued since, trim to the cap, and back off. */ + failed(batch: IngestEvent[]): void { + items = batch.concat(items); // no spread: a batch can hold up to `cap` events, too many for call arguments + trim(); + retryAt = now() + retryDelayMs(failures++); + }, + }; +} const num = (value: unknown): number | undefined => typeof value === 'number' && Number.isFinite(value) ? value : undefined; @@ -216,11 +271,11 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe const flushSize = options.flushSize ?? DEFAULT_FLUSH_SIZE; const flushIntervalMs = options.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS; const heartbeatIntervalMs = options.heartbeatIntervalMs ?? DEFAULT_HEARTBEAT_MS; - const maxBuffered = options.maxBuffered ?? DEFAULT_MAX_BUFFERED; + // At least 1: with 0 every event, the deregister included, would be evicted on arrival. + const maxBuffered = Math.max(1, options.maxBuffered ?? DEFAULT_MAX_BUFFERED); const requestTimeoutMs = options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS; let promptUse = options.promptUse; - const pending: IngestEvent[] = []; let droppedCount = 0; let closed = false; let warned = false; @@ -232,17 +287,22 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe if (warned) return; warned = true; const reason = error instanceof Error ? error.message : String(error); - console.warn(`[lousho] observe: ${what} (${reason}); events are counted in 'dropped' and the run is unaffected.`); + console.warn( + `[lousho] observe: ${what} (${reason}); retrying with backoff, events past the buffer cap are counted in 'dropped' and the run is unaffected.` + ); }; + const outbox = createOutbox(maxBuffered, now, (count) => { + droppedCount += count; + warnOnce( + `buffer is full (${maxBuffered} events); dropping the oldest`, + new SDKError('observe buffer full', 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false }) + ); + }); + const push = (event: IngestEvent) => { - if (pending.length >= maxBuffered) { - droppedCount++; - warnOnce(`buffer is full (${maxBuffered} events); dropping`, new SDKError('observe buffer full', 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false })); - return; - } - pending.push(event); - if (pending.length >= flushSize) void flush(requestTimeoutMs); + outbox.push(event); + if (outbox.size >= flushSize && outbox.ready()) void flush(requestTimeoutMs); }; const enqueue = (event: IngestEvent) => { @@ -254,12 +314,16 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe push(event); }; - const post = async (events: IngestEvent[], timeoutMs: number): Promise => { + /** Aborted by close() so a regular flush in flight can't hold the exit flush past its ~2s cap. */ + const closing = new AbortController(); + + const post = async (events: IngestEvent[], timeoutMs: number, final: boolean): Promise => { + const timeout = AbortSignal.timeout(timeoutMs); const res = await fetchFn(`${url}/api/v1/ingest`, { method: 'POST', headers, body: JSON.stringify({ batch: events }), - signal: AbortSignal.timeout(timeoutMs), + signal: final ? timeout : AbortSignal.any([timeout, closing.signal]), }); if (!res.ok) { throw new SDKError(`ingest answered ${res.status}`, 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false }); @@ -268,20 +332,27 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe // Flushes are serialized on a chain: a fire-and-forget size flush and the // exit flush can overlap, and close() must not resolve while its own POST - // is still in flight. + // is still in flight. `final` is close()'s last attempt: it ignores the + // backoff and counts what it could not send as dropped. Each link swallows + // its own errors: a rejected link would skip every later flush and surface + // as an unhandled rejection at the `void flush()` call sites. let chain: Promise = Promise.resolve(); - const flush = (timeoutMs: number): Promise => { - const run = chain.then(async () => { - if (pending.length === 0) return; - const batch = pending.splice(0, pending.length); - try { - await post(batch, timeoutMs); - } catch (error) { - // Drop-and-count by contract: a missing daemon must never slow or fail a run. - droppedCount += batch.length; - warnOnce(`could not reach the observe daemon at ${url}`, error); - } - }); + const send = async (timeoutMs: number, final: boolean): Promise => { + const batch = outbox.take(final); + if (batch.length === 0) return; + try { + await post(batch, timeoutMs, final); + outbox.delivered(); + } catch (error) { + // Never slow or fail a run: requeue for a later retry, drop-and-count only as the last resort. + // close() aborting a regular flush is not an outage: its batch rides along in the final attempt. + if (final || !closing.signal.aborted) warnOnce(`could not reach the observe daemon at ${url}`, error); + if (final) droppedCount += batch.length; + else outbox.failed(batch); + } + }; + const flush = (timeoutMs: number, final = false): Promise => { + const run = chain.then(() => send(timeoutMs, final)).catch(() => undefined); chain = run; return run; }; @@ -330,11 +401,14 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe // no process.off in this runtime } } - // Graceful exit: say goodbye, then flush with the ~2s cap. push() directly so a - // full buffer can't eat the deregister - one pending event drops instead. - if (pending.length >= maxBuffered) pending.splice(0, 1); - push({ type: 'agent.deregister', data: { agent_id: identity.agentId, ts: now() } }); - await flush(EXIT_FLUSH_TIMEOUT_MS); + // Graceful exit: say goodbye, then one final flush with the ~2s cap. Straight + // into the outbox (enqueue drops after close, push could start a regular size + // flush); a full buffer evicts its oldest event rather than the deregister. + // Aborting `closing` fails a regular flush still in flight at once - its + // batch is requeued and rides along in the final attempt. + outbox.push({ type: 'agent.deregister', data: { agent_id: identity.agentId, ts: now() } }); + closing.abort(); + await flush(EXIT_FLUSH_TIMEOUT_MS, true); }; const exitHooks: Array<[string, () => void]> = [];