fix(durability): three block-layer power-loss findings from the first fault-injection box run — all cured, matrix 15/15

An internal cross-engine fault-injection run (frozen-platter power-loss
capture) surfaced three release-gating findings; each cured in its owning
layer, each pinned:

1. WHOLE-LOG REPLAY ON UNCLEAN OPEN (the big one): log-authority replay
   only covered facts ABOVE the manifest — but live canonical entity
   writes are tmp+rename without per-file fsync, and the group-commit
   flush syncs staging + manifest, never the live tree. Power loss could
   therefore vaporize acked canonical bytes BELOW the manifest while the
   log held every fact scan-clean (measured: 299 of 301 acks lost).
   Now: a clean close stamps a clean-shutdown marker (fsynced, written
   last); every open consumes it; an UNCLEAN open under log authority
   folds the ENTIRE log into canonical — whole-entity after-images make
   the re-apply idempotent and byte-safe. Zero cost on the happy path;
   crash recovery pays one narrated fold. Recovery is replay: a crash is
   just bigger lag.

2. TORN WRITER LOCK: power loss legally leaves the lock file present but
   empty; the parse failure read as 'no holder' while the O_EXCL claim
   EEXISTed forever — a PERMANENT lockout no staleness check could clear.
   An unparseable lock is stale by definition (no live holder has one):
   unlink loudly and re-loop; a racer rewriting a valid lock first wins.

3. PAIR GUARD: flush() called metadataIndex.stampWatermark unguarded;
   a replacement metadata provider without the method killed the pair at
   first flush. All three stamp calls are optional-chained — a missing
   stamp is a verdict-side rescan, never a flush crash.

Pins: whole-log fold restores rows vanished below the manifest ·
clean-shutdown marker lifecycle (stamp/consume/re-stamp) · torn-lock
recovery with a fresh write after · stampless-provider flush.
Gates: unit 2055/2055 · integration 824 · kill-matrix 15/15.

src/brainy.ts

