David Snelling
70e4bc8a79
clearInterval() stops future heartbeat ticks but not one already in flight: a straggler tick past its ownership guards could land its atomic lock rewrite AFTER releaseWriterLock()'s unlink, re-creating the lock file as a phantom that blocks the next writer until the stale TTL expires (~60s) — the pool-eviction reopen case. Found as an ENOENT heartbeat warning during benchmark teardown; the quiet variant is the harmful one. releaseWriterLock() now awaits the in-flight tick (tracked per tick, self-clearing) before reading/unlinking, so a straggler's write always lands BEFORE the unlink and gets removed with everything else. The heartbeat's ENOENT is also now benign-by-contract (lock or directory removed under us — the next acquire recreates it); other errors stay loud. Pin: straggler-past-guards simulation — lock file absent after close, directory immediately claimable (fails on the undrained code).
321 lines
14 KiB
TypeScript
321 lines
14 KiB
TypeScript
/**
|
|
* Multi-process safety + read-only mode integration tests.
|
|
*
|
|
* Covers the surface added in 7.21.0:
|
|
* - Brainy.openReadOnly() enforces ReaderMode on every mutation entry.
|
|
* - Two concurrent writers on the same filesystem directory: second throws.
|
|
* - { force: true } overrides a live writer lock.
|
|
* - Flush-request RPC: in-process shortcut + cross-process round-trip.
|
|
* - stats(), explain(), health() return useful diagnostics in read-only mode.
|
|
*
|
|
* Uses a temp directory per `describe` block so tests don't share state.
|
|
*/
|
|
|
|
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
|
|
import { mkdtempSync, rmSync } from 'node:fs'
|
|
import { tmpdir } from 'node:os'
|
|
import { join } from 'node:path'
|
|
import { Brainy } from '../../src/brainy.js'
|
|
import { NounType } from '../../src/types/graphTypes.js'
|
|
|
|
function makeTempDir(): string {
|
|
return mkdtempSync(join(tmpdir(), 'brainy-mp-'))
|
|
}
|
|
|
|
describe('Multi-process safety + read-only mode', () => {
|
|
let dir: string
|
|
let writer: Brainy | null = null
|
|
let reader: Brainy | null = null
|
|
|
|
beforeEach(() => {
|
|
dir = makeTempDir()
|
|
})
|
|
|
|
afterEach(async () => {
|
|
if (writer) {
|
|
try { await writer.close() } catch { /* may already be closed */ }
|
|
writer = null
|
|
}
|
|
if (reader) {
|
|
try { await reader.close() } catch { /* may already be closed */ }
|
|
reader = null
|
|
}
|
|
try { rmSync(dir, { recursive: true, force: true }) } catch { /* ignore */ }
|
|
})
|
|
|
|
describe('ReaderMode enforcement', () => {
|
|
it('rejects every mutation when opened via openReadOnly()', async () => {
|
|
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await writer.init()
|
|
await writer.add({ data: 'seed entity', type: NounType.Concept })
|
|
await writer.flush()
|
|
await writer.close()
|
|
writer = null
|
|
|
|
reader = await Brainy.openReadOnly({
|
|
storage: { type: 'filesystem', path: dir }
|
|
})
|
|
|
|
expect(reader.isReadOnly).toBe(true)
|
|
await expect(reader.add({ data: 'x', type: NounType.Concept })).rejects.toThrow(/read-only/i)
|
|
await expect(reader.addMany({ items: [] })).rejects.toThrow(/read-only/i)
|
|
await expect(reader.update({ id: 'x', data: 'y' })).rejects.toThrow(/read-only/i)
|
|
await expect(reader.remove('x')).rejects.toThrow(/read-only/i)
|
|
await expect(reader.removeMany({ ids: ['x'] } as any)).rejects.toThrow(/read-only/i)
|
|
await expect(reader.relate({ from: 'x', to: 'y' } as any)).rejects.toThrow(/read-only/i)
|
|
await expect(reader.unrelate('x')).rejects.toThrow(/read-only/i)
|
|
})
|
|
|
|
it('flush() and close() are safe to call in read-only mode', async () => {
|
|
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await writer.init()
|
|
await writer.add({ data: 'thing', type: NounType.Concept })
|
|
await writer.flush()
|
|
await writer.close()
|
|
writer = null
|
|
|
|
reader = await Brainy.openReadOnly({
|
|
storage: { type: 'filesystem', path: dir }
|
|
})
|
|
await expect(reader.flush()).resolves.toBeUndefined()
|
|
await expect(reader.close()).resolves.toBeUndefined()
|
|
reader = null
|
|
})
|
|
})
|
|
|
|
describe('Writer lock', () => {
|
|
it('blocks a writer when a different live PID holds the directory', async () => {
|
|
// Simulate a cross-process lock holder by writing the lock file by hand
|
|
// with a real PID that is not ours. Node itself (PID 1 on most container
|
|
// hosts is `init`; on dev hosts it's an existing process) is always
|
|
// alive for the duration of this test.
|
|
const { mkdirSync, writeFileSync } = await import('node:fs')
|
|
const { join } = await import('node:path')
|
|
const os = await import('node:os')
|
|
mkdirSync(join(dir, 'locks'), { recursive: true })
|
|
// Use a PID we know is alive but isn't ours: the parent of our own
|
|
// process. On every Unix this is the shell or test runner.
|
|
const otherPid = (process as any).ppid || 1
|
|
writeFileSync(join(dir, 'locks', '_writer.lock'), JSON.stringify({
|
|
pid: otherPid,
|
|
hostname: os.hostname(),
|
|
startedAt: new Date().toISOString(),
|
|
lastHeartbeat: new Date().toISOString(),
|
|
version: '7.21.0',
|
|
rootDir: dir
|
|
}))
|
|
|
|
const blocked = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await expect(blocked.init()).rejects.toThrow(/another writer holds/i)
|
|
// Don't track `blocked` for afterEach cleanup since init failed.
|
|
})
|
|
|
|
it('takes over a STALE foreign lock (dead PID + old heartbeat) and claims atomically', async () => {
|
|
const { mkdirSync, writeFileSync, readFileSync } = await import('node:fs')
|
|
const { join } = await import('node:path')
|
|
const os = await import('node:os')
|
|
mkdirSync(join(dir, 'locks'), { recursive: true })
|
|
const tenMinutesAgo = new Date(Date.now() - 10 * 60 * 1000).toISOString()
|
|
writeFileSync(join(dir, 'locks', '_writer.lock'), JSON.stringify({
|
|
pid: 999999999, // no such process — provably dead
|
|
hostname: os.hostname(),
|
|
startedAt: tenMinutesAgo,
|
|
lastHeartbeat: tenMinutesAgo,
|
|
version: '8.0.0',
|
|
rootDir: dir
|
|
}))
|
|
|
|
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await writer.init() // stale takeover must succeed
|
|
|
|
const lock = JSON.parse(readFileSync(join(dir, 'locks', '_writer.lock'), 'utf-8'))
|
|
expect(lock.pid).toBe(process.pid) // the atomic wx claim installed OUR lock
|
|
})
|
|
|
|
it('the writer-locked error carries the machine-readable contract (code + lockInfo)', async () => {
|
|
const { mkdirSync, writeFileSync } = await import('node:fs')
|
|
const { join } = await import('node:path')
|
|
const os = await import('node:os')
|
|
mkdirSync(join(dir, 'locks'), { recursive: true })
|
|
const otherPid = (process as any).ppid || 1
|
|
writeFileSync(join(dir, 'locks', '_writer.lock'), JSON.stringify({
|
|
pid: otherPid,
|
|
hostname: os.hostname(),
|
|
startedAt: new Date().toISOString(),
|
|
lastHeartbeat: new Date().toISOString(),
|
|
version: '8.7.0',
|
|
rootDir: dir
|
|
}))
|
|
|
|
const blocked = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
const err: any = await blocked.init().catch((e) => e)
|
|
expect(err.code).toBe('BRAINY_WRITER_LOCKED')
|
|
expect(err.lockInfo?.pid).toBe(otherPid)
|
|
})
|
|
|
|
it('release drains an in-flight heartbeat — no phantom lock re-created after unlink', async () => {
|
|
// The race (8.9.0): clearInterval stops FUTURE heartbeat ticks, but a
|
|
// tick already in flight could land its lock rewrite AFTER release's
|
|
// unlink — re-creating the lock as a phantom that blocks the next
|
|
// writer until the stale TTL. Simulate the in-flight tick explicitly
|
|
// and prove release waits for it.
|
|
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await writer.init()
|
|
const storage: any = (writer as any).storage
|
|
|
|
// An in-flight refresh that is ALREADY PAST its ownership guards
|
|
// (captured the lock info before release ran) and lands its atomic
|
|
// rewrite slowly — the exact straggler shape; absent the drain it
|
|
// writes after the unlink.
|
|
const { join: joinPath } = await import('node:path')
|
|
const capturedInfo = { ...storage.writerLockInfo }
|
|
const lockPath = joinPath(dir, 'locks', '_writer.lock')
|
|
const slowTick = (async () => {
|
|
await new Promise((r) => setTimeout(r, 100))
|
|
await storage.writeFileAtomic(
|
|
lockPath,
|
|
JSON.stringify({ ...capturedInfo, lastHeartbeat: new Date().toISOString() })
|
|
)
|
|
})()
|
|
storage.writerHeartbeatInFlight = slowTick.catch(() => {})
|
|
|
|
await writer.close() // → releaseWriterLock must drain slowTick first
|
|
await slowTick.catch(() => {}) // both paths fully settled either way
|
|
writer = null
|
|
|
|
const { existsSync } = await import('node:fs')
|
|
const { join } = await import('node:path')
|
|
expect(existsSync(join(dir, 'locks', '_writer.lock'))).toBe(false)
|
|
|
|
// And the directory is immediately claimable — no stale-TTL wait.
|
|
const next = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await expect(next.init()).resolves.toBeUndefined()
|
|
await next.close()
|
|
})
|
|
|
|
it('allows a second in-process writer with a warning (same PID)', async () => {
|
|
// Two Brainy instances in the same Node process: not the dangerous
|
|
// cross-process case. Should succeed (with a console warning).
|
|
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await writer.init()
|
|
|
|
const second = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await expect(second.init()).resolves.toBeUndefined()
|
|
await second.close()
|
|
})
|
|
|
|
it('lets a reader open while a writer is live', async () => {
|
|
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await writer.init()
|
|
await writer.add({ data: 'concurrent', type: NounType.Concept })
|
|
await writer.flush()
|
|
|
|
reader = await Brainy.openReadOnly({
|
|
storage: { type: 'filesystem', path: dir }
|
|
})
|
|
const stats = await reader.stats()
|
|
expect(stats.mode).toBe('reader')
|
|
expect(stats.writerLock).toBeDefined()
|
|
expect(stats.writerLock!.pid).toBe(process.pid)
|
|
})
|
|
|
|
it('honors { force: true } to override an existing lock', async () => {
|
|
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await writer.init()
|
|
|
|
const second = new Brainy({ requireSubtype: false,
|
|
storage: { type: 'filesystem', path: dir },
|
|
force: true
|
|
})
|
|
await expect(second.init()).resolves.toBeUndefined()
|
|
await second.close()
|
|
})
|
|
})
|
|
|
|
describe('Flush-request RPC', () => {
|
|
it('in-process call just flushes', async () => {
|
|
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await writer.init()
|
|
const ok = await writer.requestFlush({ timeoutMs: 1000 })
|
|
expect(ok).toBe(true)
|
|
})
|
|
|
|
it('cross-instance request reaches the writer (same-process simulation)', async () => {
|
|
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await writer.init()
|
|
await writer.add({ data: 'pre-request', type: NounType.Concept })
|
|
|
|
// openReadOnly() in the same Node process — the storage will still hit
|
|
// the writer's flush watcher via the shared filesystem.
|
|
reader = await Brainy.openReadOnly({
|
|
storage: { type: 'filesystem', path: dir }
|
|
})
|
|
const acked = await reader.requestFlush({ timeoutMs: 3000 })
|
|
expect(acked).toBe(true)
|
|
})
|
|
|
|
it('returns false when no writer is running', async () => {
|
|
// Seed some data, then close the writer. The data dir exists but no
|
|
// writer is listening for flush requests.
|
|
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await writer.init()
|
|
await writer.add({ data: 'orphan', type: NounType.Concept })
|
|
await writer.flush()
|
|
await writer.close()
|
|
writer = null
|
|
|
|
reader = await Brainy.openReadOnly({
|
|
storage: { type: 'filesystem', path: dir }
|
|
})
|
|
const acked = await reader.requestFlush({ timeoutMs: 1500 })
|
|
expect(acked).toBe(false)
|
|
})
|
|
})
|
|
|
|
describe('Diagnostics on a reader', () => {
|
|
beforeEach(async () => {
|
|
writer = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
|
|
await writer.init()
|
|
await writer.add({ data: 'one', type: NounType.Concept, metadata: { tag: 'a' } })
|
|
await writer.add({ data: 'two', type: NounType.Concept, metadata: { tag: 'b' } })
|
|
await writer.flush()
|
|
})
|
|
|
|
it('stats() reports counts, mode, and writer lock metadata', async () => {
|
|
reader = await Brainy.openReadOnly({
|
|
storage: { type: 'filesystem', path: dir }
|
|
})
|
|
const stats = await reader.stats()
|
|
expect(stats.mode).toBe('reader')
|
|
// Both Concept entities the writer added should be visible to the
|
|
// reader, classified correctly as 'concept' (not 'thing'). This is
|
|
// the BR-FIND-WHERE-ZERO regression test: stats() must read from the
|
|
// column store via idMapper.size, and getNounType() must use the
|
|
// write-time type cache instead of the old hardcoded 'thing'.
|
|
expect(stats.entityCount).toBeGreaterThanOrEqual(2)
|
|
expect(stats.entitiesByType.concept).toBeGreaterThanOrEqual(2)
|
|
expect(stats.writerLock).toBeDefined()
|
|
expect(stats.writerLock!.pid).toBe(process.pid)
|
|
expect(stats.version).toBeTruthy()
|
|
})
|
|
|
|
it('explain() flags a field with no index entries', async () => {
|
|
reader = await Brainy.openReadOnly({
|
|
storage: { type: 'filesystem', path: dir }
|
|
})
|
|
const plan = await reader.explain({ where: { entityTypeXYZ: 'never-registered' } })
|
|
expect(plan.fieldPlan).toHaveLength(1)
|
|
expect(plan.fieldPlan[0].path).toBe('none')
|
|
expect(plan.warnings.length).toBeGreaterThan(0)
|
|
})
|
|
|
|
it('health() returns checks with a pass/warn/fail overall', async () => {
|
|
reader = await Brainy.openReadOnly({
|
|
storage: { type: 'filesystem', path: dir }
|
|
})
|
|
const report = await reader.health()
|
|
expect(report.checks.length).toBeGreaterThan(0)
|
|
expect(['pass', 'warn', 'fail']).toContain(report.overall)
|
|
})
|
|
})
|
|
})
|