diff --git a/CHANGELOG.md b/CHANGELOG.md index ac6b9c4..4d6cb8d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Performance + +- **Bounded immutable page payload reuse** - repeated `pages.get()` calls now + share successful in-flight and completed payload reads through an LRU bounded + by both entry count and aggregate bytes. Callers receive copies, failures + remain retryable, oversized values do not displace unrelated residents, and + `pages.open()` remains uncached and streaming. + ## [6.5.0] — 2026-07-18 ### Added diff --git a/GUIDE.md b/GUIDE.md index 3f9ef5c..4311bc5 100644 --- a/GUIDE.md +++ b/GUIDE.md @@ -163,6 +163,12 @@ lookup reads only the descriptor path and selected payload; retention and publication perform full bounded graph validation. Identical page bytes and identically constructed bundles produce identical handles. +Repeated `pages.get()` calls reuse immutable payload reads within the store's +bounded page cache. The defaults retain at most 128 payloads and 8 MiB; use +`pageCacheEntries` and `pageCacheBytes` to tune that ceiling. Every result is a +caller-owned copy. `pages.open()` remains uncached and streaming, and neither +form of access establishes retention. + ### Generic Application Publication Publication is disabled until the facade receives an explicit application-ref diff --git a/docs/API.md b/docs/API.md index 9d732ec..82232e6 100644 --- a/docs/API.md +++ b/docs/API.md @@ -80,6 +80,8 @@ new ContentAddressableStore(options); - `options.maxRestoreBufferSize` (optional): Max bytes for buffered encrypted/compressed restore (default: 536870912 / 512 MiB) - `options.maxBlobSize` (optional): Max bytes for metadata blob reads (default: 10485760 / 10 MiB) - `options.maxPageSize` (optional): Maximum immutable page bytes (default: 16777216 / 16 MiB) +- `options.pageCacheEntries` (optional): Maximum immutable page payloads retained in memory (default: 128) +- `options.pageCacheBytes` (optional): Maximum immutable page payload bytes retained in memory (default: 8388608 / 8 MiB) - `options.bundleLimits` (optional): Repository maximums for bundle members, path bytes, descriptor bytes, fanout entries, and fanout depth - `options.maxBundleNestingDepth` (optional): Maximum nested bundle depth (default: 32) - `options.compressionAdapter` (optional): CompressionPort implementation (default: NodeCompressionAdapter) @@ -1025,6 +1027,19 @@ metadata before streaming it. `get()` additionally collects the page under its effective byte bound. An imported handle above the configured maximum fails with `PAGE_TOO_LARGE` without materializing the blob. +Successful `get()` payloads share in-flight and completed immutable reads inside +one store instance. The default LRU retains at most 128 payloads and 8 MiB of +payload bytes; `pageCacheEntries` and `pageCacheBytes` may lower or raise those +bounds. Every result is copied for its caller, failed reads remain retryable, +and a payload larger than the byte budget is returned without remaining +resident. Cold misses validate object type and size before reading. Resident or +in-flight hits reuse that immutable OID validation and still enforce the current +caller's `maxBytes`. In-flight work is coalesced separately from completed LRU +residence, so its transient memory follows caller concurrency and the per-page +size bound. `open()` remains streaming and does not enter the payload cache. +This reuse is not retention evidence; callers must keep the page reachable for +the duration of use. + The canonical page token is: ```text diff --git a/docs/design/0051-bounded-page-payload-reuse/bounded-page-payload-reuse.md b/docs/design/0051-bounded-page-payload-reuse/bounded-page-payload-reuse.md new file mode 100644 index 0000000..6255d51 --- /dev/null +++ b/docs/design/0051-bounded-page-payload-reuse/bounded-page-payload-reuse.md @@ -0,0 +1,146 @@ +--- +schema: 'method.design/v1' +cycle: '0051' +slug: 'bounded-page-payload-reuse' +status: 'active' +sponsor_human: 'James Ross' +sponsor_agent: 'Codex' +created: '2026-07-18' +updated: '2026-07-18' +issue: 'https://github.com/git-stunts/git-cas/issues/85' +milestone: 'v6.5.1' +--- + +# PERF-0051 - Bounded Page Payload Reuse + +## Linked Issue + +- [#85 - perf: bound and reuse immutable page payload reads](https://github.com/git-stunts/git-cas/issues/85) + +## Decision Summary + +Reuse successful `pages.get()` payload reads in a process-local LRU keyed by +immutable page OID. Residence is bounded independently by entry count and total +payload bytes. Every caller receives a copy, rejected reads are removed, and a +value larger than the byte budget is returned to its current caller without +remaining resident or evicting unrelated entries. + +`pages.open()` remains an uncached streaming operation. The cache is an +acceleration of the already bounded collecting API, not a reason to buffer the +streaming API. + +## Current Truth + +- v6.5.0 coalesces immutable Git object metadata, exact tree entries, and bundle + descriptor bytes. +- `PageService.get()` still starts a fresh `cat-file blob` stream for every call. +- A git-warp retained-property fixture performed 16 identical reads in 2.61 + seconds, including 32 blob reads and a 1.80-second time to first reading. +- Page OIDs are immutable, and `get()` already enforces both the configured page + limit and the caller's lower operation limit. + +## Scope + +- Add page payload cache entry and byte bounds to `ContentAddressableStore`. +- Share successful in-flight and completed `get()` work by page OID. +- Preserve per-call `maxBytes` checks for warm and cold reads. +- Clone cached bytes before returning them. +- Prove rejection retry, LRU eviction, byte-budget behavior, and real-Git warm + command elimination. +- Document the residency and lifetime assumptions. + +## Non-Goals + +- Caching `pages.open()` streams. +- Caching unbounded assets, arbitrary Git blobs, refs, or mutable collection + state. +- Establishing retention. A cache hit is not a retention witness. +- Surviving destructive external pruning that races an active store instance. +- Solving path-local persistent bundle updates; that work is tracked in + [#86](https://github.com/git-stunts/git-cas/issues/86). + +## Runtime Contract + +```ts +interface ContentAddressableStoreOptions { + maxPageSize?: number; // default 16 MiB + pageCacheEntries?: number; // default 128 + pageCacheBytes?: number; // default 8 MiB +} +``` + +The cache allocates no fixed arena. Its actual residence is the sum of retained +payload byte lengths plus bounded completed-entry overhead. In-flight reads are +tracked separately so capacity pressure cannot break coalescing; their memory +follows caller concurrency and the configured per-page size bound. A configured +byte bound of zero disables non-empty completed payload residence while +preserving in-flight coalescing. + +## Correctness Invariants + +1. Cache identity is the immutable Git blob OID carried by a validated + `PageHandle`. +2. Every call parses its `PageHandle`. A cold payload miss validates object type + and size before reading; a resident or in-flight hit reuses that immutable + OID validation without another metadata access. +3. Cold misses check the caller's effective byte limit before starting a payload + read. Every resolved payload, including warm and shared in-flight hits, is + checked again against the current caller's limit. +4. Returned `Uint8Array` values never alias resident bytes. +5. Rejected work is absent before its returned promise settles. +6. An individually oversized resolved value cannot evict unrelated residents. +7. Streaming page reads never enter the payload cache. + +## Residency Posture + +The defaults retain at most 128 page payloads and 8 MiB of payload bytes. Both +bounds are configurable downward. Payloads larger than the byte budget remain +readable but are not retained. This keeps correctness independent of cache +capacity and lets low-heap consumers explicitly choose their residency ceiling. + +## Compatibility + +No persisted formats, handles, refs, witnesses, or existing method signatures +change. Constructor options are additive. `pages.get()` returns the same bytes +and errors as before; only repeat-read command count changes. + +## Alternatives Considered + +### Cache page bytes in git-warp + +Rejected. git-cas owns immutable CAS access and cache policy. A consumer-local +cache would duplicate authority and produce inconsistent residency controls. + +### Cache all blob streams in GitPersistenceAdapter + +Rejected. Persistence serves assets and metadata with different boundedness and +streaming contracts. `PageService` is the layer that knows a payload is an +admitted bounded page. + +### Cache `pages.open()` + +Rejected. It would silently turn a streaming API into a collecting API and make +memory behavior depend on access history. + +### Retain no payloads + +Rejected by measurement. Immutable metadata reuse alone leaves one Git process +and full payload read per repeated exact observation. + +## Proof Surface + +- Unit tests cover concurrent coalescing, caller-copy isolation, entry LRU, + aggregate byte bounds, oversized-value behavior, rejected-read retry, and + option validation. +- A Docker real-Git test proves an identical warm `pages.get()` performs zero + additional Git commands. +- git-warp reruns its retained-property CPU, wall, command-count, and bounded + memory scenarios against the published patch release. + +## Acceptance Criteria + +- `pages.get()` warm reads issue no payload Git command. +- Residence is bounded by entries and bytes. +- Streaming semantics and retention authority are unchanged. +- Public declarations and API documentation expose exact defaults. +- Node, Bun, and Deno release verification remains green. diff --git a/docs/design/README.md b/docs/design/README.md index ceee34a..678311f 100644 --- a/docs/design/README.md +++ b/docs/design/README.md @@ -11,6 +11,7 @@ process in [docs/method/process.md](../method/process.md). ## Active METHOD Cycles +- [0051-bounded-page-payload-reuse - bounded-page-payload-reuse](./0051-bounded-page-payload-reuse/bounded-page-payload-reuse.md) - [0050-lazy-bundle-reference-reads - lazy-bundle-reference-reads](./0050-lazy-bundle-reference-reads/lazy-bundle-reference-reads.md) - [0049-scoped-staging-workspaces — scoped-staging-workspaces](./0049-scoped-staging-workspaces/scoped-staging-workspaces.md) - [0048-scoped-cache-acquisitions — scoped-cache-acquisitions](./0048-scoped-cache-acquisitions/scoped-cache-acquisitions.md) diff --git a/docs/releases/v6.5.1.md b/docs/releases/v6.5.1.md new file mode 100644 index 0000000..8a57ff4 --- /dev/null +++ b/docs/releases/v6.5.1.md @@ -0,0 +1,35 @@ +# git-cas v6.5.1 Release Notes + +v6.5.1 completes the bounded immutable read path introduced in v6.5.0 by +coalescing repeated `pages.get()` payload reads. Exact application reads no +longer start another Git blob process when the same retained page is already +resident in the store's bounded payload cache. + +## Bounded Payload Reuse + +The page cache defaults to at most 128 entries and 8 MiB of payload bytes. It +allocates no fixed arena. Applications can configure the limits with +`pageCacheEntries` and `pageCacheBytes` when constructing the store. + +Every `get()` result is a caller-owned copy. Concurrent reads share in-flight +work, rejected reads are immediately retryable, and values larger than the byte +budget are returned to the current caller without remaining resident or +evicting unrelated entries. In-flight work is tracked separately from completed +LRU residence, so capacity pressure cannot break coalescing; transient memory +still follows caller concurrency and the configured per-page size bound. + +`pages.open()` is unchanged: it remains a streaming operation and never enters +the payload cache. Cache residence is not a retention witness, so applications +must continue to keep handles reachable through the documented lifecycle APIs. + +## Performance Evidence + +The real-Git command-count integration contract proves that an identical warm +`pages.get()` performs zero additional Git commands. git-warp will publish its +end-to-end CPU, wall-clock, Git-command, and bounded-memory deltas after adopting +this release. + +## Compatibility + +This release changes no persisted formats, handles, refs, witnesses, or existing +method signatures. The two constructor options are additive and optional. diff --git a/index.d.ts b/index.d.ts index f396a57..083c376 100644 --- a/index.d.ts +++ b/index.d.ts @@ -706,6 +706,10 @@ export interface ContentAddressableStoreOptions { maxBlobSize?: number; /** Maximum immutable page size in bytes. @default 16777216 (16 MiB) */ maxPageSize?: number; + /** Maximum immutable page payloads retained in memory. @default 128 */ + pageCacheEntries?: number; + /** Maximum immutable page payload bytes retained in memory. @default 8388608 (8 MiB) */ + pageCacheBytes?: number; /** Repository-wide maximum bundle admission limits. */ bundleLimits?: Partial; /** Maximum nested bundle depth. @default 32 */ diff --git a/index.js b/index.js index 387a7e7..1f734ea 100644 --- a/index.js +++ b/index.js @@ -133,6 +133,8 @@ export default class ContentAddressableStore { * @param {number} [options.maxRestoreBufferSize=536870912] - Max buffered restore size in bytes for encrypted/compressed restores (default 512 MiB). * @param {number} [options.maxBlobSize=10485760] - Safety limit for readBlob metadata in bytes (default 10 MiB). * @param {number} [options.maxPageSize=16777216] - Maximum immutable page size in bytes (default 16 MiB). + * @param {number} [options.pageCacheEntries=128] - Maximum immutable page payloads retained in memory. + * @param {number} [options.pageCacheBytes=8388608] - Maximum immutable page payload bytes retained in memory. * @param {object} [options.bundleLimits] - Repository-wide maximum bundle admission limits. * @param {number} [options.maxBundleNestingDepth=32] - Maximum nested bundle depth. * @param {import('./src/ports/CompressionPort.js').default} [options.compressionAdapter] - Compression adapter (default NodeCompressionAdapter). @@ -153,6 +155,8 @@ export default class ContentAddressableStore { maxRestoreBufferSize, maxBlobSize, maxPageSize, + pageCacheEntries, + pageCacheBytes, bundleLimits, maxBundleNestingDepth, compressionAdapter, @@ -173,6 +177,8 @@ export default class ContentAddressableStore { maxRestoreBufferSize, maxBlobSize, maxPageSize, + pageCacheEntries, + pageCacheBytes, bundleLimits, maxBundleNestingDepth, compressionAdapter, @@ -231,7 +237,7 @@ export default class ContentAddressableStore { }); } - /** @type {{ plumbing: *, chunkSize?: number, codec?: *, policy?: *, crypto?: *, observability?: *, merkleThreshold?: number, concurrency?: number, chunking?: *, chunker?: *, maxRestoreBufferSize?: number, maxBlobSize?: number, maxPageSize?: number, bundleLimits?: object, maxBundleNestingDepth?: number, compressionAdapter?: *, applicationRefPrefixes?: string[], clock?: { now(): Date } }} */ + /** @type {{ plumbing: *, chunkSize?: number, codec?: *, policy?: *, crypto?: *, observability?: *, merkleThreshold?: number, concurrency?: number, chunking?: *, chunker?: *, maxRestoreBufferSize?: number, maxBlobSize?: number, maxPageSize?: number, pageCacheEntries?: number, pageCacheBytes?: number, bundleLimits?: object, maxBundleNestingDepth?: number, compressionAdapter?: *, applicationRefPrefixes?: string[], clock?: { now(): Date } }} */ #config; /** @type {AssetService|null} */ #assetService = null; @@ -369,6 +375,8 @@ export default class ContentAddressableStore { this.#pageService = new PageService({ persistence: this.service.persistence, maxPageSize: cfg.maxPageSize, + pageCacheEntries: cfg.pageCacheEntries, + pageCacheBytes: cfg.pageCacheBytes, clock: cfg.clock, }); this.#bundleService = new BundleService({ @@ -620,6 +628,9 @@ export default class ContentAddressableStore { * @param {import('./src/ports/ChunkingPort.js').default} [options.chunker] - Pre-built ChunkingPort instance. * @param {number} [options.maxRestoreBufferSize=536870912] - Max buffered restore size in bytes. * @param {number} [options.maxBlobSize=10485760] - Safety limit for readBlob metadata in bytes. + * @param {number} [options.maxPageSize=16777216] - Maximum immutable page size in bytes. + * @param {number} [options.pageCacheEntries=128] - Maximum immutable page payloads retained in memory. + * @param {number} [options.pageCacheBytes=8388608] - Maximum immutable page payload bytes retained in memory. * @param {import('./src/ports/CompressionPort.js').default} [options.compressionAdapter] - Compression adapter. * @returns {Promise} */ @@ -644,6 +655,9 @@ export default class ContentAddressableStore { * @param {import('./src/ports/ChunkingPort.js').default} [options.chunker] - Pre-built ChunkingPort instance. * @param {number} [options.maxRestoreBufferSize=536870912] - Max buffered restore size in bytes. * @param {number} [options.maxBlobSize=10485760] - Safety limit for readBlob metadata in bytes. + * @param {number} [options.maxPageSize=16777216] - Maximum immutable page size in bytes. + * @param {number} [options.pageCacheEntries=128] - Maximum immutable page payloads retained in memory. + * @param {number} [options.pageCacheBytes=8388608] - Maximum immutable page payload bytes retained in memory. * @param {import('./src/ports/CompressionPort.js').default} [options.compressionAdapter] - Compression adapter. * @returns {ContentAddressableStore} */ @@ -665,6 +679,9 @@ export default class ContentAddressableStore { * @param {import('./src/ports/ChunkingPort.js').default} [options.chunker] - Pre-built ChunkingPort instance. * @param {number} [options.maxRestoreBufferSize=536870912] - Max buffered restore size in bytes. * @param {number} [options.maxBlobSize=10485760] - Safety limit for readBlob metadata in bytes. + * @param {number} [options.maxPageSize=16777216] - Maximum immutable page size in bytes. + * @param {number} [options.pageCacheEntries=128] - Maximum immutable page payloads retained in memory. + * @param {number} [options.pageCacheBytes=8388608] - Maximum immutable page payload bytes retained in memory. * @param {import('./src/ports/CompressionPort.js').default} [options.compressionAdapter] - Compression adapter. * @returns {ContentAddressableStore} */ diff --git a/src/domain/services/PageService.js b/src/domain/services/PageService.js index 5bec8e2..e26871e 100644 --- a/src/domain/services/PageService.js +++ b/src/domain/services/PageService.js @@ -4,8 +4,11 @@ import { ErrorCodes } from '../errors/index.js'; import { assertHandleObjectType, mapHandleTargetError } from '../helpers/handleTarget.js'; import PageHandle from '../value-objects/PageHandle.js'; import StagedPage from '../value-objects/StagedPage.js'; +import BoundedPromiseCache from '../../helpers/boundedPromiseCache.js'; export const DEFAULT_MAX_PAGE_SIZE = 16 * 1024 * 1024; +const DEFAULT_PAGE_CACHE_ENTRIES = 128; +const DEFAULT_PAGE_CACHE_BYTES = 8 * 1024 * 1024; const DEFAULT_CLOCK = Object.freeze({ now: () => new Date() }); /** @@ -14,19 +17,34 @@ const DEFAULT_CLOCK = Object.freeze({ now: () => new Date() }); export default class PageService { #clock; #maxPageSize; + #payloads; #persistence; /** * @param {object} options * @param {import('../../ports/GitPersistencePort.js').default} options.persistence * @param {number} [options.maxPageSize] + * @param {number} [options.pageCacheEntries] + * @param {number} [options.pageCacheBytes] * @param {{ now(): Date }} [options.clock] */ - constructor({ persistence, maxPageSize = DEFAULT_MAX_PAGE_SIZE, clock = DEFAULT_CLOCK }) { + constructor({ + persistence, + maxPageSize = DEFAULT_MAX_PAGE_SIZE, + pageCacheEntries = DEFAULT_PAGE_CACHE_ENTRIES, + pageCacheBytes = DEFAULT_PAGE_CACHE_BYTES, + clock = DEFAULT_CLOCK, + }) { PageService.#assertDependencies(persistence, clock); PageService.#assertLimit(maxPageSize, 'Configured max page size'); + PageService.#assertPositiveLimit(pageCacheEntries, 'Page cache entries'); + PageService.#assertLimit(pageCacheBytes, 'Page cache bytes'); this.#persistence = persistence; this.#maxPageSize = maxPageSize; + this.#payloads = new BoundedPromiseCache(pageCacheEntries, { + maxWeight: pageCacheBytes, + weightOf: (value) => value.byteLength, + }); this.#clock = clock; } @@ -64,7 +82,24 @@ export default class PageService { * @returns {Promise} */ async get({ handle, maxBytes }) { - return await PageService.#collect(this.open({ handle }), this.#effectiveLimit(maxBytes)); + const limit = this.#effectiveLimit(maxBytes); + const pageHandle = PageHandle.from(handle); + let payload = this.#payloads.get(pageHandle.oid); + if (payload === undefined) { + const root = await this.resolveRoot(pageHandle); + if (root.size > limit) { + throw PageService.#tooLarge(root.size, limit); + } + payload = this.#payloads.getOrCreate( + root.oid, + async () => await this.#readRoot(root), + ); + } + const bytes = await payload; + if (bytes.byteLength > limit) { + throw PageService.#tooLarge(bytes.byteLength, limit); + } + return new Uint8Array(bytes); } /** @@ -106,6 +141,17 @@ export default class PageService { return value; } + async #readRoot(root) { + try { + return await PageService.#collect( + await this.#persistence.readBlobStream(root.oid), + this.#maxPageSize, + ); + } catch (error) { + throw mapHandleTargetError(error, root.handle); + } + } + #observedAt() { const now = this.#clock.now(); if (!(now instanceof Date) || Number.isNaN(now.getTime())) { @@ -163,6 +209,14 @@ export default class PageService { } } + static #assertPositiveLimit(value, label) { + if (!Number.isSafeInteger(value) || value < 1) { + throw createCasError(`${label} must be a positive safe integer`, ErrorCodes.INVALID_OPTIONS, { + value, + }); + } + } + static #assertDependencies(persistence, clock) { const methods = ['writeBlob', 'readBlobStream', 'readObjectType', 'readObjectSize']; if (!persistence || methods.some((method) => typeof persistence[method] !== 'function')) { diff --git a/src/helpers/boundedPromiseCache.js b/src/helpers/boundedPromiseCache.js index 2bb1bb9..5d68ca0 100644 --- a/src/helpers/boundedPromiseCache.js +++ b/src/helpers/boundedPromiseCache.js @@ -4,7 +4,9 @@ */ export default class BoundedPromiseCache { /** @type {Map, weight: number }>} */ - #entries = new Map(); + #completed = new Map(); + /** @type {Map>} */ + #inFlight = new Map(); #maxEntries; #maxWeight; #totalWeight = 0; @@ -34,6 +36,21 @@ export default class BoundedPromiseCache { this.#weightOf = weightOf; } + /** + * @template T + * @param {string} key + * @returns {Promise|undefined} + */ + get(key) { + const cached = this.#completed.get(key); + if (cached !== undefined) { + this.#completed.delete(key); + this.#completed.set(key, cached); + return cached.promise; + } + return this.#inFlight.get(key); + } + /** * @template T * @param {string} key @@ -41,45 +58,30 @@ export default class BoundedPromiseCache { * @returns {Promise} */ getOrCreate(key, factory) { - const cached = this.#entries.get(key); - if (cached !== undefined) { - this.#entries.delete(key); - this.#entries.set(key, cached); - return cached.promise; + const present = this.get(key); + if (present !== undefined) { + return present; } const source = Promise.resolve().then(factory); - const entry = { promise: source, weight: 0 }; const tracked = source.then( (value) => { - if (this.#entries.get(key) === entry) { - let weight; - try { - weight = this.#resolvedWeight(value); - } catch (error) { - this.#remove(key, entry); - throw error; - } - if (weight > this.#maxWeight) { - this.#remove(key, entry); - return value; - } - entry.weight = weight; - this.#totalWeight += entry.weight; - this.#evict(); + this.#removeInFlight(key, tracked); + const weight = this.#resolvedWeight(value); + if (weight <= this.#maxWeight) { + const entry = { promise: Promise.resolve(value), weight }; + this.#completed.set(key, entry); + this.#totalWeight += weight; + this.#evictCompleted(); } return value; }, (error) => { - if (this.#entries.get(key) === entry) { - this.#remove(key, entry); - } + this.#removeInFlight(key, tracked); throw error; }, ); - entry.promise = tracked; - this.#entries.set(key, entry); - this.#evict(); + this.#inFlight.set(key, tracked); return tracked; } @@ -91,18 +93,24 @@ export default class BoundedPromiseCache { return weight; } - #evict() { - while (this.#entries.size > this.#maxEntries || this.#totalWeight > this.#maxWeight) { - const oldestKey = this.#entries.keys().next().value; - const oldest = this.#entries.get(oldestKey); - this.#remove(oldestKey, oldest); + #evictCompleted() { + while (this.#completed.size > this.#maxEntries || this.#totalWeight > this.#maxWeight) { + const oldestKey = this.#completed.keys().next().value; + const oldest = this.#completed.get(oldestKey); + this.#removeCompleted(oldestKey, oldest); } } - #remove(key, entry) { - if (entry === undefined || !this.#entries.delete(key)) { + #removeCompleted(key, entry) { + if (entry === undefined || !this.#completed.delete(key)) { return; } this.#totalWeight -= entry.weight; } + + #removeInFlight(key, promise) { + if (this.#inFlight.get(key) === promise) { + this.#inFlight.delete(key); + } + } } diff --git a/test/integration/bundle-reference-performance.test.js b/test/integration/bundle-reference-performance.test.js index a42da69..b87075d 100644 --- a/test/integration/bundle-reference-performance.test.js +++ b/test/integration/bundle-reference-performance.test.js @@ -108,6 +108,23 @@ afterAll(() => { rmSync(repoDir, { recursive: true, force: true }); }); +describe('real-Git immutable page payload reads', () => { + it('performs zero additional Git commands for an identical warm page read', async () => { + const reader = await countingReader(); + const page = await writer.pages.put({ source: Buffer.from('warm page payload') }); + + await expect(reader.cas.pages.get({ handle: page.handle })).resolves.toEqual( + new Uint8Array(Buffer.from('warm page payload')), + ); + const cold = reader.snapshot(); + await reader.cas.pages.get({ handle: page.handle }); + const warm = delta(reader.snapshot(), cold); + + expect(count(cold, 'cat-file')).toBeGreaterThan(0); + expect(total(warm)).toBe(0); + }); +}); + describe('real-Git direct bundle reference reads', () => { it('coalesces immutable Git metadata across repeated targeted reads', async () => { const reader = await countingReader(); diff --git a/test/unit/domain/services/PageService.test.js b/test/unit/domain/services/PageService.test.js index d77cd31..26e893f 100644 --- a/test/unit/domain/services/PageService.test.js +++ b/test/unit/domain/services/PageService.test.js @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import PageService from '../../../../src/domain/services/PageService.js'; import PageHandle from '../../../../src/domain/value-objects/PageHandle.js'; import StagedPage from '../../../../src/domain/value-objects/StagedPage.js'; @@ -6,11 +6,20 @@ import MemoryPersistenceAdapter from '../../../helpers/MemoryPersistenceAdapter. const OBSERVED_AT = '2026-07-13T11:00:00.000Z'; -function makePages(maxPageSize = 1024) { +function deferred() { + let resolve; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} + +function makePages(maxPageSize = 1024, cache = {}) { const persistence = new MemoryPersistenceAdapter(); const pages = new PageService({ persistence, maxPageSize, + ...cache, clock: { now: () => new Date(OBSERVED_AT) }, }); return { pages, persistence }; @@ -71,6 +80,219 @@ describe('PageService', () => { }); }); +describe('PageService immutable payload reuse', () => { + it('coalesces immutable payload reads and returns caller-owned byte copies', async () => { + const { pages, persistence } = makePages(); + const bytes = Buffer.from('cached immutable page'); + const staged = await pages.put({ source: bytes }); + const readBlobStream = vi.spyOn(persistence, 'readBlobStream'); + const readObjectType = vi.spyOn(persistence, 'readObjectType'); + const readObjectSize = vi.spyOn(persistence, 'readObjectSize'); + + const [first, concurrent] = await Promise.all([ + pages.get({ handle: staged.handle }), + pages.get({ handle: staged.handle }), + ]); + first[0] = 0; + const metadataReads = readObjectType.mock.calls.length + readObjectSize.mock.calls.length; + const warm = await pages.get({ handle: staged.handle }); + + expect(concurrent).toEqual(new Uint8Array(bytes)); + expect(warm).toEqual(new Uint8Array(bytes)); + expect(readBlobStream).toHaveBeenCalledTimes(1); + expect(readObjectType.mock.calls.length + readObjectSize.mock.calls.length).toBe(metadataReads); + }); +}); + +describe('PageService resident payload limits', () => { + it('enforces a lower operation limit on a resident payload without rereading metadata', async () => { + const { pages, persistence } = makePages(); + const staged = await pages.put({ source: Buffer.from('abcde') }); + const readBlobStream = vi.spyOn(persistence, 'readBlobStream'); + const readObjectType = vi.spyOn(persistence, 'readObjectType'); + const readObjectSize = vi.spyOn(persistence, 'readObjectSize'); + + await pages.get({ handle: staged.handle }); + const metadataReads = readObjectType.mock.calls.length + readObjectSize.mock.calls.length; + + await expect(pages.get({ handle: staged.handle, maxBytes: 4 })).rejects.toMatchObject({ + code: 'PAGE_TOO_LARGE', + meta: { observedBytes: 5, maxBytes: 4 }, + }); + expect(readBlobStream).toHaveBeenCalledTimes(1); + expect(readObjectType.mock.calls.length + readObjectSize.mock.calls.length).toBe(metadataReads); + }); +}); + +describe('PageService entry-count residency', () => { + it('evicts by entry count and rereads the least-recently-used payload', async () => { + const { pages, persistence } = makePages(1024, { + pageCacheEntries: 2, + pageCacheBytes: 1024, + }); + const first = await pages.put({ source: Buffer.from('first') }); + const second = await pages.put({ source: Buffer.from('second') }); + const third = await pages.put({ source: Buffer.from('third') }); + const readBlobStream = vi.spyOn(persistence, 'readBlobStream'); + + await pages.get({ handle: first.handle }); + await pages.get({ handle: second.handle }); + await pages.get({ handle: first.handle }); + await pages.get({ handle: third.handle }); + await pages.get({ handle: first.handle }); + await pages.get({ handle: second.handle }); + + expect(readBlobStream).toHaveBeenCalledTimes(4); + }); +}); + +describe('PageService in-flight payload reuse', () => { + it('coalesces A/B/A contention beyond the completed-entry bound', async () => { + const { pages, persistence } = makePages(1024, { + pageCacheEntries: 1, + pageCacheBytes: 1024, + }); + const first = await pages.put({ source: Buffer.from('first') }); + const second = await pages.put({ source: Buffer.from('second') }); + const gate = deferred(); + const readBlobStream = persistence.readBlobStream.bind(persistence); + const read = vi.spyOn(persistence, 'readBlobStream').mockImplementation(async (oid) => { + await gate.promise; + return await readBlobStream(oid); + }); + + const firstRead = pages.get({ handle: first.handle }); + const secondRead = pages.get({ handle: second.handle }); + const repeatedFirstRead = pages.get({ handle: first.handle }); + await vi.waitFor(() => expect(read).toHaveBeenCalledTimes(2)); + gate.resolve(); + await Promise.all([firstRead, secondRead, repeatedFirstRead]); + + expect(read).toHaveBeenCalledTimes(2); + }); +}); + +describe('PageService in-flight caller limits', () => { + it('enforces each caller limit when callers share one in-flight payload', async () => { + const { pages, persistence } = makePages(); + const staged = await pages.put({ source: Buffer.from('abcde') }); + const gate = deferred(); + const readBlobStream = persistence.readBlobStream.bind(persistence); + const read = vi.spyOn(persistence, 'readBlobStream').mockImplementation(async (oid) => { + await gate.promise; + return await readBlobStream(oid); + }); + + const permissive = pages.get({ handle: staged.handle }); + const restrictive = pages.get({ handle: staged.handle, maxBytes: 4 }); + const permissiveResult = expect(permissive).resolves.toEqual( + new Uint8Array(Buffer.from('abcde')), + ); + const restrictiveResult = expect(restrictive).rejects.toMatchObject({ + code: 'PAGE_TOO_LARGE', + meta: { observedBytes: 5, maxBytes: 4 }, + }); + await vi.waitFor(() => expect(read).toHaveBeenCalledTimes(1)); + gate.resolve(); + + await Promise.all([permissiveResult, restrictiveResult]); + expect(read).toHaveBeenCalledTimes(1); + }); +}); + +describe('PageService payload byte residency', () => { + it('coalesces in-flight work without retaining completed payloads at a zero-byte bound', async () => { + const { pages, persistence } = makePages(1024, { + pageCacheEntries: 4, + pageCacheBytes: 0, + }); + const staged = await pages.put({ source: Buffer.from('transient') }); + const readBlobStream = vi.spyOn(persistence, 'readBlobStream'); + + await Promise.all([ + pages.get({ handle: staged.handle }), + pages.get({ handle: staged.handle }), + ]); + await pages.get({ handle: staged.handle }); + + expect(readBlobStream).toHaveBeenCalledTimes(2); + }); + + it('evicts least-recently-used payloads when their aggregate bytes exceed the bound', async () => { + const { pages, persistence } = makePages(1024, { + pageCacheEntries: 4, + pageCacheBytes: 5, + }); + const first = await pages.put({ source: Buffer.from('abc') }); + const second = await pages.put({ source: Buffer.from('def') }); + const readBlobStream = vi.spyOn(persistence, 'readBlobStream'); + + await pages.get({ handle: first.handle }); + await pages.get({ handle: second.handle }); + await pages.get({ handle: second.handle }); + await pages.get({ handle: first.handle }); + + expect(readBlobStream).toHaveBeenCalledTimes(3); + }); +}); + +describe('PageService oversized payload residency', () => { + it('does not let an oversized pending payload evict unrelated cached bytes', async () => { + const { pages, persistence } = makePages(1024, { + pageCacheEntries: 4, + pageCacheBytes: 4, + }); + const small = await pages.put({ source: Buffer.from('abc') }); + const oversized = await pages.put({ source: Buffer.from('12345') }); + const oversizedGate = deferred(); + const readBlobStream = persistence.readBlobStream.bind(persistence); + const read = vi.spyOn(persistence, 'readBlobStream').mockImplementation(async (oid) => { + if (oid === oversized.handle.oid) { + await oversizedGate.promise; + } + return await readBlobStream(oid); + }); + + await pages.get({ handle: small.handle }); + const pending = pages.get({ handle: oversized.handle }); + await vi.waitFor(() => expect(read).toHaveBeenCalledTimes(2)); + await pages.get({ handle: small.handle }); + oversizedGate.resolve(); + await pending; + await pages.get({ handle: small.handle }); + await pages.get({ handle: oversized.handle }); + + expect(read).toHaveBeenCalledTimes(3); + }); +}); + +describe('PageService payload cache failure and configuration', () => { + it('removes rejected payload reads so a later call can retry', async () => { + const { pages, persistence } = makePages(); + const staged = await pages.put({ source: Buffer.from('retry') }); + const readBlobStream = persistence.readBlobStream.bind(persistence); + const read = vi.spyOn(persistence, 'readBlobStream') + .mockRejectedValueOnce(new Error('transient read failure')) + .mockImplementation(async (oid) => await readBlobStream(oid)); + + await expect(pages.get({ handle: staged.handle })).rejects.toThrow('transient read failure'); + await expect(pages.get({ handle: staged.handle })).resolves.toEqual( + new Uint8Array(Buffer.from('retry')), + ); + + expect(read).toHaveBeenCalledTimes(2); + }); + + it.each([ + ['pageCacheEntries', { pageCacheEntries: 0 }], + ['pageCacheBytes', { pageCacheBytes: -1 }], + ])('rejects invalid %s bounds', (_name, cache) => { + expect(() => makePages(1024, cache)).toThrowError( + expect.objectContaining({ code: 'INVALID_OPTIONS' }), + ); + }); +}); + describe('PageService target and source validation', () => { it('rejects a page source that is neither bytes nor iterable', async () => { const { pages } = makePages(); diff --git a/test/unit/facade/ContentAddressableStore.application-storage.test.js b/test/unit/facade/ContentAddressableStore.application-storage.test.js index 555b70c..5d2c5d8 100644 --- a/test/unit/facade/ContentAddressableStore.application-storage.test.js +++ b/test/unit/facade/ContentAddressableStore.application-storage.test.js @@ -72,6 +72,19 @@ describe('ContentAddressableStore application storage capabilities', () => { }); }); +describe('ContentAddressableStore page cache configuration', () => { + it.each([ + ['pageCacheEntries', { pageCacheEntries: 0 }], + ['pageCacheBytes', { pageCacheBytes: -1 }], + ])('threads invalid %s bounds to the page service', async (_name, cache) => { + const cas = new ContentAddressableStore({ plumbing: mockPlumbing(), ...cache }); + + await expect(cas.pages.put({ source: Buffer.from('unused') })).rejects.toMatchObject({ + code: 'INVALID_OPTIONS', + }); + }); +}); + describe('ContentAddressableStore caches', () => { it('opens a cache set through the facade namespace', async () => { const cas = new ContentAddressableStore({ plumbing: mockPlumbing() }); diff --git a/test/unit/helpers/boundedPromiseCache.test.js b/test/unit/helpers/boundedPromiseCache.test.js index 8fc207d..ef3b66a 100644 --- a/test/unit/helpers/boundedPromiseCache.test.js +++ b/test/unit/helpers/boundedPromiseCache.test.js @@ -1,6 +1,14 @@ import { describe, expect, it, vi } from 'vitest'; import BoundedPromiseCache from '../../../src/helpers/boundedPromiseCache.js'; +function deferred() { + let resolve; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} + describe('BoundedPromiseCache construction', () => { it('rejects invalid residency options', () => { expect(() => new BoundedPromiseCache(0)).toThrow('positive safe integer'); @@ -37,6 +45,28 @@ describe('BoundedPromiseCache coalescing', () => { await expect(cache.getOrCreate('retry', retryFactory)).resolves.toBe('recovered'); expect(retryFactory).toHaveBeenCalledTimes(2); }); + + it('keeps unique in-flight work coalesced beyond the completed-entry bound', async () => { + const cache = new BoundedPromiseCache(1); + const firstGate = deferred(); + const secondGate = deferred(); + const firstFactory = vi.fn().mockReturnValue(firstGate.promise); + const secondFactory = vi.fn().mockReturnValue(secondGate.promise); + + const first = cache.getOrCreate('first', firstFactory); + const second = cache.getOrCreate('second', secondFactory); + const repeated = cache.getOrCreate('first', firstFactory); + firstGate.resolve('first-value'); + secondGate.resolve('second-value'); + + await expect(Promise.all([first, second, repeated])).resolves.toEqual([ + 'first-value', + 'second-value', + 'first-value', + ]); + expect(firstFactory).toHaveBeenCalledTimes(1); + expect(secondFactory).toHaveBeenCalledTimes(1); + }); }); describe('BoundedPromiseCache rejected work', () => { @@ -90,7 +120,9 @@ describe('BoundedPromiseCache residency', () => { expect(first).toHaveBeenCalledTimes(2); expect(second).toHaveBeenCalledTimes(1); }); +}); +describe('BoundedPromiseCache oversized residency', () => { it('does not retain a value larger than the weight bound', async () => { const cache = new BoundedPromiseCache(3, { maxWeight: 2, @@ -107,6 +139,26 @@ describe('BoundedPromiseCache residency', () => { expect(oversized).toHaveBeenCalledTimes(2); expect(resident).toHaveBeenCalledTimes(1); }); + + it('does not let an oversized in-flight value displace a resident', async () => { + const cache = new BoundedPromiseCache(1, { + maxWeight: 2, + weightOf: (value) => value.length, + }); + const resident = vi.fn().mockResolvedValue('ok'); + const oversizedGate = deferred(); + const oversized = vi.fn().mockReturnValue(oversizedGate.promise); + + await cache.getOrCreate('resident', resident); + const pending = cache.getOrCreate('oversized', oversized); + await cache.getOrCreate('resident', resident); + oversizedGate.resolve('oversized'); + await pending; + await cache.getOrCreate('resident', resident); + + expect(oversized).toHaveBeenCalledTimes(1); + expect(resident).toHaveBeenCalledTimes(1); + }); }); describe('BoundedPromiseCache resolved weights', () => { diff --git a/test/unit/types/declaration-accuracy.test.js b/test/unit/types/declaration-accuracy.test.js index abdaae2..bab8980 100644 --- a/test/unit/types/declaration-accuracy.test.js +++ b/test/unit/types/declaration-accuracy.test.js @@ -113,6 +113,8 @@ describe('Application-storage declaration accuracy', () => { ); expect(declarations).toContain('iterateMembers(options: { handle: BundleHandleInput })'); expect(declarations).toContain('applicationRefPrefixes?: string[];'); + expect(declarations).toContain('pageCacheEntries?: number;'); + expect(declarations).toContain('pageCacheBytes?: number;'); expect(declarations).toContain('parentOids?: string[];'); expect(declarations).toContain('anchorRef?(options: {'); expect(declarations).toContain('deleteRef?(options: {');