diff --git a/CHANGELOG.md b/CHANGELOG.md index 7154d5a2..57c83f6b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,11 @@ All notable changes to this project will be documented in this file. See [standard-version](https://github.com/conventional-changelog/standard-version) for commit guidelines. +### [10.4.7](https://source.soulcraft.com/soulcraftlabs/open-brainy/compare/v10.4.6...v10.4.7) (2026-09-01) + +- fix(storage): counts persistence is single-flight, coalesced, and never races its own temp file (5e3b343a) + + ### [10.4.6](https://source.soulcraft.com/soulcraftlabs/open-brainy/compare/v10.4.5...v10.4.6) (2026-08-31) - fix(transact): metadata-index ops take their JSON-safe view at the crossing, not at construction (73500e7d) diff --git a/package-lock.json b/package-lock.json index 9e573da3..9add5a2e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@soulcraftlabs/brainy", - "version": "10.4.6", + "version": "10.4.7", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@soulcraftlabs/brainy", - "version": "10.4.6", + "version": "10.4.7", "license": "MIT", "dependencies": { "@msgpack/msgpack": "^3.1.2", diff --git a/package.json b/package.json index 51322998..7c74b0fd 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@soulcraftlabs/brainy", - "version": "10.4.6", + "version": "10.4.7", "brainyContract": 1, "description": "Universal Knowledge Protocol™ - World's first Triple Intelligence database unifying vector, graph, and document search in one API. Stage 3 CANONICAL: 42 nouns × 127 verbs covering 96-97% of all human knowledge.", "main": "dist/index.js", diff --git a/src/storage/adapters/baseStorageAdapter.ts b/src/storage/adapters/baseStorageAdapter.ts index cabe2e30..a90adb93 100644 --- a/src/storage/adapters/baseStorageAdapter.ts +++ b/src/storage/adapters/baseStorageAdapter.ts @@ -1089,6 +1089,10 @@ export abstract class BaseStorageAdapter implements StorageAdapter { // Counts changed since the last persist? Drives the write-through flush. protected pendingCountPersist = false + /** The one persist running right now, if any (single-flight law — see flushCounts). */ + private countPersistInFlight: Promise | null = null + /** The one trailing persist a burst has queued behind the in-flight one. */ + private countPersistTrailing: Promise | null = null /** * Get total noun count - O(1) operation @@ -1341,15 +1345,46 @@ export abstract class BaseStorageAdapter implements StorageAdapter { return } - try { - // Persist to storage (implemented by subclass) - await this.persistCounts() - this.pendingCountPersist = false - } catch (error) { - console.error('CRITICAL: Failed to flush counts to storage:', error) - // Keep pending flag set so we retry on next operation - throw error + // SINGLE-FLIGHT, COALESCED. Counts are write-through on every change, so + // a burst of writes used to launch one persist per change, all in flight + // together. Two of them inside the same millisecond shared the atomic + // writer's temp path (`.tmp--`): both wrote it, the first rename + // consumed it, the second rename found nothing — ENOENT, ~1,500 times a + // day on a busy production brain, with a full ledger write per change + // behind it. Now exactly one persist runs at a time; requests that arrive + // while it runs collapse into ONE trailing persist that carries the final + // state. A burst of N changes costs at most two writes and never races + // itself. + if (this.countPersistInFlight) { + // The in-flight write may have already serialised a stale snapshot — + // ask for one more pass after it, and let every caller in this burst + // await that same pass. + if (!this.countPersistTrailing) { + this.countPersistTrailing = this.countPersistInFlight + .catch(() => undefined) + .then(() => { + this.countPersistTrailing = null + return this.flushCounts() + }) + } + return this.countPersistTrailing } + + this.countPersistInFlight = (async () => { + try { + // Persist to storage (implemented by subclass) + this.pendingCountPersist = false + await this.persistCounts() + } catch (error) { + // Keep the flag set so the next operation retries. + this.pendingCountPersist = true + console.error('CRITICAL: Failed to flush counts to storage:', error) + throw error + } finally { + this.countPersistInFlight = null + } + })() + return this.countPersistInFlight } /** diff --git a/src/storage/adapters/fileSystemStorage.ts b/src/storage/adapters/fileSystemStorage.ts index 5ec1d88e..87b6406f 100644 --- a/src/storage/adapters/fileSystemStorage.ts +++ b/src/storage/adapters/fileSystemStorage.ts @@ -2400,8 +2400,15 @@ export class FileSystemStorage extends BaseStorage { * Atomic write via temp-file-then-rename so concurrent readers never see a * half-written lock JSON. Reused by writer-lock writes + heartbeat. */ + /** Monotonic per-process sequence so two atomic writes never share a temp path. */ + private static atomicWriteSeq = 0 + private async writeFileAtomic(filePath: string, contents: string): Promise { - const tmp = `${filePath}.tmp-${process.pid}-${Date.now()}` + // pid + timestamp alone collided: two writers of the same target inside + // one millisecond shared this path, and the loser's rename found the + // winner had already moved it (ENOENT). The sequence makes every call's + // temp path its own. + const tmp = `${filePath}.tmp-${process.pid}-${Date.now()}-${++FileSystemStorage.atomicWriteSeq}` await fs.promises.writeFile(tmp, contents) await fs.promises.rename(tmp, filePath) } diff --git a/tests/integration/counts-persist-single-flight.test.ts b/tests/integration/counts-persist-single-flight.test.ts new file mode 100644 index 00000000..5acbdcc3 --- /dev/null +++ b/tests/integration/counts-persist-single-flight.test.ts @@ -0,0 +1,111 @@ +/** + * @module tests/integration/counts-persist-single-flight + * @description Regression for a production race in FileSystemStorage's + * counts ledger: `persistCounts()` was write-through on every count change + * with no serialization, and the atomic writer named its temp file with + * millisecond granularity (`.tmp--`). Two persists inside one + * millisecond shared the temp path — both wrote it, the first rename + * consumed it, the second rename found nothing: ENOENT, ~1,500 times a day + * on a busy production brain, with a full ledger write per change behind it. + * + * Under pin: persists are single-flight and coalesced — one in flight, at + * most one trailing pass carrying the burst's final state — and every atomic + * write owns a unique temp path. A burst of N count changes costs at most + * two ledger writes, never errors, and leaves a ledger equal to memory. + */ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest' +import * as fs from 'node:fs' +import * as os from 'node:os' +import * as path from 'node:path' +import { Brainy } from '../../src/brainy.js' +import { NounType } from '../../src/types/graphTypes.js' + +describe('counts persistence is single-flight, coalesced, and never races its own temp file', () => { + let dir: string + let brain: any + + beforeEach(async () => { + process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true' + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-counts-race-')) + brain = new Brainy({ + requireSubtype: false, + storage: { type: 'filesystem', path: dir }, + dimensions: 384, + silent: true + }) + await brain.init() + }) + + afterEach(async () => { + vi.restoreAllMocks() + await brain.close() + fs.rmSync(dir, { recursive: true, force: true }) + }) + + it('a burst of concurrent count changes → at most two ledger writes, zero errors, ledger == memory', async () => { + const storage = brain.storage + const countsPath: string = storage.countsFilePath + expect(countsPath, 'the filesystem adapter persists a counts ledger').toBeTruthy() + + // Let init's own persists settle so the burst is measured alone. + await storage.flushCounts?.() + + const renameSpy = vi.spyOn(fs.promises, 'rename') + const errorSpy = vi.spyOn(console, 'error') + + // Twenty-five concurrent count changes — the shape of a write burst; each + // used to launch its own persist. + const BURST = 25 + await Promise.all( + Array.from({ length: BURST }, () => storage.scheduleCountPersist()) + ) + + const ledgerRenames = renameSpy.mock.calls.filter(([, to]) => String(to) === countsPath) + expect(ledgerRenames.length, 'single-flight + one trailing pass').toBeLessThanOrEqual(2) + expect(ledgerRenames.length, 'the burst was persisted at all').toBeGreaterThanOrEqual(1) + + const persistErrors = errorSpy.mock.calls.filter((args) => String(args[0]).includes('persisting counts')) + expect(persistErrors).toEqual([]) + + const ledger = JSON.parse(fs.readFileSync(countsPath, 'utf-8')) + expect(ledger.totalNounCount).toBe(storage.totalNounCount) + expect(ledger.totalVerbCount).toBe(storage.totalVerbCount) + }) + + it('real writes in parallel: the ledger lands complete and no persist error is logged', async () => { + const storage = brain.storage + const countsPath: string = storage.countsFilePath + const errorSpy = vi.spyOn(console, 'error') + + await Promise.all( + Array.from({ length: 12 }, (_, i) => + brain.add({ data: `burst row ${i}`, type: NounType.Thing }) + ) + ) + await storage.flushCounts?.() + + const persistErrors = errorSpy.mock.calls.filter((args) => String(args[0]).includes('persisting counts')) + expect(persistErrors).toEqual([]) + const ledger = JSON.parse(fs.readFileSync(countsPath, 'utf-8')) + expect(ledger.totalNounCount).toBe(storage.totalNounCount) + expect(await brain.getNounCount()).toBe(ledger.totalNounCount) + }) + + it('every atomic write owns its own temp path — two writes in one millisecond never collide', async () => { + const storage = brain.storage + const tmpNames: string[] = [] + vi.spyOn(fs.promises, 'writeFile').mockImplementation(async (p: any) => { + tmpNames.push(String(p)) + }) + vi.spyOn(fs.promises, 'rename').mockImplementation(async () => undefined) + const target = path.join(dir, 'probe.json') + await Promise.all([ + storage.writeFileAtomic(target, '{"a":1}'), + storage.writeFileAtomic(target, '{"a":2}'), + storage.writeFileAtomic(target, '{"a":3}') + ]) + const probeTmps = tmpNames.filter((n) => n.startsWith(`${target}.tmp-`)) + expect(probeTmps.length).toBe(3) + expect(new Set(probeTmps).size, 'no two writes shared a temp path').toBe(3) + }) +})