feat(network2): add Leios mini-protocols (LeiosNotify, LeiosFetch)

examples/p2p-discovery/src/node.rs

@@ -143,6 +143,12 @@ impl<I: Interface<AnyMessage>> MyNode<I> {
             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();

examples/p2p-responder/src/main.rs

@@ -116,6 +116,18 @@ 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");
+            }
+            ResponderEvent::LeiosVotesRequested(pid, _) => {
+                tracing::info!(%pid, "leios votes requested");
+            }
         }
     }
 

pallas-network2/benches/harness/node.rs

@@ -73,6 +73,8 @@ impl ResponderNode {
                         manager.execute(ResponderCommand::ProvidePeers(pid.clone(), vec![]));
                     }
                     ResponderEvent::TxReceived(_, _) => {}
+                    // Leios responder events are not exercised by this Praos bench.
+                    _ => {}
                 }
 
                 events.push(event);

pallas-network2/src/behavior/initiator/leiosfetch.rs

@@ -0,0 +1,157 @@
+use std::collections::VecDeque;
+
+use crate::protocol::leiosfetch as fetch_proto;
+use crate::protocol::{Bitmaps, EbId, VoteId};
+
+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),
+    /// Fetch specific votes by id.
+    Votes(Vec<VoteId>),
+}
+
+/// Sub-behavior that fetches EB bodies, transactions and votes 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");
+    }
+
+    fn send_request(
+        &self,
+        pid: &PeerId,
+        request: &FetchRequest,
+        outbound: &mut OutboundQueue<InitiatorBehavior>,
+    ) {
+        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())
+            }
+            FetchRequest::Votes(ids) => fetch_proto::Message::VotesRequest(ids.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<InitiatorBehavior>,
+    ) {
+        if let Some(response) = state.leios_fetch.drain() {
+            outbound.push_ready(BehaviorOutput::ExternalEvent(InitiatorEvent::EbFetched(
+                pid.clone(),
+                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<InitiatorBehavior>,
+    ) {
+        self.dispatch(pid, state, outbound);
+    }
+
+    fn visit_housekeeping(
+        &mut self,
+        pid: &PeerId,
+        state: &mut InitiatorState,
+        outbound: &mut OutboundQueue<InitiatorBehavior>,
+    ) {
+        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);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::protocol::Point;
+    use crate::protocol::leiosfetch::Response;
+    use crate::protocol::{RawCbor, leiosfetch as lf};
+    use crate::{OutboundQueue, behavior::ConnectionState};
+
+    fn drain_outputs(
+        outbound: &mut OutboundQueue<InitiatorBehavior>,
+    ) -> Vec<BehaviorOutput<InitiatorBehavior>> {
+        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(Response::Block(RawCbor(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);
+    }
+}

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<InitiatorBehavior>) {
+        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<InitiatorBehavior>,
+    ) {
+        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<InitiatorBehavior>,
+    ) {
+        self.dispatch(pid, state, outbound);
+    }
+
+    fn visit_housekeeping(
+        &mut self,
+        pid: &PeerId,
+        state: &mut InitiatorState,
+        outbound: &mut OutboundQueue<InitiatorBehavior>,
+    ) {
+        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<InitiatorBehavior>,
+    ) -> Vec<BehaviorOutput<InitiatorBehavior>> {
+        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());
+    }
+}