From 765146fa0bad544afc45515e7484ecf5fc9e67d1 Mon Sep 17 00:00:00 2001 From: Javier Acosta Date: Sun, 19 Jul 2026 13:42:43 -0300 Subject: [PATCH 1/3] chore: fix clippy lints for Rust 1.97 Rust 1.97 stable introduces lints that fail the workspace clippy job: - for_kv_map in pallas-validate phase1 (alonzo, babbage, conway, shelley_ma): iterate map keys directly instead of destructuring unused values - useless_borrows_in_formatting in pallas-math tests and pallas-hardano haskell_display: drop redundant references in format arguments Mechanical fixes only, no behavior change. Verified clean under both 1.93 and 1.97 clippy. Co-Authored-By: Claude Fable 5 --- pallas-hardano/src/display/haskell_display.rs | 4 ++-- pallas-math/src/math.rs | 8 ++++---- pallas-validate/src/phase1/alonzo.rs | 2 +- pallas-validate/src/phase1/babbage.rs | 2 +- pallas-validate/src/phase1/conway.rs | 2 +- pallas-validate/src/phase1/shelley_ma.rs | 2 +- 6 files changed, 10 insertions(+), 10 deletions(-) diff --git a/pallas-hardano/src/display/haskell_display.rs b/pallas-hardano/src/display/haskell_display.rs index cb382a7d..e3dec42f 100644 --- a/pallas-hardano/src/display/haskell_display.rs +++ b/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() ) } } diff --git a/pallas-math/src/math.rs b/pallas-math/src/math.rs index bd35404e..2bdc0486 100644 --- a/pallas-math/src/math.rs +++ b/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 ); } diff --git a/pallas-validate/src/phase1/alonzo.rs b/pallas-validate/src/phase1/alonzo.rs index 3e8283d9..e317d56f 100644 --- a/pallas-validate/src/phase1/alonzo.rs +++ b/pallas-validate/src/phase1/alonzo.rs @@ -909,7 +909,7 @@ fn check_minting(tx_body: &TransactionBody, mtx: &Tx) -> ValidationResult { None => Vec::new(), Some(plutus_v1_script_wits) => plutus_v1_script_wits.clone(), }; - for (policy, _) in minted_value.iter() { + for policy in minted_value.keys() { if native_script_wits .iter() .all(|native_script| compute_native_script_hash(native_script) != *policy) diff --git a/pallas-validate/src/phase1/babbage.rs b/pallas-validate/src/phase1/babbage.rs index 9144bbe6..a84e1461 100644 --- a/pallas-validate/src/phase1/babbage.rs +++ b/pallas-validate/src/phase1/babbage.rs @@ -471,7 +471,7 @@ fn check_minting(tx_body: &TransactionBody, mtx: &Tx, utxos: &UTxOs) -> Validati dbg!(&all_scripts_wits); - for (policy, _) in minted_value.iter() { + for policy in minted_value.keys() { if !all_scripts_wits.contains(policy) { return Err(PostAlonzo(MintingLacksPolicy(*policy))); } diff --git a/pallas-validate/src/phase1/conway.rs b/pallas-validate/src/phase1/conway.rs index f215e526..1f8f2617 100644 --- a/pallas-validate/src/phase1/conway.rs +++ b/pallas-validate/src/phase1/conway.rs @@ -603,7 +603,7 @@ fn check_minting(tx_body: &TransactionBody, mtx: &Tx, utxos: &UTxOs) -> Validati .chain(ref_scripts) .collect(); - for (policy, _) in minted_value.iter() { + for policy in minted_value.keys() { if !all_scripts_wits.contains(policy) { return Err(PostAlonzo(MintingLacksPolicy(*policy))); } diff --git a/pallas-validate/src/phase1/shelley_ma.rs b/pallas-validate/src/phase1/shelley_ma.rs index 2cf15e31..e6068e2a 100644 --- a/pallas-validate/src/phase1/shelley_ma.rs +++ b/pallas-validate/src/phase1/shelley_ma.rs @@ -386,7 +386,7 @@ fn check_minting(tx_body: &TransactionBody, mtx: &Tx) -> ValidationResult { .map(|x| x.clone().unwrap()) .collect(), }; - for (policy, _) in minted_value.iter() { + for policy in minted_value.keys() { if native_script_wits .iter() .all(|script| compute_script_hash(script) != *policy) From 0448ab7eb40b67a88079fd4c9c401a7785de9c85 Mon Sep 17 00:00:00 2001 From: Javier Acosta Date: Sun, 19 Jul 2026 13:43:05 -0300 Subject: [PATCH 2/3] fix(network): align txmonitor codec with the on-the-wire protocol The local-tx-monitor codec diverged from the authoritative ouroboros-network codec in three ways that made interop with real peers (cardano-node, ogmios) impossible: - MsgAwaitAcquire was guessed as label 4, which does not exist in the spec. The wire shares label 1 with MsgAcquire and peers disambiguate by protocol state (Idle = acquire, Acquired = await re-acquire). The AwaitAcquire variant now encodes as 1; a stateless decode always yields Acquire and agents map it back based on their state. - TxId was modeled as a text string. The wire format is the hard-fork-combinator GenTxId: an era-wrapped [era, hash-bytes] pair. Clients such as ogmios treat the era wrapper as significant and probe every plausible era for the same hash. - MsgGetMeasures (11) / MsgReplyGetMeasures (12), part of the protocol since node-to-client v20, were not modeled at all, so receiving them killed the decode. The client agent is repaired along the way: release() was rejected by its own outbound-state assertion, Release/AwaitAcquire/Done were missing from the assert tables, and await_acquire(), query_measures() and done() are added. The error type is renamed Error -> ClientError, matching the chainsync ClientError/ServerError convention, to make room for the server agent. BREAKING: TxId changes shape and the client error type is renamed. No working code can regress on the wire format: the previous encoding was never accepted by real peers. Co-Authored-By: Claude Fable 5 --- .../src/miniprotocols/txmonitor/client.rs | 116 +++++++++++++----- .../src/miniprotocols/txmonitor/codec.rs | 108 ++++++++++++++-- .../src/miniprotocols/txmonitor/protocol.rs | 41 ++++++- 3 files changed, 223 insertions(+), 42 deletions(-) diff --git a/pallas-network/src/miniprotocols/txmonitor/client.rs b/pallas-network/src/miniprotocols/txmonitor/client.rs index 7515834e..7b8cb8f8 100644 --- a/pallas-network/src/miniprotocols/txmonitor/client.rs +++ b/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,63 +57,69 @@ 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 { + pub async fn recv_message(&mut self) -> Result { 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; @@ -121,23 +127,38 @@ impl Client { Ok(()) } - async fn recv_while_acquiring(&mut self) -> Result { + async fn recv_while_acquiring(&mut self) -> Result { 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 { + pub async fn acquire(&mut self) -> Result { 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 { + 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; @@ -145,23 +166,23 @@ impl Client { Ok(()) } - async fn recv_while_requesting_has_tx(&mut self) -> Result { + async fn recv_while_requesting_has_tx(&mut self) -> Result { 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 { + pub async fn query_has_tx(&mut self, id: TxId) -> Result { 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; @@ -169,23 +190,23 @@ impl Client { Ok(()) } - async fn recv_while_requesting_next_tx(&mut self) -> Result, Error> { + async fn recv_while_requesting_next_tx(&mut self) -> Result, 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, Error> { + pub async fn query_next_tx(&mut self) -> Result, 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 { + ) -> Result { 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 { + pub async fn query_size_and_capacity(&mut self) -> Result { 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 { + 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 { + 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(()) + } } diff --git a/pallas-network/src/miniprotocols/txmonitor/codec.rs b/pallas-network/src/miniprotocols/txmonitor/codec.rs index 02d482a4..79058a1f 100644 --- a/pallas-network/src/miniprotocols/txmonitor/codec.rs +++ b/pallas-network/src/miniprotocols/txmonitor/codec.rs @@ -14,6 +14,10 @@ impl Encode<()> for Message { Message::Acquire => { e.array(1)?.u16(1)?; } + // shares label 1 with Acquire; peers disambiguate by state + Message::AwaitAcquire => { + e.array(1)?.u16(1)?; + } Message::Acquired(slot) => { e.array(2)?.u16(2)?; e.encode(slot)?; @@ -21,11 +25,6 @@ impl Encode<()> for Message { Message::Release => { e.array(1)?.u16(3)?; } - // TODO: confirm if this is valid, I'm just assuming that label 4 is AwaitAcquire, can't - // find the specs - Message::AwaitAcquire => { - e.array(1)?.u16(4)?; - } Message::RequestNextTx => { e.array(1)?.u16(5)?; } @@ -54,6 +53,20 @@ impl Encode<()> for Message { e.encode(sz.size_in_bytes)?; e.encode(sz.number_of_txs)?; } + Message::RequestGetMeasures => { + e.array(1)?.u16(11)?; + } + Message::ResponseGetMeasures(measures) => { + e.array(3)?.u16(12)?; + e.encode(measures.tx_count)?; + e.map(measures.measures.len() as u64)?; + for (name, sc) in &measures.measures { + e.encode(name)?; + e.array(2)?; + e.encode(sc.size)?; + e.encode(sc.capacity)?; + } + } } Ok(()) @@ -70,15 +83,14 @@ impl<'b> Decode<'b, ()> for Message { match label { 0 => Ok(Message::Done), + // label 1 is Acquire from Idle and AwaitAcquire from Acquired; a + // stateless decode can't tell them apart, agents disambiguate 1 => Ok(Message::Acquire), 2 => { let slot = d.decode()?; Ok(Message::Acquired(slot)) } 3 => Ok(Message::Release), - // TODO: confirm if this is valid, I'm just assuming that label 4 is AwaitAcquire, can't - // find the specs - 4 => Ok(Message::AwaitAcquire), 5 => Ok(Message::RequestNextTx), 6 => match d.datatype() { Ok(datatype) => match datatype { @@ -112,6 +124,28 @@ impl<'b> Decode<'b, ()> for Message { number_of_txs, })) } + 11 => Ok(Message::RequestGetMeasures), + 12 => { + let tx_count = d.decode()?; + let len = d + .map()? + .ok_or_else(|| decode::Error::message("expected definite-length map"))?; + + let mut measures = Vec::with_capacity(len as usize); + + for _ in 0..len { + let name: MeasureName = d.decode()?; + d.array()?; + let size = d.decode()?; + let capacity = d.decode()?; + measures.push((name, SizeAndCapacity { size, capacity })); + } + + Ok(Message::ResponseGetMeasures(MempoolMeasures { + tx_count, + measures, + })) + } _ => Err(decode::Error::message("can't decode Message")), } } @@ -119,6 +153,8 @@ impl<'b> Decode<'b, ()> for Message { #[cfg(test)] pub mod tests { + use super::super::protocol::*; + const EXAMPLE_RESPONSE_NEXT_TX_WITH_DATA: &str = "82068205d81859013184a5008282582003e4aea27ebacf5f50b10ac60cc84deba96569ce8a47fdf9199998d1fd16ec0601825820eebf8249544b7eefa7839510dfd58a7ed420f2254bd3bf632baea8cd0928b00102018182583901b98f57f569aba4cffc4d9c791f099374e9403ed5e2cb614eab25b78278b1312c2c271d260db425b8b9847ab142b395b4598d3c0b383aa696821a00924172a1581c09f2d4e4a5c3662f4c1e6a7d9600e9605279dbdcedb22d4507cb6e75a1435350461a0422bb35021a00029f3d031a063ec6470800a100818258208293ac2260e28a07657f77087d1d7ff5e3ced29ff4385abf60a9546e2bcbc04a5840d69ce3a8f9713513a9baf473c1be08fd17d1a85df2881dc107fb1f68ce02c8e7adcf1c91bce7fb58868908f7ac47310a8e97d95780beadcfd8493bebbb914d0df5f6"; #[test] @@ -133,13 +169,67 @@ pub mod tests { unreachable!(); } } + #[test] fn test_empty_next_tx_response() { let bytes = vec![129, 6]; let msg: super::Message = pallas_codec::minicbor::decode(&bytes).unwrap(); if let super::Message::ResponseNextTx(None) = msg { - assert_eq!(0u64, 0u64); + } else { + unreachable!(); + } + } + + #[test] + fn test_has_tx_request_roundtrip() { + let id: TxId = (5, vec![0xab; 32].into()); + let msg = super::Message::RequestHasTx(id.clone()); + + let bytes = pallas_codec::minicbor::to_vec(&msg).unwrap(); + + // [7, [era, h'ab...ab']] + let mut expected = vec![0x82, 0x07, 0x82, 0x05, 0x58, 0x20]; + expected.extend(std::iter::repeat_n(0xab, 32)); + assert_eq!(bytes, expected); + + let decoded: super::Message = pallas_codec::minicbor::decode(&bytes).unwrap(); + + if let super::Message::RequestHasTx(decoded_id) = decoded { + assert_eq!(decoded_id, id); + } else { + unreachable!(); + } + } + + #[test] + fn test_await_acquire_encodes_as_acquire() { + let bytes = pallas_codec::minicbor::to_vec(super::Message::AwaitAcquire).unwrap(); + assert_eq!(bytes, vec![0x81, 0x01]); + + let decoded: super::Message = pallas_codec::minicbor::decode(&bytes).unwrap(); + assert!(matches!(decoded, super::Message::Acquire)); + } + + #[test] + fn test_get_measures_roundtrip() { + let measures = MempoolMeasures { + tx_count: 2, + measures: vec![( + "transaction_bytes".to_string(), + SizeAndCapacity { + size: 1234, + capacity: 178176, + }, + )], + }; + + let msg = super::Message::ResponseGetMeasures(measures.clone()); + let bytes = pallas_codec::minicbor::to_vec(&msg).unwrap(); + let decoded: super::Message = pallas_codec::minicbor::decode(&bytes).unwrap(); + + if let super::Message::ResponseGetMeasures(decoded_measures) = decoded { + assert_eq!(decoded_measures, measures); } else { unreachable!(); } diff --git a/pallas-network/src/miniprotocols/txmonitor/protocol.rs b/pallas-network/src/miniprotocols/txmonitor/protocol.rs index 242ecd63..60cce50a 100644 --- a/pallas-network/src/miniprotocols/txmonitor/protocol.rs +++ b/pallas-network/src/miniprotocols/txmonitor/protocol.rs @@ -2,14 +2,23 @@ use pallas_codec::utils::TagWrap; /// Absolute slot number used to tag a mempool snapshot. pub type Slot = u64; -/// Transaction id rendered as a string (hex of the tx hash). -pub type TxId = String; /// Era number, as carried in the multi-era transaction wrapper. pub type Era = u8; +/// Raw bytes of a transaction hash. +pub type TxIdBytes = pallas_codec::utils::Bytes; +/// `(era, tx-hash-bytes)` — era-wrapped transaction id, mirroring the +/// hard-fork-combinator `GenTxId` encoding used by node-to-client peers. +/// +/// Note that peers (e.g. cardano-node, ogmios) treat the era wrapper as +/// significant for equality: the same hash wrapped in different eras is a +/// different `GenTxId`. Clients typically probe every plausible era. +pub type TxId = (Era, TxIdBytes); /// Raw CBOR bytes of a transaction body. pub type TxBody = pallas_codec::utils::Bytes; /// `(era, cbor-tag-24-wrapped body)` — the canonical mempool transaction shape. pub type Tx = (Era, TagWrap); +/// Name of a mempool measure reported by `ResponseGetMeasures`. +pub type MeasureName = String; /// Tx-monitor state-machine state. #[derive(Debug, PartialEq, Eq, Clone)] @@ -37,12 +46,36 @@ pub struct MempoolSizeAndCapacity { pub number_of_txs: u32, } +/// Current size and maximum capacity of a single mempool measure. +#[derive(Debug, PartialEq, Eq, Clone)] +pub struct SizeAndCapacity { + /// Current size of the measure. + pub size: u64, + /// Maximum capacity of the measure. + pub capacity: u64, +} + +/// Mempool measures reported by `ResponseGetMeasures` (node-to-client v20+). +#[derive(Debug, PartialEq, Eq, Clone)] +pub struct MempoolMeasures { + /// Number of transactions currently in the mempool. + pub tx_count: u32, + /// Named measures (e.g. transaction bytes, execution units) with their + /// current size and capacity. + pub measures: Vec<(MeasureName, SizeAndCapacity)>, +} + /// Tx-monitor protocol message. #[derive(Debug, Clone)] pub enum Message { /// Client → server: acquire the current mempool snapshot (non-blocking). Acquire, /// Client → server: acquire the next snapshot (blocks until it changes). + /// + /// On the wire this shares label `1` with [`Message::Acquire`]; peers + /// disambiguate by protocol state (`Idle` → acquire, `Acquired` → await + /// re-acquire). A stateless decode therefore always yields + /// [`Message::Acquire`]; agents map it back based on their state. AwaitAcquire, /// Server → client: snapshot acquired at the given slot. Acquired(Slot), @@ -52,12 +85,16 @@ pub enum Message { RequestNextTx, /// Client → server: ask for mempool size and capacity. RequestSizeAndCapacity, + /// Client → server: ask for mempool measures (node-to-client v20+). + RequestGetMeasures, /// Server → client: answer to [`Message::RequestHasTx`]. ResponseHasTx(bool), /// Server → client: next transaction (or `None` if the iteration is exhausted). ResponseNextTx(Option), /// Server → client: answer to [`Message::RequestSizeAndCapacity`]. ResponseSizeAndCapacity(MempoolSizeAndCapacity), + /// Server → client: answer to [`Message::RequestGetMeasures`]. + ResponseGetMeasures(MempoolMeasures), /// Client → server: release the current snapshot. Release, /// Client → server: terminate the protocol. From 79dbeeba3a5d3894d06f87c2e894c21a8ec2619c Mon Sep 17 00:00:00 2001 From: Javier Acosta Date: Sun, 19 Jul 2026 13:43:05 -0300 Subject: [PATCH 3/3] feat(network): add txmonitor server agent and wire it into NodeServer Adds the server side of the local-tx-monitor miniprotocol, mirroring the localstate server pattern: recv_while_idle / recv_while_acquired drive the session and typed send_* replies cover NextTx, HasTx, GetSizes and GetMeasures. Wire label 1 received in the Acquired state surfaces as ClientQueryRequest::AwaitAcquire. NodeServer now subscribes PROTOCOL_N2C_TX_MONITOR and exposes the agent via txmonitor(). Without the subscription, the multiplexer kills the whole session as soon as a client opens protocol 9 - which ogmios does on every connection, making pallas-based N2C servers (e.g. dolos) unusable behind it. The server accepts GetMeasures regardless of the negotiated handshake version (the spec gates it at v20+); pallas miniprotocol agents are version-agnostic by design, so the server errs on the lenient side. Includes a full client/server pair test over a unix socket exercising acquire, iteration, has-tx, sizes, measures, blocking re-acquire, release and termination. Co-Authored-By: Claude Fable 5 --- pallas-network/src/facades.rs | 10 + .../src/miniprotocols/txmonitor/mod.rs | 2 + .../src/miniprotocols/txmonitor/server.rs | 226 ++++++++++++++++++ pallas-network/tests/protocols.rs | 179 +++++++++++++- 4 files changed, 416 insertions(+), 1 deletion(-) create mode 100644 pallas-network/src/miniprotocols/txmonitor/server.rs diff --git a/pallas-network/src/facades.rs b/pallas-network/src/facades.rs index 713cd169..3b240257 100644 --- a/pallas-network/src/facades.rs +++ b/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, 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::::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() diff --git a/pallas-network/src/miniprotocols/txmonitor/mod.rs b/pallas-network/src/miniprotocols/txmonitor/mod.rs index 9bbd8cc0..bc68ddef 100644 --- a/pallas-network/src/miniprotocols/txmonitor/mod.rs +++ b/pallas-network/src/miniprotocols/txmonitor/mod.rs @@ -1,6 +1,8 @@ mod client; mod codec; mod protocol; +mod server; pub use client::*; pub use protocol::*; +pub use server::*; diff --git a/pallas-network/src/miniprotocols/txmonitor/server.rs b/pallas-network/src/miniprotocols/txmonitor/server.rs new file mode 100644 index 00000000..6dfdecca --- /dev/null +++ b/pallas-network/src/miniprotocols/txmonitor/server.rs @@ -0,0 +1,226 @@ +use std::fmt::Debug; +use thiserror::*; + +use super::protocol::*; +use crate::multiplexer; + +/// Errors produced by the tx-monitor server agent. +#[derive(Error, Debug)] +pub enum Error { + /// Tried to receive while we hold agency. + #[error("attempted to receive message while agency is ours")] + AgencyIsOurs, + + /// Tried to send while the peer holds agency. + #[error("attempted to send message while agency is theirs")] + AgencyIsTheirs, + + /// Inbound message is not valid for the current state. + #[error("inbound message is not valid for current state")] + InvalidInbound, + + /// Outbound message is not valid for the current state. + #[error("outbound message is not valid for current state")] + InvalidOutbound, + + /// Underlying multiplexer error. + #[error("error while sending or receiving data through the channel")] + Plexer(multiplexer::Error), +} + +/// Request received from the client while a snapshot is acquired. +#[derive(Debug)] +pub enum ClientQueryRequest { + /// Drop the current snapshot and acquire the next one (blocking until the + /// mempool changes). + AwaitAcquire, + /// Iterate to the next transaction in the snapshot. + NextTx, + /// Ask whether a specific transaction is in the snapshot. + HasTx(TxId), + /// Ask for mempool size and capacity. + GetSizes, + /// Ask for mempool measures (node-to-client v20+). + GetMeasures, + /// Release the current snapshot. + Release, +} + +/// Tx-monitor server agent. +pub struct Server(State, multiplexer::ChannelBuffer); + +impl Server { + /// Build a server over a freshly subscribed agent channel. + pub fn new(channel: multiplexer::AgentChannel) -> Self { + Self(State::Idle, multiplexer::ChannelBuffer::new(channel)) + } + + /// Current state-machine state. + pub fn state(&self) -> &State { + &self.0 + } + + /// True if the protocol has terminated. + pub fn is_done(&self) -> bool { + self.0 == State::Done + } + + fn has_agency(&self) -> bool { + matches!(self.state(), State::Acquiring | State::Busy) + } + + fn assert_agency_is_ours(&self) -> Result<(), Error> { + if !self.has_agency() { + Err(Error::AgencyIsTheirs) + } else { + Ok(()) + } + } + + fn assert_agency_is_theirs(&self) -> Result<(), Error> { + if self.has_agency() { + Err(Error::AgencyIsOurs) + } else { + Ok(()) + } + } + + fn assert_outbound_state(&self, msg: &Message) -> Result<(), Error> { + match (&self.0, msg) { + (State::Acquiring, Message::Acquired(..)) => Ok(()), + (State::Busy, Message::ResponseNextTx(..)) => Ok(()), + (State::Busy, Message::ResponseHasTx(..)) => Ok(()), + (State::Busy, Message::ResponseSizeAndCapacity(..)) => Ok(()), + (State::Busy, Message::ResponseGetMeasures(..)) => Ok(()), + _ => Err(Error::InvalidOutbound), + } + } + + fn assert_inbound_state(&self, msg: &Message) -> Result<(), Error> { + match (&self.0, msg) { + (State::Idle, Message::Acquire) => Ok(()), + (State::Idle, Message::Done) => Ok(()), + // wire label 1 in the acquired state means await re-acquire + (State::Acquired, Message::Acquire | Message::AwaitAcquire) => Ok(()), + (State::Acquired, Message::RequestNextTx) => Ok(()), + (State::Acquired, Message::RequestHasTx(..)) => Ok(()), + (State::Acquired, Message::RequestSizeAndCapacity) => Ok(()), + (State::Acquired, Message::RequestGetMeasures) => Ok(()), + (State::Acquired, Message::Release) => Ok(()), + _ => Err(Error::InvalidInbound), + } + } + + /// Low-level send. + pub async fn send_message(&mut self, msg: &Message) -> Result<(), Error> { + self.assert_agency_is_ours()?; + self.assert_outbound_state(msg)?; + self.1.send_msg_chunks(msg).await.map_err(Error::Plexer)?; + + Ok(()) + } + + /// Low-level receive. + pub async fn recv_message(&mut self) -> Result { + self.assert_agency_is_theirs()?; + let msg = self.1.recv_full_msg().await.map_err(Error::Plexer)?; + self.assert_inbound_state(&msg)?; + + Ok(msg) + } + + /// Confirm the pending acquire, tagging the snapshot with the given slot. + pub async fn send_acquired(&mut self, slot: Slot) -> Result<(), Error> { + let msg = Message::Acquired(slot); + self.send_message(&msg).await?; + self.0 = State::Acquired; + + Ok(()) + } + + /// Reply to the pending [`ClientQueryRequest::NextTx`] request. + pub async fn send_next_tx(&mut self, tx: Option) -> Result<(), Error> { + let msg = Message::ResponseNextTx(tx); + self.send_message(&msg).await?; + self.0 = State::Acquired; + + Ok(()) + } + + /// Reply to the pending [`ClientQueryRequest::HasTx`] request. + pub async fn send_has_tx(&mut self, has: bool) -> Result<(), Error> { + let msg = Message::ResponseHasTx(has); + self.send_message(&msg).await?; + self.0 = State::Acquired; + + Ok(()) + } + + /// Reply to the pending [`ClientQueryRequest::GetSizes`] request. + pub async fn send_size_and_capacity( + &mut self, + sizes: MempoolSizeAndCapacity, + ) -> Result<(), Error> { + let msg = Message::ResponseSizeAndCapacity(sizes); + self.send_message(&msg).await?; + self.0 = State::Acquired; + + Ok(()) + } + + /// Reply to the pending [`ClientQueryRequest::GetMeasures`] request. + pub async fn send_measures(&mut self, measures: MempoolMeasures) -> Result<(), Error> { + let msg = Message::ResponseGetMeasures(measures); + self.send_message(&msg).await?; + self.0 = State::Acquired; + + Ok(()) + } + + /// Wait for the next request while the protocol is in the `Idle` state. + /// Returns `None` if the client terminated the protocol. + pub async fn recv_while_idle(&mut self) -> Result, Error> { + match self.recv_message().await? { + Message::Acquire => { + self.0 = State::Acquiring; + Ok(Some(())) + } + Message::Done => { + self.0 = State::Done; + Ok(None) + } + _ => Err(Error::InvalidInbound), + } + } + + /// Wait for the next request while a snapshot is acquired. + pub async fn recv_while_acquired(&mut self) -> Result { + match self.recv_message().await? { + Message::Acquire | Message::AwaitAcquire => { + self.0 = State::Acquiring; + Ok(ClientQueryRequest::AwaitAcquire) + } + Message::RequestNextTx => { + self.0 = State::Busy; + Ok(ClientQueryRequest::NextTx) + } + Message::RequestHasTx(id) => { + self.0 = State::Busy; + Ok(ClientQueryRequest::HasTx(id)) + } + Message::RequestSizeAndCapacity => { + self.0 = State::Busy; + Ok(ClientQueryRequest::GetSizes) + } + Message::RequestGetMeasures => { + self.0 = State::Busy; + Ok(ClientQueryRequest::GetMeasures) + } + Message::Release => { + self.0 = State::Idle; + Ok(ClientQueryRequest::Release) + } + _ => Err(Error::InvalidInbound), + } + } +} diff --git a/pallas-network/tests/protocols.rs b/pallas-network/tests/protocols.rs index 43541ad4..fe95745f 100644 --- a/pallas-network/tests/protocols.rs +++ b/pallas-network/tests/protocols.rs @@ -1,6 +1,6 @@ use hex::FromHex; use pallas_codec::utils::{ - AnyCbor, AnyUInt, Bytes, CborWrap, KeyValuePairs, MaybeIndefArray, Nullable, + AnyCbor, AnyUInt, Bytes, CborWrap, KeyValuePairs, MaybeIndefArray, Nullable, TagWrap, }; use pallas_crypto::hash::Hash; use pallas_network::miniprotocols::localmsgsubmission::DmqMsgRejectReason; @@ -2095,3 +2095,180 @@ pub async fn local_message_submission_server_and_client_happy_path() { tokio::try_join!(client, server).unwrap(); } + +#[cfg(unix)] +#[tokio::test] +pub async fn txmonitor_server_and_client_happy_path() { + use pallas_network::miniprotocols::txmonitor::{ + self, ClientQueryRequest as TxMonitorRequest, MempoolMeasures, MempoolSizeAndCapacity, + SizeAndCapacity, + }; + + let expected_tx: txmonitor::Tx = (5, TagWrap::new(hex::decode("deadbeef").unwrap().into())); + let expected_id: txmonitor::TxId = (5, vec![0xab; 32].into()); + + let server = tokio::spawn({ + let expected_tx = expected_tx.clone(); + let expected_id = expected_id.clone(); + + async move { + // server setup + let socket_path = Path::new("node5.socket"); + + if socket_path.exists() { + fs::remove_file(socket_path).unwrap(); + } + + let listener = UnixListener::bind(socket_path).unwrap(); + + let mut server = pallas_network::facades::NodeServer::accept(&listener, 0) + .await + .unwrap(); + + // wait for acquire request from client + + let maybe_acquire = server.txmonitor().recv_while_idle().await.unwrap(); + + assert!(maybe_acquire.is_some()); + assert_eq!(*server.txmonitor().state(), txmonitor::State::Acquiring); + + server.txmonitor().send_acquired(123).await.unwrap(); + + assert_eq!(*server.txmonitor().state(), txmonitor::State::Acquired); + + // iterate the two-entry snapshot + + match server.txmonitor().recv_while_acquired().await.unwrap() { + TxMonitorRequest::NextTx => (), + x => panic!("unexpected message from client: {x:?}"), + } + + assert_eq!(*server.txmonitor().state(), txmonitor::State::Busy); + + server + .txmonitor() + .send_next_tx(Some(expected_tx.clone())) + .await + .unwrap(); + + assert_eq!(*server.txmonitor().state(), txmonitor::State::Acquired); + + match server.txmonitor().recv_while_acquired().await.unwrap() { + TxMonitorRequest::NextTx => (), + x => panic!("unexpected message from client: {x:?}"), + } + + server.txmonitor().send_next_tx(None).await.unwrap(); + + // answer has-tx + + match server.txmonitor().recv_while_acquired().await.unwrap() { + TxMonitorRequest::HasTx(id) => assert_eq!(id, expected_id), + x => panic!("unexpected message from client: {x:?}"), + } + + server.txmonitor().send_has_tx(true).await.unwrap(); + + // answer get-sizes + + match server.txmonitor().recv_while_acquired().await.unwrap() { + TxMonitorRequest::GetSizes => (), + x => panic!("unexpected message from client: {x:?}"), + } + + server + .txmonitor() + .send_size_and_capacity(MempoolSizeAndCapacity { + capacity_in_bytes: 178176, + size_in_bytes: 1234, + number_of_txs: 2, + }) + .await + .unwrap(); + + // answer get-measures + + match server.txmonitor().recv_while_acquired().await.unwrap() { + TxMonitorRequest::GetMeasures => (), + x => panic!("unexpected message from client: {x:?}"), + } + + server + .txmonitor() + .send_measures(MempoolMeasures { + tx_count: 2, + measures: vec![( + "transaction_bytes".to_string(), + SizeAndCapacity { + size: 1234, + capacity: 178176, + }, + )], + }) + .await + .unwrap(); + + // client re-acquires (await-acquire shares the acquire wire label) + + match server.txmonitor().recv_while_acquired().await.unwrap() { + TxMonitorRequest::AwaitAcquire => (), + x => panic!("unexpected message from client: {x:?}"), + } + + assert_eq!(*server.txmonitor().state(), txmonitor::State::Acquiring); + + server.txmonitor().send_acquired(124).await.unwrap(); + + // client releases and terminates + + match server.txmonitor().recv_while_acquired().await.unwrap() { + TxMonitorRequest::Release => (), + x => panic!("unexpected message from client: {x:?}"), + } + + assert_eq!(*server.txmonitor().state(), txmonitor::State::Idle); + + let done = server.txmonitor().recv_while_idle().await.unwrap(); + assert!(done.is_none()); + assert!(server.txmonitor().is_done()); + } + }); + + let client = tokio::spawn(async move { + tokio::time::sleep(Duration::from_secs(1)).await; + + // client setup + let socket_path = "node5.socket"; + + let mut client = NodeClient::connect(socket_path, 0).await.unwrap(); + + let slot = client.monitor().acquire().await.unwrap(); + assert_eq!(slot, 123); + + let tx = client.monitor().query_next_tx().await.unwrap(); + assert_eq!(tx, Some(expected_tx)); + + let tx = client.monitor().query_next_tx().await.unwrap(); + assert_eq!(tx, None); + + let has = client.monitor().query_has_tx(expected_id).await.unwrap(); + assert!(has); + + let sizes = client.monitor().query_size_and_capacity().await.unwrap(); + assert_eq!(sizes.capacity_in_bytes, 178176); + assert_eq!(sizes.size_in_bytes, 1234); + assert_eq!(sizes.number_of_txs, 2); + + let measures = client.monitor().query_measures().await.unwrap(); + assert_eq!(measures.tx_count, 2); + assert_eq!(measures.measures.len(), 1); + + let slot = client.monitor().await_acquire().await.unwrap(); + assert_eq!(slot, 124); + + client.monitor().release().await.unwrap(); + client.monitor().done().await.unwrap(); + }); + + tokio::try_join!(client, server).unwrap(); +}