From b7ba6d2bb71f7e2fc5962f3fd425186fbe777336 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Mon, 20 Jul 2026 18:01:12 +0200 Subject: [PATCH 1/4] fix(sync): serve TTL-filtered SWM meta past the 64,000-row snapshot ceiling (#1847) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A CG whose SWM `_meta` reached SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS (64,000) raw rows became permanently unsyncable on the meta lane: `readSwmMetaPage` passed `params.cutoffIso == null` POSITIONALLY as `fallbackOnPerSnapshotBudget`, so TTL-filtered sessions (the normal modern path) got a bounded refusal with no fallback, and the bounded snapshot applied its row/byte budget to the RAW graph BEFORE the TTL filter, so the refusal fired even when the fresh subset was tiny. Live on mainnet: 10/15 cores refusing 78-202x/day; the fifa-world-cup-2026 CG (4,600 data quads, 64,001+ meta rows) never converges (`data=0 sharedMemory=0` forever while query-remote works). THE TRAP: a naive flag-flip is NOT a fix. The fallback was disabled deliberately because the TTL-filtered paged query was `SELECT DISTINCT ?g ?s ?p ?o` + a six-predicate UNION join + global `ORDER BY ?g ?s ?p ?o` re-evaluated with a growing OFFSET per page over a mutable graph family — the #1597 listGraphs-storm class that can pin cores and gigabytes on large stores. Re-enabling the flag alone would trade a bounded refusal for a store-melter. The fix mirrors how the SWM DATA lane solved the same problem (buildFreshSwmDataGraphPlan): * buildFreshSwmMetaPlan: two small-result discovery queries per meta graph (fresh subjects by publishedAt; graph-scoped heads via the fresh-WorkspaceOperation tuple join) plus chunked VALUES row counts. The session plan caches only graph/subject/count scalars. * readBoundedFreshSwmMetaSnapshot: the snapshot now materializes only the ADMITTED rows, so the per-snapshot budget binds on what is actually served — the fifa class (64k history, small fresh subset) takes the ordinary memoized-snapshot path. Final admission still runs through filterSwmMetaSnapshotRows, the canonical in-process filter. * readFreshSwmMetaRowsPageFromPlan: if even the ADMITTED set exceeds the budget, the session degrades to whole-subject window pages walked over the plan's prefix sums — VALUES-anchored reads with NO store-side ORDER BY and NO OFFSET (rows are sorted in-process; plan subject order is compareCodePoint, identical to compareRows on ?s). Chunk row counts are verified against the plan so a mutated subject fails the session (requester restarts) instead of skipping/duplicating rows, and a seal/head row-group is always read atomically within one chunk query (never torn the way #1788 durable batching tears groups). * The store-melting TTL query is DELETED (not gated), the dead readSwmMetaRows helper is removed, and only then is fallbackOnPerSnapshotBudget enabled for TTL sessions. * Legacy cutoff-less sessions keep the existing raw-snapshot + unfiltered store-paged compatibility path, byte-for-byte. Remaining bounded refusal: a single SUBJECT above the hard 64,000-row build cap (a coherent row-group that cannot fit any budget) — pinned by test as the only refusal left, and impossible to hit through organic operation history. Mutation-tested (each reverted before commit; each killed exactly the right tests): * M1 reintroduce `params.cutoffIso == null` positional arg -> the oversized-fresh-set, paged-equivalence, plan-mutation and paged requester-reassembly tests fail (4/8); the fifa-shape test still passes, proving budget-on-filtered-set independently fixes that class. * M2 disable the chunk row-count verification -> exactly the plan-mutation session test fails. * M3 remove the single-subject cap -> exactly the pathological-subject refusal test fails. * M4 off-by-one in the window slice -> all four paged-lane correctness tests fail. Evidence (in-memory Oxigraph, default production budgets): * fifa shape: 64,026 raw rows, 26 fresh -> served in 4 pages, 22ms. * intrinsically oversized fresh set: 65,000 admitted rows -> served completely in 14 bounded pages, 774ms, no refusal. * requester fetchSyncPages reassembly across 4-row pages: every op and seal/head row-group complete, dkg:assertionVersion never stripped. sparql-scale-lint: 0 new blocking findings (the rewritten legacy query carries R2/R3 pragmas; the new plan queries are lint-clean by shape). Fixes #1847 Co-Authored-By: Claude Fable 5 --- .../agent/src/sync/responder/graph-plan.ts | 593 ++++++++++++++---- .../agent/src/sync/responder/sync-handler.ts | 6 + .../sync-responder-swm-meta-ceiling.test.ts | 519 +++++++++++++++ packages/agent/vitest.unit.config.ts | 1 + 4 files changed, 1009 insertions(+), 110 deletions(-) create mode 100644 packages/agent/test/sync-responder-swm-meta-ceiling.test.ts diff --git a/packages/agent/src/sync/responder/graph-plan.ts b/packages/agent/src/sync/responder/graph-plan.ts index eb708defa3..6e26e94c75 100644 --- a/packages/agent/src/sync/responder/graph-plan.ts +++ b/packages/agent/src/sync/responder/graph-plan.ts @@ -99,6 +99,37 @@ export interface FreshSwmDataGraphPlanMemo { ): Promise; } +interface FreshSwmMetaSubjectEntry { + subject: string; + rowCount: number; +} + +interface FreshSwmMetaGraphPlanEntry { + graph: string; + /** TTL-admitted subjects, compareCodePoint-sorted; row counts are exact at plan build. */ + subjects: readonly FreshSwmMetaSubjectEntry[]; + rowCount: number; +} + +/** + * Session plan for the TTL-filtered SWM meta phase (#1847). Holds only + * graph/subject/count scalars — never payload rows — so a 64,000+-row `_meta` + * graph costs the plan a few hundred kilobytes at most while the rows stay in + * the store until each page addresses its own bounded subject window. + */ +interface FreshSwmMetaPlan { + entries: readonly FreshSwmMetaGraphPlanEntry[]; + totalRows: number; +} + +export interface FreshSwmMetaPlanMemo { + get( + key: string, + load: () => Promise, + options?: { refresh?: boolean; requireExisting?: boolean; signal?: AbortSignal }, + ): Promise; +} + interface ExactGraphPagePlanEntry { graph: string; rowCount: number; @@ -250,8 +281,31 @@ export function createResponderFreshSwmDataGraphPlanMemo( ttlMs = 10 * 60_000, maxEntries = 32, ): FreshSwmDataGraphPlanMemo { - const cached = new Map(); - const inflight = new Map>(); + return createSessionPlanMemo(ttlMs, maxEntries); +} + +/** + * Session-scoped plan cache for the TTL-filtered SWM META phase (#1847). Same + * lifetime/refresh contract as {@link createResponderFreshSwmDataGraphPlanMemo}: + * touched on every page, offset>0 requires the existing plan so a rebuilt plan + * against a moved TTL cutoff can never make a numeric offset skip or duplicate. + */ +export function createResponderFreshSwmMetaPlanMemo( + ttlMs = 10 * 60_000, + maxEntries = 32, +): FreshSwmMetaPlanMemo { + return createSessionPlanMemo(ttlMs, maxEntries); +} + +function createSessionPlanMemo(ttlMs: number, maxEntries: number): { + get( + key: string, + load: () => Promise, + options?: { refresh?: boolean; requireExisting?: boolean; signal?: AbortSignal }, + ): Promise; +} { + const cached = new Map(); + const inflight = new Map>(); const prune = (now = Date.now()) => { for (const [key, entry] of cached) { if (now - entry.cachedAt >= ttlMs) cached.delete(key); @@ -295,38 +349,7 @@ export function createResponderExactGraphPagePlanMemo( ttlMs = 10 * 60_000, maxEntries = 32, ): ExactGraphPagePlanMemo { - const cached = new Map(); - const inflight = new Map>(); - const prune = (now = Date.now()) => { - for (const [key, entry] of cached) { - if (now - entry.cachedAt >= ttlMs) cached.delete(key); - } - }; - return { - async get(key, load, options) { - throwIfAborted(options?.signal); - const now = Date.now(); - prune(now); - const pending = inflight.get(key); - if (pending) return raceAgainstAbort(pending, options?.signal); - const existing = cached.get(key); - if (!options?.refresh && existing) { - cached.delete(key); - cached.set(key, { value: existing.value, cachedAt: now }); - return existing.value; - } - if (options?.requireExisting) return null; - if (!existing && cached.size >= maxEntries) cached.delete(cached.keys().next().value!); - const pendingLoad = load() - .then((value) => { - cached.set(key, { value, cachedAt: Date.now() }); - return value; - }) - .finally(() => inflight.delete(key)); - inflight.set(key, pendingLoad); - return raceAgainstAbort(pendingLoad, options?.signal); - }, - }; + return createSessionPlanMemo(ttlMs, maxEntries); } function createSubGraphNameMemo( @@ -465,6 +488,7 @@ export async function readSwmMetaPage(params: { rowListCacheKey?: string; refreshRowList?: boolean; refreshGeneration?: string; + freshMetaPlanMemo?: FreshSwmMetaPlanMemo; }): Promise { const graphs = swmGraphsForRegisteredSubGraphs(params.contextGraphId, params.registeredSubGraphNames, true); const graphSet = new Set(params.graphList); @@ -478,32 +502,109 @@ export async function readSwmMetaPage(params: { expiredMessage: 'Shared-memory meta sync session snapshot expired before page completion', } : undefined; - return readResponderRowsPage( - cache, - (offset, limit, signal) => readSwmMetaRowsPage( + + if (params.cutoffIso == null) { + // Legacy unfiltered sessions: unchanged bounded raw-graph snapshot with the + // existing store-paged compatibility fallback. + return readResponderRowsPage( + cache, + (offset, limit, signal) => readSwmMetaRowsPage( + params.store, + candidateGraphs, + offset, + limit, + signal, + ), + params.offset, + params.limit, + params.signal, + cache + ? () => readBoundedSwmMetaSnapshot( + params.store, + candidateGraphs, + cache, + ) + : undefined, + ); + } + + // #1847: the TTL-filtered lane. Two invariants shape it: + // + // 1. The old bounded snapshot loaded the RAW meta graph and applied the + // row/byte budget BEFORE the TTL filter, so a long-lived CG whose `_meta` + // crossed 64,000 raw rows was refused even when its fresh subset was a + // few hundred rows — and with the fallback gated off for TTL sessions the + // refusal was permanent (10/15 mainnet cores, fifa-world-cup-2026). + // 2. The old TTL fallback query (DISTINCT + UNION join + global + // `ORDER BY ?g ?s ?p ?o` + growing OFFSET over a mutable graph family) + // was gated off DELIBERATELY: it can pin cores and gigabytes on large + // stores (#1597 class). Re-enabling the flag alone would trade a bounded + // refusal for a store-melting query; that query is deleted, not revived. + // + // The fix mirrors buildFreshSwmDataGraphPlan: tiny discovery queries find the + // TTL-admitted subjects (small results, no payload sort), the session plan + // caches only graph/subject/count scalars, the snapshot materializes only the + // ADMITTED rows (so the budget now binds on what is actually served), and an + // intrinsically-oversized fresh set degrades to bounded whole-subject window + // pages from the same plan instead of failing permanently. + const cutoffIso = params.cutoffIso; + // Consume the explicit session refresh once: when the snapshot build crosses + // its budget, the immediate page-zero fallback must reuse the just-built plan + // instead of rebuilding (and re-counting) it against a moving store. + let planRefreshPending = params.refreshRowList === true; + const getPlan = async ( + pageOffset: number, + pageSignal: AbortSignal | undefined, + ): Promise => { + const loadPlan = () => buildFreshSwmMetaPlan( params.store, candidateGraphs, - params.cutoffIso, + cutoffIso, + pageSignal, + ); + const refreshPlan = pageOffset === 0 && planRefreshPending; + if (refreshPlan) planRefreshPending = false; + const plan = params.freshMetaPlanMemo && params.rowListCacheKey + ? await params.freshMetaPlanMemo.get(params.rowListCacheKey, loadPlan, { + refresh: refreshPlan, + requireExisting: pageOffset > 0, + signal: pageSignal, + }) + : await loadPlan(); + if (!plan) { + throw new Error('Shared-memory meta sync session graph plan expired before page completion'); + } + return plan; + }; + const loadStoreBoundedPage: StorePageLoader = async (offset, limit, signal) => + readFreshSwmMetaRowsPageFromPlan( + params.store, + await getPlan(offset, signal), offset, limit, + cache?.key ?? `swm-meta:${params.contextGraphId}`, signal, - ), + ); + return readResponderRowsPage( + cache, + loadStoreBoundedPage, params.offset, params.limit, params.signal, cache - ? () => readBoundedSwmMetaSnapshot( + ? async () => readBoundedFreshSwmMetaSnapshot( params.store, - candidateGraphs, - params.cutoffIso, + await getPlan(0, undefined), + cutoffIso, cache, ) : undefined, - // The TTL-filtered SPARQL fallback joins and globally sorts a mutable meta - // graph. On large stores that query is worse than a bounded refusal: it can - // consume multiple cores and gigabytes until the HTTP timeout. Unfiltered - // legacy sessions retain the existing store-paged compatibility path. - params.cutoffIso == null, + // The per-snapshot budget fallback MUST stay enabled here (#1847): it now + // degrades to the bounded plan-paged reader above, never to the deleted + // global-sort query. Passing `params.cutoffIso == null` in this position is + // the exact defect that made every 64,000-row `_meta` CG permanently + // unsyncable on mainnet. + true, ); } @@ -1952,38 +2053,15 @@ async function readRowsAcrossGraphsExcludingSubjectPrefix( .sort(compareRows); } -async function readSwmMetaRows( - store: TripleStore, - swmMetaGraphs: readonly string[], - cutoffIso: string | null, - signal?: AbortSignal, -): Promise { - const swmMetaValues = graphValues(swmMetaGraphs); - if (!swmMetaValues) return []; - const res = await store.query(` - SELECT DISTINCT ?g ?s ?p ?o WHERE { - VALUES ?g { ${swmMetaValues} } - GRAPH ?g { - ?s ?p ?o . - ${cutoffIso - ? ` - ?s <${DKG_PUBLISHED_AT}> ?ts . - FILTER(?ts >= ${sparqlString(cutoffIso)}^^)` - : ''} - } - } - `, syncResponderStoreOptions(signal, 'sync.responder.readSwmMetaRows')); - if (res.type !== 'bindings') return []; - return res.bindings - .map((row) => ({ s: row['s'], p: row['p'], o: row['o'], g: row['g'] })) - .filter((row) => row.s && row.p && row.o && row.g) - .sort(compareRows); -} - +/** + * Legacy (cutoffIso == null) bounded snapshot: reads the raw candidate meta + * graphs under the per-snapshot budget. TTL-filtered sessions use + * {@link readBoundedFreshSwmMetaSnapshot}, whose budget binds on the admitted + * fresh subset instead of the raw graph size (#1847). + */ async function readBoundedSwmMetaSnapshot( store: TripleStore, swmMetaGraphs: readonly string[], - cutoffIso: string | null, cache: RowListCache, ): Promise { const limits = cache.memo.snapshotLoadLimits ?? { @@ -2064,7 +2142,7 @@ async function readBoundedSwmMetaSnapshot( } } - return filterSwmMetaSnapshotRows(rows, cutoffIso); + return filterSwmMetaSnapshotRows(rows, null); } function filterSwmMetaSnapshotRows( @@ -2125,10 +2203,17 @@ function filterSwmMetaSnapshotRows( return rows.filter((row) => admitted.has(row.s)).sort(compareRows); } +/** + * Legacy UNFILTERED store-paged compatibility path (cutoffIso == null sessions + * only). The former TTL variant of this query — DISTINCT + a six-predicate + * UNION join + global `ORDER BY ?g ?s ?p ?o` re-evaluated with a growing + * OFFSET per page over a mutable graph family — was the #1847 store-melter and + * is deliberately DELETED, not gated: TTL-filtered sessions page from the + * session plan via {@link readFreshSwmMetaRowsPageFromPlan} instead. + */ async function readSwmMetaRowsPage( store: TripleStore, swmMetaGraphs: readonly string[], - cutoffIso: string | null, offset: number, limit: number, signal?: AbortSignal, @@ -2137,40 +2222,15 @@ async function readSwmMetaRowsPage( const safeLimit = Math.max(0, Math.floor(limit)); if (safeLimit === 0) return []; const swmMetaValues = graphValues(swmMetaGraphs); - const swmMetaClause = swmMetaValues - ? ` - VALUES ?g { ${swmMetaValues} } - GRAPH ?g { - ?s ?p ?o . - ${cutoffIso - ? ` - { - ?s <${DKG_PUBLISHED_AT}> ?ts . - } UNION { - # Graph-scoped SWM heads are current-state pointers and therefore - # intentionally have no independent publishedAt row. Bind them to - # the timestamped WorkspaceOperation they select so TTL recovery - # receives the head plus its immutable commitment atomically. - ?s <${DKG_CONTENT_SCOPE_VERSION}> ${GRAPH_KA_CONTENT_SCOPE_VERSION} ; - <${DKG_KA_UAL}> ?headUal ; - <${DKG_ASSERTION_VERSION}> ?headVersion ; - <${DKG_SHARE_OPERATION_ID}> ?shareId . - ?headOperation <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_WORKSPACE_OPERATION}> ; - <${DKG_CONTENT_SCOPE_VERSION}> ${GRAPH_KA_CONTENT_SCOPE_VERSION} ; - <${DKG_KA_UAL}> ?headUal ; - <${DKG_ASSERTION_VERSION}> ?headVersion ; - <${DKG_SHARE_OPERATION_ID}> ?shareId ; - <${DKG_PUBLISHED_AT}> ?ts . - } - FILTER(?ts >= ${sparqlString(cutoffIso)}^^)` - : ''} - } - ` - : ''; - if (!swmMetaClause) return []; + if (!swmMetaValues) return []; + // sparql-scan-allow: R2 -- ?g is bound by a finite VALUES list of pre-admitted SWM meta graph IRIs + // sparql-scan-allow: R3 -- pre-existing legacy (cutoff-less) compatibility lane, unchanged behavior; TTL sessions page from the session plan instead (#1847) const res = await store.query(` SELECT DISTINCT ?g ?s ?p ?o WHERE { - ${swmMetaClause} + VALUES ?g { ${swmMetaValues} } + GRAPH ?g { + ?s ?p ?o . + } } ORDER BY ?g ?s ?p ?o OFFSET ${safeOffset} @@ -2182,6 +2242,319 @@ async function readSwmMetaRowsPage( .filter((row) => row.s && row.p && row.o && row.g); } +const FRESH_SWM_META_PLAN_SUBJECT_CHUNK = 100; + +/** + * Discover the TTL-admitted subjects of one SWM meta graph with two + * small-result queries (no payload rows, no sort, no OFFSET): + * + * 1. subjects carrying their own fresh `publishedAt` — the + * {@link readFreshSwmRoots} shape, an indexed predicate probe whose result + * is the fresh subset, not the graph; + * 2. graph-scoped SWM heads. Heads are current-state pointers and + * intentionally have no independent publishedAt row; they are admitted via + * the timestamped WorkspaceOperation they select (same six-predicate join + * the TTL lane has always used), so TTL recovery receives the head plus its + * immutable commitment atomically. + * + * SWM meta subjects are IRIs by contract (workspace writers skolemize blank + * nodes before storage); non-IRI subjects cannot appear in a VALUES clause and + * are skipped. + */ +async function readFreshSwmMetaSubjects( + store: TripleStore, + graph: string, + cutoffIso: string, + signal?: AbortSignal, +): Promise> { + const cutoffFilter = + `FILTER(?ts >= ${sparqlString(cutoffIso)}^^)`; + const subjects = new Set(); + const freshRes = await store.query(` + SELECT DISTINCT ?s WHERE { + GRAPH <${assertSafeIri(graph)}> { + ?s <${DKG_PUBLISHED_AT}> ?ts . + ${cutoffFilter} + } + } + `, syncResponderStoreOptions(signal, 'sync.responder.readFreshSwmMetaSubjects')); + if (freshRes.type === 'bindings') { + for (const row of freshRes.bindings) { + const subject = row['s']; + if (subject && isIriTerm(subject)) subjects.add(subject); + } + } + const headRes = await store.query(` + SELECT DISTINCT ?s WHERE { + GRAPH <${assertSafeIri(graph)}> { + ?s <${DKG_CONTENT_SCOPE_VERSION}> ${GRAPH_KA_CONTENT_SCOPE_VERSION} ; + <${DKG_KA_UAL}> ?headUal ; + <${DKG_ASSERTION_VERSION}> ?headVersion ; + <${DKG_SHARE_OPERATION_ID}> ?shareId . + ?headOperation <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_WORKSPACE_OPERATION}> ; + <${DKG_CONTENT_SCOPE_VERSION}> ${GRAPH_KA_CONTENT_SCOPE_VERSION} ; + <${DKG_KA_UAL}> ?headUal ; + <${DKG_ASSERTION_VERSION}> ?headVersion ; + <${DKG_SHARE_OPERATION_ID}> ?shareId ; + <${DKG_PUBLISHED_AT}> ?ts . + ${cutoffFilter} + } + } + `, syncResponderStoreOptions(signal, 'sync.responder.readFreshSwmMetaHeadSubjects')); + if (headRes.type === 'bindings') { + for (const row of headRes.bindings) { + const subject = row['s']; + if (subject && isIriTerm(subject)) subjects.add(subject); + } + } + return subjects; +} + +function subjectValues(subjects: readonly string[]): string { + return subjects.map((subject) => `<${assertSafeIri(subject)}>`).join(' '); +} + +async function countFreshSwmMetaSubjectRows( + store: TripleStore, + graph: string, + subjects: readonly string[], + signal?: AbortSignal, +): Promise { + const countsBySubject = new Map(); + for (const chunk of chunkValues(subjects, FRESH_SWM_META_PLAN_SUBJECT_CHUNK)) { + const res = await store.query(` + SELECT ?s (COUNT(*) AS ?count) WHERE { + VALUES ?s { ${subjectValues(chunk)} } + GRAPH <${assertSafeIri(graph)}> { ?s ?p ?o } + } + GROUP BY ?s + `, syncResponderStoreOptions(signal, 'sync.responder.countFreshSwmMetaSubjectRows')); + if (res.type !== 'bindings') continue; + for (const row of res.bindings) { + const subject = row['s']; + if (subject) countsBySubject.set(subject, parseSparqlInteger(row['count'])); + } + } + return subjects + .map((subject) => ({ subject, rowCount: countsBySubject.get(subject) ?? 0 })) + .filter((entry) => entry.rowCount > 0); +} + +/** + * Build the tiny, stable pagination plan for a TTL-filtered SWM meta phase. + * Only graph/subject/count scalars are computed and cached; the payload rows + * stay in the store until a page (or the bounded snapshot) addresses its own + * subject window. Subjects are compareCodePoint-sorted so the plan's prefix + * sums agree with the compareRows order used when window rows are sorted + * in-process — no store-side ORDER BY or OFFSET is ever needed. + */ +async function buildFreshSwmMetaPlan( + store: TripleStore, + swmMetaGraphs: readonly string[], + cutoffIso: string, + signal?: AbortSignal, +): Promise { + const entries: FreshSwmMetaGraphPlanEntry[] = []; + for (const graph of dedupeStrings(swmMetaGraphs).sort(compareCodePoint)) { + throwIfAborted(signal); + const admitted = await readFreshSwmMetaSubjects(store, graph, cutoffIso, signal); + if (admitted.size === 0) continue; + const subjects = await countFreshSwmMetaSubjectRows( + store, + graph, + [...admitted].sort(compareCodePoint), + signal, + ); + if (subjects.length === 0) continue; + entries.push({ + graph, + subjects, + rowCount: subjects.reduce((sum, entry) => sum + entry.rowCount, 0), + }); + } + return { + entries, + totalRows: entries.reduce((sum, entry) => sum + entry.rowCount, 0), + }; +} + +/** + * Read ALL rows of a whole-subject window in bounded VALUES chunks. Each + * chunk's row total is verified against the plan counts: the plan's prefix + * sums are the pagination cursor, so a mutated subject must fail the session + * (the requester restarts with a fresh plan) rather than silently skip or + * duplicate rows — and a seal/head subject is always read atomically within + * one chunk query, so its row-group can never be torn by a chunk boundary. + */ +async function readFreshSwmMetaSubjectWindowRows( + store: TripleStore, + graph: string, + subjects: readonly FreshSwmMetaSubjectEntry[], + signal?: AbortSignal, +): Promise { + const rows: SyncRow[] = []; + for (const chunk of chunkValues(subjects, FRESH_SWM_META_PLAN_SUBJECT_CHUNK)) { + const expectedRows = chunk.reduce((sum, entry) => sum + entry.rowCount, 0); + const res = await store.query(` + SELECT ?s ?p ?o WHERE { + VALUES ?s { ${subjectValues(chunk.map((entry) => entry.subject))} } + GRAPH <${assertSafeIri(graph)}> { ?s ?p ?o } + } + `, { + ...syncResponderStoreOptions(signal, 'sync.responder.readFreshSwmMetaSubjectRows'), + maxResponseBytes: snapshotResponseByteLimit( + SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + ), + }); + let added = 0; + if (res.type === 'bindings') { + for (const row of res.bindings) { + const s = row['s']; + const p = row['p']; + const o = row['o']; + if (!s || !p || !o) continue; + rows.push({ s, p, o, g: graph }); + added += 1; + } + } + if (added !== expectedRows) { + throw new Error( + `Shared-memory meta sync plan changed while reading ${graph}: ` + + `expected ${expectedRows} rows for ${chunk.length} subjects, found ${added}`, + ); + } + } + return rows.sort(compareRows); +} + +/** + * Store-bounded page reader for an intrinsically-oversized TTL-filtered SWM + * meta phase. Pages advance across the plan's prefix sums; each page reads + * whole subjects (bounded by the page limit plus at most one subject's rows) + * and slices precisely. A single SUBJECT larger than the HARD build cap + * (SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS — deliberately the fixed constant, + * not the test/operator-shrinkable session budget, so a shrunken budget forces + * paged mode without refusing ordinary multi-row subjects) is the one + * remaining bounded refusal: it cannot be served as a coherent row-group + * within any budget, and unlike the graph-level cap it can only be a + * pathological writer, never organic operation history. + */ +async function readFreshSwmMetaRowsPageFromPlan( + store: TripleStore, + plan: FreshSwmMetaPlan, + offset: number, + limit: number, + budgetKey: string, + signal?: AbortSignal, +): Promise { + let skip = Math.max(0, Math.floor(offset)); + let remaining = Math.max(0, Math.floor(limit)); + if (remaining === 0 || skip >= plan.totalRows) return []; + const rows: SyncRow[] = []; + for (const entry of plan.entries) { + if (skip >= entry.rowCount) { + skip -= entry.rowCount; + continue; + } + // Select the whole-subject window covering [skip, skip + remaining). + const window: FreshSwmMetaSubjectEntry[] = []; + let windowStart = 0; + let windowRows = 0; + let beforeWindow = 0; + for (const subject of entry.subjects) { + if (beforeWindow + subject.rowCount <= skip && window.length === 0) { + beforeWindow += subject.rowCount; + continue; + } + if (window.length === 0) windowStart = beforeWindow; + if (subject.rowCount > SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS) { + throw snapshotBudgetError({ + key: budgetKey, + reason: 'snapshot_rows', + rows: subject.rowCount, + bytesEstimate: 0, + limit: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS, + }); + } + window.push(subject); + windowRows += subject.rowCount; + if (windowStart + windowRows >= skip + remaining) break; + } + if (window.length === 0) { + skip = 0; + continue; + } + const windowRowsRead = await readFreshSwmMetaSubjectWindowRows( + store, + entry.graph, + window, + signal, + ); + const page = windowRowsRead.slice(skip - windowStart, skip - windowStart + remaining); + for (const row of page) rows.push(row); + remaining -= page.length; + if (remaining <= 0) break; + skip = 0; + } + return rows; +} + +/** + * TTL-filtered bounded snapshot (#1847). The per-snapshot budget binds on the + * plan's ADMITTED row total — what will actually be served — instead of the + * raw graph size, so a 64,000-row `_meta` history with a small fresh subset + * takes the ordinary memoized-snapshot path. The collected rows then pass + * through {@link filterSwmMetaSnapshotRows}, the canonical in-process + * admission filter, exactly as the raw-graph snapshot always has; the plan's + * SPARQL discovery is a candidate superset of that filter for the canonical + * typed-literal meta writes, so both stages agree in production. + */ +async function readBoundedFreshSwmMetaSnapshot( + store: TripleStore, + plan: FreshSwmMetaPlan, + cutoffIso: string, + cache: RowListCache, +): Promise { + const limits = cache.memo.snapshotLoadLimits ?? { + maxRows: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS, + maxBytesEstimate: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + pageRows: SYNC_RESPONDER_SNAPSHOT_BUILD_PAGE_ROWS, + }; + if (plan.totalRows > limits.maxRows) { + throw snapshotBudgetError({ + key: cache.key, + reason: 'snapshot_rows', + rows: plan.totalRows, + bytesEstimate: 0, + limit: limits.maxRows, + }); + } + const rows: SyncRow[] = []; + let bytesEstimate = 0; + for (const entry of plan.entries) { + const graphRows = await readFreshSwmMetaSubjectWindowRows( + store, + entry.graph, + entry.subjects, + ); + for (const row of graphRows) { + const nextBytes = bytesEstimate + estimateStringRowHeapBytes(row.s, row.p, row.o, row.g); + if (nextBytes > limits.maxBytesEstimate) { + throw snapshotBudgetError({ + key: cache.key, + reason: 'snapshot_bytes', + rows: rows.length + 1, + bytesEstimate: nextBytes, + limit: limits.maxBytesEstimate, + }); + } + rows.push(row); + bytesEstimate = nextBytes; + } + } + return filterSwmMetaSnapshotRows(rows, cutoffIso); +} + // NOTE: keep in sync with its page-safe twin {@link readFreshSwmDataRowsPage} — // both MUST return the same SET of rows (see readDurableMetaRows note). async function readFreshSwmDataRows( diff --git a/packages/agent/src/sync/responder/sync-handler.ts b/packages/agent/src/sync/responder/sync-handler.ts index fedee5c2d1..23ab582e8b 100644 --- a/packages/agent/src/sync/responder/sync-handler.ts +++ b/packages/agent/src/sync/responder/sync-handler.ts @@ -21,6 +21,7 @@ import { createResponderGraphListMemo, createResponderExactGraphPagePlanMemo, createResponderFreshSwmDataGraphPlanMemo, + createResponderFreshSwmMetaPlanMemo, createResponderSyncRowListMemo, createResponderSubGraphRegistrationMemo, createResponderSwmAdmissionMemo, @@ -445,6 +446,10 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void { DURABLE_DATA_SYNC_SESSION_TTL_MS, SYNC_RESPONDER_SHARED_MEMORY_SNAPSHOT_LIMIT, ); + const freshSwmMetaPlanMemo = createResponderFreshSwmMetaPlanMemo( + DURABLE_DATA_SYNC_SESSION_TTL_MS, + SYNC_RESPONDER_SHARED_MEMORY_SNAPSHOT_LIMIT, + ); const durableDataExactGraphPlanMemo = createResponderExactGraphPagePlanMemo( DURABLE_DATA_SYNC_SESSION_TTL_MS, SYNC_RESPONDER_DURABLE_DATA_SNAPSHOT_LIMIT, @@ -667,6 +672,7 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void { rowListCacheKey: session?.rowListCacheKey, refreshRowList: session?.refreshRowList, refreshGeneration: session?.refreshGeneration, + freshMetaPlanMemo: freshSwmMetaPlanMemo, }); const queryDurationMs = Date.now() - queryStartedAt; const serializeStartedAt = Date.now(); diff --git a/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts b/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts new file mode 100644 index 0000000000..e70c1429c3 --- /dev/null +++ b/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts @@ -0,0 +1,519 @@ +import { describe, it, expect } from 'vitest'; +import type { OperationContext } from '@origintrail-official/dkg-core'; +import { OxigraphStore, type Quad } from '@origintrail-official/dkg-storage'; +import { + DKG_NS, + RDF_TYPE, + linesFromNquads, + registerTestSyncHandler, + subGraphRegistrationQuads, + workspaceOpQuads, + type CapturedSyncHandler, +} from './_helpers/sync-responder.js'; +import { MemorySyncCheckpointStore } from '../src/sync/checkpoint/state.js'; +import { fetchSyncPages } from '../src/sync/requester/page-fetch.js'; +import type { SyncRequestEnvelope } from '../src/sync/auth/request-build.js'; + +/** + * #1847 — SWM meta lane ceiling. A CG whose `_meta` crossed + * SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS (64,000) raw rows became permanently + * unsyncable for TTL-filtered sessions: the bounded snapshot applied its budget + * to the RAW graph before the TTL filter, and `readSwmMetaPage` passed + * `params.cutoffIso == null` POSITIONALLY as `fallbackOnPerSnapshotBudget`, so + * the refusal had no fallback. These tests seed real >64,000-row stores and + * prove the lane now serves them completely, page by page, within the DEFAULT + * production budgets — and that the deleted global-sort TTL query never runs. + */ + +const XSD_DT = 'http://www.w3.org/2001/XMLSchema#dateTime'; +const XSD_INT = 'http://www.w3.org/2001/XMLSchema#integer'; +const TTL_MS = 60_000; + +const TINY_SNAPSHOT_BUDGET = { + maxRows: 1_000_000, + maxBytesEstimate: Number.MAX_SAFE_INTEGER, + maxSnapshotRows: 1, + maxSnapshotBytesEstimate: Number.MAX_SAFE_INTEGER, +} as const; + +function freshIso(): string { + return new Date(Date.now() - 1_000).toISOString(); +} + +function staleIso(): string { + return new Date(Date.now() - 10 * TTL_MS).toISOString(); +} + +/** Graph-scoped head + selected WorkspaceOperation (11 rows), per swm-recovery shape. */ +function graphScopedHeadQuads( + cgId: string, + metaGraph: string, + ual: string, + opId: string, + timestamp: string, +): Quad[] { + const op = `urn:dkg:share:${cgId}:${opId}`; + const head = `${ual}#dkg-swm-head`; + return [ + { graph: metaGraph, subject: op, predicate: RDF_TYPE, object: `${DKG_NS}WorkspaceOperation` }, + { graph: metaGraph, subject: op, predicate: `${DKG_NS}publishedAt`, object: `"${timestamp}"^^<${XSD_DT}>` }, + { graph: metaGraph, subject: op, predicate: `${DKG_NS}shareOperationId`, object: `"${opId}"` }, + { graph: metaGraph, subject: op, predicate: `${DKG_NS}contentScopeVersion`, object: `"2"^^<${XSD_INT}>` }, + { graph: metaGraph, subject: op, predicate: `${DKG_NS}kaUal`, object: ual }, + { graph: metaGraph, subject: op, predicate: `${DKG_NS}assertionVersion`, object: `"1"^^<${XSD_INT}>` }, + { graph: metaGraph, subject: head, predicate: `${DKG_NS}contentScopeVersion`, object: `"2"^^<${XSD_INT}>` }, + { graph: metaGraph, subject: head, predicate: `${DKG_NS}kaUal`, object: ual }, + { graph: metaGraph, subject: head, predicate: `${DKG_NS}assertionVersion`, object: `"1"^^<${XSD_INT}>` }, + { graph: metaGraph, subject: head, predicate: `${DKG_NS}shareOperationId`, object: `"${opId}"` }, + { graph: metaGraph, subject: head, predicate: `${DKG_NS}assertionGraph`, object: `${metaGraph.replace(/_meta$/, '')}/0x00000000000000000000000000000000000000ab/1` }, + ]; +} + +async function insertChunked(store: OxigraphStore, quads: Quad[]): Promise { + for (let offset = 0; offset < quads.length; offset += 8_000) { + await store.insert(quads.slice(offset, offset + 8_000)); + } +} + +async function collectAllPages( + cap: CapturedSyncHandler, + base: Omit, + pageSize: number, + maxPages = 300, +): Promise<{ lines: Set; pages: number }> { + const lines = new Set(); + let pages = 0; + for (let offset = 0, page = 0; page < maxPages; page += 1, offset += pageSize) { + const out = await cap.invoke({ ...base, offset }); + const pageLines = linesFromNquads(out); + pages += 1; + for (const line of pageLines) lines.add(line); + if (pageLines.length < pageSize) break; + } + return { lines, pages }; +} + +/** Fails the test if the deleted TTL global-sort shape — or ANY OFFSET/ORDER BY + * query over an SWM meta graph — reaches the store during a TTL session. */ +function forbidSwmMetaSortOrOffsetQueries(store: OxigraphStore) { + const originalQuery = store.query.bind(store); + let windowQueries = 0; + store.query = (async (sparql: string, options?: unknown) => { + const normalized = sparql.replace(/\s+/g, ' ').trim(); + if (normalized.includes('_shared_memory_meta')) { + expect(normalized).not.toMatch(/OFFSET \d/); + expect(normalized).not.toContain('ORDER BY'); + if (normalized.includes('VALUES ?s')) windowQueries += 1; + } + return originalQuery(sparql, options as never); + }) as OxigraphStore['query']; + return { + assertWindowQueriesObserved: () => expect(windowQueries).toBeGreaterThan(0), + }; +} + +describe('SWM meta lane above the 64,000-row snapshot ceiling (#1847)', () => { + it('serves a 64,000+-row _meta with a small fresh subset completely at DEFAULT budgets (the fifa-world-cup-2026 shape)', async () => { + const cgId = 'meta-ceiling-fifa'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const stale = staleIso(); + const fresh = freshIso(); + + const quads: Quad[] = []; + // 12,800 stale operations x 5 rows = 64,000 raw rows: over the build cap. + for (let index = 0; index < 12_800; index += 1) { + quads.push(...workspaceOpQuads(cgId, `stale-${index}`, `urn:stale:root:${index}`, metaGraph, stale)); + } + // The small fresh subset that TTL sessions actually need. + const freshOpIds = ['fresh-a', 'fresh-b', 'fresh-c']; + for (const opId of freshOpIds) { + quads.push(...workspaceOpQuads(cgId, opId, `urn:fresh:root:${opId}`, metaGraph, fresh)); + } + const freshUal = 'did:dkg:hardhat:31337/0x00000000000000000000000000000000000000ab/1'; + quads.push(...graphScopedHeadQuads(cgId, metaGraph, freshUal, 'fresh-head-op', fresh)); + expect(quads.length).toBeGreaterThan(64_000); + + const store = new OxigraphStore(); + const seedStartedAt = Date.now(); + await insertChunked(store, quads); + const seedDurationMs = Date.now() - seedStartedAt; + + // DEFAULT production budgets: no snapshotBudget override. + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 7 }); + const watch = forbidSwmMetaSortOrOffsetQueries(store); + + const serveStartedAt = Date.now(); + const { lines, pages } = await collectAllPages( + cap, + { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta', limit: 7, syncSessionId: 'fifa-session' }, + 7, + ); + const serveDurationMs = Date.now() - serveStartedAt; + + // Every fresh row is served — the lane is no longer refused. + // 3 ops x 5 rows + head group 11 rows = 26. + expect(lines.size).toBe(26); + const joined = [...lines].join('\n'); + for (const opId of freshOpIds) { + expect(joined).toContain(`urn:dkg:share:${cgId}:${opId}`); + } + expect(joined).toContain(`${freshUal}#dkg-swm-head`); + expect(joined).toContain('assertionVersion'); + expect(joined).not.toContain('urn:stale:root'); + watch.assertWindowQueriesObserved(); + + // eslint-disable-next-line no-console + console.info( + `#1847 fifa-shape: raw=${quads.length} rows, fresh=26 rows, pages=${pages}, ` + + `seed=${seedDurationMs}ms, serve=${serveDurationMs}ms`, + ); + await store.close(); + }, 120_000); + + it('serves an INTRINSICALLY oversized fresh set (>64,000 admitted rows) via bounded plan paging at DEFAULT budgets', async () => { + const cgId = 'meta-ceiling-allfresh'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + + const quads: Quad[] = []; + // 13,000 FRESH operations x 5 rows = 65,000 admitted rows: even the + // filtered set exceeds the per-snapshot cap, so the session must degrade + // to plan-paged serving instead of refusing. This test is the direct + // mutation-kill for the positional `params.cutoffIso == null` defect: + // reintroduce it and this session throws the per-snapshot budget error. + for (let index = 0; index < 13_000; index += 1) { + quads.push(...workspaceOpQuads(cgId, `f${index}`, `urn:fresh:root:${index}`, metaGraph, fresh)); + } + expect(quads.length).toBe(65_000); + + const store = new OxigraphStore(); + const seedStartedAt = Date.now(); + await insertChunked(store, quads); + const seedDurationMs = Date.now() - seedStartedAt; + + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 5000 }); + const watch = forbidSwmMetaSortOrOffsetQueries(store); + + const serveStartedAt = Date.now(); + const { lines, pages } = await collectAllPages( + cap, + { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta', limit: 5000, syncSessionId: 'allfresh-session' }, + 5000, + ); + const serveDurationMs = Date.now() - serveStartedAt; + + // The complete oversized fresh set is served, page by page, no refusal. + expect(lines.size).toBe(65_000); + watch.assertWindowQueriesObserved(); + + // eslint-disable-next-line no-console + console.info( + `#1847 oversized-fresh: rows=65000, pages=${pages}, seed=${seedDurationMs}ms, serve=${serveDurationMs}ms`, + ); + await store.close(); + }, 120_000); + + it('plan-paged serving is set-equivalent to the snapshot lane across buckets, heads and stale exclusion', async () => { + const cgId = 'meta-ceiling-equiv'; + const cgPrefix = `did:dkg:context-graph:${cgId}`; + const rootMeta = `${cgPrefix}/_shared_memory_meta`; + const subMeta = `${cgPrefix}/subx/_shared_memory_meta`; + const fresh = freshIso(); + const stale = staleIso(); + + const quads: Quad[] = [ + ...subGraphRegistrationQuads(cgId, 'subx'), + ...workspaceOpQuads(cgId, 'root-fresh', 'urn:r:fresh', rootMeta, fresh), + ...workspaceOpQuads(cgId, 'root-stale', 'urn:r:stale', rootMeta, stale), + ...workspaceOpQuads(cgId, 'sub-fresh', 'urn:s:fresh', subMeta, fresh), + ...workspaceOpQuads(cgId, 'sub-stale', 'urn:s:stale', subMeta, stale), + ...graphScopedHeadQuads(cgId, rootMeta, 'did:dkg:hardhat:31337/0x00000000000000000000000000000000000000ab/7', 'head-fresh', fresh), + ...graphScopedHeadQuads(cgId, rootMeta, 'did:dkg:hardhat:31337/0x00000000000000000000000000000000000000ab/8', 'head-stale', stale), + ]; + + const canonicalStore = new OxigraphStore(); + await canonicalStore.insert(quads); + const pagedStore = new OxigraphStore(); + await pagedStore.insert(quads); + + const canonicalCap = registerTestSyncHandler(canonicalStore, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 5000 }); + const pagedCap = registerTestSyncHandler(pagedStore, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 3, + snapshotBudget: TINY_SNAPSHOT_BUDGET, + }); + const base = { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta' as const }; + + const canonical = (await collectAllPages( + canonicalCap, { ...base, limit: 5000, syncSessionId: 'canon' }, 5000, + )).lines; + const paged = (await collectAllPages( + pagedCap, { ...base, limit: 3, syncSessionId: 'paged' }, 3, + )).lines; + + expect(paged).toEqual(canonical); + const joined = [...canonical].join('\n'); + expect(joined).toContain('urn:dkg:share:meta-ceiling-equiv:root-fresh'); + expect(joined).toContain('urn:dkg:share:meta-ceiling-equiv:sub-fresh'); + expect(joined).toContain('#dkg-swm-head'); + expect(joined).toContain('/7#dkg-swm-head'); + expect(joined).not.toContain('/8#dkg-swm-head'); + expect(joined).not.toContain('root-stale'); + expect(joined).not.toContain('sub-stale'); + // Both buckets appear in the graph position. + expect(joined).toContain(`<${rootMeta}> .`); + expect(joined).toContain(`<${subMeta}> .`); + await canonicalStore.close(); + await pagedStore.close(); + }); + + it('fails the session (not silently skips/duplicates) when an admitted subject mutates between plan pages, and a fresh session recovers', async () => { + const cgId = 'meta-ceiling-mutate'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + // Deterministic subject order: op ids sort a < b < c... + const opIds = ['a', 'b', 'c', 'd', 'e', 'f']; + for (const opId of opIds) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:m:${opId}`, metaGraph, fresh)); + } + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 5, + snapshotBudget: TINY_SNAPSHOT_BUDGET, // forces plan-paged mode + }); + const base = { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta' as const, limit: 5 }; + + const page0 = linesFromNquads(await cap.invoke({ ...base, offset: 0, syncSessionId: 'M1' })); + expect(page0).toHaveLength(5); + + // Grow a subject the NEXT page's window must read. + const grownSubject = `urn:dkg:share:${cgId}:${opIds[1]}`; + await store.insert([{ graph: metaGraph, subject: grownSubject, predicate: `${DKG_NS}note`, object: '"grown"' }]); + + await expect(cap.invoke({ ...base, offset: 5, syncSessionId: 'M1' })) + .rejects.toThrow(/Shared-memory meta sync plan changed while reading/); + + // A fresh session rebuilds the plan and serves the grown store completely. + const recovered = await collectAllPages( + cap, { ...base, syncSessionId: 'M2' }, 5, + ); + expect(recovered.lines.size).toBe(6 * 5 + 1); + expect([...recovered.lines].join('\n')).toContain('"grown"'); + await store.close(); + }); + + it('keeps a bounded refusal ONLY for a single pathological subject exceeding the hard 64,000-row build cap', async () => { + const cgId = 'meta-ceiling-monster'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + // ONE fresh subject carrying 64,001 rows. Whole-subject windows are the + // consistency unit of the plan lane (they are what keeps a seal/head + // row-group atomic per #1788), so this single row-group can never be + // served coherently within the hard build cap — a bounded refusal, at + // DEFAULT budgets, is the correct answer. Ordinary multi-row subjects + // under a shrunken session budget are covered by the paged tests above. + const subject = 'urn:monster'; + const monsterQuads: Quad[] = [ + { graph: metaGraph, subject, predicate: `${DKG_NS}publishedAt`, object: `"${fresh}"^^<${XSD_DT}>` }, + ]; + for (let index = 1; index <= 64_000; index += 1) { + monsterQuads.push({ + graph: metaGraph, subject, predicate: `${DKG_NS}note`, object: `"filler-${index}"`, + }); + } + await insertChunked(store, monsterQuads); + + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 500 }); + + await expect(cap.invoke({ + contextGraphId: cgId, + includeSharedMemory: true, + phase: 'meta', + offset: 0, + limit: 500, + syncSessionId: 'monster-session', + })).rejects.toThrow(/per-snapshot rows budget/); + await store.close(); + }, 120_000); + + it('legacy cutoff-less sessions keep the unfiltered store-paged compatibility fallback', async () => { + const cgId = 'meta-ceiling-legacy'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const store = new OxigraphStore(); + const iso = '2026-06-01T00:00:00.000Z'; + for (const opId of ['x', 'y', 'z']) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:l:${opId}`, metaGraph, iso)); + } + + let legacyPagedQueries = 0; + const originalQuery = store.query.bind(store); + store.query = (async (sparql: string, options?: unknown) => { + const normalized = sparql.replace(/\s+/g, ' ').trim(); + if ( + normalized.includes('VALUES ?g') && + normalized.includes('_shared_memory_meta') && + normalized.includes('ORDER BY ?g ?s ?p ?o') && + /OFFSET \d+/.test(normalized) + ) { + // The legacy paged query must never carry the TTL join. + expect(normalized).not.toContain('publishedAt'); + expect(normalized).not.toContain('FILTER'); + legacyPagedQueries += 1; + } + return originalQuery(sparql, options as never); + }) as OxigraphStore['query']; + + // sharedMemoryTtlMs: 0 => cutoffIso == null (legacy lane), tiny budget + // forces the fallback. + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: 0, + syncPageSize: 4, + snapshotBudget: TINY_SNAPSHOT_BUDGET, + }); + const { lines } = await collectAllPages( + cap, + { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta', limit: 4, syncSessionId: 'legacy' }, + 4, + ); + expect(lines.size).toBe(15); + expect(legacyPagedQueries).toBeGreaterThan(0); + await store.close(); + }); +}); + +describe('requester reassembly of the plan-paged SWM meta lane (#1847 x #1788)', () => { + function makeCtx(): OperationContext { + return { kind: 'system', id: 'meta-ceiling-requester', startedAt: Date.now() } as never; + } + const noop = () => {}; + + /** Minimal N-Quads line parser for the fixture vocabulary (IRIs + literals). */ + function parseNquads(text: string): Quad[] { + const quads: Quad[] = []; + for (const line of text.split('\n')) { + const trimmed = line.trim(); + if (!trimmed) continue; + const match = trimmed.match(/^<([^>]+)> <([^>]+)> (.+) <([^>]+)> \.$/); + if (!match) throw new Error(`unparseable line: ${trimmed}`); + quads.push({ subject: match[1], predicate: match[2], object: match[3], graph: match[4] }); + } + return quads; + } + + async function fetchAllMeta( + cap: CapturedSyncHandler, + cgId: string, + pageSize: number, + ) { + return fetchSyncPages({ + ctx: makeCtx(), + remotePeerId: '12D3KooWMetaCeilingRemote', + contextGraphId: cgId, + includeSharedMemory: true, + phase: 'meta', + graphUri: `did:dkg:context-graph:${cgId}/_shared_memory_meta`, + deadline: Date.now() + 60_000, + syncPageTimeoutMs: 10_000, + syncRouterAttempts: 1, + syncPageRetryAttempts: 1, + syncPageSize: pageSize, + syncDeniedResponse: 'sync-denied', + debugSyncProgress: false, + protocolSync: '/origintrail/dkg/sync/1.0.0', + checkpointStore: new MemorySyncCheckpointStore(), + buildSyncRequest: async (contextGraphId, offset, limit, includeSharedMemory, _peer, phase, _snap, _since, syncSessionId) => + new TextEncoder().encode(JSON.stringify({ + contextGraphId, offset, limit, includeSharedMemory, phase, syncSessionId, + })), + parseAndFilter: async (nquadsText) => { + const quads = parseNquads(nquadsText); + return { quads, totalQuads: quads.length }; + }, + send: async (_peerId, _protocolId, data) => { + const envelope = JSON.parse(new TextDecoder().decode(data)) as SyncRequestEnvelope; + return new TextEncoder().encode(await cap.invoke(envelope)); + }, + logWarn: noop, + logInfo: noop, + logDebug: noop, + }); + } + + /** Assert no subject group lost a field to a page boundary (#1788 class). */ + function assertNoStrippedFields(quads: readonly Quad[], expectedGroups: ReadonlyMap) { + const bySubject = new Map>(); + for (const quad of quads) { + const predicates = bySubject.get(quad.subject) ?? new Set(); + predicates.add(quad.predicate); + bySubject.set(quad.subject, predicates); + } + for (const [subject, expectedPredicates] of expectedGroups) { + const predicates = bySubject.get(subject); + expect(predicates, `subject ${subject} missing entirely`).toBeDefined(); + for (const predicate of expectedPredicates) { + expect( + predicates!.has(predicate), + `subject ${subject} lost <${predicate}> across a page boundary`, + ).toBe(true); + } + } + } + + const OP_PREDICATES = [ + RDF_TYPE, + `${DKG_NS}publishedAt`, + `${DKG_NS}rootEntity`, + `${DKG_NS}contextGraphId`, + `${DKG_NS}shareOperationId`, + ] as const; + const HEAD_PREDICATES = [ + `${DKG_NS}contentScopeVersion`, + `${DKG_NS}kaUal`, + `${DKG_NS}assertionVersion`, + `${DKG_NS}shareOperationId`, + `${DKG_NS}assertionGraph`, + ] as const; + + for (const [label, snapshotBudget] of [ + ['session snapshot lane (default budgets)', undefined], + ['plan-paged lane (oversized snapshot)', TINY_SNAPSHOT_BUDGET], + ] as const) { + it(`reassembles every seal/head row-group with no stripped fields via the ${label}`, async () => { + const cgId = `meta-reassembly-${snapshotBudget ? 'paged' : 'snap'}`; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + + const expectedGroups = new Map(); + const quads: Quad[] = []; + for (let index = 0; index < 30; index += 1) { + const opId = `op-${String(index).padStart(2, '0')}`; + quads.push(...workspaceOpQuads(cgId, opId, `urn:re:${opId}`, metaGraph, fresh)); + expectedGroups.set(`urn:dkg:share:${cgId}:${opId}`, OP_PREDICATES); + } + for (let index = 0; index < 4; index += 1) { + const ual = `did:dkg:hardhat:31337/0x00000000000000000000000000000000000000ab/${index + 1}`; + quads.push(...graphScopedHeadQuads(cgId, metaGraph, ual, `head-${index}`, fresh)); + expectedGroups.set(`${ual}#dkg-swm-head`, HEAD_PREDICATES); + } + await store.insert(quads); + + // Page size 4 vs 5- and 11-row groups: every group straddles a boundary. + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 4, + ...(snapshotBudget ? { snapshotBudget } : {}), + }); + const result = await fetchAllMeta(cap, cgId, 4); + + expect(result.completed).toBe(true); + expect(result.timedOut).toBe(false); + expect(result.quads.length).toBe(quads.length); + assertNoStrippedFields(result.quads, expectedGroups); + await store.close(); + }); + } +}); diff --git a/packages/agent/vitest.unit.config.ts b/packages/agent/vitest.unit.config.ts index f89c5a1511..efa04d2fde 100644 --- a/packages/agent/vitest.unit.config.ts +++ b/packages/agent/vitest.unit.config.ts @@ -49,6 +49,7 @@ export default defineConfig({ "test/sync-responder-snapshot-cache.test.ts", "test/sync-responder-cursor.test.ts", "test/sync-responder-oversized-fallback.test.ts", + "test/sync-responder-swm-meta-ceiling.test.ts", "test/sync-responder-large-graph-stack-overflow.test.ts", "test/sync-page-frame-budget.test.ts", "test/sync-byte-budget-pages.test.ts", From bb50dc541480aef69dbc9108eec29c60caf23aac Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Mon, 20 Jul 2026 20:37:19 +0200 Subject: [PATCH 2/4] fix(sync): bound TTL meta session plans and bind subject content across pages (#1868 review) Four review findings on the #1847 lane, each with a dedicated regression: - Discovery is now bounded by construction: LIMIT-capped subject discovery (FRESH_SWM_META_PLAN_MAX_SUBJECTS) + fixed response byte caps on every plan query, with typed per-snapshot refusals; retained plans carry a scalar byte estimate charged to the process-wide responder snapshot budget as control-plane entries (LRU-evictable, globally rejected under pressure). - StoreResponseTooLargeError during TTL snapshot materialization converts to the per-snapshot snapshot_bytes budget error so the phase degrades to plan paging instead of failing outright. - Whole-subject window reads verify PER-SUBJECT counts against the plan and bind a content digest on first read, verified on every reread: same-count replacements and compensating cross-subject mutations fail the session instead of tearing or misaligning row-groups at page seams. - readResponderRowsPage optional behavior is a named options object; the fallback policy can no longer be passed positionally (the original defect). Co-Authored-By: Claude Fable 5 --- .../agent/src/sync/responder/graph-plan.ts | 464 ++++++++++++++---- .../src/sync/responder/snapshot-budget.ts | 21 +- .../agent/src/sync/responder/sync-handler.ts | 4 + .../sync-responder-swm-meta-ceiling.test.ts | 402 ++++++++++++++- 4 files changed, 783 insertions(+), 108 deletions(-) diff --git a/packages/agent/src/sync/responder/graph-plan.ts b/packages/agent/src/sync/responder/graph-plan.ts index 6e26e94c75..f98a6f2116 100644 --- a/packages/agent/src/sync/responder/graph-plan.ts +++ b/packages/agent/src/sync/responder/graph-plan.ts @@ -23,7 +23,12 @@ import { SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS, SYNC_RESPONDER_SNAPSHOT_BUILD_PAGE_ROWS, } from './snapshot-cache.js'; -import { SyncRowSnapshotBudgetError } from './snapshot-budget.js'; +import { bytesToHex } from '@noble/hashes/utils.js'; +import { sha256 } from '@noble/hashes/sha2.js'; +import { + SyncRowSnapshotBudgetError, + type SyncResponderSnapshotBudget, +} from './snapshot-budget.js'; import { estimateStringRowHeapBytes } from '../memory-telemetry.js'; import type { ChangelogSyncResponse, ChangelogDeltaRecord } from '../changelog/wire.js'; import { durableMetaDelegationSubjectAdmissionExpression } from './durable-meta-admission.js'; @@ -102,6 +107,17 @@ export interface FreshSwmDataGraphPlanMemo { interface FreshSwmMetaSubjectEntry { subject: string; rowCount: number; + /** + * Content binding for the subject's whole row-group, established on the + * FIRST window read of this session and verified on every REREAD. Row counts + * alone pass on same-count replacements, and a reread sliced at the plan's + * prefix sums could then combine rows of two different versions of one + * subject across response pages; the digest makes any content or ordering + * change of an already-served subject fail the session instead (the + * requester restarts with a fresh plan). A subject read exactly once needs + * no binding: its row-group is served whole from a single query. + */ + contentDigest?: string; } interface FreshSwmMetaGraphPlanEntry { @@ -113,13 +129,19 @@ interface FreshSwmMetaGraphPlanEntry { /** * Session plan for the TTL-filtered SWM meta phase (#1847). Holds only - * graph/subject/count scalars — never payload rows — so a 64,000+-row `_meta` - * graph costs the plan a few hundred kilobytes at most while the rows stay in - * the store until each page addresses its own bounded subject window. + * graph/subject/count scalars — never payload rows — and is bounded at + * CONSTRUCTION: discovery queries carry LIMIT/response-byte caps, the admitted + * subject cardinality is capped by {@link FRESH_SWM_META_PLAN_MAX_SUBJECTS}, + * and the retained scalar estimate is capped by the fixed snapshot build byte + * cap, so plan building can never materialize an unbounded store result. The + * retained estimate is additionally charged to the process-wide responder + * snapshot budget by the memo (see createResponderFreshSwmMetaPlanMemo). */ interface FreshSwmMetaPlan { entries: readonly FreshSwmMetaGraphPlanEntry[]; totalRows: number; + /** Estimated retained heap bytes of the plan's subject/count scalars. */ + bytesEstimate: number; } export interface FreshSwmMetaPlanMemo { @@ -289,26 +311,58 @@ export function createResponderFreshSwmDataGraphPlanMemo( * lifetime/refresh contract as {@link createResponderFreshSwmDataGraphPlanMemo}: * touched on every page, offset>0 requires the existing plan so a rebuilt plan * against a moved TTL cutoff can never make a numeric offset skip or duplicate. + * + * When a responder snapshot budget is supplied, every retained plan's scalar + * estimate is charged to the GLOBAL budget as a control-plane entry: peers + * cannot stack up to maxEntries uncharged plans, admission under global memory + * pressure fails as the quiet retryable limit, and an idle plan is LRU-evicted + * exactly like a retained row snapshot (the session then expires and the + * requester restarts it). */ export function createResponderFreshSwmMetaPlanMemo( ttlMs = 10 * 60_000, maxEntries = 32, + budget?: SyncResponderSnapshotBudget, ): FreshSwmMetaPlanMemo { - return createSessionPlanMemo(ttlMs, maxEntries); + return createSessionPlanMemo( + ttlMs, + maxEntries, + budget && { + budget, + phase: 'shared_memory', + bytesEstimate: (plan) => plan.bytesEstimate, + }, + ); +} + +interface SessionPlanBudgetAccounting { + budget: SyncResponderSnapshotBudget; + phase: 'shared_memory' | 'durable_meta' | 'durable_data'; + bytesEstimate: (value: T) => number; } -function createSessionPlanMemo(ttlMs: number, maxEntries: number): { +function createSessionPlanMemo( + ttlMs: number, + maxEntries: number, + accounting?: SessionPlanBudgetAccounting, +): { get( key: string, load: () => Promise, options?: { refresh?: boolean; requireExisting?: boolean; signal?: AbortSignal }, ): Promise; } { - const cached = new Map(); + const cached = new Map(); const inflight = new Map>(); + const deleteEntry = (key: string, reason: 'expired' | 'released' | 'replaced') => { + const entry = cached.get(key); + if (!entry) return; + cached.delete(key); + if (entry.budgetEntryId) accounting?.budget.remove(entry.budgetEntryId, reason); + }; const prune = (now = Date.now()) => { for (const [key, entry] of cached) { - if (now - entry.cachedAt >= ttlMs) cached.delete(key); + if (now - entry.cachedAt >= ttlMs) deleteEntry(key, 'expired'); } }; return { @@ -321,14 +375,44 @@ function createSessionPlanMemo(ttlMs: number, maxEntries: number): { const existing = cached.get(key); if (!options?.refresh && existing) { cached.delete(key); - cached.set(key, { value: existing.value, cachedAt: now }); + cached.set(key, { ...existing, cachedAt: now }); + if (existing.budgetEntryId) { + // Refresh the global-budget LRU position, then stay evictable: an + // entry pinned forever would let idle plans exempt themselves from + // memory-pressure eviction. + accounting?.budget.touch(existing.budgetEntryId); + accounting?.budget.release(existing.budgetEntryId); + } return existing.value; } if (options?.requireExisting) return null; - if (!existing && cached.size >= maxEntries) cached.delete(cached.keys().next().value!); + if (!existing && cached.size >= maxEntries) { + deleteEntry(cached.keys().next().value!, 'released'); + } const pendingLoad = load() .then((value) => { - cached.set(key, { value, cachedAt: Date.now() }); + const replaced = cached.get(key); + let budgetEntryId: symbol | undefined; + if (accounting) { + budgetEntryId = Symbol(key); + // Throws the typed global budget error when the process-wide + // responder budget cannot admit the plan; the failed refresh leaves + // any previously-admitted plan in place (memo entry untouched). + accounting.budget.admit({ + id: budgetEntryId, + key, + phase: accounting.phase, + rows: 0, + bytesEstimate: accounting.bytesEstimate(value), + controlPlane: true, + replaceId: replaced?.budgetEntryId, + onEvict: () => { + if (cached.get(key)?.budgetEntryId === budgetEntryId) cached.delete(key); + }, + }); + accounting.budget.release(budgetEntryId); + } + cached.set(key, { value, cachedAt: Date.now(), budgetEntryId }); return value; }) .finally(() => inflight.delete(key)); @@ -519,11 +603,13 @@ export async function readSwmMetaPage(params: { params.limit, params.signal, cache - ? () => readBoundedSwmMetaSnapshot( - params.store, - candidateGraphs, - cache, - ) + ? { + loadSnapshot: () => readBoundedSwmMetaSnapshot( + params.store, + candidateGraphs, + cache, + ), + } : undefined, ); } @@ -548,6 +634,7 @@ export async function readSwmMetaPage(params: { // intrinsically-oversized fresh set degrades to bounded whole-subject window // pages from the same plan instead of failing permanently. const cutoffIso = params.cutoffIso; + const budgetKey = cache?.key ?? `swm-meta:${params.contextGraphId}`; // Consume the explicit session refresh once: when the snapshot build crosses // its budget, the immediate page-zero fallback must reuse the just-built plan // instead of rebuilding (and re-counting) it against a moving store. @@ -560,6 +647,7 @@ export async function readSwmMetaPage(params: { params.store, candidateGraphs, cutoffIso, + budgetKey, pageSignal, ); const refreshPlan = pageOffset === 0 && planRefreshPending; @@ -582,7 +670,7 @@ export async function readSwmMetaPage(params: { await getPlan(offset, signal), offset, limit, - cache?.key ?? `swm-meta:${params.contextGraphId}`, + budgetKey, signal, ); return readResponderRowsPage( @@ -591,20 +679,22 @@ export async function readSwmMetaPage(params: { params.offset, params.limit, params.signal, - cache - ? async () => readBoundedFreshSwmMetaSnapshot( - params.store, - await getPlan(0, undefined), - cutoffIso, - cache, - ) - : undefined, - // The per-snapshot budget fallback MUST stay enabled here (#1847): it now - // degrades to the bounded plan-paged reader above, never to the deleted - // global-sort query. Passing `params.cutoffIso == null` in this position is - // the exact defect that made every 64,000-row `_meta` CG permanently - // unsyncable on mainnet. - true, + { + loadSnapshot: cache + ? async () => readBoundedFreshSwmMetaSnapshot( + params.store, + await getPlan(0, undefined), + cutoffIso, + cache, + ) + : undefined, + // The per-snapshot budget fallback MUST stay enabled here (#1847): it + // degrades to the bounded plan-paged reader above, never to the deleted + // global-sort query. This policy used to be a positional boolean, and + // passing `params.cutoffIso == null` in that position is the exact defect + // that made every 64,000-row `_meta` CG permanently unsyncable on mainnet. + fallbackOnPerSnapshotBudget: true, + }, ); } @@ -724,12 +814,14 @@ export async function readDurableMetaPage(params: { params.limit, params.signal, cache - ? () => readBoundedDurableMetaSnapshot( - params.store, - params.contextGraphId, - params.registeredSubGraphNames, - cache, - ) + ? { + loadSnapshot: () => readBoundedDurableMetaSnapshot( + params.store, + params.contextGraphId, + params.registeredSubGraphNames, + cache, + ), + } : undefined, ); } @@ -776,11 +868,7 @@ async function readGraphScopedVmManifest( key: `durable-v2-manifest:${contextGraphId}`, reason: 'snapshot_bytes', rows: 0, - bytesEstimate: typeof error.actualBytes === 'bigint' - ? Number(error.actualBytes > BigInt(Number.MAX_SAFE_INTEGER) - ? BigInt(Number.MAX_SAFE_INTEGER) - : error.actualBytes) - : error.actualBytes, + bytesEstimate: storeResponseActualBytes(error), limit: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, }); } @@ -1426,12 +1514,14 @@ async function readPagedRowsFromExactGraphPlanLoader( limit, signal, cache - ? async () => readExactGraphPlanSnapshot( - store, - await getPlan(0, undefined), - cache, - rowSnapshotLimits, - ) + ? { + loadSnapshot: async () => readExactGraphPlanSnapshot( + store, + await getPlan(0, undefined), + cache, + rowSnapshotLimits, + ), + } : undefined, ); } @@ -1476,6 +1566,15 @@ function snapshotResponseByteLimit(maxBytesEstimate: number): number { ); } +/** Clamp a store response-cap overshoot (possibly bigint) into a safe number. */ +function storeResponseActualBytes(error: StoreResponseTooLargeError): number { + return typeof error.actualBytes === 'bigint' + ? Number(error.actualBytes > BigInt(Number.MAX_SAFE_INTEGER) + ? BigInt(Number.MAX_SAFE_INTEGER) + : error.actualBytes) + : error.actualBytes; +} + function snapshotBudgetError(params: { key: string; reason: 'snapshot_rows' | 'snapshot_bytes'; @@ -1778,15 +1877,33 @@ async function loadStorePagedSnapshot( } } +/** + * Optional behavior of {@link readResponderRowsPage}, named instead of + * positional: a bare boolean in this helper's signature is how the #1847 + * production defect happened (`params.cutoffIso == null` read as the fallback + * policy), so call sites must now spell the policy out. + */ +interface ResponderRowsPageOptions { + /** Session snapshot loader; omitted phases build via the store-paged loader. */ + loadSnapshot?: () => Promise; + /** + * Whether a PER-snapshot rows/bytes budget refusal degrades to the + * store-bounded page loader for this and every later page of the session + * (defaults to true; global budget pressure always propagates). + */ + fallbackOnPerSnapshotBudget?: boolean; +} + async function readResponderRowsPage( cache: RowListCache | undefined, loadStoreBoundedPage: StorePageLoader, offset: number, limit: number, signal?: AbortSignal, - loadSnapshot?: () => Promise, - fallbackOnPerSnapshotBudget = true, + options?: ResponderRowsPageOptions, ): Promise { + const loadSnapshot = options?.loadSnapshot; + const fallbackOnPerSnapshotBudget = options?.fallbackOnPerSnapshotBudget ?? true; const safeOffset = Math.max(0, Math.floor(offset)); const safeLimit = Math.max(0, Math.floor(limit)); if (safeLimit === 0) return []; @@ -2098,16 +2215,11 @@ async function readBoundedSwmMetaSnapshot( }); } catch (error) { if (!(error instanceof StoreResponseTooLargeError)) throw error; - const actualBytes = typeof error.actualBytes === 'bigint' - ? Number(error.actualBytes > BigInt(Number.MAX_SAFE_INTEGER) - ? BigInt(Number.MAX_SAFE_INTEGER) - : error.actualBytes) - : error.actualBytes; throw snapshotBudgetError({ key: cache.key, reason: 'snapshot_bytes', rows: rows.length, - bytesEstimate: bytesEstimate + actualBytes, + bytesEstimate: bytesEstimate + storeResponseActualBytes(error), limit: limits.maxBytesEstimate, }); } @@ -2244,9 +2356,26 @@ async function readSwmMetaRowsPage( const FRESH_SWM_META_PLAN_SUBJECT_CHUNK = 100; +/** + * Hard cardinality cap for a TTL meta session plan's admitted subjects, across + * all candidate graphs of the phase. The discovery queries are LIMIT-bounded to + * this cap (plus one sentinel row), so plan construction can never materialize + * an unbounded subject set no matter how large the fresh window is: a fresh set + * beyond the cap is a typed bounded refusal, never an unbounded control-plane + * plan. Sizing: every admitted subject serves at least one row, so this cap + * alone admits sessions far past the point where they run plan-paged, while + * the retained plan stays a few megabytes at worst (also capped by the fixed + * build byte estimate below, which bounds pathological IRI lengths). + */ +export const FRESH_SWM_META_PLAN_MAX_SUBJECTS = 32_000; + /** * Discover the TTL-admitted subjects of one SWM meta graph with two - * small-result queries (no payload rows, no sort, no OFFSET): + * small-result queries (no payload rows, no sort, no OFFSET), each bounded by + * construction: LIMIT (remaining subject allowance + 1 sentinel) and the fixed + * snapshot-build response byte cap. Crossing either bound is a typed + * per-snapshot budget refusal — the plan lane's one remaining bounded refusal + * besides the single-oversized-subject case. * * 1. subjects carrying their own fresh `publishedAt` — the * {@link readFreshSwmRoots} shape, an indexed predicate probe whose result @@ -2265,26 +2394,58 @@ async function readFreshSwmMetaSubjects( store: TripleStore, graph: string, cutoffIso: string, + maxSubjects: number, + budgetKey: string, signal?: AbortSignal, ): Promise> { const cutoffFilter = `FILTER(?ts >= ${sparqlString(cutoffIso)}^^)`; + const discoveryLimit = Math.max(1, Math.floor(maxSubjects)) + 1; const subjects = new Set(); - const freshRes = await store.query(` + const runDiscovery = async (sparql: string, operation: string): Promise => { + let res; + try { + res = await store.query(sparql, { + ...syncResponderStoreOptions(signal, operation), + maxResponseBytes: snapshotResponseByteLimit( + SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + ), + }); + } catch (error) { + if (!(error instanceof StoreResponseTooLargeError)) throw error; + throw snapshotBudgetError({ + key: budgetKey, + reason: 'snapshot_bytes', + rows: subjects.size, + bytesEstimate: storeResponseActualBytes(error), + limit: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + }); + } + if (res.type !== 'bindings') return; + for (const row of res.bindings) { + const subject = row['s']; + if (subject && isIriTerm(subject)) subjects.add(subject); + } + if (subjects.size > maxSubjects) { + throw snapshotBudgetError({ + key: budgetKey, + reason: 'snapshot_rows', + rows: subjects.size, + bytesEstimate: 0, + limit: FRESH_SWM_META_PLAN_MAX_SUBJECTS, + }); + } + }; + await runDiscovery(` SELECT DISTINCT ?s WHERE { GRAPH <${assertSafeIri(graph)}> { ?s <${DKG_PUBLISHED_AT}> ?ts . ${cutoffFilter} } } - `, syncResponderStoreOptions(signal, 'sync.responder.readFreshSwmMetaSubjects')); - if (freshRes.type === 'bindings') { - for (const row of freshRes.bindings) { - const subject = row['s']; - if (subject && isIriTerm(subject)) subjects.add(subject); - } - } - const headRes = await store.query(` + LIMIT ${discoveryLimit} + `, 'sync.responder.readFreshSwmMetaSubjects'); + await runDiscovery(` SELECT DISTINCT ?s WHERE { GRAPH <${assertSafeIri(graph)}> { ?s <${DKG_CONTENT_SCOPE_VERSION}> ${GRAPH_KA_CONTENT_SCOPE_VERSION} ; @@ -2300,13 +2461,8 @@ async function readFreshSwmMetaSubjects( ${cutoffFilter} } } - `, syncResponderStoreOptions(signal, 'sync.responder.readFreshSwmMetaHeadSubjects')); - if (headRes.type === 'bindings') { - for (const row of headRes.bindings) { - const subject = row['s']; - if (subject && isIriTerm(subject)) subjects.add(subject); - } - } + LIMIT ${discoveryLimit} + `, 'sync.responder.readFreshSwmMetaHeadSubjects'); return subjects; } @@ -2318,17 +2474,35 @@ async function countFreshSwmMetaSubjectRows( store: TripleStore, graph: string, subjects: readonly string[], + budgetKey: string, signal?: AbortSignal, ): Promise { const countsBySubject = new Map(); for (const chunk of chunkValues(subjects, FRESH_SWM_META_PLAN_SUBJECT_CHUNK)) { - const res = await store.query(` - SELECT ?s (COUNT(*) AS ?count) WHERE { - VALUES ?s { ${subjectValues(chunk)} } - GRAPH <${assertSafeIri(graph)}> { ?s ?p ?o } - } - GROUP BY ?s - `, syncResponderStoreOptions(signal, 'sync.responder.countFreshSwmMetaSubjectRows')); + let res; + try { + res = await store.query(` + SELECT ?s (COUNT(*) AS ?count) WHERE { + VALUES ?s { ${subjectValues(chunk)} } + GRAPH <${assertSafeIri(graph)}> { ?s ?p ?o } + } + GROUP BY ?s + `, { + ...syncResponderStoreOptions(signal, 'sync.responder.countFreshSwmMetaSubjectRows'), + maxResponseBytes: snapshotResponseByteLimit( + SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + ), + }); + } catch (error) { + if (!(error instanceof StoreResponseTooLargeError)) throw error; + throw snapshotBudgetError({ + key: budgetKey, + reason: 'snapshot_bytes', + rows: chunk.length, + bytesEstimate: storeResponseActualBytes(error), + limit: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + }); + } if (res.type !== 'bindings') continue; for (const row of res.bindings) { const subject = row['s']; @@ -2347,25 +2521,56 @@ async function countFreshSwmMetaSubjectRows( * subject window. Subjects are compareCodePoint-sorted so the plan's prefix * sums agree with the compareRows order used when window rows are sorted * in-process — no store-side ORDER BY or OFFSET is ever needed. + * + * The plan itself is bounded by construction: subject cardinality by + * {@link FRESH_SWM_META_PLAN_MAX_SUBJECTS} (enforced inside the LIMIT-bounded + * discovery), and the retained scalar estimate by the FIXED snapshot build + * byte cap — deliberately the constant, not the test/operator-shrinkable + * session budget, so shrinking the session budget forces plan-paged mode + * without ever refusing the plan that paged mode needs (#1847 class). */ async function buildFreshSwmMetaPlan( store: TripleStore, swmMetaGraphs: readonly string[], cutoffIso: string, + budgetKey: string, signal?: AbortSignal, ): Promise { const entries: FreshSwmMetaGraphPlanEntry[] = []; + let subjectAllowance = FRESH_SWM_META_PLAN_MAX_SUBJECTS; + let bytesEstimate = 0; for (const graph of dedupeStrings(swmMetaGraphs).sort(compareCodePoint)) { throwIfAborted(signal); - const admitted = await readFreshSwmMetaSubjects(store, graph, cutoffIso, signal); + const admitted = await readFreshSwmMetaSubjects( + store, + graph, + cutoffIso, + subjectAllowance, + budgetKey, + signal, + ); if (admitted.size === 0) continue; + subjectAllowance -= admitted.size; const subjects = await countFreshSwmMetaSubjectRows( store, graph, [...admitted].sort(compareCodePoint), + budgetKey, signal, ); if (subjects.length === 0) continue; + for (const entry of subjects) { + bytesEstimate += estimateStringRowHeapBytes(entry.subject, '', '', graph); + } + if (bytesEstimate > SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE) { + throw snapshotBudgetError({ + key: budgetKey, + reason: 'snapshot_bytes', + rows: subjects.length, + bytesEstimate, + limit: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, + }); + } entries.push({ graph, subjects, @@ -2375,16 +2580,40 @@ async function buildFreshSwmMetaPlan( return { entries, totalRows: entries.reduce((sum, entry) => sum + entry.rowCount, 0), + bytesEstimate, }; } +/** Order/content digest of one subject's compareRows-sorted row-group. */ +function digestSubjectRows(rows: readonly SyncRow[]): string { + const hash = sha256.create(); + const encoder = new TextEncoder(); + for (const row of rows) { + // Length-prefixed fields: literals may contain any delimiter character. + hash.update(encoder.encode(`${row.p.length}:${row.p}${row.o.length}:${row.o}`)); + } + return bytesToHex(hash.digest()); +} + /** - * Read ALL rows of a whole-subject window in bounded VALUES chunks. Each - * chunk's row total is verified against the plan counts: the plan's prefix - * sums are the pagination cursor, so a mutated subject must fail the session - * (the requester restarts with a fresh plan) rather than silently skip or - * duplicate rows — and a seal/head subject is always read atomically within - * one chunk query, so its row-group can never be torn by a chunk boundary. + * Read ALL rows of a whole-subject window in bounded VALUES chunks, verifying + * each subject's row-group against the plan two ways. The plan's prefix sums + * are the pagination cursor, so a mutated subject must fail the session (the + * requester restarts with a fresh plan) rather than silently skip, duplicate, + * or tear rows; a seal/head subject is always read atomically within one chunk + * query, so its row-group can never be torn by a chunk boundary. + * + * 1. PER-SUBJECT row count vs the plan. An aggregate count would pass when + * two subjects in one window mutate by compensating amounts, and the + * prefix-sum slice would then duplicate or skip rows at the page seam. + * 2. Content digest, bound on the subject's first window read of this + * session and verified on every reread. Counts alone pass on a same-count + * replacement, and a reread sliced at the stale prefix sums could combine + * rows of two versions of one subject across response pages. A subject + * that is never reread needs no digest: its group is served whole from a + * single query, so a same-count change before its only read serves the + * NEWER coherent group (bounded freshness skew, like any keyset pager), + * never a hybrid. */ async function readFreshSwmMetaSubjectWindowRows( store: TripleStore, @@ -2394,7 +2623,6 @@ async function readFreshSwmMetaSubjectWindowRows( ): Promise { const rows: SyncRow[] = []; for (const chunk of chunkValues(subjects, FRESH_SWM_META_PLAN_SUBJECT_CHUNK)) { - const expectedRows = chunk.reduce((sum, entry) => sum + entry.rowCount, 0); const res = await store.query(` SELECT ?s ?p ?o WHERE { VALUES ?s { ${subjectValues(chunk.map((entry) => entry.subject))} } @@ -2406,22 +2634,36 @@ async function readFreshSwmMetaSubjectWindowRows( SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, ), }); - let added = 0; + const rowsBySubject = new Map(); if (res.type === 'bindings') { for (const row of res.bindings) { const s = row['s']; const p = row['p']; const o = row['o']; if (!s || !p || !o) continue; - rows.push({ s, p, o, g: graph }); - added += 1; + const bucket = rowsBySubject.get(s) ?? []; + bucket.push({ s, p, o, g: graph }); + rowsBySubject.set(s, bucket); } } - if (added !== expectedRows) { - throw new Error( - `Shared-memory meta sync plan changed while reading ${graph}: ` + - `expected ${expectedRows} rows for ${chunk.length} subjects, found ${added}`, - ); + for (const entry of chunk) { + const subjectRows = (rowsBySubject.get(entry.subject) ?? []).sort(compareRows); + if (subjectRows.length !== entry.rowCount) { + throw new Error( + `Shared-memory meta sync plan changed while reading ${graph}: ` + + `expected ${entry.rowCount} rows for subject ${entry.subject}, found ${subjectRows.length}`, + ); + } + const digest = digestSubjectRows(subjectRows); + if (entry.contentDigest === undefined) { + entry.contentDigest = digest; + } else if (entry.contentDigest !== digest) { + throw new Error( + `Shared-memory meta sync plan changed while reading ${graph}: ` + + `subject ${entry.subject} content changed within an active session`, + ); + } + for (const row of subjectRows) rows.push(row); } } return rows.sort(compareRows); @@ -2532,11 +2774,31 @@ async function readBoundedFreshSwmMetaSnapshot( const rows: SyncRow[] = []; let bytesEstimate = 0; for (const entry of plan.entries) { - const graphRows = await readFreshSwmMetaSubjectWindowRows( - store, - entry.graph, - entry.subjects, - ); + let graphRows; + try { + graphRows = await readFreshSwmMetaSubjectWindowRows( + store, + entry.graph, + entry.subjects, + ); + } catch (error) { + // The store's response byte cap firing during SNAPSHOT materialization is + // a per-snapshot byte overflow in disguise: the admitted set is + // intrinsically too large to hold at once, so it must degrade to the + // plan-paged reader exactly like the in-process estimate crossing the + // budget — not escape untyped and fail a syncable phase outright. The + // plan-paged reader's own bounded window reads keep the store cap + // un-translated there, so a genuinely oversized single page still + // surfaces as a hard error rather than being masked. + if (!(error instanceof StoreResponseTooLargeError)) throw error; + throw snapshotBudgetError({ + key: cache.key, + reason: 'snapshot_bytes', + rows: rows.length, + bytesEstimate: bytesEstimate + storeResponseActualBytes(error), + limit: limits.maxBytesEstimate, + }); + } for (const row of graphRows) { const nextBytes = bytesEstimate + estimateStringRowHeapBytes(row.s, row.p, row.o, row.g); if (nextBytes > limits.maxBytesEstimate) { diff --git a/packages/agent/src/sync/responder/snapshot-budget.ts b/packages/agent/src/sync/responder/snapshot-budget.ts index c9f21e6e58..e992ae9791 100644 --- a/packages/agent/src/sync/responder/snapshot-budget.ts +++ b/packages/agent/src/sync/responder/snapshot-budget.ts @@ -28,6 +28,15 @@ type SnapshotBudgetAdmission = Omit & { key: string; /** Existing entry replaced atomically after the new entry passes admission. */ replaceId?: symbol; + /** + * Control-plane entries (session pagination plans) are bounded at build time + * by their own FIXED construction caps, deliberately not by the + * operator/test-shrinkable per-snapshot limits: shrinking the per-snapshot + * budget is how a session is forced into plan-paged mode, and rejecting the + * plan itself there would turn that degradation into a refusal (#1847 + * class). Only the GLOBAL rows/bytes budget applies at admission. + */ + controlPlane?: boolean; }; export class SyncRowSnapshotBudgetError extends Error { @@ -137,11 +146,13 @@ export function createSyncResponderSnapshotBudget( return { admit(params) { - if (params.rows > limits.maxSnapshotRows) { - reject(params, 'snapshot_rows', limits.maxSnapshotRows); - } - if (params.bytesEstimate > limits.maxSnapshotBytesEstimate) { - reject(params, 'snapshot_bytes', limits.maxSnapshotBytesEstimate); + if (!params.controlPlane) { + if (params.rows > limits.maxSnapshotRows) { + reject(params, 'snapshot_rows', limits.maxSnapshotRows); + } + if (params.bytesEstimate > limits.maxSnapshotBytesEstimate) { + reject(params, 'snapshot_bytes', limits.maxSnapshotBytesEstimate); + } } const replaced = params.replaceId ? entries.get(params.replaceId) : undefined; diff --git a/packages/agent/src/sync/responder/sync-handler.ts b/packages/agent/src/sync/responder/sync-handler.ts index 23ab582e8b..8ea8e5a777 100644 --- a/packages/agent/src/sync/responder/sync-handler.ts +++ b/packages/agent/src/sync/responder/sync-handler.ts @@ -449,6 +449,10 @@ export function registerSyncHandler(params: RegisterSyncHandlerParams): void { const freshSwmMetaPlanMemo = createResponderFreshSwmMetaPlanMemo( DURABLE_DATA_SYNC_SESSION_TTL_MS, SYNC_RESPONDER_SHARED_MEMORY_SNAPSHOT_LIMIT, + // #1847 review: retained TTL meta session plans are control-plane state and + // must be charged to the same process-wide budget as retained snapshots — + // peers cannot stack uncharged plans, and global pressure evicts idle ones. + responderSnapshotBudget, ); const durableDataExactGraphPlanMemo = createResponderExactGraphPagePlanMemo( DURABLE_DATA_SYNC_SESSION_TTL_MS, diff --git a/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts b/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts index e70c1429c3..bbe7a2f1d2 100644 --- a/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts +++ b/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts @@ -1,6 +1,15 @@ import { describe, it, expect } from 'vitest'; import type { OperationContext } from '@origintrail-official/dkg-core'; -import { OxigraphStore, type Quad } from '@origintrail-official/dkg-storage'; +import { + OxigraphStore, + StoreResponseTooLargeError, + type Quad, +} from '@origintrail-official/dkg-storage'; +import { createSyncResponderSnapshotBudget } from '../src/sync/responder/snapshot-budget.js'; +import { + createResponderFreshSwmMetaPlanMemo, + FRESH_SWM_META_PLAN_MAX_SUBJECTS, +} from '../src/sync/responder/graph-plan.js'; import { DKG_NS, RDF_TYPE, @@ -304,6 +313,255 @@ describe('SWM meta lane above the 64,000-row snapshot ceiling (#1847)', () => { await store.close(); }); + it('fails the session when an already-served subject is replaced with the SAME row count (content binding, not just cardinality)', async () => { + const cgId = 'meta-ceiling-samecount'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + const opIds = ['a', 'b', 'c', 'd', 'e', 'f']; + for (const opId of opIds) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:m:${opId}`, metaGraph, fresh)); + } + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + // Page size 2 splits the 5-row subject `a` across pages, so page 1 must + // REREAD `a` and slice it at the plan's prefix sums — the exact shape + // that used to accept a same-count replacement and serve a hybrid + // row-group assembled from two versions of one subject. + syncPageSize: 2, + snapshotBudget: TINY_SNAPSHOT_BUDGET, // forces plan-paged mode + }); + const base = { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta' as const, limit: 2 }; + + const page0 = linesFromNquads(await cap.invoke({ ...base, offset: 0, syncSessionId: 'SC1' })); + expect(page0).toHaveLength(2); + + // Same-count replacement of the split subject: 5 rows before, 5 rows after. + const splitSubject = `urn:dkg:share:${cgId}:a`; + await store.delete([ + { graph: metaGraph, subject: splitSubject, predicate: `${DKG_NS}rootEntity`, object: 'urn:m:a' }, + ]); + await store.insert([ + { graph: metaGraph, subject: splitSubject, predicate: `${DKG_NS}note`, object: '"swapped"' }, + ]); + + await expect(cap.invoke({ ...base, offset: 2, syncSessionId: 'SC1' })) + .rejects.toThrow(/Shared-memory meta sync plan changed while reading/); + + // A fresh session rebuilds the plan and serves the replaced content whole. + const recovered = await collectAllPages(cap, { ...base, syncSessionId: 'SC2' }, 2); + expect(recovered.lines.size).toBe(6 * 5); + const joined = [...recovered.lines].join('\n'); + expect(joined).toContain('"swapped"'); + expect(joined).not.toContain(`<${splitSubject}> <${DKG_NS}rootEntity>`); + await store.close(); + }); + + it('serves a coherent NEW row-group when a NOT-yet-read subject mutates same-count (bounded freshness skew, never a tear)', async () => { + // Guarantee boundary, made explicit per review: whole-subject row-groups + // are the consistency unit. A subject read exactly once is served whole + // from a single query, so a same-count change BEFORE its only read serves + // the newer coherent group — the bounded skew any keyset pager has. Only a + // REREAD of a split subject binds (and verifies) content. + const cgId = 'meta-ceiling-skew'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + const opIds = ['a', 'b', 'c', 'd', 'e', 'f']; + for (const opId of opIds) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:m:${opId}`, metaGraph, fresh)); + } + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 5, // window = exactly one whole 5-row subject + snapshotBudget: TINY_SNAPSHOT_BUDGET, + }); + const base = { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta' as const, limit: 5 }; + + const page0 = linesFromNquads(await cap.invoke({ ...base, offset: 0, syncSessionId: 'SK1' })); + expect(page0).toHaveLength(5); + + // Same-count mutation of subject `b`, which page 1 will read for the FIRST time. + const nextSubject = `urn:dkg:share:${cgId}:b`; + await store.delete([ + { graph: metaGraph, subject: nextSubject, predicate: `${DKG_NS}rootEntity`, object: 'urn:m:b' }, + ]); + await store.insert([ + { graph: metaGraph, subject: nextSubject, predicate: `${DKG_NS}note`, object: '"swapped-whole"' }, + ]); + + const page1 = linesFromNquads(await cap.invoke({ ...base, offset: 5, syncSessionId: 'SK1' })); + expect(page1).toHaveLength(5); + const joined = page1.join('\n'); + // The NEW group, whole: replacement present, replaced row absent — no hybrid. + expect(joined).toContain('"swapped-whole"'); + expect(joined).not.toContain(`<${nextSubject}> <${DKG_NS}rootEntity>`); + expect(page1.every((line) => line.startsWith(`<${nextSubject}>`))).toBe(true); + await store.close(); + }); + + it('fails the session on a compensating cross-subject count mutation within one window (per-subject counts, not the window aggregate)', async () => { + const cgId = 'meta-ceiling-compensate'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + for (const opId of ['a', 'b', 'c', 'd']) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:m:${opId}`, metaGraph, fresh)); + } + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 10, + snapshotBudget: TINY_SNAPSHOT_BUDGET, + }); + const base = { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta' as const, limit: 10 }; + + // Page 0 = subjects a+b whole. Page 1's window will be subjects c+d. + const page0 = linesFromNquads(await cap.invoke({ ...base, offset: 0, syncSessionId: 'CP1' })); + expect(page0).toHaveLength(10); + + // c loses a row, d gains one: the WINDOW aggregate still totals 10, but the + // plan's prefix sums for c/d are now both wrong — an aggregate-count guard + // passes and misaligns every later slice (duplicate/skip at page seams). + await store.delete([ + { graph: metaGraph, subject: `urn:dkg:share:${cgId}:c`, predicate: `${DKG_NS}rootEntity`, object: 'urn:m:c' }, + ]); + await store.insert([ + { graph: metaGraph, subject: `urn:dkg:share:${cgId}:d`, predicate: `${DKG_NS}note`, object: '"extra"' }, + ]); + + await expect(cap.invoke({ ...base, offset: 10, syncSessionId: 'CP1' })) + .rejects.toThrow(/Shared-memory meta sync plan changed while reading/); + await store.close(); + }); + + it('degrades to plan paging when the STORE response byte cap fires during snapshot materialization (#1868 review: untyped escape)', async () => { + const cgId = 'meta-ceiling-storecap'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + const quads: Quad[] = []; + for (let index = 0; index < 40; index += 1) { + quads.push(...workspaceOpQuads(cgId, `op-${String(index).padStart(2, '0')}`, `urn:sc:${index}`, metaGraph, fresh)); + } + await store.insert(quads); + + // Emulate the storage layer's 32 MiB response cap: any whole-subject + // window query addressing MANY subjects at once (the snapshot + // materialization) throws StoreResponseTooLargeError, while the paged + // lane's small windows stay under the cap. Before the fix this error + // escaped untyped past the per-snapshot budget accounting and failed the + // phase outright instead of falling back. + let capThrows = 0; + const originalQuery = store.query.bind(store); + store.query = (async (sparql: string, options?: unknown) => { + const normalized = sparql.replace(/\s+/g, ' '); + if (normalized.includes('VALUES ?s') && !normalized.includes('COUNT(')) { + const subjectCount = (normalized.match(/ 10) { + capThrows += 1; + throw new StoreResponseTooLargeError(1024, 2048); + } + } + return originalQuery(sparql, options as never); + }) as OxigraphStore['query']; + + // DEFAULT budgets: the snapshot lane is attempted first and must degrade. + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 7 }); + const { lines } = await collectAllPages( + cap, + { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta', limit: 7, syncSessionId: 'storecap-session' }, + 7, + ); + expect(lines.size).toBe(200); + expect(capThrows).toBeGreaterThan(0); + await store.close(); + }); + + it('degrades to plan paging when the fresh snapshot crosses only the per-snapshot BYTE estimate budget', async () => { + const cgId = 'meta-ceiling-bytebudget'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + for (const opId of ['a', 'b', 'c', 'd', 'e', 'f']) { + await store.insert(workspaceOpQuads(cgId, opId, `urn:m:${opId}`, metaGraph, fresh)); + } + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 4, + snapshotBudget: { + maxRows: 1_000_000, + maxBytesEstimate: Number.MAX_SAFE_INTEGER, + maxSnapshotRows: 1_000_000, + // Well below one row's ~200-byte heap estimate: the snapshot path must + // throw the per-snapshot BYTES error (row budget never binds) and the + // session must still complete through the plan-paged reader. + maxSnapshotBytesEstimate: 64, + }, + }); + const watch = forbidSwmMetaSortOrOffsetQueries(store); + const { lines } = await collectAllPages( + cap, + { contextGraphId: cgId, includeSharedMemory: true, phase: 'meta', limit: 4, syncSessionId: 'bytebudget-session' }, + 4, + ); + expect(lines.size).toBe(30); + watch.assertWindowQueriesObserved(); + await store.close(); + }); + + it('refuses a fresh subject set beyond the plan cardinality cap as a TYPED bounded refusal, via LIMIT-bounded discovery', async () => { + const cgId = 'meta-ceiling-cardinality'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + // One row per subject: cap + 1 admitted subjects. The plan would retain a + // subject entry for every one of them — this is the reviewed unbounded + // control-plane growth (#1868), so it must refuse, bounded and typed, + // BEFORE materializing an unbounded discovery result. + const quads: Quad[] = []; + for (let index = 0; index <= FRESH_SWM_META_PLAN_MAX_SUBJECTS; index += 1) { + quads.push({ + graph: metaGraph, + subject: `urn:card:${String(index).padStart(6, '0')}`, + predicate: `${DKG_NS}publishedAt`, + object: `"${fresh}"^^<${XSD_DT}>`, + }); + } + await insertChunked(store, quads); + + // Bounded-by-construction: every TTL discovery query over the meta graph + // must carry the cap-derived LIMIT so the store can never stream an + // unbounded subject set into the plan builder. + let discoveryLimitQueries = 0; + const originalQuery = store.query.bind(store); + store.query = (async (sparql: string, options?: unknown) => { + const normalized = sparql.replace(/\s+/g, ' ').trim(); + if (normalized.includes('SELECT DISTINCT ?s') && normalized.includes('_shared_memory_meta')) { + expect(normalized).toMatch(/LIMIT \d+$/); + if (normalized.endsWith(`LIMIT ${FRESH_SWM_META_PLAN_MAX_SUBJECTS + 1}`)) { + discoveryLimitQueries += 1; + } + } + return originalQuery(sparql, options as never); + }) as OxigraphStore['query']; + + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 500 }); + await expect(cap.invoke({ + contextGraphId: cgId, + includeSharedMemory: true, + phase: 'meta', + offset: 0, + limit: 500, + syncSessionId: 'cardinality-session', + })).rejects.toThrow(/per-snapshot rows budget/); + expect(discoveryLimitQueries).toBeGreaterThan(0); + await store.close(); + }, 120_000); + it('keeps a bounded refusal ONLY for a single pathological subject exceeding the hard 64,000-row build cap', async () => { const cgId = 'meta-ceiling-monster'; const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; @@ -384,6 +642,94 @@ describe('SWM meta lane above the 64,000-row snapshot ceiling (#1847)', () => { }); }); +describe('TTL meta session plans are charged to the responder snapshot budget (#1868 review)', () => { + const plan = (bytesEstimate: number) => ({ entries: [], totalRows: 0, bytesEstimate }); + + it('admits, LRU-evicts and rejects plans via the GLOBAL budget while exempting them from per-snapshot caps', async () => { + const budget = createSyncResponderSnapshotBudget({ + maxRows: 1_000, + maxBytesEstimate: 10_000, + // Deliberately tiny per-snapshot caps: plans are control-plane entries + // bounded by their own fixed construction caps, so per-snapshot limits + // must NOT reject them (shrinking those limits is how a session is + // forced into the plan-paged mode that NEEDS the plan). + maxSnapshotRows: 1, + maxSnapshotBytesEstimate: 1, + }); + const memo = createResponderFreshSwmMetaPlanMemo(60_000, 8, budget); + + await memo.get('k1', async () => plan(4_000)); + expect(budget.stats().snapshots).toBe(1); + expect(budget.stats().bytesEstimate).toBe(4_000); + + await memo.get('k2', async () => plan(4_000)); + expect(budget.stats().snapshots).toBe(2); + + // Global pressure: admitting k3 must evict the least-recently-used idle + // plan (k1) rather than growing past the global byte budget. + await memo.get('k3', async () => plan(4_000)); + expect(budget.stats().bytesEstimate).toBe(8_000); + expect(await memo.get('k1', async () => plan(1), { requireExisting: true })).toBeNull(); + + // A plan that cannot fit even after draining evictables is a typed + // global rejection (the requester's quiet retryable limit), never an + // uncharged retention. + await expect(memo.get('kX', async () => plan(50_000))) + .rejects.toThrow(/global estimated bytes budget/); + }); + + it('memo eviction and TTL expiry release the charged bytes', async () => { + const budget = createSyncResponderSnapshotBudget({ + maxRows: 1_000, + maxBytesEstimate: 100_000, + maxSnapshotRows: 1, + maxSnapshotBytesEstimate: 1, + }); + const memo = createResponderFreshSwmMetaPlanMemo(60_000, 2, budget); + await memo.get('a', async () => plan(1_000)); + await memo.get('b', async () => plan(1_000)); + expect(budget.stats().bytesEstimate).toBe(2_000); + // maxEntries=2: inserting c evicts the memo's oldest entry AND its charge. + await memo.get('c', async () => plan(1_000)); + expect(budget.stats().snapshots).toBe(2); + expect(budget.stats().bytesEstimate).toBe(2_000); + }); + + it('the sync handler wires the responder budget through to plan admission', async () => { + const cgId = 'meta-plan-budget-wire'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + await store.insert(workspaceOpQuads(cgId, 'a', 'urn:w:a', metaGraph, fresh)); + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 5, + snapshotBudget: { + maxRows: 1_000_000, + // Global byte budget below even one plan's scalar estimate: PLAN + // admission must fail typed through the handler (proving + // registerSyncHandler passes its budget into the meta plan memo, not + // an uncharged default). maxSnapshotRows=1 keeps the ROW snapshot on + // its memoized per-snapshot refusal so it never reaches the global + // budget itself — the plan memo is the only global-budget client here. + maxBytesEstimate: 100, + maxSnapshotRows: 1, + maxSnapshotBytesEstimate: Number.MAX_SAFE_INTEGER, + }, + }); + await expect(cap.invoke({ + contextGraphId: cgId, + includeSharedMemory: true, + phase: 'meta', + offset: 0, + limit: 5, + syncSessionId: 'plan-budget-wire', + })).rejects.toThrow(/global estimated bytes budget/); + await store.close(); + }); +}); + describe('requester reassembly of the plan-paged SWM meta lane (#1847 x #1788)', () => { function makeCtx(): OperationContext { return { kind: 'system', id: 'meta-ceiling-requester', startedAt: Date.now() } as never; @@ -407,7 +753,9 @@ describe('requester reassembly of the plan-paged SWM meta lane (#1847 x #1788)', cap: CapturedSyncHandler, cgId: string, pageSize: number, + afterPage?: (pagesServed: number) => Promise, ) { + let pagesServed = 0; return fetchSyncPages({ ctx: makeCtx(), remotePeerId: '12D3KooWMetaCeilingRemote', @@ -434,7 +782,10 @@ describe('requester reassembly of the plan-paged SWM meta lane (#1847 x #1788)', }, send: async (_peerId, _protocolId, data) => { const envelope = JSON.parse(new TextDecoder().decode(data)) as SyncRequestEnvelope; - return new TextEncoder().encode(await cap.invoke(envelope)); + const out = await cap.invoke(envelope); + pagesServed += 1; + await afterPage?.(pagesServed); + return new TextEncoder().encode(out); }, logWarn: noop, logInfo: noop, @@ -516,4 +867,51 @@ describe('requester reassembly of the plan-paged SWM meta lane (#1847 x #1788)', await store.close(); }); } + + it('never completes with a hybrid row-group when a split subject is replaced same-count mid-session (#1868 review repro)', async () => { + // lupuszr's reproduction shape: ONE five-row operation, page size 1, a + // same-count replacement between pages. A count-only guard accepted the + // reread and assembled a five-row hybrid of both versions (omitting + // publishedAt); the content binding must fail the session instead, and the + // requester must never report a completed phase carrying the hybrid. + const cgId = 'meta-samecount-requester'; + const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + await store.insert(workspaceOpQuads(cgId, 'solo', 'urn:sq:solo', metaGraph, fresh)); + const subject = `urn:dkg:share:${cgId}:solo`; + + const cap = registerTestSyncHandler(store, { + sharedMemoryTtlMs: TTL_MS, + syncPageSize: 1, + snapshotBudget: TINY_SNAPSHOT_BUDGET, // plan-paged mode: every page rereads the subject + }); + + const mutateAfterFirstPage = async (pagesServed: number) => { + if (pagesServed !== 1) return; + await store.delete([ + { graph: metaGraph, subject, predicate: `${DKG_NS}publishedAt`, object: `"${fresh}"^^` }, + ]); + await store.insert([ + { graph: metaGraph, subject, predicate: `${DKG_NS}note`, object: '"replacement"' }, + ]); + }; + + let threw = false; + let result: Awaited> | undefined; + try { + result = await fetchAllMeta(cap, cgId, 1, mutateAfterFirstPage); + } catch { + threw = true; + } + if (!threw) { + expect(result!.completed).toBe(false); + } + // Whatever partial rows the requester holds, they must not mix versions: + // the pre-mutation publishedAt row and the post-mutation replacement row + // can never coexist in one assembled row-group. + const objects = (result?.quads ?? []).map((quad) => quad.object).join('\n'); + expect(objects.includes('"replacement"') && objects.includes(fresh)).toBe(false); + await store.close(); + }); }); From a92e14c30fe26dc850c727ebf28b94509dbe255d Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Mon, 20 Jul 2026 20:44:09 +0200 Subject: [PATCH 3/4] refactor(sync): extract the shared consume-once session-plan getter (#1868 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Partial take on the plan-orchestration collapse suggested in review: the exact-graph and TTL SWM meta lanes now share one createSessionPlanGetter owning refresh consumption, offset>0 require-existing, and expiry translation — the lifecycle most likely to drift between lanes. The full collapse (snapshot + fallback wiring) is deferred; those parts differ by lane for reviewed reasons and are slated for the graph-plan module split. Co-Authored-By: Claude Fable 5 --- .../agent/src/sync/responder/graph-plan.ts | 111 ++++++++++-------- 1 file changed, 61 insertions(+), 50 deletions(-) diff --git a/packages/agent/src/sync/responder/graph-plan.ts b/packages/agent/src/sync/responder/graph-plan.ts index f98a6f2116..042d936f02 100644 --- a/packages/agent/src/sync/responder/graph-plan.ts +++ b/packages/agent/src/sync/responder/graph-plan.ts @@ -635,35 +635,19 @@ export async function readSwmMetaPage(params: { // pages from the same plan instead of failing permanently. const cutoffIso = params.cutoffIso; const budgetKey = cache?.key ?? `swm-meta:${params.contextGraphId}`; - // Consume the explicit session refresh once: when the snapshot build crosses - // its budget, the immediate page-zero fallback must reuse the just-built plan - // instead of rebuilding (and re-counting) it against a moving store. - let planRefreshPending = params.refreshRowList === true; - const getPlan = async ( - pageOffset: number, - pageSignal: AbortSignal | undefined, - ): Promise => { - const loadPlan = () => buildFreshSwmMetaPlan( + const getPlan = createSessionPlanGetter( + params.freshMetaPlanMemo, + params.rowListCacheKey, + params.refreshRowList === true, + (signal) => buildFreshSwmMetaPlan( params.store, candidateGraphs, cutoffIso, budgetKey, - pageSignal, - ); - const refreshPlan = pageOffset === 0 && planRefreshPending; - if (refreshPlan) planRefreshPending = false; - const plan = params.freshMetaPlanMemo && params.rowListCacheKey - ? await params.freshMetaPlanMemo.get(params.rowListCacheKey, loadPlan, { - refresh: refreshPlan, - requireExisting: pageOffset > 0, - signal: pageSignal, - }) - : await loadPlan(); - if (!plan) { - throw new Error('Shared-memory meta sync session graph plan expired before page completion'); - } - return plan; - }; + signal, + ), + 'Shared-memory meta sync session graph plan expired before page completion', + ); const loadStoreBoundedPage: StorePageLoader = async (offset, limit, signal) => readFreshSwmMetaRowsPageFromPlan( params.store, @@ -1463,36 +1447,18 @@ async function readPagedRowsFromExactGraphPlanLoader( planMemo: ExactGraphPagePlanMemo | undefined, loadExactGraphPlan: (signal?: AbortSignal) => Promise, ): Promise { - // A small snapshot is first assembled into the row cache. If that build - // crosses its cap, readResponderRowsPage immediately retries page zero via - // the store-bounded path. Consume the explicit session refresh only once so - // that fallback reuses the exact graph/count plan instead of counting every - // graph twice. - let planRefreshPending = cache?.refresh === true; const rowSnapshotLimits = cache?.memo.snapshotLoadLimits ?? { maxRows: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_ROWS, maxBytesEstimate: SYNC_RESPONDER_SNAPSHOT_BUILD_MAX_BYTES_ESTIMATE, pageRows: SYNC_RESPONDER_SNAPSHOT_BUILD_PAGE_ROWS, }; - const getPlan = async ( - pageOffset: number, - pageSignal: AbortSignal | undefined, - ): Promise => { - const loadPlan = () => loadExactGraphPlan(pageSignal); - const refreshPlan = pageOffset === 0 && planRefreshPending; - if (refreshPlan) planRefreshPending = false; - const plan = planMemo && cache - ? await planMemo.get(cache.key, loadPlan, { - refresh: refreshPlan, - requireExisting: pageOffset > 0, - signal: pageSignal, - }) - : await loadPlan(); - if (!plan) { - throw new Error('Sync session exact-graph plan expired before page completion'); - } - return plan; - }; + const getPlan = createSessionPlanGetter( + planMemo, + cache?.key, + cache?.refresh === true, + (planSignal) => loadExactGraphPlan(planSignal), + 'Sync session exact-graph plan expired before page completion', + ); const loadPage: StorePageLoader = async (pageOffset, pageLimit, pageSignal) => { const plan = await getPlan(pageOffset, pageSignal); return readRowsPageFromExactGraphPlan( @@ -1792,6 +1758,51 @@ function isPerSnapshotBudgetError(error: unknown): error is SyncRowSnapshotBudge (error.reason === 'snapshot_rows' || error.reason === 'snapshot_bytes'); } +/** + * Session-plan getter shared by the plan-backed lanes (exact-graph and TTL SWM + * meta), owning the one lifecycle both must agree on: + * + * - the explicit session refresh is consumed exactly ONCE, so when a snapshot + * build crosses its budget, the immediate page-zero fallback reuses the + * just-built plan instead of rebuilding (and re-counting) it against a + * moving store; + * - offset>0 REQUIRES the existing plan — silently rebuilding against moved + * data would make the numeric offset skip or duplicate rows; + * - memo expiry becomes the lane's session-expired error. + * + * The SWM data lane intentionally does not use this helper: it has no snapshot + * lane, so a single plan access per page means per-call refresh semantics are + * equivalent and simpler there. + */ +function createSessionPlanGetter( + memo: { + get( + key: string, + load: () => Promise, + options?: { refresh?: boolean; requireExisting?: boolean; signal?: AbortSignal }, + ): Promise; + } | undefined, + cacheKey: string | undefined, + initialRefreshPending: boolean, + loadPlan: (signal?: AbortSignal) => Promise, + expiredMessage: string, +): (pageOffset: number, pageSignal: AbortSignal | undefined) => Promise { + let planRefreshPending = initialRefreshPending; + return async (pageOffset, pageSignal) => { + const refreshPlan = pageOffset === 0 && planRefreshPending; + if (refreshPlan) planRefreshPending = false; + const plan = memo && cacheKey + ? await memo.get(cacheKey, () => loadPlan(pageSignal), { + refresh: refreshPlan, + requireExisting: pageOffset > 0, + signal: pageSignal, + }) + : await loadPlan(pageSignal); + if (!plan) throw new Error(expiredMessage); + return plan; + }; +} + /** * Serve one responder page, owning the single budget-fallback policy for every * memoized phase. It tries the stable-snapshot cache first, but an From 0022688c65d1ef631c0c48a5786d3a9f6a332a26 Mon Sep 17 00:00:00 2001 From: Branimir Rakic Date: Tue, 21 Jul 2026 00:27:10 +0200 Subject: [PATCH 4/4] refactor(sync): move per-session digest bindings out of the immutable TTL meta plan (#1868 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FreshSwmMetaPlan is now a genuinely immutable pagination description (deep-readonly graph/subject/count scalars). The mutable per-session content-digest state that used to live on subject entries moves to a sidecar WeakMap keyed by plan instance — exactly the binding's intended lifetime: the memoized plan IS the session, a refreshed/rebuilt plan is a new object with a fresh empty binding map, and evicting or expiring the plan releases its digests with it. readFreshSwmMetaSubjectWindowRows is the only writer. Placement only: same-count replacement failure semantics are unchanged and every existing mutation test passes unmodified. Also closes the two remaining #1868 round-2 coverage asks on the plan budget thread, each proven by a killed mutant: - time-based TTL expiry (controlled clock) prunes a plan AND releases its global budget charge, distinct from the maxEntries eviction the prior test covered — a mutant that leaks the charge on expiry passes the old test and is killed only by the new one; - the plan cardinality cap binds in AGGREGATE across root and subgraph meta graphs — a mutant that resets the allowance per graph passes the single-graph cap test and is killed only by the new multi-graph test. Co-Authored-By: Claude Fable 5 --- .../agent/src/sync/responder/graph-plan.ts | 79 +++++++++++++----- .../sync-responder-swm-meta-ceiling.test.ts | 81 ++++++++++++++++++- 2 files changed, 137 insertions(+), 23 deletions(-) diff --git a/packages/agent/src/sync/responder/graph-plan.ts b/packages/agent/src/sync/responder/graph-plan.ts index 042d936f02..25290ec898 100644 --- a/packages/agent/src/sync/responder/graph-plan.ts +++ b/packages/agent/src/sync/responder/graph-plan.ts @@ -105,26 +105,15 @@ export interface FreshSwmDataGraphPlanMemo { } interface FreshSwmMetaSubjectEntry { - subject: string; - rowCount: number; - /** - * Content binding for the subject's whole row-group, established on the - * FIRST window read of this session and verified on every REREAD. Row counts - * alone pass on same-count replacements, and a reread sliced at the plan's - * prefix sums could then combine rows of two different versions of one - * subject across response pages; the digest makes any content or ordering - * change of an already-served subject fail the session instead (the - * requester restarts with a fresh plan). A subject read exactly once needs - * no binding: its row-group is served whole from a single query. - */ - contentDigest?: string; + readonly subject: string; + readonly rowCount: number; } interface FreshSwmMetaGraphPlanEntry { - graph: string; + readonly graph: string; /** TTL-admitted subjects, compareCodePoint-sorted; row counts are exact at plan build. */ - subjects: readonly FreshSwmMetaSubjectEntry[]; - rowCount: number; + readonly subjects: readonly FreshSwmMetaSubjectEntry[]; + readonly rowCount: number; } /** @@ -136,12 +125,47 @@ interface FreshSwmMetaGraphPlanEntry { * cap, so plan building can never materialize an unbounded store result. The * retained estimate is additionally charged to the process-wide responder * snapshot budget by the memo (see createResponderFreshSwmMetaPlanMemo). + * + * The plan is IMMUTABLE once built — every reader treats it as a frozen + * pagination description. The mutable per-session content-digest bindings that + * used to live on subject entries are held in a sidecar keyed by plan instance + * (see {@link sessionDigestBindingsFor}), so nothing that "reads a plan" can + * change it. */ interface FreshSwmMetaPlan { - entries: readonly FreshSwmMetaGraphPlanEntry[]; - totalRows: number; + readonly entries: readonly FreshSwmMetaGraphPlanEntry[]; + readonly totalRows: number; /** Estimated retained heap bytes of the plan's subject/count scalars. */ - bytesEstimate: number; + readonly bytesEstimate: number; +} + +/** + * Sidecar for the mutable per-session digest state of a TTL meta plan (#1868 + * review): content bindings for whole subject row-groups, established on a + * subject's FIRST window read of the session and verified on every REREAD. Row + * counts alone pass on same-count replacements, and a reread sliced at the + * plan's prefix sums could then combine rows of two different versions of one + * subject across response pages; the digest makes any content or ordering + * change of an already-served subject fail the session instead (the requester + * restarts with a fresh plan). A subject read exactly once needs no binding: + * its row-group is served whole from a single query. + * + * Keyed WEAKLY by plan object identity, which is exactly the binding's + * intended lifetime: the memoized plan IS the session (offset>0 requires the + * existing plan; refresh/rebuild produces a NEW plan object and therefore a + * fresh, empty binding map), and evicting or expiring the plan releases its + * digests with it. Map keys are `graph U+0000 subject` (NUL cannot appear + * in an IRI, so the composite key cannot collide). + */ +const freshSwmMetaSessionDigests = new WeakMap>(); + +function sessionDigestBindingsFor(plan: FreshSwmMetaPlan): Map { + let bindings = freshSwmMetaSessionDigests.get(plan); + if (!bindings) { + bindings = new Map(); + freshSwmMetaSessionDigests.set(plan, bindings); + } + return bindings; } export interface FreshSwmMetaPlanMemo { @@ -2625,11 +2649,16 @@ function digestSubjectRows(rows: readonly SyncRow[]): string { * single query, so a same-count change before its only read serves the * NEWER coherent group (bounded freshness skew, like any keyset pager), * never a hybrid. + * + * `digestBindings` is the plan's session sidecar (see + * {@link sessionDigestBindingsFor}); this reader is the only writer to it, and + * the plan itself is never mutated. */ async function readFreshSwmMetaSubjectWindowRows( store: TripleStore, graph: string, subjects: readonly FreshSwmMetaSubjectEntry[], + digestBindings: Map, signal?: AbortSignal, ): Promise { const rows: SyncRow[] = []; @@ -2666,9 +2695,11 @@ async function readFreshSwmMetaSubjectWindowRows( ); } const digest = digestSubjectRows(subjectRows); - if (entry.contentDigest === undefined) { - entry.contentDigest = digest; - } else if (entry.contentDigest !== digest) { + const digestKey = `${graph}\u0000${entry.subject}`; + const boundDigest = digestBindings.get(digestKey); + if (boundDigest === undefined) { + digestBindings.set(digestKey, digest); + } else if (boundDigest !== digest) { throw new Error( `Shared-memory meta sync plan changed while reading ${graph}: ` + `subject ${entry.subject} content changed within an active session`, @@ -2703,6 +2734,7 @@ async function readFreshSwmMetaRowsPageFromPlan( let skip = Math.max(0, Math.floor(offset)); let remaining = Math.max(0, Math.floor(limit)); if (remaining === 0 || skip >= plan.totalRows) return []; + const digestBindings = sessionDigestBindingsFor(plan); const rows: SyncRow[] = []; for (const entry of plan.entries) { if (skip >= entry.rowCount) { @@ -2741,6 +2773,7 @@ async function readFreshSwmMetaRowsPageFromPlan( store, entry.graph, window, + digestBindings, signal, ); const page = windowRowsRead.slice(skip - windowStart, skip - windowStart + remaining); @@ -2784,6 +2817,7 @@ async function readBoundedFreshSwmMetaSnapshot( } const rows: SyncRow[] = []; let bytesEstimate = 0; + const digestBindings = sessionDigestBindingsFor(plan); for (const entry of plan.entries) { let graphRows; try { @@ -2791,6 +2825,7 @@ async function readBoundedFreshSwmMetaSnapshot( store, entry.graph, entry.subjects, + digestBindings, ); } catch (error) { // The store's response byte cap firing during SNAPSHOT materialization is diff --git a/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts b/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts index bbe7a2f1d2..5f0aebf3e9 100644 --- a/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts +++ b/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect } from 'vitest'; +import { describe, it, expect, vi } from 'vitest'; import type { OperationContext } from '@origintrail-official/dkg-core'; import { OxigraphStore, @@ -562,6 +562,48 @@ describe('SWM meta lane above the 64,000-row snapshot ceiling (#1847)', () => { await store.close(); }, 120_000); + it('applies the plan cardinality cap in AGGREGATE across root and subgraph meta graphs, not per graph (#1868 review)', async () => { + const cgId = 'meta-ceiling-aggregate'; + const cgPrefix = `did:dkg:context-graph:${cgId}`; + const rootMeta = `${cgPrefix}/_shared_memory_meta`; + const subMeta = `${cgPrefix}/subagg/_shared_memory_meta`; + const fresh = freshIso(); + const store = new OxigraphStore(); + // Exactly the cap in the ROOT bucket plus ONE more fresh subject in a + // registered subgraph bucket. The subject allowance is cumulative across + // the phase's candidate graphs; a regression that reset it per graph would + // happily admit both buckets (retaining up to #graphs x cap plan entries) + // and serve this session — so it must fail this test, which demands the + // same typed bounded refusal as the single-graph overflow. + const quads: Quad[] = [...subGraphRegistrationQuads(cgId, 'subagg')]; + for (let index = 0; index < FRESH_SWM_META_PLAN_MAX_SUBJECTS; index += 1) { + quads.push({ + graph: rootMeta, + subject: `urn:agg:${String(index).padStart(6, '0')}`, + predicate: `${DKG_NS}publishedAt`, + object: `"${fresh}"^^<${XSD_DT}>`, + }); + } + quads.push({ + graph: subMeta, + subject: 'urn:agg:one-over-in-the-subgraph', + predicate: `${DKG_NS}publishedAt`, + object: `"${fresh}"^^<${XSD_DT}>`, + }); + await insertChunked(store, quads); + + const cap = registerTestSyncHandler(store, { sharedMemoryTtlMs: TTL_MS, syncPageSize: 500 }); + await expect(cap.invoke({ + contextGraphId: cgId, + includeSharedMemory: true, + phase: 'meta', + offset: 0, + limit: 500, + syncSessionId: 'aggregate-cap-session', + })).rejects.toThrow(/per-snapshot rows budget/); + await store.close(); + }, 120_000); + it('keeps a bounded refusal ONLY for a single pathological subject exceeding the hard 64,000-row build cap', async () => { const cgId = 'meta-ceiling-monster'; const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`; @@ -695,6 +737,43 @@ describe('TTL meta session plans are charged to the responder snapshot budget (# expect(budget.stats().bytesEstimate).toBe(2_000); }); + it('time-based TTL expiry prunes a plan AND releases its global charge, distinct from maxEntries eviction (#1868 review)', async () => { + const budget = createSyncResponderSnapshotBudget({ + maxRows: 1_000, + maxBytesEstimate: 100_000, + maxSnapshotRows: 1, + maxSnapshotBytesEstimate: 1, + }); + // maxEntries is deliberately roomy so ONLY the clock can remove entries: a + // regression in the time-based prune path ('expired') cannot hide behind + // the LRU/maxEntries eviction the previous test already proves. + const memo = createResponderFreshSwmMetaPlanMemo(60_000, 8, budget); + const nowSpy = vi.spyOn(Date, 'now'); + const epoch = 1_800_000_000_000; + try { + nowSpy.mockReturnValue(epoch); + await memo.get('a', async () => plan(1_000)); + expect(budget.stats().bytesEstimate).toBe(1_000); + + // One tick BEFORE the TTL boundary an unrelated get must NOT prune 'a'. + nowSpy.mockReturnValue(epoch + 59_999); + await memo.get('b', async () => plan(2_000)); + expect(budget.stats().snapshots).toBe(2); + expect(budget.stats().bytesEstimate).toBe(3_000); + + // AT the TTL boundary 'a' (still cached at epoch) must be pruned AND its + // global charge released; 'b' (age 1ms) must survive with its charge. + nowSpy.mockReturnValue(epoch + 60_000); + await memo.get('c', async () => plan(4_000)); + expect(budget.stats().snapshots).toBe(2); + expect(budget.stats().bytesEstimate).toBe(6_000); + expect(await memo.get('a', async () => plan(1), { requireExisting: true })).toBeNull(); + expect(await memo.get('b', async () => plan(1), { requireExisting: true })).not.toBeNull(); + } finally { + nowSpy.mockRestore(); + } + }); + it('the sync handler wires the responder budget through to plan admission', async () => { const cgId = 'meta-plan-budget-wire'; const metaGraph = `did:dkg:context-graph:${cgId}/_shared_memory_meta`;