diff --git a/packages/agent/src/sync/auth/request-build.ts b/packages/agent/src/sync/auth/request-build.ts index fc697b0012..b70b6bcf04 100644 --- a/packages/agent/src/sync/auth/request-build.ts +++ b/packages/agent/src/sync/auth/request-build.ts @@ -182,7 +182,15 @@ export async function buildSyncRequestEnvelope(params: BuildSyncRequestParams): const requestedLimit = Number.isSafeInteger(limit) ? Math.max(1, Math.min(limit, SYNC_BYTE_BUDGET_MAX_ROWS)) : SYNC_PAGE_SIZE; - const useByteBudgetPage = !includeSharedMemory && phase === 'data' && requestedLimit > SYNC_PAGE_SIZE; + // Advertise byte-budget page mode for durable DATA and META (#1916/#1923). + // Additive/rolling-upgrade safe both directions: an OLD responder ignores the + // meta pageMode (its meta path is not byte-budget-gated → serves legacy meta), + // and a NEW responder treats a request WITHOUT meta pageMode as non-negotiated + // (plain meta serializer). The signed `limit` still rides the 500-row legacy + // cap below, so digests stay wire-compatible. + const useByteBudgetPage = !includeSharedMemory + && (phase === 'data' || phase === 'meta') + && requestedLimit > SYNC_PAGE_SIZE; const assetUals = rawAssetUals === undefined ? undefined : requireExactAssetUals(rawAssetUals); if (!needsAuth) { diff --git a/packages/agent/src/sync/requester/page-fetch.ts b/packages/agent/src/sync/requester/page-fetch.ts index 6b98b504dc..f2c52a48e5 100644 --- a/packages/agent/src/sync/requester/page-fetch.ts +++ b/packages/agent/src/sync/requester/page-fetch.ts @@ -300,6 +300,16 @@ export async function fetchSyncPages(params: FetchSyncPagesParams): Promise SYNC_PAGE_SIZE` (the fetch wrapper passes + // SYNC_REQUEST_PAGE_SIZE=8192 for every phase), NOT a wire-negotiated + // capability. Durable meta relies on it: since #1916 the responder byte-caps + // durable-meta pages, so a page can be short for byte reasons; a requester + // that treated "short = EOF" for meta could end the phase early. Every + // testnet-canary+ requester uses 8192 here, so short≠EOF holds for meta and + // data alike; a pre-canary requester using the 500-row cap is the only one + // that would regress, and only on an oversized (>4 MiB) meta subject. const usesByteBudgetPagination = syncPageSize > SYNC_PAGE_SIZE; let activePageSize = syncPageSize; let successfulPageSize = syncPageSize; diff --git a/packages/agent/src/sync/responder/graph-plan.ts b/packages/agent/src/sync/responder/graph-plan.ts index c5b20d7598..03673ecc26 100644 --- a/packages/agent/src/sync/responder/graph-plan.ts +++ b/packages/agent/src/sync/responder/graph-plan.ts @@ -30,6 +30,7 @@ import { type SyncResponderSnapshotBudget, } from './snapshot-budget.js'; import { estimateStringRowHeapBytes } from '../memory-telemetry.js'; +import { SYNC_BYTE_BUDGET_RESPONSE_BYTES } from '../../dkg-agent-constants.js'; import type { ChangelogSyncResponse, ChangelogDeltaRecord } from '../changelog/wire.js'; import { durableMetaDelegationSubjectAdmissionExpression } from './durable-meta-admission.js'; import { exactAssetFilterKey } from '../exact-assets.js'; @@ -549,6 +550,20 @@ export function compareRows(a: SyncRow, b: SyncRow): number { ); } +/** + * The `(g, s)` identity a durable `_meta` row belongs to. Durable meta is + * ordered by `(g, s, p, o)`; a graph-scoped assertion seal is ONE `(g, s)` + * subject whose rows — including the batch-local control field + * `dkg:assertionVersion` — MUST cross the wire together in a single round, else + * the receiver's per-round completeness check drops the control fields + * permanently (#1788). Used to snap durable-meta page boundaries to subject + * boundaries. The `\n` separator cannot occur inside an IRI, so distinct + * subjects never collide. + */ +function metaSubjectKey(row: SyncRow): string { + return `${row.g}\n${row.s}`; +} + function serializeResponderRow(row: SyncRow): string { return `${formatTerm(row.s)} <${assertSafeIri(row.p)}> ${formatTerm(row.o)} <${assertSafeIri(row.g)}> .`; } @@ -557,6 +572,180 @@ export function serializeResponderRows(rows: readonly SyncRow[]): string { return rows.map(serializeResponderRow).join('\n'); } +const RESPONDER_ROW_ENCODER = new TextEncoder(); + +/** + * Serialized (N-Quads, UTF-8) wire byte length of one responder row — the ONE + * byte model for response-frame budgets, shared by the byte-budget serializer + * and the subject-atomic extend so both reason about `maxResponseBytes` the same + * way (#1916). This is distinct from `estimateStringRowHeapBytes`, which + * estimates retained HEAP size for snapshot memory budgets only. + */ +function serializedResponderRowByteLength(row: SyncRow): number { + return RESPONDER_ROW_ENCODER.encode(serializeResponderRow(row)).byteLength; +} + +/** + * How a durable-meta page handles an admitted subject (or cumulative page) that + * cannot fit the frame-safe response budget: + * - `'byte-fit'` — NEGOTIATED (testnet-canary+) requesters: byte-cap the page, + * splitting a LONE oversized FIRST subject into a byte-fitting prefix. The + * requester paginates via empty=EOF, so a short page is NOT EOF — no metadata + * loss and forward progress is guaranteed. This is the verified, bot-accepted + * #1916 behavior. + * - `'fail-loud'` — NON-NEGOTIATED / pre-`testnet-canary` legacy requesters: + * NEVER byte-fit. A legacy requester reads a short page as EOF, so a + * byte-fitting prefix would silently drop the rest of the metadata AND split a + * seal (#1788 reintroduced). Emit whole subjects up to the row limit; if the + * resulting page still cannot be produced frame-safe — a single oversized + * subject OR the cumulative row-limit-bound page — throw + * {@link DurableMetaPageFrameError} rather than return a silent short page or + * an un-sendable over-frame page. Subject atomicity (#1788) holds identically + * on both branches; only this oversized-handling differs. + */ +type MetaOversizedSubjectPolicy = 'byte-fit' | 'fail-loud'; + +/** + * A non-negotiated (legacy) durable-meta request cannot be served frame-safe: + * the subject-atomic, row-limit-bound page — a single admitted subject alone or + * the cumulative page — exceeds the frame-safe response budget, and byte-fitting + * it would return a SHORT page a legacy requester reads as EOF (silent + * partial-metadata loss + a #1788 seal split). Failing loud is strictly better + * than silent loss here: a byte-budget-negotiating requester never hits this (it + * uses empty=EOF pagination), and the oversized `_meta` subject itself is only + * reachable via unverified peer-ingest, fixed at the root by #1921. + * + * Contract: this is a HARD, NON-RETRYABLE failure. The sync responder surfaces it + * as `outcome:'error'` and re-throws it (it is deliberately NOT wrapped in a + * retryable error, and NEVER converted to an empty — EOF — body), because a retry + * cannot make an oversized subject servable to a legacy requester; the only + * resolutions are a requester upgrade or the #1921 ingest fix. The budget it is + * checked against is `SYNC_BYTE_BUDGET_RESPONSE_BYTES` (the router read cap minus + * the frame headroom) — the largest response body guaranteed to fit one transport + * frame — so a page within it is always sendable and only a genuinely oversized + * one throws. + */ +export class DurableMetaPageFrameError extends Error { + readonly contextGraphId: string; + readonly bytes: number; + readonly limit: number; + + constructor(params: { contextGraphId: string; bytes: number; limit: number }) { + super( + `Durable-meta page for "${params.contextGraphId}" cannot be served frame-safe to a ` + + `non-negotiated (legacy) requester: a subject-atomic page of ${params.bytes} bytes ` + + `exceeds the ${params.limit}-byte frame budget. A byte-budget-negotiating requester ` + + `(testnet-canary+) paginates this via empty=EOF; upgrade the requester or reject the ` + + `oversized _meta subject at ingest (#1921).`, + ); + this.name = 'DurableMetaPageFrameError'; + this.contextGraphId = params.contextGraphId; + this.bytes = params.bytes; + this.limit = params.limit; + } +} + +/** + * Exclusive end index of the SUBJECT-ATOMIC, BYTE-FITTING durable-meta page that + * starts at `start` (#1916). Walks whole `(g, s)` subjects accumulating serialized + * wire bytes (the same accounting the byte-budget serializer uses), and: + * - a subject that would push the page over `maxBytes` is DEFERRED WHOLE to the + * next page (cut on the prior subject boundary) — never split; + * - ONLY when the FIRST subject alone exceeds `maxBytes` is it emitted (the + * serializer then byte-caps that one subject — provably not a valid seal at + * that size, so splitting it is safe and guarantees forward progress); + * - so "≤ budget" and "subject-atomic" hold together, unlike a plain row-prefix + * cut which splits a small seal that sits after large-literal rows. + * + * Also bounded by `rowLimit` (the requested page size): once whole subjects + * totalling ≥ `rowLimit` rows have been accumulated, stop at that subject + * boundary — so a normal small-subject page stays ~`rowLimit` rows (extended to + * complete the straddling subject) instead of ballooning to the full byte budget. + * + * Subjects are grouped by {@link metaSubjectKey} over `rows`, which is ONE query's + * result (a cached snapshot or a single store-paged window), so a blank-node + * subject's rows are self-consistently labelled and stay grouped (same + * single-query invariant as the #1788 fix). `trailingComplete` says whether the + * LAST subject in `rows` is fully present; when false that trailing subject is + * treated as incomplete and the function returns -1 ("need more rows") so the + * caller grows the window to complete it. + */ +function subjectAtomicBudgetEnd( + rows: readonly SyncRow[], + start: number, + maxBytes: number, + rowLimit: number, + trailingComplete: boolean, + oversizedPolicy: MetaOversizedSubjectPolicy = 'byte-fit', + contextGraphId = '', +): number { + const n = rows.length; + if (start >= n) return start; + const safeMax = Math.max(1, Math.floor(maxBytes)); + const safeRowLimit = Math.max(1, Math.floor(rowLimit)); + let bytes = 0; + let lastBoundary = start; + let i = start; + while (i < n) { + let j = i; + let runBytes = 0; + const key = metaSubjectKey(rows[i]); + while (j < n && metaSubjectKey(rows[j]) === key) { + runBytes += serializedResponderRowByteLength(rows[j]) + (i === start && j === i ? 0 : 1); + j += 1; + } + // The subject is fully present iff a later subject follows it in `rows`, or + // the caller says the tail is complete. + const complete = j < n || trailingComplete; + if (bytes + runBytes > safeMax) { + // This whole subject won't fit within the budget. `bytes + runBytes` is the + // frame size of every prior (whole) subject plus this one — the smallest + // page that keeps this subject atomic. + if (oversizedPolicy === 'fail-loud') { + // NON-NEGOTIATED legacy path: we cannot byte-fit (a short page reads as + // EOF → silent metadata loss + a #1788 split) and cannot over-fill the + // frame. Whether the FIRST subject alone (`lastBoundary === start`) or the + // CUMULATIVE page (`lastBoundary > start`) overflows, the page is + // unservable frame-safe — fail LOUD instead of returning a short page. + throw new DurableMetaPageFrameError({ + contextGraphId, + bytes: bytes + runBytes, + limit: safeMax, + }); + } + if (lastBoundary > start) return lastBoundary; // defer it WHOLE; keep prior subjects + // ESCAPE HATCH: the FIRST subject alone exceeds the budget. It is provably + // not a valid seal at that size, so split it — return a byte-FITTING row + // prefix (≥ 1 row for forward progress) so the page is ≤ budget under BOTH + // the byte-budget and the plain serializer. This is the ONLY place a + // subject is split, and only on the NEGOTIATED ('byte-fit') path. + let k = start; + let prefixBytes = 0; + while (k < j) { + const rowBytes = serializedResponderRowByteLength(rows[k]) + (k === start ? 0 : 1); + if (prefixBytes + rowBytes > safeMax && k > start) break; + prefixBytes += rowBytes; + k += 1; + } + return k; + } + if (!complete) { + // Trailing subject fits SO FAR but may continue beyond `rows`. It is not + // over budget, so it should be INCLUDED whole once fully read — signal + // "need more rows" so the caller grows the window to complete it (rather + // than deferring a subject that would fit). + return -1; + } + bytes += runBytes; + lastBoundary = j; + i = j; + // Requested page size reached (at a subject boundary): stop here rather than + // keep pulling whole subjects up to the full byte budget. + if (lastBoundary - start >= safeRowLimit) return lastBoundary; + } + return lastBoundary; +} + /** * Serialize the largest prefix that fits the negotiated response target. * Pagination advances by the number of N-Quads actually parsed by the @@ -570,12 +759,11 @@ export function serializeResponderRowsWithinByteBudget( maxBytes: number, ): string { const safeMaxBytes = Math.max(1, Math.floor(maxBytes)); - const encoder = new TextEncoder(); const page: string[] = []; let bytes = 0; for (const row of rows) { const serialized = serializeResponderRow(row); - const rowBytes = encoder.encode(serialized).byteLength + (page.length > 0 ? 1 : 0); + const rowBytes = serializedResponderRowByteLength(row) + (page.length > 0 ? 1 : 0); if (page.length > 0 && bytes + rowBytes > safeMaxBytes) break; page.push(serialized); bytes += rowBytes; @@ -800,7 +988,29 @@ export async function readDurableMetaPage(params: { refreshRowList?: boolean; refreshGeneration?: string; assetUals?: readonly string[]; + /** + * Serialized-response byte budget the store-paged subject extend must not + * blow (#1916): a single admitted subject larger than this cannot be a valid + * seal/descriptor (those are ~KB), so the extend stops growing and the page is + * byte-capped at serialization, splitting that oversized subject with forward + * progress. Defaults to {@link SYNC_BYTE_BUDGET_RESPONSE_BYTES}; injectable for + * tests. + */ + maxResponseBytes?: number; + /** + * Oversized-subject policy (#1916/#1923). NEGOTIATED (testnet-canary+) + * requesters use `'byte-fit'` — the verified byte-budget behavior. A + * NON-NEGOTIATED legacy requester uses `'fail-loud'` so an oversized `_meta` + * subject fails loudly ({@link DurableMetaPageFrameError}) instead of returning + * a short page the requester would read as EOF (silent metadata loss + a #1788 + * split). Defaults to the negotiated behavior; the production caller + * (sync-handler) always spells it out from the wire `pageMode`. Subject + * atomicity holds on both branches. See {@link MetaOversizedSubjectPolicy}. + */ + oversizedSubjectPolicy?: MetaOversizedSubjectPolicy; }): Promise { + const maxResponseBytes = params.maxResponseBytes ?? SYNC_BYTE_BUDGET_RESPONSE_BYTES; + const oversizedSubjectPolicy = params.oversizedSubjectPolicy ?? 'byte-fit'; if (params.assetUals !== undefined) { if (params.assetUals.length === 0) return []; const manifest = await readGraphScopedVmManifest( @@ -828,32 +1038,68 @@ export async function readDurableMetaPage(params: { refresh: params.refreshRowList, refreshGeneration: params.refreshGeneration, expiredMessage: 'Durable meta sync session snapshot expired before page completion', + // Durable meta is byte-budget-paginated on the requester (EOF is an empty + // page, not a short one), and the subject-atomic extend / byte-cap can + // legitimately serve a page shorter than `limit`. Never release the + // session on a short page — only on the empty EOF page — or a byte-capped + // page would drop the snapshot and strand the rest of the meta (#1916). + releaseOnShortPage: false, } : undefined; - return readResponderRowsPage( + const page = await readResponderRowsPage( cache, - (offset, limit, signal) => readDurableMetaRowsPage( + (offset, limit, signal) => readDurableMetaRowsPageSubjectAtomic( params.store, params.contextGraphId, params.registeredSubGraphNames, offset, limit, + maxResponseBytes, + oversizedSubjectPolicy, signal, ), params.offset, params.limit, params.signal, - cache - ? { - loadSnapshot: () => readBoundedDurableMetaSnapshot( - params.store, - params.contextGraphId, - params.registeredSubGraphNames, - cache, - ), - } - : undefined, + { + // Durable meta is the one lane whose rows carry graph-scoped seals; snap + // its page boundaries to `(g, s)` subject boundaries so a seal's control + // fields are never split across a sync round (#1788). The cached path + // extends via `subjectAtomic`; the store-paged loader above is itself + // subject-atomic. + subjectAtomic: true, + ...(cache + ? { + loadSnapshot: () => readBoundedDurableMetaSnapshot( + params.store, + params.contextGraphId, + params.registeredSubGraphNames, + cache, + ), + } + : {}), + }, ); + // Final SUBJECT-ATOMIC pass (#1916): both lanes return a page that ends on a + // `(g, s)` boundary (cached: in-memory extend; store-paged: the loader), so its + // trailing subject is complete. On the NEGOTIATED path this trims to whole + // subjects within the response byte budget — deferring any subject that would + // not fit to the next page — so a small seal AFTER large-literal rows is never + // cut by the byte cap (no-op for the already-byte-fit store-paged page; bites + // only when the cached extend produced a > budget page). On the NON-NEGOTIATED + // ('fail-loud') path it does NOT trim (that would be a short page a legacy + // requester reads as EOF); with an unbounded row limit it walks every subject + // and throws {@link DurableMetaPageFrameError} if the whole page exceeds the + // frame budget, otherwise returns the page unchanged. + return page.slice(0, subjectAtomicBudgetEnd( + page, + 0, + maxResponseBytes, + Number.MAX_SAFE_INTEGER, + true, + oversizedSubjectPolicy, + params.contextGraphId, + )); } /** Response byte budget for one changelog delta page — keeps a page under the @@ -1989,6 +2235,18 @@ interface ResponderRowsPageOptions { * (defaults to true; global budget pressure always propagates). */ fallbackOnPerSnapshotBudget?: boolean; + /** + * Extend a served page forward so it ends on a `(g, s)` subject boundary, + * never mid-subject (#1788). Only the durable-meta lane sets this: its rows + * carry graph-scoped seals whose control fields (`dkg:assertionVersion`) are + * admitted only as a complete in-batch subject. Safe because durable meta + * uses byte-budget pagination (requester page size 8192 > the 500 legacy + * cap), so the requester never treats an over-sized page as EOF — it advances + * by the actual row count and the next OFFSET lands on the next subject. This + * option governs only the cached (snapshot) path; the store-paged loader is + * made subject-atomic at its call site. See {@link metaSubjectKey}. + */ + subjectAtomic?: boolean; } async function readResponderRowsPage( @@ -2001,9 +2259,13 @@ async function readResponderRowsPage( ): Promise { const loadSnapshot = options?.loadSnapshot; const fallbackOnPerSnapshotBudget = options?.fallbackOnPerSnapshotBudget ?? true; + const subjectAtomic = options?.subjectAtomic ?? false; const safeOffset = Math.max(0, Math.floor(offset)); const safeLimit = Math.max(0, Math.floor(limit)); if (safeLimit === 0) return []; + // No session snapshot: the store-paged loader itself enforces subject + // atomicity where required (durable meta passes a subject-atomic loader), so + // this path needs no extra handling. if (!cache) return loadStoreBoundedPage(safeOffset, safeLimit, signal); try { return await readCachedRowsPage( @@ -2012,6 +2274,7 @@ async function readResponderRowsPage( safeOffset, safeLimit, signal, + subjectAtomic, ); } catch (error) { if (!isPerSnapshotBudgetError(error) || !fallbackOnPerSnapshotBudget) throw error; @@ -2025,6 +2288,7 @@ async function readCachedRowsPage( offset: number, limit: number, signal?: AbortSignal, + subjectAtomic = false, ): Promise { const safeOffset = Math.max(0, Math.floor(offset)); const safeLimit = Math.max(0, Math.floor(limit)); @@ -2041,7 +2305,26 @@ async function readCachedRowsPage( // `rows` is an immutable snapshot. Slice it directly so serving a 500-row // page allocates only that page's backing array, never a shallow copy of the // complete snapshot first. - const page = rows.slice(safeOffset, safeOffset + safeLimit); + let pageEnd = Math.min(safeOffset + safeLimit, rows.length); + // Subject-atomic durable-meta lane (#1788): if the requested window ends + // mid-subject, EXTEND it forward until the subject changes (or the snapshot + // ends) so a graph-scoped seal is never split across a page — and therefore + // never across a sync round. Extending (never trimming) keeps a non-final + // page `>= safeLimit`, so it never trips the requester's empty-page EOF nor + // the release-on-short-page path below. Reads only in-page indices plus a + // bounded one-subject lookahead — never the whole snapshot. `_meta` subjects + // are small (a seal is 14 quads; KA descriptors ~10; membership/activity rows + // bounded), so the extension is O(one subject) and negligible against the + // #1868 64k-row meta snapshot ceiling; a pathologically large subject is + // still emitted whole rather than truncated, with the transport frame limit + // as the final guard. + if (subjectAtomic && pageEnd > safeOffset && pageEnd < rows.length) { + const trailingKey = metaSubjectKey(rows[pageEnd - 1]); + while (pageEnd < rows.length && metaSubjectKey(rows[pageEnd]) === trailingKey) { + pageEnd += 1; + } + } + const page = rows.slice(safeOffset, pageEnd); if (page.length === 0 || (cache.releaseOnShortPage !== false && page.length < safeLimit)) { cache.memo.release(cache.key, { graceMs: COMPLETED_SYNC_RESPONDER_SESSION_GRACE_MS }); } @@ -3492,6 +3775,86 @@ async function readDurableMetaRowsPage( ); } +/** + * Subject-atomic wrapper over {@link readDurableMetaRowsPage} for the + * store-paged durable-meta lane (the no-session path and the oversized-snapshot + * fallback). Guarantees the returned page ENDS on a `(g, s)` subject boundary: + * a subject straddling `limit` is emitted in full, so a graph-scoped seal's + * rows (incl. `dkg:assertionVersion`) are never split across a page — and + * therefore never across a sync round (#1788). This mirrors the cached path's + * in-memory extend for the non-snapshot lane, and holds for ANY admitted + * subject term, not only IRIs. + * + * The trailing subject is completed by RE-READING a GROWING window from the + * same offset (`limit + extra`, `extra` doubling) until that subject is fully + * contained — a later subject appears in the window, or the store is exhausted + * — then cutting at the subject boundary. Each attempt is a SINGLE ordered + * query, which is what makes this correct for a blank-node subject: Oxigraph + * relabels a blank node per query (observed `_:x` → `_:`), so a + * subject-bound re-read or a multi-query paged loop could not re-identify one, + * but within a single self-contained window its label — and therefore the + * `(g, s)` boundary comparison — is consistent. Conforming writers only emit + * IRI `_meta` subjects (metadata generators build deterministic IRIs; the + * publisher rejects blank nodes), but a blank-node subject is reachable via the + * unverified system-CG peer-ingest path, so subject atomicity must cover it too; + * enforcing the IRI invariant at ingest (defense-in-depth) is tracked + * separately. + * + * `_meta` subjects are small (a seal is 14 quads ≈ a few KB), so this converges + * in about one read. The returned page is BOTH subject-atomic AND ≤ + * `maxResponseBytes` via {@link subjectAtomicBudgetEnd}: whole subjects are + * accumulated up to the budget and a subject that would not fit is deferred whole + * to the next page — so a small seal that sits AFTER large-literal rows is never + * cut by the byte cap (the #1916 hole). Only a FIRST subject that alone exceeds + * the budget is emitted and split by the serializer (provably not a valid seal at + * that size; safe and forward-progressing). The window grows only while its + * trailing subject is still incomplete AND the budget has not been reached, so + * memory stays bounded (~one budget) and the loop terminates at EOF. When + * `oversizedPolicy` is `'fail-loud'` (a non-negotiated legacy requester) the + * page is NOT byte-fit — an oversized or cumulative-over-budget page throws + * {@link DurableMetaPageFrameError} instead (see {@link MetaOversizedSubjectPolicy}). + */ +async function readDurableMetaRowsPageSubjectAtomic( + store: TripleStore, + contextGraphId: string, + registeredSubGraphNames: readonly string[], + offset: number, + limit: number, + maxResponseBytes: number, + oversizedPolicy: MetaOversizedSubjectPolicy, + signal?: AbortSignal, +): Promise { + const safeOffset = Math.max(0, Math.floor(offset)); + const safeLimit = Math.max(0, Math.floor(limit)); + if (safeLimit === 0) return []; + const safeMaxBytes = Math.max(1, Math.floor(maxResponseBytes)); + let extra = 1; + // Bounded by construction: `extra` at least doubles each attempt and the store + // is finite, so a short (exhausted) window always terminates the loop; the + // iteration cap is a defensive ceiling a real `_meta` graph never nears. + for (let attempt = 0; attempt < 48; attempt += 1) { + const window = await readDurableMetaRowsPage( + store, + contextGraphId, + registeredSubGraphNames, + safeOffset, + safeLimit + extra, + signal, + ); + const windowExhausted = window.length < safeLimit + extra; + const end = subjectAtomicBudgetEnd(window, 0, safeMaxBytes, safeLimit, windowExhausted, oversizedPolicy, contextGraphId); + // -1 ⇒ the first subject is still incomplete and fits so far: fetch more to + // decide whether it completes within budget or is oversized. Otherwise `end` + // is the subject-atomic, byte-fitting boundary. + if (end !== -1) return window.slice(0, end); + extra = extra === 1 ? safeLimit + 1 : extra * 2; + } + throw new Error( + `durable-meta subject-atomic paging did not converge for "${contextGraphId}" at offset ${safeOffset} ` + + '(first subject exceeded the bounded window budget)', + ); +} + /** * Serve only the immutable V2 descriptors for an explicitly requested KA set. * The caller intersects the request with the confirmed manifest first, so this diff --git a/packages/agent/src/sync/responder/sync-handler.ts b/packages/agent/src/sync/responder/sync-handler.ts index b0520f5ecf..24258a6641 100644 --- a/packages/agent/src/sync/responder/sync-handler.ts +++ b/packages/agent/src/sync/responder/sync-handler.ts @@ -25,6 +25,7 @@ import { createResponderSyncRowListMemo, createResponderSubGraphRegistrationMemo, createResponderSwmAdmissionMemo, + DurableMetaPageFrameError, readCatalogPage, readDurableDataPage, readDurableMetaPage, @@ -562,6 +563,13 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void { request.pageMode === SYNC_BYTE_BUDGET_PAGE_MODE && hintedPageRows > limit; const durableDataLimit = usesByteBudgetPage ? hintedPageRows : limit; + // Durable meta negotiated its byte-budget page mode on the wire (#1916 / + // #1923). The subject-atomic byte-fit in readDurableMetaPage already bounds + // the page ≤ budget for BOTH modes, so this only selects the belt-and- + // suspenders response serializer and records the explicit contract. + const usesMetaByteBudget = !isWorkspace && + phase === 'meta' && + request.pageMode === SYNC_BYTE_BUDGET_PAGE_MODE; if (!contextGraphId || typeof contextGraphId !== 'string') { // Count this early return too — it short-circuits before limiter.run, so // without this it would never reach the syncResponseTotal{ok}/{error} @@ -778,10 +786,35 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void { refreshRowList: session?.refreshRowList, refreshGeneration: session?.refreshGeneration, assetUals, + maxResponseBytes: SYNC_BYTE_BUDGET_RESPONSE_BYTES, + // NON-NEGOTIATED legacy requesters (no wire `pageMode`) must fail + // loud on an oversized `_meta` subject rather than receive a byte-fit + // SHORT page they would read as EOF — silent metadata loss + a #1788 + // seal split. Negotiated (testnet-canary+) requesters keep the + // verified byte-fit behavior (empty=EOF pagination, so short≠EOF). + oversizedSubjectPolicy: usesMetaByteBudget ? 'byte-fit' : 'fail-loud', }); const queryDurationMs = Date.now() - queryStartedAt; const serializeStartedAt = Date.now(); - const serialized = serializeResponderRows(rows); + // Byte-cap the durable-meta response (#1916) exactly like durable data: + // the subject-atomic extend can return a whole (or oversized) subject, + // so serialize within the frame budget rather than emitting unbounded + // N-Quads. The extend keeps every valid seal well under the budget, so + // this only ever truncates a pathological oversized subject. + // + // Pagination contract: durable meta uses byte-budget pagination where a + // SHORT page is NOT EOF — only an empty page is. The requester's + // short≠EOF handling is a requester-side default (page-fetch: + // syncPageSize=8192 > SYNC_PAGE_SIZE), and since #1923 it is ALSO + // negotiated on the wire via `pageMode` (usesMetaByteBudget). The + // subject-atomic byte-fit in readDurableMetaPage already bounds every + // page ≤ budget AND to whole subjects, so both the negotiated + // (byte-budget serializer) and the non-negotiated (plain serializer) + // branches are frame-safe and never split a subject; the gate here just + // honours the explicit contract. + const serialized = usesMetaByteBudget + ? serializeResponderRowsWithinByteBudget(rows, SYNC_BYTE_BUDGET_RESPONSE_BYTES) + : serializeResponderRows(rows); if (serialized) nquads.push(serialized); const serializeDurationMs = Date.now() - serializeStartedAt; logFirstPageDetail(() => `Sync responder durable meta for "${contextGraphId}": auth=${authDurationMs}ms query=${queryDurationMs}ms serialize=${serializeDurationMs}ms`); @@ -909,6 +942,21 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void { ); throw new QuietRetryableHandlerError(err.message); } + if (err instanceof DurableMetaPageFrameError) { + // Loud, non-retryable failure (#1788/#1916): an oversized `_meta` subject + // cannot be served frame-safe to a non-negotiated legacy requester, and + // byte-fitting it would be a silent short=EOF metadata loss. Retrying + // cannot help — surface it as a hard error so the round fails visibly + // rather than completing with partial metadata. Root fix: #1921. + getMetrics().syncResponseTotal.add(1, { outcome: 'error' }); + span.setAttribute('dkg.sync_response_outcome', 'error'); + logWarn( + createOperationContext('sync'), + `Sync responder cannot serve durable meta frame-safe to a non-negotiated (legacy) ` + + `requester for "${contextGraphId}" from peer ${peerId}: ${err.message}`, + ); + throw err; + } getMetrics().syncResponseTotal.add(1, { outcome: 'error' }); throw err; }); diff --git a/packages/agent/test/sync-byte-budget-pages.test.ts b/packages/agent/test/sync-byte-budget-pages.test.ts index b127572ae5..48adeae668 100644 --- a/packages/agent/test/sync-byte-budget-pages.test.ts +++ b/packages/agent/test/sync-byte-budget-pages.test.ts @@ -60,6 +60,65 @@ describe('byte-budget sync pagination', () => { expect(request.requesterSignatureR).toMatch(/^0x/); }); + // #1916: durable META now negotiates byte-budget paging exactly like durable + // DATA. These two cases pin the request-builder's meta advertisement directly: + // a regression dropping 'meta' from the useByteBudgetPage condition would + // silently break the wire negotiation, and the handler-level tests (which + // hand-craft the pageMode field) would not catch it. + it('advertises the byte-budget page mode for a durable meta request above the legacy cap', async () => { + const wallet = ethers.Wallet.createRandom(); + const signedLimits: number[] = []; + const encoded = await buildSyncRequestEnvelope({ + contextGraphId: CG_ID, + offset: 0, + limit: SYNC_REQUEST_PAGE_SIZE, + includeSharedMemory: false, + targetPeerId: REMOTE_PEER_ID, + requesterPeerId: LOCAL_PEER_ID, + phase: 'meta', + needsAuth: true, + computeSyncDigest: (_cg, _offset, limit) => { + signedLimits.push(limit); + return new Uint8Array(32); + }, + getIdentityId: async () => 0n, + claimedAgentAddress: wallet.address, + claimedAgentPrivateKey: wallet.privateKey, + }); + + const request = JSON.parse(new TextDecoder().decode(encoded)); + // The larger hint rides while the signed legacy limit stays 500-row capped, + // so digests remain wire-compatible with an old responder. + expect(signedLimits).toEqual([SYNC_PAGE_SIZE]); + expect(request.limit).toBe(SYNC_PAGE_SIZE); + expect(request.pageMode).toBe(SYNC_BYTE_BUDGET_PAGE_MODE); + expect(request.pageRowsHint).toBe(SYNC_REQUEST_PAGE_SIZE); + }); + + it('does not advertise byte-budget paging for a durable meta request at the legacy cap', async () => { + const wallet = ethers.Wallet.createRandom(); + const encoded = await buildSyncRequestEnvelope({ + contextGraphId: CG_ID, + offset: 0, + limit: SYNC_PAGE_SIZE, + includeSharedMemory: false, + targetPeerId: REMOTE_PEER_ID, + requesterPeerId: LOCAL_PEER_ID, + phase: 'meta', + needsAuth: true, + computeSyncDigest: () => new Uint8Array(32), + getIdentityId: async () => 0n, + claimedAgentAddress: wallet.address, + claimedAgentPrivateKey: wallet.privateKey, + }); + + const request = JSON.parse(new TextDecoder().decode(encoded)); + // At the 500-row cap there is no larger page to negotiate, so the responder + // must see an unmodified legacy meta request (no pageMode field). + expect(request.pageMode).toBeUndefined(); + expect(request.pageRowsHint).toBeUndefined(); + }); + it('continues after an old responder returns a short legacy page', async () => { const requested: Array<{ offset: number; limit: number }> = []; let sends = 0; diff --git a/packages/agent/test/sync-responder-concurrent-interleaving.test.ts b/packages/agent/test/sync-responder-concurrent-interleaving.test.ts index 7369020053..5fa880f165 100644 --- a/packages/agent/test/sync-responder-concurrent-interleaving.test.ts +++ b/packages/agent/test/sync-responder-concurrent-interleaving.test.ts @@ -10,6 +10,12 @@ import { createResponderGraphListMemo, createResponderSubGraphRegistrationMemo, } from '../src/sync/responder/graph-plan.js'; +import { + SYNC_BYTE_BUDGET_MAX_ROWS, + SYNC_BYTE_BUDGET_PAGE_MODE, + SYNC_BYTE_BUDGET_RESPONSE_BYTES, + SYNC_PAGE_SIZE, +} from '../src/dkg-agent-constants.js'; import { DKG_NS, lineGraphsFromNquads, @@ -447,10 +453,14 @@ describe('sync responder pagination interleaving', () => { const cgId = 'oversized-durable-meta'; const cgPrefix = `did:dkg:context-graph:${cgId}`; const metaGraph = `${cgPrefix}/_meta`; - // Rows keyed on the CG entity subject survive readDurableMetaRows filtering. + // Three DISTINCT admitted subjects (one row each), so the page boundary + // falls on a subject boundary and the fallback pages 2 + 1. Since #1788 a + // single subject is emitted atomically and would never split across pages, + // so distinct subjects are required to exercise paging here. Activity-prefix + // subjects survive readDurableMetaRows filtering. await store.insert(Array.from({ length: 3 }, (_, i) => ({ graph: metaGraph, - subject: cgPrefix, + subject: `did:dkg:activity:${cgId}-${i}`, predicate: `http://schema.org/p${i.toString().padStart(3, '0')}`, object: `"meta-${i.toString().padStart(3, '0')}"`, }))); @@ -479,6 +489,84 @@ describe('sync responder pagination interleaving', () => { expect(new Set(linesFromNquads(`${first}\n${second}`)).size).toBe(3); }); + // Handler-level (through registerSyncHandler): the durable-meta wire branch + // handles an oversized admitted subject DIFFERENTLY by negotiation, and both + // outcomes are frame-safe with no silent metadata loss (#1788/#1916): + // - NEGOTIATED (byte-budget pageMode): the subject-atomic byte-fit chunks the + // oversized subject under the frame and pages to completion (empty=EOF, so a + // short page is not EOF). A regression returning the whole >4 MiB subject + // would fail (pageCount 1 + over-budget bytes). + // - LEGACY (no pageMode): a legacy requester reads a short page as EOF, so + // byte-fitting would silently drop the rest of the subject; instead the + // responder FAILS LOUD. A regression byte-fitting it would fail (no throw). + const oversizedMetaStore = (cgId: string): { store: OxigraphStore; rows: Quad[]; subject: string } => { + const store = new OxigraphStore(); + const metaGraph = `did:dkg:context-graph:${cgId}/_meta`; + const subject = `did:dkg:activity:${cgId}-big`; + const bigLiteral = `"${'y'.repeat(60_000)}"`; // ~60 KB per row (under the 65535 literal cap) + const rows: Quad[] = Array.from({ length: 80 }, (_, i) => ({ + graph: metaGraph, + subject, + predicate: `${DKG_NS}p${String(i).padStart(3, '0')}`, + object: bigLiteral, + })); // ~4.8 MB total > the 4 MiB budget + return { store, rows, subject }; + }; + + it('durable-meta handler byte-caps an oversized subject under the frame and pages to completion — negotiated (byte-budget pageMode) (#1916)', async () => { + const cgId = 'oversized-meta-frame-neg'; + const { store, rows } = oversizedMetaStore(cgId); + await store.insert(rows); + const cap = registerTestSyncHandler(store, { syncPageSize: SYNC_PAGE_SIZE }); + const base = { + contextGraphId: cgId, + includeSharedMemory: false, + phase: 'meta' as const, + limit: SYNC_PAGE_SIZE, + syncSessionId: `${cgId}-session`, + pageMode: SYNC_BYTE_BUDGET_PAGE_MODE, + pageRowsHint: SYNC_BYTE_BUDGET_MAX_ROWS, + }; + + const enc = new TextEncoder(); + let offset = 0; + let delivered = 0; + let pageCount = 0; + for (let guard = 0; guard < 100; guard += 1) { + const resp = await cap.invoke({ ...base, offset }); + const n = resp === '' ? 0 : linesFromNquads(resp).length; + if (n === 0) break; + // Frame-safety: every response stays within the byte budget. + expect(enc.encode(resp).byteLength).toBeLessThanOrEqual(SYNC_BYTE_BUDGET_RESPONSE_BYTES); + pageCount += 1; + delivered += n; + offset += n; + } + // Chunked (byte cap engaged, not one oversized frame) and every row delivered. + expect(pageCount).toBeGreaterThan(1); + expect(delivered).toBe(rows.length); + await store.close(); + }); + + it('durable-meta handler FAILS LOUD on an oversized subject for a legacy (no pageMode) requester, never a silent short page (#1788)', async () => { + const cgId = 'oversized-meta-frame-legacy'; + const { store, rows } = oversizedMetaStore(cgId); + await store.insert(rows); + const cap = registerTestSyncHandler(store, { syncPageSize: SYNC_PAGE_SIZE }); + // No pageMode ⇒ non-negotiated legacy requester. Byte-fitting would return a + // short page it reads as EOF (silent loss + #1788 split); the responder must + // instead surface a hard, explicit failure — not a successful short response. + await expect(cap.invoke({ + contextGraphId: cgId, + includeSharedMemory: false, + phase: 'meta', + limit: SYNC_PAGE_SIZE, + offset: 0, + syncSessionId: `${cgId}-session`, + })).rejects.toThrow(/cannot be served frame-safe/); + await store.close(); + }); + it('falls back to store-bounded paging for an oversized shared-memory meta snapshot', async () => { const store = new OxigraphStore(); const cgId = 'oversized-swm-meta'; @@ -700,11 +788,15 @@ describe('sync responder pagination interleaving', () => { const cgPrefix = `did:dkg:context-graph:${cgId}`; const metaGraph = `${cgPrefix}/_meta`; const rows: Quad[] = []; + // 100 DISTINCT admitted subjects (one row each): since #1788 a single + // subject is emitted atomically, so a deep window into ONE subject is no + // longer meaningful — distinct subjects let the deep page address a subject + // boundary. Activity-prefix subjects survive durable-meta admission. for (let index = 0; index < 100; index++) { const padded = index.toString().padStart(3, '0'); rows.push({ graph: metaGraph, - subject: cgPrefix, + subject: `did:dkg:activity:m${padded}`, predicate: `http://schema.org/p${padded}`, object: `"meta-${padded}"`, }); @@ -717,9 +809,12 @@ describe('sync responder pagination interleaving', () => { }); await store.insert(rows); - // The durable-meta read is now store-bounded (subject-membership filter - // pushed into the store via EXISTS), so a deep page is a paged store query. - const probe = watchBoundedPageQuery(store, metaGraph, 90, 5); + // The durable-meta read is store-bounded (subject-membership filter pushed + // into the store via EXISTS), so a deep page is a paged store query. Durable + // meta reads `limit + 1` rows to detect a subject straddling the page + // boundary (#1788) and serves at most `limit` when the boundary is clean, so + // the store query's LIMIT is 6 here while the served page stays 5. + const probe = watchBoundedPageQuery(store, metaGraph, 90, 6); const cap = registerTestSyncHandler(store, { syncPageSize: 5 }); const out = await cap.invoke({ contextGraphId: cgId, diff --git a/packages/agent/test/sync-responder-durable-meta-subject-atomic.test.ts b/packages/agent/test/sync-responder-durable-meta-subject-atomic.test.ts new file mode 100644 index 0000000000..397e58bd82 --- /dev/null +++ b/packages/agent/test/sync-responder-durable-meta-subject-atomic.test.ts @@ -0,0 +1,650 @@ +import { describe, expect, it } from 'vitest'; +import { contextGraphMetaGraphUri } from '@origintrail-official/dkg-core'; +import { OxigraphStore, type Quad } from '@origintrail-official/dkg-storage'; +import { + DurableMetaPageFrameError, + readDurableMetaPage, + serializeResponderRowsWithinByteBudget, + type SyncRowListMemo, +} from '../src/sync/responder/graph-plan.js'; +import { DKG_NS } from './_helpers/sync-responder.js'; + +/** + * Regression for #1788: durable sync must never split a graph-scoped assertion + * seal's `_meta` rows across a page — and therefore across a sync round — or the + * batch-local control field `dkg:assertionVersion` is admitted alone and dropped + * (13/14) permanently on curated CGs. + * + * The responder now serves durable-meta pages that always END on a `(g, s)` + * subject boundary: a subject straddling the requested row limit is emitted in + * full (the page extends past `limit`). Both the cached (session snapshot) and + * the store-paged (no-session / oversized-fallback) loaders enforce this. Meta + * uses byte-budget pagination (requester page size 8192 > the 500 legacy cap), + * so an over-sized page is transparent to the cursor: the requester advances by + * the actual row count and the next OFFSET lands on the next subject. + */ + +type Row = { s: string; p: string; o: string; g: string }; + +const CG = 'cg-1788'; +const META = contextGraphMetaGraphUri(CG); + +// A graph-scoped author seal: one subject, 14 `_meta` quads, incl. the control +// field dkg:assertionVersion whose loss is the #1788 symptom. +const SEAL_SUBJECT = `did:dkg:context-graph:${CG}/assertion/0xabcdef0123456789abcdef0123456789abcdef01/report`; +const SEAL_PREDICATES = [ + 'assertionVersion', + 'merkleRoot', + 'kaUal', + 'assertionGraph', + 'contentScopeVersion', + 'publicTripleCount', + 'privateTripleCount', + 'privateMerkleRoot', + 'contextGraph', + 'batchId', + 'status', + 'reservedUal', + 'subGraphName', + 'rootEntity', +]; +const ASSERTION_VERSION = `${DKG_NS}assertionVersion`; + +function sealRows(subject: string, graph: string): Row[] { + return SEAL_PREDICATES.map((name, index) => ({ + s: subject, + p: `${DKG_NS}${name}`, + o: `"seal-${name}-${index}"`, + g: graph, + })); +} + +/** + * Drive a durable-meta phase exactly as the requester does: advance the offset + * by the number of rows actually served each page until an empty page (EOF). + */ +async function pageThrough( + fetchPage: (offset: number, limit: number) => Promise, + limit: number, +): Promise { + const pages: Row[][] = []; + let offset = 0; + // Bounded so a cursor bug (no forward progress) fails loudly instead of hanging. + for (let guard = 0; guard < 1_000; guard += 1) { + const page = await fetchPage(offset, limit); + if (page.length === 0) return pages; + pages.push([...page]); + offset += page.length; + } + throw new Error('durable-meta paging did not terminate — cursor made no forward progress'); +} + +function sealRowCount(page: readonly Row[]): number { + return page.filter((row) => row.s === SEAL_SUBJECT).length; +} + +function assertSubjectAtomicAndComplete(pages: Row[][]): void { + // 0-or-14 invariant: no page may carry a proper subset of the seal's rows. + for (const page of pages) { + const count = sealRowCount(page); + expect([0, SEAL_PREDICATES.length]).toContain(count); + } + // The whole seal materialises exactly once, including dkg:assertionVersion. + const sealRowsSeen = pages.flat().filter((row) => row.s === SEAL_SUBJECT); + expect(sealRowsSeen).toHaveLength(SEAL_PREDICATES.length); + expect(sealRowsSeen.some((row) => row.p === ASSERTION_VERSION)).toBe(true); + // No page is empty before EOF (the loop stops on the first empty page). + for (const page of pages) expect(page.length).toBeGreaterThan(0); +} + +function assertNoDuplicatesOrGaps(pages: Row[][]): void { + const keys = pages.flat().map((row) => `${row.g}\n${row.s}\n${row.p}\n${row.o}`); + // Cursor continuity: every served row appears exactly once across all pages + // (no duplicate from the extension, no gap from the offset advance). + expect(new Set(keys).size).toBe(keys.length); +} + +describe('durable-meta subject-atomic paging (#1788)', () => { + it('cached path: extends a page across the seal boundary, never splitting it', async () => { + const limit = 10; + const fillerA: Row[] = Array.from({ length: 5 }, (_, i) => ({ + s: `urn:fillerA:${i}`, p: `${DKG_NS}label`, o: `"a-${i}"`, g: META, + })); + const fillerB: Row[] = Array.from({ length: 5 }, (_, i) => ({ + s: `urn:fillerB:${i}`, p: `${DKG_NS}label`, o: `"b-${i}"`, g: META, + })); + // The 14-row seal starts at index 5, so a raw slice(0,10) would tear it + // 5/14 — exactly the #1788 split. + const snapshot: Row[] = [...fillerA, ...sealRows(SEAL_SUBJECT, META), ...fillerB]; + const memo: SyncRowListMemo = { + get: async () => snapshot, + release: () => {}, + }; + + const pages = await pageThrough( + (offset, pageLimit) => readDurableMetaPage({ + store: {} as OxigraphStore, + contextGraphId: CG, + registeredSubGraphNames: [], + offset, + limit: pageLimit, + rowListMemo: memo, + rowListCacheKey: 'durable-meta:1788:cached', + }), + limit, + ); + + assertSubjectAtomicAndComplete(pages); + assertNoDuplicatesOrGaps(pages); + + // Page 1 extends from index 0 through the whole seal (5 filler + 14 seal), + // and stops exactly at the subject boundary — it must NOT leak fillerB. + expect(pages[0]).toHaveLength(fillerA.length + SEAL_PREDICATES.length); + expect(pages[0].some((row) => row.s.startsWith('urn:fillerB:'))).toBe(false); + // Page 2 resumes exactly after the seal with the trailing filler. + expect(sealRowCount(pages[1])).toBe(0); + expect(pages[1].map((row) => row.s)).toEqual(fillerB.map((row) => row.s)); + // Every snapshot row is delivered exactly once across the two pages. + expect(pages.flat()).toHaveLength(snapshot.length); + }); + + it('store-paged path: completes a straddling subject from the store', async () => { + const limit = 10; + const store = new OxigraphStore(); + // Filler subjects sort before the straddling subject (activity < the + // seal-carrying `did:dkg:activity:seal-report`), each contributing one row; + // all are admitted by the `did:dkg:activity:` prefix so the real durable + // admission filter runs. The 14-row straddling subject carries a literal + // dkg:assertionVersion, so this exercises the store OFFSET/LIMIT window plus + // the complete-subject continuation query end to end. + const straddling = 'did:dkg:activity:seal-report'; + const filler: Quad[] = Array.from({ length: 5 }, (_, i) => ({ + graph: META, + subject: `did:dkg:activity:f0${i}`, + predicate: `${DKG_NS}label`, + object: `"f-${i}"`, + })); + const straddlingQuads: Quad[] = SEAL_PREDICATES.map((name, index) => ({ + graph: META, + subject: straddling, + predicate: `${DKG_NS}${name}`, + object: `"seal-${name}-${index}"`, + })); + await store.insert([...filler, ...straddlingQuads]); + + const pages = await pageThrough( + async (offset, pageLimit) => { + const page = await readDurableMetaPage({ + store, + contextGraphId: CG, + registeredSubGraphNames: [], + offset, + limit: pageLimit, + }); + return page.map((row) => ({ s: row.s, p: row.p, o: row.o, g: row.g })); + }, + limit, + ); + + const straddlingCount = (page: readonly Row[]) => + page.filter((row) => row.s === straddling).length; + // 0-or-14 invariant on the straddling subject across all pages. + for (const page of pages) expect([0, SEAL_PREDICATES.length]).toContain(straddlingCount(page)); + const straddlingRows = pages.flat().filter((row) => row.s === straddling); + expect(straddlingRows).toHaveLength(SEAL_PREDICATES.length); + expect(straddlingRows.some((row) => row.p === ASSERTION_VERSION)).toBe(true); + // Page 1 extends past the limit to complete the subject (5 filler + 14). + expect(pages[0]).toHaveLength(filler.length + SEAL_PREDICATES.length); + assertNoDuplicatesOrGaps(pages); + // Every admitted row delivered exactly once; the next offset lands on EOF. + expect(pages.flat()).toHaveLength(filler.length + SEAL_PREDICATES.length); + }); + + it('store-paged path: cuts before a LATER subject when the extended window is not exhausted', async () => { + // The "completes" case above resolves the straddle at EOF (window shorter + // than requested). This exercises the OTHER branch: the grown window is FULL + // and a later subject appears in it, so the cut must drop the later subject's + // partial rows exactly at the seal boundary — otherwise the page would emit a + // partial next subject (a #1788-class split of THAT subject). A later + // multi-row subject (`z…` sorts after `seal-report`) forces this branch. + const limit = 10; + const store = new OxigraphStore(); + const straddling = 'did:dkg:activity:seal-report'; + const later = 'did:dkg:activity:zeta-later'; + const filler: Quad[] = Array.from({ length: 5 }, (_, i) => ({ + graph: META, + subject: `did:dkg:activity:f0${i}`, + predicate: `${DKG_NS}label`, + object: `"f-${i}"`, + })); + const straddlingQuads: Quad[] = SEAL_PREDICATES.map((name, index) => ({ + graph: META, + subject: straddling, + predicate: `${DKG_NS}${name}`, + object: `"seal-${name}-${index}"`, + })); + const laterQuads: Quad[] = Array.from({ length: 8 }, (_, i) => ({ + graph: META, + subject: later, + predicate: `${DKG_NS}q${String(i).padStart(2, '0')}`, + object: `"later-${i}"`, + })); + await store.insert([...filler, ...straddlingQuads, ...laterQuads]); + + const pages = await pageThrough( + async (offset, pageLimit) => { + const page = await readDurableMetaPage({ + store, + contextGraphId: CG, + registeredSubGraphNames: [], + offset, + limit: pageLimit, + }); + return page.map((row) => ({ s: row.s, p: row.p, o: row.o, g: row.g })); + }, + limit, + ); + + const countFor = (subject: string) => (page: readonly Row[]) => + page.filter((row) => row.s === subject).length; + // Both subjects are 0-or-all per page (never a proper subset). + for (const page of pages) { + expect([0, SEAL_PREDICATES.length]).toContain(countFor(straddling)(page)); + expect([0, laterQuads.length]).toContain(countFor(later)(page)); + } + // Page 1 is exactly 5 filler + 14 seal and cuts BEFORE the later subject. + expect(pages[0]).toHaveLength(filler.length + SEAL_PREDICATES.length); + expect(countFor(later)(pages[0])).toBe(0); + // The later subject resumes intact on the next page. + expect(countFor(later)(pages[1])).toBe(laterQuads.length); + assertNoDuplicatesOrGaps(pages); + expect(pages.flat()).toHaveLength(filler.length + SEAL_PREDICATES.length + laterQuads.length); + }); + + it('store-paged path: a clean subject boundary at the limit is not extended', async () => { + const limit = 5; + const store = new OxigraphStore(); + // Five single-row activity subjects: the page boundary at limit=5 falls + // exactly on a subject boundary, so no extension and no continuation query. + const quads: Quad[] = Array.from({ length: 12 }, (_, i) => ({ + graph: META, + subject: `did:dkg:activity:s${String(i).padStart(2, '0')}`, + predicate: `${DKG_NS}label`, + object: `"row-${i}"`, + })); + await store.insert(quads); + + const pages = await pageThrough( + async (offset, pageLimit) => { + const page = await readDurableMetaPage({ + store, + contextGraphId: CG, + registeredSubGraphNames: [], + offset, + limit: pageLimit, + }); + return page.map((row) => ({ s: row.s, p: row.p, o: row.o, g: row.g })); + }, + limit, + ); + + // Pages of exactly 5, 5, 2 — never extended past a clean boundary. + expect(pages.map((page) => page.length)).toEqual([5, 5, 2]); + assertNoDuplicatesOrGaps(pages); + expect(pages.flat()).toHaveLength(12); + }); + + it('store-paged path: keeps a straddling blank-node subject atomic (unverified peer-ingest)', async () => { + // A blank-node `_meta` subject is not produced by conforming first-party + // writers (metadata generators emit IRIs; the publisher rejects blank nodes) + // but IS reachable via the unverified system-CG peer-ingest path. Oxigraph + // relabels a blank node per query, so a subject-bound re-read or a + // multi-query paged loop could not re-identify it — but the growing-window + // extend re-reads ONE query per attempt, within which the label (and thus + // the `(g, s)` boundary) is self-consistent, so the straddling subject is + // served WHOLE. 14 rows (incl. a dkg:assertionVersion control field — the + // row whose loss #1788 is about) > limit ⇒ straddles; admitted via + // dkg:memoryLayer != WorkingMemory; a few IRI fillers prove the boundary cut. + const limit = 10; + const store = new OxigraphStore(); + const BNODE = '_:peerSeal'; + const bnodeQuads: Quad[] = [ + { graph: META, subject: BNODE, predicate: `${DKG_NS}memoryLayer`, object: '"LongTermMemory"' }, + { graph: META, subject: BNODE, predicate: ASSERTION_VERSION, object: '"1"' }, + ...Array.from({ length: 12 }, (_, i) => ({ + graph: META, + subject: BNODE, + predicate: `${DKG_NS}p${String(i).padStart(2, '0')}`, + object: `"v-${i}"`, + })), + ]; + const filler: Quad[] = Array.from({ length: 3 }, (_, i) => ({ + graph: META, + subject: `did:dkg:activity:z${i}`, + predicate: `${DKG_NS}label`, + object: `"z-${i}"`, + })); + await store.insert([...bnodeQuads, ...filler]); + + // Round-based: pageThrough re-queries the store on each fetch, like + // successive sync rounds. The blank node may be relabelled per query, so + // identify it structurally. + const pages = await pageThrough( + async (offset, pageLimit) => { + const page = await readDurableMetaPage({ + store, + contextGraphId: CG, + registeredSubGraphNames: [], + offset, + limit: pageLimit, + }); + return page.map((row) => ({ s: row.s, p: row.p, o: row.o, g: row.g })); + }, + limit, + ); + + const bnodeRowCount = (page: readonly Row[]) => page.filter((row) => row.s.startsWith('_:')).length; + // 0-or-all invariant: no page carries a proper subset of the blank-node subject. + for (const page of pages) expect([0, bnodeQuads.length]).toContain(bnodeRowCount(page)); + const allBnode = pages.flat().filter((row) => row.s.startsWith('_:')); + expect(allBnode).toHaveLength(bnodeQuads.length); + expect(allBnode.some((row) => row.p === ASSERTION_VERSION)).toBe(true); + // No throw, subject-atomic, and every admitted row delivered exactly once. + assertNoDuplicatesOrGaps(pages); + expect(pages.flat()).toHaveLength(bnodeQuads.length + filler.length); + }); + + it('store-paged path: bounds an oversized admitted subject under the byte budget with forward progress (#1916)', async () => { + // A single admitted subject far larger than the response budget (hostile + // peer-ingest shape). The subject-atomic extend must NOT serve it whole into + // an un-sendable oversized frame; it bounds the window and the byte-budget + // serialization caps the response, splitting the subject across pages with + // forward progress. Uses a SMALL injected budget so the case is exercised + // without inserting a 4 MiB subject. + const limit = 5; + const budget = 1_200; // bytes — small injected response budget + const store = new OxigraphStore(); + const hostile = 'did:dkg:activity:oversized'; + const quads: Quad[] = Array.from({ length: 60 }, (_, i) => ({ + graph: META, + subject: hostile, + predicate: `${DKG_NS}p${String(i).padStart(3, '0')}`, + object: `"junk-${i}"`, + })); + await store.insert(quads); + + // Drive it like the responder+requester: serve a page, byte-cap it exactly + // as the handler does, and advance by the rows that actually crossed. + const enc = new TextEncoder(); + const pages: Row[][] = []; + let offset = 0; + for (let guard = 0; guard < 1_000; guard += 1) { + const page = await readDurableMetaPage({ + store, + contextGraphId: CG, + registeredSubGraphNames: [], + offset, + limit, + maxResponseBytes: budget, + }); + // Memory bound: the extend stops growing the fetched window at the byte + // budget instead of pulling the whole oversized subject into memory. + if (guard === 0) expect(page.length).toBeLessThan(quads.length); + const serialized = serializeResponderRowsWithinByteBudget(page, budget); + // Frame-safety: the serialized response never exceeds the budget. + expect(enc.encode(serialized).byteLength).toBeLessThanOrEqual(budget); + const rowCount = serialized === '' ? 0 : serialized.split('\n').length; + if (rowCount === 0) break; + pages.push(page.slice(0, rowCount).map((row) => ({ s: row.s, p: row.p, o: row.o, g: row.g }))); + offset += rowCount; + } + // Forward progress: it was chunked (not one oversized page), terminated + // (no loop), and delivered every row exactly once. + expect(pages.length).toBeGreaterThan(1); + const all = pages.flat(); + expect(all).toHaveLength(quads.length); + expect(new Set(all.map((row) => `${row.p}\n${row.o}`)).size).toBe(quads.length); + }); + + it('store-paged path: a realistically-large-but-VALID seal stays whole under the byte budget (#1916)', async () => { + // The Lock-2 guard: a legitimate control envelope — even with max normal + // literal sizes — is far under the budget, so it is served WHOLE, never + // split. Budget sits well above the seal (~KB) and well below the frame. + // limit=10 so the 14-row seal (after 5 filler) STRADDLES and the extend runs. + const limit = 10; + const budget = 200_000; // bytes — >> a valid seal, << the ~10 MiB frame + const store = new OxigraphStore(); + const seal = 'did:dkg:activity:seal-large'; + const bigLiteral = `"${'x'.repeat(400)}"`; // a realistically large seal literal + const filler: Quad[] = Array.from({ length: 5 }, (_, i) => ({ + graph: META, + subject: `did:dkg:activity:f0${i}`, + predicate: `${DKG_NS}label`, + object: `"f-${i}"`, + })); + const sealQuads: Quad[] = SEAL_PREDICATES.map((name, i) => ({ + graph: META, + subject: seal, + predicate: `${DKG_NS}${name}`, + object: i === 0 ? bigLiteral : `"seal-${name}-${i}"`, + })); + await store.insert([...filler, ...sealQuads]); + + const page = await readDurableMetaPage({ + store, + contextGraphId: CG, + registeredSubGraphNames: [], + offset: 0, + limit, + maxResponseBytes: budget, + }); + const sealRowsInPage = page.filter((row) => row.s === seal); + // The whole 14-row seal is in this one page (extend did not split it)… + expect(sealRowsInPage).toHaveLength(SEAL_PREDICATES.length); + // …and the byte-budget serialization does not truncate it either. + const serialized = serializeResponderRowsWithinByteBudget(page, budget); + expect(new TextEncoder().encode(serialized).byteLength).toBeLessThanOrEqual(budget); + for (const q of sealQuads) { + expect(serialized).toContain(`<${q.predicate}>`); + } + // Boundary is clean: 5 filler + 14 seal, nothing beyond the seal. + expect(page).toHaveLength(filler.length + SEAL_PREDICATES.length); + }); + + it('store-paged path: large-literal rows BEFORE a valid seal never split it (byte-cap is subject-aware) (#1916)', async () => { + // The round-6 hole: a ROW-prefix byte cap would emit the large fillers + only + // the FIRST few seal rows once the budget is hit — splitting the seal across + // rounds (#1788 reintroduced). Subject-aware byte-fitting must instead DEFER + // the whole seal to the next page. Fillers (~2.5 KB) + seal (~0.7 KB) exceed + // a 3 KB budget, and limit is large so the byte budget — not the row limit — + // is what cuts. + const limit = 100; + const budget = 2_800; // bytes + const store = new OxigraphStore(); + // One filler subject (sorts BEFORE the seal, 'a' < 'z') whose single ~2.5 KB + // row nearly fills the budget, leaving room for only a FEW seal rows — so the + // 14-row seal STRADDLES the budget boundary. A row-prefix cut would emit the + // filler + the first few seal rows (partial seal); subject-aware fitting must + // defer the whole seal instead. + const filler: Quad[] = [{ + graph: META, + subject: 'did:dkg:activity:a-fill', + predicate: `${DKG_NS}label`, + object: `"${'x'.repeat(2_450)}"`, + }]; + const seal = 'did:dkg:activity:z-seal'; + const sealQuads: Quad[] = SEAL_PREDICATES.map((name, i) => ({ + graph: META, + subject: seal, + predicate: `${DKG_NS}${name}`, + object: i === 1 ? '"1"' : `"s-${i}"`, // small seal literals; keep assertionVersion + })); + await store.insert([...filler, ...sealQuads]); + + const enc = new TextEncoder(); + const pages = await pageThrough( + async (offset, pageLimit) => { + const page = await readDurableMetaPage({ + store, + contextGraphId: CG, + registeredSubGraphNames: [], + offset, + limit: pageLimit, + maxResponseBytes: budget, + }); + // Each page is subject-atomic AND within budget, so the wire serializer + // emits it whole (no truncation) — frame-safe. + expect(enc.encode(serializeResponderRowsWithinByteBudget(page, budget)).byteLength) + .toBeLessThanOrEqual(budget); + return page.map((row) => ({ s: row.s, p: row.p, o: row.o, g: row.g })); + }, + limit, + ); + + const sealCount = (page: readonly Row[]) => page.filter((row) => row.s === seal).length; + // 0-or-all: the seal is NEVER partially emitted despite the large fillers + // consuming the budget ahead of it. + for (const page of pages) expect([0, SEAL_PREDICATES.length]).toContain(sealCount(page)); + const sealRows = pages.flat().filter((row) => row.s === seal); + expect(sealRows).toHaveLength(SEAL_PREDICATES.length); + expect(sealRows.some((row) => row.p === ASSERTION_VERSION)).toBe(true); + // Progress + completeness: every row delivered once, seal deferred to a later page. + assertNoDuplicatesOrGaps(pages); + expect(pages.flat()).toHaveLength(filler.length + SEAL_PREDICATES.length); + expect(pages.length).toBeGreaterThan(1); + }); + + it('cached path: a large-literal filler before a seal never splits the seal (byte-cap is subject-aware) (#1916)', async () => { + // The cached-trim counterpart to the store-paged large-filler test above: + // readDurableMetaPage's FINAL byte-fit runs on the snapshot-derived (cached) + // page too, so it must likewise DEFER a straddling seal WHOLE rather than emit + // a row-prefix that tears it. A row-prefix cut here would split the seal + // across rounds (#1788 reintroduced on the cached path). + const limit = 100; // large ⇒ the byte budget, not the row limit, does the cut + const budget = 2_800; // bytes + // One ~2.5 KB filler subject (sorts before the seal) nearly fills the budget, + // leaving room for only a few seal rows, so the 14-row seal straddles. + const filler: Row[] = [{ + s: 'urn:aaa-fill', p: `${DKG_NS}label`, o: `"${'x'.repeat(2_450)}"`, g: META, + }]; + const seal = 'urn:zzz-seal'; + const snapshot: Row[] = [...filler, ...sealRows(seal, META)]; + const memo: SyncRowListMemo = { get: async () => snapshot, release: () => {} }; + + const enc = new TextEncoder(); + const pages = await pageThrough( + async (offset, pageLimit) => { + const page = await readDurableMetaPage({ + store: {} as OxigraphStore, + contextGraphId: CG, + registeredSubGraphNames: [], + offset, + limit: pageLimit, + rowListMemo: memo, + rowListCacheKey: 'durable-meta:1916:cached-filler', + maxResponseBytes: budget, + }); + // Each cached page is subject-atomic AND ≤ budget (frame-safe). + expect(enc.encode(serializeResponderRowsWithinByteBudget(page, budget)).byteLength) + .toBeLessThanOrEqual(budget); + return page.map((row) => ({ s: row.s, p: row.p, o: row.o, g: row.g })); + }, + limit, + ); + + const sealCount = (page: readonly Row[]) => page.filter((row) => row.s === seal).length; + // 0-or-all: the seal is never partially emitted despite the filler ahead of it. + for (const page of pages) expect([0, SEAL_PREDICATES.length]).toContain(sealCount(page)); + const sealRowsSeen = pages.flat().filter((row) => row.s === seal); + expect(sealRowsSeen).toHaveLength(SEAL_PREDICATES.length); + expect(sealRowsSeen.some((row) => row.p === ASSERTION_VERSION)).toBe(true); + assertNoDuplicatesOrGaps(pages); + expect(pages.flat()).toHaveLength(snapshot.length); + // The filler alone filled page 1; the seal was deferred whole to a later page. + expect(pages.length).toBeGreaterThan(1); + expect(sealCount(pages[0])).toBe(0); + }); + + it('legacy (non-negotiated) path: normal meta paginates in row-limit subject-atomic pages, short only at EOF (#1788)', async () => { + // A pre-testnet-canary requester (oversizedSubjectPolicy='fail-loud') never + // receives a byte-fit short page: normal-sized meta paginates by the row + // limit, extended to a subject boundary, and is short only on the final EOF + // page — the legacy short=EOF contract stays correct. No throw. + const limit = 5; + const store = new OxigraphStore(); + const quads: Quad[] = Array.from({ length: 12 }, (_, i) => ({ + graph: META, + subject: `did:dkg:activity:s${String(i).padStart(2, '0')}`, + predicate: `${DKG_NS}label`, + object: `"row-${i}"`, + })); + await store.insert(quads); + + const pages = await pageThrough( + async (offset, pageLimit) => { + const page = await readDurableMetaPage({ + store, + contextGraphId: CG, + registeredSubGraphNames: [], + offset, + limit: pageLimit, + oversizedSubjectPolicy: 'fail-loud', + }); + return page.map((row) => ({ s: row.s, p: row.p, o: row.o, g: row.g })); + }, + limit, + ); + + // Row-limit-bound pages of exactly 5, 5, 2 — no short non-EOF page, no throw. + expect(pages.map((page) => page.length)).toEqual([5, 5, 2]); + assertNoDuplicatesOrGaps(pages); + expect(pages.flat()).toHaveLength(12); + await store.close(); + }); + + it('legacy (non-negotiated) path: an oversized _meta subject fails LOUD, never a silent short page (#1788)', async () => { + // fail-loud must NOT byte-fit an oversized subject into a short page a legacy + // requester reads as EOF (silent partial-metadata loss + #1788 split). Two + // shapes throw DurableMetaPageFrameError: (a) a single subject alone over + // budget, and (b) a cumulative row-limit-bound page over budget where each + // subject individually fits. + // (a) single oversized subject + const singleStore = new OxigraphStore(); + const hostile = 'did:dkg:activity:oversized'; + await singleStore.insert(Array.from({ length: 60 }, (_, i) => ({ + graph: META, + subject: hostile, + predicate: `${DKG_NS}p${String(i).padStart(3, '0')}`, + object: `"junk-${i}"`, + }))); + await expect(readDurableMetaPage({ + store: singleStore, + contextGraphId: CG, + registeredSubGraphNames: [], + offset: 0, + limit: 5, + maxResponseBytes: 1_200, + oversizedSubjectPolicy: 'fail-loud', + })).rejects.toBeInstanceOf(DurableMetaPageFrameError); + await singleStore.close(); + + // (b) cumulative: two ~2.1 KB single-row subjects each fit a 3 KB budget + // alone, but together exceed it. A legacy page cannot include both frame-safe + // and cannot drop one (short=EOF loss), so it fails loud. + const cumulativeStore = new OxigraphStore(); + await cumulativeStore.insert([ + { graph: META, subject: 'did:dkg:activity:a-big', predicate: `${DKG_NS}label`, object: `"${'x'.repeat(2_000)}"` }, + { graph: META, subject: 'did:dkg:activity:b-big', predicate: `${DKG_NS}label`, object: `"${'y'.repeat(2_000)}"` }, + ]); + await expect(readDurableMetaPage({ + store: cumulativeStore, + contextGraphId: CG, + registeredSubGraphNames: [], + offset: 0, + limit: 5, + maxResponseBytes: 3_000, + oversizedSubjectPolicy: 'fail-loud', + })).rejects.toBeInstanceOf(DurableMetaPageFrameError); + await cumulativeStore.close(); + }); +}); diff --git a/packages/agent/test/sync-responder-snapshot-cache.test.ts b/packages/agent/test/sync-responder-snapshot-cache.test.ts index 25717737f6..d0caf4d1b3 100644 --- a/packages/agent/test/sync-responder-snapshot-cache.test.ts +++ b/packages/agent/test/sync-responder-snapshot-cache.test.ts @@ -803,7 +803,7 @@ describe('sync responder snapshot cache and budget', () => { expect(budget.stats()).toMatchObject({ snapshots: 0, rows: 0, bytesEstimate: 0 }); }); - it('extracts a cached page without iterating or copying the complete snapshot', async () => { + it('extracts a cached page with only a bounded subject-boundary lookahead', async () => { const source = Array.from({ length: 2_000 }, (_, index) => ({ s: `urn:row:${index}`, p: `${DKG_NS}label`, @@ -813,23 +813,25 @@ describe('sync responder snapshot cache and budget', () => { const pageStart = 1_000; const pageEnd = 1_500; // offset 1000 + limit 500 let wholeSnapshotIterations = 0; - let outOfPageIndexReads = 0; + // Durable meta is subject-atomic (#1788): the reader peeks the single row at + // the page boundary to detect whether a subject straddles it. That is a + // bounded one-subject-horizon lookahead, NOT a full-array copy. Reading any + // index BEFORE the page (an indexed clone / `rows.slice()` of the whole + // array) or iterating the whole snapshot remains a HARD failure, so a + // full-scan regression still fails this test. + let boundaryLookAheadReads = 0; const snapshot = new Proxy(source, { get(target, property, receiver) { if (property === Symbol.iterator) { wholeSnapshotIterations += 1; throw new Error('complete snapshot must not be iterated while extracting one page'); } - // A correct slice(offset, offset+limit) reads only in-page element - // indices (plus non-index props like `length`). Reading any index - // outside the page is a full-array copy (e.g. `rows.slice()` then slice - // again, or an indexed clone), which the iterator guard alone misses. if (typeof property === 'string' && /^\d+$/.test(property)) { const index = Number(property); - if (index < pageStart || index >= pageEnd) { - outOfPageIndexReads += 1; - throw new Error(`page extraction must not read out-of-page index ${index}`); + if (index < pageStart) { + throw new Error(`page extraction must not read pre-page index ${index} (full-array copy)`); } + if (index >= pageEnd) boundaryLookAheadReads += 1; } return Reflect.get(target, property, receiver); }, @@ -849,9 +851,13 @@ describe('sync responder snapshot cache and budget', () => { rowListCacheKey: 'cached-page', }); + // Every source subject is distinct, so the boundary subject does not + // straddle: the page stays exactly one limit wide and the reader peeks the + // boundary row exactly once, never extending. expect(page).toHaveLength(500); expect(page[0]?.s).toBe('urn:row:1000'); + expect(page[499]?.s).toBe('urn:row:1499'); expect(wholeSnapshotIterations).toBe(0); - expect(outOfPageIndexReads).toBe(0); + expect(boundaryLookAheadReads).toBe(1); }); }); diff --git a/packages/agent/vitest.unit.config.ts b/packages/agent/vitest.unit.config.ts index 90026c4481..d4b5d591ec 100644 --- a/packages/agent/vitest.unit.config.ts +++ b/packages/agent/vitest.unit.config.ts @@ -55,6 +55,7 @@ export default defineConfig({ "test/durable-meta-admission.test.ts", "test/sync-responder-log-volume.test.ts", "test/sync-responder-snapshot-cache.test.ts", + "test/sync-responder-durable-meta-subject-atomic.test.ts", "test/sync-responder-cursor.test.ts", "test/sync-responder-oversized-fallback.test.ts", "test/sync-responder-swm-meta-ceiling.test.ts",