fix(api): stream large request timelines without expanding JSON

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.

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();
+    },
+  );
+});

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<Response> {
+  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);
+  }
+}

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',

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: '请求时间线元数据、请求事件记录、单位和可空计时字段。',

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

packages/app/src/lib/cache/api-cache.ts

@@ -268,3 +268,36 @@ export function cachedJson<T>(
 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<Uint8Array>): Response {
+  const iterator = chunks[Symbol.asyncIterator]();
+  let cancelled = false;
+  const body = new ReadableStream<Uint8Array>(
+    {
+      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(),
+    },
+  });
+}

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

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<AsyncGenerator<Buffer> | 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.