1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
//! Sync-specific error types.
//!
//! [`SyncError`] covers all failure modes in the sync protocol:
//! transport failures, handshake rejections, serialization issues,
//! and shutdown coordination.
use thiserror::Error;
use super::types::InstanceId;
/// Errors specific to the sync protocol.
///
/// These are wrapped into [`crate::PulseDBError::Sync`] when propagated
/// through the public API.
#[derive(Debug, Error)]
pub enum SyncError {
/// Transport-level failure (network I/O, connection refused, etc.).
#[error("Sync transport error: {0}")]
Transport(String),
/// Handshake was rejected by the remote peer.
#[error("Sync handshake failed: {0}")]
Handshake(String),
/// Failed to serialize or deserialize a sync message.
#[error("Sync serialization error: {0}")]
Serialization(String),
/// Operation timed out waiting for a response.
#[error("Sync operation timed out")]
Timeout,
/// Connection to the remote peer was lost.
#[error("Connection to sync peer lost")]
ConnectionLost,
/// Protocol version mismatch between peers.
#[error("Sync protocol version mismatch: local v{local}, remote v{remote}")]
ProtocolVersion {
/// Local protocol version.
local: u32,
/// Remote protocol version.
remote: u32,
},
/// Wire-format preamble mismatch — caught by raw-byte inspection of the
/// 3-byte preamble *before* any deserialize, so a serializer mismatch
/// (e.g. a bincode-era peer vs a postcard-era peer) fails loud with a
/// typed error instead of yielding garbage through the decoder.
///
/// This is distinct from [`SyncError::ProtocolVersion`]: that variant is
/// protocol-*semantics* (negotiated in-band after a successful decode);
/// this variant is wire-*format* (the bytes can't be trusted to decode at
/// all). Two failure shapes feed it:
///
/// - **bad magic** (`got == None`): the leading bytes are not a PulseDB
/// sync preamble at all (truncated body, a non-PulseDB POST, or a
/// pre-4.0 no-preamble peer's body) — reported with `got: None`.
/// - **wrong version** (`got == Some(v)`): a valid magic but a
/// `wire_format_version` byte this peer does not speak.
#[error("Sync wire-format mismatch: expected wire format v{expected}, got {}", match got { Some(g) => format!("v{g}"), None => "bad/absent magic".to_string() })]
WireFormatMismatch {
/// The wire-format version this peer speaks.
expected: u8,
/// The wire-format version observed in the preamble, or `None` when the
/// magic bytes were absent/wrong (so no trustworthy version was read).
got: Option<u8>,
},
/// Received an invalid or unrecognized payload.
#[error("Invalid sync payload: {0}")]
InvalidPayload(String),
/// No cursor found for the specified peer instance.
#[error("No sync cursor found for instance {instance}")]
CursorNotFound {
/// The peer instance whose cursor was not found.
instance: InstanceId,
},
/// The sync system is shutting down.
#[error("Sync system is shutting down")]
Shutdown,
}
impl SyncError {
/// Creates a transport error with the given message.
pub fn transport(msg: impl Into<String>) -> Self {
Self::Transport(msg.into())
}
/// Creates a handshake error with the given message.
pub fn handshake(msg: impl Into<String>) -> Self {
Self::Handshake(msg.into())
}
/// Creates a serialization error with the given message.
pub fn serialization(msg: impl Into<String>) -> Self {
Self::Serialization(msg.into())
}
/// Creates an invalid payload error with the given message.
pub fn invalid_payload(msg: impl Into<String>) -> Self {
Self::InvalidPayload(msg.into())
}
/// Creates a wire-format mismatch for a **bad / absent magic** preamble.
///
/// Used when the leading bytes are not a recognizable PulseDB sync
/// preamble (truncated body, wrong magic, or a pre-preamble peer's body).
pub fn wire_format_bad_magic(expected: u8) -> Self {
Self::WireFormatMismatch {
expected,
got: None,
}
}
/// Creates a wire-format mismatch for a **wrong wire-format version**.
///
/// Used when the magic is valid but the `wire_format_version` byte names a
/// version this peer does not speak.
pub fn wire_format_version(expected: u8, got: u8) -> Self {
Self::WireFormatMismatch {
expected,
got: Some(got),
}
}
/// Returns true if this is a transport error.
pub fn is_transport(&self) -> bool {
matches!(self, Self::Transport(_))
}
/// Returns true if this is a timeout error.
pub fn is_timeout(&self) -> bool {
matches!(self, Self::Timeout)
}
/// Returns true if this is a connection lost error.
pub fn is_connection_lost(&self) -> bool {
matches!(self, Self::ConnectionLost)
}
/// Returns true if this is a shutdown error.
pub fn is_shutdown(&self) -> bool {
matches!(self, Self::Shutdown)
}
/// Returns true if this is a wire-format mismatch (bad magic OR wrong
/// wire-format version) — the typed fail-loud signal for cross-version
/// sync, distinct from a generic [`SyncError::Serialization`].
pub fn is_wire_format_mismatch(&self) -> bool {
matches!(self, Self::WireFormatMismatch { .. })
}
}
impl From<postcard::Error> for SyncError {
fn from(err: postcard::Error) -> Self {
SyncError::Serialization(err.to_string())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_sync_error_display() {
let err = SyncError::transport("connection refused");
assert_eq!(err.to_string(), "Sync transport error: connection refused");
}
#[test]
fn test_protocol_version_display() {
let err = SyncError::ProtocolVersion {
local: 1,
remote: 2,
};
assert_eq!(
err.to_string(),
"Sync protocol version mismatch: local v1, remote v2"
);
}
#[test]
fn test_sync_error_is_checks() {
assert!(SyncError::transport("x").is_transport());
assert!(SyncError::Timeout.is_timeout());
assert!(SyncError::ConnectionLost.is_connection_lost());
assert!(SyncError::Shutdown.is_shutdown());
}
#[test]
fn test_postcard_error_conversion() {
// Deserializing truncated bytes triggers a postcard error.
let bad_bytes = vec![0u8; 0]; // too short for a (u64, u64)
let postcard_err = postcard::from_bytes::<(u64, u64)>(&bad_bytes).unwrap_err();
let sync_err: SyncError = postcard_err.into();
assert!(matches!(sync_err, SyncError::Serialization(_)));
}
#[test]
fn test_wire_format_mismatch_typed_and_distinct() {
let bad_magic = SyncError::wire_format_bad_magic(3);
let wrong_ver = SyncError::wire_format_version(3, 2);
// Both are the typed wire-format signal, NOT a generic Serialization.
assert!(bad_magic.is_wire_format_mismatch());
assert!(wrong_ver.is_wire_format_mismatch());
assert!(!SyncError::serialization("x").is_wire_format_mismatch());
// Distinct from protocol-semantics ProtocolVersion.
assert!(matches!(
bad_magic,
SyncError::WireFormatMismatch { got: None, .. }
));
assert!(matches!(
wrong_ver,
SyncError::WireFormatMismatch {
expected: 3,
got: Some(2)
}
));
}
}