espresso_node/
external_event_handler.rs

1//! Should probably rename this to "external" or something
2
3use std::sync::Arc;
4
5use anyhow::{Context, Result, bail};
6use espresso_types::{PubKey, SeqTypes, v0::traits::SequencerPersistence};
7use hotshot_new_protocol::client::ClientApi;
8use hotshot_types::{
9    message::{Message, MessageKind},
10    traits::metrics::Counter,
11};
12use request_response::network::Bytes;
13use serde::{Deserialize, Serialize};
14use tokio::sync::mpsc::{Receiver, Sender, error::TrySendError};
15use vbs::{BinarySerializer, bincode_serializer::BincodeSerializer, version::StaticVersion};
16
17use crate::{consensus_handle::ConsensusHandle, context::TaskList};
18
19/// An external message that can be sent to or received from a node
20#[derive(Debug, Serialize, Deserialize, Clone)]
21pub enum ExternalMessage {
22    RequestResponse(#[serde(with = "serde_bytes")] Vec<u8>),
23}
24
25/// The external event handler
26#[derive(Clone)]
27pub struct ExternalEventHandler {
28    /// The sender to the request-response protocol
29    request_response_sender: Sender<Bytes>,
30
31    /// Counts inbound request-response messages dropped because the channel was full
32    dropped_messages: Arc<dyn Counter>,
33}
34
35#[derive(Debug)]
36pub enum OutboundMessage {
37    Direct(MessageKind<SeqTypes>, PubKey),
38}
39
40impl ExternalEventHandler {
41    pub async fn new<P>(
42        tasks: &mut TaskList,
43        request_response_sender: Sender<Bytes>,
44        outbound_message_receiver: Receiver<OutboundMessage>,
45        consensus_handle: Arc<ConsensusHandle<SeqTypes, Arc<P>>>,
46        public_key: PubKey,
47        dropped_messages: Arc<dyn Counter>,
48    ) -> Result<Self>
49    where
50        P: SequencerPersistence,
51    {
52        // Spawn the outbound message handling loop
53        tasks.spawn(
54            "ExternalEventHandler",
55            Self::outbound_message_loop(outbound_message_receiver, consensus_handle, public_key),
56        );
57
58        Ok(Self {
59            request_response_sender,
60            dropped_messages,
61        })
62    }
63
64    /// Handles an event
65    ///
66    /// # Errors
67    /// If the message type is unknown or if there is an error serializing or deserializing the message
68    pub async fn handle_event(&self, external_message_bytes: &[u8]) -> Result<()> {
69        // Deserialize the external message
70        let external_message = bincode::deserialize(external_message_bytes)
71            .with_context(|| "Failed to deserialize external message")?;
72
73        // Match the type
74        match external_message {
75            ExternalMessage::RequestResponse(request_response) => {
76                match self
77                    .request_response_sender
78                    .try_send(request_response.into())
79                {
80                    Ok(()) => Ok(()),
81                    Err(TrySendError::Full(..)) => {
82                        self.dropped_messages.add(1);
83                        bail!("request-response channel full")
84                    },
85                    Err(TrySendError::Closed(..)) => bail!("request-response channel closed"),
86                }
87            },
88        }
89    }
90
91    /// The main loop for sending outbound messages.
92    async fn outbound_message_loop<P>(
93        mut receiver: Receiver<OutboundMessage>,
94        consensus_handle: Arc<ConsensusHandle<SeqTypes, Arc<P>>>,
95        public_key: PubKey,
96    ) where
97        P: SequencerPersistence,
98    {
99        while let Some(message) = receiver.recv().await {
100            let Some(client_api) = consensus_handle.client_api().await else {
101                tracing::warn!("dropping an outbound message: consensus has shut down");
102                continue;
103            };
104            Self::send_via_coordinator(&client_api, message, public_key).await;
105        }
106    }
107
108    /// Send an outbound message through the coordinator's network.
109    ///
110    /// The coordinator's network sends external payloads over the wire
111    /// verbatim, so they must be self-framing: wrap the payload in the
112    /// versioned `Message` envelope, which the receiving side's fallback
113    /// decoder recognizes and unwraps.
114    async fn send_via_coordinator(
115        client_api: &ClientApi<SeqTypes>,
116        message: OutboundMessage,
117        public_key: PubKey,
118    ) {
119        match message {
120            OutboundMessage::Direct(kind @ MessageKind::External(_), recipient) => {
121                let message = Message {
122                    sender: public_key,
123                    kind,
124                };
125                let message_bytes =
126                    match BincodeSerializer::<StaticVersion<0, 0>>::serialize(&message) {
127                        Ok(message_bytes) => message_bytes,
128                        Err(err) => {
129                            tracing::warn!("Failed to serialize direct message: {}", err);
130                            return;
131                        },
132                    };
133                if let Err(err) = client_api
134                    .send_external_message(message_bytes, recipient)
135                    .await
136                {
137                    tracing::warn!(%err, "failed to send external message via coordinator");
138                }
139            },
140            other => {
141                tracing::warn!(message = ?other, "dropping a non-external outbound message");
142            },
143        }
144    }
145}
espresso_node/external_event_handler.rs

espresso_node/
external_event_handler.rs
