hotshot_new_protocol/
client.rs

1use std::{collections::HashMap, num::NonZeroUsize, sync::Arc};
2
3use async_trait::async_trait;
4use committable::Commitment;
5use hotshot_types::{
6    data::{EpochNumber, Leaf2, ViewNumber},
7    message::Proposal as SignedProposal,
8    simple_certificate::QuorumCertificate2,
9    simple_vote::TimeoutVote2,
10    traits::{
11        leaf_fetcher_network::LeafFetcherNetwork, node_implementation::NodeType,
12        signature_key::SignatureKey,
13    },
14    utils::StateAndDelta,
15};
16use tokio::sync::{mpsc, mpsc::error::TrySendError, oneshot};
17
18use crate::{coordinator::error::CoordinatorError, message::Proposal, state::UpdateLeaf};
19
20#[derive(Clone)]
21pub struct ClientApi<T: NodeType> {
22    tx: mpsc::Sender<ClientRequest<T>>,
23}
24
25impl<T: NodeType> ClientApi<T> {
26    pub async fn current_view(&self) -> Result<ViewNumber, QueryError> {
27        let (tx, rx) = oneshot::channel();
28        self.call(ClientRequest::CurrentView(tx), rx).await
29    }
30
31    pub async fn current_epoch(&self) -> Result<Option<EpochNumber>, QueryError> {
32        let (tx, rx) = oneshot::channel();
33        self.call(ClientRequest::CurrentEpoch(tx), rx).await
34    }
35
36    pub async fn decided_leaf(&self) -> Result<Leaf2<T>, QueryError> {
37        let (tx, rx) = oneshot::channel();
38        self.call(ClientRequest::DecidedLeaf(tx), rx).await
39    }
40
41    pub async fn decided_state(&self) -> Result<Option<Arc<T::ValidatedState>>, QueryError> {
42        let (tx, rx) = oneshot::channel();
43        self.call(ClientRequest::DecidedState(tx), rx).await
44    }
45
46    pub async fn undecided_leaves(&self) -> Result<Vec<Leaf2<T>>, QueryError> {
47        let (tx, rx) = oneshot::channel();
48        self.call(ClientRequest::UndecidedLeaves(tx), rx).await
49    }
50
51    pub async fn state(
52        &self,
53        view: ViewNumber,
54    ) -> Result<Option<Arc<T::ValidatedState>>, QueryError> {
55        let (tx, rx) = oneshot::channel();
56        self.call(ClientRequest::GetState { view, respond: tx }, rx)
57            .await
58    }
59
60    pub async fn state_and_delta(&self, view: ViewNumber) -> Result<StateAndDelta<T>, QueryError> {
61        let (tx, rx) = oneshot::channel();
62        self.call(ClientRequest::GetStateAndDelta { view, respond: tx }, rx)
63            .await
64    }
65
66    pub async fn update_leaf(&self, update: UpdateLeaf<T>) -> Result<(), QueryError> {
67        let (tx, rx) = oneshot::channel();
68        self.call(
69            ClientRequest::UpdateLeaf {
70                update,
71                respond: tx,
72            },
73            rx,
74        )
75        .await
76    }
77
78    pub async fn submit_transaction(&self, tx: T::Transaction) -> Result<(), QueryError> {
79        let (respond, rx) = oneshot::channel();
80        self.call(ClientRequest::SubmitTransaction { tx, respond }, rx)
81            .await
82    }
83
84    pub async fn request_proposal(
85        &self,
86        view: ViewNumber,
87        leaf_commitment: Commitment<Leaf2<T>>,
88    ) -> Result<SignedProposal<T, Proposal<T>>, QueryError> {
89        let (respond, rx) = oneshot::channel();
90        self.call(
91            ClientRequest::RequestProposal {
92                view,
93                leaf_commitment,
94                respond,
95            },
96            rx,
97        )
98        .await?
99    }
100
101    pub async fn send_external_message(
102        &self,
103        payload: Vec<u8>,
104        recipient: T::SignatureKey,
105    ) -> Result<(), QueryError> {
106        let (respond, rx) = oneshot::channel();
107        self.call(
108            ClientRequest::SendExternalMessage {
109                payload,
110                recipient,
111                respond,
112            },
113            rx,
114        )
115        .await?
116    }
117
118    /// Proposal participation ratios; `None` queries the current epoch.
119    pub async fn proposal_participation(
120        &self,
121        epoch: Option<EpochNumber>,
122    ) -> Result<HashMap<T::SignatureKey, f64>, QueryError> {
123        let (respond, rx) = oneshot::channel();
124        self.call(ClientRequest::ProposalParticipation { epoch, respond }, rx)
125            .await
126    }
127
128    /// Vote participation ratios; `None` queries the current epoch.
129    pub async fn vote_participation(
130        &self,
131        epoch: Option<EpochNumber>,
132    ) -> Result<HashMap<<T::SignatureKey as SignatureKey>::VerificationKeyType, f64>, QueryError>
133    {
134        let (respond, rx) = oneshot::channel();
135        self.call(ClientRequest::VoteParticipation { epoch, respond }, rx)
136            .await
137    }
138
139    /// Forward a legacy `TimeoutVote2` into the new-protocol timeout collectors.
140    pub fn try_submit_legacy_timeout_vote(&self, vote: TimeoutVote2<T>) -> Result<(), QueryError> {
141        self.try_send(ClientRequest::SubmitTimeoutVote { vote })
142    }
143
144    /// Forward the last legacy view's QC so the first new-protocol leader can
145    /// propose on it even if the cutover seed was snapshotted before it formed.
146    pub fn try_submit_legacy_high_qc(&self, qc: QuorumCertificate2<T>) -> Result<(), QueryError> {
147        self.try_send(ClientRequest::SubmitLegacyHighQc { qc })
148    }
149
150    /// Bridge sends must never block on a coordinator that hasn't started:
151    /// requests queue in the bounded channel and are dropped when it is full.
152    fn try_send(&self, request: ClientRequest<T>) -> Result<(), QueryError> {
153        self.tx.try_send(request).map_err(|err| match err {
154            TrySendError::Closed(_) => QueryError::ChannelClosed,
155            TrySendError::Full(_) => QueryError::ChannelFull,
156        })
157    }
158
159    async fn call<A>(
160        &self,
161        request: ClientRequest<T>,
162        rx: oneshot::Receiver<A>,
163    ) -> Result<A, QueryError> {
164        self.tx
165            .send(request)
166            .await
167            .map_err(|_| QueryError::ChannelClosed)?;
168        rx.await.map_err(|_| QueryError::ResponseDropped)
169    }
170}
171
172/// The coordinator client owns the receive end of the request channel.
173///
174/// The coordinator holds this and calls [`next_request`](CoordinatorClient::next_request)
175/// to process incoming requests.
176pub struct CoordinatorClient<T: NodeType> {
177    rx: mpsc::Receiver<ClientRequest<T>>,
178    api: ClientApi<T>,
179}
180
181impl<T: NodeType> Default for CoordinatorClient<T> {
182    fn default() -> Self {
183        Self::new(NonZeroUsize::new(256).expect("256 > 0"))
184    }
185}
186
187impl<T: NodeType> CoordinatorClient<T> {
188    pub fn new(capacity: NonZeroUsize) -> Self {
189        let (tx, rx) = mpsc::channel(capacity.get());
190        Self {
191            rx,
192            api: ClientApi { tx },
193        }
194    }
195
196    pub fn handle(&self) -> &ClientApi<T> {
197        &self.api
198    }
199
200    pub(crate) async fn next_request(&mut self) -> Option<ClientRequest<T>> {
201        self.rx.recv().await
202    }
203}
204
205#[allow(clippy::large_enum_variant)]
206pub(crate) enum ClientRequest<T: NodeType> {
207    CurrentView(oneshot::Sender<ViewNumber>),
208    CurrentEpoch(oneshot::Sender<Option<EpochNumber>>),
209    DecidedLeaf(oneshot::Sender<Leaf2<T>>),
210    DecidedState(oneshot::Sender<Option<Arc<T::ValidatedState>>>),
211    UndecidedLeaves(oneshot::Sender<Vec<Leaf2<T>>>),
212    GetState {
213        view: ViewNumber,
214        respond: oneshot::Sender<Option<Arc<T::ValidatedState>>>,
215    },
216    GetStateAndDelta {
217        view: ViewNumber,
218        respond: oneshot::Sender<StateAndDelta<T>>,
219    },
220    ProposalParticipation {
221        epoch: Option<EpochNumber>,
222        respond: oneshot::Sender<HashMap<T::SignatureKey, f64>>,
223    },
224    VoteParticipation {
225        epoch: Option<EpochNumber>,
226        respond:
227            oneshot::Sender<HashMap<<T::SignatureKey as SignatureKey>::VerificationKeyType, f64>>,
228    },
229    UpdateLeaf {
230        update: UpdateLeaf<T>,
231        respond: oneshot::Sender<()>,
232    },
233    SubmitTransaction {
234        tx: T::Transaction,
235        respond: oneshot::Sender<()>,
236    },
237    RequestProposal {
238        view: ViewNumber,
239        leaf_commitment: Commitment<Leaf2<T>>,
240        respond: oneshot::Sender<Result<SignedProposal<T, Proposal<T>>, QueryError>>,
241    },
242    SendExternalMessage {
243        payload: Vec<u8>,
244        recipient: T::SignatureKey,
245        respond: oneshot::Sender<Result<(), QueryError>>,
246    },
247    SubmitTimeoutVote {
248        vote: TimeoutVote2<T>,
249    },
250    SubmitLegacyHighQc {
251        qc: QuorumCertificate2<T>,
252    },
253}
254
255#[derive(Debug, thiserror::Error)]
256#[non_exhaustive]
257pub enum QueryError {
258    #[error("failed to send request. coordinator channel closed")]
259    ChannelClosed,
260
261    #[error("coordinator dropped the response")]
262    ResponseDropped,
263
264    #[error("request dropped: coordinator request queue is full")]
265    ChannelFull,
266
267    #[error("coordinator error: {0}")]
268    Coordinator(#[from] CoordinatorError),
269}
270
271/// `LeafFetcherNetwork` impl that routes catchup direct-messages through
272/// the `Coordinator`'s single owned network via [`ClientApi`].
273///
274/// The membership layer gets a clone of this so it does not need its own
275/// network handle — the `Coordinator` is the only owner of the underlying
276/// `ConnectedNetwork`.
277pub struct ClientLeafFetcherNetwork<T: NodeType> {
278    client: ClientApi<T>,
279}
280
281impl<T: NodeType> ClientLeafFetcherNetwork<T> {
282    pub fn new(client: ClientApi<T>) -> Self {
283        Self { client }
284    }
285}
286
287#[async_trait]
288impl<T: NodeType> LeafFetcherNetwork<T> for ClientLeafFetcherNetwork<T> {
289    async fn send_leaf_request(
290        &self,
291        _: ViewNumber,
292        payload: Vec<u8>,
293        recipient: T::SignatureKey,
294    ) -> anyhow::Result<()> {
295        self.client
296            .send_external_message(payload, recipient)
297            .await?;
298        Ok(())
299    }
300
301    async fn send_leaf_response(
302        &self,
303        _: ViewNumber,
304        payload: Vec<u8>,
305        recipient: T::SignatureKey,
306    ) -> anyhow::Result<()> {
307        self.client
308            .send_external_message(payload, recipient)
309            .await?;
310        Ok(())
311    }
312}
hotshot_new_protocol/client.rs

hotshot_new_protocol/
client.rs
