fix(storage): counts persistence is single-flight, coalesced, and never races its own temp file
persistCounts() was write-through on every count change with no serialization, and the atomic writer named its temp file with millisecond granularity. Two persists inside one millisecond shared the temp path: both wrote it, the first rename consumed it, the second rename found nothing — ENOENT, roughly 1,500 times a day on a busy production brain, with a full ledger write per change behind it. No data was lost (the surviving rename carried a complete ledger and the next change re-persisted), but the race was real and the write rate absurd. flushCounts() now runs exactly one persist at a time; requests arriving during it collapse into one trailing pass that carries the burst's final state — N changes cost at most two writes. writeFileAtomic() adds a per-process sequence to the temp name so no two writes can share a path. Pinned: a 25-change burst → ≤2 ledger writes, zero errors, ledger equal to memory; parallel real writes land complete; three same-instant atomic writes own three distinct temp paths.
This commit is contained in:
parent
4014e0f125
commit
5e3b343a0e
3 changed files with 162 additions and 9 deletions
|
|
@ -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<void> | null = null
|
||||
/** The one trailing persist a burst has queued behind the in-flight one. */
|
||||
private countPersistTrailing: Promise<void> | 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-<pid>-<ms>`): 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
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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)
|
||||
}
|
||||
|
|
|
|||
111
tests/integration/counts-persist-single-flight.test.ts
Normal file
111
tests/integration/counts-persist-single-flight.test.ts
Normal file
|
|
@ -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-<pid>-<ms>`). 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)
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue