fedimint_server/consensus/aleph_bft/
network.rs

1use async_channel::Sender;
2use bitcoin::hashes::{Hash, sha256};
3use fedimint_core::PeerId;
4use fedimint_core::config::P2PMessage;
5use fedimint_core::db::{Database, IDatabaseTransactionOpsCoreTyped};
6use fedimint_core::encoding::Encodable;
7use fedimint_core::module::SerdeModuleEncoding;
8use fedimint_core::module::registry::ModuleRegistry;
9use fedimint_core::net::peers::{DynP2PConnections, Recipient};
10use fedimint_core::secp256k1::schnorr;
11use fedimint_core::session_outcome::SignedSessionOutcome;
12use fedimint_core::util::FmtCompact as _;
13use fedimint_logging::LOG_CONSENSUS;
14use parity_scale_codec::{Decode, Encode, IoReader};
15use tracing::{error, trace};
16
17use super::super::db::SignedSessionOutcomeKey;
18use super::data_provider::UnitData;
19use super::keychain::Keychain;
20
21#[derive(Debug, Clone, Eq, PartialEq)]
22pub struct Hasher;
23
24impl aleph_bft::Hasher for Hasher {
25    type Hash = [u8; 32];
26
27    fn hash(input: &[u8]) -> Self::Hash {
28        input.consensus_hash::<sha256::Hash>().to_byte_array()
29    }
30}
31
32pub type NetworkData = aleph_bft::NetworkData<
33    Hasher,
34    UnitData,
35    <Keychain as aleph_bft::Keychain>::Signature,
36    <Keychain as aleph_bft::MultiKeychain>::PartialMultisignature,
37>;
38
39pub struct Network {
40    connections: DynP2PConnections<P2PMessage>,
41    signed_outcomes_sender: Sender<(PeerId, SignedSessionOutcome)>,
42    signatures_sender: Sender<(PeerId, schnorr::Signature)>,
43    db: Database,
44}
45
46impl Network {
47    pub fn new(
48        connections: DynP2PConnections<P2PMessage>,
49        signed_outcomes_sender: Sender<(PeerId, SignedSessionOutcome)>,
50        signatures_sender: Sender<(PeerId, schnorr::Signature)>,
51        db: Database,
52    ) -> Self {
53        Self {
54            connections,
55            signed_outcomes_sender,
56            signatures_sender,
57            db,
58        }
59    }
60}
61
62#[async_trait::async_trait]
63impl aleph_bft::Network<NetworkData> for Network {
64    fn send(&self, network_data: NetworkData, recipient: aleph_bft::Recipient) {
65        // convert from aleph_bft::Recipient to session::Recipient
66        let recipient = match recipient {
67            aleph_bft::Recipient::Node(node_index) => {
68                // Aleph echoes back node indices taken from messages we received, so this
69                // index may have been chosen by a peer and may not name a peer at all.
70                let Some(peer_id) = super::to_peer_id(node_index) else {
71                    trace!(
72                        target: LOG_CONSENSUS,
73                        ?node_index,
74                        "Dropping Aleph BFT message addressed to an invalid node index"
75                    );
76
77                    return;
78                };
79
80                Recipient::Peer(peer_id)
81            }
82            aleph_bft::Recipient::Everyone => Recipient::Everyone,
83        };
84
85        self.connections
86            .send(recipient, P2PMessage::Aleph(network_data.encode()));
87    }
88
89    async fn next_event(&mut self) -> Option<NetworkData> {
90        loop {
91            let (peer_id, message) = self.connections.receive().await?;
92
93            match message {
94                P2PMessage::Aleph(bytes) => {
95                    match NetworkData::decode(&mut IoReader(bytes.as_slice())) {
96                        Ok(network_data) => {
97                            // in order to bound the RAM consumption of a session we have to bound
98                            // the size of an individual unit in memory
99                            if network_data.included_data().iter().all(UnitData::is_valid) {
100                                return Some(network_data);
101                            }
102
103                            error!(
104                                target: LOG_CONSENSUS,
105                                %peer_id,
106                                "Received invalid unit data"
107                            );
108                        }
109                        Err(err) => {
110                            error!(
111                                target: LOG_CONSENSUS,
112                                %peer_id,
113                                err = %err.fmt_compact(),
114                                "Failed to decode Aleph BFT network data"
115                            );
116                        }
117                    }
118                }
119                P2PMessage::SessionSignature(signature) => {
120                    self.signatures_sender.try_send((peer_id, signature)).ok();
121                }
122                P2PMessage::SessionIndex(their_session) => {
123                    if let Some(outcome) = self
124                        .db
125                        .begin_transaction_nc()
126                        .await
127                        .get_value(&SignedSessionOutcomeKey(their_session))
128                        .await
129                    {
130                        self.connections.send(
131                            Recipient::Peer(peer_id),
132                            P2PMessage::SignedSessionOutcome(SerdeModuleEncoding::from(&outcome)),
133                        );
134                    }
135                }
136                P2PMessage::SignedSessionOutcome(encoded_outcome) => {
137                    match encoded_outcome.try_into_inner(&ModuleRegistry::default()) {
138                        Ok(outcome) => {
139                            self.signed_outcomes_sender
140                                .try_send((peer_id, outcome))
141                                .ok();
142                        }
143                        Err(err) => {
144                            error!(
145                                target: LOG_CONSENSUS,
146                                %peer_id,
147                                err = %err.fmt_compact(),
148                                "Failed to decode SignedSessionOutcome"
149                            );
150                        }
151                    }
152                }
153                message => {
154                    error!(
155                        target: LOG_CONSENSUS,
156                        %peer_id,
157                        ?message,
158                        "Received unexpected p2p message variant"
159                    );
160                }
161            }
162        }
163    }
164}
fedimint_server/consensus/aleph_bft/network.rs

fedimint_server/consensus/aleph_bft/
network.rs