@@ -11247,7 +11247,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
     {
       const wmGen = this.storage?.committedGeneration?.() ?? null
       if (wmGen !== null) {
-        this.metadataIndex.stampWatermark(wmGen)
+        // ALL THREE optional-chained: a replacement provider (the native
+        // pair swaps these managers) may not carry the stamp method — a
+        // missing stamp is a verdict-side rescan, never a flush crash.
+        ;(this.metadataIndex as { stampWatermark?: (g: number) => void }).stampWatermark?.(wmGen)
         ;(this.index as { stampWatermark?: (g: number) => void }).stampWatermark?.(wmGen)
         ;(this.graphIndex as { stampWatermark?: (g: number) => void }).stampWatermark?.(wmGen)
       }

src/db/generationStore.ts

@@ -75,6 +75,13 @@ export interface CommitBeforeImages {
 export const GENERATION_COUNTER_PATH = '_system/generation.json'
 /** Storage-root-relative path of the commit manifest. */
 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).
+ */
+export const CLEAN_SHUTDOWN_PATH = '_system/clean-shutdown.json'
 /** Storage-root-relative prefix of the per-generation record directories. */
 export const GENERATIONS_PREFIX = '_generations'
 
@@ -528,9 +535,34 @@ export class GenerationStore {
       // drift machinery at open — same as group-commit recovery.
       const authority = await readLogAuthority(this.storage)
       if (authority.authority === 'log') {
+        // TWO REPLAY TIERS, gated by the clean-shutdown marker:
+        //
+        // (1) ABOVE-MANIFEST (always): an intact fact above the manifest is
+        //     an acked write whose canonical bytes may not have survived —
+        //     replay it in and advance the manifest.
+        // (2) WHOLE-LOG (unclean shutdown only): power loss can ALSO vaporize
+        //     canonical bytes BELOW the manifest — live entity writes are
+        //     tmp+rename without per-file fsync; the group-commit flush syncs
+        //     the staging copies and the manifest, never the live tree. The
+        //     manifest therefore over-states canonical durability across a
+        //     power cut, and facts ≤ manifest can be the ONLY durable copy
+        //     of acked state (measured: 299 of 301 acks lost while the log
+        //     held every fact scan-clean). Under log authority, recovery is
+        //     REPLAY: an unclean open folds the ENTIRE log into canonical —
+        //     whole-entity after-images are idempotent, so re-applying
+        //     already-intact records is byte-safe. A clean close writes the
+        //     marker and skips all of this (zero open cost on the happy
+        //     path); crash recovery pays one narrated log fold — LC1 and
+        //     LC5 are the same code, a crash is just bigger lag.
+        const cleanShutdown = await this.readCleanShutdownMarker()
         const orphans = await this.factLog.peekFactsAbove(this.committed)
-        if (orphans.length > 0) {
-          for (const fact of orphans) {
+        const uncleanOpen = cleanShutdown === null || cleanShutdown !== this.committed
+        const factsToReplay = uncleanOpen
+          ? await this.factLog.peekFactsAbove(0)
+          : orphans
+        if (factsToReplay.length > 0) {
+          let replayed = 0
+          for (const fact of factsToReplay) {
             for (const op of fact.ops) {
               const image =
                 op.record === null
@@ -539,14 +571,17 @@ export class GenerationStore {
               if (op.kind === 'verb') await this.storage.writeVerbRaw(op.id, image)
               else await this.storage.writeNounRaw(op.id, image)
             }
-            this.committed = fact.generation
-            this.appendCommittedGen(fact.generation)
-            this.setDelta(fact.generation, {
-              nouns: new Set(fact.ops.filter((o) => o.kind === 'noun').map((o) => o.id)),
-              verbs: new Set(fact.ops.filter((o) => o.kind === 'verb').map((o) => o.id)),
-              timestamp: fact.timestamp,
-              bytes: 0
-            })
+            replayed++
+            if (fact.generation > this.committed) {
+              this.committed = fact.generation
+              this.appendCommittedGen(fact.generation)
+              this.setDelta(fact.generation, {
+                nouns: new Set(fact.ops.filter((o) => o.kind === 'noun').map((o) => o.id)),
+                verbs: new Set(fact.ops.filter((o) => o.kind === 'verb').map((o) => o.id)),
+                timestamp: fact.timestamp,
+                bytes: 0
+              })
+            }
           }
           if (this.counter < this.committed) this.counter = this.committed
           await this.persistCounterUnlocked()
@@ -559,11 +594,14 @@ export class GenerationStore {
           await this.storage.writeRawObject(MANIFEST_PATH, manifest)
           await this.storage.syncRawObjects([MANIFEST_PATH])
           prodLog.warn(
-            `[GenerationStore] log-authority recovery REPLAYED ${orphans.length} acked ` +
-              `fact(s) beyond the manifest into canonical (now committed at ${this.committed}) — ` +
-              `an acked write is never lost`
+            `[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`
           )
         }
+        // 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()
       }
       await this.factLog.open(this.committed)
     } else {
@@ -617,6 +655,37 @@ export class GenerationStore {
     await this.flushPendingSingleOps()
     this.storage.setGenerationBumpHook(undefined)
     await this.persistCounterNow()
+    // 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
+    // exactly the unclean case the marker's absence reports.
+    try {
+      await this.storage.writeRawObject(CLEAN_SHUTDOWN_PATH, { generation: this.committed })
+      await this.storage.syncRawObjects([CLEAN_SHUTDOWN_PATH])
+    } catch {
+      // A failed marker write only costs the next open a replay fold — safe.
+    }
+  }
+
+  /** Read the clean-shutdown marker's generation, or null (absent/unreadable). */
+  private async readCleanShutdownMarker(): Promise<number | null> {
+    try {
+      const raw = (await this.storage.readRawObject(CLEAN_SHUTDOWN_PATH)) as {
+        generation?: number
+      } | null
+      return raw && Number.isSafeInteger(raw.generation) ? (raw.generation as number) : null
+    } catch {
+      return null
+    }
+  }
+
+  /** Consume the clean-shutdown marker (every open; a clean close re-writes it). */
+  private async clearCleanShutdownMarker(): Promise<void> {
+    try {
+      await this.storage.deleteRawObject(CLEAN_SHUTDOWN_PATH)
+    } catch {
+      // Absent or undeletable: the conservative outcome is a future replay.
+    }
   }
 
   /**

src/storage/adapters/fileSystemStorage.ts

@@ -1785,6 +1785,32 @@ export class FileSystemStorage extends BaseStorage {
       const now = new Date().toISOString()
       const existing = await this.readWriterLock()
 
+      // TORN-LOCK RECOVERY: power loss can legally leave the lock file
+      // present but EMPTY/unparseable (the claim's non-atomic write died
+      // mid-flight). readWriterLock() reports it as null — but the O_EXCL
+      // claim below would EEXIST forever, a PERMANENT lockout no staleness
+      // check can clear (staleness needs a parsed PID). A torn lock is
+      // stale BY DEFINITION: no live holder has one (a holder either
+      // completed its write or is dead). Unlink loudly and re-loop; a
+      // racer that rewrites a VALID lock first simply wins the next read.
+      if (existing === null) {
+        try {
+          await fs.promises.access(lockFile)
+          console.warn(
+            `[brainy] Writer lock at ${lockFile} exists but is unreadable/unparseable ` +
+              `(torn write from a previous power loss) — treating as stale and removing.`
+          )
+          try {
+            await fs.promises.unlink(lockFile)
+          } catch (unlinkErr: any) {
+            if (unlinkErr.code !== 'ENOENT') throw unlinkErr
+          }
+        } catch (accessErr: any) {
+          if (accessErr.code !== 'ENOENT') throw accessErr
+          // Absent: the normal fresh-claim path below.
+        }
+      }
+
       if (existing) {
         // Same-process re-open: a second Brainy instance in this Node process
         // (e.g. test "simulate server restart" patterns, or a consumer that

tests/integration/durability-kill-matrix.test.ts

@@ -37,6 +37,7 @@
  */
 import { describe, it, expect, afterEach } from 'vitest'
 import * as fs from 'node:fs'
+import { join } from 'node:path'
 import { Brainy } from '../../src/brainy.js'
 import { NounType } from '../../src/types/graphTypes.js'
 import {
@@ -630,4 +631,71 @@ describe('durability kill matrix — crash at every commit-path step, recover by
     expect(storeOf(brain).committedGeneration()).toBe(floor)
     expect(await factGenerations(brain)).toEqual([floor])
   })
+
+  // ==========================================================================
+  // Block-layer power-loss findings (first dm-flakey run) — the three cures
+  // ==========================================================================
+
+  it('at-ack POWER LOSS BELOW THE MANIFEST — an unclean open folds the WHOLE log; acked writes committed before the flush still survive vanished canonical', async () => {
+    const { dir, brain, baselineId } = await arrangeBaseline('wlf')
+    await flipToAtAck(brain)
+    const ackedA = uid('wlf-a')
+    const ackedB = uid('wlf-b')
+    await brain.add({ id: ackedA, data: 'below manifest one', type: NounType.Document, vector: vec(2), metadata: { v: 2 } })
+    await brain.add({ id: ackedB, data: 'below manifest two', type: NounType.Document, vector: vec(3), metadata: { v: 3 } })
+    // The group-commit flush advances the manifest OVER these generations —
+    // but live canonical bytes are tmp+rename without per-file fsync, so a
+    // power cut can still take them. The fsynced facts are the durable copy.
+    await (brain as unknown as { flush(): Promise<void> }).flush()
+    await abandonAsCrashed(brain) // no clean close → no clean-shutdown marker
+    dropCanonicalNoun(dir, ackedA)
+    dropCanonicalNoun(dir, ackedB)
+
+    const reopened = await openLive(dir)
+    // The whole-log fold restores BOTH rows from facts ≤ manifest.
+    expect(((await reopened.get(ackedA)) as { metadata: { v: number } }).metadata.v).toBe(2)
+    expect(((await reopened.get(ackedB)) as { metadata: { v: number } }).metadata.v).toBe(3)
+    expect(((await reopened.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(1)
+  })
+
+  it('clean-shutdown marker: a clean close writes it, the next open consumes it (no fold on the happy path)', async () => {
+    const { dir, brain } = await arrangeBaseline('csm')
+    await flipToAtAck(brain)
+    await brain.close()
+    liveBrains.splice(liveBrains.indexOf(brain), 1)
+    // The adapter stores raw objects gzipped — accept either spelling.
+    const markerExists = () =>
+      fs.existsSync(join(dir, '_system', 'clean-shutdown.json')) ||
+      fs.existsSync(join(dir, '_system', 'clean-shutdown.json.gz'))
+    expect(markerExists(), 'clean close stamps the marker').toBe(true)
+
+    const reopened = await openLive(dir)
+    expect(markerExists(), 'open consumes the marker').toBe(false)
+    await reopened.close()
+    liveBrains.splice(liveBrains.indexOf(reopened), 1)
+    expect(markerExists(), 'the next clean close re-stamps it').toBe(true)
+  })
+
+  it('torn writer lock (empty file) — open treats it as stale and recovers; never a permanent lockout', async () => {
+    const { dir, brain } = await arrangeBaseline('tlk')
+    await brain.close()
+    liveBrains.splice(liveBrains.indexOf(brain), 1)
+    // The power-loss shape: the lock file exists but is EMPTY (torn write).
+    fs.writeFileSync(join(dir, 'locks', '_writer.lock'), '')
+
+    const reopened = await openLive(dir) // must not throw 'contended'
+    const fresh = uid('tlk-fresh')
+    await reopened.add({ id: fresh, data: 'lock recovered', type: NounType.Document, vector: vec(4), metadata: { v: 4 } })
+    expect(await reopened.get(fresh)).not.toBeNull()
+  })
+
+  it('pair guard: a metadata index without stampWatermark never crashes flush', async () => {
+    const { brain } = await arrangeBaseline('psg')
+    liveBrains.push(brain)
+    // The native pair swaps the metadata manager; the replacement may not
+    // carry the stamp method — flush must treat that as verdict-side rescan,
+    // never a TypeError at the fan-out.
+    ;(brain as unknown as { metadataIndex: { stampWatermark?: unknown } }).metadataIndex.stampWatermark = undefined
+    await expect((brain as unknown as { flush(): Promise<void> }).flush()).resolves.toBeUndefined()
+  })
 })