Compare commits

...
Author SHA1 Message Date
9494fd6c69 chore(release): v3.13.0 — cache layer hardening (per-key isolation + bypass + chunked replay + singleflight) (#67)
Per release_kit overlay (CLAUDE.md § Iron Rule 5.5):
- package.json bumped 3.12.0 → 3.13.0
- CHANGELOG.md updated with v3.13.0 entry
- README.md § Response Cache updated to document the four hardening features

This is a release preparation commit. Tag push to v3.13.0 will trigger
.github/workflows/release.yml which auto-creates the GitHub Release.

cli.js does not perform proxy-layer response caching, stampede protection,
or replay; this release only ships internal cache-layer correctness and
concurrency improvements that do not change the OpenAI-compatible wire
surface visible to clients. No client-observable wire shape change.

Co-authored-by: dtzp555 <dtzp555@gmail.com>
Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-07 17:19:44 +10:00
5ff30ac9b6 feat(cache): singleflight stampede protection on non-streaming path (#66)
cli.js does not perform proxy-layer stampede protection. The singleflight
layer is a value-add proxy operation that exists only inside OCP, between
concurrent client requests and the single upstream cli.js spawn. It does
not introduce, alter, or remove any endpoint, header, request field, or
response field that cli.js emits or expects — no client-observable wire
shape change.

Justification under ALIGNMENT.md Rule 2: the singleflight Map deduplicates
concurrent identical non-streaming cache-miss requests so only one cli.js
spawn runs per unique hash window. All followers receive the same resolved
(or rejected) content. This is a proxy-internal concurrency optimization,
not a wire-level protocol change.

Spec: docs/superpowers/specs/2026-05-07-cache-upgrade-design.md (D4)

Changes:
- keys.mjs: add singleflight(hash, fn) + getInflightStats() exports;
  in-memory Map cleared via Promise.finally() on each settlement
- server.mjs: import singleflight + getInflightStats; wrap non-streaming
  cache-enabled path through singleflight with inner recheck; add TODO
  comment in callClaudeStreaming (streaming-path dedup is explicitly out
  of scope for v3.13.0, see spec D4 streaming caveat); extend /cache/stats
  to return inflight + requesters fields (additive, no removed fields)
- test-features.mjs: 7 new PR-B singleflight tests covering basic dedup,
  failure fan-out, map cleanup (success + failure), different-hash
  independence, getInflightStats shape, and sequential-call non-sharing;
  all 31 tests pass (24 existing + 7 new)

Streaming-path singleflight is explicitly out of scope; TODO left in
callClaudeStreaming for a future follow-up ticket.

Co-authored-by: dtzp555 <dtzp555@gmail.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-07 17:16:46 +10:00
6 changed files with 202 additions and 13 deletions
+22
View File
@@ -1,5 +1,27 @@
# Changelog
## v3.13.0 — 2026-05-07
### Features (cache layer hardening)
- **Per-key cache isolation** (D1) — the cache key now includes the API key id, so distinct keys never share cache entries. Anonymous/unauthenticated callers share one `anon` pool. Hash format upgraded to `v2`; legacy v1-format rows orphan and are reaped by the existing TTL cleanup interval (no migration script).
- **`cache_control` bypass** (D2) — when a request carries an Anthropic `cache_control` annotation (top-level or nested in a content array), OCP skips its own cache entirely. The caller is using Anthropic-side prompt caching deliberately, and OCP must not interfere. A `cache_skipped{reason: cache_control_present}` log line is emitted on bypass.
- **Chunked stream replay** (D3) — when a streaming request hits the cache, the cached content is now emitted as multiple SSE chunks (80 codepoints/chunk, codepoint-safe via `Array.from()`) instead of a single large delta. Multibyte characters (CJK / emoji) stay intact.
- **Singleflight stampede protection** (D4) — concurrent identical cache-miss requests now share one upstream `cli.js` spawn instead of spawning N processes. Followers receive byte-identical responses to what the leader returns. All-or-nothing failure semantics: if the leader errors, all followers receive the same error. Streaming-path singleflight is explicitly out of scope (TODO left for follow-up).
### Behavior changes
- `/cache/stats` response now includes additive fields `inflight` and `requesters` (current in-flight singleflight entries and total waiting callers). Existing fields `entries`, `totalHits`, `sizeBytes` are preserved unchanged.
### Governance
- New ADR [`docs/adr/0005-no-multi-provider.md`](docs/adr/0005-no-multi-provider.md): OCP stays single-provider (Anthropic via `cli.js` spawn). Multi-provider gateway refactor explicitly out of scope; cache improvements are explicitly in scope.
- Design spec for this release: [`docs/superpowers/specs/2026-05-07-cache-upgrade-design.md`](docs/superpowers/specs/2026-05-07-cache-upgrade-design.md).
### No new env vars / no public API surface change
This release adds no new env vars or endpoints. All four improvements are internal correctness/concurrency upgrades to the existing `CLAUDE_CACHE_TTL`-gated cache layer. No client-observable wire shape change.
## v3.12.0 — 2026-04-25
### Features
+9 -4
View File
@@ -472,17 +472,20 @@ ocp settings cacheTTL 300000
```
**How it works:**
- Cache key = SHA-256 of `model` + `messages` + `temperature` + `max_tokens` + `top_p`
- Cache key = SHA-256 of `v2|<keyId or "anon">|model + messages + temperature + max_tokens + top_p`
- **Per-key isolation** — different API keys never share cache entries; anonymous callers share one `anon` pool
- Cache hits return instantly — no Claude CLI process spawned
- Works for both streaming and non-streaming requests
- **Streaming hits** are replayed as multiple SSE chunks (80 codepoints each), not one large delta — incremental render preserved
- **`cache_control` bypass** — if a request carries an Anthropic `cache_control` annotation (top-level or nested in `content[]`), OCP skips its own cache entirely so it doesn't interfere with Anthropic-side prompt caching
- **Singleflight stampede protection** — concurrent identical cache-miss requests share one upstream `cli.js` spawn; followers receive byte-identical responses to the leader's call. Non-streaming path only (streaming-path singleflight is a known TODO)
- Multi-turn conversations (with `session_id`) are never cached
- Expired entries are cleaned up automatically every 10 minutes
**Management:**
```bash
# View cache stats
# View cache stats (now includes singleflight in-flight counts)
curl http://127.0.0.1:3456/cache/stats
# → { "entries": 42, "totalHits": 156, "sizeBytes": 284000 }
# → { "entries": 42, "totalHits": 156, "sizeBytes": 284000, "inflight": 0, "requesters": 0 }
# Clear all cached responses
curl -X DELETE http://127.0.0.1:3456/cache
@@ -493,6 +496,8 @@ ocp settings cacheTTL 0
Cache is **disabled by default** (`CLAUDE_CACHE_TTL=0`). All data is stored locally in `~/.ocp/ocp.db`.
**Hash format upgrade in v3.13.0:** legacy `v1` cache rows from earlier versions don't match new `v2`-format lookups; they orphan and are reaped by the TTL cleanup interval within one window. No migration script required.
## How It Works
```
+30
View File
@@ -371,6 +371,36 @@ export function getCacheStats() {
return { entries: total, totalHits, sizeBytes };
}
// ── Singleflight stampede protection ──
// In-memory singleflight Map: hash → { promise, requesters }
// Deduplicates concurrent identical cache-miss flows so only one upstream call runs.
// Per ADR 0005 / spec D4: in-process scope only (single Node process per host).
const inflightMap = new Map();
export function singleflight(hash, fn) {
const existing = inflightMap.get(hash);
if (existing) {
existing.requesters++;
return existing.promise;
}
// Wrap fn() in Promise.resolve().then() so synchronous throws don't escape.
const promise = Promise.resolve().then(fn).finally(() => {
inflightMap.delete(hash);
});
inflightMap.set(hash, { promise, requesters: 1 });
return promise;
}
export function getInflightStats() {
let totalRequesters = 0;
for (const entry of inflightMap.values()) totalRequesters += entry.requesters;
return {
inflight: inflightMap.size,
requesters: totalRequesters,
};
}
// Find a key by id or name (returns { id, name } or null)
export function findKey(idOrName) {
const d = getDb();
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "open-claude-proxy",
"version": "3.12.0",
"version": "3.13.0",
"description": "OCP (Open Claude Proxy) — use your Claude Pro/Max subscription as an OpenAI-compatible API for any IDE. Works with Cline, OpenCode, Aider, Continue.dev, OpenClaw, and more.",
"type": "module",
"bin": {
+39 -7
View File
@@ -34,7 +34,7 @@ import { readFileSync, accessSync, existsSync, constants } from "node:fs";
import { fileURLToPath } from "node:url";
import { dirname, join } from "node:path";
import { homedir } from "node:os";
import { validateKey, recordUsage, getUsageByKey, getUsageTimeline, getRecentUsage, createKey, listKeys, revokeKey, closeDb, checkQuota, updateKeyQuota, getKeyQuota, findKey, cacheHash, getCachedResponse, setCachedResponse, clearCache, getCacheStats, hasCacheControl } from "./keys.mjs";
import { validateKey, recordUsage, getUsageByKey, getUsageTimeline, getRecentUsage, createKey, listKeys, revokeKey, closeDb, checkQuota, updateKeyQuota, getKeyQuota, findKey, cacheHash, getCachedResponse, setCachedResponse, clearCache, getCacheStats, hasCacheControl, singleflight, getInflightStats } from "./keys.mjs";
const __dirname = dirname(fileURLToPath(import.meta.url));
const _pkg = JSON.parse(readFileSync(join(__dirname, "package.json"), "utf8"));
@@ -590,6 +590,7 @@ function startHeartbeat(res, intervalMs, sessionId) {
// ── Call claude CLI (real streaming) ─────────────────────────────────────
// Pipes stdout from the claude process directly to SSE chunks as they arrive.
// Each data chunk becomes a proper SSE event with delta content in real time.
// TODO(cache-singleflight-stream): streaming-path singleflight is out of scope for v3.13.0; see spec D4 streaming caveat.
function callClaudeStreaming(model, messages, conversationId, res, authInfo = {}) {
const id = `chatcmpl-${randomUUID()}`;
const created = Math.floor(Date.now() / 1000);
@@ -1265,14 +1266,45 @@ async function handleChatCompletions(req, res) {
const t0Usage = Date.now();
const promptChars = messages.reduce((a, m) => a + (typeof m.content === "string" ? m.content.length : JSON.stringify(m.content).length), 0);
// Non-streaming path with stampede protection: wrap the upstream call in singleflight
// when cache is enabled and a hash is present. Concurrent identical requests share
// one upstream spawn; followers receive the same promise. Streaming-path dedup is
// explicitly out of scope (see TODO comment above callClaudeStreaming).
if (CACHE_TTL > 0 && req._cacheHash) {
try {
const content = await singleflight(req._cacheHash, async () => {
// Re-check cache inside the singleflight: a follower that enters before the
// leader finishes will wait on the shared promise (not reach here), but a
// request that races in just after the previous singleflight cleared the map
// will re-read the freshly-populated cache entry here rather than spawning.
const recheck = getCachedResponse(req._cacheHash, CACHE_TTL);
if (recheck) return recheck.response;
const c = await callClaude(model, messages, conversationId);
try { setCachedResponse(req._cacheHash, model, c); } catch (e) { logEvent("error", "cache_write_failed", { error: e.message }); }
return c;
});
const id = `chatcmpl-${randomUUID()}`;
completionResponse(res, id, model, content);
try { recordUsage({ keyId: req._authKeyId, keyName: req._authKeyName, model, promptChars, responseChars: content.length, elapsedMs: Date.now() - t0Usage, success: true }); } catch (e) { logEvent("error", "usage_record_failed", { error: e.message }); }
return;
} catch (err) {
try { recordUsage({ keyId: req._authKeyId, keyName: req._authKeyName, model, promptChars, responseChars: 0, elapsedMs: Date.now() - t0Usage, success: false }); } catch (e) { logEvent("error", "usage_record_failed", { error: e.message }); }
console.error(`[proxy] error: ${err.message}`);
if (res.headersSent || res.writableEnded || res.destroyed) {
try { res.end(); } catch {}
return;
}
const safeMessage = (err.message || "Internal error").replace(/\/[\w/.\-]+/g, "[path]");
return jsonResponse(res, 500, { error: { message: safeMessage, type: "proxy_error" } });
}
}
// Fallback: cache disabled (CACHE_TTL=0) or no _cacheHash — original path untouched.
try {
const content = await callClaude(model, messages, conversationId);
const id = `chatcmpl-${randomUUID()}`;
completionResponse(res, id, model, content);
// Write to cache
if (CACHE_TTL > 0 && req._cacheHash) {
try { setCachedResponse(req._cacheHash, model, content); } catch (e) { logEvent("error", "cache_write_failed", { error: e.message }); }
}
try { recordUsage({ keyId: req._authKeyId, keyName: req._authKeyName, model, promptChars, responseChars: content.length, elapsedMs: Date.now() - t0Usage, success: true }); } catch (e) { logEvent("error", "usage_record_failed", { error: e.message }); }
} catch (err) {
try { recordUsage({ keyId: req._authKeyId, keyName: req._authKeyName, model, promptChars, responseChars: 0, elapsedMs: Date.now() - t0Usage, success: false }); } catch (e) { logEvent("error", "usage_record_failed", { error: e.message }); }
@@ -1551,10 +1583,10 @@ const server = createServer(async (req, res) => {
});
}
// GET /cache/stats — cache statistics
// GET /cache/stats — cache statistics (entries, hits, size, inflight singleflight count)
if (pathname === "/cache/stats" && req.method === "GET") {
if (!isAdmin) return jsonResponse(res, 403, { error: "Admin access required" });
return jsonResponse(res, 200, getCacheStats());
return jsonResponse(res, 200, { ...getCacheStats(), ...getInflightStats() });
}
// DELETE /cache — clear cache
+101 -1
View File
@@ -3,7 +3,7 @@
* Integration test for Quota + Cache features.
* Tests database layer functions directly — no server needed.
*/
import { getDb, createKey, listKeys, validateKey, recordUsage, checkQuota, updateKeyQuota, getKeyQuota, findKey, cacheHash, getCachedResponse, setCachedResponse, clearCache, getCacheStats, closeDb, hasCacheControl } from "./keys.mjs";
import { getDb, createKey, listKeys, validateKey, recordUsage, checkQuota, updateKeyQuota, getKeyQuota, findKey, cacheHash, getCachedResponse, setCachedResponse, clearCache, getCacheStats, closeDb, hasCacheControl, singleflight, getInflightStats } from "./keys.mjs";
import { createHash } from "node:crypto";
import { strict as assert } from "node:assert";
import { unlinkSync } from "node:fs";
@@ -354,6 +354,106 @@ test("D3: chunked replay uses Array.from — multibyte codepoints stay intact",
}
});
// ── PR-B Singleflight tests (async) ──
async function asyncTest(name, fn) {
try {
await fn();
passed++;
console.log(`${name}`);
} catch (e) {
failed++;
console.log(`${name}: ${e.message}`);
}
}
async function runSingleflightTests() {
console.log("\nPR-B Singleflight:");
// 1. Basic dedup: 10 concurrent calls with same hash execute fn only once.
await asyncTest("basic dedup: 10 concurrent callers execute fn only once", async () => {
let callCount = 0;
const fn = () => new Promise(resolve => {
callCount++;
setTimeout(() => resolve("result-A"), 20);
});
const results = await Promise.all(Array.from({ length: 10 }, () => singleflight("sf-dedup-1", fn)));
assert.equal(callCount, 1, `fn called ${callCount} times, expected 1`);
assert.ok(results.every(r => r === "result-A"), "all 10 callers should receive the same return value");
});
// 2. Failure fan-out: all followers reject when leader rejects.
await asyncTest("failure fan-out: all followers reject with leader error", async () => {
let callCount = 0;
const fn = () => new Promise((_, reject) => {
callCount++;
setTimeout(() => reject(new Error("upstream-fail")), 20);
});
const promises = Array.from({ length: 10 }, () => singleflight("sf-fail-1", fn));
const results = await Promise.allSettled(promises);
assert.equal(callCount, 1, `fn called ${callCount} times, expected 1`);
assert.ok(results.every(r => r.status === "rejected"), "all 10 should be rejected");
assert.ok(results.every(r => r.reason?.message === "upstream-fail"), "all should share the same error message");
});
// 3a. Map cleanup after success: inflight count returns to 0 after promise resolves.
await asyncTest("map cleanup after success: inflight=0 after promise settles", async () => {
const fn = () => new Promise(resolve => setTimeout(() => resolve("done"), 10));
await singleflight("sf-cleanup-ok", fn);
const stats = getInflightStats();
assert.equal(stats.inflight, 0, `expected inflight=0 after settlement, got ${stats.inflight}`);
});
// 3b. Map cleanup after failure: inflight count returns to 0 after promise rejects.
await asyncTest("map cleanup after failure: inflight=0 after promise rejects", async () => {
const fn = () => new Promise((_, reject) => setTimeout(() => reject(new Error("fail")), 10));
try { await singleflight("sf-cleanup-fail", fn); } catch {}
const stats = getInflightStats();
assert.equal(stats.inflight, 0, `expected inflight=0 after rejection, got ${stats.inflight}`);
});
// 4. Different hashes don't share: two parallel calls with distinct hashes both execute.
await asyncTest("different hashes do not share a singleflight entry", async () => {
let countA = 0;
let countB = 0;
const fnA = () => new Promise(resolve => { countA++; setTimeout(() => resolve("A"), 20); });
const fnB = () => new Promise(resolve => { countB++; setTimeout(() => resolve("B"), 20); });
const [rA, rB] = await Promise.all([singleflight("sf-hash-A", fnA), singleflight("sf-hash-B", fnB)]);
assert.equal(countA, 1);
assert.equal(countB, 1);
assert.equal(rA, "A");
assert.equal(rB, "B");
});
// 5. getInflightStats shape: returns { inflight: number, requesters: number }.
await asyncTest("getInflightStats returns correct shape", async () => {
// Verify shape against a settled state (inflight=0 is still the right shape).
const stats = getInflightStats();
assert.equal(typeof stats.inflight, "number", "inflight should be a number");
assert.equal(typeof stats.requesters, "number", "requesters should be a number");
// Also verify live counts: start a pending fn, check inflight>0, then resolve.
const { promise: blocker, resolve: resolveBlocker } = Promise.withResolvers();
const fn = () => blocker;
const p = singleflight("sf-stats-shape", fn);
const liveStats = getInflightStats();
assert.ok(liveStats.inflight >= 1, `expected inflight>=1, got ${liveStats.inflight}`);
resolveBlocker("ok");
await p;
});
// 6. Sequential calls don't share: singleflight is for concurrent dedup only.
await asyncTest("sequential calls with same hash each execute fn independently", async () => {
let callCount = 0;
const fn = () => new Promise(resolve => { callCount++; setTimeout(() => resolve(callCount), 10); });
const r1 = await singleflight("sf-sequential", fn);
const r2 = await singleflight("sf-sequential", fn);
assert.equal(callCount, 2, `fn should have been called twice, got ${callCount}`);
assert.equal(r1, 1);
assert.equal(r2, 2);
});
}
await runSingleflightTests();
// ── Cleanup ──
closeDb();