From efcd91e6b4a8769cff8a75d28c10f037c85fe763 Mon Sep 17 00:00:00 2001
From: svatantrya <svatantrya@bitcoin-after.life>
Date: Sun, 19 Jul 2026 15:55:49 -0400
Subject: [PATCH] fix(pusher): add ZMQ diagnostic logging and heartbeat
 monitoring

- Log ZMQ subscribe confirmation with address
- Add consecutive timeout counter with periodic warn every 60s
- Log ZMQ connection restored after timeouts
- Log main_result errors instead of discarding with let _=
---
 src/bin/bal-pusher.rs | 32 ++++++++++++++++++++++++++++----
 1 file changed, 28 insertions(+), 4 deletions(-)

diff --git a/src/bin/bal-pusher.rs b/src/bin/bal-pusher.rs
index 3973e77..237d42f 100644
--- a/src/bin/bal-pusher.rs
+++ b/src/bin/bal-pusher.rs
@@ -649,24 +649,46 @@ async fn main() -> std::io::Result<()> {
     }
 
     match socket.set_subscribe(b"") {
-        Ok(_) => {}
+        Ok(_) => {
+            info!("ZMQ subscribed to all topics on {}", zmq_address);
+        }
         Err(e) => {
             error!("ZMQ subscribe failed: {}, exiting", e);
             return Ok(());
         }
     }
 
-    let _ = main_result(&cfg, network_params).await;
+    if let Err(e) = main_result(&cfg, network_params).await {
+        error!("main_result failed on startup: {}", e);
+    }
     info!("waiting new blocks..");
     socket.set_rcvtimeo(5000).unwrap(); // 5 seconds timeout
+    let mut consecutive_timeouts: u32 = 0;
     loop {
         let message = match socket.recv_multipart(0) {
             Ok(m) => m,
             Err(e) => {
-                warn!("ZMQ recv timeout or error: {}, retrying...", e);
+                consecutive_timeouts += 1;
+                if consecutive_timeouts == 1 {
+                    warn!("ZMQ recv timeout or error: {}, retrying...", e);
+                } else if consecutive_timeouts.is_multiple_of(12) {
+                    warn!(
+                        "No ZMQ messages for {}s ({} consecutive timeouts), is bitcoind ZMQ active on {}?",
+                        consecutive_timeouts * 5,
+                        consecutive_timeouts,
+                        zmq_address
+                    );
+                }
                 continue;
             }
         };
+        if consecutive_timeouts > 0 {
+            info!(
+                "ZMQ connection restored after {} consecutive timeouts",
+                consecutive_timeouts
+            );
+        }
+        consecutive_timeouts = 0;
         let topic = message[0].clone();
         let body = message[1].clone();
         debug!(
@@ -676,7 +698,9 @@ async fn main() -> std::io::Result<()> {
         trace!("ZMQ:GET BODY: {}", hex::encode(&body));
         if topic == b"hashblock" {
             info!("NEW BLOCK: {}", hex::encode(&body));
-            let _ = main_result(&cfg, network_params).await;
+            if let Err(e) = main_result(&cfg, network_params).await {
+                error!("main_result failed on new block: {}", e);
+            }
         }
         thread::sleep(Duration::from_millis(100)); // Sleep for 100ms
     }