fix(firehose): add configurable idle timeout to the block stream receive loop

graph/src/blockchain/firehose_block_stream.rs

@@ -246,7 +246,7 @@ fn stream_blocks<C: Blockchain, F: FirehoseMapper<C>>(
             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<C: Blockchain, F: FirehoseMapper<C>>(
                     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<C: Blockchain, F: FirehoseMapper<C>>(
     }
 }
 
+/// 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<Duration> {
+    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<String, std::env::VarError>) -> Option<Duration> {
+    value
+        .ok()
+        .and_then(|v| v.trim().parse::<u64>().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<S, T>(
+    stream: &mut S,
+    idle: Option<Duration>,
+) -> Result<Option<T>, ()>
+where
+    S: futures03::Stream<Item = T> + 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<C: Blockchain> {
     Proceed(BlockStreamEvent<C>, String),
     Rewind(BlockPtr),

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::<i32>(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::<i32>(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::<i32>(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::<i32>(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))
+    );
+}