From a79db434acbc1b3476ca97b979e291375af9bf86 Mon Sep 17 00:00:00 2001 From: David Snelling Date: Wed, 2 Sep 2026 10:53:54 -0700 Subject: [PATCH] =?UTF-8?q?fix(generation-store):=20commitTransaction=20re?= =?UTF-8?q?fuses=20while=20single-ops=20are=20pending=20=E2=80=94=20the=20?= =?UTF-8?q?order=20invariant=20is=20enforced,=20not=20assumed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reservedGensAsc() documented but never enforced that pending single-op generations must sort above every committed one. A direct commitTransaction() call bypassing Brainy.transact()'s flush-first step could commit a fresh generation into committedRanges above lower, still-pending ones, unsorting the committed-then-pending walk resolveManyAt relies on and returning a wrong before-image for a point-in-time read — silently. commitTransaction() now refuses via a new PendingSingleOpsUnflushedError when the pending tier is non-empty, before any staging I/O. Behavior-neutral: both sanctioned callers (Brainy.transact(), Brainy.compactHistory()) already flush first. --- src/db/errors.ts | 60 +++++ src/db/generationStore.ts | 56 +++- src/index.ts | 3 +- .../db/generationStore-commit-guard.test.ts | 254 ++++++++++++++++++ 4 files changed, 371 insertions(+), 2 deletions(-) create mode 100644 tests/unit/db/generationStore-commit-guard.test.ts diff --git a/src/db/errors.ts b/src/db/errors.ts index 3b4c1af6..da62eb0b 100644 --- a/src/db/errors.ts +++ b/src/db/errors.ts @@ -351,3 +351,63 @@ export class PendingFlushDurabilityError extends Error { this.failedAttempts = failedAttempts } } + +/** + * @description Thrown by {@link GenerationStore.commitTransaction} when the + * PENDING single-op tier is non-empty — i.e. one or more `commitSingleOp()` + * generations are buffered in memory, not yet flushed to + * `committedRanges` via `flushPendingSingleOps()`. + * + * The invariant `reservedGensAsc()` (and everything built on it — + * `resolveManyAt`, `resolveAt`, `changedBetween`, the hot-tail window) relies + * on is documented, not enforced by types: pending generations must always be + * numerically greater than every committed one, because the ONLY sanctioned + * callers of `commitTransaction()` — `Brainy.transact()` and + * `Brainy.compactHistory()` — flush the pending tier FIRST. A caller that + * invokes `commitTransaction()` directly while single-ops are still pending + * breaks that invariant: the new commit lands in `committedRanges` ABOVE + * generations still sitting in `pendingGens`, so the committed-then-pending + * concatenation `reservedGensAsc()` yields is no longer ascending. The + * concrete failure this produces is silent, not a crash: `resolveManyAt` + * walks committed ranges before pending ones, so it can report a NEWER + * generation as the "first after" a pin than an older, still-pending one that + * actually touched the id first — a wrong before-image at a point-in-time + * read, without a compensating error to warn a caller anything went wrong. + * + * This error refuses the commit outright, before any staging I/O: nothing is + * written, the generation counter reservation is untouched, and + * `committedRanges`/`pendingGens` are exactly as they were. Call + * `flushPendingSingleOps()` first (or go through `Brainy.transact()`, which + * already does). + * + * @example + * try { + * await generationStore.commitTransaction({ touched, execute }) + * } catch (err) { + * if (err instanceof PendingSingleOpsUnflushedError) { + * await generationStore.flushPendingSingleOps() + * await generationStore.commitTransaction({ touched, execute }) // now safe + * } + * } + */ +export class PendingSingleOpsUnflushedError extends Error { + /** How many un-flushed single-op generations were buffered at refusal time. */ + public readonly pendingCount: number + + /** + * @param pendingCount - `pendingGens.length` at the moment of refusal (always ≥ 1). + */ + constructor(pendingCount: number) { + super( + `commitTransaction() refused: ${pendingCount} pending single-op generation(s) ` + + `are still buffered and un-flushed. Flush the pending single-op tier before ` + + `committing a transaction — Brainy.transact() does this automatically; a ` + + `direct commitTransaction() call with pending generations would leave the ` + + `generation order unsorted (committed generations landing above lower, ` + + `still-pending ones) and make point-in-time reads (resolveManyAt/resolveAt) ` + + `return the wrong before-image. Call flushPendingSingleOps() first, then retry.` + ) + this.name = 'PendingSingleOpsUnflushedError' + this.pendingCount = pendingCount + } +} diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index da21dc61..128f905b 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -32,7 +32,13 @@ */ import { prodLog } from '../utils/logger.js' -import { GenerationCompactedError, GenerationConflictError, PendingFlushDurabilityError, StoreInconsistentError } from './errors.js' +import { + GenerationCompactedError, + GenerationConflictError, + PendingFlushDurabilityError, + PendingSingleOpsUnflushedError, + StoreInconsistentError +} from './errors.js' import type { UnreconciledRecord } from './errors.js' import { TransactionRollbackError } from '../transaction/errors.js' import type { @@ -1351,6 +1357,9 @@ export class GenerationStore { * @param args.execute - Runs the planned operation batch atomically. * @returns The committed generation and its commit timestamp. * @throws GenerationConflictError when the CAS expectation fails. + * @throws PendingSingleOpsUnflushedError when the pending single-op tier is + * non-empty — call `flushPendingSingleOps()` first (both `Brainy.transact()` + * and `Brainy.compactHistory()` already do). */ /** * The generation fact log, or `null` when the storage layer cannot host one. @@ -1425,6 +1434,13 @@ export class GenerationStore { execute: () => Promise }): Promise<{ generation: number; timestamp: number }> { return this.withMutex(async () => { + // The generation-order guard (see assertPendingSingleOpsFlushed): a + // direct commitTransaction() call while single-ops are still pending + // would commit above them, unsorting reservedGensAsc() and corrupting + // point-in-time reads. Both sanctioned callers (Brainy.transact(), + // Brainy.compactHistory()) already flush first, so this is + // behavior-neutral on every real path. + this.assertPendingSingleOpsFlushed() // A latched history-durability failure compromises the whole generation // chain — refuse a transact too (advancing the manifest past stuck, // un-durable single-op generations would be inconsistent). Same loud @@ -2294,6 +2310,37 @@ export class GenerationStore { } } + /** + * @description Throw if the pending single-op tier is non-empty. Called at + * the top of {@link commitTransaction} (the ONLY method that appends a + * fresh commit directly into {@link committedRanges} outside recovery) so + * the ordering invariant {@link reservedGensAsc}'s own doc comment states — + * "pending generations are always greater than every committed one" — is + * ENFORCED there rather than merely assumed. + * + * That invariant holds today only because both sanctioned callers flush the + * pending tier before committing: `Brainy.transact()` (src/brainy.ts, + * `await this.generationStore.flushPendingSingleOps()` immediately before + * its `commitTransaction()` call) and `Brainy.compactHistory()` + * (src/brainy.ts, the same flush immediately before its `compact()` call — + * `compact()` itself only ever RECLAIMS an existing committed prefix, so it + * cannot land a commit out of order and needs no guard of its own). A + * caller that reaches `commitTransaction()` by any other path — bypassing + * that flush — would commit a new generation into `committedRanges` ABOVE + * generations still sitting in `pendingGens`, breaking `reservedGensAsc`'s + * "committed-then-pending is already sorted" assumption and making + * `resolveManyAt`'s single ascending pass (and `resolveAt`'s consumers) + * return the WRONG before-image for a point-in-time read — silently, no + * compensating error. Refusing here, before any staging I/O, keeps the + * store untouched (nothing committed, nothing staged, the generation + * counter reservation unaffected) on every path that already flushes. + */ + private assertPendingSingleOpsFlushed(): void { + if (this.pendingGens.length > 0) { + throw new PendingSingleOpsUnflushedError(this.pendingGens.length) + } + } + /** Schedule a coalesced pending-tier flush (size trigger fires immediately on * the next microtask; otherwise a {@link PENDING_FLUSH_DELAY_MS} timer). Both * defer outside the current mutex section so the flush can re-acquire it. A @@ -2377,6 +2424,13 @@ export class GenerationStore { * committed-then-pending concatenation is already sorted — identical to the old * `[...committedGens, ...pendingGens]`. This is the union historical reads * resolve over so un-flushed single-ops are visible to pins/`asOf`. + * + * The "flush first" half of that invariant is ENFORCED, not just documented: + * {@link commitTransaction} — the only method that lands a fresh commit into + * {@link committedRanges} outside crash recovery — refuses via + * {@link assertPendingSingleOpsFlushed} whenever {@link pendingGens} is + * non-empty, so a committed generation can never land above a still-pending + * one and break this ordering. */ private *reservedGensAsc(): IterableIterator { yield* this.committedGensAsc() diff --git a/src/index.ts b/src/index.ts index edc21809..e946f15c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -231,7 +231,8 @@ export { GenerationCompactedError, StoreInconsistentError, PendingFlushDurabilityError, - CanonicalEnumerationUnavailableError + CanonicalEnumerationUnavailableError, + PendingSingleOpsUnflushedError } from './db/errors.js' export type { UnreconciledRecord } from './db/errors.js' export type { diff --git a/tests/unit/db/generationStore-commit-guard.test.ts b/tests/unit/db/generationStore-commit-guard.test.ts new file mode 100644 index 00000000..d449f8ef --- /dev/null +++ b/tests/unit/db/generationStore-commit-guard.test.ts @@ -0,0 +1,254 @@ +/** + * @module tests/unit/db/generationStore-commit-guard + * @description Pins the commit-order guard on + * `GenerationStore.commitTransaction()` (`src/db/generationStore.ts`). + * + * `reservedGensAsc()`'s own doc comment states an invariant it never + * enforced: pending single-op generations are always greater than every + * committed one, because the store's only two sanctioned callers — + * `Brainy.transact()` and `Brainy.compactHistory()` — flush the pending tier + * before committing. Nothing stopped a caller from invoking + * `commitTransaction()` directly while single-ops were still buffered: the + * fresh commit would land in `committedRanges` ABOVE those lower, + * still-pending generations, so the committed-then-pending concatenation + * `reservedGensAsc()` yields is no longer ascending — and `resolveManyAt` + * (which walks committed ranges before pending ones) would silently report a + * WRONG before-image for a point-in-time read. `commitTransaction()` now + * refuses loudly (`PendingSingleOpsUnflushedError`) instead of assuming. + * + * Four pins: + * 1. A direct `commitTransaction()` call while single-ops are pending throws + * and commits NOTHING. + * 2. The same commit succeeds once the pending tier is flushed first. + * 3. `Brainy.transact()` — which already flushes first — is unaffected + * (mirrors `tests/unit/db/generation-chain.test.ts`'s `seedX()`/`bumpX()` + * transact pin: add, then transact-update, generation advances by one + * each time, the update lands). + * 4. `reservedGensAsc()` stays ascending across a real add+transact+delete + * workload — proven by point-in-time reads (`asOf`) staying correct + * throughout, which is exactly what an ordering break would corrupt. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' +import { + GenerationStore, + GENERATIONS_PREFIX, + MANIFEST_PATH +} from '../../../src/db/generationStore.js' +import { PendingSingleOpsUnflushedError } from '../../../src/db/errors.js' +import { Brainy } from '../../../src/index.js' +import { NounType } from '../../../src/types/graphTypes.js' +import { createTestConfig, generateTestVector } from '../../helpers/test-factory.js' + +/** Precomputed embedding so Brainy-level adds skip the (slow) embedding model — + * these tests exercise the generation layer, not semantics. */ +const VEC = generateTestVector() + +// Entity ids must be UUID-shaped (the sharded storage layout derives the +// shard from the UUID hex) — same fixture convention as generationStore.test.ts. +const ID_A = '00000000-0000-4000-8000-0000000000aa' +const ID_B = '00000000-0000-4000-8000-0000000000bb' + +/** Stored-metadata fixture in the canonical shape the live write paths use + * (matches generationStore.test.ts's fixture exactly). */ +function metadataFixture(version: number): Record { + return { + noun: NounType.Document, + subtype: 'note', + data: `payload-v${version}`, + version, + createdAt: 1000, + updatedAt: 1000 + version, + _rev: version + } +} + +describe('db/GenerationStore — commitTransaction pending-tier guard (store level)', () => { + let storage: MemoryStorage + let store: GenerationStore + + beforeEach(async () => { + storage = new MemoryStorage() + await storage.init() + store = new GenerationStore(storage) + await store.open() + }) + + /** Buffer one single-op generation via commitSingleOp WITHOUT flushing — + * the pending tier that must be drained before commitTransaction(). */ + async function pendingSingleOp(id: string, version: number): Promise { + const { generation } = await store.commitSingleOp({ + touched: { nouns: [id] }, + execute: async () => { + await storage.saveNounMetadata(id, metadataFixture(version)) + } + }) + return generation + } + + /** A direct transact commit — exactly what a caller bypassing + * Brainy.transact()'s flush-first step would issue. */ + function directCommit(id: string, version: number): Promise<{ generation: number; timestamp: number }> { + return store.commitTransaction({ + touched: { nouns: [id], verbs: [] }, + execute: async () => { + await storage.saveNounMetadata(id, metadataFixture(version)) + } + }) + } + + it('PIN 1: refuses a direct commitTransaction() while single-ops are pending, and commits NOTHING', async () => { + const g1 = await pendingSingleOp(ID_A, 1) + expect(g1).toBe(1) + expect(store.committedGeneration()).toBe(0) // nothing flushed to disk yet + + let caught: unknown + try { + await directCommit(ID_B, 1) + expect.unreachable('should have thrown PendingSingleOpsUnflushedError') + } catch (err) { + caught = err + } + expect(caught).toBeInstanceOf(PendingSingleOpsUnflushedError) + expect((caught as PendingSingleOpsUnflushedError).pendingCount).toBe(1) + + // Nothing committed: the head + committed ranges are unchanged, and the + // counter never advanced for the refused attempt (the guard fires before + // a generation is even reserved). + expect(store.committedGeneration()).toBe(0) + expect(store.generation()).toBe(1) // still just the pending single-op's gen + expect(await storage.readRawObject(MANIFEST_PATH)).toBeNull() + // The guard fires BEFORE a generation is reserved (`gen = ++this.counter` + // never runs), so the refused attempt's would-be directory (generation 2, + // the next number after the pending single-op's 1) was never created. + expect(await storage.listRawObjects(`${GENERATIONS_PREFIX}/2`)).toEqual([]) + + // The refused write never touched canonical storage. + expect((await storage.readNounRaw(ID_B)).metadata).toBeNull() + + // The pending tier itself is untouched by the refused attempt — flushing + // now still commits the ORIGINAL single-op cleanly. + await store.flushPendingSingleOps() + expect(store.committedGeneration()).toBe(1) + const atG0 = await store.resolveAt('noun', ID_A, 0) + expect(atG0).toEqual({ source: 'absent' }) // the create sentinel before g1's write + }) + + it('PIN 2: the same commit succeeds once the pending tier is flushed first', async () => { + await pendingSingleOp(ID_A, 1) + await expect(directCommit(ID_B, 1)).rejects.toBeInstanceOf(PendingSingleOpsUnflushedError) + + await store.flushPendingSingleOps() + expect(store.committedGeneration()).toBe(1) + + const { generation } = await directCommit(ID_B, 1) + expect(generation).toBe(2) + expect(store.committedGeneration()).toBe(2) + expect((await storage.readNounRaw(ID_B)).metadata).toMatchObject({ version: 1 }) + }) +}) + +describe('Brainy public API — commitTransaction pending-tier guard is behavior-neutral', () => { + let brain: Brainy + + beforeEach(async () => { + brain = new Brainy(createTestConfig()) + await brain.init() + }) + afterEach(async () => { + await brain.close() + }) + + it('PIN 3: Brainy.transact() still commits normally over pending single-ops (mirrors generation-chain.test.ts\'s seedX()/bumpX() transact pin)', async () => { + const store = (brain as any).generationStore as GenerationStore + // Relative, not absolute: under the adopt-at-open default the open-time + // baseline backfill takes a generation of its own (see + // bounded-chains.test.ts's identical note), so the first user add is not + // necessarily generation 1. + const baseGen = brain.generation() + const baseCommitted = store.committedGeneration() + + const id = await brain.add({ + data: 'x', + type: NounType.Document, + subtype: 'note', + metadata: { v: 1 }, + vector: VEC + }) + // The add is a pending single-op generation — NOT yet flushed. + expect(brain.generation()).toBe(baseGen + 1) + expect(store.committedGeneration()).toBe(baseCommitted) + + // Brainy.transact() flushes the pending tier FIRST (src/brainy.ts: + // `await this.generationStore.flushPendingSingleOps()`, immediately + // before its `generationStore.commitTransaction()` call), so the guard + // never fires on this path — same shape as generation-chain.test.ts's + // seedX() (add) → bumpX() (transact update) → generation advances by one. + const db = await brain.transact([{ op: 'update', id, metadata: { v: 2 } }]) + await db.release() + + expect(brain.generation()).toBe(baseGen + 2) + expect(store.committedGeneration()).toBe(baseGen + 2) // the flushed add + the transact update + const entity = (await brain.get(id)) as any + expect(entity.metadata.v).toBe(2) + }) + + it('PIN 4: reservedGensAsc() stays ascending across a real add+transact+delete workload — point-in-time reads stay correct', async () => { + const store = (brain as any).generationStore as GenerationStore + const baseGen = brain.generation() + const baseCommitted = store.committedGeneration() + + const idX = await brain.add({ + data: 'x', + type: NounType.Document, + subtype: 'note', + metadata: { v: 1 }, + vector: VEC + }) + expect(brain.generation()).toBe(baseGen + 1) // pending (un-flushed) + + const idY = await brain.add({ + data: 'y', + type: NounType.Document, + subtype: 'note', + metadata: { v: 1 }, + vector: VEC + }) + // Pin right after BOTH adds — before the transact update — so X reads v1 + // and Y still exists at this pin, unlike the live head after the rest of + // the workload runs. + const pinAfterBothAdds = brain.generation() + expect(pinAfterBothAdds).toBe(baseGen + 2) // ALSO pending — two un-flushed single-ops + expect(store.committedGeneration()).toBe(baseCommitted) + + // A transact() flushes baseGen+1 and baseGen+2 first, then commits its + // own update as baseGen+3. If committed-vs-pending ordering ever broke, + // this is exactly the step that would land a commit ABOVE still-pending + // generations. + const db = await brain.transact([{ op: 'update', id: idX, metadata: { v: 3 } }]) + await db.release() + expect(brain.generation()).toBe(baseGen + 3) + expect(store.committedGeneration()).toBe(baseGen + 3) + + // A single-op delete, pending again (un-flushed). + await brain.remove(idY) + expect(brain.generation()).toBe(baseGen + 4) + + // A point-in-time read pinned right after the two adds (before the + // transact update) must see X's PRE-update value and Y still present. + // This is precisely what resolveManyAt/resolveAt get WRONG if committed + // and pending generations were ever interleaved out of ascending order. + const past = await brain.asOf(pinAfterBothAdds) + const xAtPin = (await past.get(idX)) as any + expect(xAtPin?.metadata?.v).toBe(1) + const yAtPin = (await past.get(idY)) as any + expect(yAtPin?.metadata?.v).toBe(1) // not yet removed, as of this pin + await past.release() + + // Live state reflects every later write, in the right order. + const xNow = (await brain.get(idX)) as any + expect(xNow.metadata.v).toBe(3) + expect(await brain.get(idY)).toBeNull() + }) +})