From 8a3d4d4766a0f40b7e0a77f39a162becb66ea8b1 Mon Sep 17 00:00:00 2001 From: Cam Quilici Date: Fri, 16 Oct 2026 14:43:17 -0500 Subject: [PATCH] fix(api): stream stored request timelines without expanding JSON --- docs/data-pipeline.md | 8 +++ .../app/api/v1/request-timeline/route.test.ts | 68 +++++++++++++++++++ .../src/app/api/v1/request-timeline/route.ts | 24 ++++++- packages/app/src/lib/api/api-documentation.ts | 4 +- packages/app/src/lib/api/api-route-catalog.ts | 9 +-- packages/app/src/lib/cache/api-cache.test.ts | 49 ++++++++++++- packages/app/src/lib/cache/api-cache.ts | 33 +++++++++ .../src/queries/request-timeline-db.test.ts | 34 +++++++++- packages/db/src/queries/request-timeline.ts | 43 ++++++++++++ 9 files changed, 262 insertions(+), 10 deletions(-) create mode 100644 packages/app/src/app/api/v1/request-timeline/route.test.ts diff --git a/docs/data-pipeline.md b/docs/data-pipeline.md index e677622a4..1ea486b8e 100644 --- a/docs/data-pipeline.md +++ b/docs/data-pipeline.md @@ -781,3 +781,11 @@ empty receipt says `no_power_points`, never that power coverage was validated. The workflow retains both the input manifest and verification receipt. Cache invalidation errors fail the workflow instead of being swallowed. Imported P75/P90 ledger edits trigger the existing reviewed override workflow. + +### Serving large request timelines + +The request-timeline API streams current-version `request_timeline_gz` directly +in bounded 1 MiB database reads. It preserves every request and the existing JSON +contract and CDN headers, without expanding the full document or putting it through +the parsed-object Blob cache. Reads follow response backpressure and stop on +cancellation. Legacy or stale timelines use the existing computation/cache path. diff --git a/packages/app/src/app/api/v1/request-timeline/route.test.ts b/packages/app/src/app/api/v1/request-timeline/route.test.ts new file mode 100644 index 000000000..6a23d5bc8 --- /dev/null +++ b/packages/app/src/app/api/v1/request-timeline/route.test.ts @@ -0,0 +1,68 @@ +import { gzipSync, gunzipSync } from 'node:zlib'; +import { NextRequest } from 'next/server'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('next/cache', () => ({ revalidateTag: vi.fn(), unstable_cache: vi.fn() })); +vi.mock('@/lib/cache/blob-cache', () => ({ + blobGet: vi.fn(() => Promise.resolve(null)), + blobSet: vi.fn(), +})); +vi.mock('@semianalysisai/inferencex-db/connection', () => ({ getDb: () => ({}) })); +vi.mock('@semianalysisai/inferencex-db/queries/request-timeline', () => ({ + getStoredRequestTimelineGz: vi.fn(), + getRequestTimeline: vi.fn(), +})); + +import { + getStoredRequestTimelineGz, + getRequestTimeline, +} from '@semianalysisai/inferencex-db/queries/request-timeline'; +import { GET } from './route'; + +const request = (query: string) => + new NextRequest(`http://localhost/api/v1/request-timeline${query}`); +beforeEach(() => vi.resetAllMocks()); + +describe('request timeline route', () => { + it('serves stored gzip unchanged without materializing the timeline', async () => { + const gz = gzipSync( + JSON.stringify({ requests: [{ cid: 'full-record', srcTrace: 'source', ack: null }] }), + ); + vi.mocked(getStoredRequestTimelineGz).mockResolvedValue( + (async function* () { + yield gz; + })(), + ); + const response = await GET(request('?id=445162')); + expect(response.status).toBe(200); + expect(Buffer.from(await response.arrayBuffer()).equals(gz)).toBe(true); + expect(getRequestTimeline).not.toHaveBeenCalled(); + }); + + it('preserves the legacy fallback and its missing-point status', async () => { + vi.mocked(getStoredRequestTimelineGz).mockResolvedValue(null); + vi.mocked(getRequestTimeline).mockResolvedValue(null); + const response = await GET(request('?id=123')); + expect(response.status).toBe(404); + }); + + it('preserves the complete legacy response', async () => { + const timeline = { version: 1, startNs: 1, endNs: 2, durationS: 1e-9, requests: [] }; + vi.mocked(getStoredRequestTimelineGz).mockResolvedValue(null); + vi.mocked(getRequestTimeline).mockResolvedValue(timeline); + const response = await GET(request('?id=123')); + expect(JSON.parse(gunzipSync(Buffer.from(await response.arrayBuffer())).toString())).toEqual( + timeline, + ); + }); + + it.each(['', '?id=0', '?id=-1', '?id=1.5', '?id=9007199254740992'])( + 'rejects invalid ids before database access: %s', + async (query) => { + const response = await GET(request(query)); + expect(response.status).toBe(400); + expect(getStoredRequestTimelineGz).not.toHaveBeenCalled(); + expect(getRequestTimeline).not.toHaveBeenCalled(); + }, + ); +}); diff --git a/packages/app/src/app/api/v1/request-timeline/route.ts b/packages/app/src/app/api/v1/request-timeline/route.ts index f4d89fd93..dc2517114 100644 --- a/packages/app/src/app/api/v1/request-timeline/route.ts +++ b/packages/app/src/app/api/v1/request-timeline/route.ts @@ -1,12 +1,16 @@ +import type { NextRequest } from 'next/server'; + import { REQUEST_TIMELINE_VERSION } from '@semianalysisai/inferencex-db/etl/compute-request-timeline'; import { getDb } from '@semianalysisai/inferencex-db/connection'; import { getRequestTimeline, + getStoredRequestTimelineGz, type RequestTimeline, } from '@semianalysisai/inferencex-db/queries/request-timeline'; -import { cachedQuery } from '@/lib/cache/api-cache'; +import { cachedGzipJson, cachedQuery } from '@/lib/cache/api-cache'; +import { PUBLIC_API_ERRORS, publicApiError } from '@/lib/api/public-api-errors'; import { idQueryRoute } from '../id-routes'; @@ -33,7 +37,23 @@ const getCachedRequestTimeline = cachedQuery( * plus TTFT, ISL, OSL, conversation id, turn index, worker id. 404 if the * point has no stored profile_export.jsonl blob. */ -export const GET = idQueryRoute({ +const legacyGET = idQueryRoute({ logLabel: 'request timeline', fetch: getCachedRequestTimeline, }); + +/** Current gzip timelines bypass parsed-object caches to keep memory bounded. */ +export async function GET(request: NextRequest): Promise { + const id = Number(request.nextUrl.searchParams.get('id')); + if (!Number.isSafeInteger(id) || id <= 0) { + return publicApiError(PUBLIC_API_ERRORS.idRequired, 400); + } + try { + const chunks = await getStoredRequestTimelineGz(getDb(), id); + if (chunks) return cachedGzipJson(chunks); + return legacyGET(request); + } catch (error) { + console.error('Error fetching request timeline:', error); + return publicApiError(PUBLIC_API_ERRORS.internal, 500); + } +} diff --git a/packages/app/src/lib/api/api-documentation.ts b/packages/app/src/lib/api/api-documentation.ts index 2e3442e50..644030164 100644 --- a/packages/app/src/lib/api/api-documentation.ts +++ b/packages/app/src/lib/api/api-documentation.ts @@ -2385,8 +2385,8 @@ export const apiOperations: readonly ApiOperation[] = [ path: '/api/v1/request-timeline', summary: text('Read a request timeline', '读取请求时间线'), description: text( - 'Returns a versioned benchmark window and per-request replay identity, source provenance, dispatch, acknowledgement, completion, token, phase, worker, and cancellation timing.', - '返回带版本号的基准测试时间窗口，以及每个请求的重放标识、来源追溯信息、调度/确认/完成/取消时间、token 指标、阶段和 worker 信息。', + 'Returns a versioned benchmark window and per-request replay identity, source provenance, dispatch, acknowledgement, completion, token, phase, worker, and cancellation timing. Current stored timelines stream directly as gzip JSON with all request records preserved; legacy timelines retain the same response shape.', + '返回带版本号的基准测试时间窗口，以及每个请求的重放标识、来源追溯信息、调度/确认/完成/取消时间、token 指标、阶段和 worker 信息。当前版本的已存储时间线以 gzip JSON 直接流式传输，保留全部请求记录；旧版时间线的响应结构不变。', ), audience: 'public', stability: 'beta', diff --git a/packages/app/src/lib/api/api-route-catalog.ts b/packages/app/src/lib/api/api-route-catalog.ts index 66e59c969..9e48315bf 100644 --- a/packages/app/src/lib/api/api-route-catalog.ts +++ b/packages/app/src/lib/api/api-route-catalog.ts @@ -880,7 +880,7 @@ export const apiRouteCatalog = [ method: 'GET', classification: 'published-read', operationId: 'get-request-timeline', - sourceSha256: 'f707e6e2f4d2eccf46882a89d73d26ba34ac4fa58e8bad925461aa7956e5a018', + sourceSha256: '792f054fc06e9e1cf59ce897cfc2b661f34eb54e501c72923409c4a05cc6973a', }, { source: 'src/app/api/v1/resident-sequence-lengths/route.ts', @@ -1439,7 +1439,7 @@ export const apiContractSourceDigests = [ }, { source: 'src/lib/cache/api-cache.ts', - sourceSha256: 'cf710c1dca9cfae794cf3d14d3dad6a153bbd7151cedc9d9f0b27b792aeade24', + sourceSha256: 'd2564d306c7794d0fb79a8996991783c6b7ef2d1780e1fff31c08b81cfbb03b9', reviewArea: { en: 'Public CDN tags, cache lifetimes, Blob key dimensions, and purge behavior.', zh: '公开 CDN 标签、缓存时长、Blob 键维度和清除行为。', @@ -1643,8 +1643,9 @@ export const apiContractSourceDigests = [ // Reviewed for gzip request-timeline storage (migration 017): the reader // prefers request_timeline_gz and falls back to legacy JSONB, and logs a // stored timeline it cannot read. It returns the same RequestTimeline - // document, nullability, 404 and caching, so the docs stand. - sourceSha256: 'c3bb72c5af3771667542f7676255fe32f8e3e28c3e5800298b663d22f45c5f2a', + // document; current stored gzip can also stream directly with unchanged + // CDN headers, bypassing the parsed-object Blob cache to bound memory. + sourceSha256: '3dc8d83421ac5b468aad7914f19fe8c9f081713dad86222ec395d01df28b3cb2', reviewArea: { en: 'Request timeline metadata, request event records, units, and nullable timings.', zh: '请求时间线元数据、请求事件记录、单位和可空计时字段。', diff --git a/packages/app/src/lib/cache/api-cache.test.ts b/packages/app/src/lib/cache/api-cache.test.ts index 4b28f5ec2..b2a5010f2 100644 --- a/packages/app/src/lib/cache/api-cache.test.ts +++ b/packages/app/src/lib/cache/api-cache.test.ts @@ -1,3 +1,4 @@ +import { gzipSync, gunzipSync } from 'node:zlib'; import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'; // Mock next/cache before importing the module under test @@ -16,7 +17,7 @@ vi.mock('./blob-cache', () => ({ blobPurge: vi.fn(), })); -import { cachedQuery, purgeAll, purgeCollectiveX, cachedJson } from './api-cache'; +import { cachedQuery, purgeAll, purgeCollectiveX, cachedJson, cachedGzipJson } from './api-cache'; import { revalidateTag, unstable_cache } from 'next/cache'; import { blobGet, blobSet, blobPurge } from './blob-cache'; @@ -332,3 +333,49 @@ describe('cachedJson', () => { expect(JSON.parse(text)).toEqual(data); }); }); + +describe('cachedGzipJson', () => { + it('preserves complete gzip bytes and the existing CDN headers', async () => { + const data = { requests: [{ cid: 'trace-é', cancelled: false, ttftMs: null }] }; + const gzip = gzipSync(JSON.stringify(data)); + async function* chunks() { + yield gzip.subarray(0, 13); + yield gzip.subarray(13); + } + const response = cachedGzipJson(chunks()); + expect(Object.fromEntries(response.headers)).toEqual( + Object.fromEntries(cachedJson(data).headers), + ); + const received = Buffer.from(await response.arrayBuffer()); + expect(received.equals(gzip)).toBe(true); + expect(JSON.parse(gunzipSync(received).toString())).toEqual(data); + expect(mockBlobGet).not.toHaveBeenCalled(); + expect(mockBlobSet).not.toHaveBeenCalled(); + }); + + it('does not fetch ahead and stops fetching after cancellation', async () => { + const next = vi.fn().mockResolvedValue({ done: false, value: new Uint8Array([1]) }); + const cleanup = vi.fn().mockResolvedValue({ done: true }); + const response = cachedGzipJson({ [Symbol.asyncIterator]: () => ({ next, return: cleanup }) }); + await Promise.resolve(); + expect(next).not.toHaveBeenCalled(); + const reader = response.body!.getReader(); + await reader.read(); + expect(next).toHaveBeenCalledTimes(1); + await reader.cancel(); + expect(cleanup).toHaveBeenCalledTimes(1); + expect(next).toHaveBeenCalledTimes(1); + }); + + it('fails the response stream rather than silently truncating on read errors', async () => { + const cleanup = vi.fn().mockResolvedValue({ done: true }); + const response = cachedGzipJson({ + [Symbol.asyncIterator]: () => ({ + next: () => Promise.reject(new Error('chunk failed')), + return: cleanup, + }), + }); + await expect(response.arrayBuffer()).rejects.toThrow('chunk failed'); + expect(cleanup).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/app/src/lib/cache/api-cache.ts b/packages/app/src/lib/cache/api-cache.ts index eb10a1e0c..d45262971 100644 --- a/packages/app/src/lib/cache/api-cache.ts +++ b/packages/app/src/lib/cache/api-cache.ts @@ -268,3 +268,36 @@ export function cachedJson( export function purgeOperatorX(): void { revalidateTag(operatorXCacheTag(), { expire: 0 }); } + +/** Serve existing gzip bytes with the same CDN contract, pulling only on demand. */ +export function cachedGzipJson(chunks: AsyncIterable): Response { + const iterator = chunks[Symbol.asyncIterator](); + let cancelled = false; + const body = new ReadableStream( + { + async pull(controller) { + try { + const next = await iterator.next(); + if (cancelled) return; + if (next.done) controller.close(); + else controller.enqueue(next.value); + } catch (error) { + if (!cancelled) controller.error(error); + await iterator.return?.(); + } + }, + async cancel() { + cancelled = true; + await iterator.return?.(); + }, + }, + { highWaterMark: 0 }, + ); + return new Response(body, { + headers: { + 'Content-Type': 'application/json', + 'Content-Encoding': 'gzip', + ...cdnHeaders(), + }, + }); +} diff --git a/packages/db/src/queries/request-timeline-db.test.ts b/packages/db/src/queries/request-timeline-db.test.ts index 8e37365df..519a499b2 100644 --- a/packages/db/src/queries/request-timeline-db.test.ts +++ b/packages/db/src/queries/request-timeline-db.test.ts @@ -9,7 +9,7 @@ import { migratedPglite, pgliteSql, type PgliteSql } from '../lib/test-pglite'; import * as agenticShared from './agentic-shared'; import { encodeRequestChartData, getRequestChartData } from './request-chart-data'; -import { getRequestTimeline } from './request-timeline'; +import { getRequestTimeline, getStoredRequestTimelineGz } from './request-timeline'; import { getTraceHistograms } from './trace-histograms'; // Expose the self-heal promise so the test can await the real write-back. @@ -192,3 +192,35 @@ describe('request timeline storage on PostgreSQL', () => { expect(writeBack).toHaveBeenCalledTimes(1); }); }); + +describe('stored gzip timeline streaming', () => { + it('returns the exact complete document in bounded slices without decoding', async () => { + const chunks = await getStoredRequestTimelineGz(sql, 102, 23); + expect(chunks).not.toBeNull(); + const bytes: Buffer[] = []; + for await (const chunk of chunks!) { + expect(chunk.length).toBeLessThanOrEqual(23); + bytes.push(chunk); + } + expect(bytes.length).toBeGreaterThan(1); + expect(gunzipRequestTimeline(Buffer.concat(bytes))).toEqual(gzipOnly); + }); + + it('leaves legacy and missing points to the existing fallback', async () => { + expect(await getStoredRequestTimelineGz(sql, 101)).toBeNull(); + expect(await getStoredRequestTimelineGz(sql, 999999)).toBeNull(); + }); + + it('rejects a disappearing row instead of completing a truncated response', async () => { + const chunks = await getStoredRequestTimelineGz(sql, 102, 23); + const first = await chunks!.next(); + expect(first.value).toHaveLength(23); + const [row] = await sql`select request_timeline_gz from agentic_trace_replay where id = 12`; + await sql`update agentic_trace_replay set request_timeline_gz = null where id = 12`; + try { + await expect(chunks!.next()).rejects.toThrow('Stored timeline changed'); + } finally { + await sql`update agentic_trace_replay set request_timeline_gz = ${row.request_timeline_gz} where id = 12`; + } + }); +}); diff --git a/packages/db/src/queries/request-timeline.ts b/packages/db/src/queries/request-timeline.ts index f67f2efc6..1a753356a 100644 --- a/packages/db/src/queries/request-timeline.ts +++ b/packages/db/src/queries/request-timeline.ts @@ -97,6 +97,49 @@ export async function readRequestTimelineGz( return Buffer.concat(slices, byteLength); } +/** + * Stream the exact stored JSON gzip without expanding the full request array. + * Return null for legacy/stale rows so callers retain the existing fallback. + */ +export async function getStoredRequestTimelineGz( + sql: DbClient, + benchmarkResultId: number, + sliceBytes = 1024 * 1024, +): Promise | null> { + if (!Number.isSafeInteger(sliceBytes) || sliceBytes <= 0) { + throw new Error('Invalid timeline slice size'); + } + const [row] = await sql` + select atr.id as trace_replay_id, octet_length(atr.request_timeline_gz) as gz_bytes + from benchmark_results br + join agentic_trace_replay atr on atr.id = br.trace_replay_id + where br.id = ${benchmarkResultId} + and atr.request_timeline_version = ${REQUEST_TIMELINE_VERSION} + and atr.request_timeline_gz is not null + `; + if (!row) return null; + const size = Number(row.gz_bytes); + if (!Number.isSafeInteger(size) || size <= 0) throw new Error('Invalid stored timeline size'); + const traceReplayId = Number(row.trace_replay_id); + return (async function* () { + for (let offset = 0; offset < size; offset += sliceBytes) { + const length = Math.min(sliceBytes, size - offset); + const [part] = await sql` + select substring(request_timeline_gz from ${offset + 1}::int for ${length}::int) as slice + from agentic_trace_replay + where id = ${traceReplayId} + and request_timeline_version = ${REQUEST_TIMELINE_VERSION} + and octet_length(request_timeline_gz) = ${size} + `; + const bytes = part?.slice; + if (!(bytes instanceof Uint8Array) || bytes.length !== length) { + throw new Error(`Stored timeline changed or ended at byte ${offset} of ${size}`); + } + yield Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength); + } + })(); +} + /** * Yield a legacy `request_timeline` JSONB as text in bounded chunks. Throws * when the timeline is missing or longer than the chunk limit.