diff --git a/src/brainy.ts b/src/brainy.ts index addb8bc2..d1ec144b 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -2245,7 +2245,8 @@ export class Brainy implements BrainyInterface { }, undefined, undefined, - [{ type: 'embed.landed', id, vector: newVector }] + [{ type: 'embed.landed', id, vector: newVector }], + 'system:embed-landing' ) this.clearPendingEmbed(id) } catch (err) { @@ -2479,7 +2480,8 @@ export class Brainy implements BrainyInterface { run: TransactionFunction, precommit?: (before: CommitBeforeImages) => void, pendingEvents?: PendingChangeEvent[], - records?: FactMarkerRecord[] + records?: FactMarkerRecord[], + origin?: string ): Promise<{ generation?: number; timestamp: number; degraded?: string[] }> { // Change-feed capture: when this write will emit, hold a reference to the // commit's before-images so `remove` events can carry the record's last @@ -2542,6 +2544,7 @@ export class Brainy implements BrainyInterface { touched, precommit: captureAndCheck, ...(records && records.length > 0 ? { records } : {}), + ...(origin ? { origin } : {}), execute: () => this.transactionManager.executeTransaction(run, { timeout: transactTimeoutBudget( @@ -8358,7 +8361,7 @@ export class Brainy implements BrainyInterface { } } }) - }) + }, undefined, undefined, undefined, 'system:adoption-backfill') } const next = await this.runOracle({ listAll: true }) // THE ONLY STOP: no progress. With uncapped listings both counts are @@ -8392,6 +8395,137 @@ export class Brainy implements BrainyInterface { return report } + /** + * @description THE ATTESTED PER-ID RECONCILE DOOR for the one divergence + * class the adoption backfill refuses BY DESIGN: `log-live-canonical-absent` + * — the log holds a live record for a row the canonical tree says does not + * exist. The engine cannot tell a legitimate pre-log deletion (the log + * missed the tombstone — the deferred-durability-era ack-window class) from + * canonical LOSS (the log holds the only surviving copy); auto-curing would + * silently destroy data in one of the two readings. A HUMAN attests which: + * + * - `attest: 'deleted'` — the row was legitimately deleted; mint the + * tombstone fact the log always lacked (canonical stays absent). The + * log's history keeps the old live record — as-of reads before the + * tombstone still see it. + * - `attest: 'restore'` — canonical lost the row; fold the log's latest + * after-image back into canonical (both sides now agree it lives). + * + * Loud, narrated, single-row, and stamped `origin: 'system:reconcile'` on + * both the tx-log entry and the commit fact. Refuses (typed) when the id's + * log and canonical already agree, when `restore` is attested but the log + * holds no record, and when canonical is PRESENT-but-different (that is + * `state-differs` — `adoptLogAuthority()`'s backfill owns it). + * + * @param id - The single entity id to reconcile. + * @param options.attest - The human's word on which reading is true. + * @returns What was done and the generation that recorded it. + * @throws When the divergence is not the attested class (nothing is written). + */ + async reconcileLogDivergence( + id: string, + options: { attest: 'deleted' | 'restore' } + ): Promise<{ reconciled: 'tombstoned' | 'restored'; id: string; generation: number }> { + await this.ensureInitialized() + this.assertWritable('reconcileLogDivergence') + + // Fold the log for THIS id (one scan; a rare operator door). + const scan = this.scanFacts() + if (!scan) { + throw new Error('reconcileLogDivergence: this store has no fact log — nothing to reconcile against') + } + let logLatest: { tombstoned: boolean; record: { metadata: unknown; vector: unknown } | null } | null = null + for await (const batch of scan.batches()) { + for (const fact of batch.facts) { + for (const op of fact.ops) { + if (op.kind === 'noun' && op.id === id) { + logLatest = + op.record === null + ? { tombstoned: true, record: null } + : { tombstoned: false, record: { metadata: op.record.metadata, vector: op.record.vector } } + } + } + } + } + const canonical = await this.storage.readNounRaw(id) + const canonicalAbsent = canonical.metadata === null && canonical.vector === null + + // Only the log-live + canonical-absent shape passes; everything else + // names its actual state and the door that owns it. + if (!logLatest || logLatest.tombstoned) { + throw new Error( + `reconcileLogDivergence(${id}): the log's latest state is ` + + `${logLatest ? 'a tombstone' : 'no record at all'} — there is no ` + + `log-live-canonical-absent divergence here. If the oracle reports this id, ` + + `re-run verifyLogAuthority() for the current class.` + ) + } + if (!canonicalAbsent) { + throw new Error( + `reconcileLogDivergence(${id}): canonical is PRESENT — this is not the ` + + `log-live-canonical-absent class. If canonical differs from the log ` + + `(state-differs), adoptLogAuthority()'s backfill cures it; nothing was written.` + ) + } + + if (options.attest === 'deleted') { + // Mint the tombstone fact the log always lacked. writeNounRaw with null + // parts is an idempotent delete; the commit fact reads canonical back + // after execute (absent) and records the tombstone. + const receipt = await this.persistSingleOp( + { nouns: [id] }, + async (tx) => { + tx.addOperation({ + name: 'ReconcileTombstone', + execute: async () => { + await this.storage.writeNounRaw(id, { metadata: null, vector: null }) + return async () => { + // Undo of an idempotent delete of an absent row: nothing. + } + } + }) + }, + undefined, + undefined, + undefined, + 'system:reconcile' + ) + prodLog.warn( + `[Brainy] reconcileLogDivergence: ${id} attested DELETED — tombstone fact minted ` + + `at generation ${receipt.generation}; the log now agrees the row is gone ` + + `(its history keeps the earlier live record).` + ) + return { reconciled: 'tombstoned', id, generation: receipt.generation! } + } + + // attest: 'restore' — the log's copy is the survivor; fold it back. + const record = logLatest.record! + const receipt = await this.persistSingleOp( + { nouns: [id] }, + async (tx) => { + tx.addOperation({ + name: 'ReconcileRestore', + execute: async () => { + await this.storage.writeNounRaw(id, record) + return async () => { + await this.storage.writeNounRaw(id, { metadata: null, vector: null }) + } + } + }) + }, + undefined, + undefined, + undefined, + 'system:reconcile' + ) + prodLog.warn( + `[Brainy] reconcileLogDivergence: ${id} attested RESTORE — the log's latest ` + + `after-image was folded back into canonical at generation ${receipt.generation}. ` + + `Derived indexes reconcile at next open/repairIndex; the row serves from canonical now.` + ) + return { reconciled: 'restored', id, generation: receipt.generation! } + } + /** * @description Read the reified transaction log — one entry per committed * generation, carrying the committed generation, the commit timestamp, and diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index 4002c1ba..7439a025 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -428,7 +428,13 @@ export class GenerationStore { private pendingGens: number[] = [] private readonly pendingBuffer = new Map< number, - { nouns: Map; verbs: Map; timestamp: number } + { + nouns: Map + verbs: Map + timestamp: number + /** Engine-origin stamp for the tx-log entry (absent = user write). */ + origin?: string + } >() /** Pending timer-coalesced flush handle (cleared on flush/close). */ private pendingFlushTimer: ReturnType | null = null @@ -1642,6 +1648,12 @@ export class GenerationStore { * surfacing that honestly. */ records?: FactMarkerRecord[] + /** + * Engine-origin stamp (`'system:embed-landing'`, `'system:adoption-backfill'`, + * `'system:reconcile'`). Rides the tx-log entry AND the commit fact's meta, + * so both records agree about WHO committed. Absent = user write. + */ + origin?: string }): Promise<{ generation: number; timestamp: number; degraded?: string[] }> { return this.withMutex(async () => { // Refuse to accept a write whose history we cannot make durable: if the @@ -1710,7 +1722,7 @@ export class GenerationStore { // incomplete for these ids until the next rebuild/repairIndex (the // egress guard prevents wrong results meanwhile). Loud, honest, // no double-write. - this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp }) + this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp, ...(args.origin ? { origin: args.origin } : {}) }) this.pendingGens.push(gen) this.extendChains(gen, nouns, verbs) // The adopted generation is committed — it gets its fact like any @@ -1723,6 +1735,7 @@ export class GenerationStore { timestamp, nouns, verbs, + ...(args.origin ? { meta: { origin: args.origin } } : {}), ...(args.records && args.records.length > 0 ? { records: args.records } : {}) }) ) @@ -1763,7 +1776,7 @@ export class GenerationStore { if (this.commitFaultInjector) this.commitFaultInjector('singleop-after-execute') // Buffer the pending generation + make it instantly visible to reads. - this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp }) + this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp, ...(args.origin ? { origin: args.origin } : {}) }) this.pendingGens.push(gen) this.extendChains(gen, nouns, verbs) // Fact log (dual-write): the acked write's AFTER-IMAGE fact, appended @@ -1802,6 +1815,7 @@ export class GenerationStore { timestamp, nouns, verbs, + ...(args.origin ? { meta: { origin: args.origin } } : {}), ...(args.records && args.records.length > 0 ? { records: args.records } : {}) }) ) @@ -1958,7 +1972,7 @@ export class GenerationStore { const deltaPath = `${dir}/tx.json` await this.storage.writeRawObject(deltaPath, delta) stagedPaths.push(deltaPath) - logEntries.push({ generation: gen, timestamp: buf.timestamp }) + logEntries.push({ generation: gen, timestamp: buf.timestamp, ...(buf.origin ? { origin: buf.origin } : {}) }) } // Test-only crash simulation. A crash here must cost only the window's diff --git a/src/db/types.ts b/src/db/types.ts index 2c7eab8f..866ca47f 100644 --- a/src/db/types.ts +++ b/src/db/types.ts @@ -412,6 +412,17 @@ export interface TxLogEntry { timestamp: number /** Transaction metadata, when supplied to `transact()`. */ meta?: Record + /** + * WHO committed. Absent = a user write (every pre-existing consumer's + * reading stays exact). Engine-originated commits stamp themselves — + * `'system:embed-landing'` (the deferred vector landing), + * `'system:adoption-backfill'` (baseline re-commits), `'system:reconcile'` + * (the attested per-id divergence door) — so activity feeds can filter on + * fact instead of collapsing near-in-time entries (a consumer refused that + * heuristic as a quiet loss, correctly; this field is the honest cure). + * The same stamp rides the commit fact's meta, so log and tx-log agree. + */ + origin?: string } // ============================================================================ diff --git a/tests/integration/txlog-origin-and-reconcile.test.ts b/tests/integration/txlog-origin-and-reconcile.test.ts new file mode 100644 index 00000000..fbe05fcf --- /dev/null +++ b/tests/integration/txlog-origin-and-reconcile.test.ts @@ -0,0 +1,142 @@ +/** + * @module tests/integration/txlog-origin-and-reconcile + * @description Two consumer-driven cures, pinned together because they share + * the origin stamp: + * + * 1. TX-LOG ORIGIN — engine-originated commits stamp `origin` on their + * tx-log entry (and the commit fact's meta) so activity feeds filter on + * fact: a downstream feed showed a "double tick" because the deferred + * vector-landing commit was indistinguishable from a user save, and the + * consumer rightly refused a time-window collapse as a quiet loss. User + * writes stay UNSTAMPED (absent origin) — the pre-existing reading of + * every consumer is exact. + * + * 2. THE RECONCILE DOOR — `log-live-canonical-absent` refuses auto-cure by + * design (a legitimate lost-tombstone deletion is indistinguishable from + * canonical loss); `reconcileLogDivergence(id, {attest})` is the human's + * door: 'deleted' mints the missing tombstone, 'restore' folds the log's + * copy back, wrong-class calls refuse typed with nothing written. + */ +import { describe, it, expect, afterEach } from 'vitest' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { Brainy } from '../../src/index.js' +import { NounType } from '../../src/types/graphTypes.js' + +type RawBox = { + storage: { + readNounRaw(id: string): Promise<{ metadata: unknown; vector: unknown }> + writeNounRaw(id: string, r: { metadata: unknown; vector: unknown }): Promise + } +} + +const dirs: string[] = [] +const brains: Brainy[] = [] +afterEach(async () => { + for (const b of brains.splice(0)) await b.close().catch(() => {}) + for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true }) +}) + +async function fsBrain(): Promise { + const dir = mkdtempSync(join(tmpdir(), 'brainy-origin-reconcile-')) + dirs.push(dir) + const brain = new Brainy({ + storage: { type: 'filesystem', path: dir }, + requireSubtype: false + }) + await brain.init() + brains.push(brain) + return brain +} + +describe('tx-log origin stamp', () => { + it('the deferred-embed landing commit is stamped system:embed-landing; the user write is not', async () => { + const brain = await fsBrain() + await brain.add({ + data: 'a row whose vector lands later', + type: NounType.Document, + metadata: { k: 1 }, + deferEmbedding: true + }) + await brain.awaitPendingEmbeds() + await brain.flush() + + const entries = await brain.transactionLog() + const system = entries.filter((e) => (e as { origin?: string }).origin === 'system:embed-landing') + const user = entries.filter((e) => !(e as { origin?: string }).origin) + expect(system.length, 'the landing commit is stamped').toBeGreaterThanOrEqual(1) + expect(user.length, 'the user add stays unstamped').toBeGreaterThanOrEqual(1) + // The feed cure in one line: filtering !origin removes the double tick. + expect(user.length).toBeLessThan(entries.length) + }, 120000) +}) + +describe('reconcileLogDivergence — the attested door', () => { + /** Manufacture the class: a live log record whose canonical row is gone. */ + async function manufactureDivergence(brain: Brainy): Promise { + const id = await brain.add({ + data: 'pre-era row whose deletion the log never saw', + type: NounType.Document, + metadata: { era: 'pre-spine' } + }) + await brain.flush() + // Delete canonical BEHIND the log's back (raw write, no generation) — + // exactly the shape a deferred-durability-era crash left behind. + const storage = (brain as unknown as RawBox).storage + await storage.writeNounRaw(id, { metadata: null, vector: null }) + return id + } + + it("attest:'deleted' mints the missing tombstone — the oracle goes green and the commit is stamped system:reconcile", async () => { + const brain = await fsBrain() + const id = await manufactureDivergence(brain) + const before = await brain.verifyLogAuthority() + expect( + before.mismatches.some((m) => m.id === id && m.reason === 'log-live-canonical-absent'), + 'the manufactured divergence is oracle-visible as the refused class' + ).toBe(true) + + const result = await brain.reconcileLogDivergence(id, { attest: 'deleted' }) + expect(result.reconciled).toBe('tombstoned') + + const after = await brain.verifyLogAuthority() + expect(after.mismatches.some((m) => m.id === id), 'the id no longer diverges').toBe(false) + expect(await brain.get(id), 'canonical stays absent').toBeNull() + + await brain.flush() + const entries = await brain.transactionLog() + expect( + entries.some((e) => (e as { origin?: string }).origin === 'system:reconcile'), + 'the reconcile commit is origin-stamped' + ).toBe(true) + }, 120000) + + it("attest:'restore' folds the log's copy back into canonical", async () => { + const brain = await fsBrain() + const id = await manufactureDivergence(brain) + + const result = await brain.reconcileLogDivergence(id, { attest: 'restore' }) + expect(result.reconciled).toBe('restored') + + const row = await brain.get(id) + expect(row, 'the log’s only copy lives again').not.toBeNull() + expect((row!.metadata as { era: string }).era).toBe('pre-spine') + expect((await brain.verifyLogAuthority()).mismatches.some((m) => m.id === id)).toBe(false) + }, 120000) + + it('wrong-class calls refuse typed with nothing written', async () => { + const brain = await fsBrain() + const id = await brain.add({ data: 'healthy row', type: NounType.Document, metadata: { n: 1 } }) + await brain.flush() + // Canonical present + log agrees: not the class — refuse, name the state. + await expect(brain.reconcileLogDivergence(id, { attest: 'deleted' })).rejects.toThrow( + /canonical is PRESENT/ + ) + expect(await brain.get(id), 'nothing was written').not.toBeNull() + // Unknown id: no log record at all — refuse, name it. + await expect( + brain.reconcileLogDivergence('00000000-0000-7000-8000-00000000dead', { attest: 'restore' }) + ).rejects.toThrow(/no record at all/) + }, 120000) +})