diff --git a/graph/src/blockchain/firehose_block_stream.rs b/graph/src/blockchain/firehose_block_stream.rs index 9ad0f2aa6e5..5385077cfcb 100644 --- a/graph/src/blockchain/firehose_block_stream.rs +++ b/graph/src/blockchain/firehose_block_stream.rs @@ -246,7 +246,7 @@ fn stream_blocks>( let result = tokio::time::timeout(Duration::from_secs(120), req).await.map_err(|x| x.into()).and_then(|x| x); match result { - Ok(stream) => { + Ok(mut stream) => { info!(&logger, "Blockstream connected"); // Track the time it takes to set up the block stream @@ -255,7 +255,33 @@ fn stream_blocks>( let mut last_response_time = Instant::now(); let mut expected_stream_end = false; - for await response in stream { + // Idle timeout for the stream. If enabled and the upstream stops sending + // frames (no message, no error, no EOF), the stream is dropped and + // re-established with backoff instead of hanging indefinitely. Disabled by + // default to preserve the current behavior. + let idle_timeout = firehose_stream_idle_timeout(); + + loop { + let next = match next_with_idle_timeout(&mut stream, idle_timeout).await { + Ok(next) => next, + Err(()) => { + error!( + logger, + "Firehose block stream idle for longer than the idle timeout, reconnecting"; + "idle_timeout_secs" => idle_timeout.map(|d| d.as_secs()).unwrap_or_default(), + ); + // Deliberate end of the stream; do not log the "completed + // unexpectedly" error below. + expected_stream_end = true; + break; + } + }; + + let response = match next { + Some(response) => response, + None => break, + }; + match process_firehose_response( &endpoint, response, @@ -338,6 +364,46 @@ fn stream_blocks>( } } +/// Returns the idle timeout to apply when waiting for the next message on a firehose +/// block stream, configured via the `GRAPH_FIREHOSE_STREAM_IDLE_TIMEOUT_SECS` environment +/// variable. The timeout is disabled (returns `None`) when the variable is missing, +/// unparseable, or set to `0` seconds. +fn firehose_stream_idle_timeout() -> Option { + idle_timeout_from_env(std::env::var("GRAPH_FIREHOSE_STREAM_IDLE_TIMEOUT_SECS")) +} + +/// Parses the value of `GRAPH_FIREHOSE_STREAM_IDLE_TIMEOUT_SECS` into an idle timeout. +/// A missing, unparseable, or zero value disables the timeout (returns `None`). +pub fn idle_timeout_from_env(value: Result) -> Option { + value + .ok() + .and_then(|v| v.trim().parse::().ok()) + .filter(|&secs| secs > 0) + .map(Duration::from_secs) +} + +/// Waits for the next item on `stream`. When `idle` is `Some(...)` the wait is bounded by +/// that duration so a stalled upstream cannot hang the receive loop forever. Returns +/// `Ok(Some(item))` when an item arrives, `Ok(None)` when the stream ends, and `Err(())` +/// when no item arrived within the idle timeout. +pub async fn next_with_idle_timeout( + stream: &mut S, + idle: Option, +) -> Result, ()> +where + S: futures03::Stream + Unpin, +{ + let next = match idle { + Some(idle) => match tokio::time::timeout(idle, stream.next()).await { + Ok(next) => next, + Err(_) => return Err(()), + }, + None => stream.next().await, + }; + + Ok(next) +} + enum BlockResponse { Proceed(BlockStreamEvent, String), Rewind(BlockPtr), diff --git a/graph/tests/firehose_idle_timeout_tests.rs b/graph/tests/firehose_idle_timeout_tests.rs new file mode 100644 index 00000000000..6e4a3602cc6 --- /dev/null +++ b/graph/tests/firehose_idle_timeout_tests.rs @@ -0,0 +1,88 @@ +// Regression tests for graphprotocol/graph-node#6689. +// +// The firehose block-stream receive loop (graph/src/blockchain/firehose_block_stream.rs) +// had no per-message idle/read timeout: if the upstream holds the HTTP/2 stream open but +// stops sending frames (no message, no error, no EOF), the subgraph would hang indexing +// indefinitely and silently until a manual restart/reassignment. +// +// The fix wraps each wait for the next stream message in a configurable idle timeout +// (GRAPH_FIREHOSE_STREAM_IDLE_TIMEOUT_SECS, disabled by default). These tests exercise the +// two new building blocks: the timeout application and the env-var parsing. + +use graph::blockchain::firehose_block_stream::{idle_timeout_from_env, next_with_idle_timeout}; +use std::env::VarError; +use std::time::Duration; + +#[test] +fn idle_timeout_from_env_parsing() { + // Missing / unparseable / zero values disable the timeout. + assert_eq!(idle_timeout_from_env(Err(VarError::NotPresent)), None); + assert_eq!( + idle_timeout_from_env(Err(VarError::NotUnicode("x".into()))), + None + ); + assert_eq!(idle_timeout_from_env(Ok("not-a-number".to_string())), None); + assert_eq!(idle_timeout_from_env(Ok("0".to_string())), None); + assert_eq!(idle_timeout_from_env(Ok(" 0 ".to_string())), None); + + // Positive values enable the timeout. + assert_eq!( + idle_timeout_from_env(Ok("30".to_string())), + Some(Duration::from_secs(30)) + ); + assert_eq!( + idle_timeout_from_env(Ok(" 5 ".to_string())), + Some(Duration::from_secs(5)) + ); +} + +#[graph::test] +async fn stream_idle_timeout_returns_item_within_deadline() { + let (tx, rx) = tokio::sync::mpsc::channel::(1); + let mut rx = tokio_stream::wrappers::ReceiverStream::new(rx); + + tx.send(1).await.unwrap(); + assert_eq!( + next_with_idle_timeout(&mut rx, Some(Duration::from_millis(100))).await, + Ok(Some(1)) + ); +} + +#[graph::test] +async fn stream_idle_timeout_breaks_stalled_stream() { + let (_tx, rx) = tokio::sync::mpsc::channel::(1); + let mut rx = tokio_stream::wrappers::ReceiverStream::new(rx); + + // No message arrives within the idle timeout: the receive loop must not hang + // indefinitely on a stalled upstream (issue #6689). + assert_eq!( + next_with_idle_timeout(&mut rx, Some(Duration::from_millis(50))).await, + Err(()) + ); +} + +#[graph::test] +async fn stream_idle_timeout_returns_none_when_stream_ends() { + let (tx, rx) = tokio::sync::mpsc::channel::(1); + let mut rx = tokio_stream::wrappers::ReceiverStream::new(rx); + + drop(tx); + assert_eq!( + next_with_idle_timeout(&mut rx, Some(Duration::from_millis(100))).await, + Ok(None) + ); +} + +#[graph::test] +async fn stream_idle_timeout_disabled_waits_for_item() { + let (tx, rx) = tokio::sync::mpsc::channel::(1); + let mut rx = tokio_stream::wrappers::ReceiverStream::new(rx); + + // With no idle timeout the helper must wait unconditionally for the next item; + // this is the default, backward-compatible behavior. + tx.send(7).await.unwrap(); + assert_eq!( + next_with_idle_timeout(&mut rx, None).await, + Ok(Some(7)) + ); +}