feat(log): v2 is the LIVE write format — envelope records with minted ints, genesis, sector seals; v1 readable forever
The cutover: new tail segments write format v2 (per-record [type, version, cipherFlag, keyId] envelope; noun/verb after-images carry dense ints MINTED AT APPEND from the id mapper — a rebuilt mapper reproduces assignments exactly; log.genesis opens every new log with the id-space width + a minted brain id; sync() seals to the header-declared sector boundary with reader-invisible pad frames). Existing v1 segments are never rewritten — per-segment decoder dispatch reads both formats and v2 facts map to the exact CommitFact shape all consumers already read. Cutover on a live v1 log: an empty v1 tail re-heads in place; a non-empty one is sealed by rotation, byte-identical. Records reserve the encryption fields (cipherFlag 0 / keyId nil are the only legal values; anything else refuses typed naming the needed newer reader) — crypto-ready with no future bump on the compat surface. Empty-records facts are legal (an all-deduped batch is a real generation — v1 semantics preserved; the refusal there tore a column-store flush mid-commit in the full suite, the consistency guard caught it loudly, and the root is fixed). Golden byte vectors pinned for the second (native) reader implementation. Pins: cutover 5/5 · codec 54 · kill-matrix stays 11/11.
This commit is contained in:
parent
73eb88d481
commit
26c6025158
6 changed files with 1372 additions and 135 deletions
|
|
@ -41,10 +41,57 @@
|
|||
* 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.
|
||||
*
|
||||
* ## 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
|
||||
|
|
@ -65,7 +112,12 @@ export function setFactCodec(impl: {
|
|||
export const FACTS_PREFIX = '_generations/facts'
|
||||
/** The facts manifest path (JSON). */
|
||||
export const FACTS_MANIFEST_PATH = `${FACTS_PREFIX}/manifest.json`
|
||||
/** Current segment format version (header field; additive-only within a major). */
|
||||
/**
|
||||
* 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
|
||||
|
|
@ -83,6 +135,29 @@ export interface FactOp {
|
|||
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). Encoder-ready by
|
||||
* design; nothing produces them yet — the deferred-embed sidecar and blob
|
||||
* lifecycle remodel onto these records 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
|
||||
|
|
@ -90,6 +165,12 @@ export interface CommitFact {
|
|||
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). */
|
||||
|
|
@ -143,6 +224,12 @@ interface FactsManifest {
|
|||
/** 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. */
|
||||
|
|
@ -251,40 +338,339 @@ function decodeFact(payload: Uint8Array): CommitFact {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
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. 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.
|
||||
* 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 } {
|
||||
): { 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 }
|
||||
return { facts: [], validBytes: 0, formatVersion: 0 }
|
||||
}
|
||||
for (let i = 0; i < MAGIC.length; i++) {
|
||||
if (bytes[i] !== MAGIC[i]) {
|
||||
throw new Error(`fact log: segment ${file} has a bad magic — not a fact segment`)
|
||||
}
|
||||
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}`)
|
||||
}
|
||||
const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
|
||||
const version = view.getUint32(8, true)
|
||||
if (version !== FACTS_FORMAT_VERSION) {
|
||||
throw new Error(
|
||||
`fact log: segment ${file} has formatVersion ${version}; this build reads ${FACTS_FORMAT_VERSION}`
|
||||
)
|
||||
}
|
||||
for (let i = 20; i < HEADER_BYTES; i++) {
|
||||
if (bytes[i] !== 0) {
|
||||
// Non-zero reserved bytes = a future format this build cannot verify.
|
||||
throw new Error(`fact log: segment ${file} has non-zero reserved header bytes — unverifiable`)
|
||||
|
||||
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) {
|
||||
|
|
@ -298,7 +684,7 @@ function parseSegment(
|
|||
facts.push(decodeFact(payload))
|
||||
offset = end
|
||||
}
|
||||
return { facts, validBytes: offset }
|
||||
return { facts, validBytes: offset, formatVersion: FACT_LOG_FORMAT_V1 }
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -318,18 +704,38 @@ export class FactLog {
|
|||
}
|
||||
/** Decoded facts of the TAIL segment (bounded by the rotation threshold). */
|
||||
private tailFacts: CommitFact[] = []
|
||||
/** Byte size of the tail segment file (valid prefix). */
|
||||
/** 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
|
||||
|
|
@ -412,11 +818,17 @@ export class FactLog {
|
|||
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.
|
||||
// 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 { facts, validBytes } = parseSegment(this.manifest.tailSegment, bytes)
|
||||
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
|
||||
|
|
@ -426,7 +838,16 @@ export class FactLog {
|
|||
`${committedGeneration} from the tail (never committed)`
|
||||
)
|
||||
}
|
||||
await this.rewriteTail(kept)
|
||||
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
|
||||
|
|
@ -441,6 +862,15 @@ export class FactLog {
|
|||
* 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) {
|
||||
|
|
@ -450,10 +880,42 @@ export class FactLog {
|
|||
}
|
||||
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)
|
||||
}
|
||||
const frame = encodeFrame(fact)
|
||||
|
||||
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)
|
||||
|
|
@ -462,8 +924,16 @@ export class FactLog {
|
|||
this.dirtySegments.add(tailPath)
|
||||
}
|
||||
|
||||
/** Fsync every segment appended since the last sync. */
|
||||
/**
|
||||
* 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()
|
||||
|
|
@ -671,7 +1141,25 @@ export class FactLog {
|
|||
`(head ${this.head}) — the fact to drop was already sealed; the log needs reopen`
|
||||
)
|
||||
}
|
||||
await this.rewriteTail(kept)
|
||||
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()
|
||||
}
|
||||
|
||||
|
|
@ -684,25 +1172,42 @@ export class FactLog {
|
|||
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}`, buildHeader(firstGeneration))
|
||||
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.
|
||||
* 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.
|
||||
// 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,
|
||||
|
|
@ -714,10 +1219,181 @@ export class FactLog {
|
|||
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}`, buildHeader(nextGeneration))
|
||||
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)
|
||||
if (typeof minted !== 'bigint' || minted <= 0n) {
|
||||
throw new Error(
|
||||
`fact log v2: the int minter returned ${String(minted)} for ${kind} ${id} — ` +
|
||||
`minted ints are positive bigints; 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). */
|
||||
|
|
@ -746,17 +1422,24 @@ export class FactLog {
|
|||
this.tailBytes = total
|
||||
}
|
||||
|
||||
/** Cut a SEALED segment back to `committedGeneration` (atomic replace). */
|
||||
/** 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 } = parseSegment(file, bytes)
|
||||
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))
|
||||
|
|
|
|||
Reference in a new issue