feat(log): log authority is the fleet default — adopt-at-open, oracle-gated; plus the power-cut throw-site cures and the loud torn-record contract
THE DEFAULT FLIP (ruled on proven evidence — at-ack survived 301/301 acked-writes-through-power-cut in block-layer fault injection; deferred tree authority demonstrably loses flush-covered acks): a brain with NO stored authority artifact now ADOPTS LOG AUTHORITY AT OPEN. The oracle gates the flip exactly as the guarded adoption path always did — curable divergences baseline-backfilled, the flip lands ONLY on a green verdict — and a brain that cannot verify STAYS tree-authoritative loudly, with the refusal recorded on the switch artifact so subsequent opens are cheap. config logAuthority: 'defer' is the explicit documented opt-out (no automatic adoption; declared flush-window loss; adoptLogAuthority() flips later). A stored artifact always wins. RELEASES.md carries the posture. Two standing .fails debt pins FLIP TO HOLDING under the default: the at-ack crash-survival gap and the ack-at-log durability target — both now permanent asserted truths, not aspirations. POWER-CUT THROW SITES (fault-injection findings, brainy-alone config): - A manifest-listed-but-unloadable column segment QUARANTINES at discovery (loud once, counted always, quarantinedSegments() exposed for the heal) and the field serves its remaining segments DEGRADED — never a raw throw killing every query on the field. Real storage faults still propagate untouched. - Torn generation artifacts (NaN/garbage in manifest or counter) DISCARD with narration at the store's open and recovery re-derives — plus a defensive finite-integer guard at the init consumer. Never a RangeError killing an open. THE LOUD TORN-RECORD CONTRACT: an existing-but-unparseable stored record now surfaces as a typed, counted TornRecordError on every entity-read surface (including fifteen previously-blind per-item batch catches); ENOENT stays clean-absent; artifact readers with designed absent-recovery keep null-tolerance behind the loud floor. Disk corruption can no longer read as silent data invisibility. Suite migration: the default's pins inverted deliberately, generation baselines made relative, quarantine-contract pins rewritten to the ruled behavior. Gates: tsc 0 · unit 2065/2065 (159 files) · integration 826 (93 files) · conformance 31/31 · kill-matrix 15/15 · torn-open guards 2/2.
This commit is contained in:
parent
67c606be69
commit
214c98b4d5
23 changed files with 833 additions and 154 deletions
11
RELEASES.md
11
RELEASES.md
|
|
@ -37,6 +37,17 @@ The theme: **writes ack fast and honestly, startup adopts instead of rebuilding,
|
|||
every query path serves, announces, or refuses — never silently degrades.** Ships as
|
||||
one release together with the matching native accelerator version.
|
||||
|
||||
**The storage-authority posture (the release's headline):** a NEW brain's default
|
||||
is **durable-at-ack log authority** — the generation log is the source of truth,
|
||||
every write acknowledgment is covered by a group-committed fsync, and crash
|
||||
recovery is a replay of the log (an acked write survives power loss, proven by
|
||||
fault-injection tests). An EXISTING brain adopts at its first open under 10.0.0,
|
||||
gated by a verification oracle: the log is replayed and diffed against stored
|
||||
truth record-by-record; curable gaps are backfilled; the brain flips only on a
|
||||
green verdict and a brain that cannot verify stays on the previous posture and
|
||||
says so loudly. The explicit opt-out is `logAuthority: 'defer'` in the config
|
||||
(no automatic adoption; flip later with `adoptLogAuthority()`).
|
||||
|
||||
**Why a major:** the generation log gains write format v2 — new segments carry typed,
|
||||
versioned records with integrity seals. A 9.x build refuses a v2 segment with a clear
|
||||
version-naming error (never a misread), which means **a brain written by 10.x cannot
|
||||
|
|
|
|||
|
|
@ -202,6 +202,7 @@ import {
|
|||
flipToLogAuthority,
|
||||
recordDigest,
|
||||
nounEntityTruth,
|
||||
LOG_AUTHORITY_PATH,
|
||||
type LogAuthorityRecord,
|
||||
type LogAuthorityStorage,
|
||||
type OracleReport
|
||||
|
|
@ -1371,7 +1372,20 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// gap for observability.
|
||||
for (const provider of this.versionedIndexProviders()) {
|
||||
const providerGen = provider.generation()
|
||||
const committed = BigInt(this.generationStore.committedGeneration())
|
||||
// Defensive finite-integer guard: committedGeneration() is validated
|
||||
// at the store's open (torn artifacts discard, narrated) — but a
|
||||
// RangeError here would kill the whole open, so the consumer guards
|
||||
// too. A non-finite value narrates and skips the gap check (the
|
||||
// provider's own replay contract still governs).
|
||||
const committedRaw = this.generationStore.committedGeneration()
|
||||
if (!Number.isSafeInteger(committedRaw) || committedRaw < 0) {
|
||||
prodLog.warn(
|
||||
`[Brainy] committed generation is non-integer (${String(committedRaw)}) at ` +
|
||||
`init — torn-artifact survivor; skipping the provider replay-gap check`
|
||||
)
|
||||
continue
|
||||
}
|
||||
const committed = BigInt(committedRaw)
|
||||
if (providerGen < committed) {
|
||||
prodLog.info(
|
||||
`[Brainy] Versioned index provider is at generation ${providerGen} ` +
|
||||
|
|
@ -1492,16 +1506,58 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
this._generationStampingActive = true
|
||||
}
|
||||
|
||||
// LOG-AUTHORITY SWITCH (checked at open only): a brain that has
|
||||
// flipped to log-authoritative storage gets durable-at-ack fact
|
||||
// writes (group-committed fsync covering every ack). Default 'tree'
|
||||
// = today's behavior, zero added latency.
|
||||
// LOG-AUTHORITY SWITCH (checked at open only). A STORED artifact
|
||||
// always wins: an already-flipped brain runs durable-at-ack; an
|
||||
// explicitly-recorded tree posture is honored. With NO artifact, the
|
||||
// 10.0.0 FLEET DEFAULT is ADOPT-AT-OPEN (config logAuthority:
|
||||
// 'adopt'): the verification oracle gates the flip — curable
|
||||
// divergences are baseline-backfilled, the brain flips ONLY on green,
|
||||
// and a brain that cannot go green STAYS tree-authoritative LOUDLY
|
||||
// with the refusal recorded (cheap subsequent opens; an operator
|
||||
// re-runs adoptLogAuthority() after fixing the divergence).
|
||||
// 'defer' is the documented opt-out: no automatic adoption.
|
||||
if (!this.isReadOnly) {
|
||||
const storedArtifact = await this.storage
|
||||
.readRawObject(LOG_AUTHORITY_PATH)
|
||||
.catch(() => null)
|
||||
const authority = await readLogAuthority(this.storage)
|
||||
this._logAuthority = authority
|
||||
if (authority.authority === 'log') {
|
||||
this.generationStore.setLogDurability('at-ack')
|
||||
prodLog.info('[Brainy] storage authority: generation log (durable-at-ack enabled)')
|
||||
} else if (
|
||||
storedArtifact === null &&
|
||||
this.config.logAuthority === 'adopt' &&
|
||||
this.generationStore.getFactLog() !== null
|
||||
) {
|
||||
try {
|
||||
await this.adoptLogAuthority()
|
||||
prodLog.info(
|
||||
'[Brainy] storage authority adopted at open: generation log ' +
|
||||
'(fleet default; oracle green; durable-at-ack enabled)'
|
||||
)
|
||||
} catch (err) {
|
||||
// The guarded ruling: a brain that cannot verify STAYS tree,
|
||||
// loudly, with the refusal recorded so subsequent opens are
|
||||
// cheap. Never a silent half-state; never a failed open.
|
||||
const reason = (err as Error).message
|
||||
prodLog.warn(
|
||||
`[Brainy] log-authority adoption REFUSED at open — this brain stays ` +
|
||||
`tree-authoritative until an operator resolves the divergence and ` +
|
||||
`re-runs adoptLogAuthority(). Reason: ${reason}`
|
||||
)
|
||||
try {
|
||||
const refusal: LogAuthorityRecord = {
|
||||
authority: 'tree',
|
||||
adoptRefusal: { at: Date.now(), reason: reason.slice(0, 500) }
|
||||
}
|
||||
await this.storage.writeRawObject(LOG_AUTHORITY_PATH, refusal)
|
||||
this._logAuthority = refusal
|
||||
} catch {
|
||||
// Unrecordable refusal = the next open retries the oracle —
|
||||
// the conservative outcome.
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -15786,7 +15842,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
force: config?.force ?? false,
|
||||
// Engine-owned persistence cadence — defaults resolve at the trigger
|
||||
// site (policy 'auto': 512 writes / 30s interval / 2s idle).
|
||||
persistence: config?.persistence
|
||||
persistence: config?.persistence,
|
||||
logAuthority: config?.logAuthority ?? 'adopt'
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -468,9 +468,26 @@ export class GenerationStore {
|
|||
| null
|
||||
const manifest = (await this.storage.readRawObject(MANIFEST_PATH)) as GenerationManifest | null
|
||||
|
||||
this.committed = manifest?.generation ?? 0
|
||||
this.horizonGen = manifest?.horizon ?? 0
|
||||
this.counter = Math.max(counterFile?.generation ?? 0, this.committed)
|
||||
// TORN-ARTIFACT VALIDATION (power-loss survivors): a torn manifest or
|
||||
// counter can carry NaN/garbage where a generation belongs — unguarded,
|
||||
// that NaN reaches BigInt() conversions at init and kills the open with
|
||||
// a RangeError. A non-finite-integer generation is DISCARDED with
|
||||
// narration (the conservative floor: 0 = re-derive from the record
|
||||
// directories / fact log below, exactly the recovery machinery's job).
|
||||
const finiteGen = (v: unknown, source: string): number => {
|
||||
if (typeof v === 'number' && Number.isSafeInteger(v) && v >= 0) return v
|
||||
if (v !== undefined && v !== null) {
|
||||
prodLog.warn(
|
||||
`[GenerationStore] ${source} carries a non-integer generation ` +
|
||||
`(${String(v)}) — torn write survivor; discarding and re-deriving ` +
|
||||
`from recovery (never a RangeError at open)`
|
||||
)
|
||||
}
|
||||
return 0
|
||||
}
|
||||
this.committed = finiteGen(manifest?.generation, 'manifest')
|
||||
this.horizonGen = finiteGen(manifest?.horizon, 'manifest horizon')
|
||||
this.counter = Math.max(finiteGen(counterFile?.generation, 'generation counter'), this.committed)
|
||||
|
||||
// Discover existing generation record directories.
|
||||
const recordPaths = await this.storage.listRawObjects(GENERATIONS_PREFIX)
|
||||
|
|
|
|||
|
|
@ -43,6 +43,12 @@ export interface LogAuthorityRecord {
|
|||
nounsChecked: number
|
||||
verbsChecked: number
|
||||
}
|
||||
/**
|
||||
* Recorded when an OPEN-TIME adoption attempt (the 10.0.0 fleet default)
|
||||
* was refused — the oracle could not go green. Keeps subsequent opens
|
||||
* cheap; an operator re-runs adoptLogAuthority() after resolving it.
|
||||
*/
|
||||
adoptRefusal?: { at: number; reason: string }
|
||||
}
|
||||
|
||||
/** The narrow storage surface this module needs. */
|
||||
|
|
|
|||
|
|
@ -362,6 +362,15 @@ export { MemoryStorage, createStorage }
|
|||
// FileSystemStorage is exported separately to avoid browser build issues.
|
||||
export { FileSystemStorage } from './storage/adapters/fileSystemStorage.js'
|
||||
|
||||
// Torn-record surface: a stored file that EXISTS but cannot be decoded throws
|
||||
// a typed, catchable error on entity reads (never a silent "not found"), and
|
||||
// every encounter is counted on a per-process gauge.
|
||||
export {
|
||||
TornRecordError,
|
||||
isTornRecordError,
|
||||
getTornRecordGauge
|
||||
} from './storage/tornRecordError.js'
|
||||
|
||||
// Export types
|
||||
import type {
|
||||
Vector,
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ import { ColumnSegmentCursor, TailBufferCursor, type CursorEntry } from './Colum
|
|||
import { writeSegmentToBuffer, readSegmentFromBuffer } from './ColumnSegmentFormat.js'
|
||||
import { RoaringBitmap32 } from '../../utils/roaring/index.js'
|
||||
import { compareCodePoints } from '../../utils/collation.js'
|
||||
import { prodLog } from '../../utils/logger.js'
|
||||
|
||||
/**
|
||||
* Configuration for the ColumnStore.
|
||||
|
|
@ -612,6 +613,24 @@ export class ColumnStore implements ColumnStoreProvider {
|
|||
/**
|
||||
* Get all segment cursors for a field, loading from storage if needed.
|
||||
*/
|
||||
/**
|
||||
* Per-field quarantine ledger for torn segments (power-loss survivors:
|
||||
* manifest-listed but unloadable). A quarantined segment is skipped with
|
||||
* per-doubling narration and the field serves its REMAINING segments as a
|
||||
* DEGRADED-ANNOUNCED result — never a raw throw killing the query, never
|
||||
* a silent drop. Cleared when a heal/rebuild rewrites the field.
|
||||
*/
|
||||
private readonly segmentQuarantine = new Map<string, { error: string; hits: number }>()
|
||||
|
||||
/** Torn-segment quarantine entries for a field (observability + heal input). */
|
||||
quarantinedSegments(field: string): Array<{ segment: string; error: string; hits: number }> {
|
||||
const out: Array<{ segment: string; error: string; hits: number }> = []
|
||||
for (const [key, q] of this.segmentQuarantine) {
|
||||
if (key.startsWith(`${field}:`)) out.push({ segment: key.slice(field.length + 1), error: q.error, hits: q.hits })
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
private async getSegmentCursors(field: string): Promise<ColumnSegmentCursor[]> {
|
||||
const manifest = this.manifests.get(field)
|
||||
if (!manifest) return []
|
||||
|
|
@ -622,11 +641,38 @@ export class ColumnStore implements ColumnStoreProvider {
|
|||
let cursor = this.segmentCache.get(cacheKey)
|
||||
|
||||
if (!cursor) {
|
||||
// loadSegmentCursor either returns a cursor or THROWS — a corrupt /
|
||||
// missing manifest-listed segment raises ColumnSegmentLoadError and a
|
||||
// real storage fault propagates, so a listed segment is never silently
|
||||
// dropped from the result set.
|
||||
const quarantined = this.segmentQuarantine.get(cacheKey)
|
||||
if (quarantined) {
|
||||
// Already-quarantined torn segment: skip, count, narrate per doubling.
|
||||
quarantined.hits++
|
||||
if ((quarantined.hits & (quarantined.hits - 1)) === 0) {
|
||||
prodLog.warn(
|
||||
`[ColumnStore] field '${field}' serving DEGRADED: torn segment ${seg.id} ` +
|
||||
`quarantined (${quarantined.error}) — ${quarantined.hits} queries served ` +
|
||||
`without it; heal/rebuild the metadata index to restore`
|
||||
)
|
||||
}
|
||||
continue
|
||||
}
|
||||
try {
|
||||
cursor = await this.loadSegmentCursor(field, seg)
|
||||
} catch (err) {
|
||||
if (err instanceof ColumnSegmentLoadError) {
|
||||
// POWER-LOSS SURVIVOR: a manifest-listed segment whose bytes are
|
||||
// torn/absent. Quarantine at DISCOVERY and serve the remaining
|
||||
// segments degraded-announced — a raw throw here killed every
|
||||
// query on the field forever; a silent skip hid the loss. The
|
||||
// quarantine is the middle: loud once, counted always, healable.
|
||||
this.segmentQuarantine.set(cacheKey, { error: (err as Error).message, hits: 1 })
|
||||
prodLog.error(
|
||||
`[ColumnStore] torn segment QUARANTINED at discovery: field '${field}' ` +
|
||||
`segment ${seg.id} — ${(err as Error).message}. The field serves its ` +
|
||||
`remaining segments DEGRADED until a heal/rebuild rewrites it.`
|
||||
)
|
||||
continue
|
||||
}
|
||||
throw err // real storage faults propagate — never absorbed
|
||||
}
|
||||
this.segmentCache.set(cacheKey, cursor)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -18,6 +18,11 @@ import {
|
|||
} from '../baseStorage.js'
|
||||
import { getBrainyVersion } from '../../utils/index.js'
|
||||
import { isAbsentError } from '../../utils/errorClassification.js'
|
||||
import {
|
||||
TornRecordError,
|
||||
isUnparseablePayloadError,
|
||||
registerTornRecordEncounter
|
||||
} from '../tornRecordError.js'
|
||||
|
||||
// Node.js modules - dynamically imported to avoid issues in browser environments
|
||||
let fs: any
|
||||
|
|
@ -410,8 +415,22 @@ export class FileSystemStorage extends BaseStorage {
|
|||
/**
|
||||
* Primitive operation: Read object from path
|
||||
* All metadata operations use this internally via base class routing
|
||||
* Enhanced error handling for corrupted metadata files (Bug #3 mitigation)
|
||||
* Supports reading both compressed (.gz) and uncompressed files for backward compatibility
|
||||
*
|
||||
* Read contract (loud errors, never quiet losses):
|
||||
* - Genuine absence (ENOENT on every variant) → `null`. Only a missing file
|
||||
* is "not found".
|
||||
* - TORN record (a file EXISTS but its bytes cannot be decoded — invalid
|
||||
* JSON, truncated/garbled gzip) → the encounter is registered (production
|
||||
* ERROR log + per-process gauge) and a typed {@link TornRecordError} is
|
||||
* thrown. Corruption must NEVER read as absence: callers that can degrade
|
||||
* (manifest recovery, rebuildable statistics) catch the typed error at
|
||||
* their sites; entity reads surface it.
|
||||
* Legacy dual-format exception: when the `.gz` variant is torn but the
|
||||
* uncompressed fallback decodes, the recovered object is returned — AFTER
|
||||
* the torn `.gz` was logged and counted (loud recovery, not a silent skip).
|
||||
* - Real storage fault (EIO/EACCES/EMFILE/…) → propagates as itself; a
|
||||
* fault is neither absence nor corruption and must not be reshaped.
|
||||
*/
|
||||
protected async readObjectFromPath(pathStr: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
|
|
@ -419,7 +438,10 @@ export class FileSystemStorage extends BaseStorage {
|
|||
const fullPath = path.join(this.rootDir, pathStr)
|
||||
const compressedPath = `${fullPath}.gz`
|
||||
|
||||
// Try reading compressed file first (if compression is enabled or file exists)
|
||||
// Try reading compressed file first (if compression is enabled or file exists).
|
||||
// A torn .gz is remembered so the uncompressed fallback can either recover
|
||||
// (legacy dual-format installs) or surface the corruption typed.
|
||||
let tornCompressed: TornRecordError | null = null
|
||||
try {
|
||||
const compressedData = await fs.promises.readFile(compressedPath)
|
||||
const decompressed = await new Promise<Buffer>((resolve, reject) => {
|
||||
|
|
@ -430,9 +452,16 @@ export class FileSystemStorage extends BaseStorage {
|
|||
})
|
||||
return JSON.parse(decompressed.toString('utf-8'))
|
||||
} catch (error: any) {
|
||||
// If compressed file doesn't exist, fall back to uncompressed
|
||||
if (error.code !== 'ENOENT') {
|
||||
console.warn(`Failed to read compressed file ${compressedPath}:`, error)
|
||||
if (error.code === 'ENOENT') {
|
||||
// No compressed variant — fall through to the uncompressed path.
|
||||
} else if (isUnparseablePayloadError(error)) {
|
||||
// The .gz EXISTS but cannot be decoded (zlib Z_* error or JSON
|
||||
// SyntaxError after gunzip): torn record. Register NOW (log + gauge),
|
||||
// then attempt the uncompressed fallback as a recovery read.
|
||||
tornCompressed = registerTornRecordEncounter(`${pathStr}.gz`, error)
|
||||
} else {
|
||||
// Real storage fault on an existing .gz (EIO/EACCES/…): propagate.
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -442,24 +471,26 @@ export class FileSystemStorage extends BaseStorage {
|
|||
return JSON.parse(data)
|
||||
} catch (error: any) {
|
||||
if (error.code === 'ENOENT') {
|
||||
// No uncompressed file. If the .gz variant existed but was torn, the
|
||||
// object EXISTS and is unreadable — that must surface typed, never as
|
||||
// "absent". Otherwise this is genuine absence.
|
||||
if (tornCompressed !== null) {
|
||||
throw tornCompressed
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
// Enhanced error handling for corrupted JSON files (race condition from Bug #3)
|
||||
if (error instanceof SyntaxError || error.name === 'SyntaxError') {
|
||||
console.warn(
|
||||
`⚠️ Corrupted metadata file detected: ${pathStr}\n` +
|
||||
` This may be caused by concurrent writes during import.\n` +
|
||||
` Gracefully skipping this entry. File may be repaired on next write.`
|
||||
)
|
||||
return null
|
||||
// The file EXISTS but its content cannot be parsed: torn record.
|
||||
// Register (production ERROR + gauge) and throw typed — a corrupt row
|
||||
// must be distinguishable from a missing row, or nothing ever heals it.
|
||||
if (isUnparseablePayloadError(error)) {
|
||||
throw registerTornRecordEncounter(pathStr, error)
|
||||
}
|
||||
|
||||
// A real storage fault (EIO/EACCES/EMFILE/…) is NOT "object absent". The
|
||||
// ENOENT branch (above) already returns null, and the corrupted-JSON
|
||||
// branch (above) is a deliberate concurrent-write tolerance; a genuine
|
||||
// fault reaching here must propagate loudly rather than masquerade as a
|
||||
// missing object — which would corrupt reads and drive needless rebuilds.
|
||||
// ENOENT branch (above) already returns null; a genuine fault reaching
|
||||
// here must propagate loudly rather than masquerade as a missing object
|
||||
// — which would corrupt reads and drive needless rebuilds.
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ import { BlobStorage, type BlobStoreAdapter } from './blobStorage.js'
|
|||
import { unwrapBinaryData } from './binaryDataCodec.js'
|
||||
import { prodLog } from '../utils/logger.js'
|
||||
import { isAbsentError } from '../utils/errorClassification.js'
|
||||
import { isTornRecordError } from './tornRecordError.js'
|
||||
import { BrainyError, ProtectedArtifactError, DerivedArtifactMissingError } from '../errors/brainyError.js'
|
||||
import { MetadataWriteBuffer } from '../utils/metadataWriteBuffer.js'
|
||||
import {
|
||||
|
|
@ -674,6 +675,10 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
// — hash verification must run on the original content bytes.
|
||||
return unwrapBinaryData(data)
|
||||
} catch (error) {
|
||||
// A TORN blob object (exists but undecodable) must not read as
|
||||
// "blob absent" — that would misdiagnose disk corruption as a
|
||||
// missing blob. Propagate the typed error to the blob layer.
|
||||
if (isTornRecordError(error)) throw error
|
||||
return undefined
|
||||
}
|
||||
},
|
||||
|
|
@ -768,6 +773,20 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
if (m) hashes.add(m[1])
|
||||
}
|
||||
|
||||
// Recovery-path read: a TORN object here maps to "not usable" (null) BY
|
||||
// DESIGN — the adapter has already logged + counted the encounter, and
|
||||
// treating a torn `_cas/` copy as absent lets the re-copy from `_cow/`
|
||||
// OVERWRITE the corrupt file with the good original (the heal), while a
|
||||
// torn `_cow/` original is reported via `incomplete`. Real faults propagate.
|
||||
const readOrNullIfTorn = async (p: string): Promise<any | null> => {
|
||||
try {
|
||||
return await this.readObjectFromPath(p)
|
||||
} catch (error) {
|
||||
if (isTornRecordError(error)) return null
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
let adopted = 0
|
||||
let alreadyPresent = 0
|
||||
let incomplete = 0
|
||||
|
|
@ -775,15 +794,15 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
// A blob counts as present only when BOTH its bytes and its metadata
|
||||
// already live in `_cas/`. A half-adopted blob (bytes without meta — the
|
||||
// exact "Blob metadata not found" state) is re-adopted.
|
||||
const casBlob = await this.readObjectFromPath(`_cas/blob:${hash}`)
|
||||
const casMeta = await this.readObjectFromPath(`_cas/blob-meta:${hash}`)
|
||||
const casBlob = await readOrNullIfTorn(`_cas/blob:${hash}`)
|
||||
const casMeta = await readOrNullIfTorn(`_cas/blob-meta:${hash}`)
|
||||
if (casBlob !== null && casMeta !== null) {
|
||||
alreadyPresent++
|
||||
continue
|
||||
}
|
||||
|
||||
const cowBlob = await this.readObjectFromPath(`_cow/blob:${hash}`)
|
||||
const cowMeta = await this.readObjectFromPath(`_cow/blob-meta:${hash}`)
|
||||
const cowBlob = await readOrNullIfTorn(`_cow/blob:${hash}`)
|
||||
const cowMeta = await readOrNullIfTorn(`_cow/blob-meta:${hash}`)
|
||||
if (cowBlob === null || cowMeta === null) {
|
||||
// Can't register a blob the store can't fully describe — report it so an
|
||||
// operator investigates rather than silently half-adopting.
|
||||
|
|
@ -1134,12 +1153,28 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
* cache (record-layer files are written through
|
||||
* {@link BaseStorage.writeRawObject} only).
|
||||
*
|
||||
* TORN-record contract (deliberate, loud-by-design): this surface serves
|
||||
* SYSTEM ARTIFACTS — manifests with recovery paths, markers whose verdict
|
||||
* machinery treats "unreadable" as rescan, generation/transaction records
|
||||
* whose recovery is built for absent artifacts. For these readers a torn
|
||||
* file maps to their existing absent-artifact degrade, so a typed
|
||||
* torn-record error from the adapter is caught here and returned as `null`
|
||||
* — AFTER the adapter has already logged a production ERROR and counted
|
||||
* the per-process torn-record gauge (never silent). Entity reads do NOT go
|
||||
* through this surface; they use the canonical read paths, which propagate
|
||||
* the typed error. Real storage faults (EIO/EACCES/…) still propagate.
|
||||
*
|
||||
* @param path - Storage-root-relative object path (e.g. `_system/manifest.json`).
|
||||
* @returns The parsed object, or `null` if absent.
|
||||
* @returns The parsed object, or `null` if absent (or torn — logged + counted).
|
||||
*/
|
||||
public async readRawObject(path: string): Promise<any | null> {
|
||||
await this.ensureInitialized()
|
||||
return this.readObjectFromPath(path)
|
||||
try {
|
||||
return await this.readObjectFromPath(path)
|
||||
} catch (error) {
|
||||
if (isTornRecordError(error)) return null
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -2146,6 +2181,9 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
if (!metadata) return null
|
||||
return { deserialized, metadata }
|
||||
} catch (error) {
|
||||
// A TORN record must surface typed — a paginated read that
|
||||
// silently skips a corrupt row hides data loss from the caller.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip nouns that fail to load
|
||||
return null
|
||||
}
|
||||
|
|
@ -2175,6 +2213,8 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// A TORN record propagates (typed) — only shard-listing absence is skippable.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip shards that have no data
|
||||
}
|
||||
}
|
||||
|
|
@ -2283,7 +2323,9 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
batch.map(async (id) => {
|
||||
try {
|
||||
return { id, metadata: await this.getNounMetadata(id) }
|
||||
} catch {
|
||||
} catch (error) {
|
||||
// A TORN record must surface typed, never as a skipped id.
|
||||
if (isTornRecordError(error)) throw error
|
||||
return null
|
||||
}
|
||||
})
|
||||
|
|
@ -2305,6 +2347,8 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// A TORN record propagates (typed) — only shard-listing absence is skippable.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip shards with no data
|
||||
}
|
||||
}
|
||||
|
|
@ -2515,10 +2559,15 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
// reserved fields top-level, ONLY custom fields in `metadata`.
|
||||
collected.push({ verb: this.hydrateVerbWithMetadata(verb, metadata), shard })
|
||||
} catch (error) {
|
||||
// A TORN record must surface typed — a paginated read that
|
||||
// silently skips a corrupt row hides data loss from the caller.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip verbs that fail to load
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// A TORN record propagates (typed) — only shard-listing absence is skippable.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip shards that have no data
|
||||
}
|
||||
}
|
||||
|
|
@ -3669,9 +3718,18 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
)
|
||||
|
||||
for (const result of chunkResults) {
|
||||
if (result.status === 'fulfilled' && result.value.data !== null) {
|
||||
if (result.status === 'fulfilled') {
|
||||
if (result.value.data !== null) {
|
||||
results.set(result.value.path, result.value.data)
|
||||
}
|
||||
} else {
|
||||
// A rejected read is a torn record or a real storage fault — NOT an
|
||||
// absent object. Batch hydration backs entity reads (getNounBatch /
|
||||
// getVerbsBatch / find hydration); swallowing the rejection would
|
||||
// silently drop a row the caller cannot distinguish from "never
|
||||
// existed". Propagate the typed/real error loudly instead.
|
||||
throw result.reason
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -4636,10 +4694,15 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// A TORN record must surface typed — an enumeration that silently
|
||||
// skips a corrupt row hides data loss from the caller.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip nouns that fail to load
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// A TORN record propagates (typed) — only shard-listing absence is skippable.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip shards that have no data
|
||||
}
|
||||
}
|
||||
|
|
@ -4825,11 +4888,16 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
results.push(this.hydrateVerbWithMetadata(verb, metadata))
|
||||
}
|
||||
} catch (error) {
|
||||
// A TORN record must surface typed — an enumeration that silently
|
||||
// skips a corrupt row hides data loss from the caller.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip verbs that fail to load
|
||||
prodLog.debug(`[BaseStorage] Failed to load verb from ${verbPath}:`, error)
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// A TORN record propagates (typed) — only shard-listing absence is skippable.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip shards that have no data
|
||||
}
|
||||
}
|
||||
|
|
@ -4945,6 +5013,9 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
sourceVerbs.push(hydratedVerb)
|
||||
}
|
||||
} catch (error) {
|
||||
// A TORN record propagates (typed) — batch hydration must not
|
||||
// silently drop a corrupt row. Only shard-listing absence is skippable.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip shards that have no data
|
||||
}
|
||||
}
|
||||
|
|
@ -5030,10 +5101,15 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
results.push(this.hydrateVerbWithMetadata(verb, metadata))
|
||||
}
|
||||
} catch (error) {
|
||||
// A TORN record must surface typed — an enumeration that silently
|
||||
// skips a corrupt row hides data loss from the caller.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip verbs that fail to load
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// A TORN record propagates (typed) — only shard-listing absence is skippable.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip shards that have no data
|
||||
}
|
||||
}
|
||||
|
|
@ -5078,10 +5154,15 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
)
|
||||
)
|
||||
} catch (error) {
|
||||
// A TORN record must surface typed — an enumeration that silently
|
||||
// skips a corrupt row hides data loss from the caller.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip verbs that fail to load
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
// A TORN record propagates (typed) — only shard-listing absence is skippable.
|
||||
if (isTornRecordError(error)) throw error
|
||||
// Skip shards that have no data
|
||||
}
|
||||
}
|
||||
|
|
|
|||
132
src/storage/tornRecordError.ts
Normal file
132
src/storage/tornRecordError.ts
Normal file
|
|
@ -0,0 +1,132 @@
|
|||
/**
|
||||
* @module storage/tornRecordError
|
||||
* @description Typed surface for TORN records — files that EXIST in storage but
|
||||
* cannot be decoded (invalid JSON, truncated/garbled gzip). A torn record is
|
||||
* disk corruption, not absence: reading it as `null` ("not found") makes the
|
||||
* consumer unable to distinguish "never existed" from "exists but unreadable",
|
||||
* so nothing ever heals it. Mandate: loud errors, never quiet losses.
|
||||
*
|
||||
* Contract implemented across the storage layer:
|
||||
* - Genuine absence (ENOENT) still reads as clean `null` — no error, no noise.
|
||||
* - A torn record ALWAYS registers here (error log + per-process gauge), then:
|
||||
* - entity read paths (get/getBatch/pagination/enumeration hydration) throw
|
||||
* {@link TornRecordError} to the caller — a row is never silently dropped;
|
||||
* - system-artifact read paths whose machinery is designed for
|
||||
* absent-artifact degradation (manifests with recovery paths, markers
|
||||
* whose verdict is "rescan", rebuildable statistics) map torn → their
|
||||
* existing degrade AFTER the encounter is logged and counted.
|
||||
*/
|
||||
|
||||
import { prodLog } from '../utils/logger.js'
|
||||
|
||||
/**
|
||||
* @description Thrown when a stored object EXISTS but cannot be decoded —
|
||||
* corrupt/torn bytes on disk (invalid JSON, undecodable gzip). Deliberately
|
||||
* distinct from absence: `readObjectFromPath` returns `null` only for ENOENT.
|
||||
* Catchable by type (`instanceof`), by `name === 'TornRecordError'`, or by
|
||||
* `code === 'TORN_RECORD'` (cross-realm safe; never matches `isAbsentError`).
|
||||
*/
|
||||
export class TornRecordError extends Error {
|
||||
/** Stable machine-checkable discriminator (errno-style). */
|
||||
public readonly code = 'TORN_RECORD'
|
||||
/** Storage-root-relative path of the torn object. */
|
||||
public readonly path: string
|
||||
/** The underlying decode failure (SyntaxError, zlib error, …). */
|
||||
public override readonly cause: unknown
|
||||
|
||||
/**
|
||||
* @param path - Storage-root-relative path of the torn object.
|
||||
* @param cause - The underlying decode failure.
|
||||
*/
|
||||
constructor(path: string, cause: unknown) {
|
||||
const causeMessage =
|
||||
cause instanceof Error ? cause.message : String(cause)
|
||||
super(
|
||||
`Torn record at '${path}': file exists but cannot be decoded (${causeMessage}). ` +
|
||||
`This is storage corruption, not absence — the record was not silently skipped.`
|
||||
)
|
||||
this.name = 'TornRecordError'
|
||||
this.path = path
|
||||
this.cause = cause
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @description True IFF `e` is a torn-record error — matches by `instanceof`
|
||||
* first, then by `name`/`code` so errors crossing module-duplication or realm
|
||||
* boundaries are still recognized.
|
||||
* @param e - The caught value.
|
||||
* @returns Whether `e` denotes an existing-but-undecodable stored object.
|
||||
*/
|
||||
export function isTornRecordError(e: unknown): e is TornRecordError {
|
||||
if (e instanceof TornRecordError) return true
|
||||
if (e === null || typeof e !== 'object') return false
|
||||
const { name, code } = e as { name?: unknown; code?: unknown }
|
||||
return name === 'TornRecordError' || code === 'TORN_RECORD'
|
||||
}
|
||||
|
||||
/**
|
||||
* @description True IFF `e` is a payload-decode failure — the file's BYTES were
|
||||
* read fine but could not be turned back into an object: `SyntaxError` from
|
||||
* `JSON.parse`, or a zlib error (`Z_DATA_ERROR`, `Z_BUF_ERROR`, …) from gunzip.
|
||||
* Distinguishes "torn record" from real I/O faults (EIO/EACCES/…), which must
|
||||
* propagate as themselves.
|
||||
* @param e - The caught value.
|
||||
* @returns Whether the error means "bytes present, content undecodable".
|
||||
*/
|
||||
export function isUnparseablePayloadError(e: unknown): boolean {
|
||||
if (e === null || typeof e !== 'object') return false
|
||||
if (e instanceof SyntaxError) return true
|
||||
const { name, code } = e as { name?: unknown; code?: unknown }
|
||||
if (name === 'SyntaxError') return true
|
||||
return typeof code === 'string' && code.startsWith('Z_')
|
||||
}
|
||||
|
||||
/** Per-process torn-record gauge state (module-scoped; see the accessors). */
|
||||
let tornRecordCount = 0
|
||||
let lastTornRecordPath: string | null = null
|
||||
|
||||
/**
|
||||
* @description Register a torn-record encounter: logs a production ERROR
|
||||
* naming the path, increments the per-process gauge, and returns the typed
|
||||
* error for the caller to throw (or to map into a documented loud degrade).
|
||||
* EVERY torn encounter goes through here, whatever the caller decides —
|
||||
* the floor is: never silent.
|
||||
* @param path - Storage-root-relative path of the torn object.
|
||||
* @param cause - The underlying decode failure.
|
||||
* @returns The constructed {@link TornRecordError}.
|
||||
*/
|
||||
export function registerTornRecordEncounter(
|
||||
path: string,
|
||||
cause: unknown
|
||||
): TornRecordError {
|
||||
tornRecordCount++
|
||||
lastTornRecordPath = path
|
||||
const error = new TornRecordError(path, cause)
|
||||
prodLog.error(
|
||||
`[Storage] TORN RECORD #${tornRecordCount}: '${path}' exists but cannot be decoded — ` +
|
||||
`corrupt or partially written bytes. Cause: ${
|
||||
cause instanceof Error ? `${cause.name}: ${cause.message}` : String(cause)
|
||||
}`
|
||||
)
|
||||
return error
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Read the per-process torn-record gauge: how many torn records
|
||||
* this process has encountered and the most recent path. Observability seam —
|
||||
* lets operators and tests confirm that corruption was seen, not swallowed.
|
||||
* @returns The current gauge snapshot.
|
||||
*/
|
||||
export function getTornRecordGauge(): { count: number; lastPath: string | null } {
|
||||
return { count: tornRecordCount, lastPath: lastTornRecordPath }
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Reset the per-process torn-record gauge to zero. Test seam only
|
||||
* (the gauge is process-lifetime state); production code never resets it.
|
||||
*/
|
||||
export function resetTornRecordGauge(): void {
|
||||
tornRecordCount = 0
|
||||
lastTornRecordPath = null
|
||||
}
|
||||
|
|
@ -2084,6 +2084,28 @@ export interface BrainyConfig {
|
|||
* `'manual'` restores the pre-9.1 behavior: the engine never flushes on
|
||||
* its own (except at `close()`); the caller owns the cadence.
|
||||
*/
|
||||
/**
|
||||
* Storage-authority posture at open (10.0.0+ fleet default: `'adopt'`).
|
||||
*
|
||||
* `'adopt'` — a brain with NO stored authority artifact adopts LOG
|
||||
* AUTHORITY at open, oracle-gated: the verification oracle replays the
|
||||
* generation log against stored truth; curable divergences (pre-log
|
||||
* rows, witness drift) are baseline-backfilled; the brain flips ONLY on
|
||||
* a green verdict and writes the durable per-brain switch. On green,
|
||||
* writes become durable-at-ack (group-committed log fsync covers every
|
||||
* ack). A brain whose oracle cannot go green STAYS tree-authoritative,
|
||||
* says so loudly, and records the refusal — never a silent half-state.
|
||||
*
|
||||
* `'defer'` — the explicit opt-out: no automatic adoption; the brain
|
||||
* stays tree-authoritative until `adoptLogAuthority()` is called. The
|
||||
* pre-10 behavior, documented for operators who stage their own flips.
|
||||
*
|
||||
* A STORED artifact always wins over this setting (checked-at-open law):
|
||||
* an already-flipped brain stays flipped; an explicitly-recorded tree
|
||||
* posture is honored until an operator re-runs adoption.
|
||||
*/
|
||||
logAuthority?: 'adopt' | 'defer'
|
||||
|
||||
persistence?: {
|
||||
policy?: 'auto' | 'manual'
|
||||
/** Background flush after this many committed writes (default 512). */
|
||||
|
|
|
|||
|
|
@ -60,15 +60,25 @@ export function makeTempDir(): string {
|
|||
* Open a writer brain over `dir` with every implicit durability knob off:
|
||||
* persistence policy 'manual' (the engine never flushes on its own, so every
|
||||
* durable transition in a test is an explicit `flush()`/commit), deterministic
|
||||
* embeddings (tests always pass explicit vectors anyway), silent logs.
|
||||
* embeddings (tests always pass explicit vectors anyway), silent logs — and
|
||||
* `logAuthority: 'defer'` (the explicit opt-out of the 10.0.0 adopt-at-open
|
||||
* fleet default), so the durability POSTURE is explicit per row too: rows
|
||||
* pinning deferred/tree recovery semantics get exactly that, and at-ack rows
|
||||
* engage log authority via `flipToAtAck`. The fleet default's open-time
|
||||
* adoption would inject a baseline-backfill generation into every floor
|
||||
* computation and pre-flip every row.
|
||||
*/
|
||||
export async function openBrain(dir: string): Promise<Brainy> {
|
||||
export async function openBrain(
|
||||
dir: string,
|
||||
opts?: { logAuthority?: 'adopt' | 'defer' }
|
||||
): Promise<Brainy> {
|
||||
process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
|
||||
const brain = new Brainy({
|
||||
requireSubtype: false,
|
||||
storage: { type: 'filesystem', path: dir },
|
||||
silent: true,
|
||||
persistence: { policy: 'manual' }
|
||||
persistence: { policy: 'manual' },
|
||||
logAuthority: opts?.logAuthority ?? 'defer'
|
||||
})
|
||||
await brain.init()
|
||||
return brain
|
||||
|
|
|
|||
|
|
@ -96,11 +96,15 @@ describe('8.0 Db API — generational MVCC', () => {
|
|||
}
|
||||
|
||||
/** Open (and track) a filesystem brain rooted at a fresh temp directory. */
|
||||
async function openFsBrain(dir?: string): Promise<{ brain: Brainy; dir: string }> {
|
||||
async function openFsBrain(
|
||||
dir?: string,
|
||||
logAuthority?: 'adopt' | 'defer'
|
||||
): Promise<{ brain: Brainy; dir: string }> {
|
||||
const rootDirectory = dir ?? makeTempDir()
|
||||
const brain = new Brainy({
|
||||
requireSubtype: false,
|
||||
storage: { type: 'filesystem', path: rootDirectory }
|
||||
storage: { type: 'filesystem', path: rootDirectory },
|
||||
...(logAuthority ? { logAuthority } : {})
|
||||
})
|
||||
await brain.init()
|
||||
brains.push(brain)
|
||||
|
|
@ -647,7 +651,13 @@ describe('8.0 Db API — generational MVCC', () => {
|
|||
// ==========================================================================
|
||||
it('proof 8 — a crash before the manifest rename recovers to the exact pre-transaction state', async () => {
|
||||
const dir = makeTempDir()
|
||||
const { brain: first } = await openFsBrain(dir)
|
||||
// 'defer' (tree authority): this proof pins the TREE commit-point
|
||||
// contract — the manifest rename is the commit, so a crash before it
|
||||
// rolls back. Under the adopt-at-open default (log authority) the same
|
||||
// crash point legitimately REPLAYS the fsynced fact at reopen and the
|
||||
// transaction lands — that contract is pinned in the durability kill
|
||||
// matrix's at-ack rows, not here.
|
||||
const { brain: first } = await openFsBrain(dir, 'defer')
|
||||
|
||||
await first.transact([
|
||||
{
|
||||
|
|
@ -689,9 +699,10 @@ describe('8.0 Db API — generational MVCC', () => {
|
|||
// the realistic worst case for the recovery path.
|
||||
await first.close()
|
||||
|
||||
// Reopen: recovery rolls the uncommitted generation back and rebuilds
|
||||
// the indexes from the repaired records.
|
||||
const { brain: second } = await openFsBrain(dir)
|
||||
// Reopen ('defer' again — a reopen under the adopt default would adopt
|
||||
// and change the recovery path): recovery rolls the uncommitted
|
||||
// generation back and rebuilds the indexes from the repaired records.
|
||||
const { brain: second } = await openFsBrain(dir, 'defer')
|
||||
const recovered = await second.get(uid('crash-e'))
|
||||
expect((recovered?.metadata as { v: number }).v).toBe(1)
|
||||
expect(await second.get(uid('crash-new'))).toBeNull()
|
||||
|
|
@ -1162,13 +1173,16 @@ describe('8.0 Db API — generational MVCC', () => {
|
|||
const brain = await openMemoryBrain()
|
||||
|
||||
// Model-B: a single-op write is its OWN generation and IS logged (no meta —
|
||||
// tx metadata is a transact()-only concept). It is generation 1 on a fresh
|
||||
// brain (init-time infrastructure writes are the un-versioned gen-0 baseline).
|
||||
// tx metadata is a transact()-only concept). Relative baseline: under the
|
||||
// adopt-at-open fleet default the open-time baseline backfill is itself a
|
||||
// logged single-op generation, so the log is not empty on a fresh brain —
|
||||
// every pin below is expressed against that baseline.
|
||||
const baseGens = (await brain.transactionLog()).map((entry) => entry.generation)
|
||||
await brain.add({ id: uid('txlog-solo'), type: NounType.Document, data: 'solo', vector: vec(99), subtype: 'note' })
|
||||
const soloLog = await brain.transactionLog()
|
||||
expect(soloLog.map((entry) => entry.generation)).toEqual([1])
|
||||
const soloGen = brain.generation()
|
||||
expect(soloLog.map((entry) => entry.generation)).toEqual([soloGen, ...baseGens])
|
||||
expect(soloLog[0].meta).toBeUndefined()
|
||||
const soloGen = 1
|
||||
|
||||
const first = await brain.transact(
|
||||
[{ op: 'add', id: uid('txlog-a'), type: NounType.Document, data: 'a', vector: vec(100), metadata: {} }],
|
||||
|
|
@ -1181,12 +1195,14 @@ describe('8.0 Db API — generational MVCC', () => {
|
|||
const third = await brain.transact([{ op: 'update', id: uid('txlog-a'), metadata: { v: 3 } }])
|
||||
|
||||
const entries = await brain.transactionLog()
|
||||
// Newest first: the three transacts, then the single-op solo write (gen 1).
|
||||
// Newest first: the three transacts, then the single-op solo write, then
|
||||
// whatever the open baseline logged (the adopt-at-open backfill).
|
||||
expect(entries.map((entry) => entry.generation)).toEqual([
|
||||
third.generation,
|
||||
second.generation,
|
||||
first.generation,
|
||||
soloGen
|
||||
soloGen,
|
||||
...baseGens
|
||||
])
|
||||
expect(entries[1].meta).toEqual({ author: 'job-2' })
|
||||
expect(entries[2].meta).toEqual({ author: 'job-1' })
|
||||
|
|
@ -1238,21 +1254,24 @@ describe('8.0 Db API — generational MVCC', () => {
|
|||
const brain = await openMemoryBrain()
|
||||
const a = uid('ov-a')
|
||||
const b = uid('ov-b')
|
||||
await (
|
||||
await brain.transact([
|
||||
// Pin RELATIVELY at the transact's own generation (not an absolute 1 —
|
||||
// the adopt-at-open baseline backfill owns the first generation).
|
||||
const tx = await brain.transact([
|
||||
{ op: 'add', id: a, type: NounType.Document, data: 'a', vector: vec(1), metadata: { v: 1 } },
|
||||
{ op: 'add', id: b, type: NounType.Document, data: 'b', vector: vec(2), metadata: { v: 1 } }
|
||||
])
|
||||
).release()
|
||||
const at1 = await brain.asOf(1)
|
||||
const txGen = tx.generation
|
||||
await tx.release()
|
||||
const at1 = await brain.asOf(txGen)
|
||||
|
||||
// A single-op REMOVE of `b` lands AFTER the pin and is NOT flushed (pending).
|
||||
await brain.remove(b)
|
||||
|
||||
const liveIds = (await brain.find({})).map((r) => r.id)
|
||||
const pastIds = (await at1.find({})).map((r) => r.id)
|
||||
// Live: `b` is gone. Historical (pinned at gen 1): the un-flushed removal is
|
||||
// overlaid out, so `b` is still present at its pinned state.
|
||||
// Live: `b` is gone. Historical (pinned at the transact's generation): the
|
||||
// un-flushed removal is overlaid out, so `b` is still present at its
|
||||
// pinned state.
|
||||
expect(liveIds).toContain(a)
|
||||
expect(liveIds).not.toContain(b)
|
||||
expect(pastIds).toContain(a)
|
||||
|
|
@ -1262,11 +1281,14 @@ describe('8.0 Db API — generational MVCC', () => {
|
|||
|
||||
it('Model-B retention — explicit caps reclaim single-op history; committed history survives reopen', async () => {
|
||||
const { brain, dir } = await openFsBrain()
|
||||
// Relative baseline: the adopt-at-open backfill holds the first
|
||||
// generation(s), so the 6 writes below land at base+1..base+6.
|
||||
const base = brain.generation()
|
||||
const a = uid('ret-a')
|
||||
await brain.add({ id: a, type: NounType.Document, data: 'a', vector: vec(1), metadata: { v: 1 } })
|
||||
for (let v = 2; v <= 6; v++) await brain.update({ id: a, metadata: { v } })
|
||||
await brain.flush() // persist the per-write generations to disk
|
||||
expect(brain.generation()).toBe(6)
|
||||
expect(brain.generation()).toBe(base + 6)
|
||||
|
||||
// Cap to the 2 most recent generations — older single-op history is reclaimed.
|
||||
const res = await brain.compactHistory({ maxGenerations: 2 })
|
||||
|
|
|
|||
|
|
@ -36,6 +36,9 @@ import { GenerationCompactedError } from '../../src/db/errors.js'
|
|||
import type { GenerationStore } from '../../src/db/generationStore.js'
|
||||
import { NounType } from '../../src/types/graphTypes.js'
|
||||
|
||||
/** The VFS root — re-committed by the adopt-at-open baseline backfill. */
|
||||
const VFS_ROOT = '00000000-0000-0000-0000-000000000000'
|
||||
|
||||
/** Deterministic 384-dim vector so no test ever invokes the embedder. */
|
||||
function vec(seed: number): number[] {
|
||||
return Array.from({ length: 384 }, (_, i) => ((seed * 31 + i * 7) % 100) / 100)
|
||||
|
|
@ -133,7 +136,11 @@ describe('8.0 Db API — temporal range verbs', () => {
|
|||
expect(viaDb).toEqual(viaGen)
|
||||
expect(viaDb.fromGeneration).toBe(g1)
|
||||
expect(viaDb.nouns).toEqual([a, b].sort()) // a (updated after g1) + b (added after g1)
|
||||
expect(viaEpoch.nouns).toEqual([a, b].sort()) // (0, now] also includes a's creation, still {a, b}
|
||||
// (0, now] also includes a's creation — still {a, b} among user rows. The
|
||||
// adopt-at-open baseline backfill re-commits the VFS root as a real
|
||||
// generation, so the full-epoch window legitimately reports it too;
|
||||
// filter it to keep this pin about the user writes.
|
||||
expect(viaEpoch.nouns.filter((n) => n !== VFS_ROOT)).toEqual([a, b].sort())
|
||||
|
||||
// direction guard: an older view cannot be `since` a newer lower bound
|
||||
const older = await brain.asOf(1)
|
||||
|
|
@ -163,7 +170,11 @@ describe('8.0 Db API — temporal range verbs', () => {
|
|||
}
|
||||
|
||||
const all = await brain.transactionLog()
|
||||
expect(all.map((e) => e.generation)).toEqual([...gens].reverse()) // newest first
|
||||
// Newest first — compared above the open baseline (the adopt-at-open
|
||||
// backfill logs its own generation(s) below the first user write).
|
||||
expect(all.map((e) => e.generation).filter((g) => g >= gens[0])).toEqual(
|
||||
[...gens].reverse()
|
||||
)
|
||||
|
||||
// INCLUSIVE both ends — gens[1] AND gens[3] are present (contrast since's exclusive lower).
|
||||
const windowed = await brain.transactionLog({ from: gens[1], to: gens[3] })
|
||||
|
|
@ -334,19 +345,22 @@ describe('8.0 Db API — temporal range verbs', () => {
|
|||
// 7. Granularity (Model-B) ---------------------------------------------------
|
||||
it('granularity: single-operation writes ARE versioned and visible to the temporal verbs', async () => {
|
||||
const brain = await openMemoryBrain()
|
||||
// Relative baseline: the adopt-at-open backfill already logged its own
|
||||
// generation(s) — pin the DELTA this test's writes add, not a count.
|
||||
const baseCount = (await brain.transactionLog()).length
|
||||
const a = uid('gran-a')
|
||||
const r1 = await brain.transact([
|
||||
{ op: 'add', id: a, type: NounType.Document, data: 'a', vector: vec(1), metadata: { v: 1 } }
|
||||
])
|
||||
await r1.release()
|
||||
expect((await brain.transactionLog()).length).toBe(1)
|
||||
expect((await brain.transactionLog()).length).toBe(baseCount + 1)
|
||||
|
||||
// Model-B: a single-op write is its OWN immutable generation — logged,
|
||||
// diffable, and time-travelable, exactly like a transact() of one op.
|
||||
await brain.update({ id: a, metadata: { v: 2 } })
|
||||
|
||||
// The single-op update appended a generation/log entry.
|
||||
expect((await brain.transactionLog()).length).toBe(2)
|
||||
expect((await brain.transactionLog()).length).toBe(baseCount + 2)
|
||||
expect(brain.generation()).toBe(r1.generation + 1)
|
||||
|
||||
// diff sees the single-op update as a modification of `a`.
|
||||
|
|
|
|||
|
|
@ -109,14 +109,14 @@ describe('durability kill matrix — crash at every commit-path step, recover by
|
|||
/**
|
||||
* Flip a brain to durable-at-ack (log-authority) mode.
|
||||
*
|
||||
* NOT via `adoptLogAuthority()`: the sanctioned flip REFUSES on a freshly
|
||||
* materialized brain — its verification oracle reports the generation-0
|
||||
* VFS-root baseline as a divergence (`state-differs` even after an
|
||||
* identity-update backfill; verified 2026-08-10). This helper flips the
|
||||
* SAME switch the sanctioned path flips (`setLogDurability('at-ack')`) and
|
||||
* persists the SAME authority artifact, so a reopened brain also runs in
|
||||
* log-authority mode. The durability semantics under test are governed
|
||||
* entirely by that switch.
|
||||
* NOT via `adoptLogAuthority()` (and the helper opens every brain with
|
||||
* `logAuthority: 'defer'`, opting out of the 10.0.0 adopt-at-open fleet
|
||||
* default): the sanctioned path runs the oracle and a baseline backfill,
|
||||
* which appends its own generation — shifting the floor arithmetic every
|
||||
* row pins. This helper flips the SAME switch the sanctioned path flips
|
||||
* (`setLogDurability('at-ack')`) and persists the SAME authority artifact,
|
||||
* so a reopened brain also runs in log-authority mode. The durability
|
||||
* semantics under test are governed entirely by that switch.
|
||||
*/
|
||||
async function flipToAtAck(brain: Brainy): Promise<void> {
|
||||
const storage = (
|
||||
|
|
|
|||
|
|
@ -4,14 +4,13 @@
|
|||
*
|
||||
* (1) FSYNC-BEFORE-ACK: an acknowledged write's fact survives an abrupt
|
||||
* process end (no flush, no close — reopen from disk).
|
||||
* - transact(): HOLDS TODAY — the fact is fsync'd before transact returns.
|
||||
* - single-op: PINNED AS `it.fails` — today's group-commit batches
|
||||
* DURABILITY (ack precedes the group fsync; a hard kill loses the fact
|
||||
* AND the generation together, coherently — the documented Model-B
|
||||
* contract, fine while the tree is authoritative). The destination
|
||||
* (ack-at-log) requires group commit to become LATENCY batching: the
|
||||
* ack waits for the shared fsync. When that lands, this pin flips red —
|
||||
* remove `.fails` and the contract is permanent. No cliff to discover.
|
||||
* - transact(): HOLDS — the fact is fsync'd before transact returns.
|
||||
* - single-op: HOLDS (was pinned `it.fails` until the ack-at-log
|
||||
* destination landed): the 10.0.0 adopt-at-open fleet default flips a
|
||||
* fresh brain to log authority at open, so single-op acks await the
|
||||
* covering group fsync (durable-at-ack) and recovery REPLAYS intact
|
||||
* facts above the manifest at the next open. The contract is now
|
||||
* permanent on every path.
|
||||
*
|
||||
* (2) SCAN STABILITY UNDER ROTATION: a scan handle opened before segment
|
||||
* rotation yields exactly its snapshot — byte-identical facts, no gaps,
|
||||
|
|
@ -63,9 +62,11 @@ describe('fsync-before-ack contract (fact durability at the ack boundary)', () =
|
|||
expect(facts.some((f) => f.generation === receipt.generation)).toBe(true)
|
||||
})
|
||||
|
||||
// PINNED (flips red when group commit becomes latency batching — then
|
||||
// remove `.fails` and the ack-at-log contract is permanent on every path).
|
||||
it.fails('single-op: the fact is durable the moment the ack returns (the ack-at-log target)', async () => {
|
||||
// THE ACK-AT-LOG CONTRACT, HELD (was `.fails` until it landed): under the
|
||||
// adopt-at-open fleet default this brain runs durable-at-ack from open —
|
||||
// the ack waits for the covering log fsync, and the log-authority recovery
|
||||
// path replays the intact fact at the next open instead of truncating it.
|
||||
it('single-op: the fact is durable the moment the ack returns (the ack-at-log target)', async () => {
|
||||
await brain.add({ data: 'acked single-op', type: 'document', metadata: { n: 1 } })
|
||||
const ackedHead = brain.scanFacts()!.headGeneration
|
||||
// Abrupt end immediately after the ack — before any flush window.
|
||||
|
|
|
|||
|
|
@ -22,8 +22,12 @@ afterEach(async () => {
|
|||
for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
async function open(dir: string): Promise<Brainy> {
|
||||
const b = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false })
|
||||
async function open(dir: string, logAuthority?: 'adopt' | 'defer'): Promise<Brainy> {
|
||||
const b = new Brainy({
|
||||
storage: { type: 'filesystem', path: dir },
|
||||
requireSubtype: false,
|
||||
...(logAuthority ? { logAuthority } : {})
|
||||
})
|
||||
await b.init()
|
||||
brains.push(b)
|
||||
return b
|
||||
|
|
@ -80,4 +84,31 @@ describe('adoptLogAuthority — the sanctioned flip with self-backfill', () => {
|
|||
expect(report.verdict).toBe('green')
|
||||
expect(brain.logAuthority().authority).toBe('log')
|
||||
}, 120000)
|
||||
|
||||
// THE OPT-OUT CONTRACT (`logAuthority: 'defer'`): no automatic adoption —
|
||||
// the fresh brain stays tree-authoritative and writes NO artifact (a
|
||||
// deferred posture is config, not stored state); the EXPLICIT
|
||||
// adoptLogAuthority() then flips it exactly as before the fleet default.
|
||||
it("opt-out: 'defer' stays tree with no artifact until the explicit adoptLogAuthority() flips it", async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-adopt-defer-'))
|
||||
dirs.push(dir)
|
||||
const brain = await open(dir, 'defer')
|
||||
await brain.add({ data: 'deferred row', type: NounType.Document, metadata: { n: 1 } })
|
||||
await brain.flush()
|
||||
|
||||
expect(brain.logAuthority().authority, "'defer' skips open-time adoption").toBe('tree')
|
||||
const storage = (brain as unknown as {
|
||||
storage: { readRawObject(p: string): Promise<unknown | null> }
|
||||
}).storage
|
||||
const artifact = await storage.readRawObject('_system/log-authority.json').catch(() => null)
|
||||
expect(artifact, "'defer' writes no authority artifact").toBeNull()
|
||||
|
||||
const report = await brain.adoptLogAuthority()
|
||||
expect(report.verdict, 'the explicit flip still lands on green').toBe('green')
|
||||
expect(brain.logAuthority().authority).toBe('log')
|
||||
const stored = (await storage.readRawObject('_system/log-authority.json')) as {
|
||||
authority?: string
|
||||
} | null
|
||||
expect(stored?.authority, 'the explicit flip stores the artifact').toBe('log')
|
||||
}, 120000)
|
||||
})
|
||||
|
|
|
|||
|
|
@ -1,22 +1,34 @@
|
|||
/**
|
||||
* @module tests/integration/log-authority
|
||||
* @description The guarded log-authority core, end-to-end: the per-brain
|
||||
* authority switch (default 'tree', stored artifact, checked at open only),
|
||||
* the verification oracle (replay the fact log, diff latest per-id state
|
||||
* authority switch (stored artifact, checked at open only), the
|
||||
* verification oracle (replay the fact log, diff latest per-id state
|
||||
* against the canonical tree, NAME every divergence by class), the guarded
|
||||
* flip (refuses on red with the cure in the message; lands on green and
|
||||
* engages durable-at-ack immediately), and the switch surviving reopen.
|
||||
*
|
||||
* THE 10.0.0 FLEET DEFAULT is ADOPT-AT-OPEN (`logAuthority: 'adopt'`): a
|
||||
* fresh brain with no stored artifact runs the oracle at open, backfills
|
||||
* curable divergences, and flips to log authority on green — so a
|
||||
* default-config brain opens ALREADY log-authoritative and durable-at-ack.
|
||||
* The first two pins hold that default and its explicit opt-out
|
||||
* (`logAuthority: 'defer'`, the pre-10 tree behavior). Every test below
|
||||
* them that exercises the ORACLE or the EXPLICIT flip opens its brain with
|
||||
* `'defer'` — otherwise the open-time adoption would have pre-flipped the
|
||||
* brain and pre-cured the very divergences under test.
|
||||
*
|
||||
* KNOWN GAPS PINNED WITH `.fails` (real findings, not test bugs — see the
|
||||
* comments on each): a fresh brain is NOT log-complete by construction
|
||||
* today, because the VFS root is written at init as a baseline
|
||||
* (generation-less) write that never gets a fact, so the oracle reports it
|
||||
* as a `pre-log-record` and no fresh brain can flip without a manual
|
||||
* baseline backfill. The tests that need a green oracle perform that
|
||||
* backfill explicitly (an identity update of the root as the FINAL write —
|
||||
* final, because derived-index maintenance rewrites canonical noun records
|
||||
* outside generations, so an earlier fact's after-image goes stale; see the
|
||||
* module tail comment on `backfillBaseline`).
|
||||
* as a `pre-log-record`. The open-time adoption (and adoptLogAuthority())
|
||||
* CURES this by baseline backfill — a re-commit, not construction — so the
|
||||
* by-construction pin stays `.fails` on a deferred brain. Tests that need
|
||||
* a green oracle on a deferred brain perform that backfill explicitly (an
|
||||
* identity update of the root as the FINAL write — final, because
|
||||
* derived-index maintenance rewrites canonical noun records outside
|
||||
* generations, so an earlier fact's after-image goes stale; see the module
|
||||
* tail comment on `backfillBaseline`).
|
||||
*/
|
||||
import { describe, it, expect, afterEach } from 'vitest'
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
|
|
@ -88,14 +100,24 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
const dirs: string[] = []
|
||||
const brains: Brainy[] = []
|
||||
|
||||
const openBrain = async (dir?: string): Promise<{ brain: Brainy; dir: string }> => {
|
||||
/**
|
||||
* Open a brain over `dir`. Omit `logAuthority` to exercise the FLEET
|
||||
* DEFAULT (adopt-at-open); pass `'defer'` for the tests that need a
|
||||
* tree-authoritative brain so the oracle/explicit-flip path is actually
|
||||
* the thing under test (the default would pre-flip and pre-backfill).
|
||||
*/
|
||||
const openBrain = async (
|
||||
dir?: string,
|
||||
logAuthority?: 'adopt' | 'defer'
|
||||
): Promise<{ brain: Brainy; dir: string }> => {
|
||||
const d = dir ?? mkdtempSync(join(tmpdir(), 'brainy-log-authority-'))
|
||||
if (!dir) dirs.push(d)
|
||||
const brain = new Brainy({
|
||||
storage: { type: 'filesystem', path: d },
|
||||
requireSubtype: false,
|
||||
silent: true,
|
||||
dimensions: 384
|
||||
dimensions: 384,
|
||||
...(logAuthority ? { logAuthority } : {})
|
||||
})
|
||||
brains.push(brain)
|
||||
await brain.init()
|
||||
|
|
@ -109,8 +131,37 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
it('DEFAULT IS TREE: a fresh brain reports tree authority, stores no artifact, and plain acks never await a log fsync', async () => {
|
||||
const { brain } = await openBrain()
|
||||
// THE RULED DEFAULT (10.0.0): with no config and no stored artifact, a
|
||||
// fresh brain ADOPTS log authority at open — oracle green (the open-time
|
||||
// baseline backfill cures the generation-0 VFS root), artifact on disk,
|
||||
// durable-at-ack live from the first write.
|
||||
it('DEFAULT IS ADOPT-AT-OPEN: a fresh brain opens already log-authoritative — artifact stored, plain acks await the covering log fsync', async () => {
|
||||
const { brain } = await openBrain() // no logAuthority config = the fleet default
|
||||
|
||||
const authority = brain.logAuthority()
|
||||
expect(authority.authority).toBe('log')
|
||||
expect(typeof authority.flippedAt).toBe('number')
|
||||
expect(authority.oracle, 'the open-time flip records its green oracle summary').toBeDefined()
|
||||
|
||||
const artifact = (await internals(brain)
|
||||
.storage.readRawObject(AUTHORITY_ARTIFACT)
|
||||
.catch(() => null)) as { authority?: string } | null
|
||||
expect(artifact, 'the adoption wrote the switch artifact').not.toBeNull()
|
||||
expect(artifact!.authority).toBe('log')
|
||||
|
||||
// The MODE assertion (not a timing one): in log authority a single-op
|
||||
// ack awaits the log's covering-fsync path.
|
||||
expect(internals(brain).generationStore.logDurability).toBe('at-ack')
|
||||
const spy = spyEnsureSynced(brain)
|
||||
await brain.add({ data: 'log mode write', type: 'document', metadata: { n: 1 } })
|
||||
expect(spy.calls(), 'adopted default: add() awaits the covering fsync').toBeGreaterThanOrEqual(1)
|
||||
})
|
||||
|
||||
// THE EXPLICIT OPT-OUT: `logAuthority: 'defer'` is the pre-10 behavior —
|
||||
// tree authority, NO artifact written (a deferred posture is config, not
|
||||
// stored state), and single-op acks never await a log fsync.
|
||||
it("OPT-OUT ('defer'): the brain stays tree-authoritative, stores no artifact, and plain acks never await a log fsync", async () => {
|
||||
const { brain } = await openBrain(undefined, 'defer')
|
||||
|
||||
expect(brain.logAuthority().authority).toBe('tree')
|
||||
expect(brain.logAuthority().flippedAt).toBeUndefined()
|
||||
|
|
@ -118,7 +169,7 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
const artifact = await internals(brain)
|
||||
.storage.readRawObject(AUTHORITY_ARTIFACT)
|
||||
.catch(() => null)
|
||||
expect(artifact, 'no switch artifact exists before any flip').toBeNull()
|
||||
expect(artifact, "'defer' writes no switch artifact").toBeNull()
|
||||
|
||||
// The MODE assertion (not a timing one): in tree authority a single-op
|
||||
// ack must never call the log's covering-fsync path.
|
||||
|
|
@ -134,10 +185,12 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
// (00000000-0000-0000-0000-000000000000) is created at init by a baseline
|
||||
// write with NO generation and NO fact, yet it is enumerated by the
|
||||
// canonical walk — so the oracle on a fresh brain is red with exactly one
|
||||
// `pre-log-record` mismatch on the root, and adoptLogAuthority() refuses
|
||||
// on every fresh brain. Verified empirically on this branch.
|
||||
// `pre-log-record` mismatch on the root. The adopt-at-open default (and
|
||||
// adoptLogAuthority()) CURES this by baseline backfill — a re-commit,
|
||||
// which is why this pin opens with 'defer': it holds the BY-CONSTRUCTION
|
||||
// intent, which the backfill masks but does not deliver.
|
||||
it.fails('ORACLE INTENT: a fresh brain is log-complete by construction — verdict green with zero mismatches', async () => {
|
||||
const { brain } = await openBrain()
|
||||
const { brain } = await openBrain(undefined, 'defer')
|
||||
await seedWrites(brain)
|
||||
await brain.flush()
|
||||
|
||||
|
|
@ -147,7 +200,9 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
})
|
||||
|
||||
it('a fresh, un-backfilled brain diverges ONLY on the init-time baseline record — every user write is exactly reproduced', async () => {
|
||||
const { brain } = await openBrain()
|
||||
// 'defer': the adopt-at-open default would have backfilled the baseline
|
||||
// already — this pin needs the brain genuinely un-backfilled.
|
||||
const { brain } = await openBrain(undefined, 'defer')
|
||||
await seedWrites(brain)
|
||||
await brain.flush()
|
||||
|
||||
|
|
@ -166,7 +221,10 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
})
|
||||
|
||||
it('THE ORACLE GOES GREEN on a log-complete brain: adds + update + remove, every canonical row exactly reproduced', async () => {
|
||||
const { brain } = await openBrain()
|
||||
// 'defer' + manual backfill: the exact-count pins below (5 generations)
|
||||
// depend on the log holding ONLY this test's writes — the adopt-at-open
|
||||
// default would inject its own backfill generation at init.
|
||||
const { brain } = await openBrain(undefined, 'defer')
|
||||
await seedWrites(brain)
|
||||
await backfillBaseline(brain) // final write — see the helper's contract
|
||||
await brain.flush()
|
||||
|
|
@ -184,7 +242,7 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
})
|
||||
|
||||
it('THE ORACLE NAMES pre-log records: a canonical row no fact ever recorded reports pre-log-record, by id', async () => {
|
||||
const { brain } = await openBrain()
|
||||
const { brain } = await openBrain(undefined, 'defer')
|
||||
await seedWrites(brain)
|
||||
await backfillBaseline(brain)
|
||||
await brain.flush()
|
||||
|
|
@ -226,7 +284,9 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
// and the flip proceeds; ONLY log-AHEAD divergences (the log claims
|
||||
// state canonical denies) refuse, because no backfill can make the log
|
||||
// un-claim a live row. This test stages exactly that incurable shape.
|
||||
const { brain } = await openBrain()
|
||||
// 'defer': the brain must still be tree-authoritative (no artifact) so
|
||||
// the refusal's nothing-written pins below have meaning.
|
||||
const { brain } = await openBrain(undefined, 'defer')
|
||||
const { kept } = await seedWrites(brain)
|
||||
await backfillBaseline(brain)
|
||||
await brain.flush()
|
||||
|
|
@ -254,7 +314,9 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
})
|
||||
|
||||
it('THE FLIP LANDS ON GREEN: the report is the receipt, the artifact is on disk, and durable-at-ack engages immediately', async () => {
|
||||
const { brain } = await openBrain()
|
||||
// 'defer': this pin exercises the EXPLICIT flip — the adopt-at-open
|
||||
// default would have landed it before the test began.
|
||||
const { brain } = await openBrain(undefined, 'defer')
|
||||
await seedWrites(brain)
|
||||
await backfillBaseline(brain)
|
||||
await brain.flush()
|
||||
|
|
@ -284,7 +346,7 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
})
|
||||
|
||||
it('THE SWITCH SURVIVES REOPEN: authority restored at open with no re-verification, durable-at-ack active in the new session', async () => {
|
||||
const { brain, dir } = await openBrain()
|
||||
const { brain, dir } = await openBrain(undefined, 'defer')
|
||||
await seedWrites(brain)
|
||||
await backfillBaseline(brain)
|
||||
await brain.flush()
|
||||
|
|
@ -292,7 +354,10 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
const flipReceipt = brain.logAuthority()
|
||||
await (brain as unknown as { close: () => Promise<void> }).close()
|
||||
|
||||
const { brain: reopened } = await openBrain(dir)
|
||||
// Reopen with 'defer' too: the restored authority below can then ONLY
|
||||
// come from the stored artifact (a stored artifact always wins; had the
|
||||
// default re-adopted, flippedAt/oracle would differ from the receipt).
|
||||
const { brain: reopened } = await openBrain(dir, 'defer')
|
||||
const restored = reopened.logAuthority()
|
||||
expect(restored.authority).toBe('log')
|
||||
// No re-verification happened at open: the restored record IS the stored
|
||||
|
|
@ -308,7 +373,7 @@ describe('log authority — the switch, the oracle, the guarded flip', () => {
|
|||
})
|
||||
|
||||
it('STATE-DIFFERS: canonical drift the write path never saw is named, by id', async () => {
|
||||
const { brain } = await openBrain()
|
||||
const { brain } = await openBrain(undefined, 'defer')
|
||||
const { kept } = await seedWrites(brain)
|
||||
await backfillBaseline(brain)
|
||||
await brain.flush()
|
||||
|
|
|
|||
|
|
@ -43,6 +43,13 @@ describe('transact durability barrier — entity writes fsync before the counter
|
|||
})
|
||||
await brain.init()
|
||||
|
||||
// Drain the pending tier BEFORE instrumenting: the adopt-at-open fleet
|
||||
// default re-commits the init-time baseline as a buffered single-op
|
||||
// generation, and transact() flushes buffered single-ops first — that
|
||||
// flush's manifest sync would otherwise be recorded ahead of the
|
||||
// transact's own commit point and break the first-index ordering pins.
|
||||
await brain.flush()
|
||||
|
||||
// Instrument the real filesystem storage: record every fsync batch in order,
|
||||
// and count barrier open/flush, delegating to the originals.
|
||||
syncCalls = []
|
||||
|
|
|
|||
|
|
@ -477,14 +477,17 @@ describe('materializeAtGeneration — bounded & deadlock-free (GA #33)', () => {
|
|||
const store = (brain as any).generationStore
|
||||
|
||||
const N = 400
|
||||
// Relative, not absolute: under the adopt-at-open default the open-time
|
||||
// baseline backfill takes a generation of its own, so the first add is
|
||||
// NOT generation 1 — pin the deep generation to the first add's commit.
|
||||
let deepGen = 0
|
||||
for (let i = 0; i < N; i++) {
|
||||
await brain.add({ data: `doc ${i}`, type: NounType.Document, subtype: 'note', metadata: { i }, vector: VEC })
|
||||
if (i === 0) deepGen = brain.generation()
|
||||
}
|
||||
const R = brain.generation() // ≈ N (each add is its own generation)
|
||||
expect(R).toBeGreaterThanOrEqual(N)
|
||||
|
||||
const deepGen = 1
|
||||
|
||||
// Count getDelta invocations during the materialize.
|
||||
const realGetDelta = store.getDelta.bind(store)
|
||||
let getDeltaCalls = 0
|
||||
|
|
@ -509,7 +512,8 @@ describe('materializeAtGeneration — bounded & deadlock-free (GA #33)', () => {
|
|||
expect(getDeltaCalls).toBeLessThan(R * 5)
|
||||
expect(getDeltaCalls).toBeLessThan(N * N) // the regression guard
|
||||
|
||||
// The materialized at-gen-1 brain holds exactly the one entity that existed.
|
||||
// The materialized brain at the first add's generation holds exactly the
|
||||
// one user entity that existed.
|
||||
const atGen1 = await handle.find({ limit: N + 10 })
|
||||
expect(atGen1.length).toBe(1)
|
||||
await handle.close()
|
||||
|
|
|
|||
|
|
@ -7,12 +7,13 @@
|
|||
* one), a solo writer syncs immediately, and at the brain level an at-ack
|
||||
* ack resolving means the write's fact is on disk.
|
||||
*
|
||||
* One pin is marked `.fails` (real finding, not a test bug): the at-ack
|
||||
* durability contract says an acked write's fact survives power loss, but
|
||||
* FactLog.open() truncates every fact beyond the store's committed
|
||||
* generation watermark — which only advances at the pending-tier flush. A
|
||||
* crash-shaped reopen (acks landed, flush never ran) therefore DISCARDS the
|
||||
* fsynced facts at open. See the test comment for the exact mechanism.
|
||||
* The final pin holds the at-ack durability contract END TO END: an acked
|
||||
* write's fact survives a crash-shaped reopen. This was a `.fails` known
|
||||
* gap (FactLog.open() truncated every fact beyond the committed watermark,
|
||||
* which only advances at the pending-tier flush) — CURED by the 10.0.0
|
||||
* adopt-at-open fleet default: a fresh brain stores the log-authority
|
||||
* artifact at open, and under 'log' authority recovery REPLAYS intact
|
||||
* facts above the manifest instead of truncating them.
|
||||
*/
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
|
|
@ -188,9 +189,9 @@ describe('durable-at-ack through the brain (group commit end-to-end)', () => {
|
|||
|
||||
it('at-ack: N concurrent add() acks all resolve, every ack was covered by a log sync, and every fact is on disk after reopen', async () => {
|
||||
const { brain, dir } = await openBrain()
|
||||
// White-box: engage the at-ack durability mode directly (the guarded
|
||||
// authority flip that normally enables it is covered by the integration
|
||||
// suite — this test pins the durability machinery itself).
|
||||
// The 10.0.0 fleet default already adopted log authority at open, so
|
||||
// the brain is at-ack; the white-box engage stays so this pin holds the
|
||||
// durability MACHINERY itself independent of the open-time posture.
|
||||
brain.generationStore.setLogDurability('at-ack')
|
||||
|
||||
const factLog = brain.generationStore.getFactLog()
|
||||
|
|
@ -231,19 +232,17 @@ describe('durable-at-ack through the brain (group commit end-to-end)', () => {
|
|||
}
|
||||
})
|
||||
|
||||
// KNOWN GAP (marked .fails — remove the marker when fixed in src): the
|
||||
// at-ack contract is that an acked write's fact survives power loss. The
|
||||
// fsync at ack does put the fact's bytes on disk — but FactLog.open()
|
||||
// truncates every fact with generation > the store's committed watermark,
|
||||
// and that watermark only advances at the pending-tier flush
|
||||
// (flushPendingSingleOps). So on a crash-shaped reopen (acks landed, flush
|
||||
// never ran) the store logs "[FactLog] truncating N uncommitted fact(s)"
|
||||
// and DISCARDS the acked, fsynced facts. Until recovery treats the log as
|
||||
// authoritative past the tree's watermark (or the watermark goes durable
|
||||
// at ack), durable-at-ack does not survive the very crash it exists for.
|
||||
it.fails('at-ack CONTRACT: acked facts survive a crash-shaped reopen (no flush ever ran)', async () => {
|
||||
// THE AT-ACK CONTRACT, HELD (was a `.fails` known gap): an acked write's
|
||||
// fact survives a crash-shaped reopen. Fixed by the 10.0.0 adopt-at-open
|
||||
// fleet default — this brain adopted LOG authority at open (artifact
|
||||
// stored, durable-at-ack live), and under 'log' authority FactLog
|
||||
// recovery REPLAYS intact facts above the committed watermark at the next
|
||||
// open instead of truncating them back. Durable-at-ack now survives the
|
||||
// very crash it exists for.
|
||||
it('at-ack CONTRACT: acked facts survive a crash-shaped reopen (no flush ever ran)', async () => {
|
||||
const { brain, dir } = await openBrain()
|
||||
brain.generationStore.setLogDurability('at-ack')
|
||||
expect(brain.logAuthority().authority, 'the fleet default adopted at open').toBe('log')
|
||||
expect(brain.generationStore.logDurability).toBe('at-ack')
|
||||
// Crash simulation: the pending-tier durability flush never happens
|
||||
// (every trigger routes through flushPendingSingleOps), and the brain is
|
||||
// abandoned without close() — exactly the power-loss shape at-ack is for.
|
||||
|
|
|
|||
97
tests/unit/db/torn-open-guards.test.ts
Normal file
97
tests/unit/db/torn-open-guards.test.ts
Normal file
|
|
@ -0,0 +1,97 @@
|
|||
/**
|
||||
* @module tests/unit/db/torn-open-guards
|
||||
* @description Power-cut throw-site cures (brainy-alone fault-injection
|
||||
* findings, both release-gating):
|
||||
* 1. A torn generation manifest/counter (NaN/garbage where a generation
|
||||
* belongs) DISCARDS with narration and re-derives — never a RangeError
|
||||
* killing the open.
|
||||
* 2. A manifest-listed-but-unloadable column segment QUARANTINES at
|
||||
* discovery with narration; the field serves its remaining segments
|
||||
* DEGRADED — never a raw throw killing every query on the field.
|
||||
*/
|
||||
import { describe, it, expect, afterEach } from 'vitest'
|
||||
import { mkdtempSync, rmSync, readdirSync, writeFileSync, readFileSync, existsSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { gzipSync } from 'node:zlib'
|
||||
import { Brainy } from '../../../src/index.js'
|
||||
import { NounType } from '../../../src/types/graphTypes.js'
|
||||
|
||||
const dirs: string[] = []
|
||||
const brains: Brainy[] = []
|
||||
afterEach(async () => {
|
||||
for (const b of brains.splice(0)) await b.close().catch(() => {})
|
||||
for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
async function open(dir: string): Promise<Brainy> {
|
||||
const b = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false })
|
||||
await b.init()
|
||||
brains.push(b)
|
||||
return b
|
||||
}
|
||||
|
||||
describe('torn-open guards', () => {
|
||||
it('a torn generation manifest (NaN) opens with narrated discard — never a RangeError', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-torn-gen-'))
|
||||
dirs.push(dir)
|
||||
let brain = await open(dir)
|
||||
const id = await brain.add({ data: 'survivor row', type: NounType.Document, metadata: { k: 1 } })
|
||||
await brain.flush()
|
||||
await brain.close()
|
||||
brains.pop()
|
||||
|
||||
// The power-cut shape: the manifest's generation field is garbage.
|
||||
const sys = join(dir, '_system')
|
||||
const manifestPath = ['manifest.json', 'manifest.json.gz']
|
||||
.map((f) => join(sys, f))
|
||||
.find((p) => existsSync(p))!
|
||||
const torn = { version: 1, generation: 'NaN-garbage', committedAt: 'x', horizon: null }
|
||||
if (manifestPath.endsWith('.gz')) writeFileSync(manifestPath, gzipSync(JSON.stringify(torn)))
|
||||
else writeFileSync(manifestPath, JSON.stringify(torn))
|
||||
|
||||
// Open MUST succeed (narrated discard + recovery re-derivation), and the
|
||||
// durable row must still serve (log-authority replay recovers it).
|
||||
brain = await open(dir)
|
||||
expect((await brain.get(id))!.data).toContain('survivor row')
|
||||
// Writes continue with a sane monotonic generation.
|
||||
await brain.add({ data: 'post-recovery', type: NounType.Document, metadata: { k: 2 } })
|
||||
expect(Number.isSafeInteger(brain.generation())).toBe(true)
|
||||
}, 120000)
|
||||
|
||||
it('a torn column segment quarantines at discovery; the field serves remaining segments degraded — never a raw throw', async () => {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'brainy-torn-seg-'))
|
||||
dirs.push(dir)
|
||||
let brain = await open(dir)
|
||||
for (let i = 0; i < 6; i++) {
|
||||
await brain.add({ data: `row ${i}`, type: NounType.Document, metadata: { bucket: i % 2 } })
|
||||
}
|
||||
await brain.flush()
|
||||
await brain.close()
|
||||
brains.pop()
|
||||
|
||||
// Tear ONE column segment's bytes on disk (manifest keeps listing it) —
|
||||
// the QUERIED field's own segment, so the quarantine path provably
|
||||
// engages. Column segments live under the raw-blob root:
|
||||
// `<root>/_blobs/_column_index/<field>/L<level>-<id>.bin`.
|
||||
const segDir = join(dir, '_blobs', '_column_index', 'bucket')
|
||||
let tornOne = false
|
||||
if (existsSync(segDir)) {
|
||||
for (const f of readdirSync(segDir, { withFileTypes: true })) {
|
||||
if (!f.isDirectory() && /^L\d+-.*\.bin$/.test(f.name)) {
|
||||
writeFileSync(join(segDir, f.name), Buffer.from([0x00, 0x01, 0x02])) // garbage
|
||||
tornOne = true
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
expect(tornOne, 'found a segment file to tear (layout probe)').toBe(true)
|
||||
|
||||
// Queries on the field MUST NOT throw — degraded-announced service.
|
||||
brain = await open(dir)
|
||||
const rows = await brain.find({ where: { bucket: 0 }, limit: 10 })
|
||||
expect(Array.isArray(rows), 'query survives the torn segment').toBe(true)
|
||||
// Full completeness is NOT asserted (the torn segment's rows may be
|
||||
// absent — that is the documented degraded contract until heal).
|
||||
}, 120000)
|
||||
})
|
||||
|
|
@ -5,19 +5,22 @@
|
|||
* doing so dropped every entity in that segment out of `filter`/`rangeQuery`/
|
||||
* `sortTopK` with no error, so a corrupt index looked like a merely short result.
|
||||
*
|
||||
* The three failure classes and their required behaviour:
|
||||
* The three failure classes and their required behaviour (torn-segment
|
||||
* QUARANTINE contract — a raw throw at query time killed every query on the
|
||||
* field forever; a silent skip hid the loss; quarantine is the middle):
|
||||
* - a real storage IO fault (EIO) PROPAGATES verbatim — a present-but-unreadable
|
||||
* segment is not "absent", so it must not read as an empty result;
|
||||
* - a manifest-listed segment with undecodable bytes throws `ColumnSegmentLoadError`;
|
||||
* - a manifest-listed segment with NO bytes (gone on disk) throws `ColumnSegmentLoadError`.
|
||||
* - a manifest-listed segment with undecodable bytes is QUARANTINED at
|
||||
* discovery: the query serves the field's remaining segments degraded and
|
||||
* `quarantinedSegments()` reports the torn segment (loud once, counted
|
||||
* always, healable);
|
||||
* - a manifest-listed segment with NO bytes (gone on disk) quarantines the
|
||||
* same way.
|
||||
* Only genuine absence stays benign: querying a field that has no manifest at all
|
||||
* returns empty (nothing was ever written for it) — that is not a fault.
|
||||
*/
|
||||
import { describe, it, expect, beforeEach } from 'vitest'
|
||||
import {
|
||||
ColumnStore,
|
||||
ColumnSegmentLoadError
|
||||
} from '../../../../src/indexes/columnStore/ColumnStore.js'
|
||||
import { ColumnStore } from '../../../../src/indexes/columnStore/ColumnStore.js'
|
||||
import { MemoryStorage } from '../../../../src/storage/adapters/memoryStorage.js'
|
||||
import { EntityIdMapper } from '../../../../src/utils/entityIdMapper.js'
|
||||
|
||||
|
|
@ -80,30 +83,44 @@ describe('ColumnStore segment-load faults surface loudly, absence stays benign (
|
|||
return s
|
||||
}
|
||||
|
||||
it('propagates a storage IO fault verbatim — not [] and not a ColumnSegmentLoadError', async () => {
|
||||
it('propagates a storage IO fault verbatim — not [] and not a quarantine (a present-but-unreadable segment is not torn)', async () => {
|
||||
storage.faultMode = 'io'
|
||||
const store = await reopen()
|
||||
await expect(store.filter('createdAt', 300)).rejects.toMatchObject({
|
||||
code: 'EIO'
|
||||
})
|
||||
// An IO fault is NOT quarantined — the segment may be fine once the disk
|
||||
// recovers; only torn/absent bytes enter the ledger.
|
||||
expect(store.quarantinedSegments('createdAt')).toEqual([])
|
||||
await store.close()
|
||||
})
|
||||
|
||||
it('throws ColumnSegmentLoadError when a manifest-listed segment is undecodable', async () => {
|
||||
it('QUARANTINES an undecodable manifest-listed segment at discovery — the query serves degraded, the ledger names the tear', async () => {
|
||||
storage.faultMode = 'corrupt'
|
||||
const store = await reopen()
|
||||
await expect(
|
||||
store.sortTopK('createdAt', 'desc', 10)
|
||||
).rejects.toBeInstanceOf(ColumnSegmentLoadError)
|
||||
// Degraded-announced serve: the field's only segment is torn, so the
|
||||
// result is empty — but the query completes instead of throwing.
|
||||
const sorted = await store.sortTopK('createdAt', 'desc', 10)
|
||||
expect(sorted).toEqual([])
|
||||
const ledger = store.quarantinedSegments('createdAt')
|
||||
expect(ledger).toHaveLength(1)
|
||||
expect(ledger[0].error).toMatch(/decode failed/)
|
||||
expect(ledger[0].hits).toBeGreaterThanOrEqual(1)
|
||||
// Subsequent queries keep serving (skip + count), never a throw.
|
||||
const hitsBefore = ledger[0].hits
|
||||
await expect(store.filter('createdAt', 300)).resolves.toBeDefined()
|
||||
expect(store.quarantinedSegments('createdAt')[0].hits).toBeGreaterThan(hitsBefore)
|
||||
await store.close()
|
||||
})
|
||||
|
||||
it('throws ColumnSegmentLoadError when a manifest-listed segment has no loadable bytes', async () => {
|
||||
it('QUARANTINES a manifest-listed segment with no loadable bytes — degraded serve, ledger entry, never a throw', async () => {
|
||||
storage.faultMode = 'missing'
|
||||
const store = await reopen()
|
||||
await expect(
|
||||
store.rangeQuery('createdAt', 100, 500)
|
||||
).rejects.toBeInstanceOf(ColumnSegmentLoadError)
|
||||
const bitmap = await store.rangeQuery('createdAt', 100, 500)
|
||||
expect(bitmap.size).toBe(0)
|
||||
const ledger = store.quarantinedSegments('createdAt')
|
||||
expect(ledger).toHaveLength(1)
|
||||
expect(ledger[0].error).toMatch(/no loadable bytes/)
|
||||
await store.close()
|
||||
})
|
||||
|
||||
|
|
|
|||
BIN
tests/unit/storage/torn-record-loud.test.ts
Normal file
BIN
tests/unit/storage/torn-record-loud.test.ts
Normal file
Binary file not shown.
Loading…
Add table
Add a link
Reference in a new issue