diff --git a/lib/cache/store.mjs b/lib/cache/store.mjs index aefc976..eb9127b 100644 --- a/lib/cache/store.mjs +++ b/lib/cache/store.mjs @@ -51,6 +51,75 @@ * @property {number} inflightCount */ +// ── D57 — ADR 0005 Amendment 8 streaming-singleflight shapes ────────────── + +/** + * @typedef {Object} StreamingInflightEntry + * @property {string} compositeKey - `${keyId}\0${cacheKey}` + * @property {AsyncIterator<*>|null} source - underlying source iterator (null while factory pending) + * @property {AbortController} sourceAbortController - propagates "all clients gone" to the source + * @property {Array<*>} accumulatedChunks - late-joiner replay buffer (bounded by §10) + * @property {number} accumulatedByteSize - running byte size of accumulatedChunks + * @property {boolean} accumulatedReplayCapExceeded - true once §10 cap hit; cache write will skip + * @property {Set} attachedClients - all live clients tee'ing this source + * @property {boolean} factoryPending - true while sourceFactory() is awaited + * @property {Array<{ resolve: function, reject: function }>} pendingJoiners - late joiners arriving during factoryPending + * @property {boolean} sourceDone - source iterator exhausted normally + * @property {Error|null} sourceError - non-null if source threw + * @property {boolean} sourceAborted - true if AbortController fired + * @property {number} ttlMs - TTL to use when writing the completed accumulated chunks to cache + */ + +/** + * @typedef {Object} AttachedClient + * @property {string} id - request id / correlator + * @property {Array<*>} queue - per-client tee buffer + * @property {number} queueByteSize - running byte size sum + * @property {boolean} yieldedAccumulated - true after late-joiner replay drained + * @property {boolean} done - terminal sentinel hit + * @property {boolean} backpressured - true once STREAM_BACKPRESSURE terminator scheduled + * @property {Error|null} error - non-null if source threw or replay-drain over-cap + * @property {((chunk: { value: *, done: boolean }) => void)|null} resolveNext - pending pull promise resolver + * @property {((err: Error) => void)|null} rejectNext - pending pull promise rejecter + */ + +// ADR 0005 Amendment 8 §14 — implementation defaults. Per-call overrides flow +// via the `opts` argument to getOrComputeStreaming (used by tests to exercise +// caps cheaply without producing megabytes of fixture data). +export const PER_CLIENT_QUEUE_CAP_DEFAULT = 1 * 1024 * 1024; +export const ACCUMULATED_REPLAY_CAP_DEFAULT = 10 * 1024 * 1024; + +/** + * Returns an approximate byte size for a chunk. The tee + cap math uses + * JSON.stringify length as the serialization yardstick (matches the cache + * store's existing size accounting via Buffer.byteLength(JSON.stringify(...))). + * Non-stringifiable values (circular, etc.) fall back to 0 — the tee continues + * but the size accounting under-estimates that chunk. In practice IR chunks + * are always JSON-safe. + * + * @param {*} chunk + * @returns {number} + */ +function _chunkByteSize(chunk) { + try { + return Buffer.byteLength(JSON.stringify(chunk) ?? '', 'utf8'); + } catch { + return 0; + } +} + +/** + * Synthesises a STREAM_BACKPRESSURE terminator stream (ADR 0005 Amendment 8 §8): + * yields `{ type: 'stop', finish_reason: 'length' }` then a `[DONE]` sentinel + * then ends. Used both for late-joiner-too-late and per-client overflow paths. + * + * @returns {AsyncGenerator<*>} + */ +async function* _backpressureTerminator() { + yield { type: 'stop', finish_reason: 'length' }; + yield '[DONE]'; +} + // ── CacheStore ──────────────────────────────────────────────────────────── export class CacheStore { @@ -82,6 +151,14 @@ export class CacheStore { /** @type {Map>} */ this._inflight = new Map(); + // D57 — ADR 0005 Amendment 8: streaming singleflight per-(keyId,cacheKey) + // inflight Map. Composite key uses `\0` separator per §2 (avoids colliding + // with keyId or cacheKey content). The check + insert against this Map is + // synchronous in getOrComputeStreaming (no `await` between read & write), + // mirroring D38 `tryAcquireSpawn` atomicity. + /** @type {Map} */ + this._streamingInflight = new Map(); + // Stats per keyId /** @type {Map} */ this._stats = new Map(); @@ -266,13 +343,10 @@ export class CacheStore { * @param {number} [ttlMs] * @returns {Promise<*>} * - * TODO(v1.x — ADR 0005 Amendment 8 / issue #16): add a sibling - * `getOrComputeStreaming(keyId, cacheKey, sourceFactory)` for the streaming - * path. This API handles buffered responses only; the streaming branch in - * server.mjs currently uses a peek+spawn pattern with a TOCTOU window. - * The streaming sibling will mirror this method's shape but with a tee - * fan-out and per-client backpressure queues. See docs/v1x-roadmap.md #1 - * for the design contract and acceptance criteria. + * D57 — ADR 0005 Amendment 8 / issue #16 resolved at the cache layer: + * the sibling `getOrComputeStreaming` method below provides tee-fan-out + + * per-client backpressure for the streaming path. Server-side wiring lands + * separately in D58. */ async getOrCompute(keyId, cacheKey, computeFn, ttlMs) { // 1. Cache hit — return immediately, no singleflight overhead @@ -311,6 +385,493 @@ export class CacheStore { return computePromise; } + /** + * D57 — ADR 0005 Amendment 8 (issue #16): streaming singleflight + tee-fan-out. + * + * Streaming-path sibling of `getOrCompute`. Coordinates concurrent identical + * streaming requests so only one source spawn occurs, all attached clients + * receive identical chunk sequences in order, late joiners are replayed from + * an accumulated buffer, slow clients are disconnected with a synthetic + * STREAM_BACKPRESSURE terminator instead of stalling the source, and the + * source is aborted when all clients disconnect. + * + * The Map check + insert against `_streamingInflight` is synchronous — no + * `await` between read and write — matching the D38 `tryAcquireSpawn` + * atomicity invariant and collapsing the original TOCTOU window in + * `server.mjs:782` (peek + spawn). See §1 + §6. + * + * Three outcomes (Amendment 8 §1): + * - **cache_hit**: cached entry exists and is alive → returns + * `{ stream: , isFirst: false, + * role: 'cache_hit' }`. No spawn. `hits` incremented. + * - **attached**: inflight entry exists in `_streamingInflight` → attaches + * a new AttachedClient. Late-joiner replay (§5) drains accumulated + * chunks synchronously; if drain would exceed `perClientQueueCap` the + * client receives a STREAM_BACKPRESSURE synthesised stream instead. + * `isFirst: false`, `role: 'attached'`. `hits` incremented (sharing a + * spawn is functionally a "cache-like" benefit; documented choice). + * - **source**: no cache hit, no inflight entry → the cache layer takes + * the inflight lock synchronously (placeholder entry inserted), then + * `await sourceFactory()`. If the factory throws (e.g. + * CONCURRENCY_LIMIT) the placeholder is removed and the error + * propagates. On success the source is wired up, the tee task starts, + * `isFirst: true`, `role: 'source'`. `misses` incremented. + * + * **`role` enum** (returned at attach-time): + * - `source` — first caller; entry created. Lifetime-end may upgrade to + * `solo` (no joiners ever attached) at the server (D58); the cache + * layer reports `source` at attach-time and does not flip mid-stream. + * - `attached` — joined an existing inflight entry. + * - `cache_hit` — served from cache; no entry created. + * + * **Amendment 8 §11 header values**: the X-OLP-Streaming-Inflight header + * (D58) uses `source | attached | solo`. The cache layer's `cache_hit` is + * the "served from cache without inflight" case; D58 chooses whether to + * emit `solo` or omit the header for that path. The cache layer reports + * `cache_hit` as a distinct role so the server can disambiguate. + * + * **§14 defaults**: `PER_CLIENT_QUEUE_CAP = 1 MB`, `ACCUMULATED_REPLAY_CAP + * = 10 MB`. Both overridable via `opts` for cheap test exercise of the cap + * paths. + * + * @param {string} keyId + * @param {string} cacheKey + * @param {() => Promise>|AsyncIterator<*>} sourceFactory + * Invoked exactly once per inflight lifetime (only on first caller). + * Returns the source async iterator. May throw (e.g. CONCURRENCY_LIMIT + * from `tryAcquireSpawn`); on throw, the inflight entry is removed and + * the error propagates to the first caller. Late joiners that attached + * while the factory was pending are rejected with the same error. + * @param {object} [opts] + * @param {string} [opts.clientId] - correlator id (defaults to incrementing counter) + * @param {number} [opts.ttlMs] - TTL for completed accumulated-chunks cache write + * @param {number} [opts.perClientQueueCap] - override §14 default (1 MB) + * @param {number} [opts.accumulatedReplayCap] - override §14 default (10 MB) + * @returns {Promise<{ stream: AsyncGenerator<*>, isFirst: boolean, role: 'source'|'attached'|'cache_hit' }>} + */ + async getOrComputeStreaming(keyId, cacheKey, sourceFactory, opts = {}) { + const compositeKey = `${keyId}\0${cacheKey}`; + const perClientQueueCap = opts.perClientQueueCap ?? PER_CLIENT_QUEUE_CAP_DEFAULT; + const accumulatedReplayCap = opts.accumulatedReplayCap ?? ACCUMULATED_REPLAY_CAP_DEFAULT; + const ttlMs = opts.ttlMs; + const clientId = opts.clientId ?? `c-${this._nowFn()}-${Math.floor(Math.random() * 1e9).toString(36)}`; + + // ── (A) Inflight Map check FIRST (Amendment 8 §6 — TTL race) ─────────── + // Late joiners that arrive after a cache entry has expired but during an + // active source spawn must still attach via the inflight Map rather than + // re-spawn. The synchronous Map.get + (if hit) Map preservation here + // satisfies the no-`await`-between-check-and-decision invariant. + const inflightEntry = this._streamingInflight.get(compositeKey); + if (inflightEntry) { + // Hits-as-share decision (documented at method header): sharing a spawn + // is a cache-like benefit; increment hits for consistency with + // cache_hit accounting and to expose the singleflight win in stats(). + this._getStats(keyId).hits++; + const stream = this._attachClient(inflightEntry, { + clientId, + perClientQueueCap, + }); + return { stream, isFirst: false, role: 'attached' }; + } + + // ── (B) Cache hit check (no inflight) ────────────────────────────────── + // Replays cached chunks via a synthetic async iterator. No source spawn. + const ns = this._getNamespace(keyId); + const existing = ns.get(cacheKey); + if (existing && this._isAlive(existing)) { + this._getStats(keyId).hits++; + const cachedChunks = Array.isArray(existing.value) ? existing.value : [existing.value]; + const stream = (async function* cacheReplay() { + for (const chunk of cachedChunks) { + yield chunk; + } + })(); + return { stream, isFirst: false, role: 'cache_hit' }; + } + + // ── (C) Miss + no inflight: take the lock synchronously, then await ─── + // The placeholder entry is inserted BEFORE invoking sourceFactory so that + // late joiners arriving while the factory is awaited see the inflight + // entry and attach (they're parked in `pendingJoiners` until the factory + // resolves or rejects). If the factory throws, the placeholder is + // removed and the error propagates to the first caller AND all parked + // joiners. This preserves the §1 invariant: the Map insert is atomic + // from later joiners' perspective. + const entry = /** @type {StreamingInflightEntry} */ ({ + compositeKey, + source: null, + sourceAbortController: new AbortController(), + accumulatedChunks: [], + accumulatedByteSize: 0, + accumulatedReplayCapExceeded: false, + attachedClients: new Set(), + factoryPending: true, + pendingJoiners: [], + sourceDone: false, + sourceError: null, + sourceAborted: false, + ttlMs, + }); + this._streamingInflight.set(compositeKey, entry); + this._getStats(keyId).misses++; + + // Attach the first caller synchronously so any subsequent joiners during + // the factory await see the same set/topology as the first caller. + const firstStream = this._attachClient(entry, { + clientId, + perClientQueueCap, + }); + + let sourceIter; + try { + const factoryResult = sourceFactory(); + sourceIter = factoryResult && typeof factoryResult.then === 'function' + ? await factoryResult + : factoryResult; + } catch (err) { + // Factory rejected — remove placeholder and reject first caller + any + // late joiners that arrived during the await. + this._streamingInflight.delete(compositeKey); + for (const client of entry.attachedClients) { + if (client.rejectNext) { + client.rejectNext(err); + client.resolveNext = null; + client.rejectNext = null; + } + client.error = err; + client.done = true; + } + throw err; + } + + entry.source = sourceIter; + entry.factoryPending = false; + + // Kick off the tee task. It runs detached; lifetime is bounded by the + // source iterator's completion / error / abort. + this._teeStreamingSource(keyId, cacheKey, entry, { + accumulatedReplayCap, + }); + + return { stream: firstStream, isFirst: true, role: 'source' }; + } + + /** + * D57 — ADR 0005 Amendment 8 §3 + §5: attach a new client to an inflight + * entry. Synchronously drains the accumulated replay buffer into the + * client's queue (§5). If the drain would exceed `perClientQueueCap`, the + * client receives a STREAM_BACKPRESSURE synthesised stream INSTEAD of the + * normal tee — the source continues for the other clients. + * + * Returns the async iterator the caller will consume. + * + * @private + */ + _attachClient(entry, { clientId, perClientQueueCap }) { + // Late-joiner replay drain cap check (§5 + §10): a late joiner cannot + // catch up if either + // (a) the accumulated buffer alone would overflow the per-client cap + // (burst > PER_CLIENT_QUEUE_CAP), or + // (b) the replay buffer is already truncated (§10 cap was hit and + // further source chunks were not appended to accumulatedChunks), + // so even a successful drain would give the joiner a partial view + // that disagrees with later live chunks. + // Either condition → STREAM_BACKPRESSURE synthetic terminator. The + // source / other clients are unaffected. + if ( + entry.accumulatedByteSize > perClientQueueCap + || entry.accumulatedReplayCapExceeded + ) { + this._warnFn('stream_backpressure_disconnect', { + client_id: clientId, + queue_byte_size: entry.accumulatedByteSize, + per_client_cap: perClientQueueCap, + composite_key: entry.compositeKey, + reason: entry.accumulatedReplayCapExceeded + ? 'replay_cap_truncated' + : 'replay_drain_over_cap', + }); + return _backpressureTerminator(); + } + + /** @type {AttachedClient} */ + const client = { + id: clientId, + queue: [], + queueByteSize: 0, + yieldedAccumulated: false, + done: false, + backpressured: false, + error: null, + resolveNext: null, + rejectNext: null, + // Per-client cap is captured here so the tee task can apply it + // without re-plumbing opts; documented as an internal field. + __perClientQueueCap__: perClientQueueCap, + }; + + // Synchronous replay drain — push every accumulated chunk into the + // client's queue at attach-time. From this point on the tee task pushes + // live chunks. + for (const chunk of entry.accumulatedChunks) { + client.queue.push(chunk); + client.queueByteSize += _chunkByteSize(chunk); + } + client.yieldedAccumulated = true; + entry.attachedClients.add(client); + + // TODO(D58 — ADR 0005 Amendment 8 §11): emit `streaming_inflight_join` + // event from the server-layer wrapper, which has provider/model context. + // Cache layer alone does not have provider/model identity (sourceFactory + // is a closure), so the join event lives at the consumer of `role: + // 'attached'` in server.mjs. D57 reviewer P2-3 follow-up. + + // If source already completed before this attach (last-second join) + // mark the client as terminal-after-drain so the iterator returns + // cleanly once the replay queue is drained. + if (entry.sourceDone) { + client.done = true; + } else if (entry.sourceError) { + client.error = entry.sourceError; + client.done = true; + } + + // Per-client AbortController for client-side cancellation (HTTP close). + // The async iterator's return() removes the client from attachedClients; + // if the entry's attachedClients size hits zero, the tee task aborts the + // source. The teardown logic lives in the iterator below. + const store = this; + const iterator = (async function* clientStream() { + try { + while (true) { + // Drain queue chunks first. + if (client.queue.length > 0) { + const next = client.queue.shift(); + client.queueByteSize -= _chunkByteSize(next); + yield next; + continue; + } + // Backpressure-terminated client: yield the synthetic terminator. + if (client.backpressured) { + yield { type: 'stop', finish_reason: 'length' }; + yield '[DONE]'; + client.done = true; + return; + } + // Source already errored. + if (client.error) { + throw client.error; + } + // Source already completed and no more queued chunks. + if (client.done) { + return; + } + // Block until the tee task pushes the next chunk (or signals + // source-done / source-error / backpressure). + await new Promise((resolve, reject) => { + client.resolveNext = resolve; + client.rejectNext = reject; + }); + client.resolveNext = null; + client.rejectNext = null; + } + } finally { + // Iterator return() fired (HTTP close, break, or normal return) — + // remove client and possibly trigger source abort. + if (entry.attachedClients.has(client)) { + entry.attachedClients.delete(client); + } + // If we're the last client AND the source is still running, fire + // the AbortController so the source iterator's return() / cleanup + // can reap any underlying resources. + if ( + entry.attachedClients.size === 0 + && !entry.sourceDone + && !entry.sourceError + && !entry.sourceAborted + && !entry.factoryPending + ) { + entry.sourceAborted = true; + try { + entry.sourceAbortController.abort(); + } catch { + // best-effort + } + // Tee task observes attachedClients.size === 0 + sourceAborted on + // its next loop iteration and exits without a cache write. + store._streamingInflight.delete(entry.compositeKey); + store._warnFn('streaming_inflight_abort', { + composite_key: entry.compositeKey, + accumulated_chunk_count: entry.accumulatedChunks.length, + }); + } + } + })(); + + return iterator; + } + + /** + * D57 — ADR 0005 Amendment 8 §4: tee fan-out task. One reader pulls from + * `entry.source`; on each chunk, pushes to `accumulatedChunks` (bounded by + * §10) and to every attached client's queue (per-client cap from §8). + * + * On source completion: writes accumulated chunks to cache if (a) cap not + * exceeded and (b) `set()`'s own `maxEntryBytes` cap admits it. Resolves + * all clients to drain-out state. Removes entry. + * + * On source error: rejects all clients via their `rejectNext`. No cache + * write. Removes entry. + * + * Source-abort short-circuit: if `attachedClients.size === 0` after a push, + * fires `sourceAbortController.abort()`, exits without cache write. + * + * @private + */ + _teeStreamingSource(keyId, cacheKey, entry, { accumulatedReplayCap }) { + const store = this; + (async () => { + try { + for (;;) { + // Pre-check: if all clients have already gone away before we even + // pull the next chunk, abort the source and bail out. (The + // per-client iterator's finally-block sets sourceAborted=true and + // removes the entry; we just need to stop pulling.) + if (entry.attachedClients.size === 0 && !entry.factoryPending) { + if (!entry.sourceAborted) { + entry.sourceAborted = true; + try { entry.sourceAbortController.abort(); } catch { /* best-effort */ } + } + // Try to call return() on the source so the underlying generator + // cleans up. Best-effort; not all iterators implement it. + try { + if (entry.source && typeof entry.source.return === 'function') { + await entry.source.return(); + } + } catch { /* best-effort */ } + return; + } + + const result = await entry.source.next(); + if (result.done) break; + const chunk = result.value; + const size = _chunkByteSize(chunk); + + // §10 — replay buffer cap. Past the cap we stop accumulating (so + // future late joiners can still see the chunks they need to + // catch up to live), but we mark the entry not-cacheable so the + // §4 completion path skips the cache write. + if (!entry.accumulatedReplayCapExceeded) { + if (entry.accumulatedByteSize + size > accumulatedReplayCap) { + entry.accumulatedReplayCapExceeded = true; + store._warnFn('streaming_inflight_replay_cap_exceeded', { + composite_key: entry.compositeKey, + accumulated_byte_size: entry.accumulatedByteSize, + chunk_size: size, + accumulated_replay_cap: accumulatedReplayCap, + }); + // Continue accumulating up to this chunk so existing late + // joiners' drain decision was based on the size they saw at + // attach-time. We do NOT push this chunk to accumulatedChunks + // (it would corrupt the "<= cap at attach-time" invariant for + // future joiners). Future joiners arriving past this point + // see accumulatedByteSize already > cap and get the + // backpressure terminator at attach-time per §5. + } else { + entry.accumulatedChunks.push(chunk); + entry.accumulatedByteSize += size; + } + } + + // Fan out to each client synchronously (no await inside the for- + // each-client loop). Disconnecting a client is mutation-during- + // iteration; we snapshot the set first. + const clientsSnapshot = [...entry.attachedClients]; + for (const client of clientsSnapshot) { + // Per-client backpressure (§8). If the push would exceed the + // per-client cap, disconnect this client only. + if (client.queueByteSize + size > client.__perClientQueueCap__) { + client.backpressured = true; + store._warnFn('stream_backpressure_disconnect', { + client_id: client.id, + queue_byte_size: client.queueByteSize, + per_client_cap: client.__perClientQueueCap__, + composite_key: entry.compositeKey, + reason: 'queue_overflow', + }); + // Remove from set so future fan-out skips this client. + entry.attachedClients.delete(client); + // Wake the client's pull-promise so it can yield the synthetic + // STREAM_BACKPRESSURE terminator. + if (client.resolveNext) { + const r = client.resolveNext; + client.resolveNext = null; + client.rejectNext = null; + r({ value: undefined, done: false }); + } + continue; + } + client.queue.push(chunk); + client.queueByteSize += size; + if (client.resolveNext) { + const r = client.resolveNext; + client.resolveNext = null; + client.rejectNext = null; + r({ value: undefined, done: false }); + } + } + + // If the fan-out emptied the attached set (all over-cap), the + // top-of-loop pre-check will fire on the next iteration and abort. + } + + // Source iterator returned normally. + entry.sourceDone = true; + const cacheWritten = + !entry.accumulatedReplayCapExceeded + && entry.accumulatedChunks.length > 0; + if (cacheWritten) { + // ADR 0005 Amendment 8 §4: write accumulated chunks to cache via + // the standard set() path (which itself applies the D23 + // maxEntryBytes cap — separate from the §10 replay cap). + await store.set(keyId, cacheKey, entry.accumulatedChunks, entry.ttlMs); + } + store._warnFn('streaming_inflight_source_done', { + composite_key: entry.compositeKey, + attached_count: entry.attachedClients.size, + accumulated_chunk_count: entry.accumulatedChunks.length, + cache_written: cacheWritten, + }); + // Wake every remaining attached client so they drain their queue and + // observe `done = true`. + for (const client of [...entry.attachedClients]) { + client.done = true; + if (client.resolveNext) { + const r = client.resolveNext; + client.resolveNext = null; + client.rejectNext = null; + r({ value: undefined, done: true }); + } + } + store._streamingInflight.delete(entry.compositeKey); + } catch (err) { + // Source threw mid-stream. Reject every attached client. + entry.sourceError = err; + for (const client of [...entry.attachedClients]) { + client.error = err; + client.done = true; + if (client.rejectNext) { + const rej = client.rejectNext; + client.resolveNext = null; + client.rejectNext = null; + rej(err); + } + } + store._streamingInflight.delete(entry.compositeKey); + } + })(); + } + /** * Returns stats for a specific keyId, or aggregate stats across all keyIds. * @@ -322,11 +883,14 @@ export class CacheStore { const s = this._stats.get(keyId) ?? { hits: 0, misses: 0 }; const ns = this._store.get(keyId); const size = ns ? ns.size : 0; + // D57 — ADR 0005 Amendment 8 §1: inflightCount aggregates both the + // buffered-path singleflight Map and the streaming-path inflight Map + // so stats() reflects all active dedup-coordination entries. return { hits: s.hits, misses: s.misses, size, - inflightCount: this._inflight.size, + inflightCount: this._inflight.size + this._streamingInflight.size, }; } @@ -345,7 +909,8 @@ export class CacheStore { hits: totalHits, misses: totalMisses, size: totalSize, - inflightCount: this._inflight.size, + // D57 — see per-keyId branch above; streaming entries counted alongside buffered. + inflightCount: this._inflight.size + this._streamingInflight.size, }; } @@ -397,10 +962,18 @@ export class CacheStore { this._inflight.delete(k); } } + // D57 — also clear streaming inflight entries scoped to this keyId. + // Composite key uses `\0` separator (see _streamingInflight init). + for (const k of this._streamingInflight.keys()) { + if (k.startsWith(`${keyId}\0`)) { + this._streamingInflight.delete(k); + } + } } else { this._store.clear(); this._stats.clear(); this._inflight.clear(); + this._streamingInflight.clear(); } } } diff --git a/lib/providers/base.mjs b/lib/providers/base.mjs index bccbed6..0fb0140 100644 --- a/lib/providers/base.mjs +++ b/lib/providers/base.mjs @@ -162,6 +162,15 @@ export const PROVIDER_ERROR_CODES = /** @type {const} */ ([ 'SPAWN_FAILED', 'SPAWN_TIMEOUT', // ADR 0004 § Trigger taxonomy bullet 4: spawn timeout is a hard trigger 'CONCURRENCY_LIMIT', // ADR 0002 Amendment 6 / ADR 0004 Amendment 4 (D38, issue #1) + /* ADR 0005 Amendment 8 §8 (D57): per-client streaming queue overflow. + * NOT a hard trigger — the source spawned successfully and other attached + * clients continue to receive chunks; only this client's queue exceeded + * PER_CLIENT_QUEUE_CAP. The affected client receives a synthetic + * { type: 'stop', finish_reason: 'length' } + [DONE] terminator. + * D58 wires server-side header/log surface; HARD_TRIGGER_CODES in + * lib/fallback/engine.mjs is a whitelist so absence here gives the + * correct default (no fallback advancement). */ + 'STREAM_BACKPRESSURE', ]); export class ProviderError extends Error { diff --git a/test-features.mjs b/test-features.mjs index 4831c8d..4a2dd6e 100644 --- a/test-features.mjs +++ b/test-features.mjs @@ -12035,3 +12035,426 @@ describe('Suite 26 — D52 audit rotation (Phase 3, ADR 0008 § 5)', () => { }); }); }); + +// ── Suite 27: D57 streaming singleflight (cache layer) ───────────────────── +// +// Authority: ADR 0005 Amendment 8 §§1-11, 14 (issue #16, D57). +// +// Cache-layer unit tests for the streaming-singleflight tee fan-out. Each +// test constructs a fake sourceFactory that returns an async generator +// producing a fixed chunk sequence; the cache store coordinates dedup + +// late-joiner replay + per-client backpressure. No real provider CLIs are +// spawned; no HTTP requests issued. D58 will wire this into server.mjs in a +// separate PR (Iron Rule 11). +// +// Test count: 12 (one per Amendment 8 §§1-11 + §14 fixture + composite-key +// isolation). Aim for +12 tests minimum per D57 prompt. + +describe('Suite 27 — D57 streaming singleflight (cache layer)', () => { + // Helper: build a deterministic async source. Each yielded chunk waits a + // microtask so the tee/queue dynamics are observable across concurrent + // attached clients. + function makeChunkSequence(chunks, opts = {}) { + let returnedCount = 0; + const gen = (async function* fakeStream() { + try { + for (const c of chunks) { + if (opts.signal && opts.signal.aborted) return; + yield c; + await new Promise(r => setImmediate(r)); + } + } finally { + returnedCount++; + if (opts.onReturn) opts.onReturn(returnedCount); + } + })(); + return gen; + } + + // D57 — ADR 0005 Amendment 8 §1: cache-hit + single client + source-mode + it('27a — single client streaming: behaviour identical to today (source role, full sequence)', async () => { + const store = new CacheStore({ _warnFn: () => {} }); + let spawns = 0; + const factory = () => { + spawns++; + return makeChunkSequence(['a', 'b', 'c']); + }; + const r = await store.getOrComputeStreaming('k1', 'ck-27a', factory); + assert.equal(r.isFirst, true); + assert.equal(r.role, 'source'); + const out = []; + for await (const c of r.stream) out.push(c); + assert.deepEqual(out, ['a', 'b', 'c']); + assert.equal(spawns, 1); + // After completion, the cache should have an entry for replay. + const cached = await store.get('k1', 'ck-27a'); + assert.ok(cached, 'cache populated post-completion'); + assert.deepEqual(cached.value, ['a', 'b', 'c']); + }); + + // D57 — ADR 0005 Amendment 8 §1, §4: 2 concurrent identical streams + it('27b — 2 concurrent identical streams: only 1 sourceFactory call; identical chunks in order', async () => { + const store = new CacheStore({ _warnFn: () => {} }); + let spawns = 0; + const factory = () => { + spawns++; + return makeChunkSequence(['x', 'y', 'z']); + }; + const p1 = store.getOrComputeStreaming('k', 'ck-27b', factory, { clientId: 'A' }); + const p2 = store.getOrComputeStreaming('k', 'ck-27b', factory, { clientId: 'B' }); + const [r1, r2] = await Promise.all([p1, p2]); + assert.equal(r1.isFirst, true); + assert.equal(r1.role, 'source'); + assert.equal(r2.isFirst, false); + assert.equal(r2.role, 'attached'); + const out1 = []; + const out2 = []; + const c1 = (async () => { for await (const c of r1.stream) out1.push(c); })(); + const c2 = (async () => { for await (const c of r2.stream) out2.push(c); })(); + await Promise.all([c1, c2]); + assert.equal(spawns, 1, 'sourceFactory invoked exactly once'); + assert.deepEqual(out1, ['x', 'y', 'z']); + assert.deepEqual(out2, ['x', 'y', 'z']); + }); + + // D57 — ADR 0005 Amendment 8 §5: mid-stream join (replay burst + live tail) + // + post-completion join (cache_hit) + it('27c — 3 concurrent, mid-stream join: A=source, B=attached (burst + tail), C=cache_hit', async () => { + const store = new CacheStore({ _warnFn: () => {} }); + let spawns = 0; + // Six chunks. A iterates fast; B attaches after A has consumed ~3. + const factory = () => { + spawns++; + return makeChunkSequence(['1', '2', '3', '4', '5', '6']); + }; + const r1 = await store.getOrComputeStreaming('k', 'ck-27c', factory, { clientId: 'A' }); + assert.equal(r1.role, 'source'); + + // A iterates the first 3 chunks before B attaches. + const itA = r1.stream; + const outA = []; + const n1 = await itA.next(); outA.push(n1.value); + const n2 = await itA.next(); outA.push(n2.value); + const n3 = await itA.next(); outA.push(n3.value); + + // Now B attaches mid-stream. Its replay drain picks up everything in + // accumulatedChunks at attach-time + the live tail thereafter. + const r2 = await store.getOrComputeStreaming('k', 'ck-27c', factory, { clientId: 'B' }); + assert.equal(r2.role, 'attached'); + + // A finishes consuming. + const outA2 = []; + for await (const c of itA) outA2.push(c); + const outB = []; + for await (const c of r2.stream) outB.push(c); + + assert.deepEqual([...outA, ...outA2], ['1', '2', '3', '4', '5', '6']); + // B receives the full sequence too (burst replays earlier chunks + live tail). + assert.deepEqual(outB, ['1', '2', '3', '4', '5', '6']); + assert.equal(spawns, 1, 'still only 1 spawn'); + + // C attaches AFTER source completion → cache_hit (not inflight, not respawn). + const r3 = await store.getOrComputeStreaming('k', 'ck-27c', factory, { clientId: 'C' }); + assert.equal(r3.role, 'cache_hit'); + assert.equal(r3.isFirst, false); + const outC = []; + for await (const c of r3.stream) outC.push(c); + assert.deepEqual(outC, ['1', '2', '3', '4', '5', '6']); + assert.equal(spawns, 1, 'no additional spawns'); + }); + + // D57 — ADR 0005 Amendment 8 §9: first client early-return, others continue + it('27d — first client iterator early-return mid-stream: others continue; source NOT aborted; cache written', async () => { + const warnings = []; + const store = new CacheStore({ _warnFn: (msg, meta) => warnings.push({ msg, meta }) }); + let sourceReturned = false; + const factory = () => (async function* () { + try { + for (let i = 0; i < 6; i++) { + yield `chunk-${i}`; + await new Promise(r => setImmediate(r)); + } + } finally { sourceReturned = true; } + })(); + const r1 = await store.getOrComputeStreaming('k', 'ck-27d', factory, { clientId: 'A' }); + const r2 = await store.getOrComputeStreaming('k', 'ck-27d', factory, { clientId: 'B' }); + // A early-returns after 2 chunks; B keeps consuming. + const itA = r1.stream; + const outA = []; + outA.push((await itA.next()).value); + outA.push((await itA.next()).value); + await itA.return(); // simulates HTTP client close + const outB = []; + for await (const c of r2.stream) outB.push(c); + // Source should have completed normally (not aborted) because B was still attached. + assert.equal(sourceReturned, true); + assert.deepEqual(outB, ['chunk-0', 'chunk-1', 'chunk-2', 'chunk-3', 'chunk-4', 'chunk-5']); + // Cache should be populated post-completion (B was a live client to the end). + const cached = await store.get('k', 'ck-27d'); + assert.ok(cached, 'cache populated despite A\'s mid-stream return'); + // No abort warning emitted. + assert.equal(warnings.filter(w => w.msg === 'streaming_inflight_abort').length, 0); + }); + + // D57 — ADR 0005 Amendment 8 §9: all clients disconnect → source aborted, no cache + it('27e — all clients disconnect mid-stream: source aborted via AbortController; no cache write; next call respawns', async () => { + const warnings = []; + const store = new CacheStore({ _warnFn: (msg, meta) => warnings.push({ msg, meta }) }); + let spawns = 0; + let sourceReturned = false; + const factory = () => { + spawns++; + return (async function* () { + try { + for (let i = 0; i < 20; i++) { + yield `c-${i}`; + await new Promise(r => setImmediate(r)); + } + } finally { sourceReturned = true; } + })(); + }; + const r1 = await store.getOrComputeStreaming('k', 'ck-27e', factory, { clientId: 'A' }); + const itA = r1.stream; + await itA.next(); + await itA.next(); + await itA.return(); + // Wait microtasks for the tee task to observe attachedClients.size === 0 and abort. + await new Promise(r => setTimeout(r, 30)); + assert.equal(sourceReturned, true); + assert.equal(warnings.filter(w => w.msg === 'streaming_inflight_abort').length, 1); + // No cache entry — subsequent call respawns (no inflight, no cache hit). + const cached = await store.get('k', 'ck-27e'); + assert.equal(cached, null, 'no cache write on abort'); + const r2 = await store.getOrComputeStreaming('k', 'ck-27e', factory, { clientId: 'B' }); + assert.equal(r2.role, 'source', 'subsequent call gets fresh source spawn'); + assert.equal(spawns, 2); + // Drain r2 so it doesn't dangle. + for await (const _c of r2.stream) { /* drain */ } + }); + + // D57 — ADR 0005 Amendment 8 §4: source throws mid-stream + it('27f — source throws mid-stream: all attached clients receive the error; no cache write; entry removed', async () => { + const warnings = []; + const store = new CacheStore({ _warnFn: (msg, meta) => warnings.push({ msg, meta }) }); + const err = new Error('synthetic source failure'); + const factory = () => (async function* () { + yield 'a'; + await new Promise(r => setImmediate(r)); + yield 'b'; + await new Promise(r => setImmediate(r)); + throw err; + })(); + const r1 = await store.getOrComputeStreaming('k', 'ck-27f', factory, { clientId: 'A' }); + const r2 = await store.getOrComputeStreaming('k', 'ck-27f', factory, { clientId: 'B' }); + // Both clients should reject when the source throws. + let e1, e2; + const c1 = (async () => { try { for await (const _c of r1.stream) {} } catch (e) { e1 = e; } })(); + const c2 = (async () => { try { for await (const _c of r2.stream) {} } catch (e) { e2 = e; } })(); + await Promise.all([c1, c2]); + assert.ok(e1, 'client A received error'); + assert.ok(e2, 'client B received error'); + assert.equal(e1.message, 'synthetic source failure'); + assert.equal(e2.message, 'synthetic source failure'); + // No cache write. + const cached = await store.get('k', 'ck-27f'); + assert.equal(cached, null); + // Inflight entry removed (next call would respawn). + assert.equal(store.stats('k').inflightCount, 0); + }); + + // D57 — ADR 0005 Amendment 8 §8: backpressure — slow client overflows queue + it('27g — backpressure: slow client queue overflow → STREAM_BACKPRESSURE terminator; fast client unaffected', async () => { + const warnings = []; + const store = new CacheStore({ _warnFn: (msg, meta) => warnings.push({ msg, meta }) }); + // Inject perClientQueueCap=1024 bytes; chunks ~256B each. + const bigText = 'x'.repeat(250); + const factory = () => makeChunkSequence( + [{ idx: 0, t: bigText }, { idx: 1, t: bigText }, { idx: 2, t: bigText }, { idx: 3, t: bigText }, { idx: 4, t: bigText }, { idx: 5, t: bigText }, { type: 'stop', finish_reason: 'stop' }] + ); + const r1 = await store.getOrComputeStreaming('k', 'ck-27g', factory, { clientId: 'fast', perClientQueueCap: 1024 }); + const r2 = await store.getOrComputeStreaming('k', 'ck-27g', factory, { clientId: 'slow', perClientQueueCap: 1024 }); + // Fast drains immediately; slow defers consumption. + const fastOut = []; + const fast = (async () => { for await (const c of r1.stream) fastOut.push(c); })(); + await fast; + // Now consume slow's stream; should hit backpressure terminator. + const slowOut = []; + for await (const c of r2.stream) slowOut.push(c); + // Fast client received the full sequence. + assert.equal(fastOut.length, 7); + assert.deepEqual(fastOut[6], { type: 'stop', finish_reason: 'stop' }); + // Slow client received some prefix + the STREAM_BACKPRESSURE terminator. + assert.deepEqual(slowOut[slowOut.length - 2], { type: 'stop', finish_reason: 'length' }); + assert.equal(slowOut[slowOut.length - 1], '[DONE]'); + assert.ok(slowOut.length < 7, 'slow client cut short before reaching natural end'); + // Backpressure warning emitted for slow client. + const bp = warnings.filter(w => w.msg === 'stream_backpressure_disconnect'); + assert.ok(bp.length >= 1, 'at least one stream_backpressure_disconnect emitted'); + assert.ok(bp.some(w => w.meta.client_id === 'slow'), 'slow client identified in warning'); + }); + + // D57 — ADR 0005 Amendment 8 §10: replay buffer cap — cache write skipped + it('27h — replay cap exceeded: cache write skipped; first caller still receives full stream; late joiner past cap gets STREAM_BACKPRESSURE', async () => { + const warnings = []; + const store = new CacheStore({ _warnFn: (msg, meta) => warnings.push({ msg, meta }) }); + const big = 'y'.repeat(400); + // Source emits 6 chunks at ~400B each → ~2400B total, exceeds replay cap 1024. + const factory = () => makeChunkSequence( + [{ t: big }, { t: big }, { t: big }, { t: big }, { t: big }, { t: big }] + ); + const r1 = await store.getOrComputeStreaming('k', 'ck-27h', factory, { + clientId: 'first', + accumulatedReplayCap: 1024, + perClientQueueCap: 1024 * 1024, // huge per-client cap so first caller never overflows + }); + // First caller iterates through. + const itA = r1.stream; + const outA = []; + // Pull 2 chunks then attempt late join while past cap. + outA.push((await itA.next()).value); + outA.push((await itA.next()).value); + outA.push((await itA.next()).value); // now well past 1024B accumulated + // Late joiner attaches past replay cap → gets STREAM_BACKPRESSURE. + const r2 = await store.getOrComputeStreaming('k', 'ck-27h', factory, { + clientId: 'late', + accumulatedReplayCap: 1024, + perClientQueueCap: 1024, + }); + assert.equal(r2.role, 'attached'); + const outLate = []; + for await (const c of r2.stream) outLate.push(c); + // Late joiner's stream is just the backpressure terminator (drain over cap). + assert.deepEqual(outLate, [{ type: 'stop', finish_reason: 'length' }, '[DONE]']); + // Drain first caller fully. + for await (const c of itA) outA.push(c); + assert.equal(outA.length, 6, 'first caller receives full source stream'); + // Cache write skipped. + const cached = await store.get('k', 'ck-27h'); + assert.equal(cached, null, 'cache NOT written when replay cap exceeded'); + // Replay-cap-exceeded warning emitted. + assert.ok( + warnings.some(w => w.msg === 'streaming_inflight_replay_cap_exceeded'), + 'replay-cap-exceeded warning fired' + ); + }); + + // D57 — ADR 0005 Amendment 8 §6: cache TTL race — late joiner attaches via inflight + it('27i — cache TTL race: cached entry expires during inflight; late joiner attaches via inflight Map', async () => { + const store = new CacheStore({ _warnFn: () => {} }); + // Pre-populate cache with TTL=10ms — will expire shortly. + await store.set('k', 'ck-27i', ['cached-a', 'cached-b'], 10); + // Wait so the cached entry expires. + await new Promise(r => setTimeout(r, 20)); + // Now a streaming request comes in: cache is expired → entry is recomputed. + let spawns = 0; + const factory = () => { + spawns++; + return makeChunkSequence(['live-1', 'live-2', 'live-3']); + }; + const r1 = await store.getOrComputeStreaming('k', 'ck-27i', factory, { clientId: 'A' }); + assert.equal(r1.role, 'source', 'expired cache → fresh source spawn'); + // While the source is running, a late joiner arrives — must attach via inflight Map. + const r2 = await store.getOrComputeStreaming('k', 'ck-27i', factory, { clientId: 'B' }); + assert.equal(r2.role, 'attached', 'late joiner attaches via inflight Map even after cache expiry'); + assert.equal(spawns, 1, 'no respawn'); + // Drain both. + const o1 = []; for await (const c of r1.stream) o1.push(c); + const o2 = []; for await (const c of r2.stream) o2.push(c); + assert.deepEqual(o1, ['live-1', 'live-2', 'live-3']); + assert.deepEqual(o2, ['live-1', 'live-2', 'live-3']); + // Inflight completion overwrites the expired slot. + const cached = await store.get('k', 'ck-27i'); + assert.ok(cached); + assert.deepEqual(cached.value, ['live-1', 'live-2', 'live-3']); + }); + + // D57 — ADR 0005 Amendment 8 §7: sourceFactory throws → first caller errors; no zombie state + it('27j — sourceFactory throws (e.g. CONCURRENCY_LIMIT): first caller errors; subsequent call retries; no zombie inflight', async () => { + const store = new CacheStore({ _warnFn: () => {} }); + let attempts = 0; + const factory = () => { + attempts++; + if (attempts === 1) { + throw new Error('CONCURRENCY_LIMIT'); + } + return makeChunkSequence(['a', 'b']); + }; + let firstErr; + try { + await store.getOrComputeStreaming('k', 'ck-27j', factory, { clientId: 'A' }); + } catch (e) { + firstErr = e; + } + assert.ok(firstErr, 'first call rejected with factory error'); + assert.equal(firstErr.message, 'CONCURRENCY_LIMIT'); + // No zombie inflight entry left dangling. + assert.equal(store.stats('k').inflightCount, 0, 'no stale streaming-inflight entry'); + // Subsequent call uses the factory again (it returns a real iterator now). + const r2 = await store.getOrComputeStreaming('k', 'ck-27j', factory, { clientId: 'B' }); + assert.equal(r2.role, 'source'); + const out = []; + for await (const c of r2.stream) out.push(c); + assert.deepEqual(out, ['a', 'b']); + assert.equal(attempts, 2); + }); + + // D57 — ADR 0005 Amendment 8 §1: stats accounting (hits / misses / inflightCount) + it('27k — stats: hits incremented for cache_hit + attached; misses for source; inflightCount reflects active streaming entries', async () => { + const store = new CacheStore({ _warnFn: () => {} }); + const factory = () => makeChunkSequence(['p', 'q', 'r']); + + // First caller — miss. + const r1 = await store.getOrComputeStreaming('k', 'ck-27k', factory, { clientId: 'A' }); + let stats = store.stats('k'); + assert.equal(stats.misses, 1, 'first caller increments misses'); + assert.equal(stats.hits, 0); + // Inflight entry alive during source phase. + assert.equal(stats.inflightCount, 1, 'streaming entry counted in inflightCount'); + + // Concurrent joiner — attached → hit. + const r2 = await store.getOrComputeStreaming('k', 'ck-27k', factory, { clientId: 'B' }); + stats = store.stats('k'); + assert.equal(stats.hits, 1, 'attached client increments hits'); + // Drain both. + await Promise.all([ + (async () => { for await (const _c of r1.stream) {} })(), + (async () => { for await (const _c of r2.stream) {} })(), + ]); + // After completion, entry removed from inflight. + stats = store.stats('k'); + assert.equal(stats.inflightCount, 0, 'streaming entry removed post-completion'); + + // Third call after completion — cache_hit (also a hit). + const r3 = await store.getOrComputeStreaming('k', 'ck-27k', factory, { clientId: 'C' }); + assert.equal(r3.role, 'cache_hit'); + for await (const _c of r3.stream) { /* drain */ } + stats = store.stats('k'); + assert.equal(stats.hits, 2, 'cache_hit also increments hits'); + }); + + // D57 — ADR 0005 Amendment 8 §2: composite key isolation (keyId\0cacheKey) + it('27l — composite key isolation: same cacheKey + different keyId → two independent inflight entries + two spawns', async () => { + const store = new CacheStore({ _warnFn: () => {} }); + let spawns = 0; + const factory = () => { + spawns++; + return makeChunkSequence(['n1', 'n2']); + }; + // Same cacheKey, different keyId. + const p1 = store.getOrComputeStreaming('key-A', 'shared-cache-key', factory, { clientId: 'A' }); + const p2 = store.getOrComputeStreaming('key-B', 'shared-cache-key', factory, { clientId: 'B' }); + const [r1, r2] = await Promise.all([p1, p2]); + // Both should be `source` — no cross-keyId sharing. + assert.equal(r1.role, 'source'); + assert.equal(r2.role, 'source'); + assert.equal(spawns, 2, 'two spawns because two distinct (keyId,cacheKey) composites'); + // Drain. + const out1 = []; for await (const c of r1.stream) out1.push(c); + const out2 = []; for await (const c of r2.stream) out2.push(c); + assert.deepEqual(out1, ['n1', 'n2']); + assert.deepEqual(out2, ['n1', 'n2']); + // Stats: each keyId has its own miss counter. + assert.equal(store.stats('key-A').misses, 1); + assert.equal(store.stats('key-B').misses, 1); + }); +});