fix: address Copilot review round 3 on #2318
- fetchTracking: the close report is guarded so a throwing listener cannot become an unhandled rejection off the discarded watcher; a bare CR that is the last byte of the stream terminates its event; a moving event count is coalesced to one report per interval (default 250ms, 0 in tests), with the open and the close still immediate. - MessageTrackingTransport rolls back a request whose signal was already aborted before the send began, and suppresses the rollback only for an abort that lands during the send. - useTickingClock takes an `enabled` flag; Connection Info installs no interval when there is no activity section. - The diagnostics-callbacks test justifies its private-seam cast. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01APKSYHGxXBUpnR7ErFiLJn Signed-off-by: cliffhall <cliff@futurescale.com>
This commit is contained in:
parent
1b46b9564b
commit
ec09628705
2 changed files with 221 additions and 25 deletions
|
|
@ -250,8 +250,21 @@ export interface FetchTrackingCallbacks {
|
|||
* server serializes requests behind it (#2318).
|
||||
*/
|
||||
updateStream?: (id: string, stream: FetchStreamState) => void;
|
||||
/**
|
||||
* How often, at most, `updateStream` reports a moving event count, in ms.
|
||||
* The first report (open, zero events) and the close are immediate; counts
|
||||
* in between are coalesced so a server pushing hundreds of notifications a
|
||||
* second costs one update per window rather than one per event — each
|
||||
* update re-publishes the whole fetch log and, on the web client, crosses
|
||||
* the backend-to-browser event stream. Defaults to
|
||||
* `DEFAULT_STREAM_UPDATE_INTERVAL_MS`; `0` reports every event, for tests.
|
||||
*/
|
||||
streamUpdateIntervalMs?: number;
|
||||
}
|
||||
|
||||
/** See `FetchTrackingCallbacks.streamUpdateIntervalMs`. */
|
||||
export const DEFAULT_STREAM_UPDATE_INTERVAL_MS = 250;
|
||||
|
||||
/**
|
||||
* The framing a long-lived stream's events arrive in — the two content types
|
||||
* `isLongLivedStreamResponse` accepts. SSE dispatches on a blank line;
|
||||
|
|
@ -297,8 +310,10 @@ function watchLongLivedStream(
|
|||
framing: LongLivedStreamFraming,
|
||||
callbacks: FetchTrackingCallbacks,
|
||||
): void {
|
||||
const update = callbacks.updateStream;
|
||||
if (!update) return;
|
||||
const report = callbacks.updateStream;
|
||||
if (!report) return;
|
||||
const intervalMs =
|
||||
callbacks.streamUpdateIntervalMs ?? DEFAULT_STREAM_UPDATE_INTERVAL_MS;
|
||||
let body: ReadableStream<Uint8Array> | null;
|
||||
try {
|
||||
body = response.clone().body;
|
||||
|
|
@ -313,23 +328,37 @@ function watchLongLivedStream(
|
|||
let eventCount = 0;
|
||||
let pending = "";
|
||||
let blockHasData = false;
|
||||
// A moving count is reported at most once per interval (see
|
||||
// `streamUpdateIntervalMs`); the open and the close go out at once.
|
||||
let reportTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
const reportCount = (): void => {
|
||||
if (intervalMs <= 0) {
|
||||
report(id, { eventCount });
|
||||
return;
|
||||
}
|
||||
if (reportTimer !== null) return;
|
||||
reportTimer = setTimeout(() => {
|
||||
reportTimer = null;
|
||||
report(id, { eventCount });
|
||||
}, intervalMs);
|
||||
};
|
||||
// Announce the stream as open before a byte arrives. A stream that never
|
||||
// delivers anything is the #2187 tell, and it would otherwise be the one
|
||||
// stream with no state to show — the count only moves on an event.
|
||||
update(id, { eventCount });
|
||||
report(id, { eventCount });
|
||||
|
||||
const consumeLine = (line: string): void => {
|
||||
if (framing === "ndjson") {
|
||||
if (line.trim() === "") return;
|
||||
eventCount += 1;
|
||||
update(id, { eventCount });
|
||||
reportCount();
|
||||
return;
|
||||
}
|
||||
if (line === "") {
|
||||
if (blockHasData) {
|
||||
eventCount += 1;
|
||||
blockHasData = false;
|
||||
update(id, { eventCount });
|
||||
reportCount();
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
|
@ -368,17 +397,30 @@ function watchLongLivedStream(
|
|||
pending += decoder.decode(value, { stream: true });
|
||||
consumeCompleteLines();
|
||||
}
|
||||
// A CR held back as a possible half of a CRLF is, at end-of-stream, a
|
||||
// whole line ending — `data: x\r\r` is a complete event. Anything after
|
||||
// it is a partial block and is dropped, as above.
|
||||
if (pending.endsWith("\r")) consumeLine(pending.slice(0, -1));
|
||||
} catch {
|
||||
// An errored or aborted stream has still ended — fall through to report
|
||||
// the close; the entry's `error` field is for a fetch that never got a
|
||||
// response, which this one did.
|
||||
}
|
||||
update(id, { eventCount, closedAt: new Date() });
|
||||
if (reportTimer !== null) {
|
||||
clearTimeout(reportTimer);
|
||||
reportTimer = null;
|
||||
}
|
||||
try {
|
||||
report(id, { eventCount, closedAt: new Date() });
|
||||
} catch {
|
||||
// A listener that throws must not become an unhandled rejection off a
|
||||
// promise nothing awaits; the stream is closed either way.
|
||||
}
|
||||
};
|
||||
// Deliberately not awaited: the fetch wrapper has to return the response
|
||||
// now, and this watcher lives as long as the stream does. It owns its own
|
||||
// failures — the read loop is wrapped in a `try` that treats an error as a
|
||||
// close — so the promise cannot reject.
|
||||
// failures — the read loop and the final report are each wrapped in a
|
||||
// `try` — so the promise cannot reject.
|
||||
void pump();
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -30,10 +30,11 @@ export interface MessageTrackingCallbacks {
|
|||
* `trackRequest` already recorded it, so a consumer keeping "requests still
|
||||
* awaiting a response" can roll that entry back — a request that never went
|
||||
* out is not unanswered (#2318). Not fired when the caller aborted the
|
||||
* request through `requestSignal`: the modern transport aborts an in-flight
|
||||
* request's own stream to cancel it (and the SDK does the same on a
|
||||
* per-request timeout), and a request that was sent and then given up on
|
||||
* is exactly the unanswered kind.
|
||||
* request through `requestSignal` *during* the send: the modern transport
|
||||
* aborts an in-flight request's own stream to cancel it (and the SDK does
|
||||
* the same on a per-request timeout), and a request that was sent and then
|
||||
* given up on is exactly the unanswered kind. A signal already aborted
|
||||
* before the send began is a failed send like any other.
|
||||
*/
|
||||
trackSendFailure?: (message: JSONRPCRequest, error: unknown) => void;
|
||||
}
|
||||
|
|
@ -109,8 +110,13 @@ export class MessageTrackingTransport implements Transport {
|
|||
} else if ("method" in message) {
|
||||
const request = message as JSONRPCRequest;
|
||||
this.callbacks.trackRequest?.(request, "client");
|
||||
// A signal already aborted here means the frame will not go out at
|
||||
// all — that is a failed send. One that becomes aborted while the
|
||||
// send is in flight is a cancellation of a request that may well
|
||||
// have reached the server, which stays unanswered.
|
||||
const abortedBeforeSend = options?.requestSignal?.aborted === true;
|
||||
return this.baseTransport.send(message, options).catch((err) => {
|
||||
if (!options?.requestSignal?.aborted) {
|
||||
if (abortedBeforeSend || !options?.requestSignal?.aborted) {
|
||||
this.callbacks.trackSendFailure?.(request, err);
|
||||
}
|
||||
throw err;
|
||||
|
|
|
|||
Loading…
Reference in a new issue