diff --git a/packages/zaino-state/src/chain_index.rs b/packages/zaino-state/src/chain_index.rs
index 92597fc80..d55af0409 100644
--- a/packages/zaino-state/src/chain_index.rs
+++ b/packages/zaino-state/src/chain_index.rs
@@ -512,8 +512,9 @@ pub struct NodeBackedChainIndex<Source: BlockchainSource = ValidatorConnector> {
 /// [`SyncTimings::default()`] produces production values (500 ms inter-iteration
 /// sleep, 250 ms initial backoff doubling up to 8 s, 10 consecutive failures
 /// before escalating to [`StatusType::CriticalError`] — ~40 s total window).
-/// [`SyncTimings::fast()`] shrinks each duration by 10× so backoff-dependent
-/// unit tests finish in ~4 s instead of ~40 s.
+/// [`SyncTimings::fast()`] is the test-only shrink: ~120 ms total backoff
+/// window with the same three observable behaviours (counter→threshold,
+/// exponential doubling, cap clamp) — see its doc for the chosen values.
 #[derive(Clone, Copy, Debug)]
 pub(crate) struct SyncTimings {
     pub(crate) interval: Duration,
@@ -535,13 +536,18 @@ impl Default for SyncTimings {
 
 #[cfg(test)]
 impl SyncTimings {
-    /// 10× faster than [`Self::default`] — test-only.
+    /// Test-only shrink: drives the backoff schedule to completion in
+    /// ~120 ms (4 sleeps: 10 + 20 + 40 + 50, with the 50 ms cap clamping
+    /// the would-be 80 ms doubling). Five `max_consecutive_failures` is
+    /// the smallest count that still exercises both exponential doubling
+    /// and `max_backoff` clamping — i.e. the same three observable
+    /// behaviours `default()` does, just compressed.
     pub(crate) const fn fast() -> Self {
         Self {
             interval: Duration::from_millis(50),
-            initial_backoff: Duration::from_millis(25),
-            max_backoff: Duration::from_millis(800),
-            max_consecutive_failures: 10,
+            initial_backoff: Duration::from_millis(10),
+            max_backoff: Duration::from_millis(50),
+            max_consecutive_failures: 5,
         }
     }
 
@@ -650,6 +656,7 @@ impl<Source: BlockchainSource> NodeBackedChainIndex<Source> {
         tokio::task::spawn(async move {
             let status = status.clone();
             let source = source.clone();
+            let mut change_rx = source.change_subscribe();
             let mut consecutive_failures: u32 = 0;
             let mut current_backoff = timings.initial_backoff;
 
@@ -689,7 +696,7 @@ impl<Source: BlockchainSource> NodeBackedChainIndex<Source> {
                         None => {
                             nfs.store(Some(Arc::new(
                                 NonFinalizedState::initialize(
-                                    source,
+                                    source.clone(),
                                     network,
                                     fs.to_reader()
                                         .get_chain_block_by_height(finalised_height)
@@ -716,7 +723,7 @@ impl<Source: BlockchainSource> NodeBackedChainIndex<Source> {
                         consecutive_failures = 0;
                         current_backoff = timings.initial_backoff;
                         status.store(StatusType::Ready);
-                        tokio::time::sleep(timings.interval).await;
+                        source::wait_or_source_change(change_rx.as_mut(), timings.interval).await;
                     }
                     Err(e) => {
                         consecutive_failures += 1;
@@ -776,6 +783,24 @@ impl<Source: BlockchainSource> NodeBackedChainIndexSubscriber<Source> {
         combined_status
     }
 
+    /// Returns a [`tokio::sync::watch::Receiver`] that wakes on every
+    /// transition of the chain-index sync loop's status.
+    ///
+    /// Used by tests to await a specific state (e.g. first transition to
+    /// [`StatusType::Ready`]) without busy-polling.
+    #[cfg(test)]
+    pub(crate) fn status_subscribe(&self) -> tokio::sync::watch::Receiver<StatusType> {
+        self.status.subscribe()
+    }
+
+    /// Returns a watch receiver tracking the mempool serve loop's most
+    /// recently observed chain tip. Forwarder for
+    /// [`mempool::MempoolSubscriber::mempool_tip`].
+    #[cfg(test)]
+    pub(crate) fn mempool_tip(&self) -> tokio::sync::watch::Receiver<crate::BlockHash> {
+        self.mempool.mempool_tip()
+    }
+
     async fn get_fullblock_bytes_from_node(
         &self,
         id: HashOrHeight,
@@ -1761,13 +1786,31 @@ impl<Source: BlockchainSource> ChainIndex for NodeBackedChainIndexSubscriber<Sou
             },
             None => None,
         };
-        let expected_chain_tip = non_finalized_snapshot.map(|snapshot| snapshot.best_tip.hash);
+
+        // Stale-snapshot check: compare the caller's snapshot against the
+        // chain-index's *current* non-finalized tip (the same source of
+        // truth as `snapshot_nonfinalized_state`), not against the mempool
+        // serve loop's `mempool_chain_tip`. The mempool polls on its own
+        // cadence, so its tip view can diverge from the chain-index's by
+        // up to one poll cycle (#1037); using it here means the API can
+        // accept a stale snapshot whenever the mempool happens to be
+        // lagging, and reject a freshly-issued snapshot whenever the
+        // mempool happens to be ahead — both are caller-visible internal
+        // contradictions. The chain-index is authoritative.
+        if let Some(expected) = non_finalized_snapshot {
+            match self.non_finalized_state.load().as_ref() {
+                Some(current) => {
+                    if expected.best_tip.hash != current.get_snapshot().best_tip.hash {
+                        return None;
+                    }
+                }
+                None => return None,
+            }
+        }
+
         let mut subscriber = self.mempool.clone();
 
-        match subscriber
-            .get_mempool_stream(expected_chain_tip)
-            .now_or_never()
-        {
+        match subscriber.get_mempool_stream(None).now_or_never() {
             Some(Ok((in_rx, _handle))) => {
                 let (out_tx, out_rx) =
                     tokio::sync::mpsc::channel::<Result<Vec<u8>, ChainIndexError>>(32);
diff --git a/packages/zaino-state/src/chain_index/mempool.rs b/packages/zaino-state/src/chain_index/mempool.rs
index 235706ce9..9ad512c0d 100644
--- a/packages/zaino-state/src/chain_index/mempool.rs
+++ b/packages/zaino-state/src/chain_index/mempool.rs
@@ -5,7 +5,7 @@ use std::{collections::HashSet, sync::Arc};
 use crate::{
     broadcast::{Broadcast, BroadcastSubscriber},
     chain_index::{
-        source::{BlockchainSource, BlockchainSourceError},
+        source::{wait_or_source_change, BlockchainSource, BlockchainSourceError},
         types::db::metadata::MempoolInfo,
     },
     error::{MempoolError, StatusError},
@@ -53,6 +53,44 @@ pub struct Mempool<T: BlockchainSource> {
     status: NamedAtomicStatus,
 }
 
+/// Fetches the current mempool snapshot (all txids paired with their
+/// serialized transactions) from a blockchain source.
+async fn fetch_mempool_snapshot<T: BlockchainSource>(
+    fetcher: &T,
+) -> Result<Vec<(MempoolKey, MempoolValue)>, MempoolError> {
+    let txids = fetcher.get_mempool_txids().await?.ok_or_else(|| {
+        MempoolError::BlockchainSourceError(BlockchainSourceError::Unrecoverable(
+            "could not fetch mempool data: mempool txid list was None".to_string(),
+        ))
+    })?;
+
+    let mut transactions = Vec::with_capacity(txids.len());
+    for txid in txids {
+        let (transaction, _location) =
+            fetcher
+                .get_transaction(txid.0.into())
+                .await?
+                .ok_or_else(|| {
+                    MempoolError::BlockchainSourceError(BlockchainSourceError::Unrecoverable(
+                        format!(
+                            "could not fetch mempool data: transaction not found for txid {txid}"
+                        ),
+                    ))
+                })?;
+
+        transactions.push((
+            MempoolKey {
+                txid: txid.to_string(),
+            },
+            MempoolValue {
+                serialized_tx: Arc::new(transaction.into()),
+            },
+        ));
+    }
+
+    Ok(transactions)
+}
+
 impl<T: BlockchainSource> Mempool<T> {
     /// Spawns a new [`Mempool`].
     #[instrument(name = "Mempool::spawn", skip(fetcher, capacity_and_shard_amount))]
@@ -60,34 +98,50 @@ impl<T: BlockchainSource> Mempool<T> {
         fetcher: T,
         capacity_and_shard_amount: Option<(usize, usize)>,
     ) -> Result<Self, MempoolError> {
-        // Wait for mempool in validator to come online.
-        loop {
-            match fetcher.get_mempool_txids().await {
-                Ok(_) => {
-                    break;
-                }
-                Err(_) => {
-                    info!("Waiting for Validator mempool to come online");
-                    tokio::time::sleep(std::time::Duration::from_secs(3)).await;
+        // Run the three independent startup probes concurrently:
+        // (1) wait for the validator mempool to come online (retry loop),
+        // (2) fetch the initial best block hash (fatal if unavailable),
+        // (3) fetch the initial mempool snapshot (retry on transient
+        //     fetch errors).
+        // None of the three depend on each other; serial ordering added
+        // two source round-trips to every indexer init for no semantic
+        // reason. `try_join!` cancels the retry loops if (2) returns a
+        // hard error.
+        let wait_for_mempool_online = async {
+            loop {
+                match fetcher.get_mempool_txids().await {
+                    Ok(_) => return Ok::<(), MempoolError>(()),
+                    Err(_) => {
+                        info!("Waiting for Validator mempool to come online");
+                        tokio::time::sleep(std::time::Duration::from_secs(3)).await;
+                    }
                 }
             }
-        }
-
-        let best_block_hash: BlockHash = match fetcher.get_best_block_hash().await {
-            Ok(block_hash_opt) => match block_hash_opt {
-                Some(hash) => hash.into(),
-                None => {
-                    return Err(MempoolError::Critical(
-                        "Error in mempool: Error connecting with validator".to_string(),
-                    ))
-                }
-            },
-            Err(_e) => {
-                return Err(MempoolError::Critical(
+        };
+        let fetch_best_block_hash = async {
+            match fetcher.get_best_block_hash().await {
+                Ok(Some(hash)) => Ok(BlockHash::from(hash)),
+                Ok(None) | Err(_) => Err(MempoolError::Critical(
                     "Error in mempool: Error connecting with validator".to_string(),
-                ))
+                )),
             }
         };
+        let fetch_initial_snapshot = async {
+            loop {
+                match fetch_mempool_snapshot(&fetcher).await {
+                    Ok(txs) => return Ok::<_, MempoolError>(txs),
+                    Err(e) => {
+                        warn!("{e}");
+                        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+                    }
+                }
+            }
+        };
+        let (_, best_block_hash, initial_transactions) = tokio::try_join!(
+            wait_for_mempool_online,
+            fetch_best_block_hash,
+            fetch_initial_snapshot,
+        )?;
 
         let (chain_tip_sender, _chain_tip_reciever) = tokio::sync::watch::channel(best_block_hash);
 
@@ -102,26 +156,11 @@ impl<T: BlockchainSource> Mempool<T> {
             },
             mempool_chain_tip: chain_tip_sender,
             sync_task_handle: None,
-            status: NamedAtomicStatus::new("Mempool", StatusType::Spawning),
+            status: NamedAtomicStatus::new("Mempool", StatusType::Ready),
         };
-
-        loop {
-            match mempool.get_mempool_transactions().await {
-                Ok(mempool_transactions) => {
-                    mempool.status.store(StatusType::Ready);
-                    mempool
-                        .state
-                        .insert_filtered_set(mempool_transactions, mempool.status.load());
-                    break;
-                }
-                Err(e) => {
-                    mempool.state.notify(mempool.status.load());
-                    warn!("{e}");
-                    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
-                    continue;
-                }
-            };
-        }
+        mempool
+            .state
+            .insert_filtered_set(initial_transactions, mempool.status.load());
 
         mempool.sync_task_handle = Some(std::sync::Mutex::new(mempool.serve().await?));
 
@@ -142,6 +181,7 @@ impl<T: BlockchainSource> Mempool<T> {
         status.store(StatusType::Spawning);
 
         let sync_handle = tokio::spawn(async move {
+            let mut change_rx = mempool.fetcher.change_subscribe();
             let mut best_block_hash: Hash;
             let mut check_block_hash: Hash;
 
@@ -207,11 +247,20 @@ impl<T: BlockchainSource> Mempool<T> {
                         .send_replace(check_block_hash.into());
                     best_block_hash = check_block_hash;
 
-                    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+                    // Cool-down before we re-fetch mempool state at the new
+                    // tip — gives the validator a moment to settle. Stays
+                    // interruptible: a follow-on `mine_blocks` must wake us
+                    // immediately, otherwise the next iteration of the test
+                    // sees a stale tip until the timer expires (#1037).
+                    wait_or_source_change(
+                        change_rx.as_mut(),
+                        std::time::Duration::from_millis(100),
+                    )
+                    .await;
                     continue;
                 }
 
-                match mempool.get_mempool_transactions().await {
+                match fetch_mempool_snapshot(&mempool.fetcher).await {
                     Ok(mempool_transactions) => {
                         status.store(StatusType::Ready);
                         state.insert_filtered_set(mempool_transactions, status.load());
@@ -230,51 +279,18 @@ impl<T: BlockchainSource> Mempool<T> {
                     return;
                 }
 
-                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+                // Wait the cadence interval, but wake immediately if the
+                // source signals new state — keeps mempool↔chain-index
+                // tip propagation aligned within a single iteration on
+                // push-capable sources (#1037 part 2/2).
+                wait_or_source_change(change_rx.as_mut(), std::time::Duration::from_millis(100))
+                    .await;
             }
         });
 
         Ok(sync_handle)
     }
 
-    /// Returns all transactions in the mempool.
-    async fn get_mempool_transactions(
-        &self,
-    ) -> Result<Vec<(MempoolKey, MempoolValue)>, MempoolError> {
-        let mut transactions = Vec::new();
-
-        let txids = self.fetcher.get_mempool_txids().await?.ok_or_else(|| {
-            MempoolError::BlockchainSourceError(BlockchainSourceError::Unrecoverable(
-                "could not fetch mempool data: mempool txid list was None".to_string(),
-            ))
-        })?;
-
-        for txid in txids {
-            let (transaction, _location) = self
-                .fetcher
-                .get_transaction(txid.0.into())
-                .await?
-                .ok_or_else(|| {
-                    MempoolError::BlockchainSourceError(
-                        crate::chain_index::source::BlockchainSourceError::Unrecoverable(format!(
-                            "could not fetch mempool data: transaction not found for txid {txid}"
-                        )),
-                    )
-                })?;
-
-            transactions.push((
-                MempoolKey {
-                    txid: txid.to_string(),
-                },
-                MempoolValue {
-                    serialized_tx: Arc::new(transaction.into()),
-                },
-            ));
-        }
-
-        Ok(transactions)
-    }
-
     /// Returns a [`MempoolSubscriber`].
     pub fn subscriber(&self) -> MempoolSubscriber {
         MempoolSubscriber {
@@ -365,6 +381,14 @@ pub struct MempoolSubscriber {
 }
 
 impl MempoolSubscriber {
+    /// Returns a clone of the watch receiver tracking the mempool's most
+    /// recently observed chain tip. Used by tests to detect the mempool↔
+    /// chain-index tip skew documented in #1037 without busy-polling.
+    #[cfg(test)]
+    pub(crate) fn mempool_tip(&self) -> tokio::sync::watch::Receiver<BlockHash> {
+        self.mempool_chain_tip.clone()
+    }
+
     /// Returns all tx currently in the mempool.
     pub async fn get_mempool(&self) -> Vec<(MempoolKey, MempoolValue)> {
         self.subscriber.get_filtered_state(&HashSet::new())
diff --git a/packages/zaino-state/src/chain_index/source.rs b/packages/zaino-state/src/chain_index/source.rs
index 4b470f8b2..24623a3ad 100644
--- a/packages/zaino-state/src/chain_index/source.rs
+++ b/packages/zaino-state/src/chain_index/source.rs
@@ -100,6 +100,41 @@ pub trait BlockchainSource: Clone + Send + Sync + 'static {
         start_index: u16,
         max_entries: Option<u16>,
     ) -> BlockchainSourceResult<Vec<([u8; 32], u32)>>;
+
+    /// Subscribe to source state-change notifications.
+    ///
+    /// Each subscriber receives every change-event on its own buffered
+    /// receiver, so wakes between iterations are preserved (no missed-wake
+    /// gap). Sync loops typically call `change_subscribe` once at startup
+    /// and `select!` `recv()` against their fixed-cadence timer, falling
+    /// through to the timer when no push notification arrives.
+    ///
+    /// The default returns `None` — poll-only sources (real validators)
+    /// pace themselves on the timer alone. Push-capable sources (test
+    /// mockchains) override to provide a live receiver.
+    fn change_subscribe(&self) -> Option<tokio::sync::broadcast::Receiver<()>> {
+        None
+    }
+}
+
+/// Sleep up to `duration`, but return early if `change_rx` resolves first.
+///
+/// Both sync loops in this module (chain-index sync, mempool serve) pace
+/// themselves on a fixed-cadence timer and want to wake immediately when the
+/// source signals new state. The two-arm `tokio::select!` is identical at
+/// every call site; this helper is the single home for the pattern. Pass
+/// `None` for poll-only sources — the helper degrades to a plain sleep.
+pub(super) async fn wait_or_source_change(
+    change_rx: Option<&mut tokio::sync::broadcast::Receiver<()>>,
+    duration: std::time::Duration,
+) {
+    match change_rx {
+        Some(rx) => tokio::select! {
+            _ = tokio::time::sleep(duration) => {}
+            _ = rx.recv() => {}
+        },
+        None => tokio::time::sleep(duration).await,
+    }
 }
 
 /// An error originating from a blockchain source.
@@ -825,6 +860,7 @@ pub(crate) mod test {
         atomic::{AtomicU32, Ordering},
         Arc,
     };
+    use tokio::sync::broadcast;
     use zebra_chain::{block::Block, orchard::tree as orchard, sapling::tree as sapling};
     use zebra_state::HashOrHeight;
 
@@ -874,6 +910,14 @@ pub(crate) mod test {
         >,
         active_chain_height: Arc<AtomicU32>,
         force_requests_against_source_to_fail: Arc<std::sync::atomic::AtomicBool>,
+        /// Pings every subscriber registered via
+        /// [`BlockchainSource::change_subscribe`] when [`Self::mine_blocks`]
+        /// advances the active height, so each can wake from its interval
+        /// timer immediately. `broadcast` (over `Notify`) gives every
+        /// subscriber its own buffered receiver, so a `mine_blocks` that
+        /// fires while one subsystem is mid-iteration is preserved on its
+        /// receiver and consumed on the next `recv().await`.
+        change_broadcast: broadcast::Sender<()>,
     }
 
     impl MockchainSource {
@@ -906,6 +950,7 @@ pub(crate) mod test {
                 force_requests_against_source_to_fail: Arc::new(
                     std::sync::atomic::AtomicBool::new(false),
                 ),
+                change_broadcast: broadcast::channel(16).0,
             }
         }
 
@@ -950,6 +995,7 @@ pub(crate) mod test {
                 force_requests_against_source_to_fail: Arc::new(
                     std::sync::atomic::AtomicBool::new(false),
                 ),
+                change_broadcast: broadcast::channel(16).0,
             }
         }
 
@@ -962,6 +1008,30 @@ pub(crate) mod test {
 
         pub(crate) fn mine_blocks(&self, blocks: u32) {
             // len() returns one-indexed length, height is zero-indexed.
+            let max_height = self.max_chain_height();
+            let advanced = self
+                .active_chain_height
+                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |current| {
+                    let target = current.saturating_add(blocks).min(max_height);
+                    if target == current {
+                        None
+                    } else {
+                        Some(target)
+                    }
+                })
+                .is_ok();
+            if advanced {
+                let _ = self.change_broadcast.send(());
+            }
+        }
+
+        /// Like [`Self::mine_blocks`] but does *not* fire the source's
+        /// change-notify. Lets the chain-index sync loop fall through to
+        /// its timer instead of waking immediately — the only way to put
+        /// the chain-index *behind* the mempool in tests, since the
+        /// mempool's serve loop polls `get_best_block_hash` directly and
+        /// always notices, notify or not.
+        pub(crate) fn mine_blocks_silent(&self, blocks: u32) {
             let max_height = self.max_chain_height();
             let _ = self.active_chain_height.fetch_update(
                 Ordering::SeqCst,
@@ -986,6 +1056,13 @@ pub(crate) mod test {
             self.active_chain_height.load(Ordering::SeqCst)
         }
 
+        /// Returns the [`BlockHash`] at [`Self::active_height`]. Used by
+        /// tip-skew tests to compare the mempool's tracked tip against the
+        /// expected post-mining hash.
+        pub(crate) fn active_block_hash(&self) -> BlockHash {
+            self.hashes[self.active_height() as usize]
+        }
+
         fn valid_height(&self, height: u32) -> Option<usize> {
             let active_chain_height = self.active_height() as usize;
             let valid_height = height as usize;
@@ -1185,5 +1262,9 @@ pub(crate) mod test {
         ) -> BlockchainSourceResult<Vec<([u8; 32], u32)>> {
             todo!()
         }
+
+        fn change_subscribe(&self) -> Option<broadcast::Receiver<()>> {
+            Some(self.change_broadcast.subscribe())
+        }
     }
 }
diff --git a/packages/zaino-state/src/chain_index/tests.rs b/packages/zaino-state/src/chain_index/tests.rs
index ec9ca5ce1..9f08a416b 100644
--- a/packages/zaino-state/src/chain_index/tests.rs
+++ b/packages/zaino-state/src/chain_index/tests.rs
@@ -2,6 +2,7 @@
 
 pub(crate) mod finalised_state;
 pub(crate) mod mempool;
+mod mempool_skew;
 mod mockchain_tests;
 mod poll;
 mod proptest_blockgen;
@@ -23,7 +24,6 @@ pub(crate) fn init_tracing() {
 
 use std::path::PathBuf;
 use tempfile::TempDir;
-use tokio::time::Duration;
 use zaino_common::{network::ActivationHeights, DatabaseConfig, Network, StorageConfig};
 
 use crate::{
@@ -45,21 +45,7 @@ async fn load_test_vectors_and_sync_chain_index(
     NodeBackedChainIndexSubscriber<MockchainSource>,
     MockchainSource,
 ) {
-    // The 2 s poll interval here is load-bearing for other tests: most
-    // callers (mockchain_tests, mempool, poll, proptest_blockgen) drop the
-    // indexer without calling `shutdown()`, relying on the background sync
-    // loop being in its post-success `interval` sleep at teardown to avoid
-    // racing with runtime shutdown. Shorter polling lets the test body
-    // return before that settle point and exposes the latent race. Tests
-    // that need faster setup should use
-    // `load_test_vectors_and_sync_chain_index_with_timings` and handle
-    // their own teardown.
-    load_with_settings(
-        active_mockchain_source,
-        SyncTimings::default(),
-        Duration::from_secs(2),
-    )
-    .await
+    load_with_settings(active_mockchain_source, SyncTimings::default()).await
 }
 
 async fn load_test_vectors_and_sync_chain_index_with_timings(
@@ -71,18 +57,12 @@ async fn load_test_vectors_and_sync_chain_index_with_timings(
     NodeBackedChainIndexSubscriber<MockchainSource>,
     MockchainSource,
 ) {
-    load_with_settings(
-        active_mockchain_source,
-        sync_timings,
-        Duration::from_millis(25),
-    )
-    .await
+    load_with_settings(active_mockchain_source, sync_timings).await
 }
 
 async fn load_with_settings(
     active_mockchain_source: bool,
     sync_timings: SyncTimings,
-    setup_poll_interval: Duration,
 ) -> (
     Vec<vectors::TestVectorBlockData>,
     NodeBackedChainIndex<MockchainSource>,
@@ -119,18 +99,35 @@ async fn load_with_settings(
         .unwrap();
     let index_reader = indexer.subscriber();
 
+    let check_height = crate::Height(match active_mockchain_source {
+        true => source.active_height() - 100,
+        false => 100,
+    });
+    let mut status = index_reader.status_subscribe();
     loop {
-        let check_height: u32 = match active_mockchain_source {
-            true => source.active_height() - 100,
-            false => 100,
-        };
-        if index_reader.finalized_state.db_height().await.unwrap()
-            == Some(crate::Height(check_height))
+        if *status.borrow_and_update() == crate::StatusType::Ready
+            && index_reader.finalized_state.db_height().await.unwrap() == Some(check_height)
         {
             break;
         }
-        tokio::time::sleep(setup_poll_interval).await;
+        status
+            .changed()
+            .await
+            .expect("ChainIndex status sender dropped before reaching Ready");
     }
 
+    // Most callers (mockchain_tests, mempool, poll, proptest_blockgen) drop
+    // the indexer without `shutdown()`, relying on the sync loop being
+    // parked in its `tokio::time::sleep(sync_timings.interval)` (see
+    // chain_index.rs sync loop) at teardown so runtime shutdown doesn't
+    // race a mid-iteration DB write. The watch-based rendezvous above
+    // wakes one statement *before* that sleep is polled, so under the
+    // multi-thread runtime there is a sub-millisecond window where the
+    // sync loop has not yet yielded. One full `interval` here lets the
+    // sync loop reach the sleep statement and stay there. Removing this
+    // requires explicit indexer shutdown (`CancellationToken`-based,
+    // tracked separately).
+    tokio::time::sleep(sync_timings.interval).await;
+
     (blocks, indexer, index_reader, source)
 }
diff --git a/packages/zaino-state/src/chain_index/tests/mempool_skew.rs b/packages/zaino-state/src/chain_index/tests/mempool_skew.rs
new file mode 100644
index 000000000..f4f436256
--- /dev/null
+++ b/packages/zaino-state/src/chain_index/tests/mempool_skew.rs
@@ -0,0 +1,178 @@
+//! Regression suite for #1037 — mempool↔chain-index tip skew.
+//!
+//! The chain-index has two independent sync tasks observing the same
+//! `BlockchainSource` on independent cadences: the chain-index sync loop
+//! and the mempool serve loop. Their tip views can diverge for a window
+//! after every block. This module pins each direction of the skew with a
+//! deterministic test so the principled fix can be TDD'd against both.
+
+use tokio::time::{timeout, Duration};
+
+use super::{load_test_vectors_and_sync_chain_index, mockchain_tests::wait_for_indexer_tip};
+use crate::{
+    chain_index::{source::test::MockchainSource, ChainIndex, NodeBackedChainIndexSubscriber},
+    BlockHash,
+};
+
+/// Waits for the next change of the mempool serve loop's `mempool_chain_tip`
+/// watch — i.e. the next time the mempool observes a new best-block hash and
+/// resets. Used by direction-#2 to act *exactly* in the window where the
+/// mempool has advanced but the chain-index has not.
+async fn wait_for_mempool_tip_change(
+    index_reader: &NodeBackedChainIndexSubscriber<MockchainSource>,
+) {
+    let mut tip = index_reader.mempool_tip();
+    tip.borrow_and_update();
+    timeout(Duration::from_secs(10), tip.changed())
+        .await
+        .unwrap_or_else(|_| panic!("mempool tip did not change within 10 s"))
+        .expect("mempool_chain_tip sender dropped");
+}
+
+/// Waits until the mempool serve loop's `mempool_chain_tip` equals
+/// `expected`, or panics after 10 s. Used to re-synchronise between
+/// iterations of property-style skew tests.
+async fn wait_for_mempool_tip(
+    index_reader: &NodeBackedChainIndexSubscriber<MockchainSource>,
+    expected: BlockHash,
+) {
+    let mut tip = index_reader.mempool_tip();
+    let work = async {
+        loop {
+            if *tip.borrow_and_update() == expected {
+                return;
+            }
+            tip.changed()
+                .await
+                .expect("mempool_chain_tip sender dropped");
+        }
+    };
+    timeout(Duration::from_secs(10), work)
+        .await
+        .unwrap_or_else(|_| panic!("mempool tip never reached expected hash within 10 s"));
+}
+
+/// Tip-skew direction #1 (#1037): chain-index sync loop ahead of mempool.
+///
+/// After `mine_blocks(1)` + `wait_for_indexer_tip`, the chain-index has
+/// reached the new tip but the mempool's serve loop may not have. Calling
+/// `get_mempool_stream` with a snapshot taken before mining must reject
+/// the snapshot as stale even when the mempool's own tip view still
+/// matches it — i.e. staleness must be defined against the *latest*
+/// observed tip, not whichever subsystem the consumer happens to read
+/// first. Companion direction (mempool ahead of chain-index) lives in
+/// the sibling test.
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn chain_index_ahead_returns_stale_stream() {
+    let (_blocks, _indexer, index_reader, mockchain) =
+        load_test_vectors_and_sync_chain_index(true).await;
+    wait_for_indexer_tip(&index_reader, mockchain.active_height()).await;
+
+    let stale_nonfinalized_snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap();
+
+    mockchain.mine_blocks(1);
+    wait_for_indexer_tip(&index_reader, mockchain.active_height()).await;
+
+    let mempool_stream = index_reader.get_mempool_stream(Some(&stale_nonfinalized_snapshot));
+
+    assert!(mempool_stream.is_none());
+}
+
+/// Tip-skew direction #2 (#1037): mempool serve loop ahead of chain-index.
+///
+/// Constructed by advancing the mockchain via `mine_blocks_silent`, which
+/// suppresses the source's `change_subscribe` broadcast. The
+/// mempool's serve loop polls `get_best_block_hash` on its own cadence,
+/// so it observes the new tip first. Once the mempool has advanced
+/// (`wait_for_mempool_tip_change`), the test takes a snapshot — which
+/// reflects the chain-index's *current* (lagging) view — and asks for a
+/// mempool stream against that snapshot.
+///
+/// Calling `get_mempool_stream(snapshot)` immediately after a
+/// `snapshot_nonfinalized_state()` from the same
+/// `NodeBackedChainIndexSubscriber` should not reject the caller's
+/// snapshot: the API just handed it out, it is the freshest chain-index
+/// view available, and refusing the stream for it is an internal
+/// contradiction (left hand says fresh, right hand says stale). Today
+/// the staleness check compares `snapshot.best_tip.hash` against
+/// `mempool_chain_tip` only, so when the mempool has moved past the
+/// chain-index, this comparison fails and the stream is dropped.
+///
+/// Expected to fail until #1037 is fixed.
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn mempool_ahead_rejects_fresh_snapshot() {
+    let (_blocks, _indexer, index_reader, mockchain) =
+        load_test_vectors_and_sync_chain_index(true).await;
+    wait_for_indexer_tip(&index_reader, mockchain.active_height()).await;
+
+    // Advance the source without notifying the chain-index sync loop,
+    // letting the mempool's poll observe the new tip first.
+    mockchain.mine_blocks_silent(1);
+    wait_for_mempool_tip_change(&index_reader).await;
+
+    // The snapshot the API just handed out should yield a valid stream.
+    let fresh_snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap();
+
+    // Sanity check: chain-index is still at the old tip in this snapshot.
+    let chain_index_tip = fresh_snapshot.get_nfs_snapshot().unwrap().best_tip.height.0;
+    assert!(
+        chain_index_tip < mockchain.active_height(),
+        "test setup: chain-index should be behind the mockchain (got chain_index={chain_index_tip}, mockchain={})",
+        mockchain.active_height(),
+    );
+
+    let mempool_stream = index_reader.get_mempool_stream(Some(&fresh_snapshot));
+
+    assert!(
+        mempool_stream.is_some(),
+        "API rejected its own freshly-issued snapshot — mempool↔chain-index tip skew (#1037 direction #2)"
+    );
+}
+
+/// Convergence-bound (#1037 success criterion): each `mine_blocks` event
+/// should bring both subsystems to the new tip within a single iteration.
+///
+/// At each iteration the test mines one block, awaits the chain-index
+/// tip via the existing event-driven helper, then samples the mempool's
+/// tracked tip *immediately*. If the mempool also converged in the same
+/// iteration, the sample reflects the new tip; otherwise the sample
+/// still equals the pre-mining hash and the iteration is counted as a
+/// lag.
+///
+/// Today the chain-index sync loop wakes on `mine_blocks` (Option-2
+/// notify) but the mempool serve loop polls on its own 100 ms cadence,
+/// so the chain-index nearly always wins the race and the mempool lags
+/// in the vast majority of iterations. Repeating the sample 20× makes
+/// the failure deterministic in practice — the probability of zero lags
+/// across 20 events under a ~10 % per-iteration win rate is negligible.
+///
+/// Once the principled fix wakes both subsystems on `mine_blocks`, the
+/// expected outcome is `lag_count == 0` deterministically.
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn tips_converge_within_bounded_time_after_mining() {
+    let (_blocks, _indexer, index_reader, mockchain) =
+        load_test_vectors_and_sync_chain_index(true).await;
+    wait_for_indexer_tip(&index_reader, mockchain.active_height()).await;
+
+    let mut lag_count = 0;
+    for _ in 0..20 {
+        let pre_mempool_tip = *index_reader.mempool_tip().borrow();
+        mockchain.mine_blocks(1);
+        let new_height = mockchain.active_height();
+        let new_hash = mockchain.active_block_hash();
+
+        wait_for_indexer_tip(&index_reader, new_height).await;
+        if *index_reader.mempool_tip().borrow() == pre_mempool_tip {
+            lag_count += 1;
+        }
+
+        // Resync mempool before the next iteration so we're not
+        // chasing multiple stacked updates.
+        wait_for_mempool_tip(&index_reader, new_hash).await;
+    }
+
+    assert_eq!(
+        lag_count, 0,
+        "mempool lagged chain-index in {lag_count}/20 mine_blocks events (#1037)"
+    );
+}
diff --git a/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs b/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs
index 88fe92376..d839ebb6a 100644
--- a/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs
+++ b/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs
@@ -1,43 +1,53 @@
 use super::load_test_vectors_and_sync_chain_index;
-use tokio::time::{sleep, Duration};
+use tokio::time::{sleep, timeout, Duration};
 use tokio_stream::StreamExt as _;
 use zebra_chain::serialization::ZcashDeserializeInto;
 
-use crate::chain_index::{
-    source::test::MockchainSource,
-    tests::{poll::poll_until, vectors::TestVectorBlockData},
-    types::{BestChainLocation, TransactionHash},
-    ChainIndex, NodeBackedChainIndexSubscriber,
+use crate::{
+    chain_index::{
+        source::test::MockchainSource,
+        tests::vectors::TestVectorBlockData,
+        types::{BestChainLocation, TransactionHash},
+        ChainIndex, NodeBackedChainIndexSubscriber,
+    },
+    StatusType,
 };
 
-/// Polls the indexer's nonfinalized-state snapshot until its best-tip height
-/// equals `expected`, or panics after a 10 s budget.
+/// Waits until the indexer's nonfinalized-state best-tip height equals
+/// `expected`, or panics after a 10 s budget.
 ///
-/// Use this wherever a test previously relied on a fixed `sleep` to hope the
-/// indexer's sync task had caught up with the mockchain tip: the indexer
-/// publishes new tips asynchronously via its background loop, and under
-/// full-suite parallel load those updates can lag well past 2 s.
-async fn wait_for_indexer_tip(
+/// Wakes on every transition of the chain-index sync loop's status (via
+/// [`NodeBackedChainIndexSubscriber::status_subscribe`]) and re-checks the
+/// snapshot. The sync loop transitions to [`StatusType::Ready`] only after a
+/// successful iteration has updated the tip, so each Ready transition is the
+/// earliest moment the test can observe a new tip — no fixed-cadence polling
+/// required.
+pub(super) async fn wait_for_indexer_tip(
     index_reader: &NodeBackedChainIndexSubscriber<MockchainSource>,
     expected: u32,
 ) {
-    poll_until(
-        "indexer tip to match expected height",
-        Duration::from_secs(10),
-        Duration::from_millis(25),
-        || async {
-            let tip = index_reader
-                .snapshot_nonfinalized_state()
+    let mut status = index_reader.status_subscribe();
+    let work = async {
+        loop {
+            if *status.borrow_and_update() == StatusType::Ready {
+                let tip = index_reader
+                    .snapshot_nonfinalized_state()
+                    .await
+                    .ok()
+                    .and_then(|s| s.get_nfs_snapshot().map(|n| n.best_tip.height.0));
+                if tip == Some(expected) {
+                    return;
+                }
+            }
+            status
+                .changed()
                 .await
-                .ok()?
-                .get_nfs_snapshot()?
-                .best_tip
-                .height
-                .0;
-            (tip == expected).then_some(())
-        },
-    )
-    .await;
+                .expect("ChainIndex status sender dropped before reaching expected tip");
+        }
+    };
+    timeout(Duration::from_secs(10), work)
+        .await
+        .unwrap_or_else(|_| panic!("indexer tip never reached {expected} within 10 s"));
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -499,22 +509,6 @@ async fn get_mempool_stream_correct_expected_chain_tip_snapshot() {
     );
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
-async fn get_mempool_stream_for_stale_snapshot() {
-    let (_blocks, _indexer, index_reader, mockchain) =
-        load_test_vectors_and_sync_chain_index(true).await;
-    wait_for_indexer_tip(&index_reader, mockchain.active_height()).await;
-
-    let stale_nonfinalized_snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap();
-
-    mockchain.mine_blocks(1);
-    wait_for_indexer_tip(&index_reader, mockchain.active_height()).await;
-
-    let mempool_stream = index_reader.get_mempool_stream(Some(&stale_nonfinalized_snapshot));
-
-    assert!(mempool_stream.is_none());
-}
-
 #[tokio::test(flavor = "multi_thread")]
 async fn get_block_height() {
     let (blocks, _indexer, index_reader, _mockchain) =
diff --git a/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs b/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs
index 48be02a88..9d1885526 100644
--- a/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs
+++ b/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs
@@ -61,7 +61,7 @@ fn passthrough_test(
                 // This number can be played with. We want to slow down
                 // sync enough to trigger passthrough without
                 // slowing down passthrough more than we need to
-                delay: Some(Duration::from_millis(100)),
+                delay: Some(Duration::from_millis(10)),
                 best_branch_cache: Arc::new(std::sync::OnceLock::new()),
                 tx_index: Arc::new(std::sync::OnceLock::new()),
             };
@@ -393,8 +393,19 @@ fn make_chain() {
             let indexer = NodeBackedChainIndex::new(mockchain.clone(), config)
                 .await
                 .unwrap();
-            tokio::time::sleep(Duration::from_secs(5)).await;
             let index_reader = indexer.subscriber();
+            let expected_block_count = segment_length * (branch_count + 1);
+            poll_until(
+                "indexer to load all branch blocks into the non-finalized state",
+                Duration::from_secs(30),
+                Duration::from_millis(50),
+                || async {
+                    let snapshot = index_reader.snapshot_nonfinalized_state().await.ok()?;
+                    let nfs = snapshot.get_nfs_snapshot()?;
+                    (nfs.blocks.len() == expected_block_count).then_some(())
+                },
+            )
+            .await;
             let snapshot = index_reader.snapshot_nonfinalized_state().await.unwrap();
             let non_finalized_snapshot = snapshot.get_nfs_snapshot().expect("not synced");
             let best_tip_hash = non_finalized_snapshot.best_tip.hash;
diff --git a/packages/zaino-state/src/chain_index/tests/sync_loop.rs b/packages/zaino-state/src/chain_index/tests/sync_loop.rs
index 257b0e8ac..2b686d720 100644
--- a/packages/zaino-state/src/chain_index/tests/sync_loop.rs
+++ b/packages/zaino-state/src/chain_index/tests/sync_loop.rs
@@ -44,40 +44,27 @@ async fn survives_transient_source_failure() {
 /// After `max_consecutive_failures` with exponential backoff, the sync loop
 /// should escalate to [`StatusType::CriticalError`].
 ///
-/// Uses [`SyncTimings::fast`] (10× shrunk) so the full backoff schedule fits
-/// in a few seconds instead of ~40 s.
+/// Uses [`SyncTimings::fast`] so the full backoff schedule fits in
+/// ~120 ms instead of ~40 s.
 #[tokio::test(flavor = "multi_thread")]
 async fn escalates_to_critical_after_persistent_failure() {
     let timings = SyncTimings::fast();
     let (_blocks, _indexer, index_reader, mockchain) =
         load_test_vectors_and_sync_chain_index_with_timings(true, timings).await;
 
-    let start = Instant::now();
     mockchain.set_failing(true);
 
     // 5× slack over the nominal backoff sum to absorb scheduling jitter and
     // the per-iteration sync work the loop performs between sleeps.
     let max_time_to_critical = timings.max_backoff_window() * 5;
-    let poll_interval = timings.initial_backoff;
-
-    loop {
-        sleep(poll_interval).await;
-
-        if index_reader.status() == StatusType::CriticalError {
-            break;
-        }
 
-        assert!(
-            start.elapsed() < max_time_to_critical,
-            "CriticalError was not reached within {max_time_to_critical:?}"
-        );
-    }
-
-    let elapsed = start.elapsed();
-    assert!(
-        elapsed < max_time_to_critical,
-        "CriticalError took {elapsed:?}, exceeding the maximum backoff window"
-    );
+    super::poll::poll_until(
+        "sync loop to escalate to CriticalError under persistent source failure",
+        max_time_to_critical,
+        timings.initial_backoff,
+        || async { (index_reader.status() == StatusType::CriticalError).then_some(()) },
+    )
+    .await;
 }
 
 /// Moved here from the integration test
diff --git a/packages/zaino-state/src/status.rs b/packages/zaino-state/src/status.rs
index 1afe177e9..ec902f62d 100644
--- a/packages/zaino-state/src/status.rs
+++ b/packages/zaino-state/src/status.rs
@@ -1,13 +1,15 @@
 //! Thread-safe status wrapper.
 //!
 //! This module provides [`AtomicStatus`], a thread-safe wrapper for [`StatusType`],
-//! and [`NamedAtomicStatus`], a variant that logs status transitions.
+//! and [`NamedAtomicStatus`], a variant that logs status transitions and supports
+//! awaiting transitions via [`NamedAtomicStatus::subscribe`].
 
 use std::sync::{
     atomic::{AtomicUsize, Ordering},
     Arc,
 };
 
+use tokio::sync::watch;
 use tracing::debug;
 
 pub use zaino_common::status::{Status, StatusType};
@@ -45,21 +47,23 @@ impl Status for AtomicStatus {
 
 /// Thread-safe status wrapper with component name for observability.
 ///
-/// Unlike [`AtomicStatus`], this variant logs all status transitions,
-/// making it easier to trace component lifecycle during debugging.
+/// Backed by a [`tokio::sync::watch`] channel: every transition is logged and
+/// every subscriber wakes via [`watch::Receiver::changed`]. Identical-value
+/// stores are no-ops — neither logged nor broadcast.
 #[derive(Debug, Clone)]
 pub struct NamedAtomicStatus {
     name: &'static str,
-    inner: Arc<AtomicUsize>,
+    inner: Arc<watch::Sender<StatusType>>,
 }
 
 impl NamedAtomicStatus {
     /// Creates a new NamedAtomicStatus with the given component name and initial status.
     pub fn new(name: &'static str, status: StatusType) -> Self {
         debug!(component = name, status = %status, "[STATUS] initial");
+        let (tx, _rx) = watch::channel(status);
         Self {
             name,
-            inner: Arc::new(AtomicUsize::new(status.into())),
+            inner: Arc::new(tx),
         }
     }
 
@@ -70,10 +74,11 @@ impl NamedAtomicStatus {
 
     /// Loads the value held in the NamedAtomicStatus.
     pub fn load(&self) -> StatusType {
-        StatusType::from(self.inner.load(Ordering::SeqCst))
+        *self.inner.borrow()
     }
 
-    /// Sets the value held in the NamedAtomicStatus, logging the transition.
+    /// Sets the value held in the NamedAtomicStatus, logging and broadcasting
+    /// the transition. Storing the current value is a no-op.
     pub fn store(&self, status: StatusType) {
         let old = self.load();
         if old != status {
@@ -83,8 +88,18 @@ impl NamedAtomicStatus {
                 to = %status,
                 "[STATUS] transition"
             );
+            self.inner.send_replace(status);
         }
-        self.inner.store(status.into(), Ordering::SeqCst);
+    }
+
+    /// Returns a [`watch::Receiver`] that observes every status transition.
+    ///
+    /// Use this to wait for a specific state with
+    /// [`watch::Receiver::changed`] / [`watch::Receiver::borrow_and_update`]
+    /// instead of busy-polling [`Self::load`].
+    #[cfg(test)]
+    pub(crate) fn subscribe(&self) -> watch::Receiver<StatusType> {
+        self.inner.subscribe()
     }
 }