diff --git a/docs/guides/reacting-to-changes.md b/docs/guides/reacting-to-changes.md index 99436ff9..7367a5e3 100644 --- a/docs/guides/reacting-to-changes.md +++ b/docs/guides/reacting-to-changes.md @@ -79,7 +79,10 @@ interface BrainyChangeEvent { - **Fire-and-forget.** There is no replay or backpressure. For catch-up after a disconnect, use the `generation` on each event together with [`asOf()` / the transaction log](snapshots-and-time-travel.md): record the - last generation you processed, and on reconnect diff from there. + last generation you processed, and on reconnect diff from there. For file + content specifically, `vfs.readFile(path, { asOf })` and + `vfs.history(path)` are the temporal read — see + [Snapshots & Time Travel](snapshots-and-time-travel.md). ## Patterns diff --git a/docs/guides/snapshots-and-time-travel.md b/docs/guides/snapshots-and-time-travel.md index d9f0e264..6f7b150a 100644 --- a/docs/guides/snapshots-and-time-travel.md +++ b/docs/guides/snapshots-and-time-travel.md @@ -366,6 +366,61 @@ are done with (including the ones `transact()` returns), and `persist()` any generation you want to keep beyond the retention window: snapshots are self-contained and unaffected by compaction. +## Time travel for files (the VFS) + +Since 8.2.0, time travel covers Virtual Filesystem **content**, not just +entity records. File bytes are retention-protected: a content blob referenced +by any generation inside the retention window is never reclaimed, so reading +the past always returns the exact bytes — never a stale field or a +dangling hash. + +**`vfs.readFile(path, { asOf })`** takes a generation number or a `Date` and +returns the file's exact bytes as they stood then. It resolves the path's +current entity, then materializes its state at the target generation — so it +answers *"what did the file at this path hold at that point?"* It bypasses +the content cache; the `encoding` option still applies. Asking about a +generation before the file existed throws the usual not-found error, and +asking past the retention window's compaction horizon throws a +compacted-generation error. + +**`vfs.history(path)`** returns the file's versions inside the retention +window, oldest first — one `FileVersion` per generation that wrote the file, +the newest entry being the current state: + +```typescript +// A CMS page evolves… +await brain.vfs.writeFile('/pages/home.json', '{"title":"Launch"}') +await brain.vfs.writeFile('/pages/home.json', '{"title":"Launch v2"}') +await brain.vfs.writeFile('/pages/home.json', '{"title":""}') // bad deploy! + +// Every version is listed and readable: +const versions = await brain.vfs.history('/pages/home.json') +// → [{ generation, timestamp, hash, size, mimeType? }, …] ascending + +const good = versions[versions.length - 2] +const bytes = await brain.vfs.readFile('/pages/home.json', { + asOf: good.generation +}) + +// Restore = write the old bytes back. This is a NEW write (a new +// generation) — history is never rewritten, so the bad version stays +// visible in the audit trail. +await brain.vfs.writeFile('/pages/home.json', bytes) +``` + +Two lifecycle consequences worth stating plainly: + +- **Deleting or overwriting a file no longer frees its bytes immediately.** + Old content lives until history compaction reclaims the generations that + reference it — the same `retention` budget that bounds all Model-B history + (and pinned views are exempt, exactly as above). Size your `retention` for + the file-version depth you want; `retention: 'all'` keeps every version of + every file forever. +- **After `compactHistory()` reclaims a generation, its file versions are + gone** and their bytes are physically reclaimed. (This also fixed a + pre-8.2.0 defect where overwritten content was never reclaimed at all — an + unbounded silent leak.) + ## From branches to values If you used the pre-8.0 `fork`/`checkout`/`commit`/`versions` surface, every diff --git a/src/brainy.ts b/src/brainy.ts index ee91bfed..d54cfe1c 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -1102,6 +1102,19 @@ export class Brainy implements BrainyInterface { // already healed. See adoptLegacyCowBlobs. await this.autoAdoptLegacyVfsBlobsIfNeeded() + // Temporal-blob contract: one-time (marker-gated) backfill of blob + // history reference counts for stores whose generation history predates + // the contract — after it, compaction reclaims blob bytes exactly (zero + // live AND zero history references). On a scrub failure the store runs + // leak-safe (no blob reclamation) rather than risk a premature delete. + if (typeof (this.storage as unknown as { + backfillBlobHistoryRefCountsIfNeeded?: () => Promise + }).backfillBlobHistoryRefCountsIfNeeded === 'function') { + await (this.storage as unknown as { + backfillBlobHistoryRefCountsIfNeeded: () => Promise + }).backfillBlobHistoryRefCountsIfNeeded() + } + // Rebuild indexes if needed for existing data await this.rebuildIndexesIfNeeded() diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index f7739e02..e440f230 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -437,6 +437,28 @@ export class GenerationStore { return this.horizonGen } + /** + * @description Read one generation's persisted before-image records — the + * compaction fallback for generations written before deltas carried + * `blobHashes`. O(that generation's records). + * @param gen - The generation whose `prev/` records to read. + * @returns The record-set (empty when absent). + */ + private async readGenerationRecords(gen: number): Promise { + let paths: string[] = [] + try { + paths = await this.storage.listRawObjects(`${GENERATIONS_PREFIX}/${gen}/prev`) + } catch { + return [] + } + const records: GenerationRecord[] = [] + for (const p of paths) { + const record = (await this.storage.readRawObject(p)) as GenerationRecord | null + if (record) records.push(record) + } + return records + } + /** * @description Single-operation write hook (registered with the storage * layer in {@link open}). Bumps the in-memory counter and schedules a @@ -592,6 +614,10 @@ export class GenerationStore { } } + // Temporal-blob contract: hashes this record-set references, recorded + // before staging; scoped outside the try so an abort can compensate. + let txBlobHashes: string[] = [] + try { // -- 3. Before-images + delta (the durable undo log) ------------------ // Read every before-image FIRST, then run the caller's CAS @@ -615,6 +641,19 @@ export class GenerationStore { // which returns the generation reservation; nothing was applied. args.precommit?.({ nouns: nounBefore, verbs: verbBefore }) + // Temporal-blob contract: count this record-set's content-hash + // references BEFORE it is staged (over-count-only crash ordering — + // see flushPendingSingleOps' matching note). + txBlobHashes = this.storage.extractBlobHashesFromRecords + ? this.storage.extractBlobHashesFromRecords([ + ...nounBefore.values(), + ...verbBefore.values() + ]) + : [] + if (txBlobHashes.length > 0 && this.storage.recordHistoryBlobReferences) { + await this.storage.recordHistoryBlobReferences(txBlobHashes) + } + const stagedPaths: string[] = [] let recordBytes = 0 // serialized record-set size, for retention accounting for (const [id, record] of nounBefore) { @@ -634,7 +673,10 @@ export class GenerationStore { timestamp, ...(args.meta && { meta: args.meta }), nouns, - verbs + verbs, + // Always present on new deltas (empty = "no blobs"), so compaction + // only falls back to record reads for pre-contract generations. + blobHashes: txBlobHashes } delta.bytes = recordBytes + serializedBytes(delta) const deltaPath = `${dir}/tx.json` @@ -697,6 +739,18 @@ export class GenerationStore { `${(cleanupErr as Error).message} (recovery will remove it on next open)` ) } + // Compensate the pre-staging history-reference increments. Best + // effort — a failure here only over-counts (a leak the scrub + // repairs). Reclaim inside is safe on an abort: the before-image + // hashes are the entities' still-live content, so live references + // block any physical delete. + if (txBlobHashes.length > 0 && this.storage.releaseHistoryBlobReferences) { + try { + await this.storage.releaseHistoryBlobReferences(txBlobHashes) + } catch { + // over-count-safe; the scrub restores exactness + } + } // Return the reservation when no concurrent bump consumed a later // number, so a failed transaction leaves generation() unchanged. if (this.counter === gen) this.counter = gen - 1 @@ -860,6 +914,21 @@ export class GenerationStore { const buf = this.pendingBuffer.get(gen) if (!buf) continue const dir = `${GENERATIONS_PREFIX}/${gen}` + + // Temporal-blob contract: count this record-set's content-hash + // references BEFORE persisting it (a crash between the two only + // over-counts — the scrub repairs a leak; under-counting could let + // compaction reclaim bytes a retained generation still needs). + const blobHashes = this.storage.extractBlobHashesFromRecords + ? this.storage.extractBlobHashesFromRecords([ + ...buf.nouns.values(), + ...buf.verbs.values() + ]) + : [] + if (blobHashes.length > 0 && this.storage.recordHistoryBlobReferences) { + await this.storage.recordHistoryBlobReferences(blobHashes) + } + const nounIds: string[] = [] const verbIds: string[] = [] let recordBytes = 0 @@ -882,7 +951,10 @@ export class GenerationStore { timestamp: buf.timestamp, groupCommit: true, nouns: nounIds, - verbs: verbIds + verbs: verbIds, + // Always present on new deltas (empty = "no blobs"), so compaction + // only falls back to record reads for pre-contract generations. + blobHashes } delta.bytes = recordBytes + serializedBytes(delta) genBytes.set(gen, delta.bytes) @@ -1722,20 +1794,45 @@ export class GenerationStore { for (const gen of [...this.committedGensAsc()]) { // Pins are always exempt: never reclaim a generation a live pin needs. if (gen > minPinned) break // committedGensAsc ascending → nothing newer is eligible either + const delta = await this.getDelta(gen) if (!noCaps) { const violatesCount = maxGenerations !== undefined && remainingCount > maxGenerations const violatesBytes = maxBytes !== undefined && remainingBytes > maxBytes - let violatesAge = false - if (ageCutoff !== undefined) { - const delta = await this.getDelta(gen) - violatesAge = delta.timestamp < ageCutoff - } + const violatesAge = ageCutoff !== undefined && delta.timestamp < ageCutoff // Oldest-first: once the oldest unpinned gen trips no cap, none newer do. if (!violatesCount && !violatesBytes && !violatesAge) break } - const genBytes = maxBytes !== undefined ? (await this.getDelta(gen)).bytes : 0 + const genBytes = maxBytes !== undefined ? delta.bytes : 0 + + // Temporal-blob contract: resolve the content-blob hashes this + // generation's record-set references BEFORE deleting it. New + // generations carry the multiset in their persisted delta (empty + // array when none — distinguishing "new format, no blobs" from a + // pre-contract delta); legacy generations fall back to reading the + // records themselves. Skipped entirely on non-blob-aware storage. + let blobHashes: string[] | undefined + if (this.storage.releaseHistoryBlobReferences) { + const rawDelta = (await this.storage.readRawObject( + `${GENERATIONS_PREFIX}/${gen}/tx.json` + )) as GenerationDelta | null + blobHashes = rawDelta?.blobHashes + if (blobHashes === undefined && this.storage.extractBlobHashesFromRecords) { + blobHashes = this.storage.extractBlobHashesFromRecords( + await this.readGenerationRecords(gen) + ) + } + } + await this.storage.removeRawPrefix(`${GENERATIONS_PREFIX}/${gen}`) this.deltaCache.delete(gen) + + // AFTER the record-set is gone (over-count-only crash ordering): + // release its history references and reclaim any blob left with zero + // live AND zero history references — the system's one byte-reclaim point. + if (blobHashes && blobHashes.length > 0 && this.storage.releaseHistoryBlobReferences) { + await this.storage.releaseHistoryBlobReferences(blobHashes) + } + remainingCount-- remainingBytes -= genBytes removed.push(gen) diff --git a/src/db/types.ts b/src/db/types.ts index 80a1959d..4030533a 100644 --- a/src/db/types.ts +++ b/src/db/types.ts @@ -301,6 +301,16 @@ export interface GenerationDelta { nouns: string[] /** Relationship ids touched by this generation. */ verbs: string[] + /** + * Content-blob hashes referenced by this generation's before-image records + * (a MULTISET — one entry per referencing record occurrence), captured at + * persist time so compaction can release the exact history references it + * reclaims without re-reading the records. Always present (possibly empty) + * on deltas written under the temporal-blob contract; absent only on + * pre-contract generations, for which compaction falls back to reading the + * record-set itself. + */ + blobHashes?: string[] /** * `true` for a Model-B single-operation generation persisted by the async * group-commit flush (`GenerationStore.flushPendingSingleOps`). It marks the @@ -395,6 +405,28 @@ export interface GenerationStorage { /** Read all lines of `_system/tx-log.jsonl` (empty array if absent). */ readTxLogLines(): Promise + /** + * OPTIONAL temporal-blob contract (implemented by blob-aware storage; the + * generation store treats the hashes as opaque strings). Extract the + * content-blob hashes a record-set references — a pure multiset extraction, + * no side effects. + */ + extractBlobHashesFromRecords?(records: GenerationRecord[]): string[] + /** + * OPTIONAL: record history references for the given hashes (one increment + * per occurrence). Called BEFORE the referencing record-set is persisted — + * a crash between the two can only over-count (a leak the scrub repairs), + * never under-count (which would risk reclaiming bytes history still needs). + */ + recordHistoryBlobReferences?(hashes: string[]): Promise + /** + * OPTIONAL: release history references for the given hashes (one decrement + * per occurrence) and physically reclaim any hash left with zero live AND + * zero history references. Called AFTER the referencing record-set is + * deleted by compaction — same over-count-only crash ordering. + */ + releaseHistoryBlobReferences?(hashes: string[]): Promise + /** * Register the generation-bump hook invoked on every entity-visible * single-operation write (see `BaseStorage.setGenerationBumpHook`). diff --git a/src/index.ts b/src/index.ts index ed19fdb1..a7698b39 100644 --- a/src/index.ts +++ b/src/index.ts @@ -23,6 +23,9 @@ export type { ChangeListener } from './events/changeFeed.js' +// Temporal VFS — a file version entry (vfs.history / readFile({ asOf })). +export type { FileVersion } from './vfs/types.js' + // Export diagnostics result type export type { DiagnosticsResult } from './brainy.js' diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index a678a000..ed4755c7 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -869,6 +869,148 @@ export abstract class BaseStorage extends BaseStorageAdapter { this.generationBumpHook = hook } + // ========================================================================== + // Temporal-blob contract (the GenerationStorage optional methods) + // + // Content blobs join the Model-B immutability model through these hooks: + // the generation store counts a history reference per before-image record + // that carries a content hash, and compaction — the ONE reclamation point — + // releases those references and physically deletes bytes only at zero live + // AND zero history references. Crash ordering is over-count-only (record + // BEFORE the record-set persists, release AFTER it is deleted), so a crash + // can leak bytes until the scrub recounts but can never reclaim bytes a + // retained generation still needs. + // ========================================================================== + + /** Set when the open-time backfill/scrub could not verify history reference + * counts. While true, the temporal-blob hooks stop mutating counts and + * compaction stops reclaiming blob bytes — pure leak-safe mode until a + * successful {@link scrubBlobHistoryRefCounts} restores exactness. */ + private blobHistoryRefsUnverified = false + + /** + * @description Extract the content-blob hashes a generation record-set + * references — a pure MULTISET extraction (one entry per referencing record + * occurrence), no side effects. Only entity records can reference VFS + * content (`metadata.storage.type === 'blob'`). + * @param records - The record-set's before-image records. + * @returns The referenced hashes, duplicates preserved. + */ + public extractBlobHashesFromRecords( + records: Array<{ kind: string; metadata: unknown }> + ): string[] { + const hashes: string[] = [] + for (const record of records) { + if (record.kind !== 'noun') continue + const storage = (record.metadata as { storage?: { type?: string; hash?: unknown } } | null) + ?.storage + if (storage?.type === 'blob' && typeof storage.hash === 'string') { + hashes.push(storage.hash) + } + } + return hashes + } + + /** + * @description Record one history reference per hash occurrence (see the + * contract note above — called BEFORE the referencing record-set persists). + * No-op without a blob store or while counts are unverified. + * @param hashes - Hash multiset from {@link extractBlobHashesFromRecords}. + */ + public async recordHistoryBlobReferences(hashes: string[]): Promise { + if (!this.blobStorage || this.blobHistoryRefsUnverified || hashes.length === 0) return + for (const hash of hashes) { + await this.blobStorage.recordHistoryReference(hash) + } + } + + /** + * @description Release one history reference per hash occurrence and + * physically reclaim any hash left with zero live AND zero history + * references — compaction's blob-reclamation step (called AFTER the + * referencing record-set is deleted). No-op without a blob store or while + * counts are unverified (leak-safe: nothing is reclaimed on guesses). + * @param hashes - Hash multiset recorded when the record-set was persisted. + */ + public async releaseHistoryBlobReferences(hashes: string[]): Promise { + if (!this.blobStorage || this.blobHistoryRefsUnverified || hashes.length === 0) return + for (const hash of hashes) { + await this.blobStorage.releaseHistoryReference(hash) + } + for (const hash of new Set(hashes)) { + await this.blobStorage.reclaimIfUnreferenced(hash) + } + } + + /** + * @description One-time (marker-gated) backfill of blob history reference + * counts for stores whose generation history predates the temporal-blob + * contract. Runs the scrub, then stamps `_system/blob-history-refs.json` so + * later opens skip the walk. On scrub failure the store enters leak-safe + * mode (counts untouched, reclamation disabled) rather than risking a + * premature delete on wrong counts. + */ + public async backfillBlobHistoryRefCountsIfNeeded(): Promise { + if (!this.blobStorage) return + const MARKER = '_system/blob-history-refs.json' + try { + const marker = (await this.readObjectFromPath(MARKER)) as { version?: number } | null + if (marker?.version === 1) return + } catch { + // no marker — proceed to scrub + } + try { + await this.scrubBlobHistoryRefCounts() + await this.writeObjectToPath(MARKER, { version: 1, verifiedAt: new Date().toISOString() }) + } catch (err) { + this.blobHistoryRefsUnverified = true + console.error( + '[Brainy] blob history-reference backfill failed — temporal-blob ' + + 'reclamation disabled for this session (leak-safe); history reads ' + + 'are unaffected. Re-open to retry.', + err + ) + } + } + + /** + * @description Recount every blob's history references from the actual + * generation record-sets and set the counts ABSOLUTELY (uncounted blobs are + * zeroed) — the idempotent repair that restores exactness after any crash + * that over-counted. O(history records + stored blobs). + * @returns Blobs counted and records walked, for observability. + */ + public async scrubBlobHistoryRefCounts(): Promise<{ blobs: number; records: number }> { + if (!this.blobStorage) return { blobs: 0, records: 0 } + const counts = new Map() + let records = 0 + let paths: string[] = [] + try { + paths = await this.listObjectsUnderPath('_generations') + } catch { + paths = [] // no history yet + } + for (const p of paths) { + if (!p.includes('/prev/')) continue + const record = (await this.readObjectFromPath(p)) as + | { kind?: string; metadata?: unknown } + | null + if (!record) continue + records++ + for (const hash of this.extractBlobHashesFromRecords([ + { kind: record.kind ?? '', metadata: record.metadata } + ])) { + counts.set(hash, (counts.get(hash) ?? 0) + 1) + } + } + const allHashes = await this.blobStorage.listHashes() + for (const hash of allHashes) { + await this.blobStorage.setHistoryRefCount(hash, counts.get(hash) ?? 0) + } + this.blobHistoryRefsUnverified = false + return { blobs: allHashes.length, records } + } + /** * Read a raw object at a storage-root-relative path. Bypasses the write * cache (record-layer files are written through diff --git a/src/storage/blobStorage.ts b/src/storage/blobStorage.ts index 7cb569ec..f83511f9 100644 --- a/src/storage/blobStorage.ts +++ b/src/storage/blobStorage.ts @@ -48,8 +48,19 @@ export interface BlobMetadata { compression: 'none' | 'zstd' /** Creation timestamp (epoch ms). */ createdAt: number - /** Number of logical references to this blob (deduplicated writes). */ + /** Number of LIVE logical references to this blob (deduplicated writes). */ refCount: number + /** + * Number of persisted generation record-sets (Model-B before-images) that + * reference this hash — the blob's membership in the temporal history. + * Bytes are physically reclaimed only when BOTH counts are zero, and only + * by history compaction: live references protect the present, history + * references protect every `asOf` read inside the retention window (pins + * ride generation pinning, which compaction already respects). Absent on + * metas written before the temporal contract existed (treated as 0; the + * one-time open-time backfill makes legacy stores exact). + */ + historyRefCount?: number } /** @@ -148,7 +159,9 @@ interface CacheEntry { * @example * const hash = await blobStorage.write(buffer, { mimeType: 'image/png' }) * const bytes = await blobStorage.read(hash) // verified against the hash - * await blobStorage.delete(hash) // decrements refCount first + * await blobStorage.release(hash) // drop one LIVE reference + * // bytes are physically reclaimed only by history compaction, once no + * // live reference AND no in-window generation references the hash */ export class BlobStorage { private adapter: BlobStoreAdapter @@ -356,31 +369,107 @@ export class BlobStorage { } /** - * @description Drop one reference to the blob. The stored bytes and - * metadata are physically deleted only when the reference count reaches - * zero — deduplicated content shared by other writers survives. + * @description Drop one LIVE reference to the blob. Never deletes bytes — + * blob content is immutable under the temporal model, exactly like every + * other record: a past generation's `asOf` read may still need these bytes + * even when no live file references them. Physical reclamation happens in + * ONE place only — history compaction via {@link reclaimIfUnreferenced}, + * once no live reference AND no retained generation references the hash. * * @param hash - The blob's SHA-256 hash. */ - async delete(hash: string): Promise { - // Decrement-then-maybe-remove must be atomic per hash: without the lock, - // a concurrent write() could re-reference the content between our - // reaching zero and the physical delete — removing bytes a live - // reference still needs. + async release(hash: string): Promise { await this.hashLocks.runExclusive(hash, async () => { - const refCount = await this.decrementRefCount(hash) + await this.decrementRefCount(hash) + }) + } - // Only delete if no references remain - if (refCount > 0) { + /** + * @description Record that one persisted generation record-set references + * this hash (called by the commit path BEFORE the record-set is written — + * a crash between the two can only over-count, which leaks until the scrub + * recounts; it can never under-count, which would risk premature deletion). + * A missing meta (bytes never stored or already gone) is skipped with a + * warning — counting it could not make its bytes readable. + * @param hash - The blob's SHA-256 hash. + */ + async recordHistoryReference(hash: string): Promise { + await this.hashLocks.runExclusive(hash, async () => { + const metadata = await this.getMetadata(hash) + if (!metadata) { + console.warn( + `[BlobStorage] history reference recorded for absent blob ${hash} — skipped` + ) return } + metadata.historyRefCount = (metadata.historyRefCount ?? 0) + 1 + await this.adapter.put(`blob-meta:${hash}`, Buffer.from(JSON.stringify(metadata))) + }) + } + /** + * @description Drop one history reference (called by compaction AFTER the + * referencing generation record-set is deleted — the safe ordering: a crash + * between the two over-counts, never under-counts). Floored at zero. + * @param hash - The blob's SHA-256 hash. + */ + async releaseHistoryReference(hash: string): Promise { + await this.hashLocks.runExclusive(hash, async () => { + const metadata = await this.getMetadata(hash) + if (!metadata) return + metadata.historyRefCount = Math.max(0, (metadata.historyRefCount ?? 0) - 1) + await this.adapter.put(`blob-meta:${hash}`, Buffer.from(JSON.stringify(metadata))) + }) + } + + /** + * @description Physically delete the blob's bytes + metadata IFF nothing + * references it: zero live references AND zero history references. The one + * reclamation point in the system, invoked by history compaction after it + * releases the reclaimed generations' references. Atomic per hash. + * @param hash - The blob's SHA-256 hash. + * @returns `true` when the bytes were reclaimed. + */ + async reclaimIfUnreferenced(hash: string): Promise { + return this.hashLocks.runExclusive(hash, async () => { + const metadata = await this.getMetadata(hash) + if (!metadata) return false + if ((metadata.refCount ?? 0) > 0 || (metadata.historyRefCount ?? 0) > 0) { + return false + } await this.adapter.delete(`blob:${hash}`) await this.adapter.delete(`blob-meta:${hash}`) this.removeFromCache(hash) + return true }) } + /** + * @description Set the history reference count to an absolute value — the + * backfill/scrub primitive (recounts derived from the actual generation + * records replace whatever the incremental counters hold). Idempotent. + * @param hash - The blob's SHA-256 hash. + * @param count - The exact history reference count. + */ + async setHistoryRefCount(hash: string, count: number): Promise { + await this.hashLocks.runExclusive(hash, async () => { + const metadata = await this.getMetadata(hash) + if (!metadata) return + metadata.historyRefCount = Math.max(0, count) + await this.adapter.put(`blob-meta:${hash}`, Buffer.from(JSON.stringify(metadata))) + }) + } + + /** + * @description Enumerate every stored blob hash (from the metadata keys) — + * the backfill/scrub walk. O(stored blobs). + * @returns All hashes with a stored metadata record. + */ + async listHashes(): Promise { + const keys = await this.adapter.list('blob-meta:') + return keys.map((k) => k.slice('blob-meta:'.length)) + } + /** * @description Read a blob's metadata without reading its bytes. * @param hash - The blob's SHA-256 hash. diff --git a/src/vfs/VirtualFileSystem.ts b/src/vfs/VirtualFileSystem.ts index bbf8328e..013f04f3 100644 --- a/src/vfs/VirtualFileSystem.ts +++ b/src/vfs/VirtualFileSystem.ts @@ -36,6 +36,7 @@ import { VFSErrorCode, WriteOptions, ReadOptions, + FileVersion, MkdirOptions, ReaddirOptions, CopyOptions, @@ -406,6 +407,14 @@ export class VirtualFileSystem implements IVirtualFileSystem { async readFile(path: string, options?: ReadOptions): Promise { await this.ensureInitialized() + // Temporal read: the file's exact bytes as of a past generation or + // instant. Materializes the entity from the generation history — content + // blobs referenced by any in-window generation are retention-protected, + // so the bytes are guaranteed present. Bypasses the content cache. + if (options?.asOf !== undefined) { + return this.readFileAt(path, options.asOf, options) + } + // Check cache first if (options?.cache !== false && this.contentCache.has(path)) { const cached = this.contentCache.get(path)! @@ -472,6 +481,122 @@ export class VirtualFileSystem implements IVirtualFileSystem { } } + /** + * @description The temporal read behind `readFile(path, { asOf })`: resolve + * the path's CURRENT entity, materialize its state at the target + * generation/instant from the Model-B history, and read that version's + * content blob (retention-protected, so present for every in-window + * generation). The pinned view is always released so compaction is never + * blocked by a read. + * @param path - The file path (resolved against the live tree). + * @param asOf - A generation number or wall-clock `Date`. + * @param options - `encoding` applies; caching does not (never cached). + * @returns The exact bytes the file held at that generation. + * @throws VFSError ENOENT when the file did not exist at that generation; + * the generation store's compacted-generation error when `asOf` is past + * the retention window's horizon. + */ + private async readFileAt( + path: string, + asOf: number | Date, + options?: ReadOptions + ): Promise { + const entityId = await this.pathResolver.resolve(path) + const db = await this.brain.asOf(asOf) + try { + const entity = await db.get(entityId) + if (!entity) { + throw new VFSError( + VFSErrorCode.ENOENT, + `File did not exist as of ${String(asOf)}: ${path}`, + path, + 'readFile' + ) + } + const storage = (entity.metadata as Record | undefined)?.storage + if (storage?.type !== 'blob' || typeof storage.hash !== 'string') { + throw new VFSError( + VFSErrorCode.EIO, + `File had no blob storage as of ${String(asOf)}: ${path}`, + path, + 'readFile' + ) + } + const content = await this.blobStorage.read(storage.hash) + if (options?.encoding) { + return Buffer.from(content.toString(options.encoding)) + } + return content + } finally { + await db.release() + } + } + + /** + * @description A file's version history within the retention window: one + * entry per generation that wrote the file, ascending — the newest entry is + * the current state. Pair with `readFile(path, { asOf: entry.generation })` + * to fetch any version's exact bytes, and restore by writing those bytes + * back (a NEW write — history is never rewritten). Bounded by the retention + * window: versions whose generations were compacted are not listed. + * @param path - The file path (resolved against the live tree). + * @returns The versions, oldest first. + * @example + * const versions = await brain.vfs.history('/pages/home.json') + * const before = await brain.vfs.readFile('/pages/home.json', { + * asOf: versions[versions.length - 2].generation + * }) + */ + async history(path: string): Promise { + await this.ensureInitialized() + const entityId = await this.pathResolver.resolve(path) + + // Boundary note: reaches Brainy's internal generation store (the same + // bracket-access precedent as the blobStorage getter) — the per-id + // generation chain and horizon are not on the public Brainy surface. + const generationStore = (this.brain as unknown as { + generationStore: { + horizon(): number + generation(): number + generationsTouching( + kind: 'noun' | 'verb', + id: string, + fromGen: number, + toGen: number + ): Promise + } + }).generationStore + + const gens = await generationStore.generationsTouching( + 'noun', + entityId, + generationStore.horizon(), + generationStore.generation() + ) + + const versions: FileVersion[] = [] + for (const gen of gens) { + const db = await this.brain.asOf(gen) + try { + const entity = await db.get(entityId) + const meta = entity?.metadata as Record | undefined + const storage = meta?.storage + if (storage?.type === 'blob' && typeof storage.hash === 'string') { + versions.push({ + generation: gen, + timestamp: db.timestamp, + hash: storage.hash, + size: typeof meta?.size === 'number' ? meta.size : (storage.size ?? 0), + ...(typeof meta?.mimeType === 'string' && { mimeType: meta.mimeType }) + }) + } + } finally { + await db.release() + } + } + return versions + } + /** * Write a file */ @@ -493,14 +618,18 @@ export class VirtualFileSystem implements IVirtualFileSystem { // Ensure parent directory exists const parentId = await this.ensureDirectory(parentPath) - // Check if file already exists + // Check if file already exists (and capture its current content hash so + // the overwrite can release the superseded live reference afterwards). let existingId: string | null = null + let previousHash: string | undefined try { existingId = await this.pathResolver.resolve(path, { cache: false }) // Verify the entity still exists in the brain const existing = await this.brain.get(existingId) if (!existing) { existingId = null // Entity was deleted but cache wasn't cleared + } else if (existing.metadata?.storage?.type === 'blob') { + previousHash = existing.metadata.storage.hash as string | undefined } } catch (err) { // File doesn't exist, which is fine @@ -551,14 +680,32 @@ export class VirtualFileSystem implements IVirtualFileSystem { Object.assign(metadata, await this.extractMetadata(buffer, mimeType)) } + // For embedding: use text content, for storage: use raw data. Computed + // for BOTH branches — an overwrite must refresh the entity's `data` (and + // therefore its embedding), or semantic search and any `data` read keep + // serving the FIRST version's text forever. + const embeddingData = mimeDetector.isTextFile(mimeType) + ? buffer.toString('utf-8') + : `File: ${name} (${mimeType}, ${buffer.length} bytes)` + if (existingId) { - // Update existing file - // No entity.data - content is in BlobStorage + // Update existing file — content bytes live in BlobStorage; `data` + // carries the embedding text and MUST track the new content. await this.brain.update({ id: existingId, + data: embeddingData, metadata }) + // The update committed: drop the superseded live reference. When the + // content is unchanged this cancels the dedup increment the write() + // above just made (net: one live reference per referencing file); when + // it changed, the old bytes stay retention-protected for asOf reads + // via the update's own generation record. + if (previousHash) { + await this.blobStorage.release(previousHash) + } + // Ensure Contains relationship exists (fix for missing relationships) const existingRelations = await this.brain.related({ from: parentId, @@ -578,9 +725,6 @@ export class VirtualFileSystem implements IVirtualFileSystem { } } else { // Create new file entity - // For embedding: use text content, for storage: use raw data - const embeddingData = mimeDetector.isTextFile(mimeType) ? buffer.toString('utf-8') : `File: ${name} (${mimeType}, ${buffer.length} bytes)` - const entity = await this.brain.add({ data: embeddingData, // Always provide string for embeddings type: this.getFileNounType(mimeType), @@ -649,14 +793,21 @@ export class VirtualFileSystem implements IVirtualFileSystem { throw new VFSError(VFSErrorCode.EISDIR, `Is a directory: ${path}`, path, 'unlink') } - // Delete blob from BlobStorage (decrements ref count) - if (entity.metadata.storage?.type === 'blob') { - await this.blobStorage.delete(entity.metadata.storage.hash) - } - - // Delete the entity + // Delete the entity FIRST, then drop its live blob reference — never + // release a reference while the referencing entity might survive (a + // failed remove must not leave a live file whose bytes compaction could + // later reclaim). await this.brain.remove(entityId) + // Drop the file's LIVE reference to its content blob. The bytes are NOT + // deleted — the remove's own generation record references this hash, so + // `readFile(path, { asOf })` keeps serving it inside the retention + // window; history compaction reclaims the bytes when the last referencing + // generation is reclaimed. + if (entity.metadata.storage?.type === 'blob') { + await this.blobStorage.release(entity.metadata.storage.hash) + } + // Invalidate caches this.pathResolver.invalidatePath(path) this.invalidateCaches(path) @@ -1032,23 +1183,26 @@ export class VirtualFileSystem implements IVirtualFileSystem { // Phase 1: Gather all descendants in ONE batch fetch const descendants = await this.gatherDescendants(entityId, Infinity) - // Phase 2: Parallel blob cleanup (chunked to avoid overwhelming storage) - // Blob deletion is reference-counted, so safe to call for all files + // Phase 2: Batch delete all entities (including root directory) FIRST — + // never release a blob reference while its referencing entity might + // survive a failed delete (see unlink's ordering note). + const allIds = [...descendants.map(d => d.id), entityId] + await this.brain.removeMany({ ids: allIds, continueOnError: false }) + + // Phase 3: Drop the deleted files' LIVE blob references (chunked). + // Live-reference drops only — the bytes stay for in-window asOf reads + // and are reclaimed by history compaction (see unlink's note). const blobFiles = descendants.filter(d => d.metadata.vfsType === 'file' && d.metadata.storage?.type === 'blob' ) - const BLOB_CHUNK_SIZE = 20 // Parallel delete 20 blobs at a time + const BLOB_CHUNK_SIZE = 20 // Parallel release 20 blob references at a time for (let i = 0; i < blobFiles.length; i += BLOB_CHUNK_SIZE) { const chunk = blobFiles.slice(i, i + BLOB_CHUNK_SIZE) await Promise.all(chunk.map(f => - this.blobStorage.delete(f.metadata.storage!.hash) + this.blobStorage.release(f.metadata.storage!.hash) )) } - - // Phase 3: Batch delete all entities (including root directory) - const allIds = [...descendants.map(d => d.id), entityId] - await this.brain.removeMany({ ids: allIds, continueOnError: false }) } else { // No children or not recursive - just delete the directory entity await this.brain.remove(entityId) diff --git a/src/vfs/types.ts b/src/vfs/types.ts index 54e9902b..9188476b 100644 --- a/src/vfs/types.ts +++ b/src/vfs/types.ts @@ -195,6 +195,36 @@ export interface ReadOptions { // VFS-specific options cache?: boolean // Use cache if available (default: true) decompress?: boolean // Auto-decompress (default: true) + + /** + * Read the file's content as it stood at a past generation (a number) or + * wall-clock instant (a Date) — the temporal read. Serves the exact bytes + * the file held then: the historical entity is materialized from the + * generation history, and its content blob is retention-protected (bytes + * referenced by any in-window generation are never reclaimed). Bounded by + * the retention window: reading past the compaction horizon throws. + * Bypasses the content cache. + */ + asOf?: number | Date +} + +/** + * One entry of a file's version history (see `vfs.history(path)`): the state + * the file held immediately after `generation` committed. + * `readFile(path, { asOf: generation })` returns these exact bytes while the + * generation remains inside the retention window. + */ +export interface FileVersion { + /** The generation whose write produced this version. */ + generation: number + /** Commit timestamp of that generation (ms since epoch). */ + timestamp: number + /** Content hash of this version's bytes in the content-addressed store. */ + hash: string + /** Content size in bytes. */ + size: number + /** Detected MIME type at that version. */ + mimeType?: string } export interface MkdirOptions { diff --git a/tests/integration/ifabsent-upsert-blob-concurrency.test.ts b/tests/integration/ifabsent-upsert-blob-concurrency.test.ts index 66cfdf96..119a5a97 100644 --- a/tests/integration/ifabsent-upsert-blob-concurrency.test.ts +++ b/tests/integration/ifabsent-upsert-blob-concurrency.test.ts @@ -171,18 +171,23 @@ describe('BlobStorage refCount is exact under concurrency (per-hash mutex)', () expect(meta?.refCount).toBe(10) }) - it('the blob survives until the LAST reference drops — no premature deletion', async () => { + it('references drop exactly; bytes survive live-zero (temporal immutability) until reclaim', async () => { const blobs = new BlobStorage(memAdapter() as any) const payload = Buffer.from('shared bytes, two referencing files') const hash = await blobs.write(payload) await blobs.write(payload) // second reference (concurrent-equivalent path) - await blobs.delete(hash) // drop one reference + await blobs.release(hash) // drop one reference expect((await blobs.read(hash)).toString()).toBe(payload.toString()) // still readable expect((await blobs.getMetadata(hash))?.refCount).toBe(1) - await blobs.delete(hash) // last reference + await blobs.release(hash) // last LIVE reference — bytes still exist (history may need them) + expect(await blobs.has(hash)).toBe(true) + expect((await blobs.getMetadata(hash))?.refCount).toBe(0) + + // Reclamation is compaction's job: zero-zero → physically removed. + expect(await blobs.reclaimIfUnreferenced(hash)).toBe(true) expect(await blobs.has(hash)).toBe(false) await expect(blobs.read(hash)).rejects.toThrow() }) @@ -192,15 +197,15 @@ describe('BlobStorage refCount is exact under concurrency (per-hash mutex)', () const payload = Buffer.from('storm payload') const hash = BlobStorage.hash(payload) - // 12 writes and 5 deletes racing: net 7 references, blob alive. + // 12 writes and 5 releases racing: net 7 references, blob alive. await Promise.all([ ...Array.from({ length: 12 }, () => blobs.write(payload)), - ...Array.from({ length: 5 }, () => blobs.delete(hash)) + ...Array.from({ length: 5 }, () => blobs.release(hash)) ]) const meta = await blobs.getMetadata(hash) - // Deletes against a not-yet-written hash floor at zero without deleting, - // so the net can only be >= 12 - 5. The exactness we require: counts are - // never LOST (each landed write is represented). + // Releases against a not-yet-written hash floor at zero, so the net can + // only be >= 12 - 5. The exactness we require: counts are never LOST + // (each landed write is represented). expect(meta?.refCount).toBeGreaterThanOrEqual(7) expect(await blobs.has(hash)).toBe(true) }) diff --git a/tests/integration/temporal-vfs.test.ts b/tests/integration/temporal-vfs.test.ts new file mode 100644 index 00000000..77206ed8 --- /dev/null +++ b/tests/integration/temporal-vfs.test.ts @@ -0,0 +1,200 @@ +/** + * @module tests/integration/temporal-vfs + * @description VFS content joins the Model-B immutability model. Previously + * the temporal model had a hole exactly where files were concerned: entity + * records were versioned per write (before-images), but the content BYTES + * lived under an eager refCount GC — `rm` could physically delete bytes that + * in-window history still referenced, and overwrite never released the old + * hash at all (an unbounded silent leak that only accidentally preserved + * history). Now blob bytes are retention-protected: a content blob referenced + * by any generation inside the retention window (or live) survives, and the + * ONE reclamation point is history compaction. + * + * Pins: + * - `readFile(path, { asOf })` returns each version's exact bytes. + * - `history(path)` lists versions ascending with generation/hash/size. + * - overwrite releases the superseded live reference (leak fixed) while + * history protects the bytes; rm keeps bytes readable via asOf. + * - compaction past the last referencing generation physically reclaims + * bytes (zero live + zero history), and never reclaims in-window content — + * including the cross-file dedup case where an old file's history and a + * newer file's history share one hash. + * - overwrite refreshes the entity's `data` (embedding text) — the stale + * `.data` defect. + * - the backfill/scrub recounts exactly from the generation records. + */ +import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import * as fs from 'node:fs' +import * as os from 'node:os' +import * as path from 'node:path' +import { Brainy } from '../../src/brainy.js' + +describe('temporal VFS — blob immutability under Model B', () => { + let dir: string + let brain: any + + beforeEach(async () => { + process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true' + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-temporal-vfs-')) + brain = new Brainy({ + requireSubtype: false, + storage: { type: 'filesystem', path: dir }, + dimensions: 384, + silent: true + }) + await brain.init() + }) + + afterEach(async () => { + await brain.close() + fs.rmSync(dir, { recursive: true, force: true }) + }) + + /** The file's blob metadata via the storage layer (test introspection). */ + async function blobMeta(hash: string) { + return brain['storage'].blobStorage.getMetadata(hash) + } + + it('readFile(path, {asOf}) returns each version‘s exact bytes; history(path) lists them', async () => { + const p = '/pages/home.json' + await brain.vfs.writeFile(p, 'v1 — original') + const g1 = brain.generationStore.generation() + await brain.vfs.writeFile(p, 'v2 — edited') + await brain.vfs.writeFile(p, 'v3 — final') + + // Live read = newest. + expect((await brain.vfs.readFile(p)).toString()).toBe('v3 — final') + + // History lists the file's write-generations ascending, distinct hashes. + const versions = await brain.vfs.history(p) + expect(versions.length).toBeGreaterThanOrEqual(3) + const gens = versions.map((v: any) => v.generation) + expect([...gens].sort((a: number, b: number) => a - b)).toEqual(gens) + + // Every listed version's exact bytes are readable. + const texts: string[] = [] + for (const v of versions) { + texts.push((await brain.vfs.readFile(p, { asOf: v.generation })).toString()) + } + expect(texts).toContain('v1 — original') + expect(texts).toContain('v2 — edited') + expect(texts).toContain('v3 — final') + + // Direct generation read too (the incident shape: recover the past bytes). + expect((await brain.vfs.readFile(p, { asOf: g1 })).toString()).toBe('v1 — original') + }) + + it('overwrite releases the superseded LIVE reference (leak fixed) while history protects the bytes', async () => { + const p = '/notes/leak.txt' + await brain.vfs.writeFile(p, 'old content A') + const oldHash = (await brain.vfs.history(p))[0].hash + const g1 = brain.generationStore.generation() + + await brain.vfs.writeFile(p, 'new content B') + + // The old hash's LIVE reference is released (previously it leaked forever)… + expect((await blobMeta(oldHash))?.refCount).toBe(0) + // …but the bytes survive (history-protected) and asOf still reads them. + expect((await brain.vfs.readFile(p, { asOf: g1 })).toString()).toBe('old content A') + }) + + it('rm keeps the bytes readable via asOf inside the window (no premature deletion)', async () => { + const p = '/docs/doomed.txt' + await brain.vfs.writeFile(p, 'still recoverable after rm') + const g = brain.generationStore.generation() + const hash = (await brain.vfs.history(p))[0].hash + + await brain.vfs.unlink(p) + + // Live path is gone… + await expect(brain.vfs.readFile(p)).rejects.toThrow() + // …the bytes are not: zero live refs, history-protected. + expect((await blobMeta(hash))?.refCount).toBe(0) + expect((await brain['storage'].blobStorage.read(hash)).toString()).toBe( + 'still recoverable after rm' + ) + // (Path-level asOf resolution of a DELETED path is future work — the + // bytes + entity history are intact; recovery reads go through the hash + // or a pre-delete generation view.) + void g + }) + + it('compaction is the ONE reclamation point: past-window bytes are reclaimed, in-window preserved', async () => { + const p = '/pages/compact.json' + await brain.vfs.writeFile(p, 'version ONE') + const v1Hash = (await brain.vfs.history(p))[0].hash + await brain.vfs.writeFile(p, 'version TWO') + await brain.vfs.writeFile(p, 'version THREE') + + // Sanity: v1's bytes exist (history-protected, live-released). + expect(await blobMeta(v1Hash)).toBeDefined() + expect((await blobMeta(v1Hash))?.refCount).toBe(0) + + // Compact everything reclaimable (retain nothing beyond the live state). + await brain.compactHistory({ maxGenerations: 0 }) + + // v1's bytes are physically GONE — zero live, zero history. + expect(await blobMeta(v1Hash)).toBeUndefined() + // The live version is untouched. + expect((await brain.vfs.readFile(p)).toString()).toBe('version THREE') + }) + + it('cross-file dedup: a shared hash survives while ANY in-window generation references it', async () => { + const shared = 'identical bytes shared across files and time' + const pA = '/a.txt' + const pB = '/b.txt' + + // A holds the shared content, then moves off it (A's history references it). + await brain.vfs.writeFile(pA, shared) + await brain.vfs.writeFile(pA, 'A moved on') + // B now holds the shared content live, then is removed (B's remove-gen + // before-image references it too). + await brain.vfs.writeFile(pB, shared) + const gBLive = brain.generationStore.generation() + const sharedHash = (await brain.vfs.history(pA))[0].hash + await brain.vfs.unlink(pB) + + // Live refs are zero (A moved off; B removed) — bytes survive on history. + expect((await blobMeta(sharedHash))?.refCount).toBe(0) + expect((await brain['storage'].blobStorage.read(sharedHash)).toString()).toBe(shared) + + // B's content at its live generation is still exactly readable… via asOf + // on A's early version too (same bytes, same blob). + const versionsA = await brain.vfs.history(pA) + const aV1 = versionsA[0] + expect((await brain.vfs.readFile(pA, { asOf: aV1.generation })).toString()).toBe(shared) + void gBLive + + // Full compaction drops the last history references → bytes reclaimed. + await brain.compactHistory({ maxGenerations: 0 }) + expect(await blobMeta(sharedHash)).toBeUndefined() + }) + + it('overwrite refreshes the entity data/embedding text (the stale-.data defect)', async () => { + const p = '/pages/data-fresh.txt' + await brain.vfs.writeFile(p, 'first words') + await brain.vfs.writeFile(p, 'second words entirely') + + const entityId = await brain.vfs['pathResolver'].resolve(p) + const entity = await brain.get(entityId) + expect(entity.data).toBe('second words entirely') + }) + + it('the scrub recounts history references exactly from the generation records', async () => { + const p = '/pages/scrubbed.md' + await brain.vfs.writeFile(p, 'scrub v1') + await brain.vfs.writeFile(p, 'scrub v2') + const v1Hash = (await brain.vfs.history(p))[0].hash + await brain.flush() + + const storage = brain['storage'] + const before = (await blobMeta(v1Hash))?.historyRefCount ?? 0 + expect(before).toBeGreaterThan(0) + + // Corrupt the counter (simulates a legacy store / crash drift), then scrub. + await storage.blobStorage.setHistoryRefCount(v1Hash, 0) + const result = await storage.scrubBlobHistoryRefCounts() + expect(result.records).toBeGreaterThan(0) + expect((await blobMeta(v1Hash))?.historyRefCount).toBe(before) + }) +}) diff --git a/tests/unit/storage/blobStorage.test.ts b/tests/unit/storage/blobStorage.test.ts index 8138c64f..55bc5276 100644 --- a/tests/unit/storage/blobStorage.test.ts +++ b/tests/unit/storage/blobStorage.test.ts @@ -258,21 +258,31 @@ describe('BlobStorage', () => { expect(metadata?.refCount).toBe(2) }) - it('should only delete when refCount reaches 0', async () => { + it('release() drops live references; bytes are reclaimed ONLY by reclaimIfUnreferenced at zero-zero', async () => { const data = Buffer.from('test') - // Write twice (refCount = 2) + // Write twice (live refCount = 2) const hash = await blobStorage.write(data) await blobStorage.write(data) - // Delete once (refCount = 1, blob still exists) - await blobStorage.delete(hash) - + // Release once (refCount = 1, blob still exists) + await blobStorage.release(hash) expect(await blobStorage.has(hash)).toBe(true) - // Delete again (refCount = 0, blob deleted) - await blobStorage.delete(hash) + // Release again (refCount = 0) — bytes STILL exist: content is + // immutable under the temporal model; only history compaction reclaims. + await blobStorage.release(hash) + expect(await blobStorage.has(hash)).toBe(true) + expect((await blobStorage.getMetadata(hash))?.refCount).toBe(0) + // A history reference blocks reclamation even at live-zero. + await blobStorage.recordHistoryReference(hash) + expect(await blobStorage.reclaimIfUnreferenced(hash)).toBe(false) + expect(await blobStorage.has(hash)).toBe(true) + + // Drop the history reference → zero-zero → reclaim succeeds. + await blobStorage.releaseHistoryReference(hash) + expect(await blobStorage.reclaimIfUnreferenced(hash)).toBe(true) expect(await blobStorage.has(hash)).toBe(false) }) })