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 _=
This commit is contained in:
2026-07-19 15:55:49 -04:00
parent cd24eda111
commit efcd91e6b4

View File

@@ -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) => {
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
}