taodeng/ocp
1
0
Fork 0
mirror of https://github.com/dtzp555-max/ocp.git synced 2026-07-22 05:25:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: singleflight follower retry on leader disconnect + quiet disconnect handling (review M1/L1/L2)

Browse Source 
Addresses the independent reviewer's APPROVE-WITH-CHANGES findings on PR #149:

M1 (MEDIUM, F2 regression) — when a singleflight LEADER disconnected while queued,
its RequestDisconnectedError rejected the SHARED promise, so live followers fell
into respondUpstreamError's generic branch and got a spurious 500 on a healthy
socket. Fix: keys.mjs singleflight() gains an optional follower-side `retryIf`
predicate. When a follower joins an existing flight and the shared promise rejects
with an error retryIf() accepts, the follower does NOT inherit the rejection — it
re-enters singleflight with its OWN fn (the map entry is guaranteed already deleted:
the delete-finally is attached upstream of the promise followers await), becoming
the new leader or joining a retrying sibling's fresh flight. The leader's own
rejection is never retried (it IS that client's disconnect). server.mjs passes
retryIf = (err) => err instanceof RequestDisconnectedError && !res.destroyed, so a
follower whose own client is also gone still propagates quietly. Callers without
retryIf keep byte-for-byte pre-existing share-everything semantics (pinned by the
existing failure-fan-out test).

L1 (LOW) — a disconnect-while-queued on the non-streaming paths was recorded as a
usage FAILURE row and logged as a [proxy] error: metric noise for a non-error.
Both non-streaming catch blocks now early-return on RequestDisconnectedError
without recordUsage(success:false) and without console.error — mirroring the
streaming path, which returns silently. The disconnect remains observable at info
level (concurrency_wait_cancelled, now also emitted with path:"tui" from
callClaudeTui for parity with acquireClaudeSlot's -p log).

L2 (LOW, test gap) — added a unit test for the abort-after-grant race: a waiter
granted its slot whose signal aborts afterward must see no rejection, no queue
corruption, and its slot released exactly once via the normal path (the semaphore
detaches the abort listener at grant; the onAbort idx===-1 guard is the in-dispatch
backstop).

Tests: +3 (2× M1 in the singleflight section, 1× L2 in the F2 section) — suite is
now 247/247 green.

ALIGNMENT.md: unchanged declaration — still local concurrency/dedup infrastructure
with no cli.js wire analogue; no endpoint, header, request field, or response field
added or changed; /health shape untouched. cli.js citation declared ABSENT per
CLAUDE.md hard-requirement #1 (not a Class A mirror change, not a Class B
contract change).

Verification: node --check server.mjs lib/tui/semaphore.mjs keys.mjs && npm test
— 247 passed, 0 failed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
taodeng
2026-07-07 22:59:02 +10:00
co-authored by Claude Fable 5
parent 26b09af5a8
commit 383dea0c82
3 changed files with 123 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server.mjs
+25 -4
View File
@@ -1297,9 +1297,15 @@ async function callClaudeTui(model, messages, _conversationId, _keyName, res) {
await tuiSemaphore.acquire(signal);
} catch (err) {
detach();
throw err instanceof SemaphoreAbortError
? new RequestDisconnectedError("client disconnected while waiting for a TUI concurrency slot")
: err;
if (err instanceof SemaphoreAbortError) {
// L1: client-driven cancellation, not an upstream failure — info, not error (mirrors
// acquireClaudeSlot's concurrency_wait_cancelled on the -p path).
logEvent("info", "concurrency_wait_cancelled", {
reason: "client_disconnected", path: "tui", inflight: tuiSemaphore.inflight, queued: tuiSemaphore.queued,
});
throw new RequestDisconnectedError("client disconnected while waiting for a TUI concurrency slot");
}
throw err;
}
detach();
// release() runs in a finally so any throw from runTuiTurn (tmux spawn failure,
@@ -2302,12 +2308,24 @@ async function handleChatCompletions(req, res) {
const c = await upstreamCall(model, messages, conversationId, req._authKeyName, res);
try { setCachedResponse(req._cacheHash, model, c); } catch (e) { logEvent("error", "cache_write_failed", { error: e.message }); }
return c;
});
},
// M1: if the LEADER disconnected while queued (F2), its RequestDisconnectedError is
// personal to the leader — a live follower must not inherit it as a spurious 500.
// retryIf makes this follower re-enter singleflight with its OWN fn (own res, own
// disconnect signal), becoming the new leader or joining a retrying sibling's flight —
// but only while OUR client is still connected. If our client is also gone, the
// rejection propagates and the RDE early-return in the catch below ends it quietly.
(err) => err instanceof RequestDisconnectedError && !res.destroyed);
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) {
// L1: a client disconnect while queued is NOT an upstream failure — mirror the
// streaming path (which returns without recording anything): no usage-failure row,
// no [proxy] error log, no error response (the socket is gone). The disconnect is
// already logged at info level (concurrency_wait_cancelled) by acquireClaudeSlot.
if (err instanceof RequestDisconnectedError) { try { res.end(); } catch {} return; }
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) {
@@ -2325,6 +2343,9 @@ async function handleChatCompletions(req, res) {
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 }); }
} catch (err) {
// L1: disconnect-while-queued — same quiet non-error outcome as the singleflight
// path above and the streaming path (see acquireClaudeSlot's info-level log).
if (err instanceof RequestDisconnectedError) { try { res.end(); } catch {} return; }
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) {