pallas_network/miniprotocols/chainsync/
client.rs

1use pallas_codec::Fragment;
2use std::marker::PhantomData;
3use thiserror::Error;
4use tracing::debug;
5
6use crate::miniprotocols::Point;
7use crate::multiplexer;
8
9use super::{BlockContent, HeaderContent, IntersectResponse, Message, State, Tip};
10
11/// Errors produced by the chain-sync client agent.
12#[derive(Error, Debug)]
13pub enum ClientError {
14    /// Tried to receive while we hold agency.
15    #[error("attempted to receive message while agency is ours")]
16    AgencyIsOurs,
17
18    /// Tried to send while the peer holds agency.
19    #[error("attempted to send message while agency is theirs")]
20    AgencyIsTheirs,
21
22    /// Inbound message is not valid for the current state.
23    #[error("inbound message is not valid for current state")]
24    InvalidInbound,
25
26    /// Outbound message is not valid for the current state.
27    #[error("outbound message is not valid for current state")]
28    InvalidOutbound,
29
30    /// None of the points offered by the client are on the server's chain.
31    #[error("no intersection point found")]
32    IntersectionNotFound,
33
34    /// Underlying multiplexer error.
35    #[error("error while sending or receiving data through the channel")]
36    Plexer(multiplexer::Error),
37}
38
39/// Outcome of a single `request_next` step.
40#[derive(Debug)]
41pub enum NextResponse<CONTENT> {
42    /// Chain advances with new content.
43    RollForward(CONTENT, Tip),
44    /// Chain rolls back to the given point.
45    RollBackward(Point, Tip),
46    /// No update is available yet; the server is awaiting one.
47    Await,
48}
49
50/// Chain-sync client agent generic over the content type (`HeaderContent` for
51/// node-to-node, `BlockContent` for node-to-client).
52pub struct Client<O>(State, multiplexer::ChannelBuffer, PhantomData<O>)
53where
54    Message<O>: Fragment;
55
56impl<O> Client<O>
57where
58    Message<O>: Fragment,
59{
60    /// Constructs a new ChainSync `Client` instance.
61    ///
62    /// # Arguments
63    ///
64    /// * `channel` - An instance of `multiplexer::AgentChannel` to be used for
65    ///   communication.
66    pub fn new(channel: multiplexer::AgentChannel) -> Self {
67        Self(
68            State::Idle,
69            multiplexer::ChannelBuffer::new(channel),
70            PhantomData {},
71        )
72    }
73
74    /// Returns the current state of the client.
75    pub fn state(&self) -> &State {
76        &self.0
77    }
78
79    /// Checks if the client is done.
80    pub fn is_done(&self) -> bool {
81        self.0 == State::Done
82    }
83
84    /// Checks if the client has agency.
85    pub fn has_agency(&self) -> bool {
86        match self.state() {
87            State::Idle => true,
88            State::CanAwait => false,
89            State::MustReply => false,
90            State::Intersect => false,
91            State::Done => false,
92        }
93    }
94
95    fn assert_agency_is_ours(&self) -> Result<(), ClientError> {
96        if !self.has_agency() {
97            Err(ClientError::AgencyIsTheirs)
98        } else {
99            Ok(())
100        }
101    }
102
103    fn assert_agency_is_theirs(&self) -> Result<(), ClientError> {
104        if self.has_agency() {
105            Err(ClientError::AgencyIsOurs)
106        } else {
107            Ok(())
108        }
109    }
110
111    fn assert_outbound_state(&self, msg: &Message<O>) -> Result<(), ClientError> {
112        match (&self.0, msg) {
113            (State::Idle, Message::RequestNext) => Ok(()),
114            (State::Idle, Message::FindIntersect(_)) => Ok(()),
115            (State::Idle, Message::Done) => Ok(()),
116            _ => Err(ClientError::InvalidOutbound),
117        }
118    }
119
120    fn assert_inbound_state(&self, msg: &Message<O>) -> Result<(), ClientError> {
121        match (&self.0, msg) {
122            (State::CanAwait, Message::RollForward(_, _)) => Ok(()),
123            (State::CanAwait, Message::RollBackward(_, _)) => Ok(()),
124            (State::CanAwait, Message::AwaitReply) => Ok(()),
125            (State::MustReply, Message::RollForward(_, _)) => Ok(()),
126            (State::MustReply, Message::RollBackward(_, _)) => Ok(()),
127            (State::Intersect, Message::IntersectFound(_, _)) => Ok(()),
128            (State::Intersect, Message::IntersectNotFound(_)) => Ok(()),
129            _ => Err(ClientError::InvalidInbound),
130        }
131    }
132
133    /// Sends a message to the server
134    ///
135    /// # Arguments
136    ///
137    /// * `msg` - A reference to the `Message` to be sent.
138    ///
139    /// # Errors
140    ///
141    /// Returns an error if the agency is not ours or if the outbound state is
142    /// invalid.
143    pub async fn send_message(&mut self, msg: &Message<O>) -> Result<(), ClientError> {
144        self.assert_agency_is_ours()?;
145        self.assert_outbound_state(msg)?;
146
147        self.1
148            .send_msg_chunks(msg)
149            .await
150            .map_err(ClientError::Plexer)?;
151
152        Ok(())
153    }
154
155    /// Receives the next message from the server.
156    ///
157    /// # Errors
158    ///
159    /// Returns an error if the agency is not theirs or if the inbound state is
160    /// invalid.
161    pub async fn recv_message(&mut self) -> Result<Message<O>, ClientError> {
162        self.assert_agency_is_theirs()?;
163
164        let msg = self.1.recv_full_msg().await.map_err(ClientError::Plexer)?;
165
166        self.assert_inbound_state(&msg)?;
167
168        Ok(msg)
169    }
170
171    /// Sends a FindIntersect message to the server.
172    ///
173    /// # Arguments
174    ///
175    /// * `points` - A vector of `Point` instances representing the points of
176    ///   intersection.
177    ///
178    /// # Errors
179    ///
180    /// Returns an error if the message cannot be sent or if it's not valid for
181    /// the current state of the client.
182    pub async fn send_find_intersect(&mut self, points: Vec<Point>) -> Result<(), ClientError> {
183        let msg = Message::FindIntersect(points);
184        self.send_message(&msg).await?;
185        self.0 = State::Intersect;
186
187        debug!("send find intersect");
188
189        Ok(())
190    }
191
192    /// Receives an IntersectResponse message from the server.
193    ///
194    /// # Errors
195    ///
196    /// Returns an error if the inbound message is invalid.
197    pub async fn recv_intersect_response(&mut self) -> Result<IntersectResponse, ClientError> {
198        debug!("waiting for intersect response");
199
200        match self.recv_message().await? {
201            Message::IntersectFound(point, tip) => {
202                self.0 = State::Idle;
203                Ok((Some(point), tip))
204            }
205            Message::IntersectNotFound(tip) => {
206                self.0 = State::Idle;
207                Ok((None, tip))
208            }
209            _ => Err(ClientError::InvalidInbound),
210        }
211    }
212
213    /// Finds the intersection point between the client's and server's chains.
214    ///
215    /// # Arguments
216    ///
217    /// * `points` - A vector of `Point` instances representing the points of
218    ///   intersection.
219    ///
220    /// # Errors
221    ///
222    /// Returns an error if the intersection point cannot be found or if there
223    /// is a communication error.
224    pub async fn find_intersect(
225        &mut self,
226        points: Vec<Point>,
227    ) -> Result<IntersectResponse, ClientError> {
228        self.send_find_intersect(points).await?;
229        self.recv_intersect_response().await
230    }
231
232    /// Send a `RequestNext` message and transition to `CanAwait`.
233    pub async fn send_request_next(&mut self) -> Result<(), ClientError> {
234        let msg = Message::RequestNext;
235        self.send_message(&msg).await?;
236        self.0 = State::CanAwait;
237
238        Ok(())
239    }
240
241    /// Receives a response while the client is in the CanAwait state.
242    ///
243    /// # Errors
244    ///
245    /// Returns an error if the inbound message is invalid.
246    pub async fn recv_while_can_await(&mut self) -> Result<NextResponse<O>, ClientError> {
247        match self.recv_message().await? {
248            Message::AwaitReply => {
249                self.0 = State::MustReply;
250                Ok(NextResponse::Await)
251            }
252            Message::RollForward(a, b) => {
253                self.0 = State::Idle;
254                Ok(NextResponse::RollForward(a, b))
255            }
256            Message::RollBackward(a, b) => {
257                self.0 = State::Idle;
258                Ok(NextResponse::RollBackward(a, b))
259            }
260            _ => Err(ClientError::InvalidInbound),
261        }
262    }
263
264    /// Receives a response while the client is in the MustReply state.
265    ///
266    /// # Errors
267    ///
268    /// Returns an error if the inbound message is invalid.
269    pub async fn recv_while_must_reply(&mut self) -> Result<NextResponse<O>, ClientError> {
270        match self.recv_message().await? {
271            Message::RollForward(a, b) => {
272                self.0 = State::Idle;
273                Ok(NextResponse::RollForward(a, b))
274            }
275            Message::RollBackward(a, b) => {
276                self.0 = State::Idle;
277                Ok(NextResponse::RollBackward(a, b))
278            }
279            _ => Err(ClientError::InvalidInbound),
280        }
281    }
282
283    /// Sends a RequestNext message to the server.
284    ///
285    /// # Errors
286    ///
287    /// Returns an error if the message cannot be sent or if the state is not
288    /// idle.
289    pub async fn request_next(&mut self) -> Result<NextResponse<O>, ClientError> {
290        debug!("requesting next block");
291
292        self.send_request_next().await?;
293
294        self.recv_while_can_await().await
295    }
296
297    /// Either requests the next block, or waits for one to become available.
298    ///
299    /// # Errors
300    ///
301    /// Returns an error if the message cannot be sent, or if the inbound
302    /// message is invalid
303    pub async fn request_or_await_next(&mut self) -> Result<NextResponse<O>, ClientError> {
304        if self.has_agency() {
305            self.request_next().await
306        } else {
307            self.recv_while_must_reply().await
308        }
309    }
310
311    /// Attempt to intersect the chain at its origin (genesis block)
312    ///
313    /// # Errors
314    ///
315    /// Returns an error if the intersection point cannot be found or if there
316    /// is a communication error.
317    pub async fn intersect_origin(&mut self) -> Result<Point, ClientError> {
318        debug!("intersecting origin");
319
320        let (point, _) = self.find_intersect(vec![Point::Origin]).await?;
321
322        point.ok_or(ClientError::IntersectionNotFound)
323    }
324
325    /// Attempts to intersect the chain at the latest known tip
326    ///
327    /// # Errors
328    ///
329    /// Returns an error if the intersection point cannot be found or if there
330    /// is a communication error.
331    pub async fn intersect_tip(&mut self) -> Result<Point, ClientError> {
332        let (_, Tip(point, _)) = self.find_intersect(vec![Point::Origin]).await?;
333
334        debug!(?point, "found tip value");
335
336        let (point, _) = self.find_intersect(vec![point]).await?;
337
338        point.ok_or(ClientError::IntersectionNotFound)
339    }
340
341    /// Send a `Done` message and terminate the protocol.
342    pub async fn send_done(&mut self) -> Result<(), ClientError> {
343        let msg = Message::Done;
344        self.send_message(&msg).await?;
345        self.0 = State::Done;
346
347        Ok(())
348    }
349}
350
351/// Node-to-node chain-sync client (streams `HeaderContent`).
352pub type N2NClient = Client<HeaderContent>;
353
354/// Node-to-client chain-sync client (streams whole `BlockContent`).
355pub type N2CClient = Client<BlockContent>;
pallas_network/miniprotocols/chainsync/client.rs

pallas_network/miniprotocols/chainsync/
client.rs
