chore: remove chain tip from mempool subscription (#1771)

Author: SantiagoPittella

CHANGELOG.md

@@ -28,6 +28,7 @@
  - Replaced NTX Builder's in-memory state management with SQLite-backed persistence; account states, notes, and transaction effects are now stored in the database and inflight state is purged on startup ([#1662](https://github.com/0xMiden/node/pull/1662)).
 - [BREAKING] Reworked `miden-remote-prover`, removing the `worker`/`proxy` distinction and simplifying to a `worker` with a request queue ([#1688](https://github.com/0xMiden/node/pull/1688)).
 - [BREAKING] Renamed `NoteRoot` protobuf message used in `GetNoteScriptByRoot` gRPC endpoints into `NoteScriptRoot` ([#1722](https://github.com/0xMiden/node/pull/1722)).
+- Removed `chain_tip` requirement from mempool subscription request ([#1771](https://github.com/0xMiden/node/pull/1771)).
 
 ### Fixes
 

crates/block-producer/src/mempool/mod.rs

@@ -586,17 +586,9 @@ impl Mempool {
     /// Creates a subscription to [`MempoolEvent`] which will be emitted in the order they occur.
     ///
     /// Only emits events which occurred after the current committed block.
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if the provided chain tip does not match the mempool's chain tip. This
-    /// prevents desync between the caller's view of the world and the mempool's event stream.
     #[instrument(target = COMPONENT, name = "mempool.subscribe", skip_all)]
-    pub fn subscribe(
-        &mut self,
-        chain_tip: BlockNumber,
-    ) -> Result<mpsc::Receiver<MempoolEvent>, BlockNumber> {
-        self.subscription.subscribe(chain_tip)
+    pub fn subscribe(&mut self) -> mpsc::Receiver<MempoolEvent> {
+        self.subscription.subscribe()
     }
 
     // STATS & INSPECTION

crates/block-producer/src/mempool/subscription.rs

@@ -43,22 +43,7 @@ impl SubscriptionProvider {
     /// Creates a new [`MempoolEvent`] subscription.
     ///
     /// This replaces any existing subscription.
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if the provided chain tip does not match the provider's. The error
-    /// value contains the provider's chain tip.
-    ///
-    /// This prevents desync between the subscribers view of the world and the mempool's event
-    /// stream.
-    pub fn subscribe(
-        &mut self,
-        chain_tip: BlockNumber,
-    ) -> Result<mpsc::Receiver<MempoolEvent>, BlockNumber> {
-        if self.chain_tip != chain_tip {
-            return Err(self.chain_tip);
-        }
-
+    pub fn subscribe(&mut self) -> mpsc::Receiver<MempoolEvent> {
         // We should leave enough space to at least send the uncommitted events (plus some extra).
         let capacity = self.inflight_txs.len().mul(2).max(1024);
         let (tx, rx) = mpsc::channel(capacity);
@@ -74,7 +59,7 @@ impl SubscriptionProvider {
             Self::send_event(&mut self.subscription, tx.clone());
         }
 
-        Ok(rx)
+        rx
     }
 
     pub(super) fn transaction_added(&mut self, tx: &AuthenticatedTransaction) {

crates/block-producer/src/server/mod.rs

@@ -414,22 +414,9 @@ impl api_server::Api for BlockProducerRpcServer {
 
     async fn mempool_subscription(
         &self,
-        request: tonic::Request<proto::block_producer::MempoolSubscriptionRequest>,
+        _request: tonic::Request<()>,
     ) -> Result<tonic::Response<Self::MempoolSubscriptionStream>, tonic::Status> {
-        let chain_tip = BlockNumber::from(request.into_inner().chain_tip);
-
-        let subscription =
-            self.mempool
-                .lock()
-                .await
-                .lock()
-                .await
-                .subscribe(chain_tip)
-                .map_err(|mempool_tip| {
-                    tonic::Status::invalid_argument(format!(
-                        "Mempool's chain tip {mempool_tip} does not match request's {chain_tip}"
-                    ))
-                })?;
+        let subscription = self.mempool.lock().await.lock().await.subscribe();
         let subscription = ReceiverStream::new(subscription);
 
         Ok(tonic::Response::new(MempoolEventSubscription { inner: subscription }))

crates/ntx-builder/src/clients/block_producer.rs

@@ -5,7 +5,6 @@ use miden_node_proto::clients::{BlockProducerClient as InnerBlockProducerClient,
 use miden_node_proto::domain::mempool::MempoolEvent;
 use miden_node_proto::generated::{self as proto};
 use miden_node_utils::FlattenResult;
-use miden_protocol::block::BlockNumber;
 use miden_protocol::transaction::ProvenTransaction;
 use miden_tx::utils::Serializable;
 use tokio_stream::StreamExt;
@@ -61,11 +60,10 @@ impl BlockProducerClient {
     #[instrument(target = COMPONENT, name = "ntx.block_producer.client.subscribe_to_mempool", skip_all, err)]
     pub async fn subscribe_to_mempool_with_retry(
         &self,
-        chain_tip: BlockNumber,
     ) -> Result<impl TryStream<Ok = MempoolEvent, Error = Status> + Send + 'static, Status> {
         let mut retry_counter = 0;
         loop {
-            match self.subscribe_to_mempool(chain_tip).await {
+            match self.subscribe_to_mempool().await {
                 Err(err) if err.code() == tonic::Code::Unavailable => {
                     // Exponential backoff with base 500ms and max 30s.
                     let backoff = Duration::from_millis(500)
@@ -89,11 +87,8 @@ impl BlockProducerClient {
 
     async fn subscribe_to_mempool(
         &self,
-        chain_tip: BlockNumber,
     ) -> Result<impl TryStream<Ok = MempoolEvent, Error = Status> + Send + 'static, Status> {
-        let request =
-            proto::block_producer::MempoolSubscriptionRequest { chain_tip: chain_tip.as_u32() };
-        let stream = self.client.clone().mempool_subscription(request).await?;
+        let stream = self.client.clone().mempool_subscription(()).await?;
 
         let stream = stream
             .into_inner()

crates/ntx-builder/src/lib.rs

@@ -204,29 +204,19 @@ impl NtxBuilderConfig {
         let store = StoreClient::new(self.store_url.clone());
         let block_producer = BlockProducerClient::new(self.block_producer_url.clone());
 
-        let (chain_tip_header, chain_mmr, mempool_events) = loop {
-            let (chain_tip_header, chain_mmr) = store
-                .get_latest_blockchain_data_with_retry()
-                .await?
-                .context("store should contain a latest block")?;
-
-            match block_producer
-                .subscribe_to_mempool_with_retry(chain_tip_header.block_num())
-                .await
-            {
-                Ok(subscription) => {
-                    let stream: MempoolEventStream = Box::pin(subscription.into_stream());
-                    break (chain_tip_header, chain_mmr, stream);
-                },
-                Err(status) if status.code() == tonic::Code::InvalidArgument => {
-                    tracing::warn!(
-                        err = %status,
-                        "mempool subscription failed due to chain tip desync, retrying"
-                    );
-                },
-                Err(err) => return Err(err).context("failed to subscribe to mempool events"),
-            }
-        };
+        // Subscribe to mempool first to ensure we don't miss any events. The subscription
+        // replays all inflight transactions, so the subscriber's state is fully reconstructed.
+        let subscription = block_producer
+            .subscribe_to_mempool_with_retry()
+            .await
+            .map_err(|err| anyhow::anyhow!(err))
+            .context("failed to subscribe to mempool events")?;
+        let mempool_events: MempoolEventStream = Box::pin(subscription.into_stream());
+
+        let (chain_tip_header, chain_mmr) = store
+            .get_latest_blockchain_data_with_retry()
+            .await?
+            .context("store should contain a latest block")?;
 
         // Store the chain tip in the DB.
         db.upsert_chain_state(chain_tip_header.block_num(), chain_tip_header.clone())

proto/proto/internal/block_producer.proto

@@ -35,27 +35,19 @@ service Api {
 
     // Subscribe to mempool events.
     //
-    // The request will be rejected if the caller and the mempool disagree on the current chain tip.
-    // This prevents potential desync issues. The caller can resolve this by resync'ing its chain state.
+    // The event stream will contain all events after the current chain tip. This includes all
+    // currently inflight events that have not yet been committed to the chain.
     //
-    // The event stream will contain all events after the chain tip. This includes all currently inflight
-    // events that have not yet been committed to the chain.
-    //
-    // Currently only a single active subscription is supported. Subscription requests will cancel the active
-    // subscription, if any.
-    rpc MempoolSubscription(MempoolSubscriptionRequest) returns (stream MempoolEvent) {}
+    // Currently only a single active subscription is supported. Subscription requests will cancel
+    // the active subscription, if any.
+    rpc MempoolSubscription(google.protobuf.Empty) returns (stream MempoolEvent) {}
 }
 
 // MEMPOOL SUBSCRIPTION
 // ================================================================================================
 
 // Request to subscribe to mempool events.
-message MempoolSubscriptionRequest {
-    // The caller's current chain height.
-    //
-    // Request will be rejected if this does not match the mempool's current view.
-    fixed32 chain_tip = 1;
-}
+message MempoolSubscriptionRequest {}
 
 // Event from the mempool.
 message MempoolEvent {