feat(zebrad): run a command on chain tip changes (block notify)

CHANGELOG.md

@@ -27,6 +27,9 @@ and this project adheres to [Semantic Versioning](https://semver.org).
   `ReadResponse::ForkPoint`) that returns the most recent block in a caller-supplied
   locator that is on the best chain — the fork point — for clients tracking chain
   reorganizations through a read-only state service.
+- Added a `[notify] block_notify_command` option that runs a command on each best-chain-tip
+  change, with `%s` replaced by the new block hash — Zebra's equivalent of `zcashd`'s
+  `-blocknotify`.
 
 ### Changed
 

zebrad/Cargo.toml

@@ -186,7 +186,7 @@ toml = { workspace = true }
 
 futures = { workspace = true }
 rayon = { workspace = true }
-tokio = { workspace = true, features = ["time", "rt-multi-thread", "macros", "tracing", "signal"] }
+tokio = { workspace = true, features = ["time", "rt-multi-thread", "macros", "tracing", "signal", "process"] }
 tokio-stream = { workspace = true, features = ["time"] }
 tower = { workspace = true, features = ["hedge", "limit"] }
 pin-project = { workspace = true }

zebrad/src/commands/start.rs

@@ -42,6 +42,9 @@
 //!  * Block Gossip Task
 //!    * runs in the background and continuously queries the state for
 //!      newly committed blocks to be gossiped to peers
+//!  * Block Notify Task
+//!    * if the user has configured a `notify.block_notify_command`, runs that command
+//!      whenever the best chain tip changes (Zebra's equivalent of zcashd's `-blocknotify`)
 //!  * Progress Task
 //!    * logs progress towards the chain tip
 //!
