From 054f1974090ac2b9988efa97117b3c4fb161895e Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 16:57:00 +0300 Subject: [PATCH 1/2] fix(observe): requeue failed ingest batches with backoff instead of dropping A failed POST now puts its batch back at the front of the outbox, trims to maxBuffered from the oldest end (evictions count in `dropped`), and backs off min(2^n * 1000, 30000) ms before the next size or interval flush. close() makes one forced final attempt and counts what it could not send as dropped. A full buffer now evicts the oldest event instead of rejecting the newest, so the deregister needs no special case. Co-Authored-By: Claude Opus 5.5 --- docs/observability.md | 8 ++- llms-full.txt | 8 ++- src/observe/reporter.test.ts | 112 +++++++++++++++++++++++++++++++++++ src/observe/reporter.ts | 106 ++++++++++++++++++++++++++------- 4 files changed, 207 insertions(+), 27 deletions(-) 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..b66f0cb5 100644 --- a/src/observe/reporter.test.ts +++ b/src/observe/reporter.test.ts @@ -215,6 +215,118 @@ 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('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..1cd19f44 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). * 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,7 +59,7 @@ 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). */ maxBuffered?: number; /** Per-request timeout (default 5000ms); the exit flush always caps at ~2000ms. */ requestTimeoutMs?: number; @@ -81,7 +83,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 +106,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) { + const 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: readonly IngestEvent[]): void { + items.unshift(...batch); + trim(); + retryAt = now() + retryDelayMs(failures++); + }, + }; +} const num = (value: unknown): number | undefined => typeof value === 'number' && Number.isFinite(value) ? value : undefined; @@ -220,7 +273,6 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe 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 +284,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) => { @@ -268,18 +325,21 @@ 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. let chain: Promise = Promise.resolve(); - const flush = (timeoutMs: number): Promise => { + const flush = (timeoutMs: number, final = false): Promise => { const run = chain.then(async () => { - if (pending.length === 0) return; - const batch = pending.splice(0, pending.length); + const batch = outbox.take(final); + if (batch.length === 0) return; try { await post(batch, timeoutMs); + outbox.delivered(); } catch (error) { - // Drop-and-count by contract: a missing daemon must never slow or fail a run. - droppedCount += batch.length; + // Never slow or fail a run: requeue for a later retry, drop-and-count only as the last resort. warnOnce(`could not reach the observe daemon at ${url}`, error); + if (final) droppedCount += batch.length; + else outbox.failed(batch); } }); chain = run; @@ -330,11 +390,11 @@ 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); + // Graceful exit: say goodbye, then one final flush with the ~2s cap. push() + // directly (enqueue drops after close); a full buffer evicts its oldest event + // rather than the deregister. push({ type: 'agent.deregister', data: { agent_id: identity.agentId, ts: now() } }); - await flush(EXIT_FLUSH_TIMEOUT_MS); + await flush(EXIT_FLUSH_TIMEOUT_MS, true); }; const exitHooks: Array<[string, () => void]> = []; From 0400f1604745111cdb4384fcbab06adb4e47bf7d Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 17:15:38 +0300 Subject: [PATCH 2/2] fix(observe): keep the flush chain alive on large requeues and cap close() at ~2s - failed(): rebuild with concat instead of unshift(...batch); a batch as large as maxBuffered overflowed the call stack, rejected the chain link, and stalled every later flush with unhandled rejections. - Each chain link now swallows its own errors, so no future throw can break the chain. - maxBuffered is clamped to at least 1 so the deregister survives. - close() aborts a regular flush still in flight (its batch is requeued and rides along in the final attempt) and no longer starts a size flush ahead of it, so the exit flush stays near its documented ~2s bound. Co-Authored-By: Claude Opus 5.5 --- src/observe/reporter.test.ts | 70 ++++++++++++++++++++++++++++++++++++ src/observe/reporter.ts | 68 +++++++++++++++++++++-------------- 2 files changed, 111 insertions(+), 27 deletions(-) diff --git a/src/observe/reporter.test.ts b/src/observe/reporter.test.ts index b66f0cb5..281633d6 100644 --- a/src/observe/reporter.test.ts +++ b/src/observe/reporter.test.ts @@ -325,6 +325,76 @@ describe('observeExporter', () => { 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 () => { diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts index 1cd19f44..eca5d4b2 100644 --- a/src/observe/reporter.ts +++ b/src/observe/reporter.ts @@ -12,7 +12,9 @@ * 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 requeues the failed batch and retries with * exponential backoff (1s base, 30s cap); only events evicted past the @@ -59,9 +61,9 @@ export interface ObserveExporterOptions { flushIntervalMs?: number; /** Heartbeat cadence (default 60000ms); `0` disables heartbeats. */ heartbeatIntervalMs?: number; - /** Buffer cap; past it the oldest events are evicted 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; @@ -120,7 +122,7 @@ const retryDelayMs = (failures: number): number => Math.min(RETRY_BASE_MS * 2 ** * there is room to keep it. */ function createOutbox(cap: number, now: () => number, onEvict: (count: number) => void) { - const items: IngestEvent[] = []; + let items: IngestEvent[] = []; let failures = 0; let retryAt = 0; const trim = (): void => { @@ -150,8 +152,8 @@ function createOutbox(cap: number, now: () => number, onEvict: (count: number) = retryAt = 0; }, /** Put a failed batch back in front of anything queued since, trim to the cap, and back off. */ - failed(batch: readonly IngestEvent[]): void { - items.unshift(...batch); + 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++); }, @@ -269,7 +271,8 @@ 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; @@ -311,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 }); @@ -326,22 +333,26 @@ 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. `final` is close()'s last attempt: it ignores the - // backoff and counts what it could not send as dropped. + // 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 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(async () => { - const batch = outbox.take(final); - if (batch.length === 0) return; - try { - await post(batch, timeoutMs); - outbox.delivered(); - } catch (error) { - // Never slow or fail a run: requeue for a later retry, drop-and-count only as the last resort. - warnOnce(`could not reach the observe daemon at ${url}`, error); - if (final) droppedCount += batch.length; - else outbox.failed(batch); - } - }); + const run = chain.then(() => send(timeoutMs, final)).catch(() => undefined); chain = run; return run; }; @@ -390,10 +401,13 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe // no process.off in this runtime } } - // Graceful exit: say goodbye, then one final flush with the ~2s cap. push() - // directly (enqueue drops after close); a full buffer evicts its oldest event - // rather than the deregister. - push({ type: 'agent.deregister', data: { agent_id: identity.agentId, ts: now() } }); + // 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); };