soulcraftlabs/open-brainy
2
0
Fork 0
Code Issues 1 Pull requests Projects Releases 342 Packages Wiki Activity Actions 5

perf(open): the pending-embed fold is bounded by a checkpoint of the SET, not an empty-only mark

Browse source 
The low-water mark shipped in 10.4.9 can only be written when the pending
set is EMPTY, because it carries no set — it means "everything at or below
G is consumed". A brain holding even one id that never lands (an embed that
keeps failing, a data-less row reaped in memory only and re-folded every
open) never drains, so it never writes a mark, so the bound never engaged on
exactly the brains whose fold is expensive: `recover-pending-embeds` re-read
the WHOLE fact log at every open, on the open's foreground.

_system/pending_embeds_checkpoint.json carries the set: { generation,
pending, writtenAt } = "as of durable generation G the pending set was
exactly this list". Open seeds the set from the list and scans from G + 1,
so the fold is O(facts since G) whether or not the set ever drains. Measured
on a 301-row brain with one stuck id: 302 facts read before, 0 after; at 601
rows, 602 before, 0 after — same pending set both ways.

THE DURABILITY LAW, by construction. A checkpoint at head H taken while the
facts up to H are still buffered would be read back after a crash that
truncated the tail: an `embed.landed` in a truncated fact would be gone from
the log while the checkpoint still recorded its id as landed, and its
landing vector went with the fact — a LOST VECTOR. So a capture is refused
unless `0 < head <= committed`, the manifest watermark below which
FactLog.open() never truncates and which the group-commit flush only
advances after fsyncing the log. The (generation, set) pair is taken in one
synchronous instant with no await between reading the generations and
snapshotting the set. The one remaining asymmetry runs the safe way: an id
enqueued in memory whose marker lands at G+1 is captured as pending at G —
one idempotent re-embed, never a loss.

Written at clean close (inside closeDurableSteps, after the generation
store's own close flushed the log and advanced the manifest), at
drain-to-empty, and on a cadence of max(64, ceil(|pending| / 64))
transitions while open — an interval that holds the mechanism's amortized
cost at <= 64 ids written per transition however large the backlog grows, so
the cure cannot reintroduce the defect class it fixes. No timer, no knob.
The debt stays armed across attempts the durability law refuses, so a write
burst does not skip a checkpoint, it defers it.

Degradation is loud and always toward a LONGER scan: a torn checkpoint
throws typed on read (the adapter's tmp+rename write means it can never
parse into a partial list) and a malformed one is refused whole, both
falling back to the low-water mark — still written, still read — and then to
generation 1. The fold narrates which bound applied and how many facts it
read, on every open, so a bound that stops engaging is visible instead of
silent.

The worker's orphan reap splits: a row that is GONE clears durably (its
tombstone is in the log, or its create never was), while a present-but
data-less row keeps clearing in memory only and is carried in the checkpoint
list, so the bounded fold and a full fold from generation 1 agree exactly.
The crash-recovery contract is unchanged: the fold stays on the open's
foreground, markers re-armed when open() returns.
This commit is contained in:
David Snelling 2026-09-02 10:08:34 -07:00
parent bc70c43d02
commit 15d4f65dcf
1 changed files with 390 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

416
src/brainy.ts
View file

@ -776,6 +776,50 @@ export class Brainy<T = any> implements BrainyInterface<T> {
private _pendingEmbedIds = new Set<string>()
private _embedWorkerFlight: Promise<void> | null = null
/**
* Ids cleared from {@link _pendingEmbedIds} with NO durable disarming record
* behind them — today exactly one case: a pending row that still EXISTS but
* carries no embeddable data, which the worker reaps in memory only. The log
* still says those ids are pending, so the pending-embed CHECKPOINT must
* carry them: the checkpoint's contract is "as of generation G the LOG's
* pending set was exactly this list", and a checkpoint that quietly dropped
* an id the log still arms would make the bounded fold disagree with a full
* fold from generation 1 — the one divergence that could lose a vector.
* Bounded by the number of such rows; an id leaves when it is re-enqueued or
* durably disarmed.
*/
private _pendingEmbedUndurableClears = new Set<string>()
/**
* Pending-set transitions (enqueue/clear) since the last checkpoint attempt —
* the checkpoint CADENCE. One mechanism, one hardcoded default, no knob and
* no timer (nothing to leave running after close).
*/
private _pendingEmbedCheckpointTransitions = 0
/**
* A checkpoint is OWED: the cadence came due (or the set drained) and no
* write has satisfied it yet. It stays armed across attempts the durability
* law refuses, so the next transition that CAN be checkpointed is.
*/
private _pendingEmbedCheckpointDue = false
/** Single-flight guard for the fire-and-forget checkpoint write. */
private _pendingEmbedCheckpointFlight: Promise<void> | null = null
/**
* What the last pending-embed recovery fold actually did — the bound it
* used, where it started, and how many facts it read. The narration's
* source, and the accounting a pin reads instead of a clock.
*/
private _pendingEmbedFoldReport: {
bound: 'checkpoint' | 'low-water' | 'genesis'
fromGeneration: number
factsScanned: number
seeded: number
pending: number
} | null = null
// OPEN-PATH FIX: the background embedding-engine warm kicked off (never
// awaited) by `performInit()` when `eagerEmbeddings` resolves true. Stored
// for observability only — `embed()`/`embeddingManager.embed()` already
@ -2434,6 +2478,47 @@ export class Brainy<T = any> implements BrainyInterface<T> {
*/
private static readonly PENDING_EMBED_LOWWATER_PATH = '_system/pending_embeds_lowwater.json'
/**
* Storage-root-relative path of the pending-embed CHECKPOINT:
* `{ generation, pending: string[], writtenAt }` — "as of durable generation
* G the pending set was exactly this list". Open seeds the set from `pending`
* and scans the log from `G + 1`, so the fold costs O(facts since G)
* REGARDLESS of whether the set ever drains.
*
* WHY IT REPLACES THE EMPTY-ONLY MARK AS THE BOUND. The low-water mark
* ({@link PENDING_EMBED_LOWWATER_PATH}) can only be written when the pending
* set is EMPTY, because it carries no set — it means "everything at or below
* G is consumed". A brain holding even ONE id that never lands (an embed that
* keeps failing; a row reaped in memory only and re-folded every open) never
* drains, so it never writes a mark, so the bound never engages on exactly
* the brains whose fold is expensive: every open re-reads the whole log. The
* checkpoint carries the set, so it needs no drain.
*
* The mark is still written and still read as the FALLBACK bound (a
* checkpoint that is absent, torn, or malformed degrades to it, and then to
* generation 1). Correctness over cost in every degradation: a stale or
* missing checkpoint only lengthens the scan.
*/
private static readonly PENDING_EMBED_CHECKPOINT_PATH = '_system/pending_embeds_checkpoint.json'
/**
* Checkpoint CADENCE BASE: attempt a checkpoint every N pending-set
* transitions (enqueues + clears) while the brain is open, on top of the
* drain-to-empty and clean-close writes. Hardcoded 90th-percentile default,
* no knob, no timer: 64 transitions is far below the cost of the fold it
* bounds and far above the per-write noise floor. An attempt that cannot
* satisfy the durability law is SKIPPED, not forced — the next transition
* retries.
*
* The interval ADAPTS to the one signal that matters, the backlog's own
* size, because a checkpoint writes the WHOLE pending list: the interval is
* `max(64, ceil(|pending| / 64))`, which holds the amortized cost of the
* mechanism at ≤ 64 ids written per transition NO MATTER how large the
* backlog grows. A term that scales with the store rather than with the
* work is exactly the defect class this file is fixing; it must not be
* reintroduced by the cure.
*/
private static readonly PENDING_EMBED_CHECKPOINT_EVERY = 64
/**
* @description Mark a deferred embed pending (MT5): the id joins the
@ -2448,6 +2533,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
*/
private enqueuePendingEmbed(id: string): FactMarkerRecord {
this._pendingEmbedIds.add(id)
// Re-armed for real: any earlier in-memory-only clear is superseded.
this._pendingEmbedUndurableClears.delete(id)
this.noteEmbedCheckpointCadence()
return { type: 'embed.pending', id, enqueuedAt: Date.now() }
}
@ -2458,11 +2546,27 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* fact) — the recovery fold consumes those; nothing here touches storage.
* One honest residue: a pending row whose entity still exists but carries
* no data is reaped in memory only, so it re-folds at the next open and
* is re-reaped there — a bounded no-op, never a lost vector.
* is re-reaped there — a bounded no-op, never a lost vector. That residue
* is the ONLY `durability: 'in-memory-only'` caller, and the checkpoint
* keeps carrying those ids so the bounded fold and a full fold from
* generation 1 agree exactly (see {@link _pendingEmbedUndurableClears}).
*
* @param id - The pending id to clear.
* @param durability - `'durable'` (default) when a record in the log at or
* below the current head disarms this id (an `embed.landed` riding the
* landing or unvector commit, or the row's tombstone — including the row
* simply not being there any more); `'in-memory-only'` when nothing in the
* log says so.
*/
private clearPendingEmbed(id: string): void {
private clearPendingEmbed(
id: string,
durability: 'durable' | 'in-memory-only' = 'durable'
): void {
this._pendingEmbedIds.delete(id)
if (durability === 'in-memory-only') this._pendingEmbedUndurableClears.add(id)
else this._pendingEmbedUndurableClears.delete(id)
if (this._pendingEmbedIds.size === 0) this.maybeWriteEmbedLowWater()
this.noteEmbedCheckpointCadence()
}
/**
@ -2498,6 +2602,223 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
}
/**
* @description Capture a pending-embed checkpoint, or refuse.
*
* THE DURABILITY LAW, satisfied by construction. The checkpoint asserts "as
* of generation G the log's pending set was exactly this list", and the next
* open TRUSTS it: it seeds the set and never reads a fact at or below G
* again. So a checkpoint may only be taken at a G whose facts are DURABLE.
* A checkpoint taken at head H while the facts up to H are still buffered
* would be read back after a crash that truncated the tail — and an
* `embed.landed` in a truncated fact would be gone from the log while the
* checkpoint still recorded its id as landed. The row's landing vector went
* with the truncated fact, so nothing would ever re-arm it: A LOST VECTOR.
*
* The gate is therefore `0 < head ≤ committed`. `committed` is the
* generation manifest's watermark — the point the store's own recovery
* treats as truth, and the point below which `FactLog.open()` never
* truncates — and the group-commit flush fsyncs the log BEFORE advancing it
* (see `GenerationStore.flushPendingSingleOps`). So every fact at or below
* `head` is fsynced and survives the crash exactly as the checkpoint
* describes it. Anything else (a head above the manifest, no log, no
* generation yet, a read-only or closed brain) REFUSES: skipping a
* checkpoint costs a longer scan next open, never a marker.
*
* The snapshot is taken SYNCHRONOUSLY with reading the two generations — no
* `await` between them — so no commit and no worker step can slip between
* "the generation I am about to claim" and "the set I claim for it".
*
* The one asymmetry, deliberately in the safe direction: an id whose
* `embed.pending` record has not been appended yet (enqueued in memory, its
* commit still in flight) is captured as pending at G although its marker
* will land at G+1 or later. Over-stating pending costs one idempotent
* re-embed attempt; under-stating it is the shape that loses a vector, and
* cannot happen — every clear either rides a durable record at or below the
* head, or is carried in {@link _pendingEmbedUndurableClears}.
*
* @returns The checkpoint payload, or `null` when this instant cannot host
* one.
*/
private captureEmbedCheckpoint(): { generation: number; pending: string[] } | null {
if (this.isReadOnly || this.closed) return null
const store = this.generationStore
if (!store) return null
const log = store.getFactLog()
if (!log) return null
// --- ONE SYNCHRONOUS INSTANT: no await until the return. ---
const generation = log.headGeneration()
const committed = store.committedGeneration()
if (!(generation > 0) || generation > committed) return null
const pending = new Set(this._pendingEmbedIds)
for (const id of this._pendingEmbedUndurableClears) pending.add(id)
// --- end of the synchronous instant. ---
return { generation, pending: [...pending] }
}
/**
* @description Fire-and-forget checkpoint write, single-flight: a burst of
* transitions never stacks writes, and because each attempt captures
* immediately before it writes, the file always ends up holding the most
* recently captured (generation, set) PAIR — and every such pair is
* independently true, so even an out-of-order landing is safe.
* {@link closeDurableSteps} awaits the flight before taking the final one.
*/
private maybeWriteEmbedCheckpoint(): void {
if (this._pendingEmbedCheckpointFlight) return
this._pendingEmbedCheckpointFlight = this.writeEmbedCheckpoint()
.then((wrote) => {
if (wrote) {
this._pendingEmbedCheckpointDue = false
this._pendingEmbedCheckpointTransitions = 0
}
})
.finally(() => {
this._pendingEmbedCheckpointFlight = null
})
}
/**
* The awaitable core of {@link maybeWriteEmbedCheckpoint}.
* @returns `true` when a checkpoint was actually written.
*/
private async writeEmbedCheckpoint(): Promise<boolean> {
const snapshot = this.captureEmbedCheckpoint()
if (!snapshot) return false
try {
// Atomic on disk: the filesystem adapter's writeRawObject is tmp+rename
// (see BaseStorage.writeRawObject), so a crash mid-write leaves either
// the previous checkpoint or the new one — never a spliced file. And a
// file that IS unreadable (a torn gzip, invalid JSON) throws typed on
// read and degrades to the fallback bound; it can never parse into a
// partial `pending` list.
//
// The file is NOT separately fsynced, and does not need to be: losing
// the rename to a power cut leaves the PREVIOUS checkpoint (or none),
// which only lengthens the next scan. The invariant that matters is the
// other direction — a checkpoint that IS visible names a generation
// whose facts are durable — and that is established by the capture gate
// above, not by this write.
await this.storage.writeRawObject(Brainy.PENDING_EMBED_CHECKPOINT_PATH, {
generation: snapshot.generation,
pending: snapshot.pending,
writtenAt: Date.now()
})
return true
} catch (err) {
prodLog.warn(
`[Brainy] pending-embed checkpoint write failed at generation ` +
`${snapshot.generation}: ${(err as Error).message} — the next open scans ` +
`from the previous checkpoint`
)
return false
}
}
/**
* @description The checkpoint cadence tick: count one pending-set transition
* and OWE a checkpoint every {@link PENDING_EMBED_CHECKPOINT_EVERY}
* transitions, plus on every drain to empty. The debt stays armed across
* attempts the durability law refuses — during a write burst the log head
* legitimately runs ahead of the manifest, so the first attempt often cannot
* be taken — and the next transition retries it. An active brain therefore
* checkpoints steadily without ever forcing a flush; an idle one relies on
* its clean close. No timer is involved, so nothing survives close().
*/
private noteEmbedCheckpointCadence(): void {
if (this.isReadOnly || this.closed) return
this._pendingEmbedCheckpointTransitions++
const listed = this._pendingEmbedIds.size + this._pendingEmbedUndurableClears.size
const every = Math.max(
Brainy.PENDING_EMBED_CHECKPOINT_EVERY,
Math.ceil(listed / Brainy.PENDING_EMBED_CHECKPOINT_EVERY)
)
if (
this._pendingEmbedIds.size === 0 ||
this._pendingEmbedCheckpointTransitions >= every
) {
this._pendingEmbedCheckpointDue = true
}
if (this._pendingEmbedCheckpointDue) this.maybeWriteEmbedCheckpoint()
}
/**
* @description Resolve the pending-embed fold's BOUND: the checkpoint first
* (a set plus a generation), then the legacy low-water mark (a generation
* only), then genesis. Every degradation is loud and lengthens the scan
* rather than shortening it — a bound that could skip a marker is never
* derived from a value this method could not fully validate.
* @returns The bound's name, the first generation to scan, and the ids to
* seed the pending set with.
*/
private async readPendingEmbedBound(): Promise<{
bound: 'checkpoint' | 'low-water' | 'genesis'
fromGeneration: number
seeded: string[]
}> {
let checkpointRejected: string | null = null
try {
const raw = await this.storage.readRawObject(Brainy.PENDING_EMBED_CHECKPOINT_PATH)
if (raw !== null && raw !== undefined) {
const parsed = Brainy.parsePendingEmbedCheckpoint(raw)
if (parsed) {
return {
bound: 'checkpoint',
fromGeneration: parsed.generation + 1,
seeded: parsed.pending
}
}
checkpointRejected = 'its shape is not { generation: number > 0, pending: string[] }'
}
} catch (err) {
// A real storage fault (EIO/EACCES/…). Corruption never lands here: the
// adapter maps a torn raw object to `null` AFTER logging it as a
// production error, so a torn checkpoint arrives as "absent" — loud at
// the adapter, and bounded here by the fallback below.
checkpointRejected = `reading it failed: ${(err as Error).message}`
}
if (checkpointRejected !== null) {
prodLog.warn(
`[Brainy] pending-embed checkpoint REFUSED (${checkpointRejected}) — falling back ` +
`to the low-water mark, else a full fold from generation 1`
)
}
try {
const mark = (await this.storage.readRawObject(Brainy.PENDING_EMBED_LOWWATER_PATH)) as {
generation?: number
} | null
if (mark && typeof mark.generation === 'number' && mark.generation > 0) {
return { bound: 'low-water', fromGeneration: mark.generation + 1, seeded: [] }
}
} catch {
// No mark (or unreadable): scan from 1 — correctness over cost.
}
return { bound: 'genesis', fromGeneration: 1, seeded: [] }
}
/**
* @description Validate a raw checkpoint object STRICTLY. Anything that is
* not exactly `{ generation: integer > 0, pending: string[] }` is refused
* whole — a partially-usable checkpoint is the one shape that could seed a
* short pending set behind a high bound, which is how a vector is lost.
* @param raw - The object read back from storage.
* @returns The validated checkpoint, or `null`.
*/
private static parsePendingEmbedCheckpoint(
raw: unknown
): { generation: number; pending: string[] } | null {
if (raw === null || typeof raw !== 'object' || Array.isArray(raw)) return null
const { generation, pending } = raw as { generation?: unknown; pending?: unknown }
if (typeof generation !== 'number' || !Number.isSafeInteger(generation) || generation <= 0) {
return null
}
if (!Array.isArray(pending) || pending.some((id) => typeof id !== 'string' || id === '')) {
return null
}
return { generation, pending: pending as string[] }
}
/**
* @description Rebuild the pending-embed set by REPLAYING the generation
* log's marker records (recovery = replay, not listing): `embed.pending`
@ -2506,14 +2827,22 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* survives the fold is exactly the set of acknowledged deferred writes
* whose vectors have not landed.
*
* BOUND: the scan starts at the advisory low-water mark
* ({@link Brainy.PENDING_EMBED_LOWWATER_PATH}) — the log head at which the
* pending set last drained to empty — so a settled brain reads only the
* facts since then, not its whole history. Without a mark (first open
* after upgrade) it scans from generation 1, once; a stale-low mark costs
* a longer scan, never a marker. The fold stays on the open's foreground —
* the crash-recovery contract pins that a reopened brain has its markers
* re-armed when open() returns — and the mark is what makes that cheap.
* BOUND: the scan starts after the pending-embed CHECKPOINT
* ({@link Brainy.PENDING_EMBED_CHECKPOINT_PATH}) — "as of durable generation
* G the pending set was exactly this list" — so the fold seeds the set from
* that list and reads only the facts after G. O(delta) whether or not the
* set ever drains, which is the whole point: the previous bound, the
* empty-only low-water mark, could not be written at all by a brain holding
* one id that never lands, so those brains re-read their whole log at every
* open. The mark remains the FALLBACK bound (checkpoint absent, torn, or
* malformed), and generation 1 the fallback below that — a brain opened for
* the first time after this change has neither a checkpoint nor, if it never
* drained, a mark, so it pays one full fold and writes a checkpoint on the
* way out. A stale bound costs a longer scan, never a marker. The fold stays
* on the open's foreground — the crash-recovery contract pins that a
* reopened brain has its markers re-armed when open() returns — and the
* bound is what makes that cheap. What it did (bound, start, facts read) is
* narrated and kept in {@link _pendingEmbedFoldReport}.
* It is SKIPPED WHOLESALE when the log has never had a v2 tail
* ({@link FactLog.hasV2History} — v1 facts cannot carry marker records),
* so pre-cutover brains pay nothing; on a mixed log the scan still reads
@ -2526,20 +2855,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
private async recoverPendingEmbedsFromLog(): Promise<void> {
const log = this.generationStore.getFactLog()
if (!log || !log.hasV2History()) return
let fromGeneration = 1
try {
const mark = (await this.storage.readRawObject(Brainy.PENDING_EMBED_LOWWATER_PATH)) as {
generation?: number
} | null
if (mark && typeof mark.generation === 'number' && mark.generation > 0) {
fromGeneration = mark.generation + 1
}
} catch {
// No mark (or unreadable): scan from 1 — correctness over cost.
}
const { bound, fromGeneration, seeded } = await this.readPendingEmbedBound()
for (const id of seeded) this._pendingEmbedIds.add(id)
let factsScanned = 0
const scan = log.scanFacts({ fromGeneration })
for await (const batch of scan.batches()) {
for (const fact of batch.facts) {
factsScanned++
for (const record of fact.records ?? []) {
if (record.type === 'embed.pending') {
this._pendingEmbedIds.add(record.id)
@ -2554,6 +2876,21 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
}
}
this._pendingEmbedFoldReport = {
bound,
fromGeneration,
factsScanned,
seeded: seeded.length,
pending: this._pendingEmbedIds.size
}
// The narration channel: an operator is entitled to hear which bound
// applied and what it cost, on every open — that is how a bound that
// silently stopped engaging (the defect this replaced) becomes visible.
prodLog.narrate(
`[Brainy] pending-embed fold: ${bound} bound → scanned ${factsScanned} fact(s) ` +
`from generation ${fromGeneration}, seeded ${seeded.length} id(s), ` +
`${this._pendingEmbedIds.size} pending`
)
}
/**
@ -2645,11 +2982,23 @@ export class Brainy<T = any> implements BrainyInterface<T> {
for (const id of batch) {
try {
const entity = await this.get(id, { includeVectors: true })
if (!entity || entity.data === undefined || entity.data === null) {
// Orphan reap: a deleted row's tombstone fact durably disarms the
// marker at the next recovery fold; a data-less-but-present row
// (edge case) re-folds and re-reaps — bounded, never a lost vector.
this.clearPendingEmbed(id)
if (!entity) {
// The row is GONE. Either it was deleted — its tombstone fact
// durably disarms the marker, at or below the head, exactly as the
// fold reads it — or its create never became durable, in which case
// the log carries no `embed.pending` for it either. Both are durable
// clears: a full fold from generation 1 reaches the same answer.
this.clearPendingEmbed(id, 'durable')
continue
}
if (entity.data === undefined || entity.data === null) {
// Orphan reap, IN MEMORY ONLY: a data-less-but-present row (edge
// case) has nothing to embed, but no record in the log says so, so
// the fold would re-arm it. Cleared here and carried in the
// checkpoint (see clearPendingEmbed) — it re-folds and re-reaps at
// the next open exactly as before: bounded, never a lost vector,
// and never a checkpoint that disagrees with the log.
this.clearPendingEmbed(id, 'in-memory-only')
continue
}
// Hang guard: a wedged embedder must not block every later pending
@ -19982,6 +20331,21 @@ export class Brainy<T = any> implements BrainyInterface<T> {
await this.stampEntityTree()
}
// Phase 1c: the pending-embed CHECKPOINT — placed HERE and not earlier
// because this is the first point in the close where the durability law it
// must satisfy actually holds: `generationStore.close()` (in Phase 1 above)
// flushed the pending single-op tier, which fsyncs the fact log and then
// advances the manifest, so `head === committed` and every fact the
// checkpoint's generation covers is durable. Taken even when the set is
// NOT empty — that is the whole difference from the low-water mark, and it
// is what makes the next open's fold O(facts since this close) on a brain
// whose pending set never drains. Awaits any in-flight cadence write first
// so the last write to the file is this one.
if (!this.isReadOnly) {
await this._pendingEmbedCheckpointFlight?.catch(() => {})
await this.writeEmbedCheckpoint()
}
// Phase 2: Close components to release resources (timers, file handles)
// Data is already safe on disk from Phase 1
await Promise.all([