From a3467e1f9b60c8dce290bf0c073a68e0234089dd Mon Sep 17 00:00:00 2001 From: David Snelling Date: Fri, 10 Jul 2026 16:43:48 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20temporal=20VFS=20=E2=80=94=20file=20con?= =?UTF-8?q?tent=20joins=20the=20Model-B=20immutability=20model?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The temporal model had a hole exactly where files were concerned: every entity write is an immutable generation with before-images, but VFS content BYTES lived under an eager refCount GC left over from the pre-8.0 design — unlink could physically destroy bytes that in-window history still referenced, and overwrite never released the old hash at all (an unbounded silent leak whose accidental byproduct was the only thing "preserving" history). Reading the past could therefore return a stale field, a dangling hash, or nothing, depending on luck. Fix: blob reclamation becomes a HISTORY decision instead of a LIVENESS decision. Each blob's metadata now carries historyRefCount alongside the live refCount: - The commit seam counts one history reference per persisted before-image record carrying a content hash (commitTransaction staging and the group-commit flush), recorded BEFORE the record-set persists and carried in the generation delta (blobHashes — always present on new deltas, so compaction only falls back to reading records for pre-contract generations). An aborted transaction compensates best-effort. - unlink/rmdir/overwrite drop ONLY the live reference (BlobStorage.delete → release; overwrite finally releases the superseded hash — cancelling the dedup increment on same-content rewrites and closing the leak), and only AFTER the canonical mutation commits, so a failed delete can never leave a live file whose bytes compaction might reclaim. - History compaction is the ONE reclamation point: after deleting a generation's record-set it releases that set's references and physically reclaims any hash at zero live AND zero history references. Pins are exempt automatically. Crash ordering is over-count-only in every path (record before persist, release after delete), so a crash can leak until the scrub recounts but can never reclaim bytes a retained generation needs. scrubBlobHistoryRefCounts() restores exactness; existing stores get a one-time marker-gated backfill on open, failing into leak-safe mode (reclamation disabled) rather than guessing. On top of the protected history, the temporal API the generational model always implied: - vfs.readFile(path, { asOf }) — the exact bytes as of a generation or Date, materialized from the history (pinned view released so compaction is never blocked by a read). - vfs.history(path) — FileVersion[] ascending ({ generation, timestamp, hash, size, mimeType? }), the newest entry being the live state. - Overwrites now refresh the file entity's data/embedding text — semantic search and the data field previously served the FIRST version's text forever (the stale-field defect a consumer's incident recovery depended on by luck). Integration suite (temporal-vfs.test.ts): per-version exact reads + history listing, leak-fix + history protection on overwrite, rm keeps bytes readable, compaction reclaims past-window bytes and preserves in-window (including the cross-file dedup case where an old file's history and a newer file's removal share one hash), data freshness, and scrub exactness. --- docs/guides/reacting-to-changes.md | 5 +- docs/guides/snapshots-and-time-travel.md | 55 +++++ src/brainy.ts | 13 ++ src/db/generationStore.ts | 113 +++++++++- src/db/types.ts | 32 +++ src/index.ts | 3 + src/storage/baseStorage.ts | 142 +++++++++++++ src/storage/blobStorage.ts | 115 ++++++++-- src/vfs/VirtualFileSystem.ts | 194 +++++++++++++++-- src/vfs/types.ts | 30 +++ .../ifabsent-upsert-blob-concurrency.test.ts | 21 +- tests/integration/temporal-vfs.test.ts | 200 ++++++++++++++++++ tests/unit/storage/blobStorage.test.ts | 24 ++- 13 files changed, 890 insertions(+), 57 deletions(-) create mode 100644 tests/integration/temporal-vfs.test.ts 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) }) })