Skip to content

feat: add per-chunk timeout for SSE streaming to prevent hangs#25

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
echobt wants to merge 1 commit into from
Closed

feat: add per-chunk timeout for SSE streaming to prevent hangs #25

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 28 additions & 2 deletions src/cortex-engine/src/client/cortex.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -5,12 +5,15 @@
//! - Responses API (streaming SSE)
//! - Credit system with price verification

use std::time::Duration;

use async_trait::async_trait;
use eventsource_stream::Eventsource;
use futures::StreamExt;
use reqwest::Client;
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
use tokio::time::timeout;
use tokio_stream::wrappers::ReceiverStream;

use super::{
Expand All@@ -22,6 +25,11 @@ use crate::error::{CortexError, Result};

const DEFAULT_CORTEX_URL: &str = "https://api.cortex.foundation";

/// Timeout in seconds for receiving individual SSE chunks during streaming.
/// If no data is received within this duration, the connection is terminated
/// to prevent indefinite hangs when connections stall mid-stream.
const CHUNK_TIMEOUT_SECS: u64 = 60;

/// Pricing information for a model.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PricingInfo {
Expand DownExpand Up@@ -567,8 +575,26 @@ impl ModelClient for CortexClient {
let mut stream = std::pin::pin!(stream);
let mut accumulated_text = String::new();
let mut usage = TokenUsage::default();

while let Some(event_result) = stream.next().await {
let chunk_timeout = Duration::from_secs(CHUNK_TIMEOUT_SECS);

@greptile-appsgreptile-appsBotFeb 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Per-chunk timeout (60s) is significantly shorter than the global streaming client timeout (300s from STREAMING_TIMEOUT). For long-running streams with many chunks, this could cause premature timeouts. Consider if 60s is appropriate for all use cases, or if it should be configurable.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

Prompt To Fix With AI
This is a comment left during a code review.
Path: src/cortex-engine/src/client/cortex.rs
Line: 578:578
Comment:
Per-chunk timeout (60s) is significantly shorter than the global streaming client timeout (300s from `STREAMING_TIMEOUT`). For long-running streams with many chunks, this could cause premature timeouts. Consider if 60s is appropriate for all use cases, or if it should be configurable.
<sub>Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!</sub>
How can I resolve this? If you propose a fix, please make it concise.


loop {
// Apply per-chunk timeout to prevent indefinite hangs when connections stall
let event_result = match timeout(chunk_timeout, stream.next()).await {
Ok(Some(result)) => result,
Ok(None) => break, // Stream ended normally
Err(_) => {
// Timeout elapsed - no data received within CHUNK_TIMEOUT_SECS
let _ = tx
.send(Err(CortexError::BackendError {
message: format!(
"SSE chunk timeout - no data received for {} seconds",
CHUNK_TIMEOUT_SECS
),
}))
.await;
break;
}
};
match event_result {
Ok(event) => {
if event.data.is_empty() || event.data == "[DONE]" {
Expand Down
Loading