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    traits::{
9        leaf_fetcher_network::LeafFetcherNetwork, node_implementation::NodeType,
10        signature_key::SignatureKey,
11    },
12    utils::StateAndDelta,
13};
14use tokio::sync::{mpsc, oneshot};
15
16use crate::{
17    block::SubmitError, coordinator::error::CoordinatorError, message::Proposal, state::UpdateLeaf,
18};
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<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            .map_err(QueryError::from)
83    }
84
85    pub async fn request_proposal(
86        &self,
87        view: ViewNumber,
88        leaf_commitment: Commitment<Leaf2<T>>,
89    ) -> Result<SignedProposal<T, Proposal<T>>, QueryError> {
90        let (respond, rx) = oneshot::channel();
91        self.call(
92            ClientRequest::RequestProposal {
93                view,
94                leaf_commitment,
95                respond,
96            },
97            rx,
98        )
99        .await?
100    }
101
102    pub async fn send_external_message(
103        &self,
104        payload: Vec<u8>,
105        recipient: T::SignatureKey,
106    ) -> Result<(), QueryError> {
107        let (respond, rx) = oneshot::channel();
108        self.call(
109            ClientRequest::SendExternalMessage {
110                payload,
111                recipient,
112                respond,
113            },
114            rx,
115        )
116        .await?
117    }
118
119    /// Proposal participation ratios; `None` queries the current epoch.
120    pub async fn proposal_participation(
121        &self,
122        epoch: Option<EpochNumber>,
123    ) -> Result<HashMap<T::SignatureKey, f64>, QueryError> {
124        let (respond, rx) = oneshot::channel();
125        self.call(ClientRequest::ProposalParticipation { epoch, respond }, rx)
126            .await
127    }
128
129    /// Vote participation ratios; `None` queries the current epoch.
130    pub async fn vote_participation(
131        &self,
132        epoch: Option<EpochNumber>,
133    ) -> Result<HashMap<<T::SignatureKey as SignatureKey>::VerificationKeyType, f64>, QueryError>
134    {
135        let (respond, rx) = oneshot::channel();
136        self.call(ClientRequest::VoteParticipation { epoch, respond }, rx)
137            .await
138    }
139
140    async fn call<A>(
141        &self,
142        request: ClientRequest<T>,
143        rx: oneshot::Receiver<A>,
144    ) -> Result<A, QueryError> {
145        self.tx
146            .send(request)
147            .await
148            .map_err(|_| QueryError::ChannelClosed)?;
149        rx.await.map_err(|_| QueryError::ResponseDropped)
150    }
151}
152
153/// The coordinator client owns the receive end of the request channel.
154///
155/// The coordinator holds this and calls [`next_request`](CoordinatorClient::next_request)
156/// to process incoming requests.
157pub struct CoordinatorClient<T: NodeType> {
158    rx: mpsc::Receiver<ClientRequest<T>>,
159    api: ClientApi<T>,
160}
161
162impl<T: NodeType> Default for CoordinatorClient<T> {
163    fn default() -> Self {
164        Self::new(NonZeroUsize::new(256).expect("256 > 0"))
165    }
166}
167
168impl<T: NodeType> CoordinatorClient<T> {
169    pub fn new(capacity: NonZeroUsize) -> Self {
170        let (tx, rx) = mpsc::channel(capacity.get());
171        Self {
172            rx,
173            api: ClientApi { tx },
174        }
175    }
176
177    pub fn handle(&self) -> &ClientApi<T> {
178        &self.api
179    }
180
181    pub(crate) async fn next_request(&mut self) -> Option<ClientRequest<T>> {
182        self.rx.recv().await
183    }
184}
185
186#[allow(clippy::large_enum_variant)]
187pub(crate) enum ClientRequest<T: NodeType> {
188    CurrentView(oneshot::Sender<ViewNumber>),
189    CurrentEpoch(oneshot::Sender<EpochNumber>),
190    DecidedLeaf(oneshot::Sender<Leaf2<T>>),
191    DecidedState(oneshot::Sender<Option<Arc<T::ValidatedState>>>),
192    UndecidedLeaves(oneshot::Sender<Vec<Leaf2<T>>>),
193    GetState {
194        view: ViewNumber,
195        respond: oneshot::Sender<Option<Arc<T::ValidatedState>>>,
196    },
197    GetStateAndDelta {
198        view: ViewNumber,
199        respond: oneshot::Sender<StateAndDelta<T>>,
200    },
201    ProposalParticipation {
202        epoch: Option<EpochNumber>,
203        respond: oneshot::Sender<HashMap<T::SignatureKey, f64>>,
204    },
205    VoteParticipation {
206        epoch: Option<EpochNumber>,
207        respond:
208            oneshot::Sender<HashMap<<T::SignatureKey as SignatureKey>::VerificationKeyType, f64>>,
209    },
210    UpdateLeaf {
211        update: UpdateLeaf<T>,
212        respond: oneshot::Sender<()>,
213    },
214    SubmitTransaction {
215        tx: T::Transaction,
216        respond: oneshot::Sender<Result<(), SubmitError>>,
217    },
218    RequestProposal {
219        view: ViewNumber,
220        leaf_commitment: Commitment<Leaf2<T>>,
221        respond: oneshot::Sender<Result<SignedProposal<T, Proposal<T>>, QueryError>>,
222    },
223    SendExternalMessage {
224        payload: Vec<u8>,
225        recipient: T::SignatureKey,
226        respond: oneshot::Sender<Result<(), QueryError>>,
227    },
228}
229
230#[derive(Debug, thiserror::Error)]
231#[non_exhaustive]
232pub enum QueryError {
233    #[error("failed to send request. coordinator channel closed")]
234    ChannelClosed,
235
236    #[error("coordinator dropped the response")]
237    ResponseDropped,
238
239    #[error("coordinator error: {0}")]
240    Coordinator(#[from] CoordinatorError),
241
242    #[error("transaction rejected: {0}")]
243    Rejected(#[from] SubmitError),
244}
245
246/// `LeafFetcherNetwork` impl that routes catchup direct-messages through
247/// the `Coordinator`'s single owned network via [`ClientApi`].
248///
249/// The membership layer gets a clone of this so it does not need its own
250/// network handle — the `Coordinator` is the only owner of the underlying
251/// network.
252pub struct ClientLeafFetcherNetwork<T: NodeType> {
253    client: ClientApi<T>,
254}
255
256impl<T: NodeType> ClientLeafFetcherNetwork<T> {
257    pub fn new(client: ClientApi<T>) -> Self {
258        Self { client }
259    }
260}
261
262#[async_trait]
263impl<T: NodeType> LeafFetcherNetwork<T> for ClientLeafFetcherNetwork<T> {
264    async fn send_leaf_request(
265        &self,
266        _: ViewNumber,
267        payload: Vec<u8>,
268        recipient: T::SignatureKey,
269    ) -> anyhow::Result<()> {
270        self.client
271            .send_external_message(payload, recipient)
272            .await?;
273        Ok(())
274    }
275
276    async fn send_leaf_response(
277        &self,
278        _: ViewNumber,
279        payload: Vec<u8>,
280        recipient: T::SignatureKey,
281    ) -> anyhow::Result<()> {
282        self.client
283            .send_external_message(payload, recipient)
284            .await?;
285        Ok(())
286    }
287}
hotshot_new_protocol/client.rs

hotshot_new_protocol/
client.rs
