feat(network): add local-tx-monitor server agent, align codec with the wire protocol

pallas-hardano/src/display/haskell_display.rs

@@ -681,8 +681,8 @@ impl HaskellDisplay for ValidityInterval {
     fn to_haskell_str(&self) -> String {
         format!(
             "(ValidityInterval {{invalidBefore = {}, invalidHereafter = {}}})",
-            &self.invalid_before.as_slot_no(),
-            &self.invalid_hereafter.as_slot_no()
+            self.invalid_before.as_slot_no(),
+            self.invalid_hereafter.as_slot_no()
         )
     }
 }

pallas-math/src/math.rs

@@ -441,8 +441,8 @@ mod tests {
                 (&one - &threshold_b).to_string(),
                 expected_threshold_b,
                 "(1 - f) *** b failed to match! - (1 - f)={}, b={}",
-                &c,
-                &b
+                c,
+                b
             );
 
             // do Taylor approximation for
@@ -467,14 +467,14 @@ mod tests {
             if a < threshold && res.estimation != ExpOrdering::LT {
                 panic!(
                     "wrong result should be leader {} should be more like {}",
-                    &temp, threshold
+                    temp, threshold
                 );
             }
 
             if a >= threshold && res.estimation != ExpOrdering::GT {
                 panic!(
                     "wrong result should not be leader {} should be more like {}",
-                    &temp, threshold
+                    temp, threshold
                 );
             }
 

pallas-network/src/facades.rs

@@ -559,6 +559,8 @@ pub struct NodeServer {
     pub statequery: localstate::Server,
     /// Local-tx-submission server.
     pub localtxsubmission: localtxsubmission::Server,
+    /// Local-tx-monitor server.
+    pub txmonitor: txmonitor::Server,
     accepted_address: Option<UnixSocketAddr>,
     accpeted_version: Option<(VersionNumber, n2c::VersionData)>,
 }
@@ -573,11 +575,13 @@ impl NodeServer {
         let cs_channel = plexer.subscribe_server(PROTOCOL_N2C_CHAIN_SYNC);
         let sq_channel = plexer.subscribe_server(PROTOCOL_N2C_STATE_QUERY);
         let localtx_channel = plexer.subscribe_server(PROTOCOL_N2C_TX_SUBMISSION);
+        let txmonitor_channel = plexer.subscribe_server(PROTOCOL_N2C_TX_MONITOR);
 
         let server_hs = handshake::Server::<n2c::VersionData>::new(hs_channel);
         let server_cs = chainsync::N2CServer::new(cs_channel);
         let server_sq = localstate::Server::new(sq_channel);
         let server_localtx = localtxsubmission::Server::new(localtx_channel);
+        let server_txmonitor = txmonitor::Server::new(txmonitor_channel);
 
         let plexer = plexer.spawn();
 
@@ -587,6 +591,7 @@ impl NodeServer {
             chainsync: server_cs,
             statequery: server_sq,
             localtxsubmission: server_localtx,
+            txmonitor: server_txmonitor,
             accepted_address: None,
             accpeted_version: None,
         }
@@ -636,6 +641,11 @@ impl NodeServer {
         &mut self.localtxsubmission
     }
 
+    /// Get mutable access to the local-tx-monitor server.
+    pub fn txmonitor(&mut self) -> &mut txmonitor::Server {
+        &mut self.txmonitor
+    }
+
     /// Remote address of the accepted local client.
     pub fn accepted_address(&self) -> Option<&UnixSocketAddr> {
         self.accepted_address.as_ref()

pallas-network/src/miniprotocols/txmonitor/client.rs

@@ -6,7 +6,7 @@ use crate::multiplexer;
 
 /// Errors produced by the tx-monitor client agent.
 #[derive(Error, Debug)]
-pub enum Error {
+pub enum ClientError {
     /// Tried to receive while we hold agency.
     #[error("attempted to receive message while agency is ours")]
     AgencyIsOurs,
@@ -57,135 +57,156 @@ impl Client {
         }
     }
 
-    fn assert_agency_is_ours(&self) -> Result<(), Error> {
+    fn assert_agency_is_ours(&self) -> Result<(), ClientError> {
         if !self.has_agency() {
-            Err(Error::AgencyIsTheirs)
+            Err(ClientError::AgencyIsTheirs)
         } else {
             Ok(())
         }
     }
 
-    fn assert_agency_is_theirs(&self) -> Result<(), Error> {
+    fn assert_agency_is_theirs(&self) -> Result<(), ClientError> {
         if self.has_agency() {
-            Err(Error::AgencyIsOurs)
+            Err(ClientError::AgencyIsOurs)
         } else {
             Ok(())
         }
     }
 
-    fn assert_outbound_state(&self, msg: &Message) -> Result<(), Error> {
+    fn assert_outbound_state(&self, msg: &Message) -> Result<(), ClientError> {
         match (&self.0, msg) {
             (State::Idle, Message::Acquire) => Ok(()),
             (State::Idle, Message::Done) => Ok(()),
-            (State::Acquired, Message::Acquire) => Ok(()),
+            (State::Acquired, Message::Acquire | Message::AwaitAcquire) => Ok(()),
             (State::Acquired, Message::RequestHasTx(..)) => Ok(()),
             (State::Acquired, Message::RequestNextTx) => Ok(()),
             (State::Acquired, Message::RequestSizeAndCapacity) => Ok(()),
-            _ => Err(Error::InvalidOutbound),
+            (State::Acquired, Message::RequestGetMeasures) => Ok(()),
+            (State::Acquired, Message::Release) => Ok(()),
+            _ => Err(ClientError::InvalidOutbound),
         }
     }
 
-    fn assert_inbound_state(&self, msg: &Message) -> Result<(), Error> {
+    fn assert_inbound_state(&self, msg: &Message) -> Result<(), ClientError> {
         match (&self.0, msg) {
             (State::Acquiring, Message::Acquired(..)) => Ok(()),
             (State::Busy, Message::ResponseHasTx(..)) => Ok(()),
             (State::Busy, Message::ResponseNextTx(..)) => Ok(()),
             (State::Busy, Message::ResponseSizeAndCapacity(..)) => Ok(()),
-            _ => Err(Error::InvalidInbound),
+            (State::Busy, Message::ResponseGetMeasures(..)) => Ok(()),
+            _ => Err(ClientError::InvalidInbound),
         }
     }
 
     /// Low-level send.
-    pub async fn send_message(&mut self, msg: &Message) -> Result<(), Error> {
+    pub async fn send_message(&mut self, msg: &Message) -> Result<(), ClientError> {
         self.assert_agency_is_ours()?;
         self.assert_outbound_state(msg)?;
-        self.1.send_msg_chunks(msg).await.map_err(Error::Plexer)?;
+        self.1
+            .send_msg_chunks(msg)
+            .await
+            .map_err(ClientError::Plexer)?;
 
         Ok(())
     }
 
     /// Low-level receive.
-    pub async fn recv_message(&mut self) -> Result<Message, Error> {
+    pub async fn recv_message(&mut self) -> Result<Message, ClientError> {
         self.assert_agency_is_theirs()?;
-        let msg = self.1.recv_full_msg().await.map_err(Error::Plexer)?;
+        let msg = self.1.recv_full_msg().await.map_err(ClientError::Plexer)?;
         self.assert_inbound_state(&msg)?;
 
         Ok(msg)
     }
 
-    async fn send_acquire(&mut self) -> Result<(), Error> {
+    async fn send_acquire(&mut self) -> Result<(), ClientError> {
         let msg = Message::Acquire;
         self.send_message(&msg).await?;
         self.0 = State::Acquiring;
 
         Ok(())
     }
 
-    async fn recv_while_acquiring(&mut self) -> Result<Slot, Error> {
+    async fn recv_while_acquiring(&mut self) -> Result<Slot, ClientError> {
         match self.recv_message().await? {
             Message::Acquired(slot) => {
                 self.0 = State::Acquired;
                 Ok(slot)
             }
-            _ => Err(Error::InvalidInbound),
+            _ => Err(ClientError::InvalidInbound),
         }
     }
 
     /// Acquire a fresh mempool snapshot and return the slot it was taken at.
-    pub async fn acquire(&mut self) -> Result<Slot, Error> {
+    pub async fn acquire(&mut self) -> Result<Slot, ClientError> {
         self.send_acquire().await?;
         self.recv_while_acquiring().await
     }
 
-    async fn send_request_has_tx(&mut self, id: TxId) -> Result<(), Error> {
+    async fn send_await_acquire(&mut self) -> Result<(), ClientError> {
+        let msg = Message::AwaitAcquire;
+        self.send_message(&msg).await?;
+        self.0 = State::Acquiring;
+
+        Ok(())
+    }
+
+    /// Release the current snapshot and block until a changed one can be
+    /// acquired. Returns the slot the new snapshot was taken at.
+    pub async fn await_acquire(&mut self) -> Result<Slot, ClientError> {
+        self.send_await_acquire().await?;
+        self.recv_while_acquiring().await
+    }
+
+    async fn send_request_has_tx(&mut self, id: TxId) -> Result<(), ClientError> {
         let msg = Message::RequestHasTx(id);
         self.send_message(&msg).await?;
         self.0 = State::Busy;
 
         Ok(())
     }
 
-    async fn recv_while_requesting_has_tx(&mut self) -> Result<bool, Error> {
+    async fn recv_while_requesting_has_tx(&mut self) -> Result<bool, ClientError> {
         match self.recv_message().await? {
             Message::ResponseHasTx(x) => {
                 self.0 = State::Acquired;
                 Ok(x)
             }
-            _ => Err(Error::InvalidInbound),
+            _ => Err(ClientError::InvalidInbound),
         }
     }
 
     /// Ask whether a transaction with the given id is in the current snapshot.
-    pub async fn query_has_tx(&mut self, id: TxId) -> Result<bool, Error> {
+    pub async fn query_has_tx(&mut self, id: TxId) -> Result<bool, ClientError> {
         self.send_request_has_tx(id).await?;
         self.recv_while_requesting_has_tx().await
     }
 
-    async fn send_request_next_tx(&mut self) -> Result<(), Error> {
+    async fn send_request_next_tx(&mut self) -> Result<(), ClientError> {
         let msg = Message::RequestNextTx;
         self.send_message(&msg).await?;
         self.0 = State::Busy;
 
         Ok(())
     }
 
-    async fn recv_while_requesting_next_tx(&mut self) -> Result<Option<Tx>, Error> {
+    async fn recv_while_requesting_next_tx(&mut self) -> Result<Option<Tx>, ClientError> {
         match self.recv_message().await? {
             Message::ResponseNextTx(x) => {
                 self.0 = State::Acquired;
                 Ok(x)
             }
-            _ => Err(Error::InvalidInbound),
+            _ => Err(ClientError::InvalidInbound),
         }
     }
 
     /// Iterate to the next transaction in the snapshot.
-    pub async fn query_next_tx(&mut self) -> Result<Option<Tx>, Error> {
+    pub async fn query_next_tx(&mut self) -> Result<Option<Tx>, ClientError> {
         self.send_request_next_tx().await?;
         self.recv_while_requesting_next_tx().await
     }
 
-    async fn send_request_size_and_capacity(&mut self) -> Result<(), Error> {
+    async fn send_request_size_and_capacity(&mut self) -> Result<(), ClientError> {
         let msg = Message::RequestSizeAndCapacity;
         self.send_message(&msg).await?;
         self.0 = State::Busy;
@@ -195,28 +216,61 @@ impl Client {
 
     async fn recv_while_requesting_size_and_capacity(
         &mut self,
-    ) -> Result<MempoolSizeAndCapacity, Error> {
+    ) -> Result<MempoolSizeAndCapacity, ClientError> {
         match self.recv_message().await? {
             Message::ResponseSizeAndCapacity(x) => {
                 self.0 = State::Acquired;
                 Ok(x)
             }
-            _ => Err(Error::InvalidInbound),
+            _ => Err(ClientError::InvalidInbound),
         }
     }
 
     /// Ask for the mempool's current size and capacity.
-    pub async fn query_size_and_capacity(&mut self) -> Result<MempoolSizeAndCapacity, Error> {
+    pub async fn query_size_and_capacity(&mut self) -> Result<MempoolSizeAndCapacity, ClientError> {
         self.send_request_size_and_capacity().await?;
         self.recv_while_requesting_size_and_capacity().await
     }
 
+    async fn send_request_measures(&mut self) -> Result<(), ClientError> {
+        let msg = Message::RequestGetMeasures;
+        self.send_message(&msg).await?;
+        self.0 = State::Busy;
+
+        Ok(())
+    }
+
+    async fn recv_while_requesting_measures(&mut self) -> Result<MempoolMeasures, ClientError> {
+        match self.recv_message().await? {
+            Message::ResponseGetMeasures(x) => {
+                self.0 = State::Acquired;
+                Ok(x)
+            }
+            _ => Err(ClientError::InvalidInbound),
+        }
+    }
+
+    /// Ask for the mempool's current measures (node-to-client v20+).
+    pub async fn query_measures(&mut self) -> Result<MempoolMeasures, ClientError> {
+        self.send_request_measures().await?;
+        self.recv_while_requesting_measures().await
+    }
+
     /// Release the current snapshot and return to the idle state.
-    pub async fn release(&mut self) -> Result<(), Error> {
+    pub async fn release(&mut self) -> Result<(), ClientError> {
         let msg = Message::Release;
         self.send_message(&msg).await?;
         self.0 = State::Idle;
 
         Ok(())
     }
+
+    /// Terminate the protocol.
+    pub async fn done(&mut self) -> Result<(), ClientError> {
+        let msg = Message::Done;
+        self.send_message(&msg).await?;
+        self.0 = State::Done;
+
+        Ok(())
+    }
 }