@@ -92,6 +95,7 @@ use crate::{
         health,
         inbound::{self, InboundSetupData, MAX_INBOUND_RESPONSE_TIME},
         mempool::{self, Mempool},
+        notify::{self, BlockNotifyError},
         sync::{self, show_block_chain_progress, VERIFICATION_PIPELINE_SCALING_MULTIPLIER},
         tokio::{RuntimeRun, TokioComponent},
         ChainSync, Inbound,
@@ -344,6 +348,21 @@ impl StartCmd {
             .in_current_span(),
         );
 
+        info!("spawning block notify task");
+        let block_notify_task_handle: tokio::task::JoinHandle<Result<(), BlockNotifyError>> =
+            if let Some(command) = config.notify.block_notify_command.clone() {
+                tokio::spawn(
+                    notify::run_block_notify(
+                        command,
+                        sync_status.clone(),
+                        chain_tip_change.clone(),
+                    )
+                    .in_current_span(),
+                )
+            } else {
+                tokio::spawn(std::future::pending().in_current_span())
+            };
+
         info!("spawning mempool queue checker task");
         let mempool_queue_checker_task_handle = mempool::QueueChecker::spawn(mempool.clone());
 
@@ -458,6 +477,7 @@ impl StartCmd {
         pin!(indexer_rpc_task_handle);
         pin!(syncer_task_handle);
         pin!(block_gossip_task_handle);
+        pin!(block_notify_task_handle);
         pin!(mempool_crawler_task_handle);
         pin!(mempool_queue_checker_task_handle);
         pin!(tx_gossip_task_handle);
@@ -511,6 +531,11 @@ impl StartCmd {
                     .map(|_| info!("chain tip block gossip task exited"))
                     .map_err(|e| eyre!(e)),
 
+                block_notify_result = &mut block_notify_task_handle => block_notify_result
+                    .expect("unexpected panic in the block notify task")
+                    .map(|_| info!("block notify task exited"))
+                    .map_err(|e| eyre!(e)),
+
                 mempool_crawl_result = &mut mempool_crawler_task_handle => mempool_crawl_result
                     .expect("unexpected panic in the mempool crawler")
                     .map(|_| info!("mempool crawler task exited"))
@@ -582,6 +607,7 @@ impl StartCmd {
         health_task_handle.abort();
         syncer_task_handle.abort();
         block_gossip_task_handle.abort();
+        block_notify_task_handle.abort();
         mempool_crawler_task_handle.abort();
         mempool_queue_checker_task_handle.abort();
         tx_gossip_task_handle.abort();

zebrad/src/components.rs

@@ -10,6 +10,7 @@ pub mod inbound;
 #[allow(missing_docs)]
 pub mod mempool;
 pub mod metrics;
+pub mod notify;
 #[allow(missing_docs)]
 pub mod sync;
 #[allow(missing_docs)]

zebrad/src/components/notify.rs

@@ -0,0 +1,174 @@
+//! A task that runs an external command whenever the best chain tip changes.
+//!
+//! This is Zebra's port of zcashd's `-blocknotify`. Whenever the node's best chain tip changes,
+//! and the node is close to the network tip (Zebra's analogue of "not in initial block download"),
+//! the configured command is run via the system shell with every `%s` replaced by the new tip's
+//! block hash. The command is run detached and never blocks block validation.
+
+use std::process::Stdio;
+
+use thiserror::Error;
+use tokio::sync::watch;
+
+use zebra_chain::block;
+use zebra_state::ChainTipChange;
+
+use crate::components::sync::SyncStatus;
+
+#[cfg(test)]
+mod tests;
+
+/// Block notify configuration section.
+///
+/// Mirrors zcashd's `-blocknotify` option.
+#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct Config {
+    /// Command run whenever the best chain tip changes, mirroring zcashd's `-blocknotify`.
+    ///
+    /// Every `%s` in the command is replaced with the new tip's block hash in `getbestblockhash`
+    /// hex format. The command is run via the system shell (`/bin/sh -c` on Unix, `cmd /C` on
+    /// Windows), detached, and never blocks block validation. It is not run during initial block
+    /// download (gated on [`SyncStatus::wait_until_close_to_tip`]).
+    ///
+    /// Best-effort per tip change: during fast sync or reorgs, intermediate tip hashes may be
+    /// coalesced and skipped — only the current best tip hash fires. Disabled when `None`.
+    pub block_notify_command: Option<String>,
+}
+
+// we like our default configs to be explicit
+#[allow(unknown_lints)]
+#[allow(clippy::derivable_impls)]
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            block_notify_command: None,
+        }
+    }
+}
+
+/// Errors that can occur in the block notify task.
+#[derive(Error, Debug)]
+pub enum BlockNotifyError {
+    /// The chain tip sender was dropped, so we can't observe further tip changes.
+    #[error("chain tip sender was dropped")]
+    TipChange(watch::error::RecvError),
+
+    /// The sync status sender was dropped, so we can't tell when we're close to the tip.
+    #[error("sync status sender was dropped")]
+    SyncStatus(watch::error::RecvError),
+}
+
+/// Run continuously, executing `command` whenever the best chain tip changes.
+///
+/// Mirrors zcashd's `-blocknotify`: each `%s` in `command` is replaced with the new tip's block
+/// hash in `getbestblockhash` hex format, and the command is run detached via the system shell.
+///
+/// The command is only run once the node is close to the network tip, which is Zebra's analogue
+/// of zcashd suppressing the callback during initial block download. If a lot of blocks are
+/// committed at once, intermediate tip hashes are coalesced into a single [`Reset`] by
+/// [`ChainTipChange`], so only the current best tip hash fires.
+///
+/// Returns an error if communication with the state or the syncer is lost.
+///
+/// [`Reset`]: zebra_state::TipAction::Reset
+pub async fn run_block_notify(
+    command: String,
+    mut sync_status: SyncStatus,
+    mut chain_tip_change: ChainTipChange,
+) -> Result<(), BlockNotifyError> {
+    info!("initializing block notify task");
+
+    loop {
+        // Block until the tip changes. This always waits for a real change, so it paces the loop.
+        // The gate below can't pace the loop: it returns immediately when synced. It doesn't skip
+        // the initial sync (the tip changes throughout it); that's the gate's job.
+        let tip_action = chain_tip_change
+            .wait_for_tip_change()
+            .await
+            .map_err(BlockNotifyError::TipChange)?;
+
+        // Gate: hold until we're close to the tip (done catching up), suppressing notifications
+        // during sync. Must sit right before the spawn so close-to-tip still holds when we fire;
+        // gating earlier could fire on an intermediate block if we fall behind again first.
+        sync_status
+            .wait_until_close_to_tip()
+            .await
+            .map_err(BlockNotifyError::SyncStatus)?;
+
+        // Grab the freshest tip: catching up can take a while, during which newer blocks may have
+        // landed and made `tip_action` stale. This peek doesn't block; fall back to `tip_action`
+        // when nothing newer is pending.
+        let (hash, height) = chain_tip_change
+            .last_tip_change()
+            .unwrap_or(tip_action)
+            .best_tip_hash_and_height();
+
+        spawn_notify_command(&command, hash, height);
+    }
+}
+
+/// Renders `command` for `hash` and spawns it detached via the system shell.
+///
+/// The command never blocks the caller: it is spawned and reaped in a separate task, so a hung
+/// command cannot stall block validation.
+fn spawn_notify_command(command: &str, hash: block::Hash, height: block::Height) {
+    let rendered = render_command(command, hash);
+
+    // Run the command via the system shell, like zcashd's `system()`: `/bin/sh -c` on Unix, and
+    // `cmd /C` on Windows.
+    #[cfg(not(target_os = "windows"))]
+    let mut cmd = {
+        let mut cmd = tokio::process::Command::new("/bin/sh");
+        cmd.arg("-c").arg(&rendered);
+        cmd
+    };
+    #[cfg(target_os = "windows")]
+    let mut cmd = {
+        let mut cmd = tokio::process::Command::new("cmd");
+        cmd.arg("/C").arg(&rendered);
+        cmd
+    };
+
+    cmd.stdin(Stdio::null())
+        .stdout(Stdio::null())
+        .stderr(Stdio::null());
+
+    // Put the command in its own process group on Unix, so signals sent to zebrad's process group
+    // (such as Ctrl-C) don't also hit the notify command.
+    #[cfg(unix)]
+    cmd.process_group(0);
+
+    let span = info_span!("block_notify_command", %hash, ?height);
+
+    match cmd.spawn() {
+        Ok(mut child) => {
+            // Reap the child asynchronously to avoid zombies, and log non-zero exits like zcashd's
+            // `runCommand`. The main loop never awaits this, so it returns immediately to await the
+            // next tip change. stdout/stderr go to `/dev/null`, so we only need the exit status.
+            tokio::spawn(async move {
+                let _enter = span.enter();
+
+                match child.wait().await {
+                    Ok(status) if !status.success() => {
+                        warn!(?rendered, ?status, "block notify command exited non-zero");
+                    }
+                    Ok(_) => {}
+                    Err(error) => {
+                        warn!(?rendered, ?error, "failed to wait on block notify command");
+                    }
+                }
+            });
+        }
+        Err(error) => warn!(?rendered, ?error, "failed to spawn block notify command"),
+    }
+}
+
+/// Replaces every `%s` in `command` with `hash` in `getbestblockhash` hex format.
+///
+/// [`block::Hash`]'s [`Display`](std::fmt::Display) impl already yields the `getbestblockhash`
+/// format (display-order hex), so the substituted value is a fixed 64-char `[0-9a-f]` string with
+/// no shell metacharacters.
+fn render_command(command: &str, hash: block::Hash) -> String {
+    command.replace("%s", &hash.to_string())
+}