diff --git a/backend/__tests__/service/podRecords.realPg.test.js b/backend/__tests__/service/podRecords.realPg.test.js index ad76ec56ad..21220c0603 100644 --- a/backend/__tests__/service/podRecords.realPg.test.js +++ b/backend/__tests__/service/podRecords.realPg.test.js @@ -176,6 +176,36 @@ real('task record route on real PostgreSQL and MongoDB', () => { ]); }); + it('serves the record in seq pages, and refuses a malformed cursor before any read', async () => { + expect((await claim()).status).toBe(200); + expect((await request(app).patch(`/api/v1/tasks/${pod._id}/TASK-1`).set(as(AGENT_TOKEN)).send({ status: 'blocked' })).status).toBe(200); + expect((await claim()).status).toBe(200); + await flushRecordWrites(); + const asOwner = generateTestToken(owner._id); + const whole = await read(asOwner); + expect(whole.headers['x-record-next-after']).toBeUndefined(); + + const first = await read(asOwner, pod._id, 'TASK-1', '?limit=2'); + expect(first.status).toBe(200); + expect(first.headers['x-record-next-after']).toBe('2'); + const rest = await read(asOwner, pod._id, 'TASK-1', '?after=2&limit=2'); + expect(rest.headers['x-record-next-after']).toBeUndefined(); + // Both pages name the whole chain's head, and together are the whole chain. + expect([first, rest].map((page) => [page.headers['x-record-entries'], page.headers['x-record-head']])) + .toEqual([['3', whole.headers['x-record-head']], ['3', whole.headers['x-record-head']]]); + expect(first.text + rest.text).toBe(whole.text); + expect(verifyRecord(first.text + rest.text)).toMatchObject({ ok: true, entries: 3 }); + const otlp = await read(asOwner, pod._id, 'TASK-1', '?format=otlp&after=2'); + expect(otlp.status).toBe(200); + expect(otlp.headers['x-record-next-after']).toBeUndefined(); + + const readChain = jest.spyOn(repository, 'readChain'); + const bad = ['?after=-1', '?after=1.5', '?after=abc', '?after=1&after=2', '?limit=0', '?limit=1001', '?limit=']; + const answers = await Promise.all(bad.map((query) => read(asOwner, pod._id, 'TASK-1', query))); + expect(answers.map((res) => [res.status, res.body.error])).toEqual(bad.map(() => [400, 'record_bad_cursor'])); + expect(readChain).not.toHaveBeenCalled(); + }); + it('refuses everyone else before any record byte is read', async () => { await claim(); await flushRecordWrites(); diff --git a/backend/__tests__/service/recordChains.realPg.test.js b/backend/__tests__/service/recordChains.realPg.test.js index 1a8b72e3f0..5cf22766b0 100644 --- a/backend/__tests__/service/recordChains.realPg.test.js +++ b/backend/__tests__/service/recordChains.realPg.test.js @@ -10,7 +10,7 @@ const path = require('path'); const { runPgMigrations } = require('../../config/run-pg-migrations'); const { createRealPgSchema } = require('../utils/realPgSchema'); const { appendRecord, readRecordExport, BOARD_ACTOR } = require('../../services/recordService'); -const { appendToChain } = require('../../repositories/recordChainRepository'); +const { appendToChain, MAX_PAGE_LINES } = require('../../repositories/recordChainRepository'); const { verifyRecord } = require('../../vendor/seats/src/record/format'); const real = process.env.INTEGRATION_TEST === 'true' ? describe : describe.skip; @@ -150,11 +150,41 @@ real('record chains on real PostgreSQL', () => { it('reads an unwritten chain as empty, and a missing store as an error', async () => { expect(await readRecordExport({ podId: POD, taskId: 'TASK-never' }, pool)) - .toEqual({ runId: null, entries: 0, head: null, jsonl: '' }); + .toEqual({ + runId: null, entries: 0, head: null, jsonl: '', nextAfter: null, + }); const { readChain, RecordStoreUnavailableError } = require('../../repositories/recordChainRepository'); await expect(readChain({ podId: POD }, null)).rejects.toBeInstanceOf(RecordStoreUnavailableError); }); + it('reads a long chain in pages that each name the whole chain\'s head and join into one that verifies', async () => { + const ref = { podId: POD, taskId: 'TASK-7' }; + const total = MAX_PAGE_LINES + 2; + await appendRecord(ref, 'bulk', Array.from({ length: total }, () => ({ kind: 'task.claimed', actor: AGENT })), pool); + + // No cursor still reads one bounded page, never the whole history. + const first = await readRecordExport(ref, pool); + expect(kinds(first.jsonl)).toHaveLength(MAX_PAGE_LINES); + expect(first).toMatchObject({ entries: total, nextAfter: MAX_PAGE_LINES }); + expect(kinds((await readRecordExport(ref, pool, { limit: MAX_PAGE_LINES * 5 })).jsonl)).toHaveLength(MAX_PAGE_LINES); + + const pages = []; + let after = 0; + while (after !== null) { + // eslint-disable-next-line no-await-in-loop + const page = await readRecordExport(ref, pool, { after, limit: 400 }); + expect(page).toMatchObject({ entries: total, head: first.head, runId: first.runId }); + pages.push(page.jsonl); + after = page.nextAfter; + } + expect(pages.map((jsonl) => kinds(jsonl).length)).toEqual([400, 400, total - 800]); + expect(kinds(pages[1])[0].seq).toBe(401); + expect(verifyRecord(pages.join(''))).toMatchObject({ ok: true, entries: total, head: first.head }); + + expect(await readRecordExport(ref, pool, { after: total, limit: 10 })) + .toMatchObject({ entries: total, jsonl: '', nextAfter: null }); + }); + it('stores lines byte for byte, so an edited row fails verification', async () => { const ref = { podId: POD, taskId: 'TASK-6' }; await appendRecord(ref, 'a', [{ kind: 'task.claimed', actor: AGENT }], pool); diff --git a/backend/__tests__/unit/routes/podRecords.otlpFullChain.test.js b/backend/__tests__/unit/routes/podRecords.otlpFullChain.test.js new file mode 100644 index 0000000000..6eb36956c5 --- /dev/null +++ b/backend/__tests__/unit/routes/podRecords.otlpFullChain.test.js @@ -0,0 +1,72 @@ +/** + * The record route pages the NDJSON read, but ?format=otlp has no paging of + * its own. An OTLP read with no cursor must cover the whole chain, not the + * first page, or a long chain exports a silently short trace. + */ +const express = require('express'); +const request = require('supertest'); + +const POD = 'aaaaaaaaaaaaaaaaaaaaaa01'; +const USER = 'bbbbbbbbbbbbbbbbbbbbbbb1'; + +const mockReadRecordExport = jest.fn(); +const mockToOtlp = jest.fn(() => ({ resourceSpans: [] })); + +jest.mock('../../../services/recordService', () => ({ + isRecordEnabled: () => true, + readRecordExport: (...args) => mockReadRecordExport(...args), + toOtlp: (...args) => mockToOtlp(...args), + RecordStoreUnavailableError: class RecordStoreUnavailableError extends Error {}, +})); +jest.mock('../../../middleware/dualAuth', () => (req, res, next) => { + req.userId = USER; + next(); +}); +jest.mock('../../../models/Pod', () => ({ + findById: () => ({ select: () => ({ lean: async () => ({ _id: POD, type: 'chat', members: [USER] }) }) }), +})); +jest.mock('../../../models/Task', () => ({ findOne: jest.fn() })); +jest.mock('../../../services/dmService', () => ({ canViewPod: async () => true })); +jest.mock('../../../middleware/ipRateLimit', () => ({ + cloudflareIpRateLimitKeyGenerator: () => 'test', +})); + +const router = require('../../../routes/podRecords'); + +const app = express(); +app.use('/api/pods', router); + +// Three pages of one line each, as readRecordExport returns them. +const pages = { + start: { jsonl: '{"seq":1}\n', nextAfter: 1 }, + 1: { jsonl: '{"seq":2}\n', nextAfter: 2 }, + 2: { jsonl: '{"seq":3}\n', nextAfter: null }, +}; + +beforeEach(() => { + jest.clearAllMocks(); + mockReadRecordExport.mockImplementation(async (_ref, _db, page = {}) => ( + page.after === undefined ? pages.start : pages[page.after] + )); +}); + +describe('GET /api/pods/:podId/record?format=otlp', () => { + test('an OTLP read with no cursor exports every page of the chain', async () => { + const res = await request(app).get(`/api/pods/${POD}/record?format=otlp`); + expect(res.status).toBe(200); + expect(mockToOtlp).toHaveBeenCalledWith('{"seq":1}\n{"seq":2}\n{"seq":3}\n'); + }); + + test('a paged NDJSON read still returns one page with its cursor', async () => { + const res = await request(app).get(`/api/pods/${POD}/record`); + expect(res.status).toBe(200); + expect(res.text).toBe('{"seq":1}\n'); + expect(res.headers['x-record-next-after']).toBe('1'); + expect(mockReadRecordExport).toHaveBeenCalledTimes(1); + }); + + test('an OTLP read with an explicit cursor stays one page', async () => { + await request(app).get(`/api/pods/${POD}/record?format=otlp&after=1`); + expect(mockToOtlp).toHaveBeenCalledWith('{"seq":2}\n'); + }); +}); diff --git a/backend/repositories/recordChainRepository.ts b/backend/repositories/recordChainRepository.ts index 707841263b..ede5d31133 100644 --- a/backend/repositories/recordChainRepository.ts +++ b/backend/repositories/recordChainRepository.ts @@ -24,8 +24,20 @@ export type StoredChain = { headSeq: number; headHash: string | null; lines: string[]; + /** The seq to pass as `after` for the next page; null when this page reached the head. */ + nextAfter: number | null; }; +/** The most lines one read returns. A chain longer than this is read in pages. */ +export const MAX_PAGE_LINES = 1000; + +/** + * A seq cursor: the lines after seq `after`, at most `limit` of them. A page's + * first line carries `prev`, the hash of line `after`, so pages read in order + * concatenate into one chain that still verifies. + */ +export type ChainPage = { after?: number; limit?: number }; + /** Thrown when no PostgreSQL is configured; a reader must not answer "empty". */ export class RecordStoreUnavailableError extends Error { constructor() { @@ -111,25 +123,33 @@ export async function appendToChain( } /** - * Read a chain's lines up to its stored head. Lines and head commit together, - * so bounding by the head read first gives a consistent prefix without a - * transaction. A chain that was never written returns null. + * Read one page of a chain's lines, bounded by its stored head. Lines and head + * commit together, so bounding by the head read first gives a consistent + * prefix without a transaction. A chain that was never written returns null. */ -export async function readChain(ref: ChainRef, db: PgPool | null = pool): Promise { +export async function readChain( + ref: ChainRef, + db: PgPool | null = pool, + page: ChainPage = {}, +): Promise { const store = requirePool(db); + const after = Math.max(0, Math.trunc(Number(page.after) || 0)); + const limit = Math.min(MAX_PAGE_LINES, Math.max(1, Math.trunc(Number(page.limit) || MAX_PAGE_LINES))); const chain = (await store.query( 'SELECT id, run_id, head_seq, head_hash FROM record_chains WHERE pod_id = $1 AND task_id = $2', [ref.podId, taskKey(ref)], )).rows[0] as { id: string; run_id: string; head_seq: number; head_hash: string | null } | undefined; if (!chain) return null; const rows = (await store.query( - 'SELECT line FROM record_lines WHERE chain_id = $1 AND seq <= $2 ORDER BY seq', - [chain.id, chain.head_seq], - )).rows as Array<{ line: string }>; + 'SELECT seq, line FROM record_lines WHERE chain_id = $1 AND seq > $2 AND seq <= $3 ORDER BY seq LIMIT $4', + [chain.id, after, chain.head_seq, limit], + )).rows as Array<{ seq: number; line: string }>; + const last = rows.length > 0 ? Number(rows[rows.length - 1].seq) : after; return { runId: chain.run_id, headSeq: chain.head_seq, headHash: chain.head_hash, lines: rows.map((row) => row.line), + nextAfter: rows.length > 0 && last < chain.head_seq ? last : null, }; } diff --git a/backend/routes/podRecords.ts b/backend/routes/podRecords.ts index f7002e89e1..b33a2868df 100644 --- a/backend/routes/podRecords.ts +++ b/backend/routes/podRecords.ts @@ -2,8 +2,9 @@ import express from 'express'; import rateLimit from 'express-rate-limit'; import mongoose from 'mongoose'; import { cloudflareIpRateLimitKeyGenerator } from '../middleware/ipRateLimit'; +import { MAX_PAGE_LINES } from '../repositories/recordChainRepository'; import { - isRecordEnabled, readRecordExport, RecordStoreUnavailableError, toOtlp, + isRecordEnabled, readRecordExport, RecordStoreUnavailableError, toOtlp, type ChainPage, } from '../services/recordService'; /* eslint-disable @typescript-eslint/no-require-imports, global-require */ @@ -30,6 +31,19 @@ type RecordReq = express.Request & { const notFound = (res: express.Response) => res.status(404).json({ error: 'record_not_found' }); +// A page cursor: `after` is a seq (0 is the start), `limit` 1..MAX_PAGE_LINES. +const SEQ = /^\d{1,9}$/; +const readPage = (query: express.Request['query']): ChainPage | null => { + const { after, limit } = query; + if (after !== undefined && (typeof after !== 'string' || !SEQ.test(after))) return null; + if (limit !== undefined && (typeof limit !== 'string' || !SEQ.test(limit) + || Number(limit) < 1 || Number(limit) > MAX_PAGE_LINES)) return null; + return { + ...(after !== undefined ? { after: Number(after) } : {}), + ...(limit !== undefined ? { limit: Number(limit) } : {}), + }; +}; + /** * A chain as record-v1 JSONL, or `?format=otlp` for the spec's OTLP/JSON * export. `taskId` null is the pod's taskless chain (grants and the calls @@ -41,11 +55,17 @@ const notFound = (res: express.Response) => res.status(404).json({ error: 'recor * credential-owner ids verbatim, and a hash-chained line can't be redacted * without breaking verification. That is the same set runTraces shows those * ids to, so admins and agent-DM observers who pass canViewPod get a 403. + * + * One read returns at most MAX_PAGE_LINES lines after `?after=`. While + * more remain, `X-Record-Next-After` names the cursor for the next page; the + * head headers always describe the whole chain. */ const serveRecord = async (req: RecordReq, res: express.Response, taskId: string | null) => { try { const { podId } = req.params; if (!mongoose.isObjectIdOrHexString(podId) || (taskId !== null && !TASK_ID.test(taskId))) return notFound(res); + const page = readPage(req.query); + if (!page) return res.status(400).json({ error: 'record_bad_cursor' }); if (!isRecordEnabled(podId)) return notFound(res); const userId = req.userId || req.user?._id || req.agentUser?._id; const pod = await Pod.findById(podId).select('members type').lean(); @@ -57,7 +77,19 @@ const serveRecord = async (req: RecordReq, res: express.Response, taskId: string let record; try { - record = await readRecordExport({ podId: String(pod._id), ...(taskId !== null ? { taskId } : {}) }); + const ref = { podId: String(pod._id), ...(taskId !== null ? { taskId } : {}) }; + record = await readRecordExport(ref, undefined, page); + // OTLP has no paging of its own: an export of one page would be a + // silently short trace. An OTLP read with no cursor covers the chain. + if (req.query.format === 'otlp' && page.after === undefined && page.limit === undefined) { + const lines = [record.jsonl]; + for (let next = record.nextAfter; next !== null;) { + const more = await readRecordExport(ref, undefined, { after: next }); + lines.push(more.jsonl); + next = more.nextAfter; + } + record = { ...record, jsonl: lines.join(''), nextAfter: null }; + } } catch (error) { // A store outage is an error, never an apparently empty record. if (!(error instanceof RecordStoreUnavailableError)) { @@ -71,6 +103,7 @@ const serveRecord = async (req: RecordReq, res: express.Response, taskId: string // The spec's outside record of the head (ยง4): it covers the last line. if (record.head) res.set('X-Record-Head', record.head); if (record.runId) res.set('X-Record-Run-Id', record.runId); + if (record.nextAfter !== null) res.set('X-Record-Next-After', String(record.nextAfter)); if (req.query.format === 'otlp') return res.json(toOtlp(record.jsonl)); res.type('application/x-ndjson; charset=utf-8'); return res.send(record.jsonl); diff --git a/backend/services/recordService.ts b/backend/services/recordService.ts index fedfad8a82..9f87083603 100644 --- a/backend/services/recordService.ts +++ b/backend/services/recordService.ts @@ -13,7 +13,7 @@ */ import { createHash } from 'crypto'; import { - appendToChain, readChain, RecordStoreUnavailableError, type ChainRef, + appendToChain, readChain, RecordStoreUnavailableError, type ChainPage, type ChainRef, } from '../repositories/recordChainRepository'; type Store = Parameters[1]; @@ -42,7 +42,7 @@ export type RecordLineInput = { actor?: RecordActor | null; data?: Record; }; -export { RecordStoreUnavailableError, type ChainRef }; +export { RecordStoreUnavailableError, type ChainPage, type ChainRef }; export const BOARD_ACTOR: RecordActor = format.BOARD_ACTOR; /** @@ -190,17 +190,24 @@ export type RecordExport = { entries: number; head: string | null; jsonl: string; + /** Pass as `after` for the next page; null once a page reaches the head. */ + nextAfter: number | null; }; -/** A chain as stored, in record-v1's file form: one line per entry, each ending in \n. */ -export async function readRecordExport(ref: ChainRef, db?: Store): Promise { - const chain = await readChain(ref, db); - if (!chain) return { runId: null, entries: 0, head: null, jsonl: '' }; +/** + * One page of a chain as stored, in record-v1's file form: one line per entry, + * each ending in \n. `entries` and `head` describe the whole chain, not the + * page, so every page names the same head. + */ +export async function readRecordExport(ref: ChainRef, db?: Store, page: ChainPage = {}): Promise { + const chain = await readChain(ref, db, page); + if (!chain) return { runId: null, entries: 0, head: null, jsonl: '', nextAfter: null }; return { runId: chain.runId, - entries: chain.lines.length, + entries: chain.headSeq, head: chain.headHash, jsonl: chain.lines.map((line) => `${line}\n`).join(''), + nextAfter: chain.nextAfter, }; } diff --git a/docs/plans/task-513-server-record.md b/docs/plans/task-513-server-record.md index 73e1c69a60..5ee2bf1800 100644 --- a/docs/plans/task-513-server-record.md +++ b/docs/plans/task-513-server-record.md @@ -83,6 +83,20 @@ endpoint. Every writer and the endpoint are off unless the pod is listed in - **Reads.** `GET /api/pods/:podId/record` serves the pod's taskless chain under the same gate as the task route. +### Decisions taken in slice 3 (TASK-561) + +- **Reads are paged by seq.** One read returns at most 1000 lines + (`MAX_PAGE_LINES`) after `?after=`, fewer with `?limit=`. A read with + no cursor is the first page, never the whole history. While lines remain, + `X-Record-Next-After` names the next cursor. `X-Record-Entries` and + `X-Record-Head` describe the whole chain on every page, so a client knows + where the chain ends before it reads that far. A page's first line carries + `prev`, the hash of line `after`. Pages read in order join into one chain + that `verifyRecord` accepts, and a single page on its own does not verify + unless it starts at seq 1. A malformed cursor gets a 400 before any record + is read. The v2 view (slice 4) calls the API from another origin. It can + read these headers only after CORS exposes them. + ## Measured inputs (2026-10-25) - TASK-504 landed in #2684, merge `ab5132313ad399d8fb990d69bc39db0ccf99e055`.