Add liveness and readiness HTTP endpoints

Cargo.toml

@@ -82,6 +82,8 @@ jsonrpsee-core = "0.24"
 jsonrpsee-types = "0.24"
 jsonrpsee = { version = "0.24", features = ["server", "macros"] }
 hyper = "1.6"
+hyper-util = "0.1"
+http-body-util = "0.1"
 
 # Hashmaps, channels, DBs
 indexmap = "2.2.6"

packages/zaino-state/src/backends/fetch.rs

@@ -150,28 +150,18 @@ impl ZcashService for FetchService {
             .await
             .unwrap();
 
+        // `NodeBackedChainIndex::new` has already started the sync in a
+        // background task; the service is returned immediately (status
+        // `Syncing`) so callers can expose a health endpoint during the
+        // initial scan. The wait-until-`Ready` gate now lives in the indexer
+        // launch path (`Indexer::launch_inner`), after the health server is up.
         let fetch_service = Self {
             fetcher,
             indexer,
             data,
             config,
         };
 
-        // wait for sync to complete, return error on sync fail.
-        loop {
-            match fetch_service.status() {
-                StatusType::Ready | StatusType::Closing => break,
-                StatusType::CriticalError => {
-                    return Err(FetchServiceError::Critical(
-                        "ChainIndex initial sync failed, check full log for details.".to_string(),
-                    ));
-                }
-                _ => {
-                    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
-                }
-            }
-        }
-
         Ok(fetch_service)
     }
 
@@ -227,6 +217,12 @@ impl Status for FetchServiceSubscriber {
     }
 }
 
+impl crate::chain_index::sync_progress::ChainSyncProgress for FetchServiceSubscriber {
+    fn sync_progress(&self) -> crate::chain_index::sync_progress::SyncProgress {
+        self.indexer.sync_progress()
+    }
+}
+
 impl FetchServiceSubscriber {
     /// Fetches the current status
     #[deprecated(note = "Use the Status trait method instead")]

packages/zaino-state/src/backends/state.rs

@@ -366,6 +366,12 @@ impl Status for StateServiceSubscriber {
     }
 }
 
+impl crate::chain_index::sync_progress::ChainSyncProgress for StateServiceSubscriber {
+    fn sync_progress(&self) -> crate::chain_index::sync_progress::SyncProgress {
+        self.indexer.sync_progress()
+    }
+}
+
 /// A subscriber to any chaintip updates
 #[derive(Clone)]
 pub struct ChainTipSubscriber {

packages/zaino-state/src/chain_index.rs

@@ -60,6 +60,8 @@ pub mod mempool;
 pub mod non_finalised_state;
 /// BlockchainSource
 pub mod source;
+/// Tip-proximity sync-progress snapshot for readiness checks.
+pub mod sync_progress;
 /// Common types used by the rest of this module
 pub mod types;
 
@@ -662,6 +664,9 @@ pub struct NodeBackedChainIndex<Source: BlockchainSource = ValidatorConnector> {
     finalized_db: std::sync::Arc<finalised_state::ZainoDB>,
     sync_loop_handle: Option<tokio::task::JoinHandle<Result<(), SyncError>>>,
     status: NamedAtomicStatus,
+    /// Tip-proximity progress, shared with the sync loop and handed to
+    /// subscribers for readiness checks.
+    sync_progress: sync_progress::SyncProgress,
     network: ZebraNetwork,
     source: Source,
     sync_timings: SyncTimings,
@@ -758,6 +763,7 @@ impl<Source: BlockchainSource> NodeBackedChainIndex<Source> {
             finalized_db,
             sync_loop_handle: None,
             status: NamedAtomicStatus::new("ChainIndex", StatusType::Spawning),
+            sync_progress: sync_progress::SyncProgress::new(),
             network: config.network.to_zebra_network(),
             source,
             sync_timings,
@@ -776,6 +782,7 @@ impl<Source: BlockchainSource> NodeBackedChainIndex<Source> {
             non_finalized_state: self.non_finalized_state.clone(),
             finalized_state: self.finalized_db.to_reader(),
             status: self.status.clone(),
+            sync_progress: self.sync_progress.clone(),
             network: self.network.clone(),
             source: self.source.clone(),
         }
@@ -820,6 +827,7 @@ impl<Source: BlockchainSource> NodeBackedChainIndex<Source> {
         let nfs = self.non_finalized_state.clone();
         let fs = self.finalized_db.clone();
         let status = self.status.clone();
+        let sync_progress = self.sync_progress.clone();
         let source = self.source.clone();
         let network = self.network.clone();
         let timings = self.sync_timings;
@@ -838,6 +846,7 @@ impl<Source: BlockchainSource> NodeBackedChainIndex<Source> {
             loop {
                 let source = source.clone();
                 let network = network.clone();
+                let sync_progress = sync_progress.clone();
                 if cancel_token.is_cancelled() {
                     return Ok(());
                 }
@@ -874,6 +883,10 @@ impl<Source: BlockchainSource> NodeBackedChainIndex<Source> {
                         })?;
                     #[cfg(feature = "prometheus")]
                     metrics::gauge!("zaino.chain.tip_height").set(chain_height.0 as f64);
+                    // Record the observed network tip up front so a readiness
+                    // probe sees the growing gap *during* a catch-up, not only
+                    // after it completes.
+                    sync_progress.record_network_tip(chain_height.0);
                     let finalised_height = finalized_height_floor(chain_height.0);
 
                     fs.sync_to_height(finalised_height, &source)
@@ -909,6 +922,11 @@ impl<Source: BlockchainSource> NodeBackedChainIndex<Source> {
                         .await?;
                     std::mem::drop(intermediate_nfs_for_scoping);
 
+                    // Reaching here means both finalized and non-finalized
+                    // state are synced up to `chain_height`; refresh the local
+                    // tip and tip-age clock for readiness.
+                    sync_progress.record_synced(chain_height.0);
+
                     Ok(())
                     } => r,
                 };
@@ -997,6 +1015,7 @@ pub struct NodeBackedChainIndexSubscriber<Source: BlockchainSource = ValidatorCo
     non_finalized_state: Arc<ArcSwapOption<crate::NonFinalizedState<Source>>>,
     finalized_state: finalised_state::reader::DbReader,
     status: NamedAtomicStatus,
+    sync_progress: sync_progress::SyncProgress,
     network: ZebraNetwork,
     source: Source,
 }
@@ -1092,6 +1111,14 @@ impl<Source: BlockchainSource> NodeBackedChainIndexSubscriber<Source> {
         combined_status
     }
 
+    /// Returns a clone of the shared [`SyncProgress`] handle for tip-proximity
+    /// readiness checks.
+    ///
+    /// [`SyncProgress`]: crate::chain_index::sync_progress::SyncProgress
+    pub fn sync_progress(&self) -> sync_progress::SyncProgress {
+        self.sync_progress.clone()
+    }
+
     /// Returns the number of transparent outputs of `txid` that are currently unspent in the
     /// finalised state. Returns 0 if `txid` is not indexed by the finalised state.
     ///

packages/zaino-state/src/chain_index/sync_progress.rs

@@ -0,0 +1,157 @@
+//! Shared sync-progress snapshot for tip-proximity readiness checks.
+//!
+//! The chain-index sync loop flips its [`StatusType`] to `Syncing` at the top
+//! of *every* iteration (even a steady-state re-check at the tip), so a
+//! readiness probe driven off the raw status would flap `ready -> not ready ->
+//! ready` on each cycle and churn the pod in and out of its k8s Service.
+//!
+//! [`SyncProgress`] is the stable signal instead, mirroring Zebra's health
+//! component: the sync loop records the network tip it observed and the height
+//! it has synced to, and a probe reports ready while the gap is small and the
+//! tip is fresh. Being one or two blocks behind for a moment does not flip
+//! readiness; falling far behind (initial sync, large re-org catch-up) or going
+//! stale (a silently dead source that stops advancing the tip) does.
+//!
+//! [`StatusType`]: crate::status::StatusType
+
+use std::{
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
+    time::{Duration, Instant},
+};
+
+/// A clone-safe, lock-free snapshot of the sync loop's progress toward the
+/// network tip. All clones share one underlying state.
+#[derive(Clone, Debug)]
+pub struct SyncProgress {
+    inner: Arc<Inner>,
+}
+
+#[derive(Debug)]
+struct Inner {
+    /// Most recently observed network/source tip height. Initialised to
+    /// `u64::MAX` so [`SyncProgress::blocks_behind`] reports "fully behind"
+    /// until the sync loop observes a real tip — a probe must never read
+    /// "caught up" before the first observation.
+    network_tip: AtomicU64,
+    /// Highest height the sync loop has fully synced to.
+    local_tip: AtomicU64,
+    /// Milliseconds (relative to `base`) at which `local_tip` was last
+    /// confirmed synced to the network tip. Drives the tip-age check.
+    last_synced_millis: AtomicU64,
+    /// Monotonic reference instant for `last_synced_millis`.
+    base: Instant,
+}
+
+impl SyncProgress {
+    /// Creates a tracker in the "not yet synced" state.
+    pub fn new() -> Self {
+        SyncProgress {
+            inner: Arc::new(Inner {
+                network_tip: AtomicU64::new(u64::MAX),
+                local_tip: AtomicU64::new(0),
+                last_synced_millis: AtomicU64::new(0),
+                base: Instant::now(),
+            }),
+        }
+    }
+
+    /// Records the network tip height observed at the start of a sync
+    /// iteration.
+    pub fn record_network_tip(&self, height: u32) {
+        self.inner
+            .network_tip
+            .store(height as u64, Ordering::Relaxed);
+    }
+
+    /// Records that the index has fully synced up to `height`, refreshing both
+    /// the local tip and the tip-age clock.
+    pub fn record_synced(&self, height: u32) {
+        self.inner.local_tip.store(height as u64, Ordering::Relaxed);
+        self.inner.last_synced_millis.store(
+            self.inner.base.elapsed().as_millis() as u64,
+            Ordering::Relaxed,
+        );
+    }
+
+    /// Estimated blocks behind the network tip. Saturates at 0 (a local tip at
+    /// or ahead of the last observed network tip is "caught up").
+    pub fn blocks_behind(&self) -> u64 {
+        let network = self.inner.network_tip.load(Ordering::Relaxed);
+        let local = self.inner.local_tip.load(Ordering::Relaxed);
+        network.saturating_sub(local)
+    }
+
+    /// Time since the local tip was last confirmed synced to the network tip.
+    /// Grows without bound if the sync loop stops succeeding (e.g. a dead
+    /// source), which a stale-tip readiness check can act on.
+    pub fn tip_age(&self) -> Duration {
+        self.inner
+            .base
+            .elapsed()
+            .saturating_sub(Duration::from_millis(
+                self.inner.last_synced_millis.load(Ordering::Relaxed),
+            ))
+    }
+}
+
+impl Default for SyncProgress {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Exposes a backend subscriber's [`SyncProgress`] handle.
+///
+/// Implemented by the service subscribers (both wrap a
+/// [`NodeBackedChainIndexSubscriber`]) so a generic caller — e.g. the daemon's
+/// health endpoint — can obtain the tip-proximity signal without naming the
+/// concrete backend.
+///
+/// [`NodeBackedChainIndexSubscriber`]: crate::chain_index::NodeBackedChainIndexSubscriber
+pub trait ChainSyncProgress {
+    /// Returns a clone of the shared sync-progress handle.
+    fn sync_progress(&self) -> SyncProgress;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn fresh_tracker_is_fully_behind() {
+        let p = SyncProgress::new();
+        // No network tip observed yet -> never reads as caught up.
+        assert_eq!(p.blocks_behind(), u64::MAX);
+    }
+
+    #[test]
+    fn blocks_behind_tracks_gap() {
+        let p = SyncProgress::new();
+        p.record_network_tip(1_000);
+        p.record_synced(900);
+        assert_eq!(p.blocks_behind(), 100);
+        p.record_synced(1_000);
+        assert_eq!(p.blocks_behind(), 0);
+    }
+
+    #[test]
+    fn local_tip_ahead_saturates_to_zero() {
+        let p = SyncProgress::new();
+        p.record_network_tip(500);
+        p.record_synced(500);
+        // A later observation lagging the synced height must not underflow.
+        p.record_network_tip(499);
+        assert_eq!(p.blocks_behind(), 0);
+    }
+
+    #[test]
+    fn record_synced_resets_tip_age() {
+        let p = SyncProgress::new();
+        p.record_synced(10);
+        // Freshly synced: age is ~0, comfortably under any sane threshold.
+        assert!(p.tip_age() < Duration::from_secs(1));
+    }
+}

packages/zaino-state/src/lib.rs

@@ -29,6 +29,8 @@ pub mod chain_index;
 
 // Core ChainIndex trait and implementations
 pub use chain_index::{ChainIndex, NodeBackedChainIndex, NodeBackedChainIndexSubscriber};
+// Tip-proximity sync-progress signal for readiness checks
+pub use chain_index::sync_progress::{ChainSyncProgress, SyncProgress};
 // Source types for ChainIndex backends
 pub use chain_index::source::{BlockchainSource, State, ValidatorConnector};
 // Supporting types

packages/zainod/Cargo.toml

@@ -70,6 +70,11 @@ tracing-subscriber = { workspace = true, features = ["fmt", "env-filter", "time"
 http = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 
+# HTTP health endpoint (/livez, /readyz)
+hyper = { workspace = true, features = ["server", "http1"] }
+hyper-util = { workspace = true, features = ["tokio"] }
+http-body-util = { workspace = true }
+
 # Metrics (behind "prometheus" feature)
 metrics = { workspace = true, optional = true }
 metrics-exporter-prometheus = { workspace = true, optional = true }