From 5ce06ed4df4bcf3a9ecdd8a733946aa22e8d9aae Mon Sep 17 00:00:00 2001
From: Morgan Mccauley <morgan@mccauley.co.nz>
Date: Thu, 18 Jul 2024 09:24:22 +1200
Subject: [PATCH] feat: Restrart block streams with stale data

---
 coordinator/src/handlers/block_streams.rs | 12 ++++++++++--
 1 file changed, 10 insertions(+), 2 deletions(-)

diff --git a/coordinator/src/handlers/block_streams.rs b/coordinator/src/handlers/block_streams.rs
index d5859390..93878d61 100644
--- a/coordinator/src/handlers/block_streams.rs
+++ b/coordinator/src/handlers/block_streams.rs
@@ -1,5 +1,7 @@
 #![cfg_attr(test, allow(dead_code))]
 
+use std::time::{Duration, SystemTime};
+
 pub use block_streamer::StreamInfo;
 
 use anyhow::Context;
@@ -241,10 +243,16 @@ impl BlockStreamsHandler {
         block_stream: &StreamInfo,
     ) -> anyhow::Result<()> {
         if let Some(health) = block_stream.health.as_ref() {
-            if !matches!(
+            let updated_at =
+                SystemTime::UNIX_EPOCH + Duration::from_secs(health.updated_at_timestamp_secs);
+
+            let stale = updated_at.elapsed().unwrap_or_default() > Duration::from_secs(30);
+            let stalled = matches!(
                 health.processing_state.try_into(),
                 Ok(ProcessingState::Stalled)
-            ) {
+            );
+
+            if !stale && !stalled {
                 return Ok(());
             }
         }