diff --git a/packages/agent/src/sync/responder/graph-plan.ts b/packages/agent/src/sync/responder/graph-plan.ts index eb708defa3..25290ec898 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'; @@ -99,6 +104,78 @@ export interface FreshSwmDataGraphPlanMemo { ): Promise; } +interface FreshSwmMetaSubjectEntry { + readonly subject: string; + readonly rowCount: number; +} + +interface FreshSwmMetaGraphPlanEntry { + readonly graph: string; + /** TTL-admitted subjects, compareCodePoint-sorted; row counts are exact at plan build. */ + readonly subjects: readonly FreshSwmMetaSubjectEntry[]; + readonly rowCount: number; +} + +/** + * Session plan for the TTL-filtered SWM meta phase (#1847). Holds only + * 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). + * + * 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 { + readonly entries: readonly FreshSwmMetaGraphPlanEntry[]; + readonly totalRows: number; + /** Estimated retained heap bytes of the plan's subject/count scalars. */ + 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 { + get( + key: string, + load: () => Promise, + options?: { refresh?: boolean; requireExisting?: boolean; signal?: AbortSignal }, + ): Promise; +} + interface ExactGraphPagePlanEntry { graph: string; rowCount: number; @@ -250,11 +327,66 @@ 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. + * + * 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, + 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, + accounting?: SessionPlanBudgetAccounting, +): { + get( + key: string, + load: () => Promise, + options?: { refresh?: boolean; requireExisting?: boolean; signal?: AbortSignal }, + ): Promise; +} { + 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 { @@ -267,14 +399,44 @@ export function createResponderFreshSwmDataGraphPlanMemo( 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)); @@ -295,38 +457,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 +596,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 +610,99 @@ 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 + ? { + loadSnapshot: () => 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; + const budgetKey = cache?.key ?? `swm-meta:${params.contextGraphId}`; + const getPlan = createSessionPlanGetter( + params.freshMetaPlanMemo, + params.rowListCacheKey, + params.refreshRowList === true, + (signal) => buildFreshSwmMetaPlan( params.store, candidateGraphs, - params.cutoffIso, + cutoffIso, + budgetKey, + signal, + ), + 'Shared-memory meta sync session graph plan expired before page completion', + ); + const loadStoreBoundedPage: StorePageLoader = async (offset, limit, signal) => + readFreshSwmMetaRowsPageFromPlan( + params.store, + await getPlan(offset, signal), offset, limit, + budgetKey, signal, - ), + ); + return readResponderRowsPage( + cache, + loadStoreBoundedPage, params.offset, params.limit, params.signal, - cache - ? () => readBoundedSwmMetaSnapshot( - params.store, - candidateGraphs, - params.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, + { + 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, + }, ); } @@ -623,12 +822,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, ); } @@ -675,11 +876,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, }); } @@ -1274,36 +1471,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( @@ -1325,12 +1504,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, ); } @@ -1375,6 +1556,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'; @@ -1592,6 +1782,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 @@ -1677,15 +1912,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 []; @@ -1952,38 +2205,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 ?? { @@ -2020,16 +2250,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, }); } @@ -2064,7 +2289,7 @@ async function readBoundedSwmMetaSnapshot( } } - return filterSwmMetaSnapshotRows(rows, cutoffIso); + return filterSwmMetaSnapshotRows(rows, null); } function filterSwmMetaSnapshotRows( @@ -2125,10 +2350,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 +2369,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 +2389,480 @@ async function readSwmMetaRowsPage( .filter((row) => row.s && row.p && row.o && row.g); } +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), 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 + * 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, + 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 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} + } + } + LIMIT ${discoveryLimit} + `, 'sync.responder.readFreshSwmMetaSubjects'); + await runDiscovery(` + 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} + } + } + LIMIT ${discoveryLimit} + `, 'sync.responder.readFreshSwmMetaHeadSubjects'); + 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[], + budgetKey: string, + signal?: AbortSignal, +): Promise { + const countsBySubject = new Map(); + for (const chunk of chunkValues(subjects, FRESH_SWM_META_PLAN_SUBJECT_CHUNK)) { + 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']; + 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. + * + * 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, + 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, + rowCount: subjects.reduce((sum, entry) => sum + entry.rowCount, 0), + }); + } + 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, 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. + * + * `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[] = []; + for (const chunk of chunkValues(subjects, FRESH_SWM_META_PLAN_SUBJECT_CHUNK)) { + 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, + ), + }); + 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; + const bucket = rowsBySubject.get(s) ?? []; + bucket.push({ s, p, o, g: graph }); + rowsBySubject.set(s, bucket); + } + } + 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); + 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`, + ); + } + for (const row of subjectRows) rows.push(row); + } + } + 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 digestBindings = sessionDigestBindingsFor(plan); + 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, + digestBindings, + 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; + const digestBindings = sessionDigestBindingsFor(plan); + for (const entry of plan.entries) { + let graphRows; + try { + graphRows = await readFreshSwmMetaSubjectWindowRows( + store, + entry.graph, + entry.subjects, + digestBindings, + ); + } 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) { + 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/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 fedee5c2d1..8ea8e5a777 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,14 @@ 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, + // #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, SYNC_RESPONDER_DURABLE_DATA_SNAPSHOT_LIMIT, @@ -667,6 +676,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..5f0aebf3e9 --- /dev/null +++ b/packages/agent/test/sync-responder-swm-meta-ceiling.test.ts @@ -0,0 +1,996 @@ +import { describe, it, expect, vi } from 'vitest'; +import type { OperationContext } from '@origintrail-official/dkg-core'; +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, + 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('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('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`; + 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('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('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`; + 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; + } + 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, + afterPage?: (pagesServed: number) => Promise, + ) { + let pagesServed = 0; + 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; + const out = await cap.invoke(envelope); + pagesServed += 1; + await afterPage?.(pagesServed); + return new TextEncoder().encode(out); + }, + 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(); + }); + } + + 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(); + }); +}); 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",