mirror of
https://github.com/dtzp555-max/olp.git
synced 2026-07-22 13:35:10 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 
|
ba69a3c13b | ||
|
|
679e3b367d | ||
|
|
9b66326e72 | ||
|
|
1062e88e77 |
@@ -6,6 +6,21 @@ All notable changes to OLP land here. Per `CLAUDE.md` release_kit overlay, this
|
||||
|
||||
(empty — Phase 4 entries land here once Phase 4 opens)
|
||||
|
||||
## v0.3.2 — 2026-05-25
|
||||
|
||||
### Post-Phase-3 cleanup batch #2 — streaming-path singleflight + TOCTOU close (D57 + D58 + D59)
|
||||
|
||||
Patch release closing v1.x roadmap #1 end-to-end. The cache layer's D4 singleflight (one spawn per identical concurrent request) was fully wired on the buffered path since v0.1 but NOT on the streaming path — N concurrent identical streaming requests each spawned their own CLI process. v0.3.2 ships the streaming sibling: tee fan-out, late-joiner replay, per-client backpressure, AbortController propagation, and TOCTOU close. 3 D-day commits (D57 + D58 + D59); ADR 0005 Amendment 8 §§1–14 implemented.
|
||||
|
||||
- **D57** (PR #36) — **cache layer.** New `cacheStore.getOrComputeStreaming(keyId, cacheKey, sourceFactory, opts) → { stream, isFirst, role }` mirroring `getOrCompute` on the streaming side. Internals: `_streamingInflight: Map<compositeKey, StreamingInflightEntry>` (composite key `keyId + '\0' + cacheKey`) with synchronous check+insert atomicity (closes TOCTOU per ADR 0005 Amendment 8 §1, §6); single-reader tee fan-out across all attached clients; late-joiner replay buffer (synchronous drain on attach; `STREAM_BACKPRESSURE` terminator if drain or replay-truncation would corrupt); per-client backpressure (`PER_CLIENT_QUEUE_CAP = 1 MB`, overridable via opts); accumulated replay cap (`ACCUMULATED_REPLAY_CAP = 10 MB`, mirrors D23 cache-entry cap); AbortController fires source-iterator return when all clients disconnect. New `'STREAM_BACKPRESSURE'` entry in `PROVIDER_ERROR_CODES` — NOT a hard trigger (whitelist-only `HARD_TRIGGER_CODES`). Suite 27 = 12 unit tests.
|
||||
- **D58** (PR #37) — **server wiring.** Streaming branch in `server.mjs` swapped from the peek+spawn pattern to `cacheStore.getOrComputeStreaming(...)`. `tryAcquireSpawn`/`releaseSpawn` moved INSIDE the `sourceFactory` closure per ADR 0005 Amendment 8 §7 (only the first caller acquires; attached joiners share the slot; release fires exactly once on source completion/error/abort). `CONCURRENCY_LIMIT` thrown by the factory triggers fallthrough to the buffered path (preserves today's behaviour). New `X-OLP-Streaming-Inflight: source | attached` HTTP header annotates per-response role (§11). New `cache_status: 'streaming_attached'` audit value tracks the singleflight win. `lib/audit-query.mjs` aggregate APIs (`aggregateRequests`, `cacheHitRateWindow`) extended with `cache_streaming_attached` / `streaming_attached` fields so the cache_status breakdown reconciles. `res.on('close')` propagates client disconnect into the tee's `attachedClients` accounting (§9). D16 truncated-not-cached invariant preserved via server-layer `cacheStore.delete` on stop-less exhaustion (the cache layer is IR-agnostic and writes accumulatedChunks on any source exhaustion; the IR-aware server deletes the entry when the source returned without a `{type:'stop'}` chunk). Suite 28 = 8 HTTP integration tests.
|
||||
- **D59** (PR #38) — **docs polish.** README § Known limitations bullet inverted to ✅ shipped marker. `docs/v1x-roadmap.md` #1 rewritten to closed state with 3-D-day breakdown. #6 (streaming SPAWN_FAILED salvage) unbundled from #1 because the tee architecture as implemented does not carry salvage semantics. Issue #16 closed with refs to PRs #36 / #37 / #38.
|
||||
- **Test count:** 603 (v0.3.1) → 623 (v0.3.2). +20 streaming-SF tests (Suite 27 = 12 unit, Suite 28 = 8 HTTP integration).
|
||||
- **Deferred sub-items (not blocking #1 closure):** (a) `X-OLP-Streaming-Inflight: solo` wire value not emitted — observable only post-stream via `streaming_inflight_source_done` log event's `attached_count: 0`. Future ADR amendment may expose via HTTP trailer. (b) `streaming_inflight_join` log event not emitted from the cache-layer `_attachClient` path because provider/model context lives in the sourceFactory closure (server-layer concern). (c) `isFirst` field returned by `getOrComputeStreaming` is unused by server.mjs (`role` supersedes); could be removed in a future cache-layer API cleanup.
|
||||
- **Authority:** ADR 0005 Amendment 8 (design ratified at D42 2026-05-25; implementation gated on maintainer "go" — fired 2026-05-25 post-v0.3.1). `docs/v1x-roadmap.md` #1 (closed). GitHub issue #16 (closed). ADR 0002 Amendment 6 (D38 `tryAcquireSpawn`/`releaseSpawn` semantics, now invoked from sourceFactory closure).
|
||||
|
||||
**Patch-release classification.** Per `release_kit.phase_rolling_mode` cross-Phase discipline + maintainer release-cut decision (this session, 2026-05-25): the new wire surface (`X-OLP-Streaming-Inflight` header + `streaming_attached` cache_status) is semver-wise a minor bump, but this is roadmap-cleanup work — NOT Phase 4 product scope. The reserved `0.4.0` identifier stays for the formal Phase 4 close. v0.3.2 ships as a patch under the Phase 4 pre-release banner. Tag push triggers `release.yml`.
|
||||
|
||||
## v0.3.1 — 2026-05-25
|
||||
|
||||
### Post-Phase-3 cleanup batch #1 (D56)
|
||||
|
||||
@@ -197,7 +197,7 @@ Phase 1 closed at v0.1.1 (multi-provider proxy core + pre-Phase-2 cleanup). Phas
|
||||
|
||||
Behaviors that work correctly at personal/family scale but have ratified follow-ups for a v1.x sprint. Single landing page: [`docs/v1x-roadmap.md`](./docs/v1x-roadmap.md).
|
||||
|
||||
- **Streaming-path singleflight not implemented.** The cache layer's D4 singleflight (one spawn per identical concurrent request) is fully wired on the buffered path but NOT on the streaming path. N concurrent identical streaming requests at v0.1 will each spawn their own CLI process. Design ratified in [ADR 0005 Amendment 8](./docs/adr/0005-cache-cross-provider.md); implementation tracked via [issue #16](https://github.com/dtzp555-max/olp/issues/16) and [v1.x roadmap #1](./docs/v1x-roadmap.md). At family scale this is observably fine — every caller still receives the correct response; the cost is N CLI processes instead of one.
|
||||
- **Streaming-path singleflight ✅ shipped (D57 + D58, 2026-05-25).** `cacheStore.getOrComputeStreaming(...)` mirrors the buffered-path `getOrCompute` and resolves the TOCTOU window between peek and spawn ([issue #16](https://github.com/dtzp555-max/olp/issues/16)). Two concurrent identical streaming requests share one CLI spawn via tee fan-out; late joiners receive accumulated replay + the live tail; per-client backpressure (`PER_CLIENT_QUEUE_CAP=1MB`) protects against slow consumers; full-disconnect aborts the source CLI via AbortController. New `X-OLP-Streaming-Inflight: source | attached` header annotates the role. New `cache_status: 'streaming_attached'` audit value tracks the singleflight win. Authority: [ADR 0005 Amendment 8](./docs/adr/0005-cache-cross-provider.md), v1.x roadmap #1.
|
||||
- **Soft triggers configured but inert.** `routing.soft_triggers` in `~/.olp/config.json` is honored by the engine's evaluation logic but `quotaStatus()` polling is not wired (ADR 0004 Amendment 2). A startup warning fires if the field is non-empty so the inert state is visible.
|
||||
- **Multi-key auth + owner gating + keygen CLI shipped at v0.2.0 (D44 + D45 + D46 + D47).** `lib/keys.mjs` (core), `lib/audit.mjs` (audit), owner-vs-guest `/health` payload trimming + `X-OLP-Fallback-Detail` policy gating, `bin/olp-keys.mjs` (keygen CLI). All 11 ADR 0007 § 10 acceptance criteria covered. v0.2.0 maintainer-merged 2026-05-25.
|
||||
|
||||
|
||||
+16
-13
@@ -8,21 +8,23 @@
|
||||
3. **Where** does the work live in the tree today (file + anchor).
|
||||
4. **When** does it need to land (trigger: load profile, security event, governance amendment).
|
||||
|
||||
**Reading order for a v1.x sprint kickoff.** Items #1–#3 are the most architecturally consequential and should be designed in dependency order: #2 (multi-key auth) blocks header gating in #1 and observability ownership in #4. #1 (streaming SF) blocks #5 (soft trigger reactivation) only if soft triggers are wired on streaming requests.
|
||||
**Reading order for a v1.x sprint kickoff.** As of 2026-05-25, #1 (streaming SF, D57+D58) and #2 (multi-key auth, Phase 2) are CLOSED, and #4 and #7 closed in D56. Remaining v1.x scope: #3 (soft trigger reactivation), #5 (provider cacheKeyFields mask), #6 (streaming SPAWN_FAILED salvage — unbundled from #1 at #1 close). All three remaining items have explicit "trigger to start" gates that have not fired.
|
||||
|
||||
---
|
||||
|
||||
## #1 — Streaming-path singleflight + TOCTOU close
|
||||
## #1 — Streaming-path singleflight + TOCTOU close — ✅ **SHIPPED (D57 + D58, 2026-05-25)**
|
||||
|
||||
- **What.** `cacheStore.getOrComputeStreaming(keyId, cacheKey, sourceFactory)` API replacing the current `peek + spawn` pattern in `server.mjs`. Per-(keyId, cacheKey) inflight Map with tee fan-out, bounded per-client backpressure queues, late-joiner replay buffer, AbortController propagation on all-disconnect.
|
||||
- **Why deferred.** Personal/family-scale single-tenant load — N concurrent identical streaming requests is an edge case that has not been reported. Each concurrent caller receives the correct response; the waste is N CLI processes instead of one.
|
||||
- **Design ADR (ratified).** [`docs/adr/0005-cache-cross-provider.md` Amendment 8](./adr/0005-cache-cross-provider.md) — full design including the inflight Map shape, tee policy, late-joiner replay, backpressure cap, D38 semaphore coordination, abort policy, cache TTL race handling, observability event set, and X-OLP-Streaming-Inflight header. Implementation acceptance criteria are in Amendment 8 §13.
|
||||
- **Tracking issue.** GitHub issue [#16](https://github.com/dtzp555-max/olp/issues/16) — STAYS OPEN as v1.x tracker. Sibling: the closed-but-not-implemented Amendment 6 deferral (D34 F1).
|
||||
- **Code anchors today.**
|
||||
- `server.mjs` lines ~782 (`preCheckHit = await cacheStore.peek(...)`) and ~811–817 (streaming branch entry) — these are the lines the new API replaces.
|
||||
- `lib/cache/store.mjs` `getOrCompute` — sibling API; the new one mirrors its shape on the streaming path.
|
||||
- **Trigger to start.** Any of: (a) report of N>1 concurrent identical streaming requests in the wild, (b) v1.x sprint planning kickoff with the maintainer explicitly opening this scope, (c) downstream feature requiring tee-streaming primitive (e.g., browser-side observer attaching to an existing stream).
|
||||
- **Estimated effort.** Design ADR ratified (Amendment 8) = 30 min done. Implementation = 200–400 lines + 15-20 tests + fresh-context reviewer pass. ~3-4 hours of subagent runtime with full Iron Rule 10 discipline.
|
||||
- **Status.** Closed. Trigger (b) fired 2026-05-25 — maintainer "go" after v0.3.1. Shipped across three D-days:
|
||||
- **D57** (PR #36) — cache layer: `cacheStore.getOrComputeStreaming(keyId, cacheKey, sourceFactory, opts) → { stream, isFirst, role }` with `_streamingInflight` Map, tee fan-out, late-joiner replay buffer, per-client backpressure (`PER_CLIENT_QUEUE_CAP=1MB`), replay cap (`ACCUMULATED_REPLAY_CAP=10MB`), AbortController propagation, synchronous Map check+insert (closes TOCTOU). Suite 27 = 12 unit tests.
|
||||
- **D58** (PR #37) — server.mjs wiring: streaming branch swap; `tryAcquireSpawn`/`releaseSpawn` moved inside `sourceFactory` closure (D38 §7 coordination); `CONCURRENCY_LIMIT` fallthrough preserved; `X-OLP-Streaming-Inflight: source | attached` header; `cache_status: 'streaming_attached'` audit value + audit-query gauge reconciliation; `res.on('close') → stream.return()` for client disconnect; D16 truncated-not-cached invariant preserved via `cacheStore.delete` on stop-less exhaustion. Suite 28 = 8 HTTP integration tests.
|
||||
- **D59** (this commit) — docs polish: README known-limitations entry inverted; this roadmap entry closed; issue #16 closed.
|
||||
- **Design authority.** [`docs/adr/0005-cache-cross-provider.md` Amendment 8](./adr/0005-cache-cross-provider.md) — implemented per spec §§1–14 across D57+D58.
|
||||
- **Tracking issue.** GitHub issue [#16](https://github.com/dtzp555-max/olp/issues/16) — CLOSED at D59 with refs to D57+D58 PRs.
|
||||
- **Final test count delta.** 603 (v0.3.1) → 623 (v0.3.2/v0.4.0). +20 tests across the SF arc.
|
||||
- **Deferred sub-items** (left here as future-work pointers, NOT blocking #1 closure):
|
||||
- `X-OLP-Streaming-Inflight: solo` value not emitted on the wire (Amendment 8 §11). It's observable only post-stream via the `streaming_inflight_source_done` log event's `attached_count: 0`. Future ADR amendment may expose via HTTP trailer.
|
||||
- `streaming_inflight_join` log event from `_attachClient` cache-layer path (carries no provider/model context). D58 emits the event from the server-layer wrapper instead; cache-layer emission would need a provider/model plumb (TODO marker at `lib/cache/store.mjs:~620`).
|
||||
- `isFirst` field returned by `getOrComputeStreaming` is currently unused by server.mjs (`role` supersedes). Could be removed in a future cache-layer API cleanup.
|
||||
|
||||
## #2 — Multi-key auth (`lib/keys.mjs`) — **PHASE 2 ACTIVE (no longer deferred)**
|
||||
|
||||
@@ -81,9 +83,10 @@
|
||||
|
||||
- **What.** Currently the streaming branch does NOT participate in D16 salvage (the salvage-on-SPAWN_FAILED + chunks pattern that the buffered path uses). Streaming SPAWN_FAILED mid-stream → the truncation marker (D35 #10) fires, but no salvage logic captures partial chunks for downstream cache reuse.
|
||||
- **Why deferred.** Less impactful than #1 — at most one client benefits per spawn event, and the buffered path already provides salvage for the bulk of requests. Streaming is the minority path.
|
||||
- **Design ADR.** Not yet ratified. Coordinated with #1 because the tee architecture changes the salvage semantics (multiple clients may want different finish_reason interpretations on source-mid-stream-failure).
|
||||
- **Status update post-#1 close (2026-05-25).** #1 was originally bundled with #6 in the design ADR (Amendment 8). The tee architecture as implemented does NOT carry salvage semantics — D57's tee writes `accumulatedChunks` to cache only on normal source completion (stop chunk seen); on SPAWN_FAILED mid-stream the cache layer rejects all clients with the error and does NOT persist partial chunks. D58 preserves D16's truncated-not-cached invariant via server-layer `cacheStore.delete` on stop-less exhaustion. #6 therefore remains independently deferrable.
|
||||
- **Design ADR.** Not yet ratified. The unbundling from #1 means #6 now needs its own ADR amendment when triggered.
|
||||
- **Tracking.** Not a GitHub issue. Tracked here.
|
||||
- **Trigger to start.** Bundled with #1 implementation work (the inflight tee architecture changes the salvage semantics, so designing them together is cheaper than serializing).
|
||||
- **Trigger to start.** First report of streaming-path SPAWN_FAILED mid-stream where partial-chunk salvage would have helped a downstream caller. Practically unlikely at family scale.
|
||||
|
||||
## #7 — AUTH_MISSING tuple path test coverage (D40 follow-up)
|
||||
|
||||
|
||||
+20
-6
@@ -227,7 +227,7 @@ export function* readAuditWindow({ startMs, endMs, olpHome, logEvent } = {}) {
|
||||
* {
|
||||
* window: { startMs, endMs },
|
||||
* request_count, status_2xx, status_4xx, status_5xx,
|
||||
* by_provider: { [providerKey]: { count, cache_hit, cache_miss, cache_bypass, fallback_count } },
|
||||
* by_provider: { [providerKey]: { count, cache_hit, cache_miss, cache_bypass, cache_streaming_attached, fallback_count } },
|
||||
* by_owner_tier: { owner: N, guest: N, anonymous: N },
|
||||
* by_path: { '/v1/chat/completions': N, '/v1/models': N, ... },
|
||||
* median_latency_ms, p95_latency_ms,
|
||||
@@ -276,12 +276,17 @@ export function aggregateRequests({ windowMs, olpHome, logEvent, _nowFn } = {})
|
||||
// By provider
|
||||
if (typeof ev.provider === 'string' && ev.provider.length > 0) {
|
||||
const p = result.by_provider[ev.provider] ??= {
|
||||
count: 0, cache_hit: 0, cache_miss: 0, cache_bypass: 0, fallback_count: 0,
|
||||
count: 0, cache_hit: 0, cache_miss: 0, cache_bypass: 0, cache_streaming_attached: 0, fallback_count: 0,
|
||||
};
|
||||
p.count++;
|
||||
if (ev.cache_status === 'hit') p.cache_hit++;
|
||||
else if (ev.cache_status === 'miss') p.cache_miss++;
|
||||
else if (ev.cache_status === 'bypass') p.cache_bypass++;
|
||||
// D58 — ADR 0005 Amendment 8 §11 + lib/audit.mjs cache_status enum: streaming
|
||||
// singleflight joiners (attached) share the source spawn but did not hit a
|
||||
// cache. Tracked separately so `count` and `cache_hit + cache_miss +
|
||||
// cache_bypass + cache_streaming_attached` reconcile.
|
||||
else if (ev.cache_status === 'streaming_attached') p.cache_streaming_attached++;
|
||||
if (typeof ev.fallback_hops === 'number' && ev.fallback_hops > 0) p.fallback_count++;
|
||||
}
|
||||
|
||||
@@ -443,7 +448,12 @@ export function spendTrendDaily({ days, olpHome, logEvent, _nowFn } = {}) {
|
||||
* `cacheStore.stats()` in server.mjs: that is the live in-process counter;
|
||||
* this is the audit-side rate scoped to the rolling window.
|
||||
*
|
||||
* { window: { startMs, endMs }, total, hit, miss, bypass, hit_rate, by_provider }
|
||||
* { window: { startMs, endMs }, total, hit, miss, bypass, streaming_attached, hit_rate, by_provider }
|
||||
*
|
||||
* `streaming_attached` (D58, ADR 0005 Amendment 8 §11): D58 streaming
|
||||
* singleflight joiners did not hit a literal cache, so they are excluded
|
||||
* from both numerator AND denominator of `hit_rate`. Tracked separately
|
||||
* so the count reconciles with `total = hit + miss + bypass + streaming_attached`.
|
||||
*
|
||||
* @param {object} args
|
||||
* @param {number} args.windowMs
|
||||
@@ -459,18 +469,22 @@ export function cacheHitRateWindow({ windowMs, olpHome, logEvent, _nowFn } = {})
|
||||
const startMs = now - windowMs;
|
||||
const endMs = now;
|
||||
|
||||
let total = 0, hit = 0, miss = 0, bypass = 0;
|
||||
let total = 0, hit = 0, miss = 0, bypass = 0, streaming_attached = 0;
|
||||
const by_provider = {};
|
||||
|
||||
for (const ev of readAuditWindow({ startMs, endMs, olpHome, logEvent })) {
|
||||
if (ev.cache_status === null || ev.cache_status === undefined) continue;
|
||||
total++;
|
||||
const p = typeof ev.provider === 'string' && ev.provider.length > 0 ? ev.provider : '__unknown__';
|
||||
const pe = by_provider[p] ??= { total: 0, hit: 0, miss: 0, bypass: 0, hit_rate: 0 };
|
||||
const pe = by_provider[p] ??= { total: 0, hit: 0, miss: 0, bypass: 0, streaming_attached: 0, hit_rate: 0 };
|
||||
pe.total++;
|
||||
if (ev.cache_status === 'hit') { hit++; pe.hit++; }
|
||||
else if (ev.cache_status === 'miss') { miss++; pe.miss++; }
|
||||
else if (ev.cache_status === 'bypass') { bypass++; pe.bypass++; }
|
||||
// D58 — ADR 0005 Amendment 8 §11: streaming singleflight joiners.
|
||||
// Excluded from hit_rate numerator + denominator (they did not hit a
|
||||
// literal cache); tracked so `total` reconciles.
|
||||
else if (ev.cache_status === 'streaming_attached') { streaming_attached++; pe.streaming_attached++; }
|
||||
}
|
||||
|
||||
// Compute hit_rate per provider + overall (excludes bypass from denominator
|
||||
@@ -484,6 +498,6 @@ export function cacheHitRateWindow({ windowMs, olpHome, logEvent, _nowFn } = {})
|
||||
|
||||
return {
|
||||
window: { startMs, endMs },
|
||||
total, hit, miss, bypass, hit_rate, by_provider,
|
||||
total, hit, miss, bypass, streaming_attached, hit_rate, by_provider,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -174,6 +174,20 @@ export function _maybeRotateAudit(args = {}) {
|
||||
* fallback_hops, tried_providers, error_code, ir_request_hash, chain_id).
|
||||
* Caller is responsible for populating fields; missing fields are
|
||||
* serialized as undefined → omitted by JSON.stringify.
|
||||
*
|
||||
* cache_status enum (free-form string; not schema-validated at append):
|
||||
* 'hit' — served from cache (buffered-replay or streaming
|
||||
* cache_hit role from ADR 0005 Amendment 8 §1).
|
||||
* 'miss' — buffered or streaming source path; provider
|
||||
* spawn fired for this request.
|
||||
* 'bypass' — D2 cache_control bypass (no cache read/write).
|
||||
* 'streaming_attached' — D58 / ADR 0005 Amendment 8 §11: client joined
|
||||
* an in-flight streaming source spawn from
|
||||
* another caller; this request did NOT spawn a
|
||||
* provider but also did NOT hit the cache (the
|
||||
* cache was empty at the inflight Map check).
|
||||
* null — pre-chain error paths (the cache layer was
|
||||
* never consulted; e.g. 401, 415, 400 IR).
|
||||
* @param {object} [opts]
|
||||
* @param {string} [opts.olpHome] - test override; defaults to ~/.olp
|
||||
* @param {(level: string, event: string, data?: object) => void} [opts.logEvent]
|
||||
|
||||
Vendored
+582
-9
@@ -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<AttachedClient>} 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<string, Promise<*>>} */
|
||||
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<string, StreamingInflightEntry>} */
|
||||
this._streamingInflight = new Map();
|
||||
|
||||
// Stats per keyId
|
||||
/** @type {Map<string, { hits: number, misses: number }>} */
|
||||
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: <async iterator over cached chunks>, 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<*>>|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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "olp",
|
||||
"version": "0.3.1",
|
||||
"version": "0.3.2",
|
||||
"description": "Personal multi-provider LLM proxy. Successor to OCP. One HTTP endpoint, multiple subscriptions behind it, automatic routing + fallback + caching.",
|
||||
"type": "module",
|
||||
"main": "server.mjs",
|
||||
|
||||
+245
-168
@@ -1117,32 +1117,21 @@ async function handleChatCompletions(req, res) {
|
||||
// truncate the response (end with no [DONE]). On error before any chunk:
|
||||
// throw so the outer handler can surface a clean error (no bytes written).
|
||||
//
|
||||
// On success: write chunks to res AND cache so subsequent identical requests
|
||||
// hit the burst-replay path.
|
||||
// D38 (issue #1): pre-acquire a concurrency slot for the streaming path so
|
||||
// saturation here behaves identically to saturation on the buffered path.
|
||||
// If acquire fails, fall through (skip this branch) — the buffered path
|
||||
// below also gates via tryAcquireSpawn and will surface a chain-exhausted
|
||||
// error through executeWithFallback (correct behaviour: a single-hop chain
|
||||
// at maxConcurrent has no other hop to advance to).
|
||||
// On success: chunks are written to res, and the cache layer (via
|
||||
// getOrComputeStreaming) writes accumulated chunks on source completion.
|
||||
//
|
||||
// Acquired here, released in the `finally` below. The gate intentionally
|
||||
// lives BEFORE the streaming-branch entry check so the buffered fallthrough
|
||||
// can re-attempt acquire from a clean slate.
|
||||
// TODO(v1.x — ADR 0005 Amendment 8 / issue #16): replace this peek+spawn
|
||||
// pattern with cacheStore.getOrComputeStreaming(...) to close the TOCTOU
|
||||
// window between line ~782 peek and the spawn at line ~846, and to make N
|
||||
// concurrent identical streaming requests share one spawn (tee-streaming).
|
||||
// Design ratified in D42; see docs/v1x-roadmap.md #1 for the trigger and
|
||||
// acceptance criteria. DO NOT remove this comment until the v1.x impl lands.
|
||||
let streamingAcquired = false;
|
||||
// D58 — ADR 0005 Amendment 8 §§7, 11, 12: streaming singleflight via
|
||||
// cacheStore.getOrComputeStreaming(...). The peek+spawn TOCTOU window
|
||||
// (server.mjs:1104 peek vs. spawn) is collapsed by the synchronous Map
|
||||
// check+insert in the cache layer (D57). The D38 tryAcquireSpawn gate now
|
||||
// lives INSIDE the sourceFactory closure: only the first caller acquires a
|
||||
// slot; attached joiners share it. On factory-throw CONCURRENCY_LIMIT, the
|
||||
// cache layer rejects → we fall through to the buffered path (same outcome
|
||||
// as today). preCheckHit gates entry exactly as before — the cache_hit
|
||||
// outcome from getOrComputeStreaming is the TTL-race safety net; in
|
||||
// practice it should not fire because preCheckHit excludes alive entries.
|
||||
// See docs/adr/0005-cache-cross-provider.md Amendment 8 + issue #16.
|
||||
if (ir.stream && chain.length === 1 && !bypassCacheForFirstHop && !preCheckHit && cacheableForFirstHop) {
|
||||
const candidatePlugin = loadedProviders.get(chain[0].provider);
|
||||
const candidateMax = candidatePlugin?.hints?.maxConcurrent ?? DEFAULT_MAX_CONCURRENT_SPAWNS;
|
||||
streamingAcquired = candidatePlugin ? tryAcquireSpawn(chain[0].provider, candidateMax) : false;
|
||||
}
|
||||
|
||||
if (ir.stream && chain.length === 1 && !bypassCacheForFirstHop && !preCheckHit && cacheableForFirstHop && streamingAcquired) {
|
||||
const streamProvider = chain[0].provider;
|
||||
const streamModel = chain[0].model;
|
||||
const streamCacheKey = computeCacheKey(streamProvider, streamModel, ir);
|
||||
@@ -1150,63 +1139,214 @@ async function handleChatCompletions(req, res) {
|
||||
|
||||
if (!streamPlugin) {
|
||||
// Provider disappeared between chain build and here (edge case).
|
||||
// Release the slot we acquired above so the counter stays balanced.
|
||||
auditCtx.provider = streamProvider;
|
||||
auditCtx.error_code = 'no_enabled_provider';
|
||||
releaseSpawn(streamProvider);
|
||||
return sendError(res, 503, `Provider ${streamProvider} is not enabled`, 'no_enabled_provider',
|
||||
olpErrorHeaders({ startMs, model: ir.model }));
|
||||
}
|
||||
|
||||
// D45 fold-in P1: populate audit ctx for the real-streaming path. Each
|
||||
// exit below (success, error-after-first-chunk, pre-first-chunk-error,
|
||||
// 503 above) leaves these fields representing the streaming attempt.
|
||||
// Error paths amend `error_code`; success leaves it null.
|
||||
auditCtx.provider = streamProvider;
|
||||
auditCtx.tried_providers = [streamProvider];
|
||||
auditCtx.cache_status = 'miss';
|
||||
// D58 — ADR 0005 Amendment 8 §7: sourceFactory wraps acquire + spawn.
|
||||
// Acquire fires ONLY for the first caller; attached clients share the
|
||||
// slot. Release fires EXACTLY ONCE when the source iterator's
|
||||
// try/finally executes (normal exhaustion, error, or iterator.return()
|
||||
// propagated from sourceAbortController via the cache layer).
|
||||
const candidateMax = streamPlugin.hints?.maxConcurrent ?? DEFAULT_MAX_CONCURRENT_SPAWNS;
|
||||
const sourceFactory = () => {
|
||||
if (!tryAcquireSpawn(streamProvider, candidateMax)) {
|
||||
const err = new ProviderError(
|
||||
`provider ${streamProvider} at maxConcurrent (${candidateMax})`,
|
||||
'CONCURRENCY_LIMIT',
|
||||
);
|
||||
err.providerName = streamProvider;
|
||||
err.maxConcurrent = candidateMax;
|
||||
err.activeSpawns = getActiveSpawnCount(streamProvider);
|
||||
throw err;
|
||||
}
|
||||
// Wrap streamPlugin.spawn in an async generator that guarantees
|
||||
// releaseSpawn fires exactly once in finally — regardless of normal
|
||||
// exhaustion, mid-stream throw, or iterator.return() from cache-layer
|
||||
// sourceAbortController propagation (§9).
|
||||
return (async function* sourceWithRelease() {
|
||||
try {
|
||||
for await (const irChunk of streamPlugin.spawn(ir, authContext)) {
|
||||
yield irChunk;
|
||||
}
|
||||
} finally {
|
||||
releaseSpawn(streamProvider);
|
||||
}
|
||||
})();
|
||||
};
|
||||
|
||||
const streamHeaders = olpHeaders({
|
||||
providerUsed: streamProvider,
|
||||
modelUsed: streamModel,
|
||||
startMs,
|
||||
cacheStatus: 'miss',
|
||||
fallbackHops: 0,
|
||||
});
|
||||
|
||||
// D14: writeHead is deferred until just before the first res.write so that
|
||||
// pre-first-chunk errors can still produce a JSON 502 (matching the buffered
|
||||
// path). Calling writeHead unconditionally here was the D14 defect.
|
||||
|
||||
const streamedChunks = [];
|
||||
let firstChunkEmitted = false;
|
||||
// D58 — invoke the cache-layer coordinator. CONCURRENCY_LIMIT from the
|
||||
// factory falls through to the buffered path (the buffered fallback
|
||||
// engine re-attempts acquire). Any other pre-stream error surfaces 502.
|
||||
let stream;
|
||||
let role;
|
||||
try {
|
||||
for await (const irChunk of streamPlugin.spawn(ir, authContext)) {
|
||||
if (irChunk.type === 'error') {
|
||||
// Error chunk from provider
|
||||
if (firstChunkEmitted) {
|
||||
// Past first-chunk boundary — can't fallback; emit truncation marker + [DONE]
|
||||
// so clients can detect the incomplete response in-band (aligns D26 F19
|
||||
// stop-less exhaustion behaviour and the catch-block fix in D35 #10).
|
||||
logEvent('warn', 'streaming_error_after_first_chunk', {
|
||||
provider: streamProvider,
|
||||
model: streamModel,
|
||||
error: irChunk.error,
|
||||
const result = await cacheStore.getOrComputeStreaming(
|
||||
keyId,
|
||||
streamCacheKey,
|
||||
sourceFactory,
|
||||
{ clientId: requestId },
|
||||
);
|
||||
stream = result.stream;
|
||||
role = result.role;
|
||||
} catch (e) {
|
||||
if (e instanceof ProviderError && e.code === 'CONCURRENCY_LIMIT') {
|
||||
// Fall through to the buffered path — the existing executeHopFn /
|
||||
// executeWithFallback machinery re-attempts acquire and surfaces a
|
||||
// chain-exhausted 502 for a single-hop chain at maxConcurrent, which
|
||||
// matches today's pre-D58 behaviour for this saturation scenario.
|
||||
logEvent('debug', 'streaming_concurrency_limit_fallthrough', {
|
||||
provider: streamProvider,
|
||||
model: streamModel,
|
||||
});
|
||||
// (proceed past this block)
|
||||
} else {
|
||||
// Any other pre-first-chunk error — surface a clean 502.
|
||||
auditCtx.provider = streamProvider;
|
||||
auditCtx.tried_providers = [streamProvider];
|
||||
auditCtx.cache_status = 'miss';
|
||||
auditCtx.error_code = e?.code ?? 'provider_error';
|
||||
logEvent('error', 'streaming_error_before_first_chunk', {
|
||||
provider: streamProvider,
|
||||
model: streamModel,
|
||||
error: e.message,
|
||||
});
|
||||
return sendError(res, 502, e.message ?? 'Provider error', 'provider_error',
|
||||
olpHeaders({ providerUsed: streamProvider, modelUsed: streamModel, startMs, cacheStatus: 'miss', fallbackHops: 0 }));
|
||||
}
|
||||
}
|
||||
|
||||
if (stream) {
|
||||
// D58 — ADR 0005 Amendment 8 §11: cache_status reflects role-of-this-
|
||||
// client at attach-time. `source` did real work (miss); `attached`
|
||||
// shared a live spawn (streaming_attached); `cache_hit` served from
|
||||
// cache (hit). See lib/audit.mjs cache_status JSDoc for the enum.
|
||||
auditCtx.provider = streamProvider;
|
||||
auditCtx.tried_providers = [streamProvider];
|
||||
if (role === 'cache_hit') {
|
||||
auditCtx.cache_status = 'hit';
|
||||
} else if (role === 'attached') {
|
||||
auditCtx.cache_status = 'streaming_attached';
|
||||
} else {
|
||||
auditCtx.cache_status = 'miss';
|
||||
}
|
||||
|
||||
// D58 — ADR 0005 Amendment 8 §9: wire HTTP req close → iterator
|
||||
// return() so a client disconnect propagates into the cache layer.
|
||||
// Without this the cache layer never learns of disconnect and the
|
||||
// source spawn may orphan (last-client-gone abort would not fire).
|
||||
// try/catch guards against concurrent return() throws (Node async
|
||||
// generators can throw if return() is invoked while iteration is
|
||||
// pending). The cache layer's iterator return() is idempotent.
|
||||
// D58 — ADR 0005 Amendment 8 §9: detect client disconnect via res
|
||||
// 'close' event (not req 'close'). For SSE responses Node emits
|
||||
// res.on('close') when the underlying socket goes away while the
|
||||
// response is still streaming. req.on('close') in Node 18+ fires only
|
||||
// after the response is fully sent, which is too late for our abort
|
||||
// propagation needs. Without this wire-up the cache layer would never
|
||||
// learn of disconnect → potential orphan spawns.
|
||||
const onClose = () => {
|
||||
try { stream.return?.(); } catch { /* best-effort */ }
|
||||
};
|
||||
res.on('close', onClose);
|
||||
|
||||
// D58 — ADR 0005 Amendment 8 §11: X-OLP-Streaming-Inflight header.
|
||||
// role === 'source' → 'source' (first caller; may upgrade to
|
||||
// 'solo' post-stream if no joiners ever
|
||||
// attached — see deferral note below).
|
||||
// role === 'attached' → 'attached'
|
||||
// role === 'cache_hit'→ omitted; the existing X-OLP-Cache: hit
|
||||
// already signals that path.
|
||||
// 'solo' (no joiners ever attached) is observable only post-stream —
|
||||
// emitted via streaming_inflight_source_done log event with
|
||||
// attached_count=0; the wire header reports `source` initially and
|
||||
// does NOT flip mid-stream. Future ADR may expose it on a trailer.
|
||||
const cacheStatusForHeader = (role === 'cache_hit') ? 'hit' : 'miss';
|
||||
const streamHeaders = olpHeaders({
|
||||
providerUsed: streamProvider,
|
||||
modelUsed: streamModel,
|
||||
startMs,
|
||||
cacheStatus: cacheStatusForHeader,
|
||||
fallbackHops: 0,
|
||||
});
|
||||
if (role === 'source' || role === 'attached') {
|
||||
streamHeaders['X-OLP-Streaming-Inflight'] = role;
|
||||
}
|
||||
|
||||
// D14: writeHead is deferred until just before the first res.write so
|
||||
// that pre-first-chunk errors can still produce a JSON 502 (matching
|
||||
// the buffered path). Calling writeHead unconditionally was the D14
|
||||
// defect.
|
||||
|
||||
const streamedChunks = [];
|
||||
let firstChunkEmitted = false;
|
||||
try {
|
||||
for await (const irChunk of stream) {
|
||||
if (irChunk.type === 'error') {
|
||||
// Error chunk from provider
|
||||
if (firstChunkEmitted) {
|
||||
// Past first-chunk boundary — emit truncation marker + [DONE].
|
||||
logEvent('warn', 'streaming_error_after_first_chunk', {
|
||||
provider: streamProvider,
|
||||
model: streamModel,
|
||||
error: irChunk.error,
|
||||
});
|
||||
auditCtx.error_code = 'streaming_error_after_first_chunk';
|
||||
res.write(irChunkToOpenAISSE({ type: 'stop', finish_reason: 'length' }, requestId, ir.model));
|
||||
res.write(SSE_DONE);
|
||||
res.end();
|
||||
return;
|
||||
}
|
||||
throw new ProviderError(irChunk.error ?? 'Provider emitted error chunk', 'SPAWN_FAILED');
|
||||
}
|
||||
|
||||
if (!res.headersSent) {
|
||||
res.writeHead(200, {
|
||||
'Content-Type': 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
Connection: 'keep-alive',
|
||||
'X-Accel-Buffering': 'no',
|
||||
...streamHeaders,
|
||||
});
|
||||
auditCtx.error_code = 'streaming_error_after_first_chunk';
|
||||
res.write(irChunkToOpenAISSE({ type: 'stop', finish_reason: 'length' }, requestId, ir.model));
|
||||
}
|
||||
streamedChunks.push(irChunk);
|
||||
res.write(irChunkToOpenAISSE(irChunk, requestId, ir.model));
|
||||
firstChunkEmitted = true;
|
||||
|
||||
if (irChunk.type === 'stop') {
|
||||
res.write(SSE_DONE);
|
||||
res.end();
|
||||
// D58 — ADR 0005 Amendment 8 §4: cache write is performed by the
|
||||
// cache layer's tee task on source completion. Server no longer
|
||||
// calls cacheStore.set from this branch; the cache layer applies
|
||||
// the same write conditions (truncated-not-cached, cacheable:false
|
||||
// opt-out, replay-cap, maxEntryBytes cap) internally.
|
||||
return;
|
||||
}
|
||||
// No bytes written yet — throw to surface a clean error.
|
||||
// auditCtx.error_code is set by the downstream catch handler (the
|
||||
// outer streaming-path catch block fills it from the thrown error).
|
||||
throw new ProviderError(irChunk.error ?? 'Provider emitted error chunk', 'SPAWN_FAILED');
|
||||
}
|
||||
|
||||
// Defer writeHead until the moment we are about to emit the first byte.
|
||||
// After this point firstChunkEmitted===true ↔ res.headersSent===true.
|
||||
// Generator exhausted without a stop chunk — emit truncation marker
|
||||
// + [DONE]. D58 — the cache layer (per ADR 0005 Amendment 8 §4) writes
|
||||
// accumulatedChunks on normal source exhaustion regardless of stop-
|
||||
// chunk semantics; IR-level stop-chunk inspection is the server's
|
||||
// responsibility. Mirror D16 truncation-eviction: the source role
|
||||
// explicitly deletes the just-written entry so subsequent identical
|
||||
// requests respawn (matches D16 buffered-path truncation behaviour
|
||||
// and ADR 0005 § "Cache write conditions" item 1). Only the source
|
||||
// role performs the delete; attached clients did not trigger the
|
||||
// write. Idempotent across concurrent source-truncation observers.
|
||||
if (streamedChunks.length > 0) {
|
||||
const truncMarker = { type: 'stop', finish_reason: 'length' };
|
||||
res.write(irChunkToOpenAISSE(truncMarker, requestId, ir.model));
|
||||
}
|
||||
if (role === 'source' && cacheableForFirstHop) {
|
||||
cacheStore.delete(keyId, streamCacheKey);
|
||||
}
|
||||
|
||||
// D35 #9: zero-chunk empty-stream — writeHead deferred (no chunks
|
||||
// yielded), emit headers + [DONE] so the response is still valid SSE.
|
||||
if (!res.headersSent) {
|
||||
res.writeHead(200, {
|
||||
'Content-Type': 'text/event-stream',
|
||||
@@ -1216,114 +1356,51 @@ async function handleChatCompletions(req, res) {
|
||||
...streamHeaders,
|
||||
});
|
||||
}
|
||||
streamedChunks.push(irChunk);
|
||||
res.write(irChunkToOpenAISSE(irChunk, requestId, ir.model));
|
||||
firstChunkEmitted = true;
|
||||
|
||||
if (irChunk.type === 'stop') {
|
||||
res.write(SSE_DONE);
|
||||
res.end();
|
||||
// Cache the buffered chunks for burst-replay on subsequent identical requests.
|
||||
// D23 defense-in-depth: cacheableForFirstHop is true here (cacheable: false
|
||||
// falls through to the buffered path, never enters this block), but the guard
|
||||
// makes the intent explicit and survives future refactors.
|
||||
if (cacheableForFirstHop) {
|
||||
await cacheStore.set(keyId, streamCacheKey, streamedChunks);
|
||||
logEvent('info', 'streaming_response_cached', {
|
||||
provider: streamProvider,
|
||||
model: streamModel,
|
||||
chunks: streamedChunks.length,
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Generator exhausted without a stop chunk — emit [DONE] but do NOT cache.
|
||||
// A stop-less exhaustion means the response is truncated (the generator
|
||||
// ended without the model signalling completion). Caching a truncated
|
||||
// response would serve wrong answers to future identical requests.
|
||||
// Compare: D16's buffered-path truncation eviction explicitly avoids
|
||||
// persisting truncated entries for the same reason.
|
||||
//
|
||||
// D26 round-3 F19: emit a synthetic truncation marker BEFORE [DONE] so
|
||||
// clients can detect the incomplete response in-band. Only emit when there
|
||||
// is actual partial content (streamedChunks.length > 0) — emitting a
|
||||
// truncation marker on an empty response is misleading.
|
||||
// The buffered D16 path synthesizes {type:'stop', finish_reason:'length'}
|
||||
// before returning; this aligns the streaming branch with that behaviour.
|
||||
if (streamedChunks.length > 0) {
|
||||
const truncMarker = { type: 'stop', finish_reason: 'length' };
|
||||
res.write(irChunkToOpenAISSE(truncMarker, requestId, ir.model));
|
||||
}
|
||||
|
||||
// D35 #9: Zero-chunk empty-stream path — writeHead is still deferred
|
||||
// (firstChunkEmitted===false) when the generator yields no chunks at all and
|
||||
// exits cleanly. Without an explicit writeHead Node auto-emits 200 with the
|
||||
// default Content-Type and none of the X-OLP-* headers.
|
||||
// A provider that yielded nothing still constitutes an attempted call, so we
|
||||
// emit the full olpHeaders (provider WAS attempted, just yielded zero chunks).
|
||||
if (!res.headersSent) {
|
||||
res.writeHead(200, {
|
||||
'Content-Type': 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
Connection: 'keep-alive',
|
||||
'X-Accel-Buffering': 'no',
|
||||
...streamHeaders,
|
||||
});
|
||||
}
|
||||
|
||||
res.write(SSE_DONE);
|
||||
res.end();
|
||||
// Loop exhausted without stop chunk = truncation. The stop-chunk completion
|
||||
// path returns earlier (above, inside the for-await loop); reaching here means
|
||||
// the generator returned without emitting stop. Never cache (per ADR 0005 cache
|
||||
// write conditions item 1: truncated responses must not persist in cache).
|
||||
if (streamedChunks.length > 0 && cacheableForFirstHop) {
|
||||
logEvent('warn', 'streaming_no_stop_chunk', {
|
||||
chunks_count: streamedChunks.length,
|
||||
provider: streamProvider,
|
||||
model: streamModel,
|
||||
});
|
||||
}
|
||||
} catch (e) {
|
||||
if (firstChunkEmitted) {
|
||||
// Past first-chunk boundary — can't fallback; emit truncation marker + [DONE]
|
||||
// so clients can detect the incomplete response in-band (aligns with D26 F19
|
||||
// stop-less exhaustion behaviour). ADR 0004 § Fallback safety: no fallback
|
||||
// after first-chunk boundary; truncation is the correct recovery.
|
||||
logEvent('warn', 'streaming_error_after_first_chunk', {
|
||||
provider: streamProvider,
|
||||
model: streamModel,
|
||||
error: e.message,
|
||||
});
|
||||
res.write(irChunkToOpenAISSE({ type: 'stop', finish_reason: 'length' }, requestId, ir.model));
|
||||
res.write(SSE_DONE);
|
||||
res.end();
|
||||
} else {
|
||||
// No bytes written — surface a clean JSON error.
|
||||
logEvent('error', 'streaming_error_before_first_chunk', {
|
||||
provider: streamProvider,
|
||||
model: streamModel,
|
||||
error: e.message,
|
||||
});
|
||||
auditCtx.error_code = e?.code ?? 'provider_error';
|
||||
if (!res.headersSent) {
|
||||
sendError(res, 502, e.message ?? 'Provider error', 'provider_error',
|
||||
olpHeaders({ providerUsed: streamProvider, modelUsed: streamModel, startMs, cacheStatus: 'miss', fallbackHops: 0 }));
|
||||
} else {
|
||||
res.end();
|
||||
if (streamedChunks.length > 0 && cacheableForFirstHop) {
|
||||
logEvent('warn', 'streaming_no_stop_chunk', {
|
||||
chunks_count: streamedChunks.length,
|
||||
provider: streamProvider,
|
||||
model: streamModel,
|
||||
});
|
||||
}
|
||||
} catch (e) {
|
||||
if (firstChunkEmitted) {
|
||||
logEvent('warn', 'streaming_error_after_first_chunk', {
|
||||
provider: streamProvider,
|
||||
model: streamModel,
|
||||
error: e.message,
|
||||
});
|
||||
res.write(irChunkToOpenAISSE({ type: 'stop', finish_reason: 'length' }, requestId, ir.model));
|
||||
res.write(SSE_DONE);
|
||||
res.end();
|
||||
} else {
|
||||
// No bytes written — surface a clean JSON error.
|
||||
logEvent('error', 'streaming_error_before_first_chunk', {
|
||||
provider: streamProvider,
|
||||
model: streamModel,
|
||||
error: e.message,
|
||||
});
|
||||
auditCtx.error_code = e?.code ?? 'provider_error';
|
||||
if (!res.headersSent) {
|
||||
sendError(res, 502, e.message ?? 'Provider error', 'provider_error',
|
||||
olpHeaders({ providerUsed: streamProvider, modelUsed: streamModel, startMs, cacheStatus: 'miss', fallbackHops: 0 }));
|
||||
} else {
|
||||
res.end();
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
// D58 — releaseSpawn lives inside sourceFactory's finally (only the
|
||||
// source caller acquired). The req-close listener is detached here
|
||||
// so it doesn't leak past the response lifetime.
|
||||
res.removeListener('close', onClose);
|
||||
}
|
||||
} finally {
|
||||
// D38 (issue #1): streaming spawn lifecycle ended — drain completed,
|
||||
// stop chunk seen, generator exhausted without stop, or any catch
|
||||
// path returned via res.end(). Release the slot acquired before
|
||||
// entering the streaming branch. The finally fires on every JS exit
|
||||
// path including the `return;` statements inside the try/catch body.
|
||||
releaseSpawn(streamProvider);
|
||||
return;
|
||||
}
|
||||
return;
|
||||
// CONCURRENCY_LIMIT fallthrough — flow continues into the buffered path
|
||||
// below.
|
||||
}
|
||||
|
||||
let fallbackResult;
|
||||
|
||||
+894
-1
@@ -9,7 +9,7 @@
|
||||
* D5 adds: Suite 9 (cache layer unit + HTTP integration), Suite 10 (Anthropic E2E gated)
|
||||
*/
|
||||
|
||||
import { describe, it, before, after } from 'node:test';
|
||||
import { describe, it, before, after, beforeEach } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { request as httpRequest } from 'node:http';
|
||||
import { EventEmitter } from 'node:events';
|
||||
@@ -12035,3 +12035,896 @@ 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);
|
||||
});
|
||||
});
|
||||
|
||||
// ── Suite 28: D58 streaming singleflight (server.mjs HTTP wiring) ──────────
|
||||
//
|
||||
// Authority: ADR 0005 Amendment 8 §§ 7, 8, 9, 11, 12 (issue #16, D58).
|
||||
//
|
||||
// HTTP-layer integration tests for the server.mjs streaming branch wired to
|
||||
// cacheStore.getOrComputeStreaming(). D57 (cache layer) tested singleflight
|
||||
// in unit form; D58 verifies the server actually plumbs it through and emits
|
||||
// the new X-OLP-Streaming-Inflight header. Provider spawn is mocked at the
|
||||
// plugin level via lp.set('anthropic', { ...real, spawn: fake }) — the
|
||||
// pattern used by Suites 15d/15e — so no real CLI is invoked.
|
||||
//
|
||||
// Test count: 8 (28a single, 28b 2-concurrent join, 28c TOCTOU pre-cache,
|
||||
// 28d mid-stream join behaviour-validated via parallel HTTP, 28e omitted in
|
||||
// favour of Suite 27g unit coverage per D58 prompt, 28f one-of-N disconnect,
|
||||
// 28g all-disconnect, 28h CONCURRENCY_LIMIT fallthrough).
|
||||
|
||||
describe('Suite 28 — D58 streaming singleflight (server.mjs HTTP wiring, ADR 0005 Amendment 8)', () => {
|
||||
let server28;
|
||||
let port28;
|
||||
let savedToken28;
|
||||
let lp28;
|
||||
let savedAnthropic28;
|
||||
|
||||
before(async () => {
|
||||
savedToken28 = process.env.CLAUDE_CODE_OAUTH_TOKEN;
|
||||
process.env.CLAUDE_CODE_OAUTH_TOKEN = 'fake-token-suite-28';
|
||||
__setProvidersEnabled({ anthropic: true });
|
||||
|
||||
const mod = await import('./server.mjs');
|
||||
lp28 = mod.loadedProviders;
|
||||
savedAnthropic28 = lp28.get('anthropic');
|
||||
mod.__clearCache();
|
||||
|
||||
server28 = mod.createOlpServer();
|
||||
await new Promise((resolve, reject) => {
|
||||
server28.listen(0, '127.0.0.1', resolve);
|
||||
server28.once('error', reject);
|
||||
});
|
||||
port28 = server28.address().port;
|
||||
});
|
||||
|
||||
after(async () => {
|
||||
// Restore original anthropic provider so other suites are unaffected.
|
||||
if (savedAnthropic28 !== undefined) {
|
||||
lp28.set('anthropic', savedAnthropic28);
|
||||
} else {
|
||||
lp28.delete('anthropic');
|
||||
}
|
||||
__resetProvidersEnabled();
|
||||
__resetSpawnImpl();
|
||||
if (savedToken28 !== undefined) {
|
||||
process.env.CLAUDE_CODE_OAUTH_TOKEN = savedToken28;
|
||||
} else {
|
||||
delete process.env.CLAUDE_CODE_OAUTH_TOKEN;
|
||||
}
|
||||
if (!server28) return;
|
||||
return new Promise(r => server28.close(r));
|
||||
});
|
||||
|
||||
/**
|
||||
* Install a custom spawn async generator on the anthropic plugin for one
|
||||
* test. `factory` is a function returning the async generator each spawn.
|
||||
* Tracks invocation count via spawnCount.
|
||||
*/
|
||||
function installFakeStreamProvider(spawnImpl) {
|
||||
const counter = { count: 0 };
|
||||
const fake = {
|
||||
...savedAnthropic28,
|
||||
spawn: async function* (ir, authContext) {
|
||||
counter.count++;
|
||||
yield* spawnImpl(ir, authContext);
|
||||
},
|
||||
};
|
||||
lp28.set('anthropic', fake);
|
||||
return counter;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fire an SSE request and collect the full body + headers. The promise
|
||||
* resolves when the server ends the response (res 'end' event).
|
||||
*/
|
||||
function makeStreamRequest(extra = {}) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const req = httpRequest({
|
||||
hostname: '127.0.0.1',
|
||||
port: port28,
|
||||
method: 'POST',
|
||||
path: '/v1/chat/completions',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
}, res => {
|
||||
let data = '';
|
||||
res.on('data', d => { data += d.toString(); });
|
||||
res.on('end', () => resolve({
|
||||
status: res.statusCode,
|
||||
body: data,
|
||||
headers: res.headers,
|
||||
}));
|
||||
res.on('error', reject);
|
||||
});
|
||||
req.on('error', reject);
|
||||
req.write(JSON.stringify({
|
||||
model: extra.model ?? 'claude-sonnet-4-6',
|
||||
messages: [{ role: 'user', content: extra.prompt ?? 'd58-default' }],
|
||||
stream: true,
|
||||
}));
|
||||
req.end();
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Fire an SSE request and abort it after `abortAfterMs` ms. Returns the
|
||||
* partial body collected up to abort. Resolves on the abort event.
|
||||
*/
|
||||
function makeAbortableStreamRequest({ prompt, abortAfterMs }) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const req = httpRequest({
|
||||
hostname: '127.0.0.1',
|
||||
port: port28,
|
||||
method: 'POST',
|
||||
path: '/v1/chat/completions',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
}, res => {
|
||||
let data = '';
|
||||
res.on('data', d => { data += d.toString(); });
|
||||
res.on('end', () => resolve({
|
||||
status: res.statusCode,
|
||||
body: data,
|
||||
headers: res.headers,
|
||||
aborted: false,
|
||||
}));
|
||||
res.on('error', () => resolve({
|
||||
status: res.statusCode,
|
||||
body: data,
|
||||
headers: res.headers,
|
||||
aborted: true,
|
||||
}));
|
||||
});
|
||||
req.on('error', () => resolve({ aborted: true }));
|
||||
req.write(JSON.stringify({
|
||||
model: 'claude-sonnet-4-6',
|
||||
messages: [{ role: 'user', content: prompt }],
|
||||
stream: true,
|
||||
}));
|
||||
req.end();
|
||||
setTimeout(() => {
|
||||
try { req.destroy(); } catch { /* ignore */ }
|
||||
}, abortAfterMs);
|
||||
});
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
const mod = await import('./server.mjs');
|
||||
mod.__clearCache();
|
||||
});
|
||||
|
||||
it('28a — single SSE request: X-OLP-Streaming-Inflight: source; cache populated; identical re-request → cache hit', async () => {
|
||||
// D58 — ADR 0005 Amendment 8 §11: single client gets `source` role,
|
||||
// X-OLP-Cache: miss. Subsequent identical request hits the cache.
|
||||
const counter = installFakeStreamProvider(async function* (_ir) {
|
||||
yield { type: 'delta', content: 'chunk-A' };
|
||||
yield { type: 'delta', content: 'chunk-B' };
|
||||
yield { type: 'stop', finish_reason: 'stop' };
|
||||
});
|
||||
|
||||
const r1 = await makeStreamRequest({ prompt: 'd58-28a' });
|
||||
assert.equal(r1.status, 200, `r1 status ${r1.status}: ${r1.body.slice(0, 200)}`);
|
||||
assert.equal(r1.headers['x-olp-streaming-inflight'], 'source',
|
||||
'r1 must emit X-OLP-Streaming-Inflight: source');
|
||||
assert.equal(r1.headers['x-olp-cache'], 'miss',
|
||||
'r1 must be X-OLP-Cache: miss');
|
||||
assert.ok(r1.body.includes('chunk-A'), 'r1 body must include chunk-A');
|
||||
assert.ok(r1.body.includes('chunk-B'), 'r1 body must include chunk-B');
|
||||
assert.ok(r1.body.includes('[DONE]'), 'r1 body must end with [DONE]');
|
||||
assert.equal(counter.count, 1, 'exactly one spawn for r1');
|
||||
|
||||
// Second identical request — cache hit, no spawn.
|
||||
const r2 = await makeStreamRequest({ prompt: 'd58-28a' });
|
||||
assert.equal(r2.status, 200);
|
||||
assert.equal(r2.headers['x-olp-cache'], 'hit', 'r2 must be cache hit');
|
||||
// X-OLP-Streaming-Inflight is omitted on cache_hit (X-OLP-Cache: hit
|
||||
// already signals that path per D58 design).
|
||||
assert.equal(r2.headers['x-olp-streaming-inflight'], undefined,
|
||||
'cache_hit path must omit X-OLP-Streaming-Inflight header');
|
||||
assert.equal(counter.count, 1, 'no additional spawn for r2');
|
||||
});
|
||||
|
||||
it('28b — 2 concurrent identical SSE: one spawn; first=source second=attached; identical chunks', async () => {
|
||||
// D58 — ADR 0005 Amendment 8 §1, §11: two concurrent identical streams
|
||||
// share one underlying spawn. First gets X-OLP-Streaming-Inflight:
|
||||
// source; second gets attached. Both bodies contain the same chunks.
|
||||
//
|
||||
// Pacing: source yields chunks with a setTimeout gap so the second
|
||||
// request has time to fire and attach mid-stream.
|
||||
const counter = installFakeStreamProvider(async function* (_ir) {
|
||||
// Slow pacing so the second request can attach before completion.
|
||||
for (const t of ['c0', 'c1', 'c2', 'c3']) {
|
||||
yield { type: 'delta', content: t };
|
||||
await new Promise(r => setTimeout(r, 20));
|
||||
}
|
||||
yield { type: 'stop', finish_reason: 'stop' };
|
||||
});
|
||||
|
||||
// Fire request 1, then a few ms later fire request 2; both run to
|
||||
// completion in parallel.
|
||||
const p1 = makeStreamRequest({ prompt: 'd58-28b' });
|
||||
// Give p1 a head start to register the inflight entry.
|
||||
await new Promise(r => setTimeout(r, 10));
|
||||
const p2 = makeStreamRequest({ prompt: 'd58-28b' });
|
||||
|
||||
const [r1, r2] = await Promise.all([p1, p2]);
|
||||
assert.equal(r1.status, 200);
|
||||
assert.equal(r2.status, 200);
|
||||
assert.equal(counter.count, 1, 'only one underlying spawn for both requests');
|
||||
|
||||
// First caller (registered the inflight entry) gets source; the other
|
||||
// gets attached. The order is set by who hit getOrComputeStreaming
|
||||
// first — we asserted that with the 10ms head start above.
|
||||
assert.equal(r1.headers['x-olp-streaming-inflight'], 'source',
|
||||
'r1 must be source role');
|
||||
assert.equal(r2.headers['x-olp-streaming-inflight'], 'attached',
|
||||
'r2 must be attached role');
|
||||
|
||||
// Both responses must contain all 4 delta chunks + [DONE].
|
||||
for (const tag of ['c0', 'c1', 'c2', 'c3', '[DONE]']) {
|
||||
assert.ok(r1.body.includes(tag), `r1 missing ${tag}`);
|
||||
assert.ok(r2.body.includes(tag), `r2 missing ${tag}`);
|
||||
}
|
||||
|
||||
// Cache_status header: source=miss, attached=hit (cache_hit role)?
|
||||
// No — attached client did NOT hit the cache (cache was empty). The
|
||||
// server distinguishes by setting cache_status header to 'miss' for
|
||||
// attached and 'miss' for source. Verify both are 'miss' on the wire
|
||||
// (X-OLP-Cache header). cache_status='streaming_attached' is the AUDIT
|
||||
// value, not the wire header value.
|
||||
assert.equal(r1.headers['x-olp-cache'], 'miss');
|
||||
assert.equal(r2.headers['x-olp-cache'], 'miss');
|
||||
});
|
||||
|
||||
it('28c — TOCTOU regression: pre-populated cache + 2 concurrent identical streams → both cache_hit, no spawn', async () => {
|
||||
// D58 — ADR 0005 Amendment 8 §6 / §11: when the cache is already
|
||||
// populated, preCheckHit gates entry to the streaming-singleflight
|
||||
// branch. The streaming-singleflight branch should not run at all.
|
||||
// 2 concurrent identical requests both replay from cache; spawn count
|
||||
// stays at 0.
|
||||
const counter = installFakeStreamProvider(async function* (_ir) {
|
||||
yield { type: 'delta', content: 'should-not-fire' };
|
||||
yield { type: 'stop', finish_reason: 'stop' };
|
||||
});
|
||||
|
||||
// Populate the cache via a first request.
|
||||
const r0 = await makeStreamRequest({ prompt: 'd58-28c' });
|
||||
assert.equal(r0.status, 200);
|
||||
assert.equal(counter.count, 1, 'r0 spawned once to populate cache');
|
||||
|
||||
// Now fire 2 concurrent identical requests — both should be cache hits.
|
||||
const [r1, r2] = await Promise.all([
|
||||
makeStreamRequest({ prompt: 'd58-28c' }),
|
||||
makeStreamRequest({ prompt: 'd58-28c' }),
|
||||
]);
|
||||
assert.equal(r1.status, 200);
|
||||
assert.equal(r2.status, 200);
|
||||
assert.equal(r1.headers['x-olp-cache'], 'hit', 'r1 must be cache hit');
|
||||
assert.equal(r2.headers['x-olp-cache'], 'hit', 'r2 must be cache hit');
|
||||
// Neither should have entered the streaming-singleflight branch.
|
||||
assert.equal(r1.headers['x-olp-streaming-inflight'], undefined,
|
||||
'r1 must not emit X-OLP-Streaming-Inflight (served from buffered cache replay)');
|
||||
assert.equal(r2.headers['x-olp-streaming-inflight'], undefined,
|
||||
'r2 must not emit X-OLP-Streaming-Inflight (served from buffered cache replay)');
|
||||
assert.equal(counter.count, 1, 'no additional spawns');
|
||||
});
|
||||
|
||||
it('28d — mid-stream join (HTTP-level): 2 concurrent SSE requests share one spawn; both bodies identical', async () => {
|
||||
// D58 — ADR 0005 Amendment 8 §5: late joiner gets accumulated burst +
|
||||
// live tail. At the HTTP level we can't observe the burst-vs-live split
|
||||
// precisely (it's internal to the cache layer), but we can verify the
|
||||
// end-to-end invariant: both clients receive the same chunk sequence
|
||||
// even when one joins mid-source. Suite 27c covers the burst-vs-live
|
||||
// split at the cache-layer unit level.
|
||||
const counter = installFakeStreamProvider(async function* (_ir) {
|
||||
// Slower pacing so the second client clearly joins mid-stream.
|
||||
for (const t of ['m0', 'm1', 'm2', 'm3', 'm4']) {
|
||||
yield { type: 'delta', content: t };
|
||||
await new Promise(r => setTimeout(r, 25));
|
||||
}
|
||||
yield { type: 'stop', finish_reason: 'stop' };
|
||||
});
|
||||
|
||||
const p1 = makeStreamRequest({ prompt: 'd58-28d' });
|
||||
// p2 joins ~40ms in — at least 2 chunks have been accumulated by then.
|
||||
await new Promise(r => setTimeout(r, 40));
|
||||
const p2 = makeStreamRequest({ prompt: 'd58-28d' });
|
||||
|
||||
const [r1, r2] = await Promise.all([p1, p2]);
|
||||
assert.equal(r1.status, 200);
|
||||
assert.equal(r2.status, 200);
|
||||
assert.equal(counter.count, 1, 'exactly one underlying spawn');
|
||||
assert.equal(r1.headers['x-olp-streaming-inflight'], 'source');
|
||||
assert.equal(r2.headers['x-olp-streaming-inflight'], 'attached');
|
||||
for (const tag of ['m0', 'm1', 'm2', 'm3', 'm4']) {
|
||||
assert.ok(r1.body.includes(tag), `r1 missing ${tag}`);
|
||||
assert.ok(r2.body.includes(tag), `r2 missing ${tag} (late-joiner replay must include burst)`);
|
||||
}
|
||||
});
|
||||
|
||||
it('28f — one of N clients disconnects mid-stream: source NOT aborted; other client completes; cache populated', async () => {
|
||||
// D58 — ADR 0005 Amendment 8 §9: when one of N attached clients drops,
|
||||
// the source continues for the remaining clients. The cache write
|
||||
// happens because the source completed normally for the surviving
|
||||
// client.
|
||||
const counter = installFakeStreamProvider(async function* (_ir) {
|
||||
for (const t of ['s0', 's1', 's2', 's3', 's4', 's5']) {
|
||||
yield { type: 'delta', content: t };
|
||||
await new Promise(r => setTimeout(r, 25));
|
||||
}
|
||||
yield { type: 'stop', finish_reason: 'stop' };
|
||||
});
|
||||
|
||||
// r1 starts and registers the inflight entry.
|
||||
const p1 = makeStreamRequest({ prompt: 'd58-28f' });
|
||||
await new Promise(r => setTimeout(r, 15));
|
||||
// r2 attaches but aborts after ~40ms.
|
||||
const p2 = makeAbortableStreamRequest({ prompt: 'd58-28f', abortAfterMs: 40 });
|
||||
|
||||
const [r1, r2] = await Promise.all([p1, p2]);
|
||||
// r1 (the source) completes normally with the full stream.
|
||||
assert.equal(r1.status, 200);
|
||||
assert.equal(r1.headers['x-olp-streaming-inflight'], 'source');
|
||||
for (const tag of ['s0', 's1', 's2', 's3', 's4', 's5', '[DONE]']) {
|
||||
assert.ok(r1.body.includes(tag), `r1 missing ${tag}`);
|
||||
}
|
||||
assert.equal(counter.count, 1, 'source spawn was NOT re-fired by the disconnect');
|
||||
|
||||
// r2 was aborted — partial body is okay; what matters is the source
|
||||
// wasn't killed (verified above by r1 completing).
|
||||
// Subsequent identical request must be a cache hit (cache was populated
|
||||
// by the source on normal completion).
|
||||
const r3 = await makeStreamRequest({ prompt: 'd58-28f' });
|
||||
assert.equal(r3.status, 200);
|
||||
assert.equal(r3.headers['x-olp-cache'], 'hit', 'cache populated after source completion');
|
||||
assert.equal(counter.count, 1, 'no additional spawn for r3');
|
||||
});
|
||||
|
||||
it('28g — ALL clients disconnect mid-stream: source aborted; no cache write; subsequent request respawns', async () => {
|
||||
// D58 — ADR 0005 Amendment 8 §9 + §4: when all attached clients
|
||||
// disconnect, the cache layer fires sourceAbortController.abort() and
|
||||
// does NOT write the cache. A subsequent identical request must spawn
|
||||
// afresh (no inflight entry left dangling).
|
||||
//
|
||||
// Implementation detail: the fake spawn's async generator must respect
|
||||
// the abort signal (or simply have its iterator.return() called when
|
||||
// the spawn is iterated by the tee task — see Suite 27 unit tests).
|
||||
// For an async generator with `await new Promise(setTimeout)` between
|
||||
// yields, calling .return() naturally propagates because the for-await
|
||||
// exits cleanly via the try/finally.
|
||||
let sourceFinished = false;
|
||||
const counter = installFakeStreamProvider(async function* (_ir) {
|
||||
try {
|
||||
for (let i = 0; i < 30; i++) {
|
||||
yield { type: 'delta', content: `g${i}` };
|
||||
await new Promise(r => setTimeout(r, 20));
|
||||
}
|
||||
yield { type: 'stop', finish_reason: 'stop' };
|
||||
sourceFinished = true;
|
||||
} finally {
|
||||
// intentionally no-op; the "finished without abort" signal lives in
|
||||
// sourceFinished above.
|
||||
}
|
||||
});
|
||||
|
||||
// Only one client; abort it after ~40ms (well before completion).
|
||||
const r1 = await makeAbortableStreamRequest({ prompt: 'd58-28g', abortAfterMs: 40 });
|
||||
assert.equal(counter.count, 1, 'spawn fired once');
|
||||
// Wait for the cache layer to observe attachedClients.size === 0 and abort.
|
||||
await new Promise(r => setTimeout(r, 100));
|
||||
assert.equal(sourceFinished, false, 'source was aborted, not completed');
|
||||
|
||||
// Subsequent identical request must respawn (no cache, no inflight).
|
||||
const r2 = await makeStreamRequest({ prompt: 'd58-28g' });
|
||||
assert.equal(r2.status, 200);
|
||||
assert.equal(r2.headers['x-olp-cache'], 'miss', 'no cache write after abort');
|
||||
assert.equal(r2.headers['x-olp-streaming-inflight'], 'source', 'fresh source role');
|
||||
assert.equal(counter.count, 2, 'r2 triggered a fresh spawn');
|
||||
});
|
||||
|
||||
it('28h — CONCURRENCY_LIMIT fallthrough: different cacheKeys at maxConcurrent=1 → first succeeds; second falls through to buffered path', async () => {
|
||||
// D58 — ADR 0005 Amendment 8 §7: CONCURRENCY_LIMIT thrown by
|
||||
// sourceFactory falls through to the buffered path. Different
|
||||
// cacheKeys do NOT share an inflight entry (they're not singleflight
|
||||
// candidates), so request 2's sourceFactory throws and the streaming
|
||||
// branch is bypassed for that request. The buffered path then
|
||||
// re-attempts acquire; since maxConcurrent=1 and request 1 still has
|
||||
// the slot, it surfaces a chain-exhausted 502 (single-hop saturation).
|
||||
//
|
||||
// We install a custom provider with maxConcurrent=1 + slow stream to
|
||||
// hold the slot while request 2 fires.
|
||||
let releaseSig;
|
||||
const releaseGate = new Promise(r => { releaseSig = r; });
|
||||
const counter = { count: 0 };
|
||||
const fake = {
|
||||
...savedAnthropic28,
|
||||
hints: { ...savedAnthropic28.hints, maxConcurrent: 1 },
|
||||
spawn: async function* (_ir, _ctx) {
|
||||
counter.count++;
|
||||
// First spawn holds the slot until releaseGate is signalled.
|
||||
yield { type: 'delta', content: `h${counter.count}-0` };
|
||||
await releaseGate;
|
||||
yield { type: 'stop', finish_reason: 'stop' };
|
||||
},
|
||||
};
|
||||
lp28.set('anthropic', fake);
|
||||
|
||||
try {
|
||||
// Fire request 1 with prompt-A; it acquires the slot and stalls.
|
||||
const p1 = makeStreamRequest({ prompt: 'd58-28h-A' });
|
||||
await new Promise(r => setTimeout(r, 30));
|
||||
// Fire request 2 with prompt-B (different cache key) — should
|
||||
// CONCURRENCY_LIMIT in factory, fall through to buffered path,
|
||||
// which will also fail acquire → chain-exhausted 502.
|
||||
const r2 = await makeStreamRequest({ prompt: 'd58-28h-B' });
|
||||
// The buffered fallback engine surfaces a chain-exhausted error as
|
||||
// 502 for single-hop saturation per pre-D58 behaviour.
|
||||
assert.ok(r2.status === 502 || r2.status === 503,
|
||||
`expected 502/503 chain-exhausted, got ${r2.status}: ${r2.body.slice(0, 200)}`);
|
||||
|
||||
// Release request 1's stall.
|
||||
releaseSig();
|
||||
const r1 = await p1;
|
||||
assert.equal(r1.status, 200, 'r1 must complete normally');
|
||||
// Exactly one spawn — request 2 never spawned (CONCURRENCY_LIMIT).
|
||||
assert.equal(counter.count, 1, 'only one underlying spawn');
|
||||
} finally {
|
||||
// In case of an early failure, release the gate so promises settle.
|
||||
try { releaseSig(); } catch { /* already released */ }
|
||||
}
|
||||
});
|
||||
|
||||
it('28i — stop-less exhaustion: source generator returns without {type:"stop"} → cache NOT populated (D58 follow-up to D58 reviewer P2-2)', async () => {
|
||||
// ADR 0005 § "Cache write conditions" item 1 (D16 truncated-not-cached
|
||||
// invariant): if the source generator exhausts without emitting a stop
|
||||
// chunk, the response is treated as truncated and MUST NOT persist in
|
||||
// cache. D57's cache layer is IR-agnostic and writes accumulatedChunks
|
||||
// on any source exhaustion; D58's server.mjs handles the IR semantics by
|
||||
// calling cacheStore.delete(...) immediately after on the no-stop path.
|
||||
// This test pins the end-to-end behaviour from the HTTP layer.
|
||||
const counter = { count: 0 };
|
||||
const fake = {
|
||||
...savedAnthropic28,
|
||||
spawn: async function* (_ir, _ctx) {
|
||||
counter.count++;
|
||||
yield { type: 'delta', content: 'no-stop-1' };
|
||||
yield { type: 'delta', content: 'no-stop-2' };
|
||||
// Generator returns WITHOUT a stop chunk — truncation path.
|
||||
},
|
||||
};
|
||||
lp28.set('anthropic', fake);
|
||||
const r1 = await makeStreamRequest({ prompt: 'd58-28i' });
|
||||
assert.equal(r1.status, 200, '28i r1: SSE response 200');
|
||||
assert.equal(counter.count, 1, '28i r1: spawned once');
|
||||
// The synthetic truncation marker {type:"stop", finish_reason:"length"}
|
||||
// should appear (D26 F19 in-band signal); [DONE] terminator follows.
|
||||
assert.ok(r1.body.includes('"finish_reason":"length"'),
|
||||
`28i r1: truncation marker expected; body=${r1.body.slice(0, 300)}`);
|
||||
assert.ok(r1.body.includes('[DONE]'), '28i r1: [DONE] terminator');
|
||||
// Subsequent identical request must respawn (cache was NOT populated).
|
||||
const r2 = await makeStreamRequest({ prompt: 'd58-28i' });
|
||||
assert.equal(r2.status, 200, '28i r2: SSE response 200');
|
||||
assert.equal(counter.count, 2, '28i r2: second request triggered a fresh spawn (no cache reuse for truncated entry)');
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user