diff --git a/.gitmodules b/.gitmodules
index f91662e8..fdfda6ea 100644
--- a/.gitmodules
+++ b/.gitmodules
@@ -1,3 +1,4 @@
[submodule "cardano-blueprint"]
path = cardano-blueprint
url = https://github.com/cardano-scaling/cardano-blueprint.git
+ branch = leios-prototype
diff --git a/Cargo.toml b/Cargo.toml
index 1bbfe294..fac0fb79 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -32,6 +32,7 @@ members = [
"examples/p2p-discovery",
"examples/p2p-responder",
"examples/p2p-initiator",
+ "examples/leios-testnet",
]
[workspace.dependencies]
diff --git a/cardano-blueprint b/cardano-blueprint
index 594f39a8..188183b3 160000
--- a/cardano-blueprint
+++ b/cardano-blueprint
@@ -1 +1 @@
-Subproject commit 594f39a80aec1787866a69b1ace3dd38d9eee2a4
+Subproject commit 188183b37081fa012fa890236edb7771f96ae92f
diff --git a/examples/leios-testnet/Cargo.toml b/examples/leios-testnet/Cargo.toml
new file mode 100644
index 00000000..e61da563
--- /dev/null
+++ b/examples/leios-testnet/Cargo.toml
@@ -0,0 +1,17 @@
+[package]
+name = "leios-testnet"
+version = "0.1.0"
+edition.workspace = true
+rust-version.workspace = true
+publish = false
+
+[package.metadata.release]
+release = false
+
+[dependencies]
+pallas-network2 = { path = "../../pallas-network2" }
+pallas-codec = { path = "../../pallas-codec" }
+hex = "0.4.3"
+tokio = { version = "1.27.0", features = ["rt-multi-thread", "macros", "time"] }
+tracing = "0.1.41"
+tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
diff --git a/examples/leios-testnet/src/main.rs b/examples/leios-testnet/src/main.rs
new file mode 100644
index 00000000..721cda20
--- /dev/null
+++ b/examples/leios-testnet/src/main.rs
@@ -0,0 +1,290 @@
+//! Connect a node to the Cardano **Leios** ("Musashi Dojo") testnet and follow
+//! its endorsement layer.
+//!
+//! Leios (CIP-0164) is an overlay on top of Ouroboros Praos: alongside the
+//! normal Praos chain, producers diffuse **Endorser Blocks (EBs)** — compact
+//! lists of transaction references — which are then voted on and certified back
+//! into a ranking block. Pallas speaks this overlay through two node-to-node
+//! mini-protocols that ride the *same* connection as Praos once a Leios-capable
+//! handshake version is negotiated:
+//!
+//! - **leios-notify** (server-push): the relay announces / offers new EBs and
+//! their transactions, and diffuses full votes inline. Surfaced as
+//! [`InitiatorEvent::EbNotification`].
+//! - **leios-fetch** (client-pull): we request the EB body or a subset of its
+//! transactions. Surfaced as [`InitiatorEvent::EbFetched`].
+//!
+//! The only thing that "turns Leios on" is the handshake: we must propose an
+//! N2N version `>= LEIOS_MIN_VERSION` (v15, the Dijkstra era) with the testnet's
+//! network magic. The default [`InitiatorBehavior`] only proposes a mainnet v13
+//! handshake, so this example swaps in a v15-capable version table.
+//!
+//! Run with:
+//!
+//! ```sh
+//! RUST_LOG=info cargo run -p leios-testnet
+//! ```
+
+use std::collections::HashMap;
+use std::time::Duration;
+
+use pallas_codec::minicbor::{Decoder, data::Type};
+use pallas_network2::{
+ Manager, PeerId,
+ behavior::{
+ AnyMessage,
+ initiator::{
+ Config as HandshakeConfig, HandshakeBehavior, InitiatorBehavior, InitiatorCommand,
+ InitiatorEvent,
+ },
+ },
+ interface::TcpInterface,
+ protocol::{
+ EbId, Point,
+ handshake::n2n::{LEIOS_MIN_VERSION, VersionTable},
+ leiosfetch, leiosnotify,
+ },
+};
+use tokio::{select, time::Interval};
+
+/// Public bootstrap relay for the Leios "Musashi Dojo" testnet.
+///
+/// This is a throwaway, continuously-resetting devnet — if the connection is
+/// refused, check
+/// for the current relay address and network magic.
+const LEIOS_RELAY: &str = "leios-node.play.dev.cardano.org:3001";
+
+/// Network magic for the Leios testnet (a nod to CIP-0164).
+const LEIOS_TESTNET_MAGIC: u64 = 164;
+
+/// Chain-sync intersection point to start following from, so we sync near the
+/// tip instead of replaying the chain from origin.
+///
+/// The Musashi testnet resets periodically; if sync stalls or the intersection
+/// is not found, replace this with a current point from the chain.
+const INTERSECT_SLOT: u64 = 2724123;
+const INTERSECT_HASH: &str = "daea7e4b9754244bceb713b269771756450d2fe19ee551f7cb24ba65fa841aeb";
+
+/// How many of an EB's transactions to fetch per request. We only request txs a
+/// peer has offered (so it holds the closure), and bound each request to one
+/// 64-tx bitmap window — requesting a whole large EB at once can exceed the
+/// relay's per-response limits. A real client pages across windows as needed.
+const MAX_TXS_PER_FETCH: usize = 64;
+
+struct LeiosNode {
+ network: Manager, InitiatorBehavior, AnyMessage>,
+ housekeeping_interval: Interval,
+ /// Transaction count per EB, learned by decoding the EB body. Used to size a
+ /// correct tx bitmap when the peer later offers that EB's transactions.
+ eb_tx_counts: HashMap,
+}
+
+impl LeiosNode {
+ fn new() -> Self {
+ let interface = TcpInterface::new();
+
+ // The default behavior only proposes a mainnet v13 handshake, which does
+ // not carry the Leios overlay. Swap in a version table that proposes
+ // v11..=v15 with the testnet magic so the peer can negotiate v15 and
+ // enable leios-notify / leios-fetch. The rest of the behavior (chain-sync,
+ // block-fetch, keepalive, ...) is left at its defaults.
+ let behavior = InitiatorBehavior {
+ handshake: HandshakeBehavior::new(HandshakeConfig {
+ supported_version: VersionTable::v11_and_above_with_query(
+ LEIOS_TESTNET_MAGIC,
+ false,
+ ),
+ }),
+ ..Default::default()
+ };
+
+ let network = Manager::new(interface, behavior);
+
+ Self {
+ network,
+ housekeeping_interval: tokio::time::interval(Duration::from_secs(3)),
+ eb_tx_counts: HashMap::new(),
+ }
+ }
+
+ fn handle_event(&mut self, event: InitiatorEvent) {
+ match event {
+ InitiatorEvent::PeerInitialized(pid, (version, _data)) => {
+ let leios = version >= LEIOS_MIN_VERSION;
+ tracing::info!(%pid, version, leios, "peer initialized");
+ if !leios {
+ tracing::warn!(
+ %pid,
+ version,
+ min = LEIOS_MIN_VERSION,
+ "peer negotiated a pre-Leios version; no EBs will be diffused"
+ );
+ }
+ }
+
+ // --- Praos chain-sync (runs underneath Leios) ---
+ InitiatorEvent::IntersectionFound(pid, point, tip) => {
+ tracing::info!(%pid, ?point, tip = %fmt_eb(&tip.0), "intersection found");
+ self.network.execute(InitiatorCommand::ContinueSync(pid));
+ }
+ InitiatorEvent::BlockHeaderReceived(pid, header, tip) => {
+ tracing::info!(%pid, variant = header.variant, tip_block = tip.1, "header received");
+ self.network.execute(InitiatorCommand::ContinueSync(pid));
+ }
+ InitiatorEvent::RollbackReceived(pid, point, tip) => {
+ tracing::warn!(%pid, ?point, tip_block = tip.1, "rollback received");
+ self.network.execute(InitiatorCommand::ContinueSync(pid));
+ }
+
+ // --- Leios: server-pushed announcements / offers (leios-notify) ---
+ InitiatorEvent::EbNotification(pid, notification) => {
+ self.handle_notification(pid, notification);
+ }
+
+ // --- Leios: pulled bodies / txs (leios-fetch) ---
+ InitiatorEvent::EbFetched(pid, eb, response) => match response {
+ leiosfetch::Response::Block(body) => {
+ // The EB body is a `{ tx_hash => size }` map; its entry count
+ // is the EB's transaction count. Remember it so we can size a
+ // correct tx bitmap once the peer offers the transactions.
+ let n = eb_tx_count(body.raw_bytes());
+ tracing::info!(%pid, eb = %fmt_eb(&eb), bytes = body.raw_bytes().len(), txs = n, "EB body fetched");
+ self.eb_tx_counts.insert(eb, n);
+ }
+ leiosfetch::Response::BlockTxs { txs } => {
+ let total: usize = txs.iter().map(|tx| tx.raw_bytes().len()).sum();
+ tracing::info!(%pid, eb = %fmt_eb(&eb), count = txs.len(), bytes = total, "EB transactions fetched");
+ for (i, tx) in txs.iter().enumerate() {
+ tracing::debug!(%pid, index = i, bytes = tx.raw_bytes().len(), " tx");
+ }
+ }
+ },
+
+ other => {
+ tracing::debug!(?other, "unhandled event");
+ }
+ }
+ }
+
+ fn handle_notification(&mut self, pid: PeerId, notification: leiosnotify::Notification) {
+ match notification {
+ leiosnotify::Notification::BlockOffer(eb_id, size) => {
+ tracing::info!(%pid, eb = %fmt_eb(&eb_id), size, "EB offered → fetching body");
+ // Pull the body to learn the EB's tx count; we fetch the txs
+ // later, when the peer offers them (`BlockTxsOffer`).
+ self.network.execute(InitiatorCommand::FetchEb(pid, eb_id));
+ }
+ leiosnotify::Notification::BlockAnnouncement(raw) => {
+ tracing::info!(%pid, bytes = raw.raw_bytes().len(), "EB announced");
+ }
+ leiosnotify::Notification::BlockTxsOffer(eb_id) => {
+ // The peer signals it holds this EB's transaction closure, so it
+ // can serve a BlockTxsRequest. Requesting txs a peer has NOT
+ // offered makes the prototype relay reset the connection, so we
+ // only fetch in response to this offer, sized from the body's tx
+ // count (learned when we fetched the body).
+ match self.eb_tx_counts.get(&eb_id).copied() {
+ Some(n) if n > 0 => {
+ let want = n.min(MAX_TXS_PER_FETCH);
+ tracing::info!(%pid, eb = %fmt_eb(&eb_id), want, total = n, "txs offered → fetching");
+ self.network.execute(InitiatorCommand::FetchEbTxs(
+ pid,
+ eb_id,
+ leiosfetch::Bitmaps::all(want),
+ ));
+ }
+ _ => {
+ tracing::info!(%pid, eb = %fmt_eb(&eb_id), "txs offered (body not yet fetched)");
+ }
+ }
+ }
+ leiosnotify::Notification::Votes(votes) => {
+ tracing::info!(%pid, count = votes.len(), "votes received");
+ }
+ }
+ }
+
+ async fn tick(&mut self) {
+ select! {
+ _ = self.housekeeping_interval.tick() => {
+ tracing::debug!("housekeeping tick");
+ self.network.execute(InitiatorCommand::Housekeeping);
+ }
+ evt = self.network.poll_next() => {
+ if let Some(evt) = evt {
+ self.handle_event(evt);
+ }
+ }
+ }
+ }
+
+ async fn run_forever(&mut self) {
+ loop {
+ self.tick().await;
+ }
+ }
+}
+
+/// Formats an EB reference (`[slot, hash]`) for logging.
+fn fmt_eb(eb: &Point) -> String {
+ match eb {
+ Point::Origin => "origin".to_string(),
+ Point::Specific(slot, hash) => format!("{slot}@{}", hex::encode(hash)),
+ }
+}
+
+/// Counts the transactions in an EB body, which is a `{ tx_hash => size }` CBOR
+/// map — the number of entries is the transaction count.
+fn eb_tx_count(body: &[u8]) -> usize {
+ let mut d = Decoder::new(body);
+ match d.map() {
+ Ok(Some(n)) => n as usize,
+ // Indefinite-length map: count key/value pairs until the break marker.
+ Ok(None) => {
+ let mut n = 0;
+ while !matches!(d.datatype(), Ok(Type::Break)) {
+ if d.skip().is_err() || d.skip().is_err() {
+ break;
+ }
+ n += 1;
+ }
+ n
+ }
+ Err(_) => 0,
+ }
+}
+
+#[tokio::main]
+async fn main() {
+ tracing_subscriber::fmt()
+ .with_env_filter(
+ tracing_subscriber::EnvFilter::try_from_default_env()
+ .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
+ )
+ .init();
+
+ let mut node = LeiosNode::new();
+
+ let peer = LEIOS_RELAY
+ .parse()
+ .expect("LEIOS_RELAY should be a valid host:port");
+
+ tracing::info!(
+ relay = LEIOS_RELAY,
+ magic = LEIOS_TESTNET_MAGIC,
+ "connecting to Leios testnet"
+ );
+
+ node.network.execute(InitiatorCommand::IncludePeer(peer));
+ // Start chain-sync near the tip (not origin) so we follow live blocks without
+ // replaying the whole chain. The Leios overlay diffuses EBs over the same
+ // connection independently of where we are in chain-sync.
+ let intersect = Point::Specific(
+ INTERSECT_SLOT,
+ hex::decode(INTERSECT_HASH).expect("INTERSECT_HASH should be valid hex"),
+ );
+ node.network
+ .execute(InitiatorCommand::StartSync(vec![intersect]));
+
+ node.run_forever().await;
+}
diff --git a/examples/p2p-discovery/src/node.rs b/examples/p2p-discovery/src/node.rs
index 7ed928d0..a5c556d5 100644
--- a/examples/p2p-discovery/src/node.rs
+++ b/examples/p2p-discovery/src/node.rs
@@ -143,6 +143,12 @@ impl> MyNode {
InitiatorEvent::TxRequested(pid, _) => {
tracing::info!(%pid, "tx requested");
}
+ InitiatorEvent::EbNotification(pid, _) => {
+ tracing::info!(%pid, "leios notification received");
+ }
+ InitiatorEvent::EbFetched(pid, _, _) => {
+ tracing::info!(%pid, "leios fetch received");
+ }
}
self.enqueue_next_cmds();
diff --git a/examples/p2p-responder/src/main.rs b/examples/p2p-responder/src/main.rs
index 36be7fed..3d4ae230 100644
--- a/examples/p2p-responder/src/main.rs
+++ b/examples/p2p-responder/src/main.rs
@@ -116,6 +116,15 @@ impl MockResponderNode {
ResponderEvent::TxReceived(pid, _tx) => {
tracing::info!(%pid, "tx received");
}
+ ResponderEvent::EbNotificationRequested(pid) => {
+ tracing::info!(%pid, "leios notification requested");
+ }
+ ResponderEvent::EbRequested(pid, _) => {
+ tracing::info!(%pid, "leios eb requested");
+ }
+ ResponderEvent::EbTxsRequested(pid, _, _) => {
+ tracing::info!(%pid, "leios eb txs requested");
+ }
}
}
diff --git a/pallas-codec/src/utils.rs b/pallas-codec/src/utils.rs
index 9d1b71dd..f1f66bac 100644
--- a/pallas-codec/src/utils.rs
+++ b/pallas-codec/src/utils.rs
@@ -1233,6 +1233,13 @@ pub struct AnyCbor {
}
impl AnyCbor {
+ /// Wraps already-encoded CBOR bytes verbatim, without re-encoding. Use this
+ /// to embed a pre-serialized CBOR item; contrast with [`AnyCbor::from_encode`],
+ /// which encodes a value into CBOR.
+ pub fn from_raw_bytes(inner: Vec) -> Self {
+ Self { inner }
+ }
+
pub fn raw_bytes(&self) -> &[u8] {
&self.inner
}
@@ -1257,6 +1264,13 @@ impl AnyCbor {
}
}
+impl From> for AnyCbor {
+ /// Wraps already-encoded CBOR bytes verbatim (see [`AnyCbor::from_raw_bytes`]).
+ fn from(inner: Vec) -> Self {
+ Self { inner }
+ }
+}
+
impl Deref for AnyCbor {
type Target = Vec;
diff --git a/pallas-network2/Cargo.toml b/pallas-network2/Cargo.toml
index 3fb3c0a4..8e27bf9e 100644
--- a/pallas-network2/Cargo.toml
+++ b/pallas-network2/Cargo.toml
@@ -24,11 +24,12 @@ byteorder.workspace = true
socket2.workspace = true
thiserror.workspace = true
hex.workspace = true
+cddl = { version = "0.10", optional = true }
[features]
emulation = ["tokio", "rand"]
default = ["emulation"]
-blueprint = []
+blueprint = ["dep:cddl"]
[dev-dependencies]
tokio = { version = "1.45.1", features = ["full", "test-util"] }
diff --git a/pallas-network2/benches/harness/node.rs b/pallas-network2/benches/harness/node.rs
index dc630ba6..cc5296e4 100644
--- a/pallas-network2/benches/harness/node.rs
+++ b/pallas-network2/benches/harness/node.rs
@@ -73,6 +73,10 @@ impl ResponderNode {
manager.execute(ResponderCommand::ProvidePeers(pid.clone(), vec![]));
}
ResponderEvent::TxReceived(_, _) => {}
+ // Leios responder events are not exercised by this Praos bench.
+ ResponderEvent::EbNotificationRequested(..)
+ | ResponderEvent::EbRequested(..)
+ | ResponderEvent::EbTxsRequested(..) => {}
}
events.push(event);
diff --git a/pallas-network2/src/behavior/initiator/leiosfetch.rs b/pallas-network2/src/behavior/initiator/leiosfetch.rs
new file mode 100644
index 00000000..95d13143
--- /dev/null
+++ b/pallas-network2/src/behavior/initiator/leiosfetch.rs
@@ -0,0 +1,200 @@
+use std::collections::VecDeque;
+
+use crate::protocol::EbId;
+use crate::protocol::leiosfetch::{self as fetch_proto, Bitmaps};
+
+use crate::{BehaviorOutput, InterfaceCommand, OutboundQueue, PeerId, behavior::AnyMessage};
+
+use super::{InitiatorBehavior, InitiatorEvent, InitiatorState, PeerVisitor};
+
+/// A pending leios-fetch request targeting a specific peer.
+#[derive(Debug, Clone)]
+pub enum FetchRequest {
+ /// Fetch a complete EB body.
+ Block(EbId),
+ /// Fetch a subset of an EB's transactions.
+ BlockTxs(EbId, Bitmaps),
+}
+
+/// Sub-behavior that fetches EB bodies and transactions from peers.
+///
+/// Requests are queued (each targeting the peer that should serve it) and sent
+/// one at a time per peer during housekeeping, when that peer is idle. Responses
+/// are surfaced as [`InitiatorEvent::EbFetched`].
+#[derive(Default)]
+pub struct LeiosFetchBehavior {
+ requests: VecDeque<(PeerId, FetchRequest)>,
+}
+
+impl LeiosFetchBehavior {
+ /// Queues a fetch request to be served by the given peer.
+ pub fn enqueue(&mut self, pid: PeerId, request: FetchRequest) {
+ self.requests.push_back((pid, request));
+ tracing::info!(total = self.requests.len(), "new leios-fetch request");
+ }
+
+ /// Drops any queued requests targeting `pid`. Called when the peer goes away
+ /// so requests don't leak or get re-sent to a later reconnection of the same
+ /// `PeerId` (which may no longer hold the offered EB).
+ fn purge(&mut self, pid: &PeerId) {
+ self.requests.retain(|(p, _)| p != pid);
+ }
+
+ fn send_request(
+ &self,
+ pid: &PeerId,
+ request: &FetchRequest,
+ outbound: &mut OutboundQueue,
+ ) {
+ let msg = match request {
+ FetchRequest::Block(point) => fetch_proto::Message::BlockRequest(point.clone()),
+ FetchRequest::BlockTxs(point, bitmaps) => {
+ fetch_proto::Message::BlockTxsRequest(point.clone(), bitmaps.clone())
+ }
+ };
+
+ outbound.push_ready(BehaviorOutput::InterfaceCommand(InterfaceCommand::Send(
+ pid.clone(),
+ AnyMessage::LeiosFetch(msg),
+ )));
+ }
+
+ /// Drains a pending response from the peer state and emits the corresponding
+ /// external event.
+ fn dispatch(
+ &self,
+ pid: &PeerId,
+ state: &mut InitiatorState,
+ outbound: &mut OutboundQueue,
+ ) {
+ if let Some((eb, response)) = state.leios_fetch.drain() {
+ outbound.push_ready(BehaviorOutput::ExternalEvent(InitiatorEvent::EbFetched(
+ pid.clone(),
+ eb,
+ response,
+ )));
+ }
+ }
+}
+
+fn peer_is_available(state: &InitiatorState) -> bool {
+ state.is_initialized()
+ && state.supports_leios()
+ && matches!(state.leios_fetch, fetch_proto::State::Idle(None))
+}
+
+impl PeerVisitor for LeiosFetchBehavior {
+ fn visit_inbound_msg(
+ &mut self,
+ pid: &PeerId,
+ state: &mut InitiatorState,
+ outbound: &mut OutboundQueue,
+ ) {
+ self.dispatch(pid, state, outbound);
+ }
+
+ fn visit_housekeeping(
+ &mut self,
+ pid: &PeerId,
+ state: &mut InitiatorState,
+ outbound: &mut OutboundQueue,
+ ) {
+ if !peer_is_available(state) {
+ return;
+ }
+
+ // Serve the first queued request targeting this peer.
+ if let Some(idx) = self.requests.iter().position(|(p, _)| p == pid) {
+ let (_, request) = self.requests.remove(idx).expect("index just found");
+ self.send_request(pid, &request, outbound);
+ }
+ }
+
+ fn visit_disconnected(
+ &mut self,
+ pid: &PeerId,
+ _state: &mut InitiatorState,
+ _outbound: &mut OutboundQueue,
+ ) {
+ self.purge(pid);
+ }
+
+ fn visit_errored(
+ &mut self,
+ pid: &PeerId,
+ _state: &mut InitiatorState,
+ _outbound: &mut OutboundQueue,
+ ) {
+ self.purge(pid);
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::protocol::Point;
+ use crate::protocol::leiosfetch::Response;
+ use crate::protocol::{AnyCbor, leiosfetch as lf};
+ use crate::{OutboundQueue, behavior::ConnectionState};
+
+ fn drain_outputs(
+ outbound: &mut OutboundQueue,
+ ) -> Vec> {
+ outbound.drain_ready()
+ }
+
+ #[test]
+ fn dispatch_emits_fetched_event_once() {
+ let b = LeiosFetchBehavior::default();
+ let pid = PeerId::test(1);
+ let mut state = InitiatorState::new();
+ let mut outbound = OutboundQueue::new();
+
+ state.leios_fetch = lf::State::Idle(Some((
+ Point::Origin,
+ Response::Block(AnyCbor::from_raw_bytes(vec![0x01])),
+ )));
+
+ b.dispatch(&pid, &mut state, &mut outbound);
+ assert!(drain_outputs(&mut outbound).iter().any(|o| matches!(
+ o,
+ BehaviorOutput::ExternalEvent(InitiatorEvent::EbFetched(..))
+ )));
+
+ b.dispatch(&pid, &mut state, &mut outbound);
+ assert!(drain_outputs(&mut outbound).is_empty());
+ }
+
+ #[test]
+ fn housekeeping_sends_request_for_available_peer() {
+ let mut b = LeiosFetchBehavior::default();
+ let pid = PeerId::test(1);
+ let mut outbound = OutboundQueue::new();
+
+ b.enqueue(pid.clone(), FetchRequest::Block(Point::Origin));
+
+ // peer not ready → request stays queued
+ let mut state = InitiatorState::new();
+ state.connection = ConnectionState::Initialized; // but supports_leios() is false
+ b.visit_housekeeping(&pid, &mut state, &mut outbound);
+ assert!(drain_outputs(&mut outbound).is_empty());
+ assert_eq!(b.requests.len(), 1);
+ }
+
+ #[test]
+ fn disconnect_purges_queued_requests() {
+ let mut b = LeiosFetchBehavior::default();
+ let pid = PeerId::test(1);
+ let mut state = InitiatorState::new();
+ let mut outbound = OutboundQueue::new();
+
+ b.enqueue(pid.clone(), FetchRequest::Block(Point::Origin));
+ b.enqueue(PeerId::test(2), FetchRequest::Block(Point::Origin));
+ assert_eq!(b.requests.len(), 2);
+
+ // Disconnecting pid drops only its queued request.
+ b.visit_disconnected(&pid, &mut state, &mut outbound);
+ assert_eq!(b.requests.len(), 1);
+ assert!(b.requests.iter().all(|(p, _)| p != &pid));
+ }
+}
diff --git a/pallas-network2/src/behavior/initiator/leiosnotify.rs b/pallas-network2/src/behavior/initiator/leiosnotify.rs
new file mode 100644
index 00000000..3712e9b6
--- /dev/null
+++ b/pallas-network2/src/behavior/initiator/leiosnotify.rs
@@ -0,0 +1,126 @@
+use crate::protocol::leiosnotify as notify_proto;
+
+use crate::{BehaviorOutput, InterfaceCommand, OutboundQueue, PeerId, behavior::AnyMessage};
+
+use super::{InitiatorBehavior, InitiatorEvent, InitiatorState, PeerVisitor};
+
+/// Sub-behavior that drives the leios-notify pull loop and surfaces EB
+/// announcements/offers received from peers.
+///
+/// `RequestNext` is issued during housekeeping whenever the peer is initialized,
+/// negotiated a Leios-capable version, and the protocol is idle with nothing
+/// pending — yielding a continuous notification loop paced by housekeeping.
+#[derive(Default)]
+pub struct LeiosNotifyBehavior;
+
+impl LeiosNotifyBehavior {
+ fn request_next(&self, pid: &PeerId, outbound: &mut OutboundQueue) {
+ tracing::debug!("requesting next leios notification");
+
+ outbound.push_ready(BehaviorOutput::InterfaceCommand(InterfaceCommand::Send(
+ pid.clone(),
+ AnyMessage::LeiosNotify(notify_proto::Message::RequestNext),
+ )));
+ }
+
+ /// Drains a pending notification from the peer state and emits the
+ /// corresponding external event.
+ fn dispatch(
+ &self,
+ pid: &PeerId,
+ state: &mut InitiatorState,
+ outbound: &mut OutboundQueue,
+ ) {
+ if let Some(notification) = state.leios_notify.drain() {
+ outbound.push_ready(BehaviorOutput::ExternalEvent(
+ InitiatorEvent::EbNotification(pid.clone(), notification),
+ ));
+ }
+ }
+}
+
+fn peer_ready(state: &InitiatorState) -> bool {
+ state.is_initialized() && state.supports_leios()
+}
+
+impl PeerVisitor for LeiosNotifyBehavior {
+ fn visit_inbound_msg(
+ &mut self,
+ pid: &PeerId,
+ state: &mut InitiatorState,
+ outbound: &mut OutboundQueue,
+ ) {
+ self.dispatch(pid, state, outbound);
+ }
+
+ fn visit_housekeeping(
+ &mut self,
+ pid: &PeerId,
+ state: &mut InitiatorState,
+ outbound: &mut OutboundQueue,
+ ) {
+ if !peer_ready(state) {
+ return;
+ }
+
+ // Only request when idle with nothing pending; the Sent event will move
+ // the protocol to Busy before the next housekeeping pass, avoiding
+ // duplicate requests.
+ if matches!(state.leios_notify, notify_proto::State::Idle(None)) {
+ self.request_next(pid, outbound);
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::protocol::Point;
+ use crate::protocol::leiosnotify::Notification;
+ use crate::{OutboundQueue, behavior::ConnectionState};
+
+ fn drain_outputs(
+ outbound: &mut OutboundQueue,
+ ) -> Vec> {
+ outbound.drain_ready()
+ }
+
+ #[test]
+ fn dispatch_emits_notification_event() {
+ let b = LeiosNotifyBehavior;
+ let pid = PeerId::test(1);
+ let mut state = InitiatorState::new();
+ let mut outbound = OutboundQueue::new();
+
+ state.leios_notify =
+ notify_proto::State::Idle(Some(Notification::BlockOffer(Point::Origin, 10)));
+
+ b.dispatch(&pid, &mut state, &mut outbound);
+
+ let outputs = drain_outputs(&mut outbound);
+ assert!(outputs.iter().any(|o| matches!(
+ o,
+ BehaviorOutput::ExternalEvent(InitiatorEvent::EbNotification(..))
+ )));
+ // drained, so a second dispatch emits nothing
+ b.dispatch(&pid, &mut state, &mut outbound);
+ assert!(drain_outputs(&mut outbound).is_empty());
+ }
+
+ #[test]
+ fn housekeeping_requests_next_only_when_ready_and_idle() {
+ let mut b = LeiosNotifyBehavior;
+ let pid = PeerId::test(1);
+ let mut outbound = OutboundQueue::new();
+
+ // not initialized → nothing
+ let mut state = InitiatorState::new();
+ b.visit_housekeeping(&pid, &mut state, &mut outbound);
+ assert!(drain_outputs(&mut outbound).is_empty());
+
+ // initialized but no Leios version → nothing
+ state.connection = ConnectionState::Initialized;
+ b.visit_housekeeping(&pid, &mut state, &mut outbound);
+ assert!(drain_outputs(&mut outbound).is_empty());
+ }
+}
diff --git a/pallas-network2/src/behavior/initiator/mod.rs b/pallas-network2/src/behavior/initiator/mod.rs
index ce6e42c4..7020ec0a 100644
--- a/pallas-network2/src/behavior/initiator/mod.rs
+++ b/pallas-network2/src/behavior/initiator/mod.rs
@@ -13,6 +13,8 @@ mod connection;
mod discovery;
mod handshake;
mod keepalive;
+mod leiosfetch;
+mod leiosnotify;
mod promotion;
pub use blockfetch::*;
@@ -21,6 +23,8 @@ pub use connection::*;
pub use discovery::*;
pub use handshake::*;
pub use keepalive::*;
+pub use leiosfetch::*;
+pub use leiosnotify::*;
pub use promotion::*;
/// A visitor trait that allows sub-behaviors to react to peer lifecycle events.
@@ -143,6 +147,8 @@ pub struct InitiatorState {
pub(crate) blockfetch: proto::blockfetch::State,
pub(crate) chainsync: proto::chainsync::State,
pub(crate) tx_submission: proto::txsubmission::State,
+ pub(crate) leios_notify: proto::leiosnotify::State,
+ pub(crate) leios_fetch: proto::leiosfetch::State,
pub(crate) violation: bool,
pub(crate) error_count: u32,
pub(crate) continue_sync: bool,
@@ -160,6 +166,8 @@ impl InitiatorState {
blockfetch: proto::blockfetch::State::default(),
chainsync: crate::protocol::chainsync::State::default(),
tx_submission: crate::protocol::txsubmission::State::default(),
+ leios_notify: proto::leiosnotify::State::default(),
+ leios_fetch: proto::leiosfetch::State::default(),
violation: false,
error_count: 0,
continue_sync: false,
@@ -197,6 +205,16 @@ impl InitiatorState {
val > 0
}
+ /// Returns the negotiated N2N protocol version, if the handshake completed.
+ pub fn accepted_version(&self) -> Option {
+ super::accepted_version(&self.handshake)
+ }
+
+ /// Returns true if the negotiated version carries the Leios overlay.
+ pub fn supports_leios(&self) -> bool {
+ super::supports_leios(&self.handshake)
+ }
+
/// Applies a message to the corresponding mini-protocol state machine.
pub fn apply_msg(&mut self, msg: &AnyMessage) {
match msg {
@@ -266,6 +284,28 @@ impl InitiatorState {
self.tx_submission = new;
}
+ AnyMessage::LeiosNotify(msg) => {
+ let result = self.leios_notify.apply(msg);
+
+ let Ok(new) = result else {
+ tracing::warn!("leios notify violation");
+ self.violation = true;
+ return;
+ };
+
+ self.leios_notify = new;
+ }
+ AnyMessage::LeiosFetch(msg) => {
+ let result = self.leios_fetch.apply(msg);
+
+ let Ok(new) = result else {
+ tracing::warn!("leios fetch violation");
+ self.violation = true;
+ return;
+ };
+
+ self.leios_fetch = new;
+ }
}
}
@@ -279,6 +319,8 @@ impl InitiatorState {
self.blockfetch = proto::blockfetch::State::default();
self.chainsync = proto::chainsync::State::default();
self.tx_submission = proto::txsubmission::State::default();
+ self.leios_notify = proto::leiosnotify::State::default();
+ self.leios_fetch = proto::leiosfetch::State::default();
self.continue_sync = false;
self.violation = false;
}
@@ -307,6 +349,10 @@ pub enum InitiatorCommand {
proto::txsubmission::EraTxId,
proto::txsubmission::EraTxBody,
),
+ /// Request a complete EB body from a peer (leios-fetch).
+ FetchEb(PeerId, proto::EbId),
+ /// Request a subset of an EB's transactions from a peer (leios-fetch).
+ FetchEbTxs(PeerId, proto::EbId, proto::leiosfetch::Bitmaps),
/// Ban a peer, preventing future connections.
BanPeer(PeerId),
/// Demote a peer back to cold status.
@@ -332,6 +378,10 @@ pub enum InitiatorEvent {
BlockBodyReceived(PeerId, proto::blockfetch::Body),
/// The remote peer requested a transaction via tx-submission.
TxRequested(PeerId, proto::txsubmission::EraTxId),
+ /// An EB announcement or offer was received via leios-notify.
+ EbNotification(PeerId, proto::leiosnotify::Notification),
+ /// An EB body or transactions were received via leios-fetch, for the given EB.
+ EbFetched(PeerId, proto::EbId, proto::leiosfetch::Response),
}
/// The main initiator behavior that orchestrates outbound Cardano connections.
@@ -348,6 +398,8 @@ pub struct InitiatorBehavior {
pub discovery: discovery::DiscoveryBehavior,
pub blockfetch: blockfetch::BlockFetchBehavior,
pub chainsync: chainsync::ChainSyncBehavior,
+ pub leiosnotify: leiosnotify::LeiosNotifyBehavior,
+ pub leiosfetch: leiosfetch::LeiosFetchBehavior,
pub peers: HashMap,
pub outbound: OutboundQueue,
}
@@ -361,6 +413,8 @@ macro_rules! all_visitors {
$self.discovery.$method($pid, $state, &mut $self.outbound);
$self.blockfetch.$method($pid, $state, &mut $self.outbound);
$self.chainsync.$method($pid, $state, &mut $self.outbound);
+ $self.leiosnotify.$method($pid, $state, &mut $self.outbound);
+ $self.leiosfetch.$method($pid, $state, &mut $self.outbound);
};
}
@@ -564,6 +618,16 @@ impl Behavior for InitiatorBehavior {
InitiatorCommand::SendTx(..) => {
tracing::warn!("SendTx not yet implemented");
}
+ InitiatorCommand::FetchEb(pid, point) => {
+ tracing::debug!("fetch eb command");
+ self.leiosfetch
+ .enqueue(pid, leiosfetch::FetchRequest::Block(point));
+ }
+ InitiatorCommand::FetchEbTxs(pid, point, bitmaps) => {
+ tracing::debug!("fetch eb txs command");
+ self.leiosfetch
+ .enqueue(pid, leiosfetch::FetchRequest::BlockTxs(point, bitmaps));
+ }
}
}
}
@@ -572,7 +636,8 @@ impl Behavior for InitiatorBehavior {
mod tests {
use super::*;
use crate::protocol::{
- MAINNET_MAGIC, Point, blockfetch as bf, chainsync as cs, handshake, keepalive, peersharing,
+ AnyCbor, MAINNET_MAGIC, Point, blockfetch as bf, chainsync as cs, handshake, keepalive,
+ leiosfetch as lf, leiosnotify as ln, peersharing,
};
use crate::testing::BehaviorOutputExt;
use crate::{InterfaceError, InterfaceEvent};
@@ -608,6 +673,23 @@ mod tests {
drain_outputs(behavior);
}
+ /// Completes a handshake negotiating a Leios-capable version (15).
+ fn complete_handshake_leios(behavior: &mut InitiatorBehavior, pid: &PeerId) {
+ let version_data =
+ handshake::n2n::VersionData::new(MAINNET_MAGIC, false, Some(1), Some(false));
+ let mut values = HashMap::new();
+ values.insert(15u64, version_data.clone());
+ let version_table = handshake::VersionTable { values };
+
+ let propose = AnyMessage::Handshake(handshake::Message::Propose(version_table));
+ behavior.handle_io(InterfaceEvent::Sent(pid.clone(), propose));
+ drain_outputs(behavior);
+
+ let accept = AnyMessage::Handshake(handshake::Message::Accept(15, version_data));
+ behavior.handle_io(InterfaceEvent::Recv(pid.clone(), vec![accept]));
+ drain_outputs(behavior);
+ }
+
// ---- Kept: genuinely cross-cutting tests ----
#[tokio::test]
@@ -913,4 +995,72 @@ mod tests {
"should send RequestRange after handshake completes"
);
}
+
+ #[tokio::test]
+ async fn leios_notify_and_fetch_flow() {
+ // Composition: handshake negotiates v15 → supports_leios → notify pull
+ // loop surfaces an offer → fetch command pulls the EB body.
+ tokio::time::pause();
+
+ let mut behavior = InitiatorBehavior::default();
+ let pid = PeerId::test(20);
+ let eb = Point::new(7, vec![0xAB; 32]);
+
+ behavior.execute(InitiatorCommand::IncludePeer(pid.clone()));
+ behavior.execute(InitiatorCommand::Housekeeping);
+ drain_outputs(&mut behavior);
+
+ behavior.handle_io(InterfaceEvent::Connected(pid.clone()));
+ drain_outputs(&mut behavior);
+ complete_handshake_leios(&mut behavior, &pid);
+ assert!(behavior.peers.get(&pid).unwrap().supports_leios());
+
+ // Housekeeping drives the notify pull loop once Leios is negotiated.
+ behavior.execute(InitiatorCommand::Housekeeping);
+ let outputs = drain_outputs(&mut behavior);
+ assert!(
+ outputs.has_send(|m| matches!(m, AnyMessage::LeiosNotify(ln::Message::RequestNext))),
+ "should request next leios notification"
+ );
+
+ // Server moves to Busy (our request was sent), then offers an EB.
+ behavior.handle_io(InterfaceEvent::Sent(
+ pid.clone(),
+ AnyMessage::LeiosNotify(ln::Message::RequestNext),
+ ));
+ drain_outputs(&mut behavior);
+
+ let offer = AnyMessage::LeiosNotify(ln::Message::BlockOffer(eb.clone(), 99));
+ behavior.handle_io(InterfaceEvent::Recv(pid.clone(), vec![offer]));
+ let outputs = drain_outputs(&mut behavior);
+ assert!(
+ outputs.has_event(|e| matches!(e, InitiatorEvent::EbNotification(..))),
+ "should surface the EB offer as an event"
+ );
+
+ // Application requests the EB body; housekeeping sends the fetch request.
+ behavior.execute(InitiatorCommand::FetchEb(pid.clone(), eb.clone()));
+ behavior.execute(InitiatorCommand::Housekeeping);
+ let outputs = drain_outputs(&mut behavior);
+ assert!(
+ outputs.has_send(|m| matches!(m, AnyMessage::LeiosFetch(lf::Message::BlockRequest(_)))),
+ "should send a leios-fetch block request"
+ );
+
+ // Server delivers the EB body, surfaced as EbFetched.
+ behavior.handle_io(InterfaceEvent::Sent(
+ pid.clone(),
+ AnyMessage::LeiosFetch(lf::Message::BlockRequest(eb.clone())),
+ ));
+ drain_outputs(&mut behavior);
+
+ let block =
+ AnyMessage::LeiosFetch(lf::Message::Block(AnyCbor::from_raw_bytes(vec![1, 2, 3])));
+ behavior.handle_io(InterfaceEvent::Recv(pid.clone(), vec![block]));
+ let outputs = drain_outputs(&mut behavior);
+ assert!(
+ outputs.has_event(|e| matches!(e, InitiatorEvent::EbFetched(..))),
+ "should surface the fetched EB body as an event"
+ );
+ }
}
diff --git a/pallas-network2/src/behavior/mod.rs b/pallas-network2/src/behavior/mod.rs
index 21bd464e..0ad94157 100644
--- a/pallas-network2/src/behavior/mod.rs
+++ b/pallas-network2/src/behavior/mod.rs
@@ -10,6 +10,28 @@ pub mod responder;
// Re-export initiator types for backward compatibility
pub use initiator::*;
+/// Returns the negotiated N2N protocol version number from a completed handshake.
+///
+/// Shared by `InitiatorState` and `ResponderState` so the version-gating logic
+/// lives in one place.
+pub(crate) fn accepted_version(
+ handshake: &proto::handshake::State,
+) -> Option {
+ match handshake {
+ proto::handshake::State::Done(proto::handshake::DoneState::Accepted(num, _)) => Some(*num),
+ _ => None,
+ }
+}
+
+/// Returns true if the negotiated N2N version carries the Leios overlay.
+pub(crate) fn supports_leios(
+ handshake: &proto::handshake::State,
+) -> bool {
+ accepted_version(handshake)
+ .map(|v| v >= proto::handshake::n2n::LEIOS_MIN_VERSION)
+ .unwrap_or(false)
+}
+
/// A unified message type that wraps all supported mini-protocol messages.
#[derive(Debug, Clone)]
pub enum AnyMessage {
@@ -25,6 +47,10 @@ pub enum AnyMessage {
BlockFetch(proto::blockfetch::Message),
/// A tx-submission protocol message.
TxSubmission(proto::txsubmission::Message),
+ /// A leios-notify protocol message.
+ LeiosNotify(proto::leiosnotify::Message),
+ /// A leios-fetch protocol message.
+ LeiosFetch(proto::leiosfetch::Message),
}
fn try_decode_msg(buffer: &mut Vec) -> Option {
@@ -54,6 +80,8 @@ impl Message for AnyMessage {
AnyMessage::PeerSharing(_) => proto::peersharing::CHANNEL_ID,
AnyMessage::BlockFetch(_) => proto::blockfetch::CHANNEL_ID,
AnyMessage::TxSubmission(_) => proto::txsubmission::CHANNEL_ID,
+ AnyMessage::LeiosNotify(_) => proto::leiosnotify::CHANNEL_ID,
+ AnyMessage::LeiosFetch(_) => proto::leiosfetch::CHANNEL_ID,
}
}
@@ -65,6 +93,8 @@ impl Message for AnyMessage {
AnyMessage::PeerSharing(msg) => pallas_codec::minicbor::to_vec(msg).unwrap(),
AnyMessage::BlockFetch(msg) => pallas_codec::minicbor::to_vec(msg).unwrap(),
AnyMessage::TxSubmission(msg) => pallas_codec::minicbor::to_vec(msg).unwrap(),
+ AnyMessage::LeiosNotify(msg) => pallas_codec::minicbor::to_vec(msg).unwrap(),
+ AnyMessage::LeiosFetch(msg) => pallas_codec::minicbor::to_vec(msg).unwrap(),
}
}
@@ -78,6 +108,8 @@ impl Message for AnyMessage {
proto::txsubmission::CHANNEL_ID => {
try_decode_msg(payload).map(AnyMessage::TxSubmission)
}
+ proto::leiosnotify::CHANNEL_ID => try_decode_msg(payload).map(AnyMessage::LeiosNotify),
+ proto::leiosfetch::CHANNEL_ID => try_decode_msg(payload).map(AnyMessage::LeiosFetch),
x => {
tracing::warn!(channel = x, "unsupported channel, skipping payload");
payload.clear();
diff --git a/pallas-network2/src/behavior/responder/leiosfetch.rs b/pallas-network2/src/behavior/responder/leiosfetch.rs
new file mode 100644
index 00000000..1cdf837a
--- /dev/null
+++ b/pallas-network2/src/behavior/responder/leiosfetch.rs
@@ -0,0 +1,38 @@
+use crate::{BehaviorOutput, OutboundQueue, PeerId, protocol::leiosfetch as fetch_proto};
+
+use super::{ResponderBehavior, ResponderEvent, ResponderPeerVisitor, ResponderState};
+
+/// Responder sub-behavior that surfaces leios-fetch requests from peers as
+/// events, so the application can answer with the corresponding `Provide*`
+/// responder commands.
+#[derive(Default)]
+pub struct LeiosFetchResponder;
+
+impl ResponderPeerVisitor for LeiosFetchResponder {
+ fn visit_inbound_msg(
+ &mut self,
+ pid: &PeerId,
+ state: &mut ResponderState,
+ outbound: &mut OutboundQueue,
+ ) {
+ if !state.is_initialized() {
+ return;
+ }
+
+ let event = match &state.leios_fetch {
+ fetch_proto::State::AwaitingBlock(point) => {
+ Some(ResponderEvent::EbRequested(pid.clone(), point.clone()))
+ }
+ fetch_proto::State::AwaitingBlockTxs(point, bitmaps) => Some(
+ ResponderEvent::EbTxsRequested(pid.clone(), point.clone(), bitmaps.clone()),
+ ),
+ // Idle/Done carry no inbound request to surface.
+ _ => None,
+ };
+
+ if let Some(event) = event {
+ tracing::debug!("leios fetch requested");
+ outbound.push_ready(BehaviorOutput::ExternalEvent(event));
+ }
+ }
+}
diff --git a/pallas-network2/src/behavior/responder/leiosnotify.rs b/pallas-network2/src/behavior/responder/leiosnotify.rs
new file mode 100644
index 00000000..8844a752
--- /dev/null
+++ b/pallas-network2/src/behavior/responder/leiosnotify.rs
@@ -0,0 +1,29 @@
+use crate::{BehaviorOutput, OutboundQueue, PeerId, protocol::leiosnotify as notify_proto};
+
+use super::{ResponderBehavior, ResponderEvent, ResponderPeerVisitor, ResponderState};
+
+/// Responder sub-behavior that surfaces leios-notify `RequestNext` from peers as
+/// [`ResponderEvent::EbNotificationRequested`], so the application can answer
+/// with an announcement or offer (via the `Provide*` responder commands).
+#[derive(Default)]
+pub struct LeiosNotifyResponder;
+
+impl ResponderPeerVisitor for LeiosNotifyResponder {
+ fn visit_inbound_msg(
+ &mut self,
+ pid: &PeerId,
+ state: &mut ResponderState,
+ outbound: &mut OutboundQueue,
+ ) {
+ if !state.is_initialized() {
+ return;
+ }
+
+ if matches!(state.leios_notify, notify_proto::State::Busy) {
+ tracing::debug!("leios notification requested");
+ outbound.push_ready(BehaviorOutput::ExternalEvent(
+ ResponderEvent::EbNotificationRequested(pid.clone()),
+ ));
+ }
+ }
+}
diff --git a/pallas-network2/src/behavior/responder/mod.rs b/pallas-network2/src/behavior/responder/mod.rs
index 3267eeef..05decdbd 100644
--- a/pallas-network2/src/behavior/responder/mod.rs
+++ b/pallas-network2/src/behavior/responder/mod.rs
@@ -15,6 +15,8 @@ pub mod chainsync;
pub mod connection;
pub mod handshake;
pub mod keepalive;
+pub mod leiosfetch;
+pub mod leiosnotify;
pub mod peersharing;
pub mod txsubmission;
@@ -93,6 +95,8 @@ pub struct ResponderState {
pub(crate) blockfetch: proto::blockfetch::State,
pub(crate) chainsync: proto::chainsync::State,
pub(crate) tx_submission: proto::txsubmission::State,
+ pub(crate) leios_notify: proto::leiosnotify::State,
+ pub(crate) leios_fetch: proto::leiosfetch::State,
pub(crate) violation: bool,
pub(crate) error_count: u32,
pub(crate) violations_counter: Option>,
@@ -128,6 +132,16 @@ impl ResponderState {
}
}
+ /// Returns the negotiated N2N protocol version, if the handshake completed.
+ pub fn accepted_version(&self) -> Option {
+ super::accepted_version(&self.handshake)
+ }
+
+ /// Returns true if the negotiated version carries the Leios overlay.
+ pub fn supports_leios(&self) -> bool {
+ super::supports_leios(&self.handshake)
+ }
+
fn record_violation(&self, protocol: &'static str) {
if let Some(counter) = &self.violations_counter {
counter.add(1, &[opentelemetry::KeyValue::new("protocol", protocol)]);
@@ -209,6 +223,30 @@ impl ResponderState {
self.tx_submission = new;
}
+ AnyMessage::LeiosNotify(msg) => {
+ let result = self.leios_notify.apply(msg);
+
+ let Ok(new) = result else {
+ tracing::warn!("leios notify violation");
+ self.violation = true;
+ self.record_violation("leiosnotify");
+ return;
+ };
+
+ self.leios_notify = new;
+ }
+ AnyMessage::LeiosFetch(msg) => {
+ let result = self.leios_fetch.apply(msg);
+
+ let Ok(new) = result else {
+ tracing::warn!("leios fetch violation");
+ self.violation = true;
+ self.record_violation("leiosfetch");
+ return;
+ };
+
+ self.leios_fetch = new;
+ }
}
}
@@ -221,6 +259,8 @@ impl ResponderState {
self.blockfetch = proto::blockfetch::State::default();
self.chainsync = proto::chainsync::State::default();
self.tx_submission = proto::txsubmission::State::default();
+ self.leios_notify = proto::leiosnotify::State::default();
+ self.leios_fetch = proto::leiosfetch::State::default();
self.violation = false;
}
}
@@ -244,6 +284,24 @@ pub enum ResponderCommand {
ProvideBlocks(PeerId, Vec),
/// Send a list of peer addresses to a peer via peer-sharing.
ProvidePeers(PeerId, Vec),
+ /// Announce an EB to a peer via leios-notify (raw RB header CBOR).
+ ProvideEbAnnouncement(PeerId, proto::AnyCbor),
+ /// Offer an EB body to a peer via leios-notify, with its size in bytes.
+ ProvideEbOffer(PeerId, proto::EbId, u32),
+ /// Offer an EB's transactions to a peer via leios-notify.
+ ProvideEbTxsOffer(PeerId, proto::EbId),
+ /// Diffuse full votes to a peer inline via leios-notify.
+ ProvideVotes(PeerId, Vec),
+ /// Deliver an EB body to a peer via leios-fetch.
+ ProvideEb(PeerId, proto::leiosfetch::EndorserBlockCbor),
+ /// Deliver EB transactions to a peer via leios-fetch, echoing the request's
+ /// point and bitmaps.
+ ProvideEbTxs(
+ PeerId,
+ proto::EbId,
+ proto::leiosfetch::Bitmaps,
+ Vec,
+ ),
/// Ban a peer and disconnect them.
BanPeer(PeerId),
/// Disconnect a peer gracefully.
@@ -267,6 +325,12 @@ pub enum ResponderEvent {
PeersRequested(PeerId, u8),
/// A transaction was received from a peer via tx-submission.
TxReceived(PeerId, proto::txsubmission::EraTxBody),
+ /// A peer requested the next notification via leios-notify.
+ EbNotificationRequested(PeerId),
+ /// A peer requested an EB body via leios-fetch.
+ EbRequested(PeerId, proto::EbId),
+ /// A peer requested a subset of an EB's transactions via leios-fetch.
+ EbTxsRequested(PeerId, proto::EbId, proto::leiosfetch::Bitmaps),
}
/// The main responder behavior that handles inbound Cardano connections.
@@ -282,6 +346,8 @@ pub struct ResponderBehavior {
pub blockfetch: blockfetch::BlockFetchResponder,
pub peersharing: peersharing::PeerSharingResponder,
pub txsubmission: txsubmission::TxSubmissionResponder,
+ pub leiosnotify: leiosnotify::LeiosNotifyResponder,
+ pub leiosfetch: leiosfetch::LeiosFetchResponder,
pub peers: HashMap,
pub outbound: OutboundQueue,
@@ -306,6 +372,8 @@ impl Default for ResponderBehavior {
blockfetch: Default::default(),
peersharing: Default::default(),
txsubmission: Default::default(),
+ leiosnotify: Default::default(),
+ leiosfetch: Default::default(),
peers: Default::default(),
outbound: Default::default(),
violations_counter,
@@ -324,6 +392,8 @@ macro_rules! all_visitors {
$self
.txsubmission
.$method($pid, $state, &mut $self.outbound);
+ $self.leiosnotify.$method($pid, $state, &mut $self.outbound);
+ $self.leiosfetch.$method($pid, $state, &mut $self.outbound);
};
}
@@ -373,6 +443,14 @@ impl ResponderBehavior {
self.txsubmission
.visit_inbound_msg(pid, state, &mut self.outbound);
}
+ AnyMessage::LeiosNotify(_) => {
+ self.leiosnotify
+ .visit_inbound_msg(pid, state, &mut self.outbound);
+ }
+ AnyMessage::LeiosFetch(_) => {
+ self.leiosfetch
+ .visit_inbound_msg(pid, state, &mut self.outbound);
+ }
}
});
}
@@ -515,6 +593,15 @@ impl ResponderBehavior {
)));
}
+ /// Pushes a single message to be sent to the given peer.
+ fn send_msg(&mut self, pid: &PeerId, msg: AnyMessage) {
+ self.outbound
+ .push_ready(BehaviorOutput::InterfaceCommand(InterfaceCommand::Send(
+ pid.clone(),
+ msg,
+ )));
+ }
+
fn ban_peer(&mut self, pid: &PeerId) {
self.connection.banned_peers.insert(pid.clone());
self.outbound.push_ready(BehaviorOutput::InterfaceCommand(
@@ -609,6 +696,46 @@ impl Behavior for ResponderBehavior {
tracing::debug!("provide peers command");
self.provide_peers(&pid, peers);
}
+ ResponderCommand::ProvideEbAnnouncement(pid, header) => {
+ self.send_msg(
+ &pid,
+ AnyMessage::LeiosNotify(proto::leiosnotify::Message::BlockAnnouncement(header)),
+ );
+ }
+ ResponderCommand::ProvideEbOffer(pid, point, size) => {
+ self.send_msg(
+ &pid,
+ AnyMessage::LeiosNotify(proto::leiosnotify::Message::BlockOffer(point, size)),
+ );
+ }
+ ResponderCommand::ProvideEbTxsOffer(pid, point) => {
+ self.send_msg(
+ &pid,
+ AnyMessage::LeiosNotify(proto::leiosnotify::Message::BlockTxsOffer(point)),
+ );
+ }
+ ResponderCommand::ProvideVotes(pid, votes) => {
+ self.send_msg(
+ &pid,
+ AnyMessage::LeiosNotify(proto::leiosnotify::Message::Votes(votes)),
+ );
+ }
+ ResponderCommand::ProvideEb(pid, block) => {
+ self.send_msg(
+ &pid,
+ AnyMessage::LeiosFetch(proto::leiosfetch::Message::Block(block)),
+ );
+ }
+ ResponderCommand::ProvideEbTxs(pid, point, bitmaps, txs) => {
+ self.send_msg(
+ &pid,
+ AnyMessage::LeiosFetch(proto::leiosfetch::Message::BlockTxs {
+ point,
+ bitmaps,
+ txs,
+ }),
+ );
+ }
ResponderCommand::BanPeer(pid) => {
tracing::debug!("ban peer command");
self.ban_peer(&pid);
diff --git a/pallas-network2/src/emulation/happy.rs b/pallas-network2/src/emulation/happy.rs
index 46deb0c1..c9414f93 100644
--- a/pallas-network2/src/emulation/happy.rs
+++ b/pallas-network2/src/emulation/happy.rs
@@ -108,6 +108,50 @@ fn reply_block_fetch_ok(
}
}
+fn reply_leios_notify_ok(
+ pid: PeerId,
+ msg: proto::leiosnotify::Message,
+ queue: &mut emulation::ReplyQueue,
+) {
+ match msg {
+ proto::leiosnotify::Message::RequestNext => {
+ tracing::debug!("received leios notify request");
+
+ let offer =
+ proto::leiosnotify::Message::BlockOffer(proto::Point::new(1, vec![0xAB; 32]), 3);
+ queue.push_jittered_msg(pid, AnyMessage::LeiosNotify(offer), Duration::from_secs(0));
+ }
+ proto::leiosnotify::Message::Done => {}
+ _ => queue.push_jittered_disconnect(pid, Duration::from_secs(0)),
+ }
+}
+
+fn reply_leios_fetch_ok(
+ pid: PeerId,
+ msg: proto::leiosfetch::Message,
+ queue: &mut emulation::ReplyQueue,
+) {
+ match msg {
+ proto::leiosfetch::Message::BlockRequest(_) => {
+ tracing::debug!("received leios block request");
+
+ let block =
+ proto::leiosfetch::Message::Block(proto::AnyCbor::from_raw_bytes(b"abc".to_vec()));
+ queue.push_jittered_msg(pid, AnyMessage::LeiosFetch(block), Duration::from_secs(0));
+ }
+ proto::leiosfetch::Message::BlockTxsRequest(point, bitmaps) => {
+ let txs = proto::leiosfetch::Message::BlockTxs {
+ point,
+ bitmaps,
+ txs: vec![],
+ };
+ queue.push_jittered_msg(pid, AnyMessage::LeiosFetch(txs), Duration::from_secs(0));
+ }
+ proto::leiosfetch::Message::Done => {}
+ _ => queue.push_jittered_disconnect(pid, Duration::from_secs(0)),
+ }
+}
+
/// Emulation rules that always accept connections and reply successfully to all
/// mini-protocol messages (happy path).
#[derive(Default)]
@@ -129,7 +173,10 @@ impl emulation::Rules for HappyRules {
AnyMessage::KeepAlive(msg) => reply_keepalive_ok(pid, msg, jitter, queue),
AnyMessage::PeerSharing(msg) => reply_peer_sharing_ok(pid, msg, jitter, queue),
AnyMessage::BlockFetch(msg) => reply_block_fetch_ok(pid, msg, queue),
- _ => todo!(),
+ AnyMessage::LeiosNotify(msg) => reply_leios_notify_ok(pid, msg, queue),
+ AnyMessage::LeiosFetch(msg) => reply_leios_fetch_ok(pid, msg, queue),
+ // chain-sync / tx-submission are not modelled by the happy emulator.
+ _ => queue.push_jittered_disconnect(pid, Duration::from_secs(0)),
};
}
diff --git a/pallas-network2/src/protocol/cddl.rs b/pallas-network2/src/protocol/cddl.rs
new file mode 100644
index 00000000..63c18da7
--- /dev/null
+++ b/pallas-network2/src/protocol/cddl.rs
@@ -0,0 +1,69 @@
+//! Shared test scaffolding for exercising mini-protocol codecs against the
+//! vendored cardano-blueprint CDDL schemas.
+//!
+//! Today this hosts CBOR-vs-CDDL *conformance* checks; it is also where the
+//! upcoming round-trip helpers will live, so codec tests across protocols share
+//! one home rather than re-deriving the same plumbing.
+//!
+//! A protocol's conformance test needs only two protocol-specific things: a
+//! `self_contained()` that turns its blueprint `messages.cddl` into a schema
+//! `cddl-rs` can parse, and a table of [`conforms!`] cases (one per message).
+//! Everything common — the CDDL preprocessing, the scalar prelude, and the
+//! encode-then-validate kernel — lives here.
+//!
+//! The validating pieces are gated on the `blueprint` feature (they pull in the
+//! `cddl` crate and `include_str!` the submodule); the [`conforms!`] macro is
+//! always available so call sites compile feature-independently, expanding to a
+//! `#[cfg(feature = "blueprint")]` test.
+
+/// Emits one `#[test]` (gated on the `blueprint` feature) that encodes `$msg`
+/// with our `Encode` impl and asserts the bytes conform to CDDL rule `$rule`,
+/// using the schema produced by the `$schema` builder in scope at the call site.
+///
+/// ```ignore
+/// conforms!(done_conforms, self_contained, "msgClientDone", Message::Done);
+/// ```
+macro_rules! conforms {
+ ($name:ident, $schema:path, $rule:literal, $msg:expr) => {
+ #[cfg(feature = "blueprint")]
+ #[test]
+ fn $name() {
+ $crate::protocol::cddl::assert_conforms(
+ &$schema(),
+ $rule,
+ &pallas_codec::minicbor::to_vec(&$msg).unwrap(),
+ );
+ }
+ };
+}
+pub(crate) use conforms;
+
+/// Minimal prelude for the scalar types the blueprint CDDLs `;# import` from
+/// `base`. The opaque/word types collapse to `uint`/`bytes` because cddl-rs
+/// can't resolve the imports and the bit-widths aren't what these tests check.
+#[cfg(feature = "blueprint")]
+pub(crate) const BASE_PRELUDE: &str =
+ "slotno = uint\nhash = bytes\nword16 = uint\nword32 = uint\nword64 = uint\n";
+
+/// Rewrites a vendored blueprint CDDL into something cddl-rs can parse: drops the
+/// `;# import` pragmas and the `base.` namespace it can't resolve. Protocol-
+/// specific relaxations (e.g. opaque payloads to `any`) are layered on by the
+/// caller before appending [`BASE_PRELUDE`].
+#[cfg(feature = "blueprint")]
+pub(crate) fn preprocess(schema: &str) -> String {
+ schema
+ .lines()
+ .filter(|l| !l.trim_start().starts_with(";#"))
+ .collect::>()
+ .join("\n")
+ .replace("base.", "")
+}
+
+/// Validates `cbor` against rule `rule` of `schema`, panicking with the rule name
+/// on any mismatch. The single touch point for the `cddl` crate's API.
+#[cfg(feature = "blueprint")]
+pub(crate) fn assert_conforms(schema: &str, rule: &str, cbor: &[u8]) {
+ let doc = format!("start = {rule}\n{schema}");
+ ::cddl::validate_cbor_from_slice(&doc, cbor, None)
+ .unwrap_or_else(|e| panic!("`{rule}` does not conform to CDDL: {e}"));
+}
diff --git a/pallas-network2/src/protocol/common.rs b/pallas-network2/src/protocol/common.rs
index c164ecba..d9ac7e77 100644
--- a/pallas-network2/src/protocol/common.rs
+++ b/pallas-network2/src/protocol/common.rs
@@ -118,3 +118,16 @@ impl<'b> Decode<'b, ()> for Point {
}
}
}
+
+// ---------------------------------------------------------------------------
+// Leios shared wire primitives
+//
+// Reference to an Endorser Block, shared across the Leios mini-protocols
+// (`leiosnotify`, `leiosfetch`). It lives here next to [`Point`] following the
+// same convention used for other cross-protocol types. Heavy payloads (EB
+// bodies, transactions, votes) are carried as verbatim CBOR via
+// [`pallas_codec::utils::AnyCbor`] in the owning protocol modules.
+// ---------------------------------------------------------------------------
+
+/// Reference to an Endorser Block, encoded as a `[slot, eb_hash]` point.
+pub type EbId = Point;
diff --git a/pallas-network2/src/protocol/handshake/n2n.rs b/pallas-network2/src/protocol/handshake/n2n.rs
index 520da9d6..d921abff 100644
--- a/pallas-network2/src/protocol/handshake/n2n.rs
+++ b/pallas-network2/src/protocol/handshake/n2n.rs
@@ -12,9 +12,16 @@ const PROTOCOL_V11: u64 = 11;
const PROTOCOL_V12: u64 = 12;
const PROTOCOL_V13: u64 = 13;
const PROTOCOL_V14: u64 = 14;
+const PROTOCOL_V15: u64 = 15;
const PEER_SHARING_DISABLED: u8 = 0;
+/// Minimum N2N protocol version that carries the Leios overlay (the Dijkstra-era
+/// version 15). When a connection negotiates this version or higher, the
+/// leios-notify / leios-fetch mini-protocols are available. The version data
+/// shape is unchanged from v13/v14.
+pub const LEIOS_MIN_VERSION: u64 = PROTOCOL_V15;
+
impl VersionTable {
/// Builds a version table for v4 and above (deprecated, delegates to v7).
#[deprecated(note = "no longer supported by spec")]
@@ -118,6 +125,15 @@ impl VersionTable {
Some(query),
),
),
+ (
+ PROTOCOL_V15,
+ VersionData::new(
+ network_magic,
+ true,
+ Some(PEER_SHARING_DISABLED),
+ Some(query),
+ ),
+ ),
]
.into_iter()
.collect::>();
@@ -169,6 +185,15 @@ impl VersionTable {
Some(query),
),
),
+ (
+ PROTOCOL_V15,
+ VersionData::new(
+ network_magic,
+ true,
+ Some(PEER_SHARING_DISABLED),
+ Some(query),
+ ),
+ ),
]
.into_iter()
.collect::>();
diff --git a/pallas-network2/src/protocol/leiosfetch.rs b/pallas-network2/src/protocol/leiosfetch.rs
new file mode 100644
index 00000000..2075e009
--- /dev/null
+++ b/pallas-network2/src/protocol/leiosfetch.rs
@@ -0,0 +1,483 @@
+//! LeiosFetch mini-protocol implementation.
+//!
+//! Client-pull protocol for fetching Endorser Block (EB) bodies and their
+//! transactions (selected by a compact bitmap), discovered via
+//! [`super::leiosnotify`]. The client issues one request and the server replies
+//! with the matching response, returning to idle.
+//!
+//! Wire format and state machine follow the authoritative `leios-fetch` CDDL on
+//! the `leios-prototype` branch of cardano-blueprint (protocol id 19), which is
+//! the network spec of record while CIP-0164's network chapter stabilises.
+
+use std::collections::BTreeMap;
+
+use pallas_codec::minicbor::{Decode, Decoder, Encode, Encoder, decode, encode};
+
+use super::{AnyCbor, EbId, Error};
+
+/// Protocol channel number for node-to-node leios-fetch.
+pub const CHANNEL_ID: u16 = 19;
+
+/// Raw CBOR of an Endorser Block body (`{ hash32 => word32 }`).
+pub type EndorserBlockCbor = AnyCbor;
+
+/// Raw CBOR of a single transaction.
+pub type TxCbor = AnyCbor;
+
+/// A transaction-subset selector for leios-fetch block-txs requests.
+///
+/// Each key indexes a 64-transaction window (window `n` covers txs
+/// `64*n .. 64*n+63`); each set bit in the `u64` value selects a transaction
+/// within that window.
+///
+/// **Wire note:** this *must* serialize as an indefinite-length CBOR map
+/// (`0xbf … 0xff`). The Leios prototype rejects a definite-length map and resets
+/// the connection. Decoding accepts either form (a [`BTreeMap`] keeps key order
+/// deterministic).
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct Bitmaps(pub BTreeMap);
+
+impl Bitmaps {
+ /// Selects the first `count` transactions of an EB (txs `0..count`).
+ ///
+ /// Transactions are addressed within 64-tx windows; tx `i` of a window is the
+ /// **most-significant** bit (tx 0 → bit 63), matching the wire convention.
+ /// `count == 0` selects nothing.
+ pub fn all(count: usize) -> Self {
+ let mut m = BTreeMap::new();
+ let mut remaining = count;
+ let mut window = 0u16;
+ while remaining > 0 {
+ let bits = remaining.min(64);
+ // Top `bits` bits set (MSB-first: tx 0 of the window is bit 63).
+ let mask = if bits == 64 {
+ u64::MAX
+ } else {
+ !((1u64 << (64 - bits)) - 1)
+ };
+ m.insert(window, mask);
+ remaining -= bits;
+ window += 1;
+ }
+ Bitmaps(m)
+ }
+
+ /// Selects the transactions at the given sequence indices within an EB, using
+ /// the same MSB-first 64-tx window convention as [`Bitmaps::all`].
+ pub fn from_indices(indices: impl IntoIterator- ) -> Self {
+ let mut m = BTreeMap::new();
+ for offset in indices {
+ let window = (offset / 64) as u16;
+ let bit = 63 - (offset % 64);
+ *m.entry(window).or_insert(0) |= 1u64 << bit;
+ }
+ Bitmaps(m)
+ }
+}
+
+impl Encode<()> for Bitmaps {
+ fn encode(
+ &self,
+ e: &mut Encoder,
+ _ctx: &mut (),
+ ) -> Result<(), encode::Error> {
+ e.begin_map()?;
+ for (k, v) in &self.0 {
+ e.u16(*k)?;
+ e.u64(*v)?;
+ }
+ e.end()?;
+
+ Ok(())
+ }
+}
+
+impl<'b> Decode<'b, ()> for Bitmaps {
+ fn decode(d: &mut Decoder<'b>, _ctx: &mut ()) -> Result {
+ // minicbor's `BTreeMap` decoder transparently handles both definite and
+ // indefinite-length maps.
+ Ok(Bitmaps(d.decode()?))
+ }
+}
+
+/// A leios-fetch mini-protocol message.
+#[derive(Debug, Clone)]
+pub enum Message {
+ /// Client requests a complete EB body.
+ BlockRequest(EbId),
+ /// Server delivers an EB body (raw CBOR).
+ Block(EndorserBlockCbor),
+ /// Client requests a subset of an EB's transactions, selected by bitmap.
+ BlockTxsRequest(EbId, Bitmaps),
+ /// Server delivers transactions for an EB, echoing the request's point and
+ /// bitmaps: `[3, point, bitmaps, tx_list]`.
+ BlockTxs {
+ /// Echoed EB point.
+ point: EbId,
+ /// Echoed bitmap selector.
+ bitmaps: Bitmaps,
+ /// The requested transactions, as raw CBOR.
+ txs: Vec,
+ },
+ /// Client terminates the protocol.
+ Done,
+}
+
+/// A response delivered by the server, retained (with the EB it answers) in the
+/// idle state until the consumer drains it (mirrors the chain-sync `Data` idiom).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Response {
+ /// An EB body (raw CBOR).
+ Block(EndorserBlockCbor),
+ /// Transactions delivered for an EB. (The echoed point/bitmaps from the wire
+ /// are dropped — the EB is carried alongside the response by the state.)
+ BlockTxs {
+ /// The delivered transactions.
+ txs: Vec,
+ },
+}
+
+/// State machine for the leios-fetch mini-protocol.
+///
+/// The `Awaiting*` states retain the request parameters so a responder can serve
+/// them; the `Idle` state retains the delivered response — paired with the
+/// [`EbId`] it answers — until the consumer drains it.
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub enum State {
+ /// Client has agency; can issue a request or finish.
+ Idle(Option<(EbId, Response)>),
+ /// Server has agency; will deliver the requested EB body.
+ AwaitingBlock(EbId),
+ /// Server has agency; will deliver transactions for the requested EB.
+ AwaitingBlockTxs(EbId, Bitmaps),
+ /// The protocol has terminated.
+ Done,
+}
+
+impl Default for State {
+ fn default() -> Self {
+ State::Idle(None)
+ }
+}
+
+impl State {
+ /// Applies a message to the current state, returning the new state.
+ pub fn apply(&self, msg: &Message) -> Result {
+ match self {
+ State::Idle(_) => match msg {
+ Message::BlockRequest(p) => Ok(State::AwaitingBlock(p.clone())),
+ Message::BlockTxsRequest(p, b) => Ok(State::AwaitingBlockTxs(p.clone(), b.clone())),
+ Message::Done => Ok(State::Done),
+ _ => Err(Error::InvalidOutbound),
+ },
+ State::AwaitingBlock(eb) => match msg {
+ Message::Block(b) => {
+ Ok(State::Idle(Some((eb.clone(), Response::Block(b.clone())))))
+ }
+ _ => Err(Error::InvalidInbound),
+ },
+ State::AwaitingBlockTxs(eb, _) => match msg {
+ // The wire form echoes point/bitmaps; we keep only the txs and pair
+ // them with the EB from our request state.
+ Message::BlockTxs { txs, .. } => Ok(State::Idle(Some((
+ eb.clone(),
+ Response::BlockTxs { txs: txs.clone() },
+ )))),
+ _ => Err(Error::InvalidInbound),
+ },
+ State::Done => Err(Error::InvalidOutbound),
+ }
+ }
+
+ /// Takes any pending response (with the EB it answers), leaving the state
+ /// idle. Returns `None` if there is nothing to drain or the protocol is not
+ /// idle.
+ pub fn drain(&mut self) -> Option<(EbId, Response)> {
+ match self {
+ State::Idle(r) => r.take(),
+ _ => None,
+ }
+ }
+}
+
+impl Encode<()> for Message {
+ fn encode(
+ &self,
+ e: &mut Encoder,
+ _ctx: &mut (),
+ ) -> Result<(), encode::Error> {
+ match self {
+ Message::BlockRequest(point) => {
+ e.array(2)?.u16(0)?;
+ e.encode(point)?;
+ }
+ Message::Block(block) => {
+ e.array(2)?.u16(1)?;
+ e.encode(block)?;
+ }
+ Message::BlockTxsRequest(point, bitmaps) => {
+ e.array(3)?.u16(2)?;
+ e.encode(point)?;
+ e.encode(bitmaps)?;
+ }
+ Message::BlockTxs {
+ point,
+ bitmaps,
+ txs,
+ } => {
+ e.array(4)?.u16(3)?;
+ e.encode(point)?;
+ e.encode(bitmaps)?;
+ e.encode(txs)?;
+ }
+ Message::Done => {
+ e.array(1)?.u16(9)?;
+ }
+ }
+
+ Ok(())
+ }
+}
+
+impl<'b> Decode<'b, ()> for Message {
+ fn decode(d: &mut Decoder<'b>, _ctx: &mut ()) -> Result {
+ d.array()?;
+ let label = d.u16()?;
+
+ match label {
+ 0 => Ok(Message::BlockRequest(d.decode()?)),
+ 1 => Ok(Message::Block(d.decode()?)),
+ 2 => {
+ let point = d.decode()?;
+ let bitmaps = d.decode()?;
+ Ok(Message::BlockTxsRequest(point, bitmaps))
+ }
+ 3 => {
+ let point = d.decode()?;
+ let bitmaps = d.decode()?;
+ let txs = d.decode()?;
+ Ok(Message::BlockTxs {
+ point,
+ bitmaps,
+ txs,
+ })
+ }
+ 9 => Ok(Message::Done),
+ _ => Err(decode::Error::message(
+ "unknown variant for leiosfetch message",
+ )),
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::protocol::Point;
+ #[cfg(feature = "blueprint")]
+ use crate::protocol::cddl;
+ use crate::protocol::cddl::conforms;
+ use pallas_codec::minicbor;
+ use std::collections::BTreeMap;
+
+ fn point() -> EbId {
+ Point::Specific(99, vec![0xCD; 32])
+ }
+
+ fn bitmaps() -> Bitmaps {
+ let mut m = BTreeMap::new();
+ m.insert(0u16, 0xff00u64);
+ Bitmaps(m)
+ }
+
+ fn raw(bytes: [u8; 3]) -> AnyCbor {
+ AnyCbor::from_raw_bytes(minicbor::to_vec(bytes).unwrap())
+ }
+
+ fn reencode(msg: &Message) -> Vec {
+ minicbor::to_vec(msg).unwrap()
+ }
+
+ fn roundtrip_eq(msg: &Message) {
+ let bytes = reencode(msg);
+ let back: Message = minicbor::decode(&bytes).unwrap();
+ assert_eq!(reencode(&back), bytes);
+ }
+
+ #[test]
+ fn message_roundtrips() {
+ roundtrip_eq(&Message::BlockRequest(point()));
+ roundtrip_eq(&Message::Block(AnyCbor::from_raw_bytes(
+ minicbor::to_vec([1u8, 2]).unwrap(),
+ )));
+ roundtrip_eq(&Message::BlockTxsRequest(point(), bitmaps()));
+ roundtrip_eq(&Message::BlockTxs {
+ point: point(),
+ bitmaps: bitmaps(),
+ txs: vec![raw([8, 8, 8])],
+ });
+ roundtrip_eq(&Message::Done);
+ }
+
+ #[test]
+ fn block_txs_roundtrip() {
+ let msg = Message::BlockTxs {
+ point: point(),
+ bitmaps: bitmaps(),
+ txs: vec![raw([8, 8, 8])],
+ };
+ let bytes = reencode(&msg);
+ // envelope is a 4-element array: [tag=3, point, bitmaps, txs]
+ assert_eq!(bytes[0], 0x84);
+ let back: Message = minicbor::decode(&bytes).unwrap();
+ assert!(matches!(back, Message::BlockTxs { .. }));
+ }
+
+ #[test]
+ fn state_transitions_and_drain() {
+ assert_eq!(State::default(), State::Idle(None));
+ assert_eq!(
+ State::Idle(None)
+ .apply(&Message::BlockRequest(point()))
+ .unwrap(),
+ State::AwaitingBlock(point())
+ );
+
+ let mut idle = State::AwaitingBlock(point())
+ .apply(&Message::Block(raw([1, 2, 3])))
+ .unwrap();
+ assert_eq!(
+ idle,
+ State::Idle(Some((point(), Response::Block(raw([1, 2, 3])))))
+ );
+ assert_eq!(
+ idle.drain(),
+ Some((point(), Response::Block(raw([1, 2, 3]))))
+ );
+ assert_eq!(idle, State::Idle(None));
+ assert_eq!(idle.drain(), None);
+ }
+
+ #[test]
+ fn awaiting_states_retain_request_params() {
+ assert_eq!(
+ State::Idle(None)
+ .apply(&Message::BlockTxsRequest(point(), bitmaps()))
+ .unwrap(),
+ State::AwaitingBlockTxs(point(), bitmaps())
+ );
+ }
+
+ #[test]
+ fn illegal_transition_errors() {
+ assert!(matches!(
+ State::Idle(None).apply(&Message::Block(raw([1, 2, 3]))),
+ Err(Error::InvalidOutbound)
+ ));
+ assert!(matches!(
+ State::AwaitingBlock(point()).apply(&Message::BlockRequest(point())),
+ Err(Error::InvalidInbound)
+ ));
+ }
+
+ #[test]
+ fn bitmaps_all_is_msb_first() {
+ assert_eq!(Bitmaps::all(0).0.len(), 0);
+ assert_eq!(Bitmaps::all(1).0.get(&0), Some(&(1u64 << 63)));
+ assert_eq!(Bitmaps::all(64).0.get(&0), Some(&u64::MAX));
+ let two = Bitmaps::all(65);
+ assert_eq!(two.0.get(&0), Some(&u64::MAX));
+ assert_eq!(two.0.get(&1), Some(&(1u64 << 63)));
+ }
+
+ #[test]
+ fn bitmaps_from_indices_is_msb_first() {
+ // tx 0 -> window 0 bit 63; tx 65 -> window 1 bit 62.
+ let b = Bitmaps::from_indices([0usize, 65]);
+ assert_eq!(b.0.get(&0), Some(&(1u64 << 63)));
+ assert_eq!(b.0.get(&1), Some(&(1u64 << 62)));
+ }
+
+ #[test]
+ fn bitmaps_encode_is_indefinite() {
+ let mut m = BTreeMap::new();
+ m.insert(0u16, 0xffff_ffff_ffff_ffffu64);
+ m.insert(1u16, 0x0000_0000_0001_0000u64);
+ let bm = Bitmaps(m);
+
+ let bytes = minicbor::to_vec(&bm).unwrap();
+ // indefinite-length map marker, terminated by break
+ assert_eq!(bytes[0], 0xbf, "bitmaps must use an indefinite-length map");
+ assert_eq!(*bytes.last().unwrap(), 0xff, "must be break-terminated");
+
+ let back: Bitmaps = minicbor::decode(&bytes).unwrap();
+ assert_eq!(back, bm);
+ }
+
+ #[test]
+ fn bitmaps_decode_accepts_definite() {
+ // A definite-length map { 0: 1 } encoded as 0xa1 00 01
+ let definite = [0xa1u8, 0x00, 0x01];
+ let back: Bitmaps = minicbor::decode(&definite).unwrap();
+ assert_eq!(back.0.get(&0), Some(&1u64));
+ }
+
+ // --- CBOR-vs-CDDL conformance (run with `--features blueprint`) ---
+ //
+ // Each `conforms!` below emits one `#[test]` that encodes a sample message
+ // with our `Encode` impl and validates the bytes against the vendored
+ // cardano-blueprint leios-fetch CDDL (via the shared `cddl` helper),
+ // so a spec change (tag, arity, the bitmaps shape) fails the matching test.
+ // The EB body / txs are opaque raw CBOR here, so they are validated as `any`.
+
+ /// Turns the vendored leios-fetch CDDL into a schema cddl-rs can parse. On top
+ /// of the shared preprocessing this relaxes the opaque sub-structures (the EB
+ /// body and `tx.tx`) to `any`, since they are raw CBOR in our codec.
+ #[cfg(feature = "blueprint")]
+ fn self_contained() -> String {
+ let body = cddl::preprocess(include_str!(
+ "../../../cardano-blueprint/src/network/node-to-node/leios-fetch/messages.cddl"
+ ))
+ .replace(
+ "endorser_block = { * hash => word32 }",
+ "endorser_block = any",
+ )
+ .replace("tx.tx", "tx_tx");
+ format!("{body}\n{}tx_tx = any\n", cddl::BASE_PRELUDE)
+ }
+
+ conforms!(
+ block_request_conforms,
+ self_contained,
+ "msgLeiosBlockRequest",
+ Message::BlockRequest(point())
+ );
+ conforms!(
+ block_conforms,
+ self_contained,
+ "msgLeiosBlock",
+ Message::Block(raw([1, 2, 3]))
+ );
+ conforms!(
+ block_txs_request_conforms,
+ self_contained,
+ "msgLeiosBlockTxsRequest",
+ Message::BlockTxsRequest(point(), Bitmaps::all(8))
+ );
+ conforms!(
+ block_txs_conforms,
+ self_contained,
+ "msgLeiosBlockTxs",
+ Message::BlockTxs {
+ point: point(),
+ bitmaps: Bitmaps::all(8),
+ txs: vec![raw([1, 2, 3])],
+ }
+ );
+ conforms!(
+ done_conforms,
+ self_contained,
+ "msgClientDone",
+ Message::Done
+ );
+}
diff --git a/pallas-network2/src/protocol/leiosnotify.rs b/pallas-network2/src/protocol/leiosnotify.rs
new file mode 100644
index 00000000..a988da06
--- /dev/null
+++ b/pallas-network2/src/protocol/leiosnotify.rs
@@ -0,0 +1,323 @@
+//! LeiosNotify mini-protocol implementation.
+//!
+//! Server-push protocol by which a peer announces new Endorser Blocks (EBs),
+//! offers their bodies and transactions for eager fetching over
+//! [`super::leiosfetch`], and diffuses full votes inline. The client repeatedly
+//! asks for the next notification; the server replies with exactly one
+//! announcement/offer and returns to idle.
+//!
+//! Wire format and state machine follow the authoritative `leios-notify` CDDL on
+//! the `leios-prototype` branch of cardano-blueprint (protocol id 18), which is
+//! the network spec of record while CIP-0164's network chapter stabilises.
+
+use pallas_codec::minicbor::{Decode, Decoder, Encode, Encoder, decode, encode};
+
+use super::{AnyCbor, EbId, Error};
+
+/// Protocol channel number for node-to-node leios-notify.
+pub const CHANNEL_ID: u16 = 18;
+
+/// Raw CBOR of a single Leios vote (persistent or non-persistent).
+pub type VoteCbor = AnyCbor;
+
+/// A leios-notify mini-protocol message.
+#[derive(Debug, Clone)]
+pub enum Message {
+ /// Client requests the next notification.
+ RequestNext,
+ /// Server announces an EB via the raw CBOR of the announcing RB header.
+ BlockAnnouncement(AnyCbor),
+ /// Server offers an EB body it can deliver, with its size in bytes.
+ BlockOffer(EbId, u32),
+ /// Server offers the transactions of an EB it can deliver.
+ BlockTxsOffer(EbId),
+ /// Server diffuses full votes inline: `[4, [vote, ...]]`.
+ Votes(Vec),
+ /// Client terminates the protocol.
+ Done,
+}
+
+/// A notification delivered by the server, retained in the idle state until the
+/// consumer drains it (mirrors the chain-sync `Data` idiom).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Notification {
+ /// An EB announcement (raw CBOR of the announcing RB header).
+ BlockAnnouncement(AnyCbor),
+ /// An EB body is available, with its size in bytes.
+ BlockOffer(EbId, u32),
+ /// The transactions of an EB are available.
+ BlockTxsOffer(EbId),
+ /// Full votes diffused inline by the server.
+ Votes(Vec),
+}
+
+/// State machine for the leios-notify mini-protocol.
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub enum State {
+ /// Client has agency; can request the next notification or finish. Carries
+ /// any not-yet-drained notification delivered by the server.
+ Idle(Option),
+ /// Server has agency; will deliver one announcement/offer.
+ Busy,
+ /// The protocol has terminated.
+ Done,
+}
+
+impl Default for State {
+ fn default() -> Self {
+ State::Idle(None)
+ }
+}
+
+impl State {
+ /// Applies a message to the current state, returning the new state.
+ pub fn apply(&self, msg: &Message) -> Result {
+ match self {
+ State::Idle(_) => match msg {
+ Message::RequestNext => Ok(State::Busy),
+ Message::Done => Ok(State::Done),
+ _ => Err(Error::InvalidOutbound),
+ },
+ State::Busy => match msg {
+ Message::BlockAnnouncement(h) => Ok(State::Idle(Some(
+ Notification::BlockAnnouncement(h.clone()),
+ ))),
+ Message::BlockOffer(p, s) => {
+ Ok(State::Idle(Some(Notification::BlockOffer(p.clone(), *s))))
+ }
+ Message::BlockTxsOffer(p) => {
+ Ok(State::Idle(Some(Notification::BlockTxsOffer(p.clone()))))
+ }
+ Message::Votes(v) => Ok(State::Idle(Some(Notification::Votes(v.clone())))),
+ _ => Err(Error::InvalidInbound),
+ },
+ State::Done => Err(Error::InvalidOutbound),
+ }
+ }
+
+ /// Takes any pending notification, leaving the state idle. Returns `None` if
+ /// there is nothing to drain or the protocol is not idle.
+ pub fn drain(&mut self) -> Option {
+ match self {
+ State::Idle(n) => n.take(),
+ _ => None,
+ }
+ }
+}
+
+impl Encode<()> for Message {
+ fn encode(
+ &self,
+ e: &mut Encoder,
+ _ctx: &mut (),
+ ) -> Result<(), encode::Error> {
+ match self {
+ Message::RequestNext => {
+ e.array(1)?.u16(0)?;
+ }
+ Message::BlockAnnouncement(header) => {
+ e.array(2)?.u16(1)?;
+ e.encode(header)?;
+ }
+ Message::BlockOffer(point, size) => {
+ e.array(3)?.u16(2)?;
+ e.encode(point)?;
+ e.u32(*size)?;
+ }
+ Message::BlockTxsOffer(point) => {
+ e.array(2)?.u16(3)?;
+ e.encode(point)?;
+ }
+ Message::Votes(votes) => {
+ e.array(2)?.u16(4)?;
+ e.encode(votes)?;
+ }
+ Message::Done => {
+ e.array(1)?.u16(5)?;
+ }
+ }
+
+ Ok(())
+ }
+}
+
+impl<'b> Decode<'b, ()> for Message {
+ fn decode(d: &mut Decoder<'b>, _ctx: &mut ()) -> Result {
+ d.array()?;
+ let label = d.u16()?;
+
+ match label {
+ 0 => Ok(Message::RequestNext),
+ 1 => Ok(Message::BlockAnnouncement(d.decode()?)),
+ 2 => {
+ let point = d.decode()?;
+ let size = d.u32()?;
+ Ok(Message::BlockOffer(point, size))
+ }
+ 3 => Ok(Message::BlockTxsOffer(d.decode()?)),
+ 4 => Ok(Message::Votes(d.decode()?)),
+ 5 => Ok(Message::Done),
+ _ => Err(decode::Error::message(
+ "unknown variant for leiosnotify message",
+ )),
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::protocol::Point;
+ #[cfg(feature = "blueprint")]
+ use crate::protocol::cddl;
+ use crate::protocol::cddl::conforms;
+ use pallas_codec::minicbor;
+
+ fn point() -> EbId {
+ Point::Specific(42, vec![0xAB; 32])
+ }
+
+ fn roundtrip(msg: &Message) -> Message {
+ let bytes = minicbor::to_vec(msg).unwrap();
+ minicbor::decode(&bytes).unwrap()
+ }
+
+ #[test]
+ fn message_roundtrips() {
+ let cases = vec![
+ Message::RequestNext,
+ Message::BlockAnnouncement(AnyCbor::from_raw_bytes(
+ minicbor::to_vec([1u8, 2, 3]).unwrap(),
+ )),
+ Message::BlockOffer(point(), 12345),
+ Message::BlockTxsOffer(point()),
+ Message::Votes(vec![
+ AnyCbor::from_raw_bytes(minicbor::to_vec([9u8, 8, 7, 6]).unwrap()),
+ AnyCbor::from_raw_bytes(minicbor::to_vec([5u8, 4, 3, 2]).unwrap()),
+ ]),
+ Message::Done,
+ ];
+
+ for msg in &cases {
+ let back = roundtrip(msg);
+ // compare via re-encode since Message is not PartialEq
+ assert_eq!(
+ minicbor::to_vec(&back).unwrap(),
+ minicbor::to_vec(msg).unwrap()
+ );
+ }
+ }
+
+ #[test]
+ fn state_transitions_and_drain() {
+ let s = State::default();
+ assert_eq!(s, State::Idle(None));
+
+ let busy = s.apply(&Message::RequestNext).unwrap();
+ assert_eq!(busy, State::Busy);
+
+ let mut idle = busy.apply(&Message::BlockTxsOffer(point())).unwrap();
+ assert_eq!(
+ idle,
+ State::Idle(Some(Notification::BlockTxsOffer(point())))
+ );
+
+ // draining yields the notification once and leaves the state idle/empty
+ assert_eq!(idle.drain(), Some(Notification::BlockTxsOffer(point())));
+ assert_eq!(idle, State::Idle(None));
+ assert_eq!(idle.drain(), None);
+
+ let done = idle.apply(&Message::Done).unwrap();
+ assert_eq!(done, State::Done);
+ }
+
+ #[test]
+ fn illegal_transitions_error() {
+ // offer while idle is invalid
+ assert!(matches!(
+ State::Idle(None).apply(&Message::BlockTxsOffer(point())),
+ Err(Error::InvalidOutbound)
+ ));
+ // request while busy is invalid
+ assert!(matches!(
+ State::Busy.apply(&Message::RequestNext),
+ Err(Error::InvalidInbound)
+ ));
+ }
+
+ // --- CBOR-vs-CDDL conformance (run with `--features blueprint`) ---
+ //
+ // Each `conforms!` below emits one `#[test]` that encodes a sample message
+ // with our `Encode` impl and validates the bytes against the vendored
+ // cardano-blueprint leios-notify CDDL (via the shared `cddl` helper),
+ // so a spec change (tag, arity, the inline vote shape) fails the matching
+ // test.
+
+ /// Turns the vendored leios-notify CDDL into a schema cddl-rs can parse. This
+ /// protocol has no opaque sub-structures, so the shared preprocessing plus the
+ /// scalar prelude is all that's needed.
+ #[cfg(feature = "blueprint")]
+ fn self_contained() -> String {
+ let body = cddl::preprocess(include_str!(
+ "../../../cardano-blueprint/src/network/node-to-node/leios-notify/messages.cddl"
+ ));
+ format!("{body}\n{}", cddl::BASE_PRELUDE)
+ }
+
+ /// A conformant `vote = [slot, eb_hash, voter_id, vote_signature(.size 48)]`.
+ #[cfg(feature = "blueprint")]
+ fn vote() -> VoteCbor {
+ let mut buf = Vec::new();
+ Encoder::new(&mut buf)
+ .array(4)
+ .unwrap()
+ .u64(7)
+ .unwrap()
+ .bytes(&[0xAB; 32])
+ .unwrap()
+ .u16(3)
+ .unwrap()
+ .bytes(&[0xEE; 48])
+ .unwrap();
+ AnyCbor::from_raw_bytes(buf)
+ }
+
+ conforms!(
+ request_next_conforms,
+ self_contained,
+ "msgLeiosNotificationRequestNext",
+ Message::RequestNext
+ );
+ conforms!(
+ block_announcement_conforms,
+ self_contained,
+ "msgLeiosBlockAnnouncement",
+ Message::BlockAnnouncement(AnyCbor::from_raw_bytes(
+ minicbor::to_vec([1u8, 2, 3]).unwrap()
+ ))
+ );
+ conforms!(
+ block_offer_conforms,
+ self_contained,
+ "msgLeiosBlockOffer",
+ Message::BlockOffer(point(), 1234)
+ );
+ conforms!(
+ block_txs_offer_conforms,
+ self_contained,
+ "msgLeiosBlockTxsOffer",
+ Message::BlockTxsOffer(point())
+ );
+ conforms!(
+ votes_conforms,
+ self_contained,
+ "msgLeiosVotes",
+ Message::Votes(vec![vote()])
+ );
+ conforms!(
+ done_conforms,
+ self_contained,
+ "msgClientDone",
+ Message::Done
+ );
+}
diff --git a/pallas-network2/src/protocol/mod.rs b/pallas-network2/src/protocol/mod.rs
index c35208f0..9afdbc10 100644
--- a/pallas-network2/src/protocol/mod.rs
+++ b/pallas-network2/src/protocol/mod.rs
@@ -2,14 +2,20 @@
mod common;
+#[cfg(test)]
+pub(crate) mod cddl;
+
pub mod blockfetch;
pub mod chainsync;
pub mod handshake;
pub mod keepalive;
+pub mod leiosfetch;
+pub mod leiosnotify;
pub mod peersharing;
pub mod txsubmission;
pub use common::*;
+pub use pallas_codec::utils::AnyCbor;
/// Errors that can occur when applying a message to a mini-protocol state
/// machine.
diff --git a/pallas-network2/tests/harness/node.rs b/pallas-network2/tests/harness/node.rs
index 7ca40d51..bcb7c525 100644
--- a/pallas-network2/tests/harness/node.rs
+++ b/pallas-network2/tests/harness/node.rs
@@ -91,6 +91,10 @@ impl ResponderNode {
ResponderEvent::TxReceived(pid, _) => {
tracing::info!(%pid, "responder: tx received");
}
+ // Leios responder events are not exercised by this harness.
+ ResponderEvent::EbNotificationRequested(..)
+ | ResponderEvent::EbRequested(..)
+ | ResponderEvent::EbTxsRequested(..) => {}
}
events.push(event);