From ff43de1ada9f2c48c6d62e2723aebf0e9aeddb30 Mon Sep 17 00:00:00 2001 From: David Snelling Date: Wed, 12 Aug 2026 16:56:08 -0700 Subject: [PATCH] =?UTF-8?q?feat(recovery):=20the=20fold-checkpoint=20bound?= =?UTF-8?q?=20=E2=80=94=20crash=20folds=20(checkpoint,=20head],=20never=20?= =?UTF-8?q?the=20whole=20log=20twice?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fold checkpoint (_system/fold-checkpoint.json) is stamped strictly after a canonical-sync barrier over every live entity touched since the last stamp (syncEntityCanonical: ids → canonical paths → fsync; an absent file fsyncs its parent directory so deletes are as durable as writes). An unclean open under log authority now folds only (checkpoint, head]; the chain bootstraps at an empty brain's adoption (three-phase hooks around adoptLogAuthority) or at a brain's first whole-log fold — existing brains converge at their first crash with zero regression. Rollback restores sync immediately; abort paths feed the barrier; a failed barrier retains the old bound (bigger fold later, never a lost write). Five structural pins including boundedness itself. Also: the production-shaped write-flow gate leg (mixed traffic racing flushes, crash mid-traffic, every ack survives — from a consumer-reported gate miss), and two release-ceremony cures (tag-first push so the publish never queues behind the release commit's CI run; raw-curl npmjs shasum probe with propagation grace instead of a one-shot false divergence). --- scripts/release.sh | 39 ++- src/brainy.ts | 20 ++ src/db/generationStore.ts | 257 +++++++++++++++++- src/db/types.ts | 12 + src/storage/adapters/fileSystemStorage.ts | 7 + src/storage/baseStorage.ts | 23 ++ .../integration/fold-checkpoint-bound.test.ts | 200 ++++++++++++++ .../write-flow-production-shape.test.ts | 149 ++++++++++ 8 files changed, 695 insertions(+), 12 deletions(-) create mode 100644 tests/integration/fold-checkpoint-bound.test.ts create mode 100644 tests/integration/write-flow-production-shape.test.ts diff --git a/scripts/release.sh b/scripts/release.sh index ce2d0882..03d60ac2 100755 --- a/scripts/release.sh +++ b/scripts/release.sh @@ -177,8 +177,15 @@ echo -e "${GREEN}✅ Tag created${NC}\n" # Step 9: Push to origin — The Source is the one home (ruled 2026-07-23; the # old public GitHub repo is archived history, no longer part of any release). -echo -e "${BLUE}8️⃣ Pushing to origin...${NC}" -git push --follow-tags origin "$CURRENT_BRANCH" +# TAG FIRST, branch second — deliberately two pushes: the runner is +# sequential, and a combined push can queue the release commit's ci.yml run +# AHEAD of the tag's publish-source run (observed on 10.0.0: the publish sat +# ~37 minutes behind a redundant CI run of the very commit the local gates +# had just proven). Pushing the tag alone queues the publish immediately; +# the branch push (and its ci.yml run) follows behind it, harmlessly. +echo -e "${BLUE}8️⃣ Pushing to origin (tag first — the publish must never queue behind CI)...${NC}" +git push origin "v${NEW_VERSION}" +git push origin "$CURRENT_BRANCH" echo -e "${GREEN}✅ Pushed to origin${NC}\n" # Step 10: The home publish (The Source, source.soulcraft.com) is CI's job @@ -227,14 +234,32 @@ npm publish "$SOURCE_TARBALL" --tag "$NPM_TAG" "--@soulcraft:registry=https://re rm -rf "$STOREFRONT_TMP" # Brainy is the only PUBLIC @soulcraft package — verify visibility after every publish. npm access get status @soulcraft/brainy "--@soulcraft:registry=https://registry.npmjs.org/" || true -# Verify the pair is byte-identical by registry-reported shasum — divergence here -# means the storefront leg must be treated as failed, loudly. +# Verify the pair is byte-identical by registry-reported shasum — divergence +# here means the storefront leg must be treated as failed, loudly. RETRIED +# with raw curl: npmjs metadata propagates with a lag measured in minutes, +# and a one-shot npm-view probe fired a false DIVERGENCE on 10.0.0 while a +# raw curl of the registry document already confirmed byte-identity. The +# probe now reads the registry JSON directly (no npm cache in the path) and +# gives propagation up to 5 minutes before calling the pair divergent. +NPMJS_VERIFY_ATTEMPTS=20 +NPMJS_VERIFY_INTERVAL_S=15 # 20 × 15s = 5 minutes of propagation grace SOURCE_SHA=$(npm view "@soulcraft/brainy@${NEW_VERSION}" dist.shasum "--@soulcraft:registry=${SOURCE_NPM_REG}" 2>/dev/null || echo "source-unavailable") -NPMJS_SHA=$(npm view "@soulcraft/brainy@${NEW_VERSION}" dist.shasum "--@soulcraft:registry=https://registry.npmjs.org/" 2>/dev/null || echo "npmjs-unavailable") -if [ "$SOURCE_SHA" = "$NPMJS_SHA" ]; then +PAIR_IDENTICAL=false +for ((attempt = 1; attempt <= NPMJS_VERIFY_ATTEMPTS; attempt++)); do + NPMJS_SHA=$(curl -fsSL "https://registry.npmjs.org/@soulcraft%2Fbrainy" 2>/dev/null \ + | node -e "let d='';process.stdin.on('data',c=>d+=c).on('end',()=>{try{const v=JSON.parse(d).versions[process.argv[1]];console.log(v?v.dist.shasum:'')}catch{console.log('')}})" "${NEW_VERSION}" \ + || echo "") + if [ -n "$NPMJS_SHA" ] && [ "$SOURCE_SHA" = "$NPMJS_SHA" ]; then + PAIR_IDENTICAL=true + break + fi + echo -e "${YELLOW} … npmjs metadata not settled (attempt ${attempt}/${NPMJS_VERIFY_ATTEMPTS}: '${NPMJS_SHA:-absent}' vs '${SOURCE_SHA}'); retrying in ${NPMJS_VERIFY_INTERVAL_S}s${NC}" + sleep "$NPMJS_VERIFY_INTERVAL_S" +done +if [ "$PAIR_IDENTICAL" = true ]; then echo -e "${GREEN}✅ Published to npmjs — byte-identical pair (shasum ${NPMJS_SHA})${NC}\n" else - echo -e "${RED}❌ REGISTRY DIVERGENCE: The Source shasum ${SOURCE_SHA} != npmjs shasum ${NPMJS_SHA} — investigate before announcing${NC}\n" + echo -e "${RED}❌ REGISTRY DIVERGENCE: The Source shasum ${SOURCE_SHA} != npmjs shasum ${NPMJS_SHA} after ${NPMJS_VERIFY_ATTEMPTS} attempts — investigate before announcing${NC}\n" exit 1 fi diff --git a/src/brainy.ts b/src/brainy.ts index 785d10e5..dc97b82f 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -8239,6 +8239,23 @@ export class Brainy implements BrainyInterface { async adoptLogAuthority(): Promise { await this.ensureInitialized() this.assertWritable('adoptLogAuthority') + // Fold-checkpoint chain, phase 1: a FRESH brain (no committed + // generations) arms the chain now so the backfill's re-commits below + // feed the canonical-sync accumulator — its first stamp is then total. + // A non-fresh flip skips (the store refuses the arm); its chain starts + // at the first recovery fold instead. Disarmed on any failure below. + this.generationStore.beginFoldCheckpointBootstrap() + try { + return await this.adoptLogAuthorityInner() + } catch (err) { + this.generationStore.abandonFoldCheckpointBootstrap() + throw err + } + } + + /** The adoption body — see {@link Brainy.adoptLogAuthority} (which owns the + * fold-checkpoint bootstrap arm/disarm around it). */ + private async adoptLogAuthorityInner(): Promise { let report = await this.verifyLogAuthority() // BASELINE BACKFILL: curable divergences are rows whose CANONICAL truth @@ -8334,6 +8351,9 @@ export class Brainy implements BrainyInterface { report ) this.generationStore.setLogDurability('at-ack') + // Fold-checkpoint chain, phase 2: the flip is recorded — open the stamp + // gate so the next flush/close barrier writes the first checkpoint. + this.generationStore.completeFoldCheckpointBootstrap() return report } diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index 8f3bf625..837ea90a 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -78,10 +78,21 @@ export const MANIFEST_PATH = '_system/manifest.json' /** * The clean-shutdown marker (log-authority recovery gate): written+fsynced at * a clean close carrying the committed generation; CONSUMED at every open. - * Absent or generation-mismatched at open = unclean shutdown = the whole-log - * replay fold. Its absence is always safe (costs one replay, loses nothing). + * Absent or generation-mismatched at open = unclean shutdown = the replay + * fold, bounded below by the fold checkpoint when one is stored (whole-log + * without one). Its absence is always safe (costs one fold, loses nothing). */ export const CLEAN_SHUTDOWN_PATH = '_system/clean-shutdown.json' +/** + * The fold checkpoint (log-authority recovery BOUND): `{ generation: G }` + * asserts that every entity whose latest fact is ≤ G has durable canonical + * bytes — so an unclean open folds only `(G, head]` instead of the whole log. + * Stamped strictly AFTER a canonical-sync barrier over every live entity + * touched since the last stamp (stamp-after-data); absent or torn = fold from + * 0 (always safe, just bigger). The chain of stamps starts only at a provable + * point: an empty brain, or the end of a whole-log fold. + */ +export const FOLD_CHECKPOINT_PATH = '_system/fold-checkpoint.json' /** Storage-root-relative prefix of the per-generation record directories. */ export const GENERATIONS_PREFIX = '_generations' @@ -219,6 +230,37 @@ export class GenerationStore { /** Compaction horizon — record-sets ≤ this are reclaimed. */ private horizonGen = 0 + /** + * Fold-checkpoint accumulator: every entity whose CANONICAL live bytes were + * (re)written since the last stamped checkpoint. Drained by + * {@link advanceFoldCheckpointUnlocked} — synced first, stamped after; on a + * failed barrier the drained ids merge back so the checkpoint can never + * advance past unsynced bytes. Fed only while the chain is valid (see + * {@link foldCheckpointChainValid}) so tree-authority brains never grow it. + */ + private checkpointDirtyNouns = new Set() + /** @see checkpointDirtyNouns — the verb half of the accumulator. */ + private checkpointDirtyVerbs = new Set() + /** + * Whether the checkpoint chain is PROVABLY sound for this brain: true when + * a stored checkpoint exists (induction), the brain opened empty (vacuous), + * or a whole-log fold just re-applied every fact (base case). While false, + * checkpoints are never stamped and the fold bound stays 0 — the honest + * 10.0 contract, upgraded at the brain's first recovery fold. + */ + private foldCheckpointChainValid = false + /** Last stamped fold-checkpoint generation (0 = none / fold from origin). */ + private foldCheckpoint = 0 + /** + * Whether this brain's stored authority is the log — set from the stored + * artifact at open, or by {@link completeFoldCheckpointBootstrap} when an + * in-session adoption flips it. Checkpoints are only ever STAMPED under log + * authority (the artifact bounds the log fold, which only log-authority + * recovery runs); the dirty accumulator may fill slightly earlier, during + * an adoption in flight (see {@link beginFoldCheckpointBootstrap}). + */ + private authorityIsLog = false + /** * Committed generations whose record dirs exist, stored as a SORTED, DISJOINT, * ascending list of INCLUSIVE `[start, end]` intervals (a run-length set). @@ -552,6 +594,7 @@ export class GenerationStore { // drift machinery at open — same as group-commit recovery. const authority = await readLogAuthority(this.storage) if (authority.authority === 'log') { + this.authorityIsLog = true // TWO REPLAY TIERS, gated by the clean-shutdown marker: // // (1) ABOVE-MANIFEST (always): an intact fact above the manifest is @@ -574,8 +617,22 @@ export class GenerationStore { const cleanShutdown = await this.readCleanShutdownMarker() const orphans = await this.factLog.peekFactsAbove(this.committed) const uncleanOpen = cleanShutdown === null || cleanShutdown !== this.committed + // FOLD-CHECKPOINT BOUND: a stored checkpoint G proves every entity + // whose latest fact is ≤ G has durable canonical bytes (each stamp + // followed a canonical-sync barrier), so the unclean fold only needs + // (G, head] — entities untouched since G are already safe, entities + // touched after G get their latest after-image re-applied. Absent or + // invalid checkpoint = fold from 0 (the 10.0 whole-log contract). + const checkpoint = await this.readFoldCheckpoint() + const foldBound = checkpoint ?? 0 + // Chain validity: induction (a stored stamp), vacuous truth (an empty + // brain has no bytes to assert), or — set below — the base case (a + // whole-log fold re-applies and re-syncs every entity in the log). + this.foldCheckpointChainValid = checkpoint !== null || this.committed === 0 + this.foldCheckpoint = foldBound + if (uncleanOpen) this.foldCheckpointChainValid = true const factsToReplay = uncleanOpen - ? await this.factLog.peekFactsAbove(0) + ? await this.factLog.peekFactsAbove(foldBound) : orphans if (factsToReplay.length > 0) { let replayed = 0 @@ -587,6 +644,7 @@ export class GenerationStore { : { metadata: op.record.metadata, vector: op.record.vector } if (op.kind === 'verb') await this.storage.writeVerbRaw(op.id, image) else await this.storage.writeNounRaw(op.id, image) + this.noteCheckpointDirty(op.kind, op.id) } replayed++ if (fact.generation > this.committed) { @@ -612,10 +670,21 @@ export class GenerationStore { await this.storage.syncRawObjects([MANIFEST_PATH]) prodLog.warn( `[GenerationStore] log-authority recovery replayed ${replayed} fact(s) into ` + - `canonical (${uncleanOpen ? 'WHOLE-LOG fold — unclean shutdown' : 'above-manifest'}; ` + - `committed at ${this.committed}) — an acked write is never lost` + `canonical (${ + uncleanOpen + ? foldBound > 0 + ? `BOUNDED fold above checkpoint ${foldBound} — unclean shutdown` + : 'WHOLE-LOG fold — unclean shutdown' + : 'above-manifest' + }; committed at ${this.committed}) — an acked write is never lost` ) } + // A recovery fold re-applied (and the barrier below re-syncs) every + // entity in (bound, head] — stamp the checkpoint at the new committed + // watermark so the NEXT crash folds only its own tail. This is also + // the chain's base case: the first whole-log fold of a pre-checkpoint + // brain covers every entity in the log, so its stamp is total. + if (uncleanOpen) await this.advanceFoldCheckpointUnlocked() // The marker is consumed: any session that can write invalidates it // at first commit (see the commit paths); a clean close re-writes it. await this.clearCleanShutdownMarker() @@ -672,6 +741,12 @@ export class GenerationStore { await this.flushPendingSingleOps() this.storage.setGenerationBumpHook(undefined) await this.persistCounterNow() + // Fold-checkpoint barrier BEFORE the clean-shutdown marker: entities that + // reached the accumulator outside the pending tier (transact commits, + // aborted-write restores) get their canonical bytes synced and the stamp + // advanced, so the marker below never vouches for bytes the checkpoint + // chain hasn't proven durable. + await this.advanceFoldCheckpoint() // Clean-shutdown marker (log-authority recovery gate): everything above // is durable; stamp the committed generation so the next open can adopt // instead of folding the log. Written LAST — a crash before this line is @@ -705,6 +780,131 @@ export class GenerationStore { } } + /** + * Read the fold checkpoint's generation, or `null` when absent, torn, or + * implausible (> committed) — every invalid shape degrades to the safe + * whole-log fold, never to a bound that could skip an acked write. + */ + private async readFoldCheckpoint(): Promise { + try { + const raw = (await this.storage.readRawObject(FOLD_CHECKPOINT_PATH)) as { + generation?: number + } | null + const gen = raw?.generation + if (!Number.isSafeInteger(gen) || (gen as number) < 0) return null + if ((gen as number) > this.committed) { + prodLog.warn( + `[GenerationStore] fold checkpoint ${gen} is ahead of the manifest ` + + `(${this.committed}) — ignoring it; recovery folds the whole log` + ) + return null + } + return gen as number + } catch { + return null + } + } + + /** + * Record that an entity's canonical live bytes were (re)written and are not + * yet covered by a checkpoint stamp. Gated on chain validity so brains + * without a sound chain (tree authority, or log authority before its first + * recovery fold) never accumulate — they keep the fold-from-0 contract. + */ + private noteCheckpointDirty(kind: 'noun' | 'verb', id: string): void { + if (!this.foldCheckpointChainValid) return + if (kind === 'verb') this.checkpointDirtyVerbs.add(id) + else this.checkpointDirtyNouns.add(id) + } + + /** + * The canonical-sync barrier + checkpoint stamp (must run under the commit + * mutex or in single-threaded open). Drains the dirty accumulator, makes + * those entities' canonical bytes durable via the adapter barrier, and only + * THEN stamps `_system/fold-checkpoint.json` at the committed watermark — + * stamp-after-data, always. On any failure the drained ids merge back and + * the stored checkpoint stays where it was: the bound can lag (a bigger + * fold later) but can never overstate durability (a lost write, outlawed). + */ + private async advanceFoldCheckpointUnlocked(): Promise { + if (!this.foldCheckpointChainValid || !this.authorityIsLog || !this.factLog) return + const nouns = [...this.checkpointDirtyNouns] + const verbs = [...this.checkpointDirtyVerbs] + const target = this.committed + if (nouns.length === 0 && verbs.length === 0 && target === this.foldCheckpoint) return + this.checkpointDirtyNouns = new Set() + this.checkpointDirtyVerbs = new Set() + try { + if (nouns.length > 0 || verbs.length > 0) { + await this.storage.syncEntityCanonical?.(nouns, verbs) + } + await this.storage.writeRawObject(FOLD_CHECKPOINT_PATH, { generation: target }) + await this.storage.syncRawObjects([FOLD_CHECKPOINT_PATH]) + this.foldCheckpoint = target + } catch (err) { + for (const id of nouns) this.checkpointDirtyNouns.add(id) + for (const id of verbs) this.checkpointDirtyVerbs.add(id) + prodLog.warn( + `[GenerationStore] fold-checkpoint barrier failed at generation ${target} ` + + `(${(err as Error).message}) — checkpoint stays at ${this.foldCheckpoint}; ` + + `recovery would fold from there (bigger, never lossy). Will retry next flush.` + ) + } + } + + /** + * @description Public, mutex-serialized fold-checkpoint advance — called by + * `close()` after the final flush so entities touched by paths that do not + * ride the pending tier (e.g. `transact()`) are covered before the + * clean-shutdown marker is written. + */ + async advanceFoldCheckpoint(): Promise { + return this.withMutex(() => this.advanceFoldCheckpointUnlocked()) + } + + /** + * @description Adoption-time chain bootstrap, phase 1 — called by + * `adoptLogAuthority()` BEFORE its oracle/backfill passes. Only a FRESH + * brain (committed === 0) may bootstrap here: with no committed + * generations the chain's assertion is vacuously true, and arming it now + * means the baseline backfill's own re-commits feed the dirty accumulator, + * so the first stamp after the flip covers them. A non-fresh flip skips + * this (returns false) — its chain starts at the brain's first recovery + * fold instead, because only a whole-log fold can prove coverage of + * entities written before the log existed. + */ + beginFoldCheckpointBootstrap(): boolean { + if (this.committed !== 0 || this.foldCheckpointChainValid) { + return this.foldCheckpointChainValid + } + this.foldCheckpointChainValid = true + this.foldCheckpoint = 0 + return true + } + + /** + * @description Adoption-time chain bootstrap, phase 2 — called after + * `flipToLogAuthority` records the flip. Opens the stamp gate; the next + * flush/close barrier writes the first checkpoint. + */ + completeFoldCheckpointBootstrap(): void { + this.authorityIsLog = true + } + + /** + * @description Adoption-time chain bootstrap, abort — called when an + * adoption attempt throws or refuses after phase 1. Disarms the chain and + * drops the accumulator so a tree-authority brain never accumulates or + * stamps. (If the chain was valid BEFORE the attempt — a stored checkpoint + * exists — it stays valid; only a phase-1 arm is undone.) + */ + abandonFoldCheckpointBootstrap(): void { + if (this.authorityIsLog) return + this.foldCheckpointChainValid = false + this.checkpointDirtyNouns = new Set() + this.checkpointDirtyVerbs = new Set() + } + /** * @description TEST-ONLY: install (or clear, with `undefined`) a fault * injector that is invoked at each {@link CommitFaultPhase} of the commit @@ -1238,6 +1438,13 @@ export class GenerationStore { this.historyBytesTotal += delta.bytes ?? 0 } this.extendChains(gen, nouns, verbs) + // Fold-checkpoint accounting: the write barrier above already synced + // this batch's canonical footprint on adapters that have one, but the + // accumulator entry is the belt — an adapter without a write barrier + // still gets these ids covered by the next checkpoint barrier, and a + // redundant fsync of already-durable bytes is cheap and idempotent. + for (const id of nouns) this.noteCheckpointDirty('noun', id) + for (const id of verbs) this.noteCheckpointDirty('verb', id) const logEntry: TxLogEntry = { generation: gen, timestamp, ...(args.meta && { meta: args.meta }) } await this.storage.appendTxLogLine(JSON.stringify(logEntry)) @@ -1249,6 +1456,13 @@ export class GenerationStore { if (crashSimulated) { throw err } + // Fold-checkpoint accounting: an abort's rollback restores are raw + // canonical writes that never reach the transaction write barrier + // (flushWriteBarrier only runs on the commit path) — feed them so the + // next checkpoint barrier syncs the restored bytes before any stamp + // vouches for them. + for (const id of nouns) this.noteCheckpointDirty('noun', id) + for (const id of verbs) this.noteCheckpointDirty('verb', id) // The trapdoor for a batch: if rollback FAILED to fully apply, canonical // storage may be inconsistent. A batch is never adopted forward (its // other ops were rolled back — partial commit would break atomicity), so @@ -1476,6 +1690,13 @@ export class GenerationStore { await args.execute() } catch (err) { this.inTransact = false + // Fold-checkpoint accounting: execute() ran, so canonical bytes for + // the touched ids changed — whether they now hold the new images, a + // restored rollback, or (the trapdoor) something indeterminate, the + // next checkpoint stamp must not assert their durability without a + // barrier over whatever is actually there. + for (const id of nouns) this.noteCheckpointDirty('noun', id) + for (const id of verbs) this.noteCheckpointDirty('verb', id) // A failed rollback (TransactionRollbackError) may have left canonical // storage inconsistent — the trapdoor. Reconcile against the // before-images to decide the honest response (David's ruling: @@ -1530,6 +1751,10 @@ export class GenerationStore { throw err } this.inTransact = false + // Fold-checkpoint accounting: the live canonical write is applied — it + // must ride the next canonical-sync barrier before any stamp covers it. + for (const id of nouns) this.noteCheckpointDirty('noun', id) + for (const id of verbs) this.noteCheckpointDirty('verb', id) // Test-only crash simulation (direct call — a throw propagates with no // cleanup, exactly like a process death; recovery-on-open restores the // contract). A crash here must cost only the never-returned ack: the @@ -1801,6 +2026,16 @@ export class GenerationStore { for (const entry of logEntries) { await this.storage.appendTxLogLine(JSON.stringify(entry)) } + + // Fold-checkpoint barrier: the window's LIVE canonical bytes (the acked + // writes themselves — the staging sync above covered only their history + // copies) become durable here, and only then does the checkpoint stamp + // advance to the new committed watermark. This is what keeps crash + // recovery's log fold bounded to (checkpoint, head] instead of the + // whole log. A failure inside is absorbed by the barrier (it warns, + // retains the accumulator, and leaves the old bound standing) — history + // durability above already succeeded, so the flush itself is good. + await this.advanceFoldCheckpointUnlocked() }) } @@ -2961,6 +3196,8 @@ export class GenerationStore { private async rollBackUncommittedGeneration(gen: number): Promise { const dir = `${GENERATIONS_PREFIX}/${gen}` const prevPaths = await this.storage.listRawObjects(`${dir}/prev`) + const restoredNouns: string[] = [] + const restoredVerbs: string[] = [] for (const recordPath of prevPaths) { const id = recordIdFromPath(recordPath) if (id === null) continue @@ -2969,10 +3206,20 @@ export class GenerationStore { const image = { metadata: record.metadata, vector: record.vector } if (record.kind === 'verb') { await this.storage.writeVerbRaw(id, image) + restoredVerbs.push(id) } else { await this.storage.writeNounRaw(id, image) + restoredNouns.push(id) } } + // Make the restores durable IMMEDIATELY (this runs at open, before the + // fold-checkpoint chain state is even read): a restored before-image + // replaces bytes a stored checkpoint may already vouch for, so it must + // reach disk with the same certainty — otherwise a power cut could let + // the rolled-back write's bytes resurrect past a bounded fold. + if (restoredNouns.length > 0 || restoredVerbs.length > 0) { + await this.storage.syncEntityCanonical?.(restoredNouns, restoredVerbs) + } await this.storage.removeRawPrefix(dir) prodLog.warn( `[GenerationStore] rolled back uncommitted generation ${gen} ` + diff --git a/src/db/types.ts b/src/db/types.ts index a7f86a89..2c7eab8f 100644 --- a/src/db/types.ts +++ b/src/db/types.ts @@ -462,6 +462,18 @@ export interface GenerationStorage { /** @see beginWriteBarrier — fsync every canonical write since begin. */ flushWriteBarrier?(): Promise + /** + * OPTIONAL fold-checkpoint durability barrier: make the listed entities' + * CANONICAL live objects durable — fsync each present metadata/vector file + * AND the parent directory entry of each absent one (so a delete is as + * durable as a write). The generation store may only advance the fold + * checkpoint (`_system/fold-checkpoint.json`) after this resolves; the + * checkpoint bounds crash recovery's log fold to `(checkpoint, head]`. + * Adapters whose writes are durable per-call may leave this undefined — + * the store then treats canonical durability as immediate. + */ + syncEntityCanonical?(nouns: string[], verbs: string[]): Promise + /** Read an entity's raw stored metadata+vector objects. */ readNounRaw(id: string): Promise<{ metadata: any | null; vector: any | null }> /** Restore an entity's raw stored objects (`null` part ⇒ delete that file). */ diff --git a/src/storage/adapters/fileSystemStorage.ts b/src/storage/adapters/fileSystemStorage.ts index fea817d5..81c6e545 100644 --- a/src/storage/adapters/fileSystemStorage.ts +++ b/src/storage/adapters/fileSystemStorage.ts @@ -799,6 +799,7 @@ export class FileSystemStorage extends BaseStorage { for (const objectPath of paths) { const fullPath = path.join(this.rootDir, objectPath) + let synced = false for (const candidate of [`${fullPath}.gz`, fullPath]) { let handle: any try { @@ -813,8 +814,14 @@ export class FileSystemStorage extends BaseStorage { await handle.close() } parentDirs.add(path.dirname(fullPath)) + synced = true break } + // An absent path is a state too: fsync the parent directory so a + // completed unlink is durable (a delete must survive power loss as + // surely as a write — otherwise a bounded log fold could let a + // tombstoned record resurrect from a lost directory update). + if (!synced) parentDirs.add(path.dirname(fullPath)) } for (const dir of parentDirs) { diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index 23003ede..c3b3b3bf 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -1390,6 +1390,29 @@ export abstract class BaseStorage extends BaseStorageAdapter { void paths } + /** + * Fold-checkpoint durability barrier: make the listed entities' canonical + * live objects durable. Maps each id to its canonical metadata + vector + * paths and delegates to {@link BaseStorage.syncRawObjects}, whose + * filesystem override fsyncs present files (and their rename directory + * entries) and the parent directory of absent ones — so deletes are as + * durable as writes. The generation store advances the fold checkpoint + * only after this resolves (stamp-after-data). + * + * @param nouns - Entity ids whose canonical objects must be durable. + * @param verbs - Relationship ids whose canonical objects must be durable. + */ + public async syncEntityCanonical(nouns: string[], verbs: string[]): Promise { + const paths: string[] = [] + for (const id of nouns) { + paths.push(getNounMetadataPath(id), getNounVectorPath(id)) + } + for (const id of verbs) { + paths.push(getVerbMetadataPath(id), getVerbVectorPath(id)) + } + if (paths.length > 0) await this.syncRawObjects(paths) + } + /** * Read an entity's raw stored objects — the exact bytes at its canonical * metadata + vector paths (write-cache coherent). Used by the generation diff --git a/tests/integration/fold-checkpoint-bound.test.ts b/tests/integration/fold-checkpoint-bound.test.ts new file mode 100644 index 00000000..ce074dcf --- /dev/null +++ b/tests/integration/fold-checkpoint-bound.test.ts @@ -0,0 +1,200 @@ +/** + * @module tests/integration/fold-checkpoint-bound + * @description The fold-checkpoint bound (crash recovery's log fold, bounded): + * `_system/fold-checkpoint.json` at generation G asserts every entity whose + * latest fact is ≤ G has DURABLE canonical bytes — each stamp strictly follows + * a canonical-sync barrier over every live entity touched since the last one + * (stamp-after-data). An unclean open then folds only `(G, head]` instead of + * the whole log. These pins prove the four load-bearing properties: + * + * 1. The stamp exists and tracks the committed watermark (flush + close). + * 2. The fold is genuinely BOUNDED — facts ≤ G are skipped — while facts in + * `(G, head]` are re-applied even BELOW the manifest. + * 3. A failed barrier NEVER advances the stamp (the bound can lag, growing + * a later fold — it can never overstate durability, losing a write). + * 4. A pre-checkpoint brain (the 10.0 shape) bootstraps its chain at its + * first whole-log fold; a tree-authority brain never stamps at all. + */ +import { describe, it, expect, afterEach, vi } from 'vitest' +import * as fs from 'node:fs' +import * as zlib from 'node:zlib' +import { join } from 'node:path' +import { Brainy } from '../../src/brainy.js' +import { NounType } from '../../src/types/graphTypes.js' +import { + abandonAsCrashed, + dropCanonicalNoun, + makeTempDir, + openBrain, + storeOf +} from '../helpers/durabilityKillMatrix.js' + +const CHECKPOINT = join('_system', 'fold-checkpoint.json') + +/** Read the fold-checkpoint artifact's generation from disk, or null. */ +function readCheckpoint(dir: string): number | null { + for (const candidate of [join(dir, `${CHECKPOINT}.gz`), join(dir, CHECKPOINT)]) { + if (!fs.existsSync(candidate)) continue + const raw = fs.readFileSync(candidate) + const text = candidate.endsWith('.gz') ? zlib.gunzipSync(raw).toString('utf8') : raw.toString('utf8') + const parsed = JSON.parse(text) as { generation?: number } + return Number.isSafeInteger(parsed.generation) ? (parsed.generation as number) : null + } + return null +} + +function removeArtifact(dir: string, rel: string): void { + for (const candidate of [join(dir, `${rel}.gz`), join(dir, rel)]) { + fs.rmSync(candidate, { force: true }) + } +} + +function committedOf(brain: Brainy): number { + return (storeOf(brain) as unknown as { committed: number }).committed +} + +describe('fold-checkpoint bound — crash recovery folds (checkpoint, head], never less durability than stamped', () => { + const dirs: string[] = [] + const liveBrains: Brainy[] = [] + afterEach(async () => { + vi.restoreAllMocks() + for (const b of liveBrains.splice(0)) await b.close().catch(() => {}) + for (const d of dirs.splice(0)) fs.rmSync(d, { recursive: true, force: true }) + }) + function trackDir(): string { + const dir = makeTempDir() + dirs.push(dir) + return dir + } + + it('a fresh adopt brain stamps at flush and again at close — the stamp tracks the committed watermark', async () => { + const dir = trackDir() + const brain = await openBrain(dir, { logAuthority: 'adopt' }) + liveBrains.push(brain) + expect(brain.logAuthority().authority).toBe('log') + + await brain.add({ data: 'first', type: NounType.Document, metadata: { n: 1 } }) + await brain.add({ data: 'second', type: NounType.Document, metadata: { n: 2 } }) + await brain.flush() + const afterFlush = readCheckpoint(dir) + expect(afterFlush).toBe(committedOf(brain)) + expect(afterFlush!).toBeGreaterThan(0) + + await brain.add({ data: 'third', type: NounType.Document, metadata: { n: 3 } }) + const closingCommit = liveBrains.pop()! + await closingCommit.close() + // Close flushes, so the stamp advanced with it — and the clean-shutdown + // marker it writes afterward never vouches for bytes the stamp has not. + expect(readCheckpoint(dir)).toBeGreaterThanOrEqual(afterFlush!) + }, 120000) + + it('BOUNDED fold: facts ≤ checkpoint are skipped, facts in (checkpoint, head] are re-applied even below the manifest; a failed barrier retains the old bound', async () => { + const dir = trackDir() + const brain = await openBrain(dir, { logAuthority: 'adopt' }) + liveBrains.push(brain) + + // Window 1 — flushed and stamped: the checkpoint's covered past. + const idA = await brain.add({ data: 'covered by the stamp', type: NounType.Document, metadata: { w: 1 } }) + await brain.flush() + const checkpoint1 = readCheckpoint(dir) + expect(checkpoint1).toBe(committedOf(brain)) + + // Window 2 — committed BELOW a new manifest but with the checkpoint stamp + // FAILING: the barrier throws once, so the manifest advances while the + // stamp stays at checkpoint1 (pin 3: a failed barrier never advances it). + const storage = (brain as unknown as { + storage: { syncEntityCanonical(n: string[], v: string[]): Promise } + }).storage + const realBarrier = storage.syncEntityCanonical.bind(storage) + let failedOnce = false + vi.spyOn(storage, 'syncEntityCanonical').mockImplementation(async (n: string[], v: string[]) => { + if (!failedOnce) { + failedOnce = true + throw new Error('injected barrier failure (device hiccup)') + } + return realBarrier(n, v) + }) + const idB = await brain.add({ data: 'below manifest, above checkpoint', type: NounType.Document, metadata: { w: 2 } }) + await brain.flush() + expect(failedOnce).toBe(true) + expect(readCheckpoint(dir)).toBe(checkpoint1) // stamp did NOT advance + expect(committedOf(brain)).toBeGreaterThan(checkpoint1!) // manifest DID + + // Crash. Vaporize BOTH canonical records: idB's fact lives in + // (checkpoint, manifest] — the bounded fold MUST restore it; idA's fact + // is ≤ checkpoint — the fold must SKIP it (its loss here is synthetic: + // the stamp's barrier fsynced it, a power cut cannot take it, and the + // skip is exactly what makes the fold bounded instead of whole-log). + await abandonAsCrashed(liveBrains.pop()!) + dropCanonicalNoun(dir, idA) + dropCanonicalNoun(dir, idB) + + const reopened = await openBrain(dir, { logAuthority: 'adopt' }) + liveBrains.push(reopened) + const restoredB = await reopened.get(idB) + expect(restoredB, 'a fact above the checkpoint is re-applied even below the manifest').not.toBeNull() + const skippedA = await reopened.get(idA) + expect(skippedA, 'a fact at-or-below the checkpoint is outside the fold — the bound is real').toBeNull() + // And recovery re-stamped at its new committed watermark. + expect(readCheckpoint(dir)).toBe(committedOf(reopened)) + }, 120000) + + it('a pre-checkpoint brain (the 10.0 shape) folds the WHOLE log once, then its chain is established', async () => { + const dir = trackDir() + const brain = await openBrain(dir, { logAuthority: 'adopt' }) + liveBrains.push(brain) + const idA = await brain.add({ data: 'ten-point-oh resident', type: NounType.Document, metadata: { era: '10.0' } }) + await brain.flush() + await liveBrains.pop()!.close() + + // Rewind the brain to the 10.0 shape: no checkpoint artifact, and an + // unclean shutdown (marker gone) — exactly what an existing fleet brain + // looks like at its first crash under 10.1. + removeArtifact(dir, CHECKPOINT) + removeArtifact(dir, join('_system', 'clean-shutdown.json')) + dropCanonicalNoun(dir, idA) + + const reopened = await openBrain(dir, { logAuthority: 'adopt' }) + liveBrains.push(reopened) + expect(await reopened.get(idA), 'no checkpoint ⇒ whole-log fold ⇒ every acked write restored').not.toBeNull() + const stamped = readCheckpoint(dir) + expect(stamped, 'the first whole-log fold is the chain’s base case — it stamps').toBe(committedOf(reopened)) + }, 120000) + + it('a tree-authority brain never stamps a checkpoint', async () => { + const dir = trackDir() + const brain = await openBrain(dir, { logAuthority: 'defer' }) + liveBrains.push(brain) + expect(brain.logAuthority().authority).not.toBe('log') + await brain.add({ data: 'tree resident', type: NounType.Document, metadata: { n: 1 } }) + await brain.flush() + await liveBrains.pop()!.close() + expect(readCheckpoint(dir)).toBeNull() + }, 120000) + + it('a delete rides the barrier: the tombstoned id is in the synced set and the stamp advances past it', async () => { + const dir = trackDir() + const brain = await openBrain(dir, { logAuthority: 'adopt' }) + liveBrains.push(brain) + const id = await brain.add({ data: 'short-lived', type: NounType.Document, metadata: { n: 1 } }) + await brain.flush() + + const storage = (brain as unknown as { + storage: { syncEntityCanonical(n: string[], v: string[]): Promise } + }).storage + const seen: string[][] = [] + const realBarrier = storage.syncEntityCanonical.bind(storage) + vi.spyOn(storage, 'syncEntityCanonical').mockImplementation(async (n: string[], v: string[]) => { + seen.push([...n]) + return realBarrier(n, v) + }) + + await brain.remove(id) + await brain.flush() + expect( + seen.some((nouns) => nouns.includes(id)), + 'the deleted id must reach the canonical barrier (absence is durable state too)' + ).toBe(true) + expect(readCheckpoint(dir)).toBe(committedOf(brain)) + }, 120000) +}) diff --git a/tests/integration/write-flow-production-shape.test.ts b/tests/integration/write-flow-production-shape.test.ts new file mode 100644 index 00000000..f33cdc88 --- /dev/null +++ b/tests/integration/write-flow-production-shape.test.ts @@ -0,0 +1,149 @@ +/** + * @module tests/integration/write-flow-production-shape + * @description The production-shaped WRITE-FLOW gate leg. A downstream + * deployment's release gate went all-green on snapshots and rehearsal reads + * while two write-path defects (pad-frame constructibility, a counter rewind + * after a successful append) waited in ordinary WRITE flows — deferred + * embedding retries plus background history-flush concurrency wearing the + * stacks. This leg runs that exact shape, permanently: + * + * - concurrent mixed writes (adds, deferred-embed adds, updates, removes) + * - racing explicit flushes (the history tier's group commit, mid-traffic) + * - then the three laws: every ack is readable truth, the fact log is + * STRICTLY ascending end-to-end, and no write is ever refused. + * + * Part two crashes the brain mid-traffic (no close — RAM discarded) and + * requires every acked write back after reopen: the at-ack contract under + * the same production shape, not under a synthetic single write. + */ +import { describe, it, expect, afterEach } from 'vitest' +import * as fs from 'node:fs' +import { Brainy } from '../../src/brainy.js' +import { NounType } from '../../src/types/graphTypes.js' +import { + abandonAsCrashed, + factGenerations, + makeTempDir, + openBrain +} from '../helpers/durabilityKillMatrix.js' + +describe('write-flow production shape — the pair gate leg from a consumer-reported miss', () => { + const dirs: string[] = [] + const liveBrains: Brainy[] = [] + afterEach(async () => { + for (const b of liveBrains.splice(0)) await b.close().catch(() => {}) + for (const d of dirs.splice(0)) fs.rmSync(d, { recursive: true, force: true }) + }) + function trackDir(): string { + const dir = makeTempDir() + dirs.push(dir) + return dir + } + + async function runTrafficWave( + brain: Brainy, + wave: number, + perWave: number + ): Promise<{ kept: string[]; removed: string[] }> { + const kept: string[] = [] + const removed: string[] = [] + const work: Promise[] = [] + for (let i = 0; i < perWave; i++) { + const n = wave * perWave + i + if (i % 4 === 0) { + // Deferred-embed add — the retry-marker flow that wore the defect. + work.push( + brain + .add({ data: `deferred payload ${n}`, type: NounType.Document, metadata: { n, defer: true }, deferEmbedding: true }) + .then((id) => void kept.push(id)) + ) + } else if (i % 4 === 1) { + // Add, then update it in the same wave (two generations, same id). + work.push( + brain.add({ data: `versioned payload ${n}`, type: NounType.Document, metadata: { n, v: 1 } }).then(async (id) => { + kept.push(id) + await brain.update({ id, metadata: { n, v: 2 } }) + }) + ) + } else if (i % 4 === 2) { + // Add, then remove — a durable tombstone is an ack too. + work.push( + brain.add({ data: `ephemeral payload ${n}`, type: NounType.Document, metadata: { n } }).then(async (id) => { + await brain.remove(id) + removed.push(id) + }) + ) + } else { + work.push( + brain.add({ data: `plain payload ${n}`, type: NounType.Document, metadata: { n } }).then((id) => void kept.push(id)) + ) + } + // Race the history tier's group commit against live traffic. + if (i % 5 === 3) work.push(brain.flush()) + } + // NO REFUSALS: every promise must resolve — a single rejection here is + // the refusal-loop costume this leg exists to catch. + await Promise.all(work) + return { kept, removed } + } + + it('three waves of mixed traffic with racing flushes: every ack is truth, the log is strictly ascending, nothing refused', async () => { + const dir = trackDir() + const brain = await openBrain(dir, { logAuthority: 'adopt' }) + liveBrains.push(brain) + expect(brain.logAuthority().authority).toBe('log') + + const kept: string[] = [] + const removed: string[] = [] + for (let wave = 0; wave < 3; wave++) { + const result = await runTrafficWave(brain, wave, 20) + kept.push(...result.kept) + removed.push(...result.removed) + } + await brain.flush() + + for (const id of kept) { + expect(await brain.get(id), `acked write ${id} must be readable truth`).not.toBeNull() + } + for (const id of removed) { + expect(await brain.get(id), `acked remove ${id} must hold`).toBeNull() + } + + const gens = await factGenerations(brain) + expect(gens.length).toBeGreaterThan(0) + for (let i = 1; i < gens.length; i++) { + expect(gens[i], 'fact log strictly ascending end-to-end').toBeGreaterThan(gens[i - 1]) + } + + // Clean reopen: the same truth survives a restart. + await liveBrains.pop()!.close() + const reopened = await openBrain(dir, { logAuthority: 'adopt' }) + liveBrains.push(reopened) + for (const id of kept.slice(0, 10)) { + expect(await reopened.get(id)).not.toBeNull() + } + }, 240000) + + it('crash mid-traffic: every acked write survives the reopen (the at-ack law under the production shape)', async () => { + const dir = trackDir() + const brain = await openBrain(dir, { logAuthority: 'adopt' }) + liveBrains.push(brain) + + const { kept, removed } = await runTrafficWave(brain, 0, 24) + // No close, no flush — the process "dies" holding its RAM. + await abandonAsCrashed(liveBrains.pop()!) + + const reopened = await openBrain(dir, { logAuthority: 'adopt' }) + liveBrains.push(reopened) + for (const id of kept) { + expect(await reopened.get(id), `acked write ${id} must survive the crash`).not.toBeNull() + } + for (const id of removed) { + expect(await reopened.get(id), `acked remove ${id} must survive the crash`).toBeNull() + } + const gens = await factGenerations(reopened) + for (let i = 1; i < gens.length; i++) { + expect(gens[i], 'fact log strictly ascending after recovery').toBeGreaterThan(gens[i - 1]) + } + }, 240000) +})