Skip to content

fix(api): finalize bifrost streaming usage on close - #4616

Closed
KooshaPari wants to merge 1 commit into
cheaper-inference:release/v3.8.32from
KooshaPari:fix/issue-4595-bifrost-stream
Closed

KooshaPari wants to merge 1 commit into
cheaper-inference:release/v3.8.32from
KooshaPari:fix/issue-4595-bifrost-stream

Conversation

@KooshaPari

Copy link
Copy Markdown
Contributor

Fixes #4595. Validation: targeted bifrost-sidecar test, typecheck:core, eslint on the route and test, prettier, and git diff --check.

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request introduces streaming support for the Bifrost relay route by wrapping response streams with a finalizer to record usage and clean up resources upon stream completion. It also adds a corresponding unit test. The reviewer provided valuable feedback, pointing out that keeping the connection timeout active during the entire stream will prematurely abort long-running generations, and recommended clearing it immediately after the initial fetch. Additionally, the reviewer suggested obtaining the stream reader lazily inside the stream's start method to prevent eager locking, and advised against logging client-side cancellations as errors to avoid skewing metrics.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

Comment on lines 304 to 308
});
} finally {
} catch (err) {
clearTimeout(tid);
throw err;
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Keeping the BIFROST_TIMEOUT_MS active during the entire streaming response means that any stream taking longer than 30 seconds (the default) will be prematurely aborted by the AbortController. In LLM applications, streaming responses can easily exceed 30 seconds for longer generations.

The timeout should only cover the initial connection and header retrieval (i.e., cleared as soon as fetch resolves), rather than the entire stream transfer. This matches the original behavior where clearTimeout(tid) was called in the finally block of the fetch call.

Suggested change
});
} finally {
} catch (err) {
clearTimeout(tid);
throw err;
}
});
clearTimeout(tid);
} catch (err) {
clearTimeout(tid);
throw err;
}

Comment on lines +208 to +221
assert.equal(clearedTimeouts.length, 0);

streamController?.enqueue(encoder.encode('data: {"delta":"hi"}\n\n'));
assert.equal(relayDb.getRelayLogs(token.id).length, 0);
assert.equal(clearedTimeouts.length, 0);

streamController?.close();
assert.equal(await res.text(), 'data: {"delta":"hi"}\n\n');

const logs = relayDb.getRelayLogs(token.id);
assert.equal(logs.length, 1);
assert.equal(logs[0]?.status, "success");
assert.equal(logs[0]?.status_code, 200);
assert.equal(clearedTimeouts.length, 1);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Since the connection timeout should be cleared immediately after the initial response headers are received (when POST resolves), we should update the test assertions to expect clearedTimeouts.length to be 1 as soon as the response is returned, rather than waiting for the stream to close.

  assert.equal(clearedTimeouts.length, 1);

  streamController?.enqueue(encoder.encode('data: {\"delta\":\"hi\"}\\n\\n'));
  assert.equal(relayDb.getRelayLogs(token.id).length, 0);
  assert.equal(clearedTimeouts.length, 1);

  streamController?.close();
  assert.equal(await res.text(), 'data: {\"delta\":\"hi\"}\\n\\n');

  const logs = relayDb.getRelayLogs(token.id);
  assert.equal(logs.length, 1);
  assert.equal(logs[0]?.status, \"success\");
  assert.equal(logs[0]?.status_code, 200);
  assert.equal(clearedTimeouts.length, 1);

Comment on lines +83 to +97
function wrapStreamWithFinalizer(
source: ReadableStream<Uint8Array>,
finalize: (status?: "success" | "error") => void
): ReadableStream<Uint8Array> {
const reader = source.getReader();
let finalized = false;

const finalizeOnce = (status?: "success" | "error") => {
if (finalized) return;
finalized = true;
finalize(status);
};

return new ReadableStream<Uint8Array>({
async pull(controller) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Calling source.getReader() immediately when wrapStreamWithFinalizer is called locks the source stream eagerly. If the returned stream is never started or read, the underlying stream remains locked. It is safer and more idiomatic to obtain the reader inside the start method of the returned ReadableStream.

function wrapStreamWithFinalizer(
  source: ReadableStream<Uint8Array>,
  finalize: (status?: \"success\" | \"error\") => void
): ReadableStream<Uint8Array> {
  let reader: ReadableStreamDefaultReader<Uint8Array>;
  let finalized = false;

  const finalizeOnce = (status?: \"success\" | \"error\") => {
    if (finalized) return;
    finalized = true;
    finalize(status);
  };

  return new ReadableStream<Uint8Array>({
    start() {
      reader = source.getReader();
    },
    async pull(controller) {

Comment on lines +112 to +118
async cancel(reason) {
try {
await reader.cancel(reason);
} finally {
finalizeOnce("error");
}
},

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Logging client-side stream cancellations as "error" might skew your error metrics, making normal user actions (like stopping generation or closing the tab) look like system or upstream failures. Consider whether these should be logged as "success" (or a specific "cancelled" status if supported) instead of "error".

Suggested change
async cancel(reason) {
try {
await reader.cancel(reason);
} finally {
finalizeOnce("error");
}
},
async cancel(reason) {
try {
await reader.cancel(reason);
} finally {
finalizeOnce();
}
},

@diegosouzapw

Copy link
Copy Markdown
Collaborator

Obrigado pela contribuição, @KooshaPari! Infelizmente este fix ficou redundante: o mesmo problema (finalizar o uso do streaming do bifrost no fechamento/cancel/erro do stream) já foi resolvido e mergeado via #4612 (finalizeReadableStream em src/app/api/v1/relay/chat/completions/bifrost/route.ts), com a mesma garantia finalize-once e tratamento adicional de timeout→504. Como o código já está na release/v3.8.34, vou fechar este PR — o crédito pela necessidade está reconhecido aqui e o comportamento que você buscava já está em produção. 🙏

@KooshaPari

Copy link
Copy Markdown
Contributor Author

Acknowledging closure as duplicate of #4612 (thanks @diegosouzapw for the credit + the design tradeoff notes). I've now opened PR #4881 (koosha/refactor/bifrost-lazy-reader-cancel-status -> release/v3.8.34) which addresses the two open items from gemini-code-assist's review on this PR that #4612 did not pick up:

@gemini-code-assist

Copy link
Copy Markdown
Contributor

Warning

You have reached your daily quota limit. Please wait up to 24 hours and I will start processing your requests again!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants