radicle_protocol/service/
session.rs

1use std::collections::VecDeque;
2use std::sync::mpsc;
3use std::{fmt, time};
4
5use radicle::node::{FetchResult, Severity};
6use radicle::node::{Link, Timestamp};
7pub use radicle::node::{PingState, State};
8use radicle::storage::refs::RefsAt;
9
10use crate::service::message;
11use crate::service::message::Message;
12use crate::service::{Address, LocalDuration, LocalTime, NodeId, Outbox, RepoId, Rng};
13
14/// Time after which a connection is considered stable.
15pub const CONNECTION_STABLE_THRESHOLD: LocalDuration = LocalDuration::from_mins(1);
16/// Maximum items in the fetch queue.
17pub const MAX_FETCH_QUEUE_SIZE: usize = 128;
18
19#[derive(thiserror::Error, Debug, Clone, Copy)]
20pub enum Error {
21    /// The remote peer sent an invalid announcement timestamp,
22    /// for eg. a timestamp far in the future.
23    #[error("invalid announcement timestamp: {0}")]
24    InvalidTimestamp(InvalidTimestamp),
25    /// The remote peer sent git protocol messages while we were expecting
26    /// gossip messages. Or vice-versa.
27    #[error("protocol mismatch")]
28    ProtocolMismatch,
29    /// The remote peer did something that violates the protocol rules.
30    #[error("peer misbehaved")]
31    Misbehavior,
32    /// The remote peer timed out.
33    #[error("peer timed out")]
34    Timeout,
35}
36
37impl Error {
38    pub(crate) fn future_timestamp(theirs: Timestamp, ours: Timestamp) -> Self {
39        Self::InvalidTimestamp(InvalidTimestamp::Future { theirs, ours })
40    }
41}
42
43#[derive(thiserror::Error, Debug, Clone, Copy)]
44pub enum InvalidTimestamp {
45    #[error("{theirs} appears too far in the future compared to {ours}")]
46    Future { theirs: Timestamp, ours: Timestamp },
47}
48
49impl Error {
50    /// Return the severity for this error.
51    pub fn severity(&self) -> Severity {
52        match self {
53            Self::InvalidTimestamp(_) => Severity::High,
54            Self::ProtocolMismatch => Severity::High,
55            Self::Misbehavior => Severity::High,
56            Self::Timeout => Severity::Low,
57        }
58    }
59}
60
61/// Error when trying to queue a fetch.
62#[derive(thiserror::Error, Debug, Clone)]
63pub enum QueueError {
64    /// The item already exists in the queue.
65    #[error("item is already queued")]
66    Duplicate(QueuedFetch),
67    /// The queue is at capacity.
68    #[error("queue capacity reached")]
69    CapacityReached(QueuedFetch),
70}
71
72impl QueueError {
73    /// Get the inner [`QueuedFetch`].
74    pub fn inner(&self) -> &QueuedFetch {
75        match self {
76            Self::Duplicate(f) => f,
77            Self::CapacityReached(f) => f,
78        }
79    }
80}
81
82/// Fetch waiting to be processed, in the fetch queue.
83#[derive(Debug, Clone)]
84pub struct QueuedFetch {
85    /// Repo being fetched.
86    pub rid: RepoId,
87    /// Peer being fetched from.
88    pub from: NodeId,
89    /// Refs being fetched.
90    pub refs_at: Vec<RefsAt>,
91    /// The timeout given for the fetch request.
92    pub timeout: time::Duration,
93    /// Result channel.
94    pub channel: Option<mpsc::Sender<FetchResult>>,
95}
96
97impl PartialEq for QueuedFetch {
98    fn eq(&self, other: &Self) -> bool {
99        self.rid == other.rid
100            && self.from == other.from
101            && self.refs_at == other.refs_at
102            && self.channel.is_none()
103            && other.channel.is_none()
104    }
105}
106
107/// A peer session. Each connected peer will have one session.
108#[derive(Debug, Clone)]
109pub struct Session {
110    /// Peer id.
111    pub id: NodeId,
112    /// Peer address.
113    pub addr: Address,
114    /// Connection direction.
115    pub link: Link,
116    /// Whether we should attempt to re-connect
117    /// to this peer upon disconnection.
118    pub persistent: bool,
119    /// Peer connection state.
120    pub state: State,
121    /// Peer subscription.
122    pub subscribe: Option<message::Subscribe>,
123    /// Last time a message was received from the peer.
124    pub last_active: LocalTime,
125
126    /// Connection attempts. For persistent peers, Tracks
127    /// how many times we've attempted to connect. We reset this to zero
128    /// upon successful connection, once the connection is stable.
129    attempts: usize,
130    /// Source of entropy.
131    rng: Rng,
132}
133
134impl fmt::Display for Session {
135    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
136        let mut attrs = Vec::new();
137        let state = self.state.to_string();
138
139        if self.link.is_inbound() {
140            attrs.push("inbound");
141        } else {
142            attrs.push("outbound");
143        }
144        if self.persistent {
145            attrs.push("persistent");
146        }
147        attrs.push(state.as_str());
148
149        write!(f, "{} [{}]", self.id, attrs.join(" "))
150    }
151}
152
153impl From<&Session> for radicle::node::Session {
154    fn from(s: &Session) -> Self {
155        Self {
156            nid: s.id,
157            link: if s.link.is_inbound() {
158                radicle::node::Link::Inbound
159            } else {
160                radicle::node::Link::Outbound
161            },
162            addr: s.addr.clone(),
163            state: s.state.clone(),
164        }
165    }
166}
167
168impl Session {
169    pub fn outbound(id: NodeId, addr: Address, persistent: bool, rng: Rng) -> Self {
170        Self {
171            id,
172            addr,
173            state: State::Initial,
174            link: Link::Outbound,
175            subscribe: None,
176            persistent,
177            last_active: LocalTime::default(),
178            attempts: 1,
179            rng,
180        }
181    }
182
183    pub fn inbound(id: NodeId, addr: Address, persistent: bool, rng: Rng, time: LocalTime) -> Self {
184        Self {
185            id,
186            addr,
187            state: State::Connected {
188                since: time,
189                ping: PingState::default(),
190                latencies: VecDeque::default(),
191                stable: false,
192            },
193            link: Link::Inbound,
194            subscribe: None,
195            persistent,
196            last_active: time,
197            attempts: 0,
198            rng,
199        }
200    }
201
202    pub fn is_connecting(&self) -> bool {
203        matches!(self.state, State::Attempted)
204    }
205
206    pub fn is_stable(&self) -> bool {
207        matches!(self.state, State::Connected { stable: true, .. })
208    }
209
210    pub fn is_connected(&self) -> bool {
211        self.state.is_connected()
212    }
213
214    pub fn is_disconnected(&self) -> bool {
215        matches!(self.state, State::Disconnected { .. })
216    }
217
218    pub fn is_initial(&self) -> bool {
219        matches!(self.state, State::Initial)
220    }
221
222    pub fn attempts(&self) -> usize {
223        self.attempts
224    }
225
226    /// Run 'idle' task for session.
227    pub fn idle(&mut self, now: LocalTime) {
228        if let State::Connected {
229            since,
230            ref mut stable,
231            ..
232        } = self.state
233            && now >= since
234            && now.duration_since(since) >= CONNECTION_STABLE_THRESHOLD
235        {
236            *stable = true;
237            // Reset number of attempts for stable connections.
238            self.attempts = 0;
239        }
240    }
241
242    pub fn to_attempted(&mut self) {
243        assert!(
244            self.is_initial(),
245            "Can only transition to 'attempted' state from 'initial' state"
246        );
247        self.state = State::Attempted;
248        self.attempts += 1;
249    }
250
251    pub fn to_connected(&mut self, since: LocalTime) {
252        self.last_active = since;
253
254        if let State::Connected { .. } = &self.state {
255            log::debug!(target: "service", "Session {} is already in 'connected' state, resetting..", self.id);
256        };
257        self.state = State::Connected {
258            since,
259            ping: PingState::default(),
260            latencies: VecDeque::default(),
261            stable: false,
262        };
263    }
264
265    /// Move the session state to "disconnected". Returns any pending RID
266    /// that was requested.
267    pub fn to_disconnected(&mut self, since: LocalTime, retry_at: LocalTime) {
268        self.state = State::Disconnected { since, retry_at };
269    }
270
271    /// Return to initial state from disconnected state. This state transition
272    /// happens when we attempt to re-connect to a disconnected peer.
273    pub fn to_initial(&mut self) {
274        assert!(
275            self.is_disconnected(),
276            "Can only transition to 'initial' state from 'disconnected' state"
277        );
278        self.state = State::Initial;
279    }
280
281    pub fn ping(&mut self, since: LocalTime, reactor: &mut Outbox) -> Result<(), Error> {
282        if let State::Connected { ping, .. } = &mut self.state {
283            let msg = message::Ping::new(&mut self.rng);
284            *ping = PingState::AwaitingResponse {
285                len: msg.ponglen,
286                since,
287            };
288            reactor.write(self, Message::Ping(msg));
289        }
290        Ok(())
291    }
292}
radicle_protocol/service/session.rs

radicle_protocol/service/
session.rs
