From d6a41417e6b7bc7b23fe2be7949185a1070c9b75 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sat, 12 Sep 2026 23:33:28 +0800 Subject: [PATCH 1/2] refactor(authority): unify retained journals and validate scan snapshots Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- .../authority-three-arm-rehearsal.py | 15 ++ .../coordination/authority_journal_scan.ts | 64 +++++++ .../coordination/authority_store_codec.ts | 2 +- .../authority_store_transactions.ts | 71 +++++++- .../coordination/file_authority_store.ts | 139 +++------------- .../coordination/nokv_authority_store.ts | 156 +++--------------- .../postgresql_authority_store.ts | 60 ++----- .../coordination/sqlite_authority_store.ts | 27 ++- .../test_authority_journal_readback.py | 33 ++++ .../authority_journal_scan.test.ts | 67 ++++++++ .../authority_scan_conformance.ts | 69 ++++++++ .../authority_store_conformance.ts | 2 + .../authority_store_transactions.test.ts | 50 +++++- ...gresql_authority_store.integration.test.ts | 58 +++++++ 14 files changed, 487 insertions(+), 326 deletions(-) create mode 100644 loopx/control_plane/coordination/authority_journal_scan.ts create mode 100644 tests/control_plane/test_authority_journal_readback.py create mode 100644 tests/control_plane_ts/authority_journal_scan.test.ts create mode 100644 tests/control_plane_ts/authority_scan_conformance.ts diff --git a/examples/control_plane/authority-three-arm-rehearsal.py b/examples/control_plane/authority-three-arm-rehearsal.py index a06ac0092c..4cc28e7a03 100644 --- a/examples/control_plane/authority-three-arm-rehearsal.py +++ b/examples/control_plane/authority-three-arm-rehearsal.py @@ -135,6 +135,20 @@ assert.equal(loaded.status, 'loaded', `${name} readback failed`); const receipt = await store.readReceipt(`${name}-three-arm-archive`); assert.equal(receipt.status, 'found', `${name} receipt missing`); + const firstPage = await store.scanCommitted(null, 1); + assert.equal(firstPage.status, 'page', `${name} first journal page failed`); + assert.equal(firstPage.has_more, true); + assert.deepEqual(firstPage.transactions[0].projection, request.initial); + const finalPage = await store.scanCommitted(firstPage.next_cursor, 1); + assert.equal(finalPage.status, 'page', `${name} final journal page failed`); + assert.equal(finalPage.has_more, false); + assert.deepEqual(finalPage.transactions[0].projection, loaded.head); + assert.equal(finalPage.transactions[0].provider_revision, loaded.provider_revision); + assert.deepEqual(finalPage.transactions[0].receipts, receipt.receipts); + const end = await store.scanCommitted(finalPage.next_cursor, 1); + assert.deepEqual(end, {status: 'page', transactions: [], + next_cursor: finalPage.next_cursor, has_more: false}); + assert.deepEqual(await store.loadAuthority(), loaded, 'journal reads changed authority'); results[name] = {archived, head: loaded.head}; } @@ -231,6 +245,7 @@ active_lease_count_after: activeLeases.length, moved_ids_sha256_prefix: movedDigest.slice(0, 16), provider_heads_exact: true, + journal_pages_exact: true, legacy_active_semantics_exact: true, relative_order_exact: true, non_target_semantics_unchanged: true, diff --git a/loopx/control_plane/coordination/authority_journal_scan.ts b/loopx/control_plane/coordination/authority_journal_scan.ts new file mode 100644 index 0000000000..97bbb752ba --- /dev/null +++ b/loopx/control_plane/coordination/authority_journal_scan.ts @@ -0,0 +1,64 @@ +/** A page proves a contiguous segment of one retained journal snapshot. + * Storage effects and snapshot acquisition remain with each provider. */ +import type {AuthorityStoreCommittedTransaction, AuthorityStoreHead, + AuthorityStoreReadFailure, AuthorityStoreScanResult} from "./authority_store.ts"; +import {AuthorityStoreProtocolError, canonicalAuthorityBytes, parseAuthorityCursor} from "./authority_store_codec.ts"; + +export class AuthorityJournalScan { + readonly after: string | null; + readonly offset: bigint; + readonly limit: number; + + private constructor(after: string | null, offset: bigint, limit: number) { + this.after = after; this.offset = offset; this.limit = limit; + } + + static prepare(after: string | null, limit: number): AuthorityJournalScan | AuthorityStoreReadFailure { + try { + const offset = parseAuthorityCursor(after); + if (!Number.isSafeInteger(limit) || limit < 1) { + throw new AuthorityStoreProtocolError("scan limit must be a positive safe integer"); + } + return new AuthorityJournalScan(after, offset, limit); + } catch (error) { + if (!(error instanceof AuthorityStoreProtocolError)) throw error; + return {status: "failed", reason_code: "invalid_scan_request", reason: error.message}; + } + } + + rangeFailure(headCursor: string | null): AuthorityStoreReadFailure | null { + return this.offset > parseAuthorityCursor(headCursor) + ? {status: "failed", reason_code: "scan_cursor_out_of_range", + reason: "scan cursor is ahead of the provider head"} : null; + } + + /** Providers fetch up to limit + 1 rows inside the same snapshot as head. + * The extra row proves has_more and must pass the same validation. */ + page(rows: readonly AuthorityStoreCommittedTransaction[], head: AuthorityStoreHead | null): AuthorityStoreScanResult { + const range = this.rangeFailure(head?.cursor ?? null); + if (range) return range; + const remaining = parseAuthorityCursor(head?.cursor ?? null) - this.offset; + const requested = BigInt(this.limit) + 1n; + const expected = remaining < requested ? remaining : requested; + if (BigInt(rows.length) !== expected) { + throw new AuthorityStoreProtocolError("committed scan does not cover its retained snapshot interval"); + } + const operations = new Set(); + for (const [index, row] of rows.entries()) { + if (parseAuthorityCursor(row.cursor) !== this.offset + BigInt(index) + 1n) { + throw new AuthorityStoreProtocolError("committed scan cursor lineage is invalid"); + } + if (operations.has(row.operation_id)) { + throw new AuthorityStoreProtocolError("committed scan operation identity is duplicated"); + } + operations.add(row.operation_id); + if (head && row.cursor === head.cursor && (row.provider_revision !== head.provider_revision || + !canonicalAuthorityBytes(row.projection).equals(canonicalAuthorityBytes(head.head)))) { + throw new AuthorityStoreProtocolError("committed scan head lineage is invalid"); + } + } + const transactions = structuredClone(rows.slice(0, this.limit)); + return {status: "page", transactions, + next_cursor: transactions.at(-1)?.cursor ?? this.after, has_more: rows.length > this.limit}; + } +} diff --git a/loopx/control_plane/coordination/authority_store_codec.ts b/loopx/control_plane/coordination/authority_store_codec.ts index e7b0ca13c5..a386352b7a 100644 --- a/loopx/control_plane/coordination/authority_store_codec.ts +++ b/loopx/control_plane/coordination/authority_store_codec.ts @@ -110,7 +110,7 @@ export function canonicalAuthoritySha256(value: unknown): string { export function parseAuthorityCursor(value: string | null): bigint { if (value === null) return 0n; - if (!/^[1-9]\d*$/.test(value)) { + if (typeof value !== "string" || !/^[1-9]\d*$/.test(value)) { throw new AuthorityStoreProtocolError("provider cursor is invalid"); } return BigInt(value); diff --git a/loopx/control_plane/coordination/authority_store_transactions.ts b/loopx/control_plane/coordination/authority_store_transactions.ts index 28ab4dab7d..328926a142 100644 --- a/loopx/control_plane/coordination/authority_store_transactions.ts +++ b/loopx/control_plane/coordination/authority_store_transactions.ts @@ -1,5 +1,5 @@ import type { JsonObject } from "../effect_program.ts"; -import type { AuthorityStoreCommittedTransaction } from "./authority_store.ts"; +import type { AuthorityStoreCommit, AuthorityStoreCommittedTransaction } from "./authority_store.ts"; import { AuthorityStoreProtocolError, canonicalAuthorityObject, @@ -7,6 +7,8 @@ import { hasExactAuthorityKeys, isAuthorityJsonObject, requireAuthorityStoreId, + parseAuthorityCursor, + canonicalAuthorityBytes, } from "./authority_store_codec.ts"; /** Shared wire decoder used by every authority provider. */ @@ -14,8 +16,10 @@ export function decodeAuthorityTransaction(value: unknown): AuthorityStoreCommit if (!isAuthorityJsonObject(value) || !hasExactAuthorityKeys(value, [ "cursor", "provider_revision", "operation_id", "events", "projection", "receipts", ])) throw new AuthorityStoreProtocolError("committed transaction is invalid"); + const cursor = requireAuthorityStoreId(value.cursor, "transaction cursor"); + parseAuthorityCursor(cursor); return { - cursor: requireAuthorityStoreId(value.cursor, "transaction cursor"), + cursor, provider_revision: requireAuthorityStoreId(value.provider_revision, "transaction provider revision"), operation_id: requireAuthorityStoreId(value.operation_id, "operation id"), events: canonicalAuthorityObjectList(value.events, "transaction events"), @@ -24,10 +28,6 @@ export function decodeAuthorityTransaction(value: unknown): AuthorityStoreCommit }; } -export function cloneAuthorityTransaction(value: AuthorityStoreCommittedTransaction): AuthorityStoreCommittedTransaction { - return structuredClone(value); -} - export function transactionForRevision(value: AuthorityStoreCommittedTransaction): JsonObject { return { cursor: value.cursor, @@ -37,3 +37,62 @@ export function transactionForRevision(value: AuthorityStoreCommittedTransaction receipts: value.receipts, }; } + +/** File and NoKV retain the whole journal in one envelope. Their wire headers, + * physical CAS, and revision algorithms remain provider-owned. */ +export interface RetainedAuthorityJournal { + provider_revision: string; + cursor: string; + head: JsonObject; + committed: AuthorityStoreCommittedTransaction[]; +} + +type RevisionInput = ReturnType; +export type JournalRevision = (previous: string | null, transaction: RevisionInput) => string; + +export function decodeRetainedAuthorityJournal(value: JsonObject, label: string, + revisionFor: JournalRevision): RetainedAuthorityJournal { + const revision = requireAuthorityStoreId(value.provider_revision, "provider revision"); + const cursor = requireAuthorityStoreId(value.cursor, "provider cursor"); + const head = canonicalAuthorityObject(value.head, `${label} head`); + if (!Array.isArray(value.committed)) { + throw new AuthorityStoreProtocolError(`${label} history is invalid`); + } + const committed = value.committed.map(decodeAuthorityTransaction); + if (committed.length === 0 || parseAuthorityCursor(cursor) !== BigInt(committed.length)) { + throw new AuthorityStoreProtocolError(`${label} lineage is invalid`); + } + let previous: string | null = null; + const operations = new Set(); + for (const [index, entry] of committed.entries()) { + if (parseAuthorityCursor(entry.cursor) !== BigInt(index + 1)) { + throw new AuthorityStoreProtocolError(`${label} cursor lineage is invalid`); + } + if (operations.has(entry.operation_id)) { + throw new AuthorityStoreProtocolError(`${label} operation identity is duplicated`); + } + operations.add(entry.operation_id); + if (entry.provider_revision !== revisionFor(previous, transactionForRevision(entry))) { + throw new AuthorityStoreProtocolError(`${label} revision lineage is invalid`); + } + previous = entry.provider_revision; + } + const last = committed.at(-1)!; + if (last.cursor !== cursor || last.provider_revision !== revision || + !canonicalAuthorityBytes(last.projection).equals(canonicalAuthorityBytes(head))) { + throw new AuthorityStoreProtocolError(`${label} head lineage is invalid`); + } + return {provider_revision: revision, cursor, head, committed}; +} + +/** Build only after the provider checked revision and operation uniqueness in + * its write boundary. This function neither grants admission nor persists. */ +export function appendRetainedAuthorityJournal(current: RetainedAuthorityJournal | null, + commit: AuthorityStoreCommit, revisionFor: JournalRevision): RetainedAuthorityJournal { + const cursor = (parseAuthorityCursor(current?.cursor ?? null) + 1n).toString(); + const base = {cursor, operation_id: commit.operation_id, events: commit.events, + projection: commit.next_projection, receipts: commit.receipts}; + const revision = revisionFor(current?.provider_revision ?? null, base); + return {provider_revision: revision, cursor, head: commit.next_projection, + committed: [...(current?.committed ?? []), {...base, provider_revision: revision}]}; +} diff --git a/loopx/control_plane/coordination/file_authority_store.ts b/loopx/control_plane/coordination/file_authority_store.ts index b89fe6b297..2ef5818dfb 100644 --- a/loopx/control_plane/coordination/file_authority_store.ts +++ b/loopx/control_plane/coordination/file_authority_store.ts @@ -8,7 +8,6 @@ import { withFileMutationLock } from "../effect_runtime_io.ts"; import type { AuthorityStore, AuthorityStoreCommit, - AuthorityStoreCommittedTransaction, AuthorityStoreCommitResult, AuthorityStoreIdentityResult, AuthorityStoreLoadResult, @@ -20,27 +19,22 @@ import { AuthorityStoreProtocolError, isAuthorityJsonObject, hasExactAuthorityKeys, - canonicalAuthorityObjectList, - canonicalAuthorityObject, - authorityUnicodeCompare, canonicalAuthorityBytes, normalizeAuthorityStoreCommit, - parseAuthorityCursor, requireAuthorityStoreId, } from "./authority_store_codec.ts"; -import { cloneAuthorityTransaction, decodeAuthorityTransaction, transactionForRevision } from "./authority_store_transactions.ts"; +import {appendRetainedAuthorityJournal, decodeRetainedAuthorityJournal, + type RetainedAuthorityJournal, transactionForRevision} from "./authority_store_transactions.ts"; +import {AuthorityJournalScan} from "./authority_journal_scan.ts"; const FILE_AUTHORITY_STORE_SCHEMA = "loopx_file_authority_store_v0"; const STORE_IDENTITY_PATTERN = /^file:[0-9a-f]{32}$/; -interface FileAuthorityStoreDocument extends JsonObject { +interface FileAuthorityStoreDocument extends JsonObject, RetainedAuthorityJournal { schema_version: typeof FILE_AUTHORITY_STORE_SCHEMA; goal_id: string; - provider_revision: string; - cursor: string; store_identity: string; - head: JsonObject; - committed: AuthorityStoreCommittedTransaction[]; + } class FileStoreUnavailableError extends Error {} @@ -127,51 +121,9 @@ function decodeDocument( if (value.store_identity !== storeIdentity) { throw new AuthorityStoreProtocolError("file authority store lineage mismatch"); } - const revision = requireAuthorityStoreId(value.provider_revision, "provider revision"); - const cursor = requireAuthorityStoreId(value.cursor, "provider cursor"); - const head = canonicalAuthorityObject(value.head, "file authority store head"); - if (!Array.isArray(value.committed)) { - throw new AuthorityStoreProtocolError("file authority store history is invalid"); - } - const committed = value.committed.map(decodeAuthorityTransaction); - if (committed.length === 0 || parseAuthorityCursor(cursor) !== BigInt(committed.length)) { - throw new AuthorityStoreProtocolError("file authority store lineage is invalid"); - } - let previousRevision: string | null = null; - const operationIds = new Set(); - for (const [index, entry] of committed.entries()) { - if (parseAuthorityCursor(entry.cursor) !== BigInt(index + 1)) { - throw new AuthorityStoreProtocolError("file authority store cursor lineage is invalid"); - } - if (operationIds.has(entry.operation_id)) { - throw new AuthorityStoreProtocolError("file authority store operation identity is duplicated"); - } - operationIds.add(entry.operation_id); - const expectedRevision = providerRevision( - goalId, - storeIdentity, - previousRevision, - transactionForRevision(entry), - ); - if (entry.provider_revision !== expectedRevision) { - throw new AuthorityStoreProtocolError("file authority store revision lineage is invalid"); - } - previousRevision = entry.provider_revision; - } - const last = committed.at(-1)!; - if ( - last.cursor !== cursor || last.provider_revision !== revision || - !canonicalAuthorityBytes(last.projection).equals(canonicalAuthorityBytes(head)) - ) throw new AuthorityStoreProtocolError("file authority store head lineage is invalid"); - return { - schema_version: FILE_AUTHORITY_STORE_SCHEMA, - goal_id: goalId, - store_identity: storeIdentity, - provider_revision: revision, - cursor, - head, - committed, - }; + return {schema_version: FILE_AUTHORITY_STORE_SCHEMA, goal_id: goalId, store_identity: storeIdentity, + ...decodeRetainedAuthorityJournal(value, "file authority store", (previous, transaction) => + providerRevision(goalId, storeIdentity, previous, transaction))}; } function readFailure(error: unknown): AuthorityStoreReadFailure { @@ -345,33 +297,11 @@ export class FileAuthorityStore implements AuthorityStore { current_cursor: current.cursor, }; } - const cursor = (parseAuthorityCursor(current?.cursor ?? null) + 1n).toString(); - const base = { - cursor, - operation_id: normalized.operation_id, - events: normalized.events, - projection: normalized.next_projection, - receipts: normalized.receipts, - }; - const revision = providerRevision( - this.goalId, - identity, - current?.provider_revision ?? null, - base, - ); - const transaction: AuthorityStoreCommittedTransaction = { - ...base, - provider_revision: revision, - }; - const document: FileAuthorityStoreDocument = { - schema_version: FILE_AUTHORITY_STORE_SCHEMA, - goal_id: this.goalId, - store_identity: identity, - provider_revision: revision, - cursor, - head: normalized.next_projection, - committed: [...(current?.committed ?? []), transaction], - }; + const journal = appendRetainedAuthorityJournal(current, normalized, (previous, transaction) => + providerRevision(this.goalId, identity, previous, transaction)); + const {cursor, provider_revision: revision} = journal; + const document: FileAuthorityStoreDocument = {schema_version: FILE_AUTHORITY_STORE_SCHEMA, + goal_id: this.goalId, store_identity: identity, ...journal}; try { await this.replaceDurably(this.path, canonicalAuthorityBytes(document)); } catch (error) { @@ -423,43 +353,16 @@ export class FileAuthorityStore implements AuthorityStore { } async scanCommitted(afterCursor: string | null, limit: number): Promise { - let offset: bigint; - try { - offset = parseAuthorityCursor(afterCursor); - if (!Number.isSafeInteger(limit) || limit < 1) { - throw new AuthorityStoreProtocolError("scan limit must be a positive safe integer"); - } - } catch (error) { - return { - status: "failed", - reason_code: "invalid_scan_request", - reason: error instanceof Error ? error.message : "invalid scan request", - }; - } + const scan = AuthorityJournalScan.prepare(afterCursor, limit); + if (!(scan instanceof AuthorityJournalScan)) return scan; try { const document = await this.readDocument(); - if (!document) { - return { status: "page", transactions: [], next_cursor: afterCursor, has_more: false }; - } - const headCursor = parseAuthorityCursor(document.cursor); - if (offset > headCursor || offset > BigInt(Number.MAX_SAFE_INTEGER)) { - return { - status: "failed", - reason_code: "scan_cursor_out_of_range", - reason: "scan cursor is ahead of the provider head", - }; - } - const start = Number(offset); - const transactions = document.committed.slice(start, start + limit).map(cloneAuthorityTransaction); - return { - status: "page", - transactions, - next_cursor: transactions.at(-1)?.cursor ?? afterCursor, - has_more: start + transactions.length < document.committed.length, - }; - } catch (error) { - return readFailure(error); - } + if (!document) return scan.page([], null); + const range = scan.rangeFailure(document.cursor); + if (range) return range; + const start = Number(scan.offset); + return scan.page(document.committed.slice(start, start + limit + 1), document); + } catch (error) { return readFailure(error); } } /** diff --git a/loopx/control_plane/coordination/nokv_authority_store.ts b/loopx/control_plane/coordination/nokv_authority_store.ts index c655e3389f..365af80e20 100644 --- a/loopx/control_plane/coordination/nokv_authority_store.ts +++ b/loopx/control_plane/coordination/nokv_authority_store.ts @@ -17,14 +17,13 @@ import { AuthorityStoreProtocolError, isAuthorityJsonObject, hasExactAuthorityKeys, - canonicalAuthorityObjectList, - canonicalAuthorityObject, canonicalAuthorityBytes, normalizeAuthorityStoreCommit, - parseAuthorityCursor, requireAuthorityStoreId, } from "./authority_store_codec.ts"; -import { cloneAuthorityTransaction, decodeAuthorityTransaction, transactionForRevision } from "./authority_store_transactions.ts"; +import {appendRetainedAuthorityJournal, decodeRetainedAuthorityJournal, + type RetainedAuthorityJournal, transactionForRevision} from "./authority_store_transactions.ts"; +import {AuthorityJournalScan} from "./authority_journal_scan.ts"; const NOKV_AUTHORITY_STORE_SCHEMA = "loopx_nokv_authority_store_v0"; const DEFAULT_MAX_ENVELOPE_BYTES = 16 * 1024 * 1024; @@ -77,16 +76,13 @@ export interface NoKVAuthorityStoreOptions { max_envelope_bytes?: number; } -interface NoKVAuthorityStoreDocument extends JsonObject { +interface NoKVAuthorityStoreDocument extends JsonObject, RetainedAuthorityJournal { schema_version: typeof NOKV_AUTHORITY_STORE_SCHEMA; tenant_id: string; goal_id: string; store_identity: string; storage_generation: number; - provider_revision: string; - cursor: string; - head: JsonObject; - committed: AuthorityStoreCommittedTransaction[]; + } type EnvelopeReadResult = @@ -162,65 +158,13 @@ function decodeDocument( "NoKV authority store storage generation does not match read metadata", ); } - const revision = requireAuthorityStoreId(value.provider_revision, "provider revision"); - const cursor = requireAuthorityStoreId(value.cursor, "provider cursor"); - const head = canonicalAuthorityObject(value.head, "NoKV authority store head"); - if (!Array.isArray(value.committed)) { - throw new AuthorityStoreProtocolError("NoKV authority store history is invalid"); - } - const committed = value.committed.map(decodeAuthorityTransaction); - if ( - committed.length === 0 || - storageGeneration !== committed.length || - parseAuthorityCursor(cursor) !== BigInt(committed.length) - ) { + if (!Array.isArray(value.committed) || storageGeneration !== value.committed.length) { throw new AuthorityStoreProtocolError("NoKV authority store generation lineage is invalid"); } - let previousRevision: string | null = null; - const operationIds = new Set(); - for (const [index, entry] of committed.entries()) { - const generation = index + 1; - if (parseAuthorityCursor(entry.cursor) !== BigInt(generation)) { - throw new AuthorityStoreProtocolError("NoKV authority store cursor lineage is invalid"); - } - if (operationIds.has(entry.operation_id)) { - throw new AuthorityStoreProtocolError( - "NoKV authority store operation identity is duplicated", - ); - } - operationIds.add(entry.operation_id); - const expectedRevision = providerRevision( - tenantId, - goalId, - storeIdentity, - generation, - previousRevision, - transactionForRevision(entry), - ); - if (entry.provider_revision !== expectedRevision) { - throw new AuthorityStoreProtocolError("NoKV authority store revision lineage is invalid"); - } - previousRevision = entry.provider_revision; - } - const last = committed.at(-1)!; - if ( - last.cursor !== cursor || - last.provider_revision !== revision || - !canonicalAuthorityBytes(last.projection).equals(canonicalAuthorityBytes(head)) - ) { - throw new AuthorityStoreProtocolError("NoKV authority store head lineage is invalid"); - } - return { - schema_version: NOKV_AUTHORITY_STORE_SCHEMA, - tenant_id: tenantId, - goal_id: goalId, - store_identity: storeIdentity, - storage_generation: storageGeneration, - provider_revision: revision, - cursor, - head, - committed, - }; + return {schema_version: NOKV_AUTHORITY_STORE_SCHEMA, tenant_id: tenantId, goal_id: goalId, + store_identity: storeIdentity, storage_generation: storageGeneration, + ...decodeRetainedAuthorityJournal(value, "NoKV authority store", (previous, transaction) => + providerRevision(tenantId, goalId, storeIdentity, Number(transaction.cursor), previous, transaction))}; } function readFailure(error: unknown): AuthorityStoreReadFailure { @@ -437,38 +381,13 @@ export class NoKVAuthorityStore implements AuthorityStore { current_cursor: currentDocument.cursor, }; } - const cursor = (parseAuthorityCursor(currentDocument?.cursor ?? null) + 1n).toString(); const generation = (current.status === "loaded" ? current.generation : 0) + 1; - const base = { - cursor, - operation_id: normalized.operation_id, - events: normalized.events, - projection: normalized.next_projection, - receipts: normalized.receipts, - }; - const revision = providerRevision( - this.tenantId, - this.goalId, - current.identity, - generation, - currentDocument?.provider_revision ?? null, - base, - ); - const transaction: AuthorityStoreCommittedTransaction = { - ...base, - provider_revision: revision, - }; - const document: NoKVAuthorityStoreDocument = { - schema_version: NOKV_AUTHORITY_STORE_SCHEMA, - tenant_id: this.tenantId, - goal_id: this.goalId, - store_identity: current.identity, - storage_generation: generation, - provider_revision: revision, - cursor, - head: normalized.next_projection, - committed: [...(currentDocument?.committed ?? []), transaction], - }; + const journal = appendRetainedAuthorityJournal(currentDocument, normalized, (previous, transaction) => + providerRevision(this.tenantId, this.goalId, current.identity, generation, previous, transaction)); + const transaction = journal.committed.at(-1)!; + const document: NoKVAuthorityStoreDocument = {schema_version: NOKV_AUTHORITY_STORE_SCHEMA, + tenant_id: this.tenantId, goal_id: this.goalId, store_identity: current.identity, + storage_generation: generation, ...journal}; const payload = canonicalAuthorityBytes(document); if (payload.byteLength > this.maxEnvelopeBytes) { return { @@ -579,41 +498,16 @@ export class NoKVAuthorityStore implements AuthorityStore { afterCursor: string | null, limit: number, ): Promise { - let offset: bigint; - try { - offset = parseAuthorityCursor(afterCursor); - if (!Number.isSafeInteger(limit) || limit < 1) { - throw new AuthorityStoreProtocolError("scan limit must be a positive safe integer"); - } - } catch (error) { - return { - status: "failed", - reason_code: "invalid_scan_request", - reason: error instanceof Error ? error.message : "invalid scan request", - }; - } + const scan = AuthorityJournalScan.prepare(afterCursor, limit); + if (!(scan instanceof AuthorityJournalScan)) return scan; const result = await this.readEnvelope(); - if (result.status === "missing") { - return { status: "page", transactions: [], next_cursor: afterCursor, has_more: false }; - } + if (result.status === "missing") return scan.page([], null); if (result.status !== "loaded") return result; - const headCursor = parseAuthorityCursor(result.document.cursor); - if (offset > headCursor || offset > BigInt(Number.MAX_SAFE_INTEGER)) { - return { - status: "failed", - reason_code: "scan_cursor_out_of_range", - reason: "scan cursor is ahead of the provider head", - }; - } - const start = Number(offset); - const transactions = result.document.committed - .slice(start, start + limit) - .map(cloneAuthorityTransaction); - return { - status: "page", - transactions, - next_cursor: transactions.at(-1)?.cursor ?? afterCursor, - has_more: start + transactions.length < result.document.committed.length, - }; + try { + const range = scan.rangeFailure(result.document.cursor); + if (range) return range; + const start = Number(scan.offset); + return scan.page(result.document.committed.slice(start, start + limit + 1), result.document); + } catch (error) { return readFailure(error); } } } diff --git a/loopx/control_plane/coordination/postgresql_authority_store.ts b/loopx/control_plane/coordination/postgresql_authority_store.ts index 53a87e3898..fba9c1cfc9 100644 --- a/loopx/control_plane/coordination/postgresql_authority_store.ts +++ b/loopx/control_plane/coordination/postgresql_authority_store.ts @@ -1,8 +1,8 @@ +import {AuthorityJournalScan} from "./authority_journal_scan.ts"; import type { JsonObject } from "../effect_program.ts"; import type { AuthorityStore, AuthorityStoreCommit, - AuthorityStoreCommittedTransaction, AuthorityStoreCommitResult, AuthorityStoreIdentityResult, AuthorityStoreLoadResult, @@ -321,7 +321,7 @@ async function beginTenantTransaction( tenantId: string, options: { readOnly: boolean }, ): Promise { - await connection.query(options.readOnly ? "BEGIN READ ONLY" : "BEGIN"); + await connection.query(options.readOnly ? "BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY" : "BEGIN"); try { const context = oneRow(await connection.query( "SELECT set_config('loopx.tenant_id', $1, TRUE) AS tenant_id", @@ -713,19 +713,8 @@ export class PostgreSqlAuthorityStore implements AuthorityStore { afterCursor: string | null, limit: number, ): Promise { - let offset: bigint; - try { - offset = parseAuthorityCursor(afterCursor); - if (!Number.isSafeInteger(limit) || limit < 1) { - throw new AuthorityStoreProtocolError("scan limit must be a positive safe integer"); - } - } catch (error) { - return { - status: "failed", - reason_code: "invalid_scan_request", - reason: error instanceof Error ? error.message : "invalid scan request", - }; - } + const scan = AuthorityJournalScan.prepare(afterCursor, limit); + if (!(scan instanceof AuthorityJournalScan)) return scan; try { return await this.readInTenantTransaction(async (connection) => { const storeIdentity = await requireStoreIdentity(connection); @@ -733,45 +722,22 @@ export class PostgreSqlAuthorityStore implements AuthorityStore { await connection.query(SELECT_HEAD_SQL, [this.tenantId, this.goalId]), "PostgreSQL authority head", ); - if (current === null) { - return { - status: "page", - transactions: [], - next_cursor: afterCursor, - has_more: false, - } as const; - } + if (current === null) return scan.page([], null); const head = decodeHeadRow(current); - if (offset > BigInt(head.cursor)) { - return { - status: "failed", - reason_code: "scan_cursor_out_of_range", - reason: "scan cursor is ahead of the provider head", - } as const; - } + const snapshot = head.head === null ? null : {cursor: head.cursor, + provider_revision: providerRevisionToken(storeIdentity, head.provider_revision), head: head.head}; + const range = scan.rangeFailure(snapshot?.cursor ?? null); + if (range) return range; const result = rows(await connection.query( `${SELECT_TRANSACTION_COLUMNS_SQL} WHERE commit.tenant_id = $1 AND commit.goal_id = $2 AND commit.cursor > $3::bigint ORDER BY commit.cursor LIMIT $4`, - [this.tenantId, this.goalId, offset.toString(), (BigInt(limit) + 1n).toString()], + [this.tenantId, this.goalId, scan.offset.toString(), (BigInt(limit) + 1n).toString()], )).map(decodeTransactionRow); - const hasMore = result.length > limit; - const page = result.slice(0, limit); - const transactions: AuthorityStoreCommittedTransaction[] = page.map((value) => ({ - cursor: value.cursor, - provider_revision: providerRevisionToken(storeIdentity, value.provider_revision), - operation_id: value.operation_id, - events: structuredClone(value.events), - projection: structuredClone(value.projection), - receipts: structuredClone(value.receipts), - })); - return { - status: "page", - transactions, - next_cursor: transactions.at(-1)?.cursor ?? afterCursor, - has_more: hasMore, - } as const; + const transactions = result.map(value => ({...value, + provider_revision: providerRevisionToken(storeIdentity, value.provider_revision)})); + return scan.page(transactions, snapshot); }); } catch (error) { return readFailure(error); diff --git a/loopx/control_plane/coordination/sqlite_authority_store.ts b/loopx/control_plane/coordination/sqlite_authority_store.ts index 068e7d6b76..854ff4a933 100644 --- a/loopx/control_plane/coordination/sqlite_authority_store.ts +++ b/loopx/control_plane/coordination/sqlite_authority_store.ts @@ -1,3 +1,4 @@ +import {AuthorityJournalScan} from "./authority_journal_scan.ts"; import { createHash, randomUUID } from "node:crypto"; import { existsSync, mkdirSync } from "node:fs"; import { createRequire } from "node:module"; @@ -9,7 +10,7 @@ import type { AuthorityStore, AuthorityStoreCommit, AuthorityStoreCommitResult, AuthorityStoreReceiptResult, AuthorityStoreScanResult } from "./authority_store.ts"; import { AuthorityStoreProtocolError, canonicalAuthorityBytes, canonicalAuthorityObject, canonicalAuthorityObjectList, canonicalAuthoritySha256, normalizeAuthorityStoreCommit, - parseAuthorityCursor, requireAuthorityStoreId } from "./authority_store_codec.ts"; + requireAuthorityStoreId } from "./authority_store_codec.ts"; const SCHEMA = "loopx_sqlite_authority_store_v0"; const IDENTITY = /^sqlite:[0-9a-f]{32}$/; @@ -260,27 +261,21 @@ export class SqliteAuthorityStore implements AuthorityStore { } async scanCommitted(afterCursor: string | null, limit: number): Promise { - let offset: bigint; - try { - offset = parseAuthorityCursor(afterCursor); - if (!Number.isSafeInteger(limit) || limit < 1) protocol("Scan limit must be a positive safe integer"); - } catch (error) { return {status: "failed", reason_code: "invalid_scan_request", - reason: error instanceof Error ? error.message : "Invalid scan request"}; } + const scan = AuthorityJournalScan.prepare(afterCursor, limit); + if (!(scan instanceof AuthorityJournalScan)) return scan; let db: DatabaseSync | null = null; try { db = this.open(false); - if (!db) return {status: "page", transactions: [], next_cursor: afterCursor, has_more: false}; + if (!db) return scan.page([], null); db.exec("BEGIN"); const identity = this.identity(db); - const head = BigInt(this.current(db)?.cursor ?? "0"); - if (offset > head) return {status: "failed", reason_code: "scan_cursor_out_of_range", reason: "Scan cursor is ahead of the provider head"}; + const current = this.current(db); + const range = scan.rangeFailure(current?.cursor ?? null); + if (range) return range; const rows = db.prepare(`SELECT ${ROW_COLUMNS} FROM commits WHERE cursor > ? ORDER BY cursor LIMIT ?`); - const result = rows.all(offset, BigInt(limit) + 1n); - // Validate the lookahead row too: it is evidence for has_more. - const verified = result.map(row => this.transaction(row, identity)); - const transactions = verified.slice(0, limit); - return {status: "page", transactions, next_cursor: transactions.at(-1)?.cursor ?? afterCursor, - has_more: result.length > limit}; + const verified = rows.all(scan.offset, BigInt(limit) + 1n).map(row => this.transaction(row, identity)); + return scan.page(verified, current ? {cursor: current.cursor, + provider_revision: current.provider_revision, head: current.projection} : null); } catch (error) { return readFailure(error); } finally { db?.close(); } } diff --git a/tests/control_plane/test_authority_journal_readback.py b/tests/control_plane/test_authority_journal_readback.py new file mode 100644 index 0000000000..ee76b89103 --- /dev/null +++ b/tests/control_plane/test_authority_journal_readback.py @@ -0,0 +1,33 @@ +"""Actual Python -> managed TS -> File readback, without an active Goal.""" +from pathlib import Path + +import pytest + +from loopx.control_plane.coordination.local_authority_shadow_adapter import ( + read_local_authority_shadow, +) + + +@pytest.mark.parametrize("cursor", [None, "1", "9007199254740993"]) +def test_empty_journal_checkpoint_survives_runtime_transport(tmp_path: Path, cursor: str | None) -> None: + root = tmp_path / "runtime" + directory = root / "authority-shadow" / "file" / "synthetic-journal" + directory.mkdir(parents=True) + identity = directory / "store-identity" + identity.write_text("file:" + "a" * 32, encoding="ascii") + before = identity.read_bytes() + + result = read_local_authority_shadow( + runtime_root=root, goal_id="synthetic-journal", store_kind="legacy_observation", + scan_after_cursor=cursor, scan_limit=1, + ) + + if cursor is None: + assert result["status"] == "missing" + assert result["scan"] == {"transactions": [], "next_cursor": None, "has_more": False} + else: + assert result["status"] == "failed" + assert result["reason_code"] == "scan_cursor_out_of_range" + assert result["scan"] is None + assert identity.read_bytes() == before + assert sorted(path.name for path in directory.iterdir()) == ["store-identity"] diff --git a/tests/control_plane_ts/authority_journal_scan.test.ts b/tests/control_plane_ts/authority_journal_scan.test.ts new file mode 100644 index 0000000000..bf3dd6056d --- /dev/null +++ b/tests/control_plane_ts/authority_journal_scan.test.ts @@ -0,0 +1,67 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {AuthorityJournalScan} from "../../loopx/control_plane/coordination/authority_journal_scan.ts"; +import {decodeAuthorityTransaction} from "../../loopx/control_plane/coordination/authority_store_transactions.ts"; +import {parseAuthorityCursor} from "../../loopx/control_plane/coordination/authority_store_codec.ts"; + +const row = (cursor: string) => ({cursor, provider_revision: `revision-${cursor}`, + operation_id: `operation-${cursor}`, events: [{sequence: cursor}], receipts: [{accepted: true}], + projection: {nested: {sequence: cursor}}}); +const head = (cursor: string) => ({cursor, provider_revision: `revision-${cursor}`, head: row(cursor).projection}); +const scan = (after: string | null, limit: number) => { + const result = AuthorityJournalScan.prepare(after, limit); + assert.ok(result instanceof AuthorityJournalScan); + return result; +}; + +test("scan snapshot rejects gaps, repeated identities, short pages and unproved lookahead", () => { + const candidates = [[], [row("1")], [row("1"), row("3")], + [row("1"), {...row("2"), operation_id: "operation-1"}], + [row("1"), {...row("2"), cursor: "01"}], + [row("1"), row("2"), row("3")]]; + for (const rows of candidates) { + assert.throws(() => scan(null, 1).page(rows, head("3")), /committed scan|provider cursor/); + } + assert.throws(() => scan("1", 10).page([row("2")], head("3")), /snapshot interval/); +}); + +test("a final row and a lookahead head must match the observed head exactly", () => { + for (const limit of [1, 2]) { + for (const corrupt of [{...head("2"), provider_revision: "other"}, + {...head("2"), head: {nested: {sequence: "forged"}}}]) { + assert.throws(() => scan(null, limit).page([row("1"), row("2")], corrupt), /head lineage/); + } + } +}); + +test("scan checkpoints preserve bigint precision and output value isolation", () => { + const offset = "9007199254740993"; + const next = "9007199254740994"; + const rows = [row(next)]; + const page = scan(offset, Number.MAX_SAFE_INTEGER).page(rows, head(next)); + assert.equal(page.status, "page"); + if (page.status !== "page") return; + assert.equal(page.next_cursor, next); + assert.equal(page.has_more, false); + page.transactions[0]!.projection.nested = "changed"; + assert.deepEqual(rows[0]!.projection.nested, {sequence: next}); + assert.deepEqual(scan(next, 1).page([], head(next)), + {status: "page", transactions: [], next_cursor: next, has_more: false}); +}); + +test("scan requests reject malformed types before a provider effect", () => { + for (const limit of [0, -1, 1.5, Infinity, NaN, "1", 1n, Number.MAX_SAFE_INTEGER + 1]) { + const result = AuthorityJournalScan.prepare(null, limit as number); + assert.ok(!(result instanceof AuthorityJournalScan)); + assert.equal(result.reason_code, "invalid_scan_request"); + } + for (const cursor of [1, 1n, ["1"], {toString: () => "1"}, undefined]) { + assert.throws(() => parseAuthorityCursor(cursor as string), /provider cursor/); + } +}); + +test("transaction decoder requires a canonical positive decimal cursor", () => { + for (const cursor of ["0", "01", "-1", "1.0", "unknown", 1, null]) { + assert.throws(() => decodeAuthorityTransaction({...row("1"), cursor}), /cursor/); + } +}); diff --git a/tests/control_plane_ts/authority_scan_conformance.ts b/tests/control_plane_ts/authority_scan_conformance.ts new file mode 100644 index 0000000000..66cd931e9c --- /dev/null +++ b/tests/control_plane_ts/authority_scan_conformance.ts @@ -0,0 +1,69 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts"; +import {productionScaleCoordinationFixture} from "./production_scale_coordination_fixture.ts"; + +/** Scan consumers must distinguish complete history from an invalid checkpoint. + * Expectations follow the retained, contiguous journal contract, not a provider. */ +export function registerAuthorityScanConformance(name: string, factory: AuthorityStoreConformanceFactory): void { + test(`${name} scan contract: empty history cannot acknowledge a positive checkpoint`, async t => { + const {store} = await factory(t); + assert.deepEqual(await store.scanCommitted(null, 1), + {status: "page", transactions: [], next_cursor: null, has_more: false}); + for (const cursor of ["1", "9007199254740993"]) { + const result = await store.scanCommitted(cursor, 1); + assert.equal(result.status, "failed", JSON.stringify(result)); + if (result.status === "failed") assert.equal(result.reason_code, "scan_cursor_out_of_range"); + } + assert.deepEqual(await store.loadAuthority(), {status: "missing"}); + }); + + test(`${name} scan contract: runtime cursor types never coerce into checkpoints`, async t => { + const {store} = await factory(t); + for (const cursor of [1, 1n, ["1"], {toString: () => "1"}, undefined, "0", "01", " 1"]) { + const result = await store.scanCommitted(cursor as string, 1); + assert.equal(result.status, "failed"); + if (result.status === "failed") assert.equal(result.reason_code, "invalid_scan_request"); + } + assert.deepEqual(await store.loadAuthority(), {status: "missing"}); + }); + + test(`${name} scan contract: complex historical pages retain exact identity and isolation`, async t => { + const {store, contender} = await factory(t); + const fixture = productionScaleCoordinationFixture("goal-scan-fixture"); + const original = structuredClone(fixture.projection); + let revision: string | null = null; + for (let index = 1; index <= 4; index++) { + const result = await store.commitAuthority({expected_provider_revision: revision, + operation_id: `history-${index}`, next_projection: {...original, authority_revision: index}, + events: [{kind: "projection_checkpoint", sequence: index}], + receipts: [{operation_id: `history-${index}`, sequence: index, evidence: {valid: true}}]}); + assert.equal(result.status, "applied"); + if (result.status !== "applied") return; + revision = result.provider_revision; + } + const before = await store.loadAuthority(); + let cursor: string | null = null; + const seen: string[] = []; + for (const limit of [1, 2, Number.MAX_SAFE_INTEGER]) { + const page = await store.scanCommitted(cursor, limit); + assert.equal(page.status, "page", JSON.stringify(page)); + if (page.status !== "page") return; + for (const entry of page.transactions) { + seen.push(entry.operation_id); + assert.deepEqual(entry.projection.todos, original.todos); + assert.deepEqual(entry.projection.leases, original.leases); + } + cursor = page.next_cursor; + assert.equal(page.has_more, cursor !== "4"); + if (page.transactions.length) page.transactions[0]!.projection.todos = []; + } + assert.deepEqual(seen, ["history-1", "history-2", "history-3", "history-4"]); + assert.deepEqual(await contender.loadAuthority(), before); + assert.deepEqual(await store.scanCommitted("4", 1), + {status: "page", transactions: [], next_cursor: "4", has_more: false}); + const receipt = await store.readReceipt("history-1"); + assert.equal(receipt.status, "found"); + if (receipt.status === "found") assert.equal(receipt.cursor, "1"); + }); +} diff --git a/tests/control_plane_ts/authority_store_conformance.ts b/tests/control_plane_ts/authority_store_conformance.ts index e8d3fa680c..582ed8d9bc 100644 --- a/tests/control_plane_ts/authority_store_conformance.ts +++ b/tests/control_plane_ts/authority_store_conformance.ts @@ -1,3 +1,4 @@ +import {registerAuthorityScanConformance} from "./authority_scan_conformance.ts"; import assert from "node:assert/strict"; import { createHash } from "node:crypto"; import test from "node:test"; @@ -204,6 +205,7 @@ export function registerAuthorityStoreConformance( providerName: string, factory: AuthorityStoreConformanceFactory, ): void { + registerAuthorityScanConformance(providerName, factory); registerNativePlanningUpdateConformance(providerName, factory); for (const native of [false, true]) test(`${providerName} conformance: standing revocation survives canonical ordering and archive (${native ? "native" : "legacy"})`, async (t) => { const {store} = await factory(t); diff --git a/tests/control_plane_ts/authority_store_transactions.test.ts b/tests/control_plane_ts/authority_store_transactions.test.ts index e100a17f4a..4c267215f2 100644 --- a/tests/control_plane_ts/authority_store_transactions.test.ts +++ b/tests/control_plane_ts/authority_store_transactions.test.ts @@ -20,7 +20,6 @@ import { type NoKVStoreIdentityResult, } from "../../loopx/control_plane/coordination/nokv_authority_store.ts"; import { - cloneAuthorityTransaction, decodeAuthorityTransaction, transactionForRevision, } from "../../loopx/control_plane/coordination/authority_store_transactions.ts"; @@ -287,10 +286,47 @@ test("file and NoKV scan results are isolated clones", async (t) => { } }); -test("transaction clone preserves the canonical logical projection", () => { - const decoded = decodeAuthorityTransaction(logicalFixtureTransaction()); - const cloned = cloneAuthorityTransaction(decoded); - assert.deepEqual(transactionForRevision(cloned), expectedRevisionInput()); - (cloned.projection as MutableRecord).changed = true; - assert.equal((decoded.projection as MutableRecord).changed, undefined); + +const journalMutations: readonly [string, (document: MutableRecord) => void][] = [ + ["missing middle", d => (d.committed as MutableRecord[]).splice(1, 1)], + ["repeated operation", d => {const rows = d.committed as MutableRecord[]; rows[1]!.operation_id = rows[0]!.operation_id;}], + ["reordered history", d => (d.committed as MutableRecord[]).reverse()], + ["historical payload", d => {(d.committed as MutableRecord[])[0]!.receipts = [{forged: true}];}], + ["last revision", d => {(d.committed as MutableRecord[]).at(-1)!.provider_revision = "wrong";}], + ["head projection", d => {d.head = {forged: true};}], + ["head cursor", d => {d.cursor = "2";}], + ["head revision", d => {d.provider_revision = "wrong";}], + ["empty history", d => {d.committed = [];}], +]; + +test("retained journal lineage violations fail every consumer without rewriting history", async t => { + for (const [name, createProvider] of providerFactories) { + await t.test(name, async () => { + const provider = await createProvider(); + try { + for (let index = 2; index <= 3; index++) { + const loaded = await provider.store.loadAuthority(); + assert.equal(loaded.status, "loaded"); + if (loaded.status !== "loaded") return; + assert.equal((await provider.store.commitAuthority({...seedCommit, + operation_id: `history-${index}`, expected_provider_revision: loaded.provider_revision})).status, "applied"); + } + const original = await provider.readDocument(); + for (const [label, mutate] of journalMutations) { + const broken = structuredClone(original); + mutate(broken); + await provider.writeDocument(broken); + for (const response of [await provider.store.loadAuthority(), + await provider.store.readReceipt(seedCommit.operation_id), + await provider.store.scanCommitted(null, 1), + await provider.store.commitAuthority({...seedCommit, operation_id: "after-corruption", + expected_provider_revision: original.provider_revision as string})]) { + assert.equal(response.status, "failed", label); + if (response.status === "failed") assert.equal(response.reason_code, "provider_protocol_violation", label); + } + assert.deepEqual(await provider.readDocument(), broken, label); + } + } finally { await provider.cleanup(); } + }); + } }); diff --git a/tests/control_plane_ts/postgresql_authority_store.integration.test.ts b/tests/control_plane_ts/postgresql_authority_store.integration.test.ts index 82d44b3237..7807db4296 100644 --- a/tests/control_plane_ts/postgresql_authority_store.integration.test.ts +++ b/tests/control_plane_ts/postgresql_authority_store.integration.test.ts @@ -129,6 +129,64 @@ if (database && installed) { }; }); + test("PostgreSQL scan binds head and rows to one snapshot during concurrent commit", async t => { + await installed; + const options = {tenant_id: `tenant-${randomUUID()}`, goal_id: `goal-${randomUUID()}`}; + t.after(() => cleanScope(options.tenant_id, options.goal_id)); + const writer = new PostgreSqlAuthorityStore(database, options); + const first = await writer.commitAuthority(commit(null, "snapshot-first", 1, 1)); + assert.equal(first.status, "applied"); + if (first.status !== "applied") return; + let interleaved = false; + const reader = new PostgreSqlAuthorityStore({connect: async () => { + const connection = await database.connect(); + return {...connection, query: async (sql, values) => { + const result = await connection.query(sql, values); + if (!interleaved && sql.includes("FROM loopx_control_plane.authority_heads")) { + interleaved = true; + assert.equal((await writer.commitAuthority(commit(first.provider_revision, + "snapshot-second", 2, 2))).status, "applied"); + } + return result; + }}; + }}, options); + const page = await reader.scanCommitted(null, 10); + assert.equal(interleaved, true); + assert.equal(page.status, "page", JSON.stringify(page)); + if (page.status !== "page") return; + assert.deepEqual(page.transactions.map(row => row.operation_id), ["snapshot-first"]); + assert.equal(page.has_more, false); + const next = await reader.scanCommitted(page.next_cursor, 10); + assert.equal(next.status, "page"); + if (next.status === "page") assert.deepEqual(next.transactions.map(row => row.operation_id), ["snapshot-second"]); + }); + + for (const removedCursor of [2, 3]) { + test(`PostgreSQL scan rejects a missing retained row at cursor ${removedCursor}`, async t => { + await installed; + const options = {tenant_id: `tenant-${randomUUID()}`, goal_id: `goal-${randomUUID()}`}; + t.after(() => cleanScope(options.tenant_id, options.goal_id)); + const store = new PostgreSqlAuthorityStore(database, options); + let revision: string | null = null; + for (let index = 1; index <= 3; index++) { + const result = await store.commitAuthority(commit(revision, `gap-${index}`, index, index)); + assert.equal(result.status, "applied"); + if (result.status !== "applied") return; + revision = result.provider_revision; + } + await pool!.query("DELETE FROM loopx_control_plane.authority_commits WHERE tenant_id=$1 AND goal_id=$2 AND cursor=$3", + [options.tenant_id, options.goal_id, removedCursor]); + for (const limit of [1, 10]) { + const result = await store.scanCommitted(removedCursor === 3 ? "1" : null, limit); + assert.equal(result.status, "failed", JSON.stringify(result)); + if (result.status === "failed") assert.equal(result.reason_code, "provider_protocol_violation"); + } + const head = await store.loadAuthority(); + assert.equal(head.status, "loaded"); + if (head.status === "loaded") assert.equal(head.provider_revision, revision); + }); + } + test("PostgreSQL provider scopes identical goals and operations by tenant", async (t) => { await installed; const goalId = `goal-${randomUUID()}`; From c1f513a44163c7158d1c1d04bf10fbbb71e1953a Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Sat, 12 Sep 2026 23:33:28 +0800 Subject: [PATCH 2/2] docs(authority): define contiguous journal scan and snapshot semantics Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- ...shared-goal-authority-state-provider-v0.md | 25 +++++++++++++++++++ ...-goal-authority-state-provider-v0.zh-CN.md | 20 +++++++++++++++ .../typescript-control-plane-migration-v0.md | 19 ++++++++++++++ ...script-control-plane-migration-v0.zh-CN.md | 13 ++++++++++ 4 files changed, 77 insertions(+) diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md index 765b1bbf0d..6de929c984 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md @@ -758,6 +758,31 @@ separate schemas and roles, and relate them only through opaque identities or digests. Provider-specific payloads do not enter the provider-neutral LoopX schema. +#### Retained-journal scan contract + +`scanCommitted(after_cursor, limit)` binds the head, rows and lookahead to one +read snapshot. The current providers retain a contiguous journal from cursor 1; +`null` is the sole start checkpoint, and every supplied cursor is a positive +canonical decimal string. A positive checkpoint beyond the head, including an +empty store, fails with `scan_cursor_out_of_range` rather than acknowledging +successful exhaustion. Malformed runtime values fail before storage access. + +The shared TS scan owner checks the exact requested interval, including the +lookahead row that proves `has_more`. Missing, repeated or reordered rows and a +last transaction inconsistent with the snapshot head fail as protocol violations. +PostgreSQL metadata/head/row reads use repeatable read; File/NoKV validate one +retained envelope and SQLite keeps its existing read transaction. This does not +introduce a snapshot token across pages: later calls may observe later commits. +A page does not certify rows before its checkpoint, arbitrary payload integrity, +or a future compacted/segmented history format. + +File and NoKV share journal decoding and append construction in the existing +transaction module, while retaining their own revision digest inputs, identities, +CAS and durability effects. Validation covers all four adapters with the shared +complex fixture, real PostgreSQL concurrent commits and disposable corrupted +rows, plus isolated real-source File/PostgreSQL pagination. No active Goal +migration, default-provider change or D1–D3 qualification is implied. + #### 6.2.2 Target store contract after the reference CAS slice The current Stage 2/3 reference implementation deliberately uses the smaller diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md index f61352253e..e1602677f3 100644 --- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md +++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md @@ -638,6 +638,26 @@ commit-marker protocol，并通过新的合同 review。 LoopX 控制面记录与应用领域记录，只通过 opaque identity 或 digest 建立关联。 provider-specific payload 不进入 provider-neutral 的 LoopX schema。 +#### 保留 journal 的扫描合同 + +`scanCommitted(after_cursor, limit)` 将 head、记录和 lookahead 绑定到同一个读取 +snapshot。当前 provider 保留从 cursor 1 起的连续 journal；`null` 是唯一起点， +其他游标必须是规范的正十进制字符串。超过 head 的正数 checkpoint（包括空存储） +返回 `scan_cursor_out_of_range`，不能确认“已成功读完”；非法运行时类型在访问 +存储前拒绝。 + +共享 TS scan owner 验证请求区间及用于证明 `has_more` 的 lookahead 行。 +缺行、重复、乱序及末条 transaction 与 snapshot head 不一致均为协议错误。 +PostgreSQL metadata/head/row 使用 repeatable read；File/NoKV 验证同一个保留 +历史的 envelope，SQLite 保留原读事务。此合同不增加跨页 snapshot token，后续 +调用可以看到后续提交；也不证明 checkpoint 之前全部历史、任意 payload 的完整性， +或未来压缩/分段历史格式。 + +File 与 NoKV 在既有 transaction 模块共用 journal 解码及 append 构造，各自 +保留版本摘要输入、identity、CAS 与持久化副作用。验证包含四个 adapter 的复杂 +fixture、真实 PostgreSQL 并发提交及一次性损坏行，以及真实来源隔离副本上的 +File/PostgreSQL 分页。不迁移活跃 Goal，不切换默认 provider，不宣布 D1–D3 合格。 + #### 6.2.2 参考 CAS 切片之后的目标 store contract 当前 Stage 2/3 参考实现刻意使用上面的较小 `load` / `compare_and_put` document diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index d924fe7397..0d69816274 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -771,6 +771,25 @@ import, provider qualification, soak, or D3 cutover requirements. authority separate. Unknown observations cannot settle Todo/replan work. Explicitly disclose any semantic correction; do not label it full parity. +The retained-journal read boundary now shares one TS owner for scan admission, +checkpoint range, contiguous page coverage, lookahead and final-head agreement. +File and NoKV also share retained-history validation and append construction; +provider revision hashes, physical locks/CAS and backend headers remain local. +This retires duplicated storage-protocol knowledge without a new RPC, Python +bridge, capability or provider. The existing coordination internal owner is +sufficient; built-in File and optional NoKV/SQLite/PostgreSQL implementations +retain their deployment boundaries. + +Intentional corrections: a positive checkpoint against an empty store is +`scan_cursor_out_of_range`; non-string cursors are `invalid_scan_request`; +a missing/reordered retained row or contradictory final head cannot produce a +successful page. PostgreSQL read operations use one repeatable-read snapshot, +so a concurrent commit appears on the next call instead of mixing newer rows +with an older head. The scan proves its requested interval, not an audit of +history before that checkpoint. Successful schemas, File/NoKV persisted bytes, +request identity and revision algorithms remain compatible. This supports T3/D1 +readers but does not finish Todo writers, retention/compaction or promotion. + **T4 — collect full-writer retirement after durability cutover.** - Depends on T1–T3 and the shared RFC's [D1–D3](shared-goal-authority-state-provider-v0.md#durability-execution-cards), including owner approval diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index 5c0b9da00f..7bdfa4780a 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -594,6 +594,19 @@ user role。历史节点不会进入活动工作或 lease lane。无法识别的 - 区分历史监督、canonical 义务与 settlement 权威；unknown 不能结清 Todo/replan。 有意语义修正单独披露，不标成全量 parity。 +保留 journal 的读取边界现由同一个 TS owner 负责扫描参数、checkpoint 范围、 +分页连续性、lookahead 和末行/head 一致性。File 与 NoKV 同时共用历史校验及 +append 构造，版本哈希、物理锁/CAS 和后端头字段仍归各 provider。这删除了重复 +存储协议知识，没有新增 RPC、Python bridge、capability 或 provider；既有 +coordination 内部 owner 足够，File 内置及 NoKV/SQLite/PostgreSQL 可选部署边界不变。 + +明确修正：空存储上的正数 checkpoint 返回 `scan_cursor_out_of_range`，非字符串 +游标返回 `invalid_scan_request`；历史缺行、乱序或末行/head 矛盾不能返回成功分页。 +PostgreSQL 读取使用同一个 repeatable-read snapshot，并发提交在下一次调用可见， +不会将较新的行混入较旧 head。扫描只证明请求区间，不审计 checkpoint 之前的全部 +历史。合法结果 schema、File/NoKV 持久字节、请求身份及版本算法保持兼容。这支持 +T3/D1 reader，未完成全部 Todo writer、retention/compaction 或 promotion。 + **T4 — durable cutover 后兑现完整 writer 删除。** - 前提是 T1–T3 和 shared RFC 的 [D1–D3](shared-goal-authority-state-provider-v0.zh-CN.md#持久化执行卡)，包括 owner 批准及明确的 legacy 迁移窗口。