fedimint_server/consensus/aleph_bft/
network.rs

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

fedimint_server/consensus/aleph_bft/
network.rs
