diff --git a/.fallowrc.json b/.fallowrc.json index 439f702d..eeb4c009 100644 --- a/.fallowrc.json +++ b/.fallowrc.json @@ -22,6 +22,7 @@ "src/deploy/shims/pi.worker.ts", "src/deploy/shims/isolatedVm.worker.ts", "src/agentDir/__fixtures__/**/*.{ts,js}", + "src/observe/__fixtures__/*.mts", "examples/agent-dir/tools/*.ts", "registry/**/*.ts" ], diff --git a/docs/observability.md b/docs/observability.md index 86e71cb7..c6efd471 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -413,10 +413,25 @@ The daemon also serves two stores agents can use at runtime: 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 +2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. A +backlog drains in batches one right after another, each at most 1000 events +and 4 MB of JSON (under the daemon's per-request caps of 1000 events and +5 MB). Only these are dropped and counted in the reporter's `dropped`: 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 — +first); events still unsent after the exit flush's final attempt; a single +event too large for any request (over 5 MB serialized, for example with +content capture on); and a batch the daemon refuses outright (`400`/`413`), +which is never retried. + +A process that dies of an uncaught exception or an unhandled rejection still +reports its last run: the reporter watches `uncaughtExceptionMonitor` and +sends what it holds, oldest first, holding up the crash for up to 1 s per +reporter. Node's error output and exit code stay as they were, and a script +that survives the throw (its own `uncaughtException` handler, or a capture +callback such as the REPL's) keeps reporting. An explicit `process.exit()` or +a SIGKILL still loses what is buffered. + +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 9123744c..1af0e749 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -14079,10 +14079,25 @@ The daemon also serves two stores agents can use at runtime: 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 +2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. A +backlog drains in batches one right after another, each at most 1000 events +and 4 MB of JSON (under the daemon's per-request caps of 1000 events and +5 MB). Only these are dropped and counted in the reporter's `dropped`: 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 — +first); events still unsent after the exit flush's final attempt; a single +event too large for any request (over 5 MB serialized, for example with +content capture on); and a batch the daemon refuses outright (`400`/`413`), +which is never retried. + +A process that dies of an uncaught exception or an unhandled rejection still +reports its last run: the reporter watches `uncaughtExceptionMonitor` and +sends what it holds, oldest first, holding up the crash for up to 1 s per +reporter. Node's error output and exit code stay as they were, and a script +that survives the throw (its own `uncaughtException` handler, or a capture +callback such as the REPL's) keeps reporting. An explicit `process.exit()` or +a SIGKILL still loses what is buffered. + +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/deploy/adapters/cloudflare.test.ts b/src/deploy/adapters/cloudflare.test.ts index 6af8e859..a74a781c 100644 --- a/src/deploy/adapters/cloudflare.test.ts +++ b/src/deploy/adapters/cloudflare.test.ts @@ -162,7 +162,7 @@ describe('CloudflareWorkerAdapter', () => { it('findNodeBuiltinReferences accepts only the allowlisted getBuiltinModule probes', () => { const probes = - 'loadBuiltinModule2("node:diagnostics_channel"); loadBuiltinModule4("node:dns"); process.getBuiltinModule?.("node:async_hooks"); loadBuiltinModule("node:module"); await loadNodeModule("node:dns"); getBuiltinModule("node:fs");'; + 'loadBuiltinModule2("node:diagnostics_channel"); loadBuiltinModule4("node:dns"); process.getBuiltinModule?.("node:async_hooks"); loadBuiltinModule("node:module"); await loadNodeModule("node:dns"); getBuiltinModule("node:fs"); getBuiltinModule(\'node:worker_threads\');'; expect(findNodeBuiltinReferences(probes)).toEqual([]); // Any other builtin, or an allowlisted id outside a probe call, is still a leak. expect(findNodeBuiltinReferences('loadBuiltinModule("node:net"); const id = "node:dns"; import("node:async_hooks");')).toEqual([ diff --git a/src/deploy/adapters/cloudflare.ts b/src/deploy/adapters/cloudflare.ts index 63134471..7d009e58 100644 --- a/src/deploy/adapters/cloudflare.ts +++ b/src/deploy/adapters/cloudflare.ts @@ -325,10 +325,12 @@ export default { fetch, scheduled }; * the worker graph via createAgent's LOUSHO_OBSERVE wiring; every access is * guarded by `typeof process === 'undefined'` and degrades to an * in-memory reporter identity on runtimes without builtins. + * - node:worker_threads: src/observe/blockingPost.ts (the reporter's crash + * flush), same guard; a runtime without it reports the send as failed. * Only these ids, and only as the argument of a `getBuiltinModule` / * `loadBuiltinModule` / `loadNodeModule` call, are exempt; any other `node:` string still fails. */ -const WORKER_SAFE_BUILTIN_PROBE = /\b(?:get|load)(?:Node|Builtin)Module\d*(?:\?\.)?\(\s*(["'`])node:(?:module|dns|diagnostics_channel|async_hooks|fs|crypto|os|path)\1\s*\)/g; +const WORKER_SAFE_BUILTIN_PROBE = /\b(?:get|load)(?:Node|Builtin)Module\d*(?:\?\.)?\(\s*(["'`])node:(?:module|dns|diagnostics_channel|async_hooks|fs|crypto|os|path|worker_threads)\1\s*\)/g; /** Returns every `node:`-prefixed module specifier referenced in `source`. */ export function findNodeBuiltinReferences(source: string): string[] { diff --git a/src/observe/__fixtures__/crashAfterRun.mts b/src/observe/__fixtures__/crashAfterRun.mts new file mode 100644 index 00000000..ae553b3a --- /dev/null +++ b/src/observe/__fixtures__/crashAfterRun.mts @@ -0,0 +1,60 @@ +/** + * Spawned by reporter.test.ts ('crash flush'): build a reporter with its + * default exit hooks, finish one span (buffered: the interval flush is far + * off and the batch is below the size threshold), then die without closing + * it - the way an agent script dies of a bug after a run. + * + * node --import tsx crashAfterRun.mts + * + * - throw: a top-level throw (Node reports it with origin 'unhandledRejection') + * - reject: an unhandled promise rejection + * - inflight: flushSize 1, so the span's batch is already in a POST the + * collector holds open when a timer callback throws + * - backlog: 1500 spans pending (past the daemon's 1000-event batch cap), then a top-level throw + * - bigbacklog: 30 spans of ~200KB input (~6MB, past the byte budget), then a top-level throw + * - handled: the script has its own 'uncaughtException' listener, so it + * survives the throw, finishes a second span and closes normally + * - capture: the same through process.setUncaughtExceptionCaptureCallback (REPL, domains) + * + * Env: LOUSHO_OBSERVE (collector URL), LOUSHO_FIXTURE_CWD (identity dir). + */ +import { observeExporter } from '../reporter'; + +const mode = process.argv[2] ?? 'throw'; +const reporter = observeExporter({ + url: process.env.LOUSHO_OBSERVE, + cwd: process.env.LOUSHO_FIXTURE_CWD, + project: 'crash-fixture', + agentName: 'crash-fixture', + heartbeatIntervalMs: 0, + ...(mode === 'inflight' ? { flushSize: 1 } : mode === 'backlog' || mode === 'bigbacklog' ? { flushSize: 10_000 } : {}), +}); +const finish = (id: string, attributes: Record = {}): void => { + const span = { id, name: 'invoke_agent crash-fixture', attributes, startTime: Date.now() }; + reporter.onSpanStart(span); + reporter.onSpanEnd({ ...span, endTime: Date.now() }); +}; +finish('crash-span'); +if (mode === 'bigbacklog') for (let i = 0; i < 30; i++) finish(`big-${i}`, { 'gen_ai.input.messages': 'x'.repeat(200 * 1024) }); +if (mode === 'backlog') for (let i = 1; i < 1500; i++) finish(`crash-span-${i}`); + +if (mode === 'reject') { + void Promise.reject(new Error('fixture: unhandled rejection after run')); +} else if (mode === 'inflight') { + setTimeout(() => { + throw new Error('fixture: uncaught exception after run'); + }, 300); +} else if (mode === 'handled' || mode === 'capture') { + const survive = (error: Error): void => { + console.error(`handled: ${error.message}`); + finish('after-span'); + void reporter.close(); + }; + if (mode === 'capture') process.setUncaughtExceptionCaptureCallback(survive); + else process.on('uncaughtException', survive); + setTimeout(() => { + throw new Error('fixture: uncaught exception after run'); + }, 10); +} else { + throw new Error('fixture: uncaught exception after run'); +} diff --git a/src/observe/blockingPost.ts b/src/observe/blockingPost.ts new file mode 100644 index 00000000..f988c60b --- /dev/null +++ b/src/observe/blockingPost.ts @@ -0,0 +1,67 @@ +/** + * POSTs that block the calling thread until they settle - the reporter's + * crash flush. `uncaughtExceptionMonitor` listeners run synchronously and + * Node exits the moment they return, so an ordinary `fetch` started there + * never even opens its socket (measured: 0 connections). Here the requests + * run in a throwaway worker thread with its own event loop while this + * thread sleeps on `Atomics.wait`; the worker marks each delivered body + * in a shared array and wakes this thread when it is done. + * + * The worker is given no `execArgv` (no loaders such as tsx: it only needs + * the global `fetch`) and is left running when this returns: the process is + * about to exit, which terminates it. + * + * Builtins load through `getBuiltinModule` (builtins.ts): this module ships + * in the Cloudflare Worker bundle, where it reports nothing delivered + * instead of breaking the bundle. + */ +import { getBuiltinModule } from './builtins'; + +interface WorkerThreadsLike { + Worker: new (code: string, options: { eval: true; execArgv: string[]; workerData: unknown }) => { unref(): void }; +} + +/** Shared slots: [0] 0 while the worker runs, 1 once it stopped; [1 + i] 1 once body i got a 2xx. */ +const DONE = 0; +const FIRST_BODY = 1; + +/** + * The worker's whole program (CommonJS, `eval: true`): the bodies in order, + * one at a time, all inside the one deadline. A body the server answers with + * an error is skipped; no answer at all (down, deadline) stops it. + * The id literal is double-quoted so the bundle check's probe matcher sees it. + */ +const WORKER_SOURCE = ` +const { workerData: d } = process.getBuiltinModule("node:worker_threads"); +const flag = new Int32Array(d.flag); +(async () => { + for (let i = 0; i < d.bodies.length; i++) { + const left = d.deadline - Date.now(); + if (left <= 0) break; + const res = await fetch(d.url, { method: 'POST', headers: d.headers, body: d.bodies[i], signal: AbortSignal.timeout(left) }); + if (res.ok) Atomics.store(flag, ${FIRST_BODY} + i, 1); + } +})().catch(() => undefined).finally(() => { Atomics.store(flag, ${DONE}, 1); Atomics.notify(flag, ${DONE}); }); +`; + +/** + * POST each of `bodies` to `url` in order, blocking at most `timeoutMs` in + * all (worker start-up included). Returns, per body, whether the server + * answered it 2xx in time; all false on a runtime without worker threads. + */ +export function blockingPost(url: string, headers: Record, bodies: readonly string[], timeoutMs: number): boolean[] { + const none = bodies.map(() => false); + if (bodies.length === 0) return none; + try { + const threads = getBuiltinModule('node:worker_threads'); + if (!threads) return none; + const flag = new Int32Array(new SharedArrayBuffer(4 * (FIRST_BODY + bodies.length))); + const deadline = Date.now() + timeoutMs; + const worker = new threads.Worker(WORKER_SOURCE, { eval: true, execArgv: [], workerData: { flag: flag.buffer, url, headers, bodies, deadline } }); + worker.unref(); // never keeps a process alive past a timed-out wait + Atomics.wait(flag, DONE, 0, timeoutMs); + return bodies.map((_, i) => Atomics.load(flag, FIRST_BODY + i) === 1); + } catch { + return none; + } +} diff --git a/src/observe/builtins.ts b/src/observe/builtins.ts new file mode 100644 index 00000000..3e135c0f --- /dev/null +++ b/src/observe/builtins.ts @@ -0,0 +1,14 @@ +/** + * Lazy Node builtins for the observe modules that also ship in the Worker + * bundle (identity.ts, blockingPost.ts): `undefined` on runtimes without + * them, never an import for the bundler to resolve. + * + * Named to match the worker leak check's exempted call pattern (see + * WORKER_SAFE_BUILTIN_PROBE in src/deploy/adapters/cloudflare.ts): a builtin + * id literal stays out of the flagged set only while it appears as this + * function's argument at the call site. Do not rename. + */ +export function getBuiltinModule(name: string): T | undefined { + if (typeof process === 'undefined') return undefined; + return (process.getBuiltinModule?.(name) as T | undefined) ?? undefined; +} diff --git a/src/observe/identity.ts b/src/observe/identity.ts index 9f6c5e33..2f041246 100644 --- a/src/observe/identity.ts +++ b/src/observe/identity.ts @@ -18,6 +18,8 @@ * breaking non-Node runtimes. */ +import { getBuiltinModule } from './builtins'; + /** The slices of the builtins used here, typed so the lazy loads stay honest. */ interface FsLike { existsSync(path: string): boolean; @@ -36,15 +38,8 @@ interface PathLike { resolve(...parts: string[]): string; } -// Named to match the worker leak check's exempted call pattern (see -// WORKER_SAFE_BUILTIN_PROBE in src/deploy/adapters/cloudflare.ts): the -// the builtin id literals below stay out of the flagged set only while they -// appear as this function's argument. Do not rename. -function getBuiltinModule(name: string): T | undefined { - // Reporters bundled for non-Node runtimes have no fs; identity falls back to memory. - if (typeof process === 'undefined') return undefined; - return (process.getBuiltinModule?.(name) as T | undefined) ?? undefined; -} +// Reporters bundled for non-Node runtimes have no fs; identity falls back to memory. +// Keep each id literal as getBuiltinModule's direct argument (see builtins.ts). const lazyFs = () => getBuiltinModule('node:fs') as FsLike | undefined; const lazyCrypto = () => getBuiltinModule('node:crypto') as CryptoLike | undefined; diff --git a/src/observe/reporter.test.ts b/src/observe/reporter.test.ts index 281633d6..28c5fce6 100644 --- a/src/observe/reporter.test.ts +++ b/src/observe/reporter.test.ts @@ -5,12 +5,16 @@ * drops-and-counts without ever throwing into the run, and the env * (`LOUSHO_OBSERVE`) wiring composes with an explicit exporter. */ +import { spawn } from 'node:child_process'; import * as fs from 'node:fs'; +import * as http from 'node:http'; +import type { AddressInfo } from 'node:net'; import * as os from 'node:os'; import * as path from 'node:path'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { GenAiAttr, GenAiOperation, LegacyAttr, SdkAttr } from '../execution/semconv'; import type { Span, TraceExporter } from '../execution/tracing'; +import { blockingPost } from './blockingPost'; import type { IngestEvent } from './envelope'; import { fanOutExporters, observeExporter, observeExporterFromEnv, withObserveExporter, type ObserveReporter } from './reporter'; @@ -326,33 +330,133 @@ describe('observeExporter', () => { expect(r.dropped).toBe(3); }); - it('requeues a batch too large to spread into call arguments, and the flush chain survives', async () => { + it('drains a backlog past the daemon batch cap in successive capped batches, 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 ids: string[] = []; 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'); + const batch = (JSON.parse(String(init?.body)) as { batch: IngestEvent[] }).batch; + sizes.push(batch.length); + if (sizes.length === 1) throw new TypeError('fetch failed'); // the daemon is down + for (const e of batch) if (e.type === 'observation') ids.push(e.data.obs_id); 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 }); + const total = 2500; + const { r } = reporter({ fetchFn, 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]); + await vi.advanceTimersByTimeAsync(1000); // rejected: the oldest 1000 are requeued + await vi.advanceTimersByTimeAsync(1000); // the daemon is back: one flush drains it all, batch after batch + expect(sizes).toEqual([1000, 1000, 1000, 500]); + expect(ids).toEqual(Array.from({ length: total }, (_, i) => `s${i}`)); // every event, oldest first expect(r.dropped).toBe(0); await r.close(); - expect(sizes).toHaveLength(3); // the chain still runs close()'s flush (the deregister) + expect(sizes).toHaveLength(5); // 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("close()'s final attempt sends a backlog in capped batches and drops the rest when one fails", async () => { + const sizes: number[] = []; + const fetchFn = vi.fn(async (_input: string | URL | Request, init?: RequestInit): Promise => { + sizes.push((JSON.parse(String(init?.body)) as { batch: IngestEvent[] }).batch.length); + return new Response('{}', { status: sizes.length === 2 ? 500 : 202 }); + }) as unknown as typeof fetch; + vi.spyOn(console, 'warn').mockImplementation(() => undefined); + const { r } = reporter({ fetchFn, flushSize: 10_000 }); + for (let i = 0; i < 2500; i++) r.onSpanEnd({ id: `s${i}`, name: 'n', attributes: {}, startTime: 1, endTime: 2 }); + await r.close(); + expect(sizes).toEqual([1000, 1000]); // the second failed: it and the remaining 501 (500 + the deregister) are dropped + expect(r.dropped).toBe(1501); + }); + }); + + describe('byte cap', () => { + const BUDGET = 4 * 1024 * 1024; // BATCH_BUDGET_BYTES + /** A finished span whose input is `bytes` of ASCII. */ + const bigSpan = (id: string, bytes: number): Span => + makeSpan({ id, attributes: { [GenAiAttr.OPERATION_NAME]: GenAiOperation.CHAT, [GenAiAttr.INPUT_MESSAGES]: 'x'.repeat(bytes) } }); + /** A fetch that records each body's size and ids, answering `status(n)` to the n-th call (1-based). */ + function sizedFetch(status: (n: number) => number = () => 202) { + const calls: { bytes: number; ids: string[] }[] = []; + const fetchFn = vi.fn(async (_input: string | URL | Request, init?: RequestInit): Promise => { + const body = String(init?.body); + const batch = (JSON.parse(body) as { batch: IngestEvent[] }).batch; + calls.push({ bytes: Buffer.byteLength(body), ids: batch.flatMap((e) => (e.type === 'observation' ? [e.data.obs_id] : [])) }); + return new Response('{}', { status: status(calls.length) }); + }) as unknown as typeof fetch; + return { fetchFn, calls }; + } + + it('cuts batches of large events at the byte budget, all delivered in order', async () => { + const { fetchFn, calls } = sizedFetch(); + const { r } = reporter({ fetchFn, flushSize: 1000 }); + for (let i = 0; i < 50; i++) { + const span = bigSpan(`big${i}`, 200 * 1024); + r.onSpanStart(span); + r.onSpanEnd(span); + } + await r.close(); + expect(calls.length).toBeGreaterThanOrEqual(3); // ~10MB of events + for (const call of calls) expect(call.bytes).toBeLessThanOrEqual(BUDGET); + expect(calls.flatMap((c) => c.ids)).toEqual(Array.from({ length: 50 }, (_, i) => `big${i}`)); + expect(r.dropped).toBe(0); + }); + + it('drops and counts one event too large for any request, and keeps delivering', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); + const { fetchFn, calls } = sizedFetch(); + const { r } = reporter({ fetchFn, flushSize: 1000 }); + for (const span of [bigSpan('huge', 6 * 1024 * 1024), makeSpan({ id: 'after' })]) { + r.onSpanStart(span); + r.onSpanEnd(span); + } + expect(r.dropped).toBe(1); + expect(warn).toHaveBeenCalledOnce(); + expect(String(warn.mock.calls[0][0])).toMatch(/larger than the daemon's 5242880-byte request cap/); + await r.close(); + expect(calls.flatMap((c) => c.ids)).toEqual(['after']); + expect(r.dropped).toBe(1); + }); + + it('sends one event over the budget but under the hard cap on its own', async () => { + // The input is carried twice (input + attrs): ~4.4MB of event, over the 4MB budget, under the 5MB cap. + const { fetchFn, calls } = sizedFetch(); + const { r } = reporter({ fetchFn, flushSize: 1000 }); + for (const span of [makeSpan({ id: 'before' }), bigSpan('large', 2.2 * 1024 * 1024), makeSpan({ id: 'after' })]) { + r.onSpanStart(span); + r.onSpanEnd(span); + } + await r.close(); + expect(calls.map((c) => c.ids)).toEqual([['before'], ['large'], ['after']]); + expect(r.dropped).toBe(0); + }); + + it('drops and counts a batch the daemon answers 413, and keeps delivering', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); + const { fetchFn, calls } = sizedFetch((n) => (n === 1 ? 413 : 202)); + const { r } = reporter({ fetchFn, flushSize: 2 }); + const finish = (id: string) => { + const span = makeSpan({ id }); + r.onSpanStart(span); + r.onSpanEnd(span); + }; + finish('a'); + finish('b'); // a size flush: answered 413 + await vi.waitFor(() => expect(r.dropped).toBe(2)); + finish('c'); + finish('d'); + await r.close(); + expect(calls[0].ids).toEqual(['a', 'b']); + expect(calls.flatMap((c) => c.ids)).toEqual(['a', 'b', 'c', 'd']); // the refused batch is not resent + expect(r.dropped).toBe(2); // the refused batch, not requeued + expect(String(warn.mock.calls[0][0])).toContain('refused a batch of 2 events (413)'); + }); }); it('clamps maxBuffered to 1 so the deregister survives', async () => { @@ -445,6 +549,137 @@ describe('observeExporter', () => { }); }); +describe('crash flush', () => { + const fixture = path.join(__dirname, '__fixtures__', 'crashAfterRun.mts'); + const repoRoot = path.resolve(__dirname, '..', '..'); + type Mode = 'throw' | 'reject' | 'inflight' | 'handled' | 'capture' | 'backlog' | 'bigbacklog'; + + /** + * Run the fixture against an in-test collector; resolves with its exit and + * the batch of every ingest request received. `holdFirst` never answers the + * first request (a POST still in flight when the process dies). + */ + async function crash(mode: Mode, holdFirst = false): Promise<{ code: number | null; stderr: string; batches: IngestEvent[][] }> { + const batches: IngestEvent[][] = []; + const server = http.createServer((req, res) => { + let body = ''; + req.on('data', (chunk: Buffer) => (body += chunk.toString())); + req.on('end', () => { + if (req.url === '/api/v1/ingest') batches.push((JSON.parse(body) as { batch: IngestEvent[] }).batch); + if (holdFirst && batches.length === 1) return; + res.writeHead(202, { 'content-type': 'application/json' }).end('{}'); + }); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const { port } = server.address() as AddressInfo; + try { + const child = spawn(process.execPath, ['--import', 'tsx', fixture, mode], { + cwd: repoRoot, + env: { ...process.env, LOUSHO_OBSERVE: `http://127.0.0.1:${port}`, LOUSHO_FIXTURE_CWD: cwd }, + stdio: ['ignore', 'ignore', 'pipe'], + }); + let stderr = ''; + child.stderr.on('data', (chunk: Buffer) => (stderr += chunk.toString())); + const code = await new Promise((resolve) => child.on('close', resolve)); + return { code, stderr, batches }; + } finally { + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + } + } + const obsIds = (batch: IngestEvent[]): string[] => batch.flatMap((e) => (e.type === 'observation' ? [e.data.obs_id] : [])); + + it('flushes pending events when the process dies of an uncaught exception', async () => { + const { code, stderr, batches } = await crash('throw'); + expect(code).toBe(1); // Node's own crash exit code, unchanged + expect(stderr).toContain('fixture: uncaught exception after run'); + expect(batches.flatMap(obsIds)).toEqual(['crash-span']); + expect(batches.flat().map((e) => e.type)).toContain('agent.deregister'); + }, 30_000); + + it('flushes pending events when the process dies of an unhandled rejection', async () => { + const { code, stderr, batches } = await crash('reject'); + expect(code).toBe(1); + expect(stderr).toContain('fixture: unhandled rejection after run'); + expect(batches.flatMap(obsIds)).toEqual(['crash-span']); + }, 30_000); + + it('sends a crash backlog past the daemon batch cap as successive capped POSTs', async () => { + const { code, batches } = await crash('backlog'); + expect(code).toBe(1); + expect(batches.map((b) => b.length)).toEqual([1000, 501]); // 1500 observations + the deregister, oldest first + expect(batches.flatMap(obsIds)).toEqual(['crash-span', ...Array.from({ length: 1499 }, (_, i) => `crash-span-${i + 1}`)]); + }, 30_000); + + it('cuts a crash backlog of large events at the byte budget', async () => { + const { code, batches } = await crash('bigbacklog'); + expect(code).toBe(1); + expect(batches.length).toBeGreaterThanOrEqual(2); // 30 x 200KB = ~6MB + for (const batch of batches) expect(Buffer.byteLength(JSON.stringify({ batch }))).toBeLessThanOrEqual(4 * 1024 * 1024); + expect(batches.flatMap(obsIds)).toEqual(['crash-span', ...Array.from({ length: 30 }, (_, i) => `big-${i}`)]); + }, 30_000); + + it('resends the batch a POST still in flight was carrying when the process died', async () => { + const { code, batches } = await crash('inflight', true); + expect(code).toBe(1); + expect(batches.map(obsIds)).toEqual([['crash-span'], ['crash-span']]); + }, 30_000); + + it.each([['handled'], ['capture']] as const)('keeps reporting when the script survives the throw itself (%s)', async (mode) => { + const { code, stderr, batches } = await crash(mode); + expect(code).toBe(0); // the user's listener / capture callback kept the process alive; the reporter did not exit it + expect(stderr).toContain('handled: fixture: uncaught exception after run'); + expect(batches.flatMap(obsIds)).toEqual(['crash-span', 'after-span']); + }, 30_000); + + it('only flushes, and stays hooked, when an uncaughtException listener keeps the process alive', async () => { + // vitest itself listens for 'uncaughtException', so this is the survivable-throw path. + const before = process.listenerCount('uncaughtExceptionMonitor'); + const { r, posted } = reporter({ registerExitHooks: true }); + for (const id of ['one', 'two']) { + const span = makeSpan({ id }); + r.onSpanStart(span); + r.onSpanEnd(span); + // Call the listeners directly: emitting the event for real would also reach vitest's own monitors. + const monitor = process.listeners('uncaughtExceptionMonitor').at(-1); + monitor?.(new Error('survived'), 'uncaughtException'); + await vi.waitFor(() => expect(observations(posted).map((e) => e.data.obs_id)).toContain(id)); + } + expect(process.listenerCount('uncaughtExceptionMonitor')).toBe(before + 1); + expect(allEvents(posted).map((e) => e.type)).not.toContain('agent.deregister'); // not closed + }); + + it('removes the crash hook on close()', async () => { + const before = process.listenerCount('uncaughtExceptionMonitor'); + const { r } = reporter({ registerExitHooks: true }); + expect(process.listenerCount('uncaughtExceptionMonitor')).toBe(before + 1); + await r.close(); + expect(process.listenerCount('uncaughtExceptionMonitor')).toBe(before); + }); +}); + +describe('blockingPost', () => { + it('delivers nothing to an unreachable collector', () => { + expect(blockingPost('http://127.0.0.1:1/api/v1/ingest', {}, ['{}', '{}'], 1000)).toEqual([false, false]); + expect(blockingPost('http://127.0.0.1:1/api/v1/ingest', {}, [], 1000)).toEqual([]); + }); + + it('gives up after its timeout', async () => { + // Blocking this thread means an in-process server can never answer: the wait must end on its own. + const server = http.createServer(() => undefined); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const { port } = server.address() as AddressInfo; + try { + const t0 = Date.now(); + expect(blockingPost(`http://127.0.0.1:${port}/`, {}, ['{}'], 200)).toEqual([false]); + expect(Date.now() - t0).toBeLessThan(1000); + } finally { + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + } + }); +}); + describe('env wiring', () => { it('returns undefined when LOUSHO_OBSERVE is unset or off', () => { expect(observeExporterFromEnv({ env: {} })).toBeUndefined(); diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts index eca5d4b2..e607ef79 100644 --- a/src/observe/reporter.ts +++ b/src/observe/reporter.ts @@ -14,8 +14,10 @@ * id (the `TraceLine.id`), so a resent batch upserts by key. * - 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. + * attempt with the 2s timeout), and on a crash (an uncaught exception or + * unhandled rejection: one POST that blocks the dying process for at most + * 1s, see blockingPost.ts; Node's crash output and exit code are untouched). + * SIGKILL loss is accepted and bounded: nothing persists mid 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 @@ -33,7 +35,8 @@ import { GenAiAttr, GenAiOperation, LegacyAttr, SdkAttr } from '../execution/sem import { SDKError } from '../execution/errors'; import type { Span, TraceExporter } from '../execution/tracing'; import { VERSION } from '../version'; -import { OBSERVE_DEFAULT_PORT, type IngestEvent, type ObservationData, type ObservationKind, type PromptUseData } from './envelope'; +import { blockingPost } from './blockingPost'; +import { OBSERVE_DEFAULT_PORT, OBSERVE_MAX_BATCH_EVENTS, OBSERVE_MAX_BODY_BYTES, type IngestEvent, type ObservationData, type ObservationKind, type PromptUseData } from './envelope'; import { observeTargetFromEnv } from './emit'; import { hostName, resolveIdentity, resolveProject } from './identity'; @@ -108,21 +111,81 @@ const DEFAULT_HEARTBEAT_MS = 60_000; const DEFAULT_MAX_BUFFERED = 5000; const DEFAULT_REQUEST_TIMEOUT_MS = 5000; const EXIT_FLUSH_TIMEOUT_MS = 2000; +/** How long a crash may block the dying process to deliver what is pending. */ +const CRASH_FLUSH_TIMEOUT_MS = 1000; 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 JSON a POST carries for `entries`: byte for byte what + * `JSON.stringify({ batch: events })` would produce, built from each event's + * JSON serialized once on arrival. + */ +const bodyOf = (entries: readonly OutboxEntry[]): string => `{"batch":[${entries.map((e) => e.json).join(',')}]}`; + +/** `{"batch":[]}`: the envelope's bytes around the events. */ +const ENVELOPE_BYTES = 12; +/** + * A batch stops growing before its body would pass this - kept under the + * daemon's OBSERVE_MAX_BODY_BYTES (5MB, a `413` past it) with room to spare. + */ +const BATCH_BUDGET_BYTES = 4 * 1024 * 1024; + +/** One pending event, serialized once when queued so batching can measure it. */ +interface OutboxEntry { + json: string; + bytes: number; +} + +const utf8 = new TextEncoder(); + +/** An event as an outbox entry; `undefined` when its body alone would pass the daemon's hard cap. */ +function entryOf(event: IngestEvent): OutboxEntry | undefined { + const json = JSON.stringify(event); + const bytes = utf8.encode(json).length; + return ENVELOPE_BYTES + bytes > OBSERVE_MAX_BODY_BYTES ? undefined : { json, bytes }; +} + +/** + * Where the batch starting at `start` ends: at most OBSERVE_MAX_BATCH_EVENTS + * events (the daemon rejects more) and BATCH_BUDGET_BYTES of body, measured + * incrementally. The first event is always taken, so one event over the + * budget (but under the hard cap) goes out alone instead of wedging the queue. + */ +function batchEnd(entries: readonly OutboxEntry[], start: number): number { + let end = start; + let bytes = ENVELOPE_BYTES; + while (end < entries.length && end - start < OBSERVE_MAX_BATCH_EVENTS) { + bytes += entries[end].bytes + (end > start ? 1 : 0); // the comma + if (end > start && bytes > BATCH_BUDGET_BYTES) break; + end++; + } + return end; +} + +/** `entries` cut into consecutive batches, oldest first. */ +function batchesOf(entries: readonly OutboxEntry[]): OutboxEntry[][] { + const batches: OutboxEntry[][] = []; + for (let start = 0; start < entries.length; ) { + const end = batchEnd(entries, start); + batches.push(entries.slice(start, end)); + start = end; + } + return batches; +} + /** * 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. + * there is room to keep it. A batch is cut by {@link batchEnd}, oldest first. */ function createOutbox(cap: number, now: () => number, onEvict: (count: number) => void) { - let items: IngestEvent[] = []; + let items: OutboxEntry[] = []; let failures = 0; let retryAt = 0; const trim = (): void => { @@ -139,21 +202,29 @@ function createOutbox(cap: number, now: () => number, onEvict: (count: number) = ready(): boolean { return now() >= retryAt; }, - push(event: IngestEvent): void { - items.push(event); + /** Queue `event`; false (nothing queued) when it is too large for any request. */ + push(event: IngestEvent): boolean { + const entry = entryOf(event); + if (!entry) return false; + items.push(entry); trim(); + return true; }, - /** 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) : []; + /** The oldest batch, for one send - or nothing while backing off, unless `force` (the final attempt). */ + take(force = false): OutboxEntry[] { + return force || now() >= retryAt ? items.splice(0, batchEnd(items, 0)) : []; + }, + /** Everything pending, uncut (the crash flush; counting what a failed final attempt leaves). */ + drain(): OutboxEntry[] { + return 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 + failed(batch: OutboxEntry[]): void { + items = batch.concat(items); // no spread: keeps call arguments bounded whatever the batch size trim(); retryAt = now() + retryDelayMs(failures++); }, @@ -300,8 +371,14 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe ); }); + /** Count events the daemon will never take - too large, or a batch it refused - and warn once. */ + const discard = (count: number, why: string): void => { + droppedCount += count; + warnOnce(why, new SDKError('observe event rejected', 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false })); + }; + const push = (event: IngestEvent) => { - outbox.push(event); + if (!outbox.push(event)) discard(1, `an event is larger than the daemon's ${OBSERVE_MAX_BODY_BYTES}-byte request cap; dropping it`); if (outbox.size >= flushSize && outbox.ready()) void flush(requestTimeoutMs); }; @@ -317,17 +394,16 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe /** 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 => { + /** One ingest POST; resolves with the status, rejects only when no answer arrived. */ + const post = async (entries: OutboxEntry[], 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 }), + body: bodyOf(entries), signal: final ? timeout : AbortSignal.any([timeout, closing.signal]), }); - if (!res.ok) { - throw new SDKError(`ingest answered ${res.status}`, 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false }); - } + return res.status; }; // Flushes are serialized on a chain: a fire-and-forget size flush and the @@ -337,18 +413,45 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe // 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 send = async (timeoutMs: number, final: boolean): Promise => { - const batch = outbox.take(final); - if (batch.length === 0) return; + /** The batch a POST is carrying right now: a crash resends it, since that POST can never finish. */ + let inFlight: OutboxEntry[] = []; + /** One POST; false when it failed (the batch is then requeued, or - final - dropped with everything left). */ + const sendBatch = async (batch: OutboxEntry[], timeoutMs: number, final: boolean): Promise => { + inFlight = batch; try { - await post(batch, timeoutMs, final); + const status = await post(batch, timeoutMs, final); + // A batch the daemon refuses as such (too large, malformed) would be refused again on every retry. + if (status === 400 || status === 413) discard(batch.length, `the observe daemon refused a batch of ${batch.length} events (${status}); dropping it`); + else if (status < 200 || status >= 300) { + throw new SDKError(`ingest answered ${status}`, 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false }); + } outbox.delivered(); + return true; } 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; + if (final) droppedCount += batch.length + outbox.drain().length; else outbox.failed(batch); + return false; + } finally { + inFlight = []; + } + }; + /** + * Send batch after batch while a cap cut the last one short: a backlog + * (after an outage) drains at once instead of one batch per interval. + * Regular flushes give each POST its own timeout; the final attempt shares + * one deadline. + */ + const send = async (timeoutMs: number, final: boolean): Promise => { + const deadline = Date.now() + timeoutMs; + for (;;) { + const batch = outbox.take(final); + const cut = outbox.size > 0; // read before the await: events queued meanwhile wait for their own trigger + if (batch.length === 0) return; + if (!(await sendBatch(batch, final ? Math.max(1, deadline - Date.now()) : timeoutMs, final))) return; + if (!cut) return; } }; const flush = (timeoutMs: number, final = false): Promise => { @@ -390,8 +493,9 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe timers.push(t); } - const close = async (): Promise => { - if (closed) return; + /** Stop timers and hooks and queue the goodbye; false when already closed. Shared by close() and the crash flush. */ + const shutdown = (): boolean => { + if (closed) return false; closed = true; for (const t of timers) clearInterval(t); for (const [signal, handler] of exitHooks) { @@ -401,16 +505,37 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe // no process.off in this runtime } } - // 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. + // 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. outbox.push({ type: 'agent.deregister', data: { agent_id: identity.agentId, ts: now() } }); + return true; + }; + + const close = async (): Promise => { + if (!shutdown()) return; + // Graceful exit: one final flush with the ~2s cap. Aborting `closing` fails + // a regular flush still in flight at once - its batch is requeued and + // rides along in the final attempt. closing.abort(); await flush(EXIT_FLUSH_TIMEOUT_MS, true); }; + /** + * The process is dying of an uncaught exception or unhandled rejection: + * Node exits as soon as the monitor returns, so no async send can finish. + * Send everything pending, oldest first - the batch a regular POST was + * carrying included (re-sent events upsert by key) - as successive POSTs, + * cut like any other batch, that together block for at most 1s. + */ + const crashFlush = (): void => { + if (!shutdown()) return; + const batches = batchesOf(inFlight.concat(outbox.drain())); + const delivered = blockingPost(`${url}/api/v1/ingest`, headers, batches.map(bodyOf), CRASH_FLUSH_TIMEOUT_MS); + batches.forEach((batch, i) => { + if (!delivered[i]) droppedCount += batch.length; + }); + }; + const exitHooks: Array<[string, () => void]> = []; if (options.registerExitHooks !== false && typeof process !== 'undefined' && typeof process.once === 'function') { // 'beforeExit' fires while async work still runs; the in-flight POST keeps the loop @@ -420,13 +545,28 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe const onSignal = (code: number) => () => { void close().finally(() => process.exit(code)); }; - const hooks: Array<[NodeJS.Signals | 'beforeExit', () => void]> = [ + // 'uncaughtExceptionMonitor' observes a crash without handling it: Node still + // prints the error and exits with its own code. It also sees an unhandled + // rejection that crashes the process (origin 'unhandledRejection'), and stays + // silent when one does not (--unhandled-rejections=warn, or a user listener), + // so no 'unhandledRejection' listener is added - adding one would itself stop + // the crash. With an 'uncaughtException' listener or a capture callback + // (REPL, domains) the process survives: keep reporting, just flush now. + const survives = () => process.listenerCount('uncaughtException') > 0 || process.hasUncaughtExceptionCaptureCallback?.() === true; + const onCrash = () => { + if (survives()) void flush(requestTimeoutMs); + else crashFlush(); + }; + const hooks: Array<[string, () => void]> = [ ['beforeExit', onBeforeExit], ['SIGINT', onSignal(130)], ['SIGTERM', onSignal(143)], + ['uncaughtExceptionMonitor', onCrash], ]; for (const [signal, handler] of hooks) { - process.once(signal as NodeJS.Signals, handler); + // The monitor stays on: a process that survived one throw can still die of the next. + if (signal === 'uncaughtExceptionMonitor') process.on(signal, handler); + else process.once(signal as NodeJS.Signals, handler); exitHooks.push([signal, handler]); } }