David Snelling
bc70c43d02
Some checks failed
Delta Gate / Delta gate — candidate vs control (push) Waiting to run
CI / Node 22 (push) Successful in 12m27s
CI / Node 24 (push) Successful in 12m29s
CI / Integration + conformance (Node 22) (push) Failing after 17m33s
CI / Bun (latest) (push) Successful in 12m28s
Every log-authority open asks the fact log one question — is there a fact
above the committed pointer? — and answered it by reading and CRC-decoding
EVERY segment file the manifest names. MEASURED in production on a 16k-row
brain at generation ~478,819: 34-37 seconds inside `generation-store-open-fold`
on every open, including the clean one where the answer is always "nothing".
The manifest already knows. A sealed segment's `lastGeneration` is written at
seal time, and the seal order has been the same since the log was introduced:
`rotate()` fsyncs the tail's bytes FIRST ("sealed segments are always fully
durable"), builds the entry from the content that fsync covered, and only then
flips the manifest — atomically, fsynced, and in the same write re-pointing
`tailSegment`, so a sealed file is never appended to again. A crash in that
order is safe in the pruning direction: before the manifest write the segment
is still the TAIL and is read whole; after it, the entry describes bytes that
were already durable. The only later mutation of a sealed segment is open()'s
straddle truncation, which removes facts and re-derives the entry from the
actual bytes — a recorded bound can drift DOWN with its file, never up.
So `lastGeneration = L` proves the file holds no fact above L, and both
manifest-direct passes (`peekFactsAbove` and its streaming twin, the recovery
fold) now read only the unsealed tail, entries with no numeric
`lastGeneration` — legacy or hand-repaired manifests, never prune what you
cannot prove — and entries whose recorded maximum is actually above the bound.
The open narrates what it read and what it pruned when the log holds more than
one segment.
Pinned in tests/integration/factlog-open-prune.test.ts, from the log's own
counters rather than a clock: a clean reopen over five sealed segments reads
exactly the tail (1 of 6) and finds nothing; a real SIGKILLed writer that
sealed segments holding facts above the committed pointer has those segments
READ, and its peek, its fold stream and its rollback all match the unpruned
full scan fact for fact; a manifest entry missing `lastGeneration` is read.
1613 lines
67 KiB
TypeScript
1613 lines
67 KiB
TypeScript
/**
|
|
* @module db/factLog
|
|
* @description The generation FACT LOG — an append-only, CRC-framed record of
|
|
* every committed generation as an AFTER-IMAGE "fact": what each touched
|
|
* entity/relationship BECAME (or a body-less tombstone when it was removed).
|
|
* This is the dual-write half of the log-canonical transition: today the
|
|
* before-image history + canonical tree remain authoritative; the fact log is
|
|
* appended at the same commit points and reconciled to committed truth at
|
|
* open, so consumers (index heals, replays, scans) can read one sequential,
|
|
* self-verifying stream instead of walking the entity tree.
|
|
*
|
|
* ## Wire format (frozen; additive-only within a major)
|
|
*
|
|
* Fact (msgpack, POSITIONAL array — the segment header's formatVersion
|
|
* governs the schema):
|
|
*
|
|
* fact := [ generation:u64, timestamp:u64, ops, meta|nil, blobHashes|nil ]
|
|
* op := [ kind:u8 (0=noun, 1=verb), id:bin16 (raw uuid bytes),
|
|
* record:[metaLeg, vecLeg] | nil ] // nil = TOMBSTONE
|
|
*
|
|
* Segment file (`_generations/facts/seg-<firstGeneration, zero-padded 20>.bfl`):
|
|
*
|
|
* header := magic "BFACTS\0\0" (8B) | formatVersion:u32 LE |
|
|
* firstGeneration:u64 LE | reserved 12B (ZEROED, verified)
|
|
* frame := length:u32 LE | crc32c:u32 LE (of payload) | payload
|
|
*
|
|
* A fact is never split across segments; a torn tail (length overrun or CRC
|
|
* mismatch) terminates that segment's scan — everything before it is intact.
|
|
* Zero-padded names make lexicographic order == generation order.
|
|
*
|
|
* ## Invariant
|
|
*
|
|
* After {@link FactLog.open}, the log contains EXACTLY the committed prefix:
|
|
* facts are appended BEFORE the commit point (inside the same durability
|
|
* window), so a crash can only leave the log AHEAD of committed truth — open
|
|
* truncates any fact beyond the committed generation. Absent generation =
|
|
* never committed; present = committed. A scan can never see an uncommitted
|
|
* fact.
|
|
*
|
|
* The manifest (`_generations/facts/manifest.json`, JSON — forensics stay
|
|
* terminal-readable) is the single source of truth for the segment SET;
|
|
* rotation flips it atomically (write-new → fsync → rename) BEFORE the new
|
|
* tail's first byte exists, so no segment file is ever unaccounted for. Its
|
|
* per-segment `firstGeneration`/`lastGeneration` are LOAD-BEARING at open: a
|
|
* recovery pass looking for facts above a bound reads only the segments those
|
|
* bounds cannot rule out (the prune law — see `segmentsHoldingFactsAbove`).
|
|
*
|
|
* ## Mixed-version logs (the v2 live-write cutover)
|
|
*
|
|
* The segment header's `formatVersion` selects the decoder PER SEGMENT:
|
|
* v1 segments (ops-shaped facts, the format above) stay readable forever and
|
|
* are NEVER rewritten; a NEW tail segment writes the v2 format
|
|
* (`src/db/factLogFormat.ts` — record envelope, minted dense ints, genesis,
|
|
* sector seals) whenever the int minter is installed ({@link FactLog.setIntMinter} —
|
|
* the brain wires it from the metadata index's id mapper right after init).
|
|
* A bare `FactLog` with no minter keeps writing v1 (there is no authority
|
|
* that could reproduce int assignments, and 0 is never written). Cutover
|
|
* mechanics on an existing v1 log: an EMPTY v1 tail is re-headed to v2 in
|
|
* place; a non-empty v1 tail is sealed by an immediate rotation and the new
|
|
* tail is v2. Decoded v2 facts map back to the SAME {@link CommitFact} shape
|
|
* v1 consumers read (noun/verb ops with `{metadata, vector} | null` records) —
|
|
* the vector wrapper object is reconstructed from the record's metadata leg
|
|
* through the reserved-field hydration law (see `commitFactFromV2`).
|
|
*
|
|
* V2 tails additionally: write the `log.genesis` record (id-space width 64 +
|
|
* the brain id, minted once into the manifest's additive `brainId` field) as
|
|
* the first record of the FIRST fact of a brand-new log, and seal every
|
|
* `sync()` to the header-declared sector size with pad frames that are
|
|
* invisible to readers (torn-page defense at group-commit boundaries).
|
|
*/
|
|
import { encode as defaultEncode, decode as defaultDecode } from '@msgpack/msgpack'
|
|
import { crc32c } from '../utils/crc32c.js'
|
|
import { prodLog } from '../utils/logger.js'
|
|
import {
|
|
FACT_LOG_FORMAT_V1,
|
|
FACT_LOG_FORMAT_V2,
|
|
DEFAULT_SEAL_SIZE,
|
|
parseSegmentHeader,
|
|
encodeSegmentHeaderV2,
|
|
encodeFactV2,
|
|
decodeFact as decodeFormatFact,
|
|
decodeGroupV2,
|
|
encodePadFrame,
|
|
minPadFrameBytes,
|
|
type CommitFactV2,
|
|
type LogRecord,
|
|
type EmbedPendingRecord,
|
|
type EmbedLandedRecord,
|
|
type BlobManifestRecord,
|
|
type BootstrapBaselineRecord,
|
|
type ProjectionNoteRecord
|
|
} from './factLogFormat.js'
|
|
import {
|
|
splitNounMetadataRecord
|
|
} from '../types/reservedFields.js'
|
|
import { NounType } from '../types/graphTypes.js'
|
|
import { v4 as uuidv4 } from '../universal/uuid.js'
|
|
|
|
// Swappable msgpack implementation — defaults to the JS codec; a native
|
|
// provider (registered via the plugin registry's 'msgpack' key) may replace
|
|
// it. Byte-compatibility is the contract (positional arrays, bin16 ids).
|
|
let msgpackEncode: (value: unknown) => Uint8Array = defaultEncode
|
|
let msgpackDecode: (bytes: Uint8Array) => unknown = defaultDecode
|
|
|
|
/** Replace the msgpack encode/decode implementation at runtime. */
|
|
export function setFactCodec(impl: {
|
|
encode: (value: unknown) => Uint8Array
|
|
decode: (bytes: Uint8Array) => unknown
|
|
}): void {
|
|
msgpackEncode = impl.encode
|
|
msgpackDecode = impl.decode
|
|
}
|
|
|
|
/** Storage-root-relative home of the fact log. */
|
|
export const FACTS_PREFIX = '_generations/facts'
|
|
/** The facts manifest path (JSON). */
|
|
export const FACTS_MANIFEST_PATH = `${FACTS_PREFIX}/manifest.json`
|
|
/**
|
|
* The v1 segment format version — the MANIFEST's formatVersion gate and the
|
|
* header value of v1 (minter-less) tails. NOT the live-write ceiling: new
|
|
* tails write `FACT_LOG_FORMAT_V2` (src/db/factLogFormat.ts) whenever the
|
|
* int minter is installed; both versions are read forever, per segment.
|
|
*/
|
|
export const FACTS_FORMAT_VERSION = 1
|
|
/** Rotation threshold: seal the tail segment once it exceeds this many bytes. */
|
|
const SEGMENT_ROTATE_BYTES = 8 * 1024 * 1024
|
|
/** Segment header: magic(8) + formatVersion(4) + firstGeneration(8) + reserved(12). */
|
|
const HEADER_BYTES = 32
|
|
const MAGIC = new Uint8Array([0x42, 0x46, 0x41, 0x43, 0x54, 0x53, 0x00, 0x00]) // "BFACTS\0\0"
|
|
/** Frame prefix: length(4) + crc32c(4). */
|
|
const FRAME_PREFIX_BYTES = 8
|
|
|
|
/** One write inside a fact: what the id became (or a tombstone). */
|
|
export interface FactOp {
|
|
kind: 'noun' | 'verb'
|
|
id: string
|
|
/** The AFTER-IMAGE legs, or `null` for a tombstone (the id was removed). */
|
|
record: { metadata: unknown | null; vector: unknown | null } | null
|
|
}
|
|
|
|
/**
|
|
* V2-native records beyond noun/verb ops that a fact may carry through the
|
|
* ENCODER (types 6/7/8/9/10 of the v2 registry: embed markers, blob
|
|
* manifests, projection notes, bootstrap baselines). The deferred-embedding
|
|
* lifecycle PRODUCES types 6/7 today: `embed.pending` rides the deferred
|
|
* write's own commit fact and `embed.landed` rides the background worker's
|
|
* landing commit (recovery folds the pair back out of the log at open). The
|
|
* blob lifecycle remodels onto type 8 in a later leg.
|
|
*/
|
|
export type FactMarkerRecord =
|
|
| EmbedPendingRecord
|
|
| EmbedLandedRecord
|
|
| BlobManifestRecord
|
|
| ProjectionNoteRecord
|
|
| BootstrapBaselineRecord
|
|
|
|
/**
|
|
* Mints the dense integer handle for an entity/verb id at fact-append time —
|
|
* REQUIRED to be reproducible: a rebuilt id mapper must reproduce the same
|
|
* assignments exactly, so the only legal implementation delegates to the
|
|
* metadata index's id mapper (`getOrAssign`). Returns a POSITIVE bigint; a
|
|
* minter that cannot resolve its mapper throws — an int of 0 is never written.
|
|
*/
|
|
export type FactIntMinter = (kind: 'noun' | 'verb', id: string) => bigint
|
|
|
|
/** One committed generation, as scanned back out of the log. */
|
|
export interface CommitFact {
|
|
generation: number
|
|
timestamp: number
|
|
ops: FactOp[]
|
|
meta?: Record<string, unknown>
|
|
blobHashes?: string[]
|
|
/**
|
|
* V2-native marker records riding this fact (see {@link FactMarkerRecord}).
|
|
* Optional and additive: absent on every v1 fact and on every fact the
|
|
* current writers produce; requires a v2 tail to encode.
|
|
*/
|
|
records?: FactMarkerRecord[]
|
|
}
|
|
|
|
/** The telemetry a scan batch carries (frozen shape). */
|
|
export interface FactScanBatch {
|
|
facts: CommitFact[]
|
|
firstGeneration: number
|
|
lastGeneration: number
|
|
factCount: number
|
|
byteSize: number
|
|
segmentId: string
|
|
}
|
|
|
|
/**
|
|
* Liveness bound on a scan's FIRST batch (Stage-2 co-freeze, D1 contract):
|
|
* `batches()` must yield its first batch — or fail loudly — within this many
|
|
* ms of the first pull. A backlogged or damaged store may be SLOW, but it may
|
|
* never be SILENT: a consumer awaiting the first batch is otherwise
|
|
* indistinguishable from a wedge (the exact failure shape a production heal
|
|
* hit against a generations-backlogged brain).
|
|
*/
|
|
export const SCANFACTS_FIRST_BATCH_MS = 10_000
|
|
|
|
/** The telemetry a scan OPEN returns (frozen shape). */
|
|
export interface FactScanHandle {
|
|
headGeneration: number
|
|
segmentCount: number
|
|
approxFactCount: number
|
|
/**
|
|
* Ordered batches; a detected gap aborts LOUDLY, never a silent skip.
|
|
* Liveness contract: the FIRST batch resolves or rejects within
|
|
* {@link SCANFACTS_FIRST_BATCH_MS} of the first pull — never a silent hang.
|
|
*/
|
|
batches: () => AsyncGenerator<FactScanBatch>
|
|
/** Close telemetry — the invariant cross-check, valid after iteration ends. */
|
|
summary: () => { factsYielded: number; segmentsRead: number }
|
|
}
|
|
|
|
/** Manifest entry for a sealed segment. */
|
|
interface SegmentEntry {
|
|
file: string
|
|
firstGeneration: number
|
|
lastGeneration: number
|
|
facts: number
|
|
bytes: number
|
|
}
|
|
|
|
/** The facts manifest (JSON on disk). */
|
|
interface FactsManifest {
|
|
formatVersion: number
|
|
segments: SegmentEntry[]
|
|
/** The append target. Its true content is established by scanning (crash tolerance). */
|
|
tailSegment: string | null
|
|
updatedAt: string
|
|
/**
|
|
* This brain's stable id (additive, v2 cutover): minted as a uuid at the
|
|
* first v2 tail creation and never changed; the `log.genesis` record
|
|
* carries it. Absent on logs that have never had a v2 tail.
|
|
*/
|
|
brainId?: string
|
|
}
|
|
|
|
/** The narrow byte-level storage surface the fact log rides. */
|
|
export interface FactLogStorage {
|
|
appendRawBytes(path: string, bytes: Uint8Array): Promise<void>
|
|
readRawBytes(path: string): Promise<Uint8Array | null>
|
|
writeRawBytes(path: string, bytes: Uint8Array): Promise<void>
|
|
rawByteSize(path: string): Promise<number | null>
|
|
readRawObject(path: string): Promise<any | null>
|
|
writeRawObject(path: string, data: any): Promise<void>
|
|
syncRawObjects(paths: string[]): Promise<void>
|
|
deleteRawObject(path: string): Promise<void>
|
|
}
|
|
|
|
/** True when the storage adapter exposes every primitive the fact log needs. */
|
|
export function storageSupportsFactLog(storage: unknown): storage is FactLogStorage {
|
|
const s = storage as Record<string, unknown>
|
|
return (
|
|
typeof s.appendRawBytes === 'function' &&
|
|
typeof s.readRawBytes === 'function' &&
|
|
typeof s.writeRawBytes === 'function' &&
|
|
typeof s.rawByteSize === 'function'
|
|
)
|
|
}
|
|
|
|
/** uuid string → 16 raw bytes (bin16 on the wire). */
|
|
function uuidToBytes(id: string): Uint8Array {
|
|
const hex = id.replace(/-/g, '')
|
|
if (hex.length !== 32) {
|
|
// Non-uuid ids (legacy/natural keys) ride as UTF-8 with a length prefix
|
|
// marker impossible for uuids: we refuse instead — the write API has
|
|
// guaranteed uuid ids since 8.0, so anything else is a corruption signal.
|
|
throw new Error(`fact log: id is not a uuid: ${id}`)
|
|
}
|
|
const bytes = new Uint8Array(16)
|
|
for (let i = 0; i < 16; i++) {
|
|
bytes[i] = parseInt(hex.slice(i * 2, i * 2 + 2), 16)
|
|
}
|
|
return bytes
|
|
}
|
|
|
|
/** 16 raw bytes → canonical lowercase uuid string. */
|
|
function bytesToUuid(bytes: Uint8Array): string {
|
|
let hex = ''
|
|
for (let i = 0; i < 16; i++) hex += bytes[i].toString(16).padStart(2, '0')
|
|
return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`
|
|
}
|
|
|
|
/** Zero-padded segment filename: lexicographic order == generation order. */
|
|
function segmentFileName(firstGeneration: number): string {
|
|
return `seg-${String(firstGeneration).padStart(20, '0')}.bfl`
|
|
}
|
|
|
|
/** Build a segment header. Reserved bytes are ZEROED (and verified on open). */
|
|
function buildHeader(firstGeneration: number): Uint8Array {
|
|
const header = new Uint8Array(HEADER_BYTES)
|
|
header.set(MAGIC, 0)
|
|
const view = new DataView(header.buffer)
|
|
view.setUint32(8, FACTS_FORMAT_VERSION, true)
|
|
view.setBigUint64(12, BigInt(firstGeneration), true)
|
|
// bytes 20..31 stay zero (reserved)
|
|
return header
|
|
}
|
|
|
|
/** Encode one fact into a framed record (length + crc32c + msgpack payload). */
|
|
function encodeFrame(fact: CommitFact): Uint8Array {
|
|
const payload = msgpackEncode([
|
|
fact.generation,
|
|
fact.timestamp,
|
|
fact.ops.map((op) => [
|
|
op.kind === 'noun' ? 0 : 1,
|
|
uuidToBytes(op.id),
|
|
op.record === null ? null : [op.record.metadata, op.record.vector]
|
|
]),
|
|
fact.meta ?? null,
|
|
fact.blobHashes && fact.blobHashes.length > 0 ? fact.blobHashes : null
|
|
])
|
|
const frame = new Uint8Array(FRAME_PREFIX_BYTES + payload.length)
|
|
const view = new DataView(frame.buffer)
|
|
view.setUint32(0, payload.length, true)
|
|
view.setUint32(4, crc32c(payload), true)
|
|
frame.set(payload, FRAME_PREFIX_BYTES)
|
|
return frame
|
|
}
|
|
|
|
/** Decode one msgpack payload back into a CommitFact. */
|
|
function decodeFact(payload: Uint8Array): CommitFact {
|
|
const raw = msgpackDecode(payload) as unknown[]
|
|
const [generation, timestamp, ops, meta, blobHashes] = raw as [
|
|
number,
|
|
number,
|
|
Array<[number, Uint8Array, [unknown, unknown] | null]>,
|
|
Record<string, unknown> | null,
|
|
string[] | null
|
|
]
|
|
return {
|
|
generation: Number(generation),
|
|
timestamp: Number(timestamp),
|
|
ops: ops.map(([kind, idBytes, record]) => ({
|
|
kind: kind === 0 ? ('noun' as const) : ('verb' as const),
|
|
id: bytesToUuid(idBytes),
|
|
record: record === null ? null : { metadata: record[0] ?? null, vector: record[1] ?? null }
|
|
})),
|
|
...(meta ? { meta } : {}),
|
|
...(blobHashes && blobHashes.length > 0 ? { blobHashes } : {})
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Deep-normalize a decoded v2 JSON position (metadata legs, meta maps,
|
|
* notes) back to plain-JSON values: the v2 codec decodes msgpack int64/uint64
|
|
* as `bigint` (its u64 wire discipline), but canonical records are JSON — a
|
|
* metadata timestamp like `createdAt: 1786…` must come back as the NUMBER it
|
|
* was encoded from. Safe-range bigints narrow exactly; anything beyond the
|
|
* safe-integer range in a JSON position refuses loudly (it cannot have come
|
|
* from a JSON write).
|
|
*/
|
|
function normalizeWireJson(value: unknown): unknown {
|
|
if (typeof value === 'bigint') {
|
|
if (value > BigInt(Number.MAX_SAFE_INTEGER) || value < -BigInt(Number.MAX_SAFE_INTEGER)) {
|
|
throw new Error(
|
|
`fact log v2: decoded integer ${value} exceeds the JS safe-integer range in a JSON position`
|
|
)
|
|
}
|
|
return Number(value)
|
|
}
|
|
if (Array.isArray(value)) return value.map(normalizeWireJson)
|
|
if (value && typeof value === 'object' && !(value instanceof Uint8Array)) {
|
|
const out: Record<string, unknown> = {}
|
|
for (const [k, v] of Object.entries(value)) out[k] = normalizeWireJson(v)
|
|
return out
|
|
}
|
|
return value
|
|
}
|
|
|
|
/**
|
|
* JSON-serialization equivalence for a v2 ENCODE-side JSON position: drop
|
|
* undefined-valued object keys and map undefined array elements to null —
|
|
* exactly what `JSON.stringify` does when canonical records are persisted.
|
|
* Commit facts are built from write-cache-WARM objects that may still carry
|
|
* undefined-valued engine keys (`service: undefined`, …) which the durable
|
|
* JSON never had; msgpack would preserve them as nil (the v1 capture's known
|
|
* wart), so the v2 capture — the future storage authority — sanitizes to the
|
|
* DURABLE truth instead.
|
|
*/
|
|
function toJsonSafe(value: unknown): unknown {
|
|
if (value === undefined) return null
|
|
if (Array.isArray(value)) return value.map((v) => (v === undefined ? null : toJsonSafe(v)))
|
|
if (value && typeof value === 'object' && !(value instanceof Uint8Array)) {
|
|
const out: Record<string, unknown> = {}
|
|
for (const [k, v] of Object.entries(value)) {
|
|
if (v === undefined) continue
|
|
out[k] = toJsonSafe(v)
|
|
}
|
|
return out
|
|
}
|
|
return value
|
|
}
|
|
|
|
/** Mirror of the storage layer's stored-timestamp normalization, minus its
|
|
* `Date.now()` fallback (a DECODER must be deterministic — an unreadable
|
|
* timestamp is omitted, and the divergence surfaces via the oracle). */
|
|
function reconstructTimestamp(value: unknown): number | undefined {
|
|
if (typeof value === 'number' && value > 0) return value
|
|
if (
|
|
value !== null &&
|
|
typeof value === 'object' &&
|
|
typeof (value as { seconds?: unknown }).seconds === 'number'
|
|
) {
|
|
return (value as { seconds: number }).seconds * 1000
|
|
}
|
|
return undefined
|
|
}
|
|
|
|
/**
|
|
* Rebuild a noun's canonical VECTOR-FILE wrapper from a v2 after-image —
|
|
* the read-side of the hydration law. Canonical noun vector files hold the
|
|
* denormalized enumerable entity (`{id, vector, connections, level, type,
|
|
* …reserved fields…, metadata}` — the write path's composition); the v2
|
|
* record deliberately carries only the ENTITY state (metadata leg + embedding
|
|
* floats), because connections/level are derived HNSW residue with their own
|
|
* rebuild paths (empty in every 8.x write) and the denormalized top-level
|
|
* fields are projections of the metadata leg. This reconstruction applies
|
|
* the SAME split/hydrate law the storage layer uses
|
|
* (`splitNounMetadataRecord` — the single source of truth in
|
|
* src/types/reservedFields.ts; field map mirrors
|
|
* `BaseStorage.hydrateNounWithMetadata`, undefined keys omitted exactly as
|
|
* JSON serialization omits them), so in the no-drift case the reconstructed
|
|
* wrapper digests byte-equal to canonical. A drifted denormalized copy
|
|
* surfaces as an oracle `state-differs` — named, never silently absorbed.
|
|
*/
|
|
export function reconstructNounWrapper(
|
|
id: string,
|
|
metadataLeg: unknown,
|
|
floats: number[]
|
|
): Record<string, unknown> {
|
|
const { reserved, custom } = splitNounMetadataRecord(
|
|
(metadataLeg ?? null) as Record<string, unknown> | null
|
|
)
|
|
const wrapper: Record<string, unknown> = {
|
|
id,
|
|
vector: floats,
|
|
connections: {},
|
|
level: 0,
|
|
type: (reserved.noun as string) || NounType.Thing
|
|
}
|
|
if (reserved.subtype !== undefined) wrapper.subtype = reserved.subtype
|
|
if (reserved.visibility !== undefined) wrapper.visibility = reserved.visibility
|
|
const createdAt = reconstructTimestamp(reserved.createdAt)
|
|
if (createdAt !== undefined) wrapper.createdAt = createdAt
|
|
const updatedAt = reconstructTimestamp(reserved.updatedAt)
|
|
if (updatedAt !== undefined) wrapper.updatedAt = updatedAt
|
|
if (reserved.confidence !== undefined) wrapper.confidence = reserved.confidence
|
|
if (reserved.weight !== undefined) wrapper.weight = reserved.weight
|
|
if (reserved.service !== undefined) wrapper.service = reserved.service
|
|
if (reserved.data !== undefined) wrapper.data = reserved.data
|
|
if (reserved.createdBy !== undefined) wrapper.createdBy = reserved.createdBy
|
|
wrapper._rev = typeof reserved._rev === 'number' ? reserved._rev : 1
|
|
wrapper.metadata = custom
|
|
return wrapper
|
|
}
|
|
|
|
/** Coerce a candidate embedding to `number[]`: plain arrays pass through
|
|
* (element-checked); numeric typed arrays (the JS HNSW rebuild path stores
|
|
* `Float32Array` vectors on the memory adapter) widen via `Array.from`. */
|
|
function floatsOf(candidate: unknown, context: string): number[] | undefined {
|
|
if (Array.isArray(candidate)) {
|
|
for (const el of candidate) {
|
|
if (typeof el !== 'number') {
|
|
throw new Error(`fact log v2: ${context} vector carries a non-number element`)
|
|
}
|
|
}
|
|
return candidate as number[]
|
|
}
|
|
if (ArrayBuffer.isView(candidate) && !(candidate instanceof DataView)) {
|
|
return Array.from(candidate as unknown as ArrayLike<number>)
|
|
}
|
|
return undefined
|
|
}
|
|
|
|
/** Extract the embedding float array from a canonical vector value: a bare
|
|
* float array (or numeric typed array) passes through; a wrapper object
|
|
* yields its `vector` floats; `null` stays `null`; anything else refuses
|
|
* loudly. */
|
|
function embeddingLegOf(value: unknown, context: string): number[] | null {
|
|
if (value === null || value === undefined) return null
|
|
const direct = floatsOf(value, context)
|
|
if (direct !== undefined) return direct
|
|
if (typeof value === 'object') {
|
|
const nested = floatsOf((value as { vector?: unknown }).vector, context)
|
|
if (nested !== undefined) return nested
|
|
}
|
|
throw new Error(
|
|
`fact log v2: ${context} has a canonical vector record with no float vector — ` +
|
|
`cannot encode its after-image`
|
|
)
|
|
}
|
|
|
|
/**
|
|
* Map one decoded v2 fact to the {@link CommitFact} shape every consumer
|
|
* already reads: noun/verb after-images and tombstones become ops (vector
|
|
* wrappers reconstructed — see {@link reconstructNounWrapper}); a
|
|
* `batch.meta` record becomes `meta` when the fact position carries none;
|
|
* `log.genesis` is log-level metadata (its width was verified at decode) and
|
|
* is not an op; marker records surface on the additive `records` field so
|
|
* nothing is silently dropped. Decoded JSON positions are normalized back
|
|
* from the codec's bigint discipline ({@link normalizeWireJson}).
|
|
*/
|
|
function commitFactFromV2(f: CommitFactV2): CommitFact {
|
|
const ops: FactOp[] = []
|
|
const markers: FactMarkerRecord[] = []
|
|
let batchMeta: Record<string, unknown> | undefined
|
|
for (const r of f.records) {
|
|
switch (r.type) {
|
|
case 'noun.afterImage': {
|
|
const metadata = normalizeWireJson(r.metadata) ?? null
|
|
let vector: unknown | null = null
|
|
if (r.vectorLeg !== null) {
|
|
if (!Array.isArray(r.vectorLeg)) {
|
|
throw new Error(
|
|
`fact log v2: noun.afterImage ${r.id} carries a vector ref — this reader ` +
|
|
`resolves inline vectors only (refs are a later leg); refusing`
|
|
)
|
|
}
|
|
vector = reconstructNounWrapper(r.id, metadata, r.vectorLeg)
|
|
}
|
|
ops.push({ kind: 'noun', id: r.id, record: { metadata, vector } })
|
|
break
|
|
}
|
|
case 'noun.tombstone':
|
|
ops.push({ kind: 'noun', id: r.id, record: null })
|
|
break
|
|
case 'verb.afterImage': {
|
|
const metadata = normalizeWireJson(r.metadata) ?? null
|
|
if (r.vectorLeg !== null && !Array.isArray(r.vectorLeg)) {
|
|
throw new Error(
|
|
`fact log v2: verb.afterImage ${r.id} carries a vector ref — this reader ` +
|
|
`resolves inline vectors only (refs are a later leg); refusing`
|
|
)
|
|
}
|
|
// The canonical verb vector-file wrapper: endpoints + verb name ride
|
|
// as first-class v2 wire fields precisely so this reconstruction is
|
|
// exact ({id, vector, connections:{}, verb, sourceId, targetId} —
|
|
// verbs carry no `level`).
|
|
const vector: Record<string, unknown> = {
|
|
id: r.id,
|
|
vector: r.vectorLeg ?? [],
|
|
connections: {},
|
|
verb: r.verb,
|
|
sourceId: r.sourceId,
|
|
targetId: r.targetId
|
|
}
|
|
ops.push({ kind: 'verb', id: r.id, record: { metadata, vector } })
|
|
break
|
|
}
|
|
case 'verb.tombstone':
|
|
ops.push({ kind: 'verb', id: r.id, record: null })
|
|
break
|
|
case 'batch.meta':
|
|
batchMeta = normalizeWireJson(r.meta) as Record<string, unknown>
|
|
break
|
|
case 'log.genesis':
|
|
break // the log's birth certificate — log-level metadata, not an op
|
|
case 'projection.note':
|
|
markers.push({ ...r, note: normalizeWireJson(r.note) as Record<string, unknown> })
|
|
break
|
|
case 'bootstrap.baseline':
|
|
markers.push({ ...r, metadata: normalizeWireJson(r.metadata) })
|
|
break
|
|
default:
|
|
// embed.pending / embed.landed / blob.manifest carry no loose JSON maps.
|
|
markers.push(r)
|
|
break
|
|
}
|
|
}
|
|
const meta = f.meta ? (normalizeWireJson(f.meta) as Record<string, unknown>) : batchMeta
|
|
return {
|
|
generation: f.generation,
|
|
timestamp: f.timestamp,
|
|
ops,
|
|
...(meta ? { meta } : {}),
|
|
...(f.blobHashes && f.blobHashes.length > 0 ? { blobHashes: f.blobHashes } : {}),
|
|
...(markers.length > 0 ? { records: markers } : {})
|
|
}
|
|
}
|
|
|
|
/** One intact v2 frame's extent inside a segment (byte-slicing support). */
|
|
interface V2FrameExtent {
|
|
/** Byte offset just past this frame. */
|
|
end: number
|
|
/** The frame's generation (0 for pad filler). */
|
|
generation: number
|
|
/** True when the frame is a pad (invisible filler). */
|
|
isPad: boolean
|
|
}
|
|
|
|
/** Walk a v2 segment's intact frames (torn-tail terminated), returning each
|
|
* frame's extent — the byte-level view truncation slices against, so kept
|
|
* frames are never re-encoded (byte-immutability of CRC-covered frames). */
|
|
function walkV2Frames(bytes: Uint8Array): V2FrameExtent[] {
|
|
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
|
|
const extents: V2FrameExtent[] = []
|
|
let offset = HEADER_BYTES
|
|
while (offset + FRAME_PREFIX_BYTES <= bytes.length) {
|
|
const length = view.getUint32(offset, true)
|
|
const expectedCrc = view.getUint32(offset + 4, true)
|
|
const start = offset + FRAME_PREFIX_BYTES
|
|
const end = start + length
|
|
if (end > bytes.length) break // torn tail
|
|
const payload = bytes.subarray(start, end)
|
|
if (crc32c(payload) !== expectedCrc) break // torn tail
|
|
const fact = decodeFormatFact(payload, FACT_LOG_FORMAT_V2, {
|
|
expectedIdSpaceWidth: 64
|
|
}) as CommitFactV2
|
|
extents.push({ end, generation: fact.generation, isPad: fact.records.length === 0 })
|
|
offset = end
|
|
}
|
|
return extents
|
|
}
|
|
|
|
/**
|
|
* The byte offset a v2 segment is cut at to keep exactly the facts with
|
|
* `generation ≤ keepThrough`: the end of the last kept FACT frame (pads
|
|
* between kept facts sit inside the retained span; pads after the cut are
|
|
* dropped and re-sealed at the next sync). When nothing is dropped the cut
|
|
* lands after the last intact frame — trailing pads retained, only a torn
|
|
* suffix (if any) removed.
|
|
*/
|
|
function v2CutOffset(extents: V2FrameExtent[], keepThrough: number): number {
|
|
let cut = HEADER_BYTES
|
|
let lastIntactEnd = HEADER_BYTES
|
|
for (const e of extents) {
|
|
lastIntactEnd = e.end
|
|
if (e.isPad) continue
|
|
if (e.generation <= keepThrough) {
|
|
cut = e.end
|
|
} else {
|
|
return cut // first beyond-keep fact: everything from here (pads included) goes
|
|
}
|
|
}
|
|
return lastIntactEnd
|
|
}
|
|
|
|
/**
|
|
* Parse a segment's bytes: verify the header, then walk frames until the end
|
|
* or a torn tail (length overrun / CRC mismatch), which terminates the walk —
|
|
* everything before it is intact. The header's formatVersion selects the
|
|
* decoder: the v1 walk below is byte-identical to the original v1 reader;
|
|
* v2 segments decode through the reference codec (`decodeGroupV2`, pads
|
|
* invisible, id-space width verified at 64 — a disagreeing genesis throws
|
|
* the codec's typed `GenesisWidthMismatchError`). Returns the decoded facts
|
|
* plus the byte length of the VALID prefix (header + intact frames), which
|
|
* reconciliation uses to cut a torn tail without re-encoding.
|
|
*/
|
|
function parseSegment(
|
|
file: string,
|
|
bytes: Uint8Array
|
|
): { facts: CommitFact[]; validBytes: number; formatVersion: number; sealSize?: number } {
|
|
if (bytes.length < HEADER_BYTES) {
|
|
prodLog.warn(`[FactLog] segment ${file} shorter than its header — treating as empty`)
|
|
return { facts: [], validBytes: 0, formatVersion: 0 }
|
|
}
|
|
let header: { formatVersion: number; sealSize?: number }
|
|
try {
|
|
header = parseSegmentHeader(bytes.subarray(0, HEADER_BYTES))
|
|
} catch (err) {
|
|
throw new Error(`fact log: segment ${file}: ${(err as Error).message}`)
|
|
}
|
|
|
|
if (header.formatVersion === FACT_LOG_FORMAT_V2) {
|
|
const group = decodeGroupV2(bytes.subarray(HEADER_BYTES), { expectedIdSpaceWidth: 64 })
|
|
return {
|
|
facts: group.facts.map(commitFactFromV2),
|
|
validBytes: HEADER_BYTES + group.validBytes,
|
|
formatVersion: FACT_LOG_FORMAT_V2,
|
|
sealSize: header.sealSize
|
|
}
|
|
}
|
|
|
|
// v1 walk — byte-identical to the original reader.
|
|
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
|
|
const facts: CommitFact[] = []
|
|
let offset = HEADER_BYTES
|
|
while (offset + FRAME_PREFIX_BYTES <= bytes.length) {
|
|
const length = view.getUint32(offset, true)
|
|
const expectedCrc = view.getUint32(offset + 4, true)
|
|
const start = offset + FRAME_PREFIX_BYTES
|
|
const end = start + length
|
|
if (end > bytes.length) break // torn tail: frame length overruns the file
|
|
const payload = bytes.subarray(start, end)
|
|
if (crc32c(payload) !== expectedCrc) break // torn tail: payload CRC mismatch
|
|
facts.push(decodeFact(payload))
|
|
offset = end
|
|
}
|
|
return { facts, validBytes: offset, formatVersion: FACT_LOG_FORMAT_V1 }
|
|
}
|
|
|
|
/**
|
|
* THE PRUNE LAW — which segment files a pass looking for facts ABOVE
|
|
* `committedGeneration` actually has to read, and how many the manifest's own
|
|
* recorded bounds took off the table.
|
|
*
|
|
* A sealed segment's `lastGeneration` is written at SEAL time and never
|
|
* mutated upward afterwards ({@link FactLog.rotate}, unchanged since the log
|
|
* was introduced): the tail's bytes are fsynced FIRST (`await this.sync()` —
|
|
* "sealed segments are always fully durable"), the entry is then built from
|
|
* the content that fsync covered, and only then does the manifest flip —
|
|
* atomically (tmp+rename) and fsynced — which in the SAME write re-points
|
|
* `tailSegment` at a new file, so the sealed file is never appended to again.
|
|
* A crash anywhere in that order is safe in the pruning direction: crash
|
|
* before the manifest write and the segment is still the TAIL (read whole);
|
|
* crash after it and the entry describes bytes that were already durable. The
|
|
* only later mutation of a sealed segment is `open()`'s straddle truncation,
|
|
* which REMOVES facts and re-derives the entry from the actual bytes — so a
|
|
* recorded bound can drift DOWN with its file, never up.
|
|
*
|
|
* Therefore: `lastGeneration = L` proves the file holds no fact above L, and
|
|
* a pass above `committedGeneration >= L` can skip it whole — no read, no
|
|
* CRC decode, no msgpack. What the manifest cannot PROVE is never pruned: an
|
|
* entry with no numeric `lastGeneration` (a legacy or hand-repaired manifest)
|
|
* is read, and the unsealed tail is always read.
|
|
*
|
|
* This is the difference between an open that costs O(whole fact log) and one
|
|
* that costs O(the facts that could matter). MEASURED in production: a 16k-row
|
|
* brain at generation ~478,819 paid 34-37s of segment reads and CRC decoding
|
|
* in `generation-store-open-fold` on EVERY open — to answer a question whose
|
|
* answer, after a clean close, is always "nothing".
|
|
*/
|
|
function segmentsHoldingFactsAbove(
|
|
stored: FactsManifest,
|
|
committedGeneration: number
|
|
): { files: string[]; pruned: number } {
|
|
const files: string[] = []
|
|
let pruned = 0
|
|
for (const entry of stored.segments) {
|
|
const last = (entry as Partial<SegmentEntry>).lastGeneration
|
|
if (typeof last === 'number' && Number.isFinite(last) && last <= committedGeneration) {
|
|
pruned++
|
|
continue
|
|
}
|
|
files.push(entry.file)
|
|
}
|
|
if (stored.tailSegment) files.push(stored.tailSegment)
|
|
return { files, pruned }
|
|
}
|
|
|
|
/**
|
|
* Say what the open actually read. One line, and only when the log holds more
|
|
* than one segment (a single-segment log has nothing to prune and nothing to
|
|
* report) — the operator's receipt that the open is paying for the tail, not
|
|
* for the whole history.
|
|
*/
|
|
function narrateAboveScan(
|
|
pass: string,
|
|
committedGeneration: number,
|
|
read: number,
|
|
pruned: number
|
|
): void {
|
|
if (read + pruned <= 1) return
|
|
prodLog.narrate(
|
|
`[FactLog] ${pass} above generation ${committedGeneration}: ${read} segment(s) read, ` +
|
|
`${pruned} pruned of ${read + pruned} (sealed at or below the bound)`
|
|
)
|
|
}
|
|
|
|
/**
|
|
* The generation fact log. One instance per open store; every method assumes
|
|
* the single-writer discipline the generation store already enforces (calls
|
|
* arrive under its commit mutex).
|
|
*/
|
|
export class FactLog {
|
|
private readonly storage: FactLogStorage
|
|
/** Rotation threshold (bytes); tests may lower it to exercise rotation. */
|
|
private readonly rotateBytes: number
|
|
private manifest: FactsManifest = {
|
|
formatVersion: FACTS_FORMAT_VERSION,
|
|
segments: [],
|
|
tailSegment: null,
|
|
updatedAt: new Date(0).toISOString()
|
|
}
|
|
/** Decoded facts of the TAIL segment (bounded by the rotation threshold). */
|
|
private tailFacts: CommitFact[] = []
|
|
/** Byte size of the tail segment file (valid prefix, pads included —
|
|
* pads count toward bytes but NEVER toward facts). */
|
|
private tailBytes = 0
|
|
/** Highest generation in the log (0 = empty). */
|
|
private head = 0
|
|
/** Segment paths appended since the last sync (the fsync batch). */
|
|
private readonly dirtySegments = new Set<string>()
|
|
/** The TAIL segment's on-disk format version (selects the live encoder). */
|
|
private tailVersion: number = FACT_LOG_FORMAT_V1
|
|
/** The tail's sector-seal size (v2 tails; from its header on reopen). */
|
|
private tailSealSize: number = DEFAULT_SEAL_SIZE
|
|
/** The v2 int minter (see {@link FactIntMinter}); null = v1 live writes. */
|
|
private intMinter: FactIntMinter | null = null
|
|
|
|
constructor(storage: FactLogStorage, options?: { rotateBytes?: number }) {
|
|
this.storage = storage
|
|
this.rotateBytes = options?.rotateBytes ?? SEGMENT_ROTATE_BYTES
|
|
}
|
|
|
|
/**
|
|
* Install the v2 int minter — the capability gate for v2 LIVE WRITES.
|
|
* With a minter installed, every NEW tail segment writes the v2 format and
|
|
* after-image records carry minted dense ints; without one, live writes
|
|
* stay v1 (no authority could reproduce int assignments, and 0 is never
|
|
* written). The brain wires this from the metadata index's id mapper right
|
|
* after the index is ready; an existing v1 tail cuts over on the next
|
|
* append (empty tail: re-headed in place; non-empty: sealed by rotation).
|
|
*/
|
|
setIntMinter(mint: FactIntMinter): void {
|
|
this.intMinter = mint
|
|
}
|
|
|
|
/** The highest committed generation the log holds (0 = empty). */
|
|
headGeneration(): number {
|
|
return this.head
|
|
}
|
|
|
|
/**
|
|
* True when this log has EVER had a v2 tail — the manifest's `brainId` is
|
|
* minted at every v2 tail creation seam and never removed (the tail-version
|
|
* check is a belt-and-braces second signal). Only v2 facts can carry marker
|
|
* records, so marker folds (e.g. the deferred-embed recovery scan) skip
|
|
* v1-only logs WHOLESALE on this one cheap check — no segment is read.
|
|
*/
|
|
hasV2History(): boolean {
|
|
return this.manifest.brainId !== undefined || this.tailVersion === FACT_LOG_FORMAT_V2
|
|
}
|
|
|
|
/**
|
|
* STREAMING twin of {@link FactLog.peekFactsAbove} for the recovery fold:
|
|
* yields facts above the bound one SEGMENT at a time, ascending, without
|
|
* ever materializing the whole log (a production first-boot fold OOM-class
|
|
* allocation storm came from exactly that — GBs of decoded after-images in
|
|
* one array while the process looked hung). Memory is one segment's worth.
|
|
* Works manifest-direct (safe before {@link FactLog.open}). Ordering is
|
|
* structural (segments rotate in order; appends are ordered within one) and
|
|
* ASSERTED — a violation aborts loudly, never a silent misordered replay.
|
|
*
|
|
* Reads only the segments that CAN hold a fact above the bound — see
|
|
* {@link segmentsHoldingFactsAbove}. A bounded fold above a high checkpoint
|
|
* therefore reads its own tail, not the whole history it already proved
|
|
* durable.
|
|
*/
|
|
async *streamFactsAbove(committedGeneration: number): AsyncGenerator<CommitFact[], void> {
|
|
const stored = (await this.storage.readRawObject(FACTS_MANIFEST_PATH)) as FactsManifest | null
|
|
if (!stored || typeof stored !== 'object' || !Array.isArray(stored.segments)) return
|
|
if (stored.formatVersion !== FACTS_FORMAT_VERSION) return
|
|
const { files, pruned } = segmentsHoldingFactsAbove(stored, committedGeneration)
|
|
narrateAboveScan('recovery fold', committedGeneration, files.length, pruned)
|
|
let lastGen = committedGeneration
|
|
for (const file of files) {
|
|
const bytes = await this.storage.readRawBytes(`${FACTS_PREFIX}/${file}`)
|
|
if (bytes === null) continue
|
|
const { facts } = parseSegment(file, bytes)
|
|
const batch: CommitFact[] = []
|
|
for (const f of facts) {
|
|
if (f.generation <= committedGeneration) continue
|
|
if (f.generation <= lastGen) {
|
|
throw new Error(
|
|
`fact log: streamFactsAbove found non-ascending generations ` +
|
|
`(${f.generation} after ${lastGen} in ${file}) — refusing to replay out of order`
|
|
)
|
|
}
|
|
lastGen = f.generation
|
|
batch.push(f)
|
|
}
|
|
if (batch.length > 0) yield batch
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Read (without truncating) every intact fact ABOVE a generation — the
|
|
* log-authority recovery surface: after a crash, facts beyond the
|
|
* manifest watermark that survived with valid CRCs are ACKED writes in
|
|
* durable-at-ack mode, and the owner REPLAYS them instead of letting
|
|
* open() truncate them. Must be called BEFORE open() (it reads the raw
|
|
* segments directly; the torn tail's invalid suffix is ignored exactly
|
|
* like open() would).
|
|
*
|
|
* Reads only the segments that CAN hold such a fact — see
|
|
* {@link segmentsHoldingFactsAbove}. This runs on EVERY log-authority open,
|
|
* including the clean one where the answer is always empty, so the segments
|
|
* the manifest already proves irrelevant are never opened at all.
|
|
*/
|
|
async peekFactsAbove(committedGeneration: number): Promise<CommitFact[]> {
|
|
const stored = (await this.storage.readRawObject(FACTS_MANIFEST_PATH)) as FactsManifest | null
|
|
if (!stored || typeof stored !== 'object' || !Array.isArray(stored.segments)) return []
|
|
if (stored.formatVersion !== FACTS_FORMAT_VERSION) return []
|
|
const out: CommitFact[] = []
|
|
const { files, pruned } = segmentsHoldingFactsAbove(stored, committedGeneration)
|
|
narrateAboveScan('above-manifest peek', committedGeneration, files.length, pruned)
|
|
for (const file of files) {
|
|
const bytes = await this.storage.readRawBytes(`${FACTS_PREFIX}/${file}`)
|
|
if (bytes === null) continue
|
|
const { facts } = parseSegment(file, bytes)
|
|
for (const f of facts) {
|
|
if (f.generation > committedGeneration) out.push(f)
|
|
}
|
|
}
|
|
out.sort((a, b) => a.generation - b.generation)
|
|
return out
|
|
}
|
|
|
|
/**
|
|
* Open the log and reconcile it to committed truth: read the manifest,
|
|
* establish the tail's intact content (torn-tail scan), then TRUNCATE any
|
|
* fact with `generation > committedGeneration` — those never committed (a
|
|
* crash between fact-append and the commit point). After open, the log is
|
|
* exactly the committed prefix.
|
|
*/
|
|
async open(committedGeneration: number): Promise<void> {
|
|
const stored = (await this.storage.readRawObject(FACTS_MANIFEST_PATH)) as FactsManifest | null
|
|
if (stored && typeof stored === 'object' && Array.isArray(stored.segments)) {
|
|
if (stored.formatVersion !== FACTS_FORMAT_VERSION) {
|
|
throw new Error(
|
|
`fact log: manifest formatVersion ${stored.formatVersion}; this build reads ${FACTS_FORMAT_VERSION}`
|
|
)
|
|
}
|
|
this.manifest = stored
|
|
}
|
|
|
|
// Drop sealed segments that sit ENTIRELY beyond committed truth (a crash
|
|
// right after a rotation whose facts never committed), newest first.
|
|
while (this.manifest.segments.length > 0) {
|
|
const last = this.manifest.segments[this.manifest.segments.length - 1]
|
|
if (last.firstGeneration > committedGeneration) {
|
|
prodLog.warn(
|
|
`[FactLog] dropping sealed segment ${last.file} (generations ${last.firstGeneration}..` +
|
|
`${last.lastGeneration} never committed)`
|
|
)
|
|
await this.storage.deleteRawObject(`${FACTS_PREFIX}/${last.file}`)
|
|
this.manifest.segments.pop()
|
|
await this.persistManifest()
|
|
} else if (last.lastGeneration > committedGeneration) {
|
|
// A sealed segment STRADDLING committed truth: cut it back.
|
|
await this.truncateSegmentTo(last.file, committedGeneration)
|
|
const cut = await this.reloadSegmentEntry(last.file)
|
|
this.manifest.segments[this.manifest.segments.length - 1] = cut
|
|
await this.persistManifest()
|
|
break
|
|
} else {
|
|
break
|
|
}
|
|
}
|
|
|
|
// Establish the tail: scan its intact prefix, then truncate beyond
|
|
// committed truth (the common crash shape: buffered single-op facts whose
|
|
// counter never went durable).
|
|
if (this.manifest.tailSegment) {
|
|
const tailPath = `${FACTS_PREFIX}/${this.manifest.tailSegment}`
|
|
const bytes = await this.storage.readRawBytes(tailPath)
|
|
if (bytes === null) {
|
|
// Manifest named a tail whose first byte never landed — an empty
|
|
// tail. Its header (and format version) is established at the next
|
|
// append (see the tail-provisioning ladder there).
|
|
this.tailFacts = []
|
|
this.tailBytes = 0
|
|
} else {
|
|
const parsed = parseSegment(this.manifest.tailSegment, bytes)
|
|
const { facts, validBytes } = parsed
|
|
this.tailVersion =
|
|
parsed.formatVersion === FACT_LOG_FORMAT_V2 ? FACT_LOG_FORMAT_V2 : FACT_LOG_FORMAT_V1
|
|
this.tailSealSize = parsed.sealSize ?? DEFAULT_SEAL_SIZE
|
|
const kept = facts.filter((f) => f.generation <= committedGeneration)
|
|
if (kept.length !== facts.length || validBytes !== bytes.length) {
|
|
const dropped = facts.length - kept.length
|
|
if (dropped > 0) {
|
|
prodLog.warn(
|
|
`[FactLog] truncating ${dropped} uncommitted fact(s) beyond generation ` +
|
|
`${committedGeneration} from the tail (never committed)`
|
|
)
|
|
}
|
|
if (this.tailVersion === FACT_LOG_FORMAT_V2) {
|
|
// V2: byte-slice at frame boundaries — CRC-covered frames are
|
|
// byte-immutable; a truncation never re-encodes what it keeps.
|
|
const cut = v2CutOffset(walkV2Frames(bytes), committedGeneration)
|
|
await this.storage.writeRawBytes(tailPath, bytes.subarray(0, cut))
|
|
this.tailFacts = kept
|
|
this.tailBytes = cut
|
|
} else {
|
|
await this.rewriteTail(kept)
|
|
}
|
|
} else {
|
|
this.tailFacts = facts
|
|
this.tailBytes = validBytes
|
|
}
|
|
}
|
|
}
|
|
|
|
this.head = this.computeHead()
|
|
}
|
|
|
|
/**
|
|
* Append one committed generation's fact. NOT durable until {@link sync} —
|
|
* the caller batches durability at its commit barrier (transact syncs in
|
|
* the same call; Model-B group-commit syncs at flush).
|
|
*
|
|
* Tail provisioning (in order): a missing tail starts one; a named tail
|
|
* whose header never landed (manifest-first crash) gets its header now; an
|
|
* existing V1 tail cuts over to v2 once the minter is installed (empty:
|
|
* re-headed in place, non-empty: sealed by rotation — v1 segments are never
|
|
* rewritten); a full tail rotates. The frame then encodes in the TAIL's
|
|
* format: v2 tails carry after-image records with minted ints (and the
|
|
* genesis record on the very first fact of a brand-new log); v1 tails keep
|
|
* the v1 wire format byte-identically.
|
|
*/
|
|
async append(fact: CommitFact): Promise<void> {
|
|
if (fact.generation <= this.head) {
|
|
throw new Error(
|
|
`fact log: non-monotonic append (generation ${fact.generation} ≤ head ${this.head})`
|
|
)
|
|
}
|
|
if (this.manifest.tailSegment === null) {
|
|
await this.startTail(fact.generation)
|
|
} else if (this.tailBytes === 0) {
|
|
await this.reinitializeTailHeader()
|
|
} else if (this.intMinter !== null && this.tailVersion === FACT_LOG_FORMAT_V1) {
|
|
if (this.tailFacts.length === 0 && this.tailBytes <= HEADER_BYTES) {
|
|
await this.upgradeEmptyTailToV2()
|
|
} else {
|
|
await this.rotate(fact.generation)
|
|
}
|
|
} else if (this.tailBytes >= this.rotateBytes) {
|
|
await this.rotate(fact.generation)
|
|
}
|
|
|
|
let frame: Uint8Array
|
|
if (this.tailVersion === FACT_LOG_FORMAT_V2) {
|
|
const records = this.buildV2Records(fact)
|
|
if (this.needsGenesis()) {
|
|
if (this.ensureBrainId()) await this.persistManifest()
|
|
records.unshift(this.genesisRecord())
|
|
}
|
|
frame = encodeFactV2({
|
|
generation: fact.generation,
|
|
timestamp: fact.timestamp,
|
|
records,
|
|
...(fact.meta ? { meta: toJsonSafe(fact.meta) as Record<string, unknown> } : {}),
|
|
...(fact.blobHashes && fact.blobHashes.length > 0 ? { blobHashes: fact.blobHashes } : {})
|
|
})
|
|
} else {
|
|
if (fact.records && fact.records.length > 0) {
|
|
throw new Error(
|
|
`fact log: marker records (${fact.records.map((r) => r.type).join(', ')}) require a ` +
|
|
`v2 tail segment — this log's tail is v1 (no int minter installed); refusing rather ` +
|
|
`than silently dropping them`
|
|
)
|
|
}
|
|
frame = encodeFrame(fact)
|
|
}
|
|
const tailPath = `${FACTS_PREFIX}/${this.manifest.tailSegment}`
|
|
await this.storage.appendRawBytes(tailPath, frame)
|
|
this.tailFacts.push(fact)
|
|
this.tailBytes += frame.length
|
|
this.head = fact.generation
|
|
this.dirtySegments.add(tailPath)
|
|
}
|
|
|
|
/**
|
|
* Fsync every segment appended since the last sync. SEALS AT SYNC: a v2
|
|
* tail is first padded to its sector-seal boundary (one pad frame,
|
|
* invisible to readers; a gap smaller than the smallest constructible pad
|
|
* frame pads through one extra sector — the codec's rule), so every
|
|
* durability barrier leaves the tail sector-aligned: a torn page can only
|
|
* tear INSIDE the group being written, never a previously-sealed one.
|
|
*/
|
|
async sync(): Promise<void> {
|
|
await this.padTailToSealBoundary()
|
|
if (this.dirtySegments.size === 0) return
|
|
const paths = [...this.dirtySegments]
|
|
this.dirtySegments.clear()
|
|
await this.storage.syncRawObjects(paths)
|
|
}
|
|
|
|
// --- GROUP COMMIT ON THE LOG (durable-at-ack mode) ------------------------
|
|
// Classic group commit: concurrent writers append, then join ONE fsync
|
|
// whose completion releases every covered ack. Two slots — the running
|
|
// sync and at most one queued behind it — give the covering guarantee:
|
|
// an append followed by ensureSynced() is always covered, because the
|
|
// sync it awaits STARTS after the append landed (a running sync that
|
|
// may have snapshotted earlier is never joined; the queued one is).
|
|
private syncRunning: Promise<void> | null = null
|
|
private syncQueued: Promise<void> | null = null
|
|
|
|
/**
|
|
* Await a sync that covers every byte appended before this call. Many
|
|
* concurrent callers share one fsync (solo caller = immediate sync). The
|
|
* durability contract of an acked write in log-durable mode: this promise
|
|
* resolving means the caller's frames survive power loss.
|
|
*/
|
|
async ensureSynced(): Promise<void> {
|
|
if (this.syncQueued) {
|
|
// A sync that has NOT started yet exists — it will snapshot after our
|
|
// append, so it covers us.
|
|
return this.syncQueued
|
|
}
|
|
if (this.syncRunning) {
|
|
// The running sync may have snapshotted before our append — queue the
|
|
// next one behind it and join that.
|
|
const queued = this.syncRunning
|
|
.catch(() => {})
|
|
.then(() => {
|
|
// Promote: the queued sync becomes the running one.
|
|
this.syncQueued = null
|
|
this.syncRunning = this.sync().finally(() => {
|
|
this.syncRunning = null
|
|
})
|
|
return this.syncRunning
|
|
})
|
|
this.syncQueued = queued
|
|
return queued
|
|
}
|
|
this.syncRunning = this.sync().finally(() => {
|
|
this.syncRunning = null
|
|
})
|
|
return this.syncRunning
|
|
}
|
|
|
|
/**
|
|
* Open a scan over committed facts. The scan runs against a MANIFEST
|
|
* SNAPSHOT (sealed segments + the tail's decoded facts at open) — exactly-
|
|
* once per fact, inclusive bounds, stable under concurrent appends. Gaps
|
|
* abort LOUDLY: a missing generation inside a segment's declared range is
|
|
* corruption, never silently skipped.
|
|
*/
|
|
scanFacts(options?: {
|
|
fromGeneration?: number
|
|
toGeneration?: number
|
|
kinds?: Array<'noun' | 'verb'>
|
|
batchSize?: number
|
|
/** Test override for the first-batch liveness bound (default {@link SCANFACTS_FIRST_BATCH_MS}). */
|
|
firstBatchTimeoutMs?: number
|
|
}): FactScanHandle {
|
|
const from = options?.fromGeneration ?? 1
|
|
const to = options?.toGeneration ?? this.head
|
|
const kinds = options?.kinds
|
|
const batchSize = Math.max(1, options?.batchSize ?? 256)
|
|
|
|
// Snapshot: the segment list + tail content as of NOW.
|
|
const segments = this.manifest.segments.filter(
|
|
(s) => s.lastGeneration >= from && s.firstGeneration <= to
|
|
)
|
|
const tailSnapshot = this.tailFacts.filter((f) => f.generation >= from && f.generation <= to)
|
|
const tailId = this.manifest.tailSegment ?? 'tail'
|
|
const approxFactCount =
|
|
segments.reduce((sum, s) => sum + s.facts, 0) + tailSnapshot.length
|
|
|
|
let factsYielded = 0
|
|
let segmentsRead = 0
|
|
const storage = this.storage
|
|
|
|
async function* batches(this: void): AsyncGenerator<FactScanBatch> {
|
|
let expectedNext = 0 // gap detection: generations are monotonic, not necessarily dense
|
|
const emit = (facts: CommitFact[], segmentId: string, byteSize: number): FactScanBatch => ({
|
|
facts,
|
|
firstGeneration: facts[0].generation,
|
|
lastGeneration: facts[facts.length - 1].generation,
|
|
factCount: facts.length,
|
|
byteSize,
|
|
segmentId
|
|
})
|
|
const filterOps = (fact: CommitFact): CommitFact =>
|
|
kinds
|
|
? { ...fact, ops: fact.ops.filter((op) => kinds.includes(op.kind)) }
|
|
: fact
|
|
|
|
for (const entry of segments) {
|
|
const bytes = await storage.readRawBytes(`${FACTS_PREFIX}/${entry.file}`)
|
|
if (bytes === null) {
|
|
throw new Error(
|
|
`fact log: sealed segment ${entry.file} is MISSING — the log is damaged; aborting scan`
|
|
)
|
|
}
|
|
const { facts } = parseSegment(entry.file, bytes)
|
|
segmentsRead++
|
|
const inRange = facts.filter((f) => f.generation >= from && f.generation <= to)
|
|
for (const f of inRange) {
|
|
if (f.generation <= expectedNext - 1) {
|
|
throw new Error(`fact log: out-of-order fact ${f.generation} in ${entry.file} — aborting scan`)
|
|
}
|
|
expectedNext = f.generation + 1
|
|
}
|
|
for (let i = 0; i < inRange.length; i += batchSize) {
|
|
const slice = inRange.slice(i, i + batchSize).map(filterOps)
|
|
if (slice.length === 0) continue
|
|
factsYielded += slice.length
|
|
yield emit(slice, entry.file, slice.reduce((n, f) => n + encodeFrame(f).length, 0))
|
|
}
|
|
}
|
|
|
|
if (tailSnapshot.length > 0) {
|
|
segmentsRead++
|
|
for (const f of tailSnapshot) {
|
|
if (f.generation <= expectedNext - 1) {
|
|
throw new Error(`fact log: out-of-order fact ${f.generation} in the tail — aborting scan`)
|
|
}
|
|
expectedNext = f.generation + 1
|
|
}
|
|
for (let i = 0; i < tailSnapshot.length; i += batchSize) {
|
|
const slice = tailSnapshot.slice(i, i + batchSize).map(filterOps)
|
|
factsYielded += slice.length
|
|
yield emit(slice, tailId, slice.reduce((n, f) => n + encodeFrame(f).length, 0))
|
|
}
|
|
}
|
|
}
|
|
|
|
// Liveness wrapper: the FIRST pull races the contract deadline. Only the
|
|
// first — the bound is time-to-first-batch (proof the producer is alive),
|
|
// not per-batch pacing; and it runs only while a pull is actually pending,
|
|
// so consumer think-time between pulls never counts against the producer.
|
|
const firstBatchTimeoutMs = options?.firstBatchTimeoutMs ?? SCANFACTS_FIRST_BATCH_MS
|
|
async function* batchesWithLiveness(this: void): AsyncGenerator<FactScanBatch> {
|
|
const inner = batches()
|
|
let timer: NodeJS.Timeout | undefined
|
|
try {
|
|
const deadline = new Promise<never>((_, reject) => {
|
|
timer = setTimeout(
|
|
() =>
|
|
reject(
|
|
new Error(
|
|
`fact log: scanFacts produced no first batch within ${firstBatchTimeoutMs}ms ` +
|
|
`(liveness contract) — the store is wedged or unreadably slow; aborting scan LOUDLY ` +
|
|
`instead of hanging the consumer.`
|
|
)
|
|
),
|
|
firstBatchTimeoutMs
|
|
)
|
|
timer.unref?.()
|
|
})
|
|
const first = await Promise.race([inner.next(), deadline])
|
|
if (first.done) return
|
|
yield first.value
|
|
} finally {
|
|
clearTimeout(timer)
|
|
}
|
|
yield* inner
|
|
}
|
|
|
|
return {
|
|
headGeneration: this.head,
|
|
segmentCount: segments.length + (tailSnapshot.length > 0 ? 1 : 0),
|
|
approxFactCount,
|
|
batches: batchesWithLiveness,
|
|
summary: () => ({ factsYielded, segmentsRead })
|
|
}
|
|
}
|
|
|
|
/**
|
|
* The mmap fast path (capability handoff): the immutable sealed segment
|
|
* files covering `fromGeneration`, in order. The TAIL is deliberately NOT
|
|
* included — it is append-mutable; consumers read it via {@link scanFacts}.
|
|
*/
|
|
segmentPaths(options?: { fromGeneration?: number }): string[] {
|
|
const from = options?.fromGeneration ?? 1
|
|
return this.manifest.segments
|
|
.filter((s) => s.lastGeneration >= from)
|
|
.map((s) => `${FACTS_PREFIX}/${s.file}`)
|
|
}
|
|
|
|
/**
|
|
* Drop every fact with `generation > keepThrough` — the in-session abort
|
|
* compensation: a transact appends its fact BEFORE the commit point, so a
|
|
* real (non-crash) abort after the append must take the fact back out. The
|
|
* dropped facts can only live in the TAIL (they were just appended); the
|
|
* rewrite is atomic and bounded by the rotation threshold.
|
|
*/
|
|
async dropAbove(keepThrough: number): Promise<void> {
|
|
if (this.head <= keepThrough) return
|
|
const kept = this.tailFacts.filter((f) => f.generation <= keepThrough)
|
|
if (kept.length === this.tailFacts.length) {
|
|
throw new Error(
|
|
`fact log: dropAbove(${keepThrough}) found no droppable facts in the tail ` +
|
|
`(head ${this.head}) — the fact to drop was already sealed; the log needs reopen`
|
|
)
|
|
}
|
|
if (this.tailVersion === FACT_LOG_FORMAT_V2) {
|
|
// V2: byte-slice at frame boundaries (kept frames stay byte-identical;
|
|
// pads between kept facts are retained inside the prefix, trailing pads
|
|
// go and the next sync re-seals). The dropped frames may be unsynced —
|
|
// readRawBytes is read-after-write coherent over the append path.
|
|
const file = this.manifest.tailSegment
|
|
if (!file) return
|
|
const tailPath = `${FACTS_PREFIX}/${file}`
|
|
const bytes = await this.storage.readRawBytes(tailPath)
|
|
if (bytes === null) {
|
|
throw new Error(`fact log: dropAbove(${keepThrough}) cannot read the tail segment ${file}`)
|
|
}
|
|
const cut = v2CutOffset(walkV2Frames(bytes), keepThrough)
|
|
await this.storage.writeRawBytes(tailPath, bytes.subarray(0, cut))
|
|
this.tailFacts = kept
|
|
this.tailBytes = cut
|
|
} else {
|
|
await this.rewriteTail(kept)
|
|
}
|
|
this.head = this.computeHead()
|
|
}
|
|
|
|
// -- internals -------------------------------------------------------------
|
|
|
|
private computeHead(): number {
|
|
if (this.tailFacts.length > 0) return this.tailFacts[this.tailFacts.length - 1].generation
|
|
const sealed = this.manifest.segments
|
|
if (sealed.length > 0) return sealed[sealed.length - 1].lastGeneration
|
|
return 0
|
|
}
|
|
|
|
/** The header bytes for a NEW tail: v2 whenever the minter is installed. */
|
|
private newTailHeader(firstGeneration: number): Uint8Array {
|
|
return this.intMinter !== null
|
|
? encodeSegmentHeaderV2(firstGeneration, DEFAULT_SEAL_SIZE)
|
|
: buildHeader(firstGeneration)
|
|
}
|
|
|
|
/** Record the just-created tail's format in memory (mirrors its header). */
|
|
private noteFreshTail(): void {
|
|
this.tailVersion = this.intMinter !== null ? FACT_LOG_FORMAT_V2 : FACT_LOG_FORMAT_V1
|
|
this.tailSealSize = DEFAULT_SEAL_SIZE
|
|
}
|
|
|
|
/** Create the very first tail segment (manifest-first, then header bytes). */
|
|
private async startTail(firstGeneration: number): Promise<void> {
|
|
const file = segmentFileName(firstGeneration)
|
|
this.manifest.tailSegment = file
|
|
if (this.intMinter !== null) this.ensureBrainId()
|
|
await this.persistManifest()
|
|
await this.storage.appendRawBytes(`${FACTS_PREFIX}/${file}`, this.newTailHeader(firstGeneration))
|
|
this.tailFacts = []
|
|
this.tailBytes = HEADER_BYTES
|
|
this.noteFreshTail()
|
|
}
|
|
|
|
/**
|
|
* Seal the tail into the manifest and start a new one. Manifest-first: the
|
|
* flip both seals the old tail AND names the new one atomically, so no
|
|
* segment file ever exists unaccounted for. The NEW tail's format follows
|
|
* the minter gate ({@link newTailHeader}) — this is also the v1→v2 cutover
|
|
* seam for a non-empty v1 tail (sealed as-is, never rewritten).
|
|
*/
|
|
private async rotate(nextGeneration: number): Promise<void> {
|
|
const sealedFile = this.manifest.tailSegment
|
|
if (!sealedFile) return
|
|
// Seal what the tail actually holds (sync() also sector-seals a v2 tail).
|
|
await this.sync() // sealed segments are always fully durable
|
|
const entry: SegmentEntry = {
|
|
file: sealedFile,
|
|
firstGeneration: this.tailFacts[0]?.generation ?? nextGeneration,
|
|
lastGeneration: this.tailFacts[this.tailFacts.length - 1]?.generation ?? nextGeneration - 1,
|
|
facts: this.tailFacts.length,
|
|
bytes: this.tailBytes
|
|
}
|
|
const newFile = segmentFileName(nextGeneration)
|
|
this.manifest.segments.push(entry)
|
|
this.manifest.tailSegment = newFile
|
|
if (this.intMinter !== null) this.ensureBrainId()
|
|
await this.persistManifest()
|
|
await this.storage.appendRawBytes(`${FACTS_PREFIX}/${newFile}`, this.newTailHeader(nextGeneration))
|
|
this.tailFacts = []
|
|
this.tailBytes = HEADER_BYTES
|
|
this.noteFreshTail()
|
|
}
|
|
|
|
/**
|
|
* The v1→v2 cutover for an EMPTY v1 tail: re-head it in place (nothing but
|
|
* the 32-byte header exists, so no v1 frame is ever rewritten). Also the
|
|
* cheapest cutover shape: brand-new brains whose first tail predates the
|
|
* minter installation converge here on their first post-install append.
|
|
*/
|
|
private async upgradeEmptyTailToV2(): Promise<void> {
|
|
const file = this.manifest.tailSegment
|
|
if (!file) return
|
|
if (this.ensureBrainId()) await this.persistManifest()
|
|
const first = this.segmentFirstGenerationFromName(file)
|
|
const path = `${FACTS_PREFIX}/${file}`
|
|
await this.storage.writeRawBytes(path, encodeSegmentHeaderV2(first, DEFAULT_SEAL_SIZE))
|
|
this.tailBytes = HEADER_BYTES
|
|
this.tailVersion = FACT_LOG_FORMAT_V2
|
|
this.tailSealSize = DEFAULT_SEAL_SIZE
|
|
this.dirtySegments.add(path)
|
|
}
|
|
|
|
/**
|
|
* A manifest-named tail whose header never landed (crash between the
|
|
* manifest flip and the first header byte — previously this appended
|
|
* frames into a headerless file the next open could not parse): write the
|
|
* header now, in the CURRENT format gate.
|
|
*/
|
|
private async reinitializeTailHeader(): Promise<void> {
|
|
const file = this.manifest.tailSegment
|
|
if (!file) return
|
|
if (this.intMinter !== null && this.ensureBrainId()) await this.persistManifest()
|
|
const first = this.segmentFirstGenerationFromName(file)
|
|
const path = `${FACTS_PREFIX}/${file}`
|
|
await this.storage.writeRawBytes(path, this.newTailHeader(first))
|
|
this.tailBytes = HEADER_BYTES
|
|
this.noteFreshTail()
|
|
this.dirtySegments.add(path)
|
|
}
|
|
|
|
/** True when the NEXT appended fact is the first fact of a brand-new v2
|
|
* log — the one that must open with the log.genesis record. */
|
|
private needsGenesis(): boolean {
|
|
return (
|
|
this.tailVersion === FACT_LOG_FORMAT_V2 &&
|
|
this.manifest.segments.length === 0 &&
|
|
this.tailFacts.length === 0
|
|
)
|
|
}
|
|
|
|
/** Mint the brain id into the manifest if absent; true when it changed. */
|
|
private ensureBrainId(): boolean {
|
|
if (this.manifest.brainId) return false
|
|
this.manifest.brainId = uuidv4()
|
|
return true
|
|
}
|
|
|
|
/** The log's birth certificate (id-space width 64 — the only width this
|
|
* writer mints; a reader expecting another width refuses at decode). */
|
|
private genesisRecord(): LogRecord {
|
|
const brainId = this.manifest.brainId
|
|
if (!brainId) {
|
|
throw new Error(
|
|
'fact log v2: genesis requires a brainId in the facts manifest — invariant violated'
|
|
)
|
|
}
|
|
return { type: 'log.genesis', idSpaceWidth: 64, brainId, createdAt: Date.now() }
|
|
}
|
|
|
|
/**
|
|
* Convert one CommitFact's ops (+ optional marker records) to v2 wire
|
|
* records, MINTING ints at append time: entity/verb ints come from the
|
|
* injected minter (the metadata index's id mapper — the one authority a
|
|
* rebuild reproduces exactly). Verb endpoints and the verb name ride as
|
|
* first-class wire fields, lifted from the canonical verb vector wrapper.
|
|
* Every refusal here is loud — an after-image without a mintable int, a
|
|
* verb without endpoints, or a vector record without floats fails the
|
|
* WRITE, never writes a 0.
|
|
*/
|
|
private buildV2Records(fact: CommitFact): LogRecord[] {
|
|
const mint = (kind: 'noun' | 'verb', id: string): bigint => {
|
|
if (this.intMinter === null) {
|
|
throw new Error(
|
|
`fact log v2: no int minter is installed — cannot mint the ${kind} int for ${id}; ` +
|
|
`refusing to write a v2 after-image (an int of 0 is never written)`
|
|
)
|
|
}
|
|
const minted = this.intMinter(kind, id)
|
|
// Reserved-root exemption: int 0 is legitimate for exactly one id —
|
|
// the all-zeros VFS root, minted 0 by construction at genesis on
|
|
// existing brains. Zero anywhere else is a corrupt mint.
|
|
const isReservedRoot =
|
|
minted === 0n && id === '00000000-0000-0000-0000-000000000000'
|
|
if (typeof minted !== 'bigint' || minted < 0n || (minted === 0n && !isReservedRoot)) {
|
|
throw new Error(
|
|
`fact log v2: the int minter returned ${String(minted)} for ${kind} ${id} — ` +
|
|
`minted ints are positive bigints (int 0 reserved for the VFS root alone); ` +
|
|
`refusing to write`
|
|
)
|
|
}
|
|
return minted
|
|
}
|
|
|
|
const records: LogRecord[] = []
|
|
for (const op of fact.ops) {
|
|
if (op.kind === 'noun') {
|
|
if (op.record === null) {
|
|
records.push({ type: 'noun.tombstone', id: op.id })
|
|
continue
|
|
}
|
|
records.push({
|
|
type: 'noun.afterImage',
|
|
id: op.id,
|
|
entityInt: mint('noun', op.id),
|
|
metadata: toJsonSafe(op.record.metadata ?? null),
|
|
vectorLeg: embeddingLegOf(op.record.vector, `noun ${op.id}`)
|
|
})
|
|
} else {
|
|
if (op.record === null) {
|
|
records.push({ type: 'verb.tombstone', id: op.id })
|
|
continue
|
|
}
|
|
const wrapper = op.record.vector as Record<string, unknown> | null
|
|
const verbName = wrapper?.verb
|
|
const sourceId = wrapper?.sourceId
|
|
const targetId = wrapper?.targetId
|
|
if (
|
|
typeof verbName !== 'string' ||
|
|
typeof sourceId !== 'string' ||
|
|
typeof targetId !== 'string'
|
|
) {
|
|
throw new Error(
|
|
`fact log v2: verb ${op.id} has no canonical endpoints (verb/sourceId/targetId ` +
|
|
`live in its vector record, which is missing or torn) — refusing to write an ` +
|
|
`after-image that could not be replayed`
|
|
)
|
|
}
|
|
const floats = floatsOf(wrapper?.vector, `verb ${op.id}`) ?? []
|
|
records.push({
|
|
type: 'verb.afterImage',
|
|
id: op.id,
|
|
verbInt: mint('verb', op.id),
|
|
metadata: toJsonSafe(op.record.metadata ?? null),
|
|
vectorLeg: floats,
|
|
verb: verbName,
|
|
sourceId,
|
|
sourceInt: mint('noun', sourceId),
|
|
targetId,
|
|
targetInt: mint('noun', targetId)
|
|
})
|
|
}
|
|
}
|
|
for (const marker of fact.records ?? []) records.push(marker)
|
|
return records
|
|
}
|
|
|
|
/**
|
|
* Pad a v2 tail to its next sector-seal boundary with ONE pad frame —
|
|
* called from {@link sync} so alignment holds at every durability barrier.
|
|
* Pads count toward {@link tailBytes} but never toward facts (they are
|
|
* invisible to every reader); a gap smaller than the smallest constructible
|
|
* pad frame pads through one extra sector (the codec's rule). No-op for v1
|
|
* tails, empty tails, and already-aligned tails.
|
|
*/
|
|
private async padTailToSealBoundary(): Promise<void> {
|
|
if (this.tailVersion !== FACT_LOG_FORMAT_V2) return
|
|
const file = this.manifest.tailSegment
|
|
if (!file || this.tailBytes <= HEADER_BYTES) return
|
|
const remainder = this.tailBytes % this.tailSealSize
|
|
if (remainder === 0) return
|
|
let padBytes = this.tailSealSize - remainder
|
|
if (padBytes < minPadFrameBytes()) padBytes += this.tailSealSize
|
|
const tailPath = `${FACTS_PREFIX}/${file}`
|
|
await this.storage.appendRawBytes(tailPath, encodePadFrame(padBytes))
|
|
this.tailBytes += padBytes
|
|
this.dirtySegments.add(tailPath)
|
|
}
|
|
|
|
/** Atomically persist the manifest (write-new → fsync → rename downstream). */
|
|
private async persistManifest(): Promise<void> {
|
|
this.manifest.updatedAt = new Date().toISOString()
|
|
await this.storage.writeRawObject(FACTS_MANIFEST_PATH, this.manifest)
|
|
await this.storage.syncRawObjects([FACTS_MANIFEST_PATH])
|
|
}
|
|
|
|
/** Rewrite the tail segment to hold exactly `facts` (atomic replace). */
|
|
private async rewriteTail(facts: CommitFact[]): Promise<void> {
|
|
const file = this.manifest.tailSegment
|
|
if (!file) return
|
|
const first = facts[0]?.generation ?? this.segmentFirstGenerationFromName(file)
|
|
const parts: Uint8Array[] = [buildHeader(first)]
|
|
for (const f of facts) parts.push(encodeFrame(f))
|
|
const total = parts.reduce((n, p) => n + p.length, 0)
|
|
const merged = new Uint8Array(total)
|
|
let offset = 0
|
|
for (const p of parts) {
|
|
merged.set(p, offset)
|
|
offset += p.length
|
|
}
|
|
await this.storage.writeRawBytes(`${FACTS_PREFIX}/${file}`, merged)
|
|
this.tailFacts = facts
|
|
this.tailBytes = total
|
|
}
|
|
|
|
/** Cut a SEALED segment back to `committedGeneration` (atomic replace).
|
|
* v2 segments byte-slice at frame boundaries (kept frames — pads
|
|
* included — are never re-encoded); the v1 re-encode path is unchanged. */
|
|
private async truncateSegmentTo(file: string, committedGeneration: number): Promise<void> {
|
|
const path = `${FACTS_PREFIX}/${file}`
|
|
const bytes = await this.storage.readRawBytes(path)
|
|
if (bytes === null) return
|
|
const { facts, formatVersion } = parseSegment(file, bytes)
|
|
const kept = facts.filter((f) => f.generation <= committedGeneration)
|
|
prodLog.warn(
|
|
`[FactLog] truncating sealed segment ${file} to generation ${committedGeneration} ` +
|
|
`(${facts.length - kept.length} uncommitted fact(s) dropped)`
|
|
)
|
|
if (formatVersion === FACT_LOG_FORMAT_V2) {
|
|
const cut = v2CutOffset(walkV2Frames(bytes), committedGeneration)
|
|
await this.storage.writeRawBytes(path, bytes.subarray(0, cut))
|
|
return
|
|
}
|
|
const first = kept[0]?.generation ?? this.segmentFirstGenerationFromName(file)
|
|
const parts: Uint8Array[] = [buildHeader(first)]
|
|
for (const f of kept) parts.push(encodeFrame(f))
|
|
const total = parts.reduce((n, p) => n + p.length, 0)
|
|
const merged = new Uint8Array(total)
|
|
let offset = 0
|
|
for (const p of parts) {
|
|
merged.set(p, offset)
|
|
offset += p.length
|
|
}
|
|
await this.storage.writeRawBytes(path, merged)
|
|
}
|
|
|
|
/** Re-derive a sealed segment's manifest entry from its actual bytes. */
|
|
private async reloadSegmentEntry(file: string): Promise<SegmentEntry> {
|
|
const bytes = await this.storage.readRawBytes(`${FACTS_PREFIX}/${file}`)
|
|
const { facts, validBytes } = bytes
|
|
? parseSegment(file, bytes)
|
|
: { facts: [] as CommitFact[], validBytes: 0 }
|
|
return {
|
|
file,
|
|
firstGeneration: facts[0]?.generation ?? this.segmentFirstGenerationFromName(file),
|
|
lastGeneration: facts[facts.length - 1]?.generation ?? 0,
|
|
facts: facts.length,
|
|
bytes: validBytes
|
|
}
|
|
}
|
|
|
|
/** Parse the zero-padded firstGeneration back out of a segment filename. */
|
|
private segmentFirstGenerationFromName(file: string): number {
|
|
const match = /^seg-(\d{20})\.bfl$/.exec(file)
|
|
return match ? Number(match[1]) : 0
|
|
}
|
|
}
|