diff --git a/src/brainy.ts b/src/brainy.ts index dc97b82f..d7313855 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -9229,7 +9229,13 @@ export class Brainy implements BrainyInterface { } const floorGeneration = this.generationStore.generation() - await this.storage.restoreFromDirectory(path) + // The swap runs inside the generation store's exclusive section: pending + // flush timers are disarmed and buffers discarded BEFORE any directory is + // removed, so a background flush can never write into `_system/` mid-swap + // (the ENOTEMPTY race a checkpoint stamp once hit). + await this.generationStore.runStateReplacement(() => + this.storage.restoreFromDirectory(path) + ) await this.generationStore.reopenAfterRestore(floorGeneration) // If the entity-id mapper is a NATIVE provider with a `rebuild()`, reload it diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index 837ea90a..4002c1ba 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -3127,6 +3127,28 @@ export class GenerationStore { * are never reissued. * @param floorGeneration - The counter value before the restore. */ + /** + * @description Run a wholesale state replacement (the restore swap) + * EXCLUSIVELY: under the commit mutex, with the pending flush timer + * disarmed and the pending tier + fold-checkpoint accumulator discarded + * FIRST — so no background flush can write into `_system/` while the + * replacement is removing and swapping directories. Observed without this: + * a checkpoint stamp raced restore's directory removal and the swap died + * ENOTEMPTY mid-flight. The discarded in-memory state describes the store + * being replaced — `reopenAfterRestore` (which the caller runs next) + * rebuilds everything from the restored bytes. + */ + async runStateReplacement(replace: () => Promise): Promise { + return this.withMutex(async () => { + this.clearPendingFlushTimer() + this.pendingGens = [] + this.pendingBuffer.clear() + this.checkpointDirtyNouns = new Set() + this.checkpointDirtyVerbs = new Set() + await replace() + }) + } + async reopenAfterRestore(floorGeneration: number): Promise { await this.withMutex(async () => { this.deltaCache.clear() @@ -3139,6 +3161,27 @@ export class GenerationStore { this.clearPendingFlushTimer() this.pendingGens = [] this.pendingBuffer.clear() + // The fold-checkpoint accumulator described the replaced state too. + this.checkpointDirtyNouns = new Set() + this.checkpointDirtyVerbs = new Set() + this.foldCheckpointChainValid = false + this.foldCheckpoint = 0 + // A RESTORE IS AN UNCLEAN EVENT, by construction: the snapshot's files + // were just bulk-copied WITHOUT per-file fsync, so a power cut here can + // tear them — yet the snapshot may CARRY the source brain's + // clean-shutdown marker and fold checkpoint, which would together + // suppress exactly the recovery fold that cures such a tear. Delete + // both BEFORE reopening: the open below then treats the store as + // uncleanly shut, folds the restored log into canonical, barrier-syncs + // what it re-applied, and stamps a FRESH checkpoint — the restored + // state becomes durably founded at restore time instead of inheriting + // the source brain's assertions about bytes this disk never synced. + try { + await this.storage.deleteRawObject(CLEAN_SHUTDOWN_PATH) + } catch { /* absent is fine — same outcome */ } + try { + await this.storage.deleteRawObject(FOLD_CHECKPOINT_PATH) + } catch { /* absent is fine — fold from 0 */ } this.opened = false // open() re-reads counter/manifest and re-registers the bump hook. await this.open() diff --git a/tests/integration/fold-checkpoint-bound.test.ts b/tests/integration/fold-checkpoint-bound.test.ts index ce074dcf..60bcce5e 100644 --- a/tests/integration/fold-checkpoint-bound.test.ts +++ b/tests/integration/fold-checkpoint-bound.test.ts @@ -172,6 +172,41 @@ describe('fold-checkpoint bound — crash recovery folds (checkpoint, head], nev expect(readCheckpoint(dir)).toBeNull() }, 120000) + it('restore is an UNCLEAN event: the snapshot’s stamps do not survive — the reopen fold re-founds and re-stamps the restored state', async () => { + const dir = trackDir() + const brain = await openBrain(dir, { logAuthority: 'adopt' }) + liveBrains.push(brain) + const idA = await brain.add({ data: 'survives the restore', type: NounType.Document, metadata: { n: 1 } }) + await brain.flush() + + const snapDir = join(trackDir(), 'snap') + const db = brain.now() + await (db as unknown as { persist(p: string): Promise }).persist(snapDir) + await (db as unknown as { release(): Promise }).release() + + // Advance the live brain past the snapshot: a later write, a later flush, + // a later checkpoint stamp — none of which may survive the restore. + const idB = await brain.add({ data: 'must not survive', type: NounType.Document, metadata: { n: 2 } }) + await brain.flush() + const stampBeforeRestore = readCheckpoint(dir) + expect(stampBeforeRestore).toBe(committedOf(brain)) + + // Unflushed traffic in flight at restore time — the quiesced swap discards + // it under the mutex instead of letting its flush timer race the swap + // (the ENOTEMPTY class). + await brain.add({ data: 'in-flight at restore', type: NounType.Document, metadata: { n: 3 } }) + await brain.restore(snapDir, { confirm: true }) + + expect(await brain.get(idA), 'snapshot state restored').not.toBeNull() + expect(await brain.get(idB), 'post-snapshot state replaced').toBeNull() + // The stamp on disk is the REOPEN FOLD's fresh assertion about the + // restored (and now barrier-synced) bytes — at the restored watermark, + // strictly below the pre-restore stamp that must not survive. + const stampAfterRestore = readCheckpoint(dir) + expect(stampAfterRestore).toBe(committedOf(brain)) + expect(stampAfterRestore!).toBeLessThan(stampBeforeRestore!) + }, 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' })