diff --git a/src/brainy.ts b/src/brainy.ts index 6d04f78d..57ad0c73 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -11247,7 +11247,10 @@ export class Brainy implements BrainyInterface { { 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) } diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index 422f062f..c25e2326 100644 --- a/src/db/generationStore.ts +++ b/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 { + 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 { + try { + await this.storage.deleteRawObject(CLEAN_SHUTDOWN_PATH) + } catch { + // Absent or undeletable: the conservative outcome is a future replay. + } } /** diff --git a/src/storage/adapters/fileSystemStorage.ts b/src/storage/adapters/fileSystemStorage.ts index 5eb4785a..c719b63d 100644 --- a/src/storage/adapters/fileSystemStorage.ts +++ b/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 diff --git a/tests/integration/durability-kill-matrix.test.ts b/tests/integration/durability-kill-matrix.test.ts index 1e543bc1..35540e5a 100644 --- a/tests/integration/durability-kill-matrix.test.ts +++ b/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 }).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 }).flush()).resolves.toBeUndefined() + }) })