kevy_replicate/replica_error.rs
1//! Error surface of [`crate::replica::ReplicaClient`] — split from
2//! `replica.rs` so each file stays under the 500-LOC house rule.
3//! Re-exported from [`crate::replica`], so caller paths are
4//! unchanged.
5
6use crate::wire::WireError;
7use std::io;
8
9/// Errors a replica client can surface to its driver loop.
10///
11/// ```
12/// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
13/// use kevy_replicate::replica::{ReplicaClient, ReplicaError};
14///
15/// # let stream = kevy_replicate::wire::encode_frame(3, &argv(&["SET", "k", "v"]));
16/// # let (addr, _primary) = fake_primary(b"+ACK 1 0\r\n", stream);
17/// let mut client = ReplicaClient::connect(addr, "replica-a", 0)?;
18/// match client.next_event() {
19/// Some(Err(ReplicaError::OffsetGap { .. })) => { /* resync through a snapshot */ }
20/// Some(Err(e)) => return Err(e.into()), // drop the link and reconnect
21/// other => panic!("unexpected {other:?}"),
22/// }
23/// # Ok::<(), Box<dyn std::error::Error>>(())
24/// ```
25#[derive(Debug)]
26#[non_exhaustive]
27pub enum ReplicaError {
28 /// Primary closed the connection or never replied during the
29 /// handshake / `+ACK` exchange.
30 ///
31 /// ```
32 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
33 /// use kevy_replicate::replica::{ReplicaClient, ReplicaError};
34 ///
35 /// # let (addr, _primary) = fake_primary(b"", Vec::new());
36 /// // the primary hangs up without an `+ACK`
37 /// let err = ReplicaClient::connect(addr, "replica-a", 0).unwrap_err();
38 /// assert!(matches!(err, ReplicaError::HandshakeRejected));
39 /// ```
40 HandshakeRejected,
41 /// `+ACK` line was malformed (didn't start with `+ACK `, didn't
42 /// parse the offset).
43 ///
44 /// ```
45 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
46 /// use kevy_replicate::replica::{ReplicaClient, ReplicaError};
47 ///
48 /// # let (addr, _primary) = fake_primary(b"-ERR not a primary\r\n", Vec::new());
49 /// let err = ReplicaClient::connect(addr, "replica-a", 0).unwrap_err();
50 /// assert!(matches!(err, ReplicaError::AckMalformed));
51 /// ```
52 AckMalformed,
53 /// Peer closed the connection mid-frame; reconnect to resume.
54 ///
55 /// ```
56 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
57 /// use kevy_replicate::replica::{ReplicaClient, ReplicaError};
58 ///
59 /// # let frame = kevy_replicate::wire::encode_frame(0, &argv(&["SET", "k", "v"]));
60 /// # let (addr, _primary) = fake_primary(b"+ACK 1 0\r\n", frame[..frame.len() - 3].to_vec());
61 /// let mut client = ReplicaClient::connect(addr, "replica-a", 0)?;
62 /// assert!(matches!(client.next_event(), Some(Err(ReplicaError::Truncated))));
63 /// # Ok::<(), ReplicaError>(())
64 /// ```
65 Truncated,
66 /// Wire-level decode error (envelope shape wrong, payload
67 /// malformed, etc.).
68 ///
69 /// ```
70 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
71 /// use kevy_replicate::replica::{ReplicaClient, ReplicaError};
72 /// use kevy_replicate::wire::WireError;
73 ///
74 /// # let (addr, _primary) = fake_primary(b"+ACK 1 0\r\n", b":1\r\n".to_vec());
75 /// let mut client = ReplicaClient::connect(addr, "replica-a", 0)?;
76 /// let event = client.next_event();
77 /// assert!(matches!(event, Some(Err(ReplicaError::Frame(WireError::BadEnvelope)))));
78 /// # Ok::<(), ReplicaError>(())
79 /// ```
80 Frame(WireError),
81 /// Frame arrived with an offset other than the expected next.
82 /// Caller should trigger a full snapshot resync.
83 ///
84 /// ```
85 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
86 /// use kevy_replicate::replica::{ReplicaClient, ReplicaError};
87 ///
88 /// # let stream = kevy_replicate::wire::encode_frame(12, &argv(&["SET", "k", "v"]));
89 /// # let (addr, _primary) = fake_primary(b"+ACK 1 10\r\n", stream);
90 /// let mut client = ReplicaClient::connect(addr, "replica-a", 10)?;
91 /// let Some(Err(ReplicaError::OffsetGap { expected, got })) = client.next_event() else { panic!() };
92 /// assert_eq!((expected, got), (10, 12)); // frames 10 and 11 never arrived
93 /// # Ok::<(), ReplicaError>(())
94 /// ```
95 OffsetGap {
96 /// The offset the client expected next (= `last_seen + 1`).
97 ///
98 /// ```
99 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
100 /// use kevy_replicate::replica::{ReplicaClient, ReplicaError};
101 ///
102 /// # let stream = kevy_replicate::wire::encode_frame(12, &argv(&["SET", "k", "v"]));
103 /// # let (addr, _primary) = fake_primary(b"+ACK 1 10\r\n", stream);
104 /// let mut client = ReplicaClient::connect(addr, "replica-a", 10)?;
105 /// let Some(Err(ReplicaError::OffsetGap { expected, .. })) = client.next_event() else {
106 /// panic!("expected a gap")
107 /// };
108 /// assert_eq!(expected, 10);
109 /// # Ok::<(), ReplicaError>(())
110 /// ```
111 expected: u64,
112 /// The offset the primary actually sent.
113 ///
114 /// ```
115 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
116 /// use kevy_replicate::replica::{ReplicaClient, ReplicaError};
117 ///
118 /// # let stream = kevy_replicate::wire::encode_frame(12, &argv(&["SET", "k", "v"]));
119 /// # let (addr, _primary) = fake_primary(b"+ACK 1 10\r\n", stream);
120 /// let mut client = ReplicaClient::connect(addr, "replica-a", 10)?;
121 /// let Some(Err(ReplicaError::OffsetGap { got, .. })) = client.next_event() else {
122 /// panic!("expected a gap")
123 /// };
124 /// assert_eq!(got, 12);
125 /// # Ok::<(), ReplicaError>(())
126 /// ```
127 got: u64,
128 },
129 /// While streaming a snapshot, the primary sent bytes that were
130 /// neither a snapshot chunk nor `+SNAPSHOT_END`. Interleaving live
131 /// frames inside a snapshot is forbidden (see `docs/snapshot.md`).
132 ///
133 /// ```
134 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
135 /// use kevy_replicate::replica::{ReplicaClient, ReplicaError};
136 /// use kevy_replicate::wire;
137 ///
138 /// # let mut stream = wire::encode_snapshot_begin();
139 /// # stream.extend(wire::encode_frame(0, &argv(&["SET", "k", "v"])));
140 /// # let (addr, _primary) = fake_primary(b"+ACK 1 0\r\n", stream);
141 /// let mut client = ReplicaClient::connect(addr, "replica-a", 0)?;
142 /// client.next_event(); // SnapshotBegin
143 /// assert!(matches!(client.next_event(), Some(Err(ReplicaError::UnexpectedInSnapshot))));
144 /// # Ok::<(), ReplicaError>(())
145 /// ```
146 UnexpectedInSnapshot,
147 /// `next_frame` was called but the next event is a snapshot
148 /// marker / chunk. Callers that want the snapshot-aware surface
149 /// must use [`crate::replica::ReplicaClient::next_event`].
150 ///
151 /// ```
152 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
153 /// use kevy_replicate::replica::{ReplicaClient, ReplicaError};
154 ///
155 /// # let (addr, _primary) = fake_primary(b"+ACK 1 0\r\n", kevy_replicate::wire::encode_snapshot_begin());
156 /// let mut client = ReplicaClient::connect(addr, "replica-a", 0)?;
157 /// assert!(matches!(client.next_frame(), Some(Err(ReplicaError::SnapshotInProgress))));
158 /// # Ok::<(), ReplicaError>(())
159 /// ```
160 SnapshotInProgress,
161 /// Underlying socket I/O failure.
162 ///
163 /// ```
164 /// use kevy_replicate::replica::{ReplicaClient, ReplicaError};
165 ///
166 /// // nothing listens on a port whose listener was just dropped
167 /// let addr = std::net::TcpListener::bind("127.0.0.1:0")?.local_addr()?;
168 /// let err = ReplicaClient::connect(addr, "replica-a", 0).unwrap_err();
169 /// let ReplicaError::Io(io) = err else { panic!("expected an I/O error, got {err}") };
170 /// assert_eq!(io.kind(), std::io::ErrorKind::ConnectionRefused);
171 /// # Ok::<(), std::io::Error>(())
172 /// ```
173 Io(io::Error),
174}
175
176impl std::fmt::Display for ReplicaError {
177 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
178 match self {
179 Self::HandshakeRejected => write!(f, "primary rejected replication handshake"),
180 Self::AckMalformed => write!(f, "primary sent malformed +ACK"),
181 Self::Truncated => write!(f, "replication stream truncated by peer"),
182 Self::Frame(e) => write!(f, "replication frame decode error: {e}"),
183 Self::OffsetGap { expected, got } => {
184 write!(f, "replication offset gap: expected {expected}, got {got}")
185 }
186 Self::UnexpectedInSnapshot => {
187 write!(f, "primary sent non-chunk bytes mid-snapshot")
188 }
189 Self::SnapshotInProgress => {
190 write!(f, "snapshot in progress; use next_event() to consume")
191 }
192 Self::Io(e) => write!(f, "replication socket I/O error: {e}"),
193 }
194 }
195}
196
197impl std::error::Error for ReplicaError {
198 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
199 match self {
200 Self::Frame(e) => Some(e),
201 Self::Io(e) => Some(e),
202 _ => None,
203 }
204 }
205}
206
207impl From<io::Error> for ReplicaError {
208 fn from(e: io::Error) -> Self {
209 ReplicaError::Io(e)
210 }
211}
212
213impl From<WireError> for ReplicaError {
214 fn from(e: WireError) -> Self {
215 match e {
216 WireError::Truncated => ReplicaError::Truncated,
217 other => ReplicaError::Frame(other),
218 }
219 }
220}
221
222#[cfg(test)]
223mod tests {
224 use super::*;
225
226 #[test]
227 fn from_io_error_wraps_into_io_variant() {
228 let e: ReplicaError = io::Error::new(io::ErrorKind::ConnectionRefused, "x").into();
229 assert!(matches!(e, ReplicaError::Io(_)));
230 }
231
232 #[test]
233 fn from_wire_error_truncated_maps_to_truncated() {
234 let e: ReplicaError = WireError::Truncated.into();
235 assert!(matches!(e, ReplicaError::Truncated));
236 }
237
238 #[test]
239 fn the_source_chain_reaches_the_frame_or_socket_error_and_stops_elsewhere() {
240 use std::error::Error;
241 let frame: ReplicaError = WireError::NegativeOffset(-3).into();
242 assert_eq!(
243 frame.to_string(),
244 "replication frame decode error: wire offset is negative: -3"
245 );
246 assert_eq!(
247 frame.source().map(ToString::to_string).as_deref(),
248 Some("wire offset is negative: -3")
249 );
250 let io: ReplicaError = io::Error::new(io::ErrorKind::ConnectionReset, "peer reset").into();
251 assert_eq!(io.to_string(), "replication socket I/O error: peer reset");
252 assert_eq!(io.source().map(ToString::to_string).as_deref(), Some("peer reset"));
253 for e in
254 [ReplicaError::HandshakeRejected, ReplicaError::Truncated, ReplicaError::AckMalformed]
255 {
256 assert!(e.source().is_none(), "{e} has no source");
257 }
258 }
259
260 #[test]
261 fn from_wire_error_other_maps_to_frame() {
262 let e: ReplicaError = WireError::BadEnvelope.into();
263 assert!(matches!(e, ReplicaError::Frame(_)));
264 }
265}