1#![doc = include_str!("../README.md")]
2
3use sha2::{Digest, Sha256};
4
5pub type LogIndex = u64;
6pub type Epoch = u64;
7pub type ConfigId = u64;
8pub type NodeId = String;
9pub type ClusterId = String;
10
11#[non_exhaustive]
16#[derive(Clone, Copy, Debug, Eq, PartialEq)]
17pub enum ErrorCategory {
18 InvalidRequest,
19 Authentication,
20 Conflict,
21 Unavailable,
22 ResourceExhausted,
23 Internal,
24 Unknown,
25}
26
27#[derive(Clone, Debug, Eq, PartialEq)]
29pub struct ErrorClassification {
30 code: String,
31 category: ErrorCategory,
32 retryable: bool,
33}
34
35impl ErrorClassification {
36 pub fn new(code: impl Into<String>, category: ErrorCategory, retryable: bool) -> Self {
37 Self {
38 code: code.into(),
39 category,
40 retryable,
41 }
42 }
43
44 pub fn from_server_code(code: impl Into<String>, retryable: bool) -> Self {
45 let code = code.into();
46 let category = match code.as_str() {
47 "invalid_request" | "invalid_json" | "invalid_content_type" => {
48 ErrorCategory::InvalidRequest
49 }
50 "unauthorized" => ErrorCategory::Authentication,
51 "request_conflict" | "precondition_failed" => ErrorCategory::Conflict,
52 "unavailable"
53 | "durability_unavailable"
54 | "write_timeout"
55 | "writes_unavailable"
56 | "configuration_transition"
57 | "contention"
58 | "winner_limit_exceeded"
59 | "leader_unavailable"
60 | "snapshot_required" => ErrorCategory::Unavailable,
61 "resource_exhausted" | "overloaded" | "payload_too_large" => {
62 ErrorCategory::ResourceExhausted
63 }
64 "data_root_locked"
65 | "unsupported_ack_mode"
66 | "execution_profile_mismatch"
67 | "storage_error"
68 | "reconciliation_error"
69 | "invariant_violation"
70 | "fatal"
71 | "task_failed" => ErrorCategory::Internal,
72 _ => ErrorCategory::Unknown,
73 };
74 Self {
75 code,
76 category,
77 retryable,
78 }
79 }
80
81 pub fn code(&self) -> &str {
82 &self.code
83 }
84
85 pub const fn category(&self) -> ErrorCategory {
86 self.category
87 }
88
89 pub const fn retryable(&self) -> bool {
90 self.retryable
91 }
92}
93
94pub const RECOVERY_ANCHOR_FORMAT_VERSION: u32 = 2;
95
96#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, serde::Deserialize, serde::Serialize)]
97pub struct LogHash([u8; 32]);
98
99impl LogHash {
100 pub const ZERO: Self = Self([0; 32]);
101
102 pub const fn from_bytes(bytes: [u8; 32]) -> Self {
103 Self(bytes)
104 }
105
106 pub const fn as_bytes(&self) -> &[u8; 32] {
107 &self.0
108 }
109
110 pub fn digest(parts: &[&[u8]]) -> Self {
111 let mut hasher = Sha256::new();
112 for part in parts {
113 hasher.update(part);
114 }
115 Self(hasher.finalize().into())
116 }
117
118 pub fn to_hex(self) -> String {
119 let mut out = String::with_capacity(64);
120 for byte in self.0 {
121 out.push(hex_char(byte >> 4));
122 out.push(hex_char(byte & 0x0f));
123 }
124 out
125 }
126
127 pub fn from_hex(hex: &str) -> Option<Self> {
128 if hex.len() != 64 {
129 return None;
130 }
131
132 let mut bytes = [0; 32];
133 for (index, chunk) in hex.as_bytes().chunks_exact(2).enumerate() {
134 bytes[index] = (hex_value(chunk[0])? << 4) | hex_value(chunk[1])?;
135 }
136 Some(Self(bytes))
137 }
138}
139
140#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
141#[serde(deny_unknown_fields)]
142pub struct LogAnchor {
143 index: LogIndex,
144 hash: LogHash,
145}
146
147impl LogAnchor {
148 pub const fn new(index: LogIndex, hash: LogHash) -> Self {
149 Self { index, hash }
150 }
151
152 pub const fn index(&self) -> LogIndex {
153 self.index
154 }
155
156 pub const fn hash(&self) -> LogHash {
157 self.hash
158 }
159}
160
161#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
162#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
163pub enum StopBinding {
164 Unbound,
165 Bound {
166 successor: SuccessorDescriptor,
167 stop_command_hash: LogHash,
168 },
169}
170
171#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
172#[serde(tag = "phase", rename_all = "snake_case", deny_unknown_fields)]
173pub enum ConfigurationState {
174 Active {
175 config_id: ConfigId,
176 digest: LogHash,
177 },
178 Stopped {
179 config_id: ConfigId,
180 digest: LogHash,
181 stop: LogAnchor,
182 binding: StopBinding,
183 },
184}
185
186impl ConfigurationState {
187 pub const fn active(config_id: ConfigId, digest: LogHash) -> Self {
188 Self::Active { config_id, digest }
189 }
190
191 pub const fn stopped(
192 config_id: ConfigId,
193 digest: LogHash,
194 stop: LogAnchor,
195 binding: StopBinding,
196 ) -> Self {
197 Self::Stopped {
198 config_id,
199 digest,
200 stop,
201 binding,
202 }
203 }
204
205 pub const fn config_id(&self) -> ConfigId {
206 match self {
207 Self::Active { config_id, .. } | Self::Stopped { config_id, .. } => *config_id,
208 }
209 }
210
211 pub const fn digest(&self) -> LogHash {
212 match self {
213 Self::Active { digest, .. } | Self::Stopped { digest, .. } => *digest,
214 }
215 }
216
217 pub const fn stop(&self) -> Option<&LogAnchor> {
218 match self {
219 Self::Active { .. } => None,
220 Self::Stopped { stop, .. } => Some(stop),
221 }
222 }
223
224 pub const fn is_active(&self) -> bool {
225 matches!(self, Self::Active { .. })
226 }
227
228 pub fn validate_entry(&self, entry: &LogEntry) -> Result<Self, ConfigurationTransitionError> {
229 if entry.recompute_hash() != entry.hash {
230 return Err(ConfigurationTransitionError::EntryHashMismatch);
231 }
232 let change = if entry.entry_type == EntryType::ConfigChange {
233 Some(
234 ConfigChange::recognize_parts(entry.entry_type, &entry.payload)
235 .map_err(|_| ConfigurationTransitionError::InvalidConfigChange)?,
236 )
237 } else {
238 None
239 };
240
241 match (self, change) {
242 (
243 Self::Active { config_id, digest },
244 Some(ConfigChange::Stop {
245 config_id: stop_config_id,
246 config_digest,
247 }),
248 ) if entry.config_id == *config_id
249 && stop_config_id == *config_id
250 && (*digest == LogHash::ZERO || config_digest == *digest) =>
251 {
252 Ok(Self::stopped(
253 *config_id,
254 config_digest,
255 LogAnchor::new(entry.index, entry.hash),
256 StopBinding::Unbound,
257 ))
258 }
259 (Self::Active { config_id, digest }, Some(ConfigChange::BoundStop { successor }))
260 if entry.cluster_id == successor.cluster_id
261 && entry.config_id == *config_id
262 && successor.predecessor_config_id == *config_id
263 && successor.predecessor_config_digest == *digest =>
264 {
265 let stop_command_hash = (ConfigChange::BoundStop {
266 successor: successor.clone(),
267 })
268 .to_stored_command()
269 .hash();
270 Ok(Self::Stopped {
271 config_id: *config_id,
272 digest: *digest,
273 stop: LogAnchor::new(entry.index, entry.hash),
274 binding: StopBinding::Bound {
275 successor,
276 stop_command_hash,
277 },
278 })
279 }
280 (Self::Active { config_id, .. }, None) if entry.config_id == *config_id => {
281 Ok(self.clone())
282 }
283 (Self::Active { .. }, _) => Err(ConfigurationTransitionError::ConfigurationMismatch),
284 (
285 Self::Stopped {
286 config_id: predecessor_id,
287 stop,
288 binding: StopBinding::Unbound,
289 ..
290 },
291 Some(ConfigChange::ActivationBarrier {
292 config_id,
293 config_digest,
294 stop_slot,
295 prefix_hash,
296 }),
297 ) if predecessor_id.checked_add(1) == Some(config_id)
298 && entry.config_id == config_id
299 && stop.index().checked_add(1) == Some(entry.index)
300 && entry.prev_hash == stop.hash()
301 && stop_slot == stop.index()
302 && prefix_hash == stop.hash() =>
303 {
304 Ok(Self::active(config_id, config_digest))
305 }
306 (
307 Self::Stopped {
308 config_id: predecessor_id,
309 digest: predecessor_digest,
310 stop,
311 binding:
312 StopBinding::Bound {
313 successor: authorized_successor,
314 stop_command_hash: authorized_stop_command_hash,
315 },
316 },
317 Some(ConfigChange::BoundActivationBarrier {
318 successor,
319 stop_slot,
320 prefix_hash,
321 stop_command_hash,
322 }),
323 ) if successor.predecessor_config_id == *predecessor_id
324 && successor.predecessor_config_digest == *predecessor_digest
325 && &successor == authorized_successor
326 && entry.cluster_id == successor.cluster_id
327 && entry.config_id == successor.config_id
328 && stop.index().checked_add(1) == Some(entry.index)
329 && entry.prev_hash == stop.hash()
330 && stop_slot == stop.index()
331 && prefix_hash == stop.hash()
332 && *authorized_stop_command_hash
335 == (ConfigChange::BoundStop {
336 successor: authorized_successor.clone(),
337 })
338 .to_stored_command()
339 .hash()
340 && stop_command_hash == *authorized_stop_command_hash =>
341 {
342 Ok(Self::active(successor.config_id, successor.config_digest))
343 }
344 (Self::Stopped { .. }, _) => Err(ConfigurationTransitionError::InvalidActivation),
345 }
346 }
347}
348
349#[derive(Clone, Copy, Debug, Eq, PartialEq)]
350pub enum ConfigurationTransitionError {
351 EntryHashMismatch,
352 InvalidConfigChange,
353 ConfigurationMismatch,
354 InvalidActivation,
355}
356
357impl std::fmt::Display for ConfigurationTransitionError {
358 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
359 write!(f, "invalid configuration transition: {self:?}")
360 }
361}
362
363impl std::error::Error for ConfigurationTransitionError {}
364
365#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
366#[serde(deny_unknown_fields)]
367pub struct SnapshotIdentity {
368 snapshot_id: String,
369 digest: LogHash,
370 size_bytes: u64,
371 executor_fingerprint: LogHash,
372}
373
374impl SnapshotIdentity {
375 pub fn new(
376 snapshot_id: impl Into<String>,
377 digest: LogHash,
378 size_bytes: u64,
379 executor_fingerprint: LogHash,
380 ) -> Self {
381 Self {
382 snapshot_id: snapshot_id.into(),
383 digest,
384 size_bytes,
385 executor_fingerprint,
386 }
387 }
388
389 pub fn snapshot_id(&self) -> &str {
390 &self.snapshot_id
391 }
392
393 pub const fn digest(&self) -> LogHash {
394 self.digest
395 }
396
397 pub const fn size_bytes(&self) -> u64 {
398 self.size_bytes
399 }
400
401 pub const fn executor_fingerprint(&self) -> LogHash {
402 self.executor_fingerprint
403 }
404}
405
406#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
407#[serde(deny_unknown_fields)]
408pub struct RecoveryAnchor {
409 format_version: u32,
410 cluster_id: ClusterId,
411 epoch: Epoch,
412 config_id: ConfigId,
413 configuration_state: ConfigurationState,
414 recovery_generation: u64,
415 compacted: LogAnchor,
416 snapshot: SnapshotIdentity,
417}
418
419impl RecoveryAnchor {
420 pub fn new(
421 cluster_id: impl Into<ClusterId>,
422 epoch: Epoch,
423 configuration_state: ConfigurationState,
424 recovery_generation: u64,
425 compacted: LogAnchor,
426 snapshot: SnapshotIdentity,
427 ) -> Self {
428 Self {
429 format_version: RECOVERY_ANCHOR_FORMAT_VERSION,
430 cluster_id: cluster_id.into(),
431 epoch,
432 config_id: configuration_state.config_id(),
433 configuration_state,
434 recovery_generation,
435 compacted,
436 snapshot,
437 }
438 }
439
440 pub const fn format_version(&self) -> u32 {
441 self.format_version
442 }
443
444 pub fn cluster_id(&self) -> &str {
445 &self.cluster_id
446 }
447
448 pub const fn epoch(&self) -> Epoch {
449 self.epoch
450 }
451
452 pub const fn config_id(&self) -> ConfigId {
453 self.config_id
454 }
455
456 pub const fn configuration_state(&self) -> &ConfigurationState {
457 &self.configuration_state
458 }
459
460 pub const fn recovery_generation(&self) -> u64 {
461 self.recovery_generation
462 }
463
464 pub const fn compacted(&self) -> &LogAnchor {
465 &self.compacted
466 }
467
468 pub const fn snapshot(&self) -> &SnapshotIdentity {
469 &self.snapshot
470 }
471
472 pub const fn executor_fingerprint(&self) -> LogHash {
473 self.snapshot.executor_fingerprint()
474 }
475}
476
477impl<'de> serde::Deserialize<'de> for RecoveryAnchor {
478 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
479 where
480 D: serde::Deserializer<'de>,
481 {
482 #[derive(serde::Deserialize)]
483 #[serde(deny_unknown_fields)]
484 struct Wire {
485 format_version: u32,
486 cluster_id: ClusterId,
487 epoch: Epoch,
488 config_id: ConfigId,
489 configuration_state: ConfigurationState,
490 recovery_generation: u64,
491 compacted: LogAnchor,
492 snapshot: SnapshotIdentity,
493 }
494
495 let wire = Wire::deserialize(deserializer)?;
496 if wire.format_version != RECOVERY_ANCHOR_FORMAT_VERSION
497 || wire.configuration_state.config_id() != wire.config_id
498 {
499 return Err(serde::de::Error::custom(
500 "invalid recovery anchor configuration state",
501 ));
502 }
503 Ok(Self {
504 format_version: wire.format_version,
505 cluster_id: wire.cluster_id,
506 epoch: wire.epoch,
507 config_id: wire.config_id,
508 configuration_state: wire.configuration_state,
509 recovery_generation: wire.recovery_generation,
510 compacted: wire.compacted,
511 snapshot: wire.snapshot,
512 })
513 }
514}
515
516fn hex_char(value: u8) -> char {
517 match value {
518 0..=9 => (b'0' + value) as char,
519 10..=15 => (b'a' + value - 10) as char,
520 _ => unreachable!("nibble out of range"),
521 }
522}
523
524fn hex_value(value: u8) -> Option<u8> {
525 match value {
526 b'0'..=b'9' => Some(value - b'0'),
527 b'a'..=b'f' => Some(value - b'a' + 10),
528 b'A'..=b'F' => Some(value - b'A' + 10),
529 _ => None,
530 }
531}
532
533#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
534pub struct EntryId {
535 pub epoch: Epoch,
536 pub index: LogIndex,
537}
538
539#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
540pub enum EntryType {
541 Command,
542 ConfigChange,
543 SnapshotBarrier,
544 SnapshotPublished,
545 Noop,
546}
547
548impl EntryType {
549 pub const fn as_u8(self) -> u8 {
550 match self {
551 Self::Command => 1,
552 Self::ConfigChange => 2,
553 Self::SnapshotBarrier => 3,
554 Self::SnapshotPublished => 4,
555 Self::Noop => 5,
556 }
557 }
558
559 pub const fn from_u8(value: u8) -> Option<Self> {
560 match value {
561 1 => Some(Self::Command),
562 2 => Some(Self::ConfigChange),
563 3 => Some(Self::SnapshotBarrier),
564 4 => Some(Self::SnapshotPublished),
565 5 => Some(Self::Noop),
566 _ => None,
567 }
568 }
569}
570
571#[derive(
572 Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, serde::Deserialize, serde::Serialize,
573)]
574#[serde(rename_all = "snake_case")]
575pub enum ExecutionProfile {
576 #[serde(rename = "sql")]
577 Sqlite,
578 Graph,
579 Kv,
580}
581
582impl ExecutionProfile {
583 pub const fn as_str(self) -> &'static str {
584 match self {
585 Self::Sqlite => "sql",
586 Self::Graph => "graph",
587 Self::Kv => "kv",
588 }
589 }
590
591 pub const fn wire_id(self) -> u8 {
592 match self {
593 Self::Sqlite => 1,
594 Self::Graph => 2,
595 Self::Kv => 3,
596 }
597 }
598
599 pub const fn from_wire_id(value: u8) -> Option<Self> {
600 match value {
601 1 => Some(Self::Sqlite),
602 2 => Some(Self::Graph),
603 3 => Some(Self::Kv),
604 _ => None,
605 }
606 }
607}
608
609impl std::fmt::Display for ExecutionProfile {
610 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
611 formatter.write_str(self.as_str())
612 }
613}
614
615impl std::str::FromStr for ExecutionProfile {
616 type Err = ExecutionProfileParseError;
617
618 fn from_str(value: &str) -> Result<Self, Self::Err> {
619 match value {
620 "sql" => Ok(Self::Sqlite),
621 "graph" => Ok(Self::Graph),
622 "kv" => Ok(Self::Kv),
623 _ => Err(ExecutionProfileParseError),
624 }
625 }
626}
627
628#[derive(Clone, Copy, Debug, Eq, PartialEq)]
629pub struct ExecutionProfileParseError;
630
631impl std::fmt::Display for ExecutionProfileParseError {
632 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
633 formatter.write_str("expected execution profile `sql`, `graph`, or `kv`")
634 }
635}
636
637impl std::error::Error for ExecutionProfileParseError {}
638
639const REPLICATED_COMMAND_MAGIC: &[u8; 4] = b"QCMD";
640pub const REPLICATED_COMMAND_FORMAT_VERSION: u16 = 1;
641const REPLICATED_COMMAND_HEADER_BYTES: usize = 15;
642
643#[derive(Clone, Debug, Eq, PartialEq)]
644pub struct ReplicatedCommandEnvelope {
645 profile: ExecutionProfile,
646 command_version: u16,
647 request_id: String,
648 body: Vec<u8>,
649}
650
651impl ReplicatedCommandEnvelope {
652 pub fn new(
653 profile: ExecutionProfile,
654 command_version: u16,
655 request_id: impl Into<String>,
656 body: Vec<u8>,
657 ) -> Result<Self, CommandEnvelopeError> {
658 let request_id = request_id.into();
659 validate_command_envelope_fields(command_version, &request_id, &body)?;
660 Ok(Self {
661 profile,
662 command_version,
663 request_id,
664 body,
665 })
666 }
667
668 pub const fn profile(&self) -> ExecutionProfile {
669 self.profile
670 }
671
672 pub const fn command_version(&self) -> u16 {
673 self.command_version
674 }
675
676 pub fn request_id(&self) -> &str {
677 &self.request_id
678 }
679
680 pub fn body(&self) -> &[u8] {
681 &self.body
682 }
683
684 pub fn encode(&self) -> Result<Vec<u8>, CommandEnvelopeError> {
685 validate_command_envelope_fields(self.command_version, &self.request_id, &self.body)?;
686 let request_id_len = u16::try_from(self.request_id.len())
687 .map_err(|_| CommandEnvelopeError::RequestIdTooLong)?;
688 let body_len =
689 u32::try_from(self.body.len()).map_err(|_| CommandEnvelopeError::BodyTooLong)?;
690 let capacity = REPLICATED_COMMAND_HEADER_BYTES
691 .checked_add(self.request_id.len())
692 .and_then(|size| size.checked_add(self.body.len()))
693 .ok_or(CommandEnvelopeError::LengthOverflow)?;
694 let mut encoded = Vec::with_capacity(capacity);
695 encoded.extend_from_slice(REPLICATED_COMMAND_MAGIC);
696 encoded.extend_from_slice(&REPLICATED_COMMAND_FORMAT_VERSION.to_be_bytes());
697 encoded.push(self.profile.wire_id());
698 encoded.extend_from_slice(&self.command_version.to_be_bytes());
699 encoded.extend_from_slice(&request_id_len.to_be_bytes());
700 encoded.extend_from_slice(&body_len.to_be_bytes());
701 encoded.extend_from_slice(self.request_id.as_bytes());
702 encoded.extend_from_slice(&self.body);
703 Ok(encoded)
704 }
705
706 pub fn decode(encoded: &[u8]) -> Result<Self, CommandEnvelopeError> {
707 if encoded.len() < REPLICATED_COMMAND_HEADER_BYTES {
708 return Err(CommandEnvelopeError::Truncated);
709 }
710 if encoded.get(..4) != Some(REPLICATED_COMMAND_MAGIC) {
711 return Err(CommandEnvelopeError::InvalidMagic);
712 }
713 let format_version = u16::from_be_bytes([encoded[4], encoded[5]]);
714 if format_version != REPLICATED_COMMAND_FORMAT_VERSION {
715 return Err(CommandEnvelopeError::UnsupportedFormatVersion(
716 format_version,
717 ));
718 }
719 let profile = ExecutionProfile::from_wire_id(encoded[6])
720 .ok_or(CommandEnvelopeError::UnknownExecutionProfile(encoded[6]))?;
721 let command_version = u16::from_be_bytes([encoded[7], encoded[8]]);
722 let request_id_len = usize::from(u16::from_be_bytes([encoded[9], encoded[10]]));
723 let body_len = usize::try_from(u32::from_be_bytes([
724 encoded[11],
725 encoded[12],
726 encoded[13],
727 encoded[14],
728 ]))
729 .map_err(|_| CommandEnvelopeError::LengthOverflow)?;
730 let request_id_end = REPLICATED_COMMAND_HEADER_BYTES
731 .checked_add(request_id_len)
732 .ok_or(CommandEnvelopeError::LengthOverflow)?;
733 let body_end = request_id_end
734 .checked_add(body_len)
735 .ok_or(CommandEnvelopeError::LengthOverflow)?;
736 if encoded.len() < body_end {
737 return Err(CommandEnvelopeError::Truncated);
738 }
739 if encoded.len() != body_end {
740 return Err(CommandEnvelopeError::TrailingBytes);
741 }
742 let request_id = std::str::from_utf8(
743 encoded
744 .get(REPLICATED_COMMAND_HEADER_BYTES..request_id_end)
745 .ok_or(CommandEnvelopeError::Truncated)?,
746 )
747 .map_err(|_| CommandEnvelopeError::InvalidRequestIdUtf8)?
748 .to_owned();
749 let body = encoded
750 .get(request_id_end..body_end)
751 .ok_or(CommandEnvelopeError::Truncated)?
752 .to_vec();
753 Self::new(profile, command_version, request_id, body)
754 }
755}
756
757fn validate_command_envelope_fields(
758 command_version: u16,
759 request_id: &str,
760 body: &[u8],
761) -> Result<(), CommandEnvelopeError> {
762 if command_version == 0 {
763 return Err(CommandEnvelopeError::InvalidCommandVersion);
764 }
765 if request_id.is_empty() {
766 return Err(CommandEnvelopeError::EmptyRequestId);
767 }
768 u16::try_from(request_id.len()).map_err(|_| CommandEnvelopeError::RequestIdTooLong)?;
769 u32::try_from(body.len()).map_err(|_| CommandEnvelopeError::BodyTooLong)?;
770 Ok(())
771}
772
773#[derive(Clone, Copy, Debug, Eq, PartialEq)]
774pub enum CommandEnvelopeError {
775 InvalidMagic,
776 UnsupportedFormatVersion(u16),
777 UnknownExecutionProfile(u8),
778 InvalidCommandVersion,
779 EmptyRequestId,
780 RequestIdTooLong,
781 BodyTooLong,
782 InvalidRequestIdUtf8,
783 Truncated,
784 LengthOverflow,
785 TrailingBytes,
786}
787
788impl std::fmt::Display for CommandEnvelopeError {
789 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
790 match self {
791 Self::InvalidMagic => formatter.write_str("replicated command magic is invalid"),
792 Self::UnsupportedFormatVersion(version) => {
793 write!(
794 formatter,
795 "unsupported replicated command format version {version}"
796 )
797 }
798 Self::UnknownExecutionProfile(profile) => {
799 write!(formatter, "unknown execution profile wire id {profile}")
800 }
801 Self::InvalidCommandVersion => {
802 formatter.write_str("replicated command version must be positive")
803 }
804 Self::EmptyRequestId => formatter.write_str("replicated command request id is empty"),
805 Self::RequestIdTooLong => {
806 formatter.write_str("replicated command request id exceeds u16 length")
807 }
808 Self::BodyTooLong => formatter.write_str("replicated command body exceeds u32 length"),
809 Self::InvalidRequestIdUtf8 => {
810 formatter.write_str("replicated command request id is not UTF-8")
811 }
812 Self::Truncated => formatter.write_str("replicated command is truncated"),
813 Self::LengthOverflow => formatter.write_str("replicated command length overflow"),
814 Self::TrailingBytes => formatter.write_str("replicated command has trailing bytes"),
815 }
816 }
817}
818
819impl std::error::Error for CommandEnvelopeError {}
820
821#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
822pub enum CommandKind {
823 Deterministic,
824 ReadBarrier,
825}
826
827#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
828pub struct Command {
829 kind: CommandKind,
830 payload: Vec<u8>,
831}
832
833impl Command {
834 pub fn new(kind: CommandKind, payload: Vec<u8>) -> Self {
835 Self { kind, payload }
836 }
837
838 pub const fn kind(&self) -> CommandKind {
839 self.kind
840 }
841
842 pub fn payload(&self) -> &[u8] {
843 &self.payload
844 }
845}
846
847#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
848pub struct StoredCommand {
849 pub entry_type: EntryType,
850 pub payload: Vec<u8>,
851}
852
853impl StoredCommand {
854 pub fn new(entry_type: EntryType, payload: Vec<u8>) -> Self {
855 Self {
856 entry_type,
857 payload,
858 }
859 }
860
861 pub fn hash(&self) -> LogHash {
862 let entry_type = [self.entry_type.as_u8()];
863 LogHash::digest(&[b"rhiza-command-v2", &entry_type, &self.payload])
864 }
865}
866
867const CONFIG_CHANGE_MAGIC: &[u8; 4] = b"QCFG";
868
869#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
870pub struct SuccessorDescriptor {
871 cluster_id: ClusterId,
872 predecessor_config_id: ConfigId,
873 predecessor_config_digest: LogHash,
874 config_id: ConfigId,
875 config_digest: LogHash,
876 members: Vec<NodeId>,
877}
878
879impl<'de> serde::Deserialize<'de> for SuccessorDescriptor {
880 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
881 where
882 D: serde::Deserializer<'de>,
883 {
884 #[derive(serde::Deserialize)]
885 #[serde(deny_unknown_fields)]
886 struct Wire {
887 cluster_id: ClusterId,
888 predecessor_config_id: ConfigId,
889 predecessor_config_digest: LogHash,
890 config_id: ConfigId,
891 config_digest: LogHash,
892 members: Vec<NodeId>,
893 }
894
895 let wire = Wire::deserialize(deserializer)?;
896 if !wire.members.windows(2).all(|pair| pair[0] < pair[1]) {
897 return Err(serde::de::Error::custom(
898 "successor members are not canonical",
899 ));
900 }
901 let encoded_digest = wire.config_digest;
902 let descriptor = Self::new(
903 wire.cluster_id,
904 wire.predecessor_config_id,
905 wire.predecessor_config_digest,
906 wire.config_id,
907 wire.members,
908 )
909 .map_err(serde::de::Error::custom)?;
910 if descriptor.config_digest != encoded_digest {
911 return Err(serde::de::Error::custom(
912 "successor membership digest mismatch",
913 ));
914 }
915 Ok(descriptor)
916 }
917}
918
919impl SuccessorDescriptor {
920 pub fn new(
921 cluster_id: impl Into<ClusterId>,
922 predecessor_config_id: ConfigId,
923 predecessor_config_digest: LogHash,
924 config_id: ConfigId,
925 members: Vec<NodeId>,
926 ) -> Result<Self, ConfigChangeDecodeError> {
927 let cluster_id = cluster_id.into();
928 if cluster_id.is_empty() || cluster_id.len() > usize::from(u16::MAX) {
929 return Err(ConfigChangeDecodeError);
930 }
931 if predecessor_config_id.checked_add(1) != Some(config_id) {
932 return Err(ConfigChangeDecodeError);
933 }
934 let mut canonical = members;
935 let original_len = canonical.len();
936 canonical.sort();
937 canonical.dedup();
938 if canonical.len() != original_len
939 || !(3..=7).contains(&canonical.len())
940 || canonical
941 .iter()
942 .any(|member| member.is_empty() || member.len() > usize::from(u16::MAX))
943 {
944 return Err(ConfigChangeDecodeError);
945 }
946 let config_digest = canonical_membership_digest(&canonical)?;
947 Ok(Self {
948 cluster_id,
949 predecessor_config_id,
950 predecessor_config_digest,
951 config_id,
952 config_digest,
953 members: canonical,
954 })
955 }
956
957 pub fn cluster_id(&self) -> &str {
958 &self.cluster_id
959 }
960
961 pub const fn predecessor_config_id(&self) -> ConfigId {
962 self.predecessor_config_id
963 }
964
965 pub const fn predecessor_config_digest(&self) -> LogHash {
966 self.predecessor_config_digest
967 }
968
969 pub const fn config_id(&self) -> ConfigId {
970 self.config_id
971 }
972
973 pub const fn digest(&self) -> LogHash {
974 self.config_digest
975 }
976
977 pub fn members(&self) -> &[NodeId] {
978 &self.members
979 }
980}
981
982pub fn canonical_membership_digest(members: &[NodeId]) -> Result<LogHash, ConfigChangeDecodeError> {
983 if !(3..=7).contains(&members.len())
984 || members
985 .iter()
986 .any(|member| member.is_empty() || member.len() > usize::from(u16::MAX))
987 || !members.windows(2).all(|pair| pair[0] < pair[1])
988 {
989 return Err(ConfigChangeDecodeError);
990 }
991 let encoded_len = 14 + members.len() * 8 + members.iter().map(String::len).sum::<usize>();
992 let mut encoded = Vec::with_capacity(encoded_len);
993 encoded.extend_from_slice(b"QMEM\0\x01");
994 encoded.extend_from_slice(&(members.len() as u64).to_be_bytes());
995 for member in members {
996 encoded.extend_from_slice(&(member.len() as u64).to_be_bytes());
997 encoded.extend_from_slice(member.as_bytes());
998 }
999 Ok(LogHash::digest(&[&encoded]))
1000}
1001
1002#[derive(Clone, Debug, Eq, PartialEq)]
1003pub enum ConfigChange {
1004 Stop {
1005 config_id: ConfigId,
1006 config_digest: LogHash,
1007 },
1008 ActivationBarrier {
1009 config_id: ConfigId,
1010 config_digest: LogHash,
1011 stop_slot: LogIndex,
1012 prefix_hash: LogHash,
1013 },
1014 BoundStop {
1015 successor: SuccessorDescriptor,
1016 },
1017 BoundActivationBarrier {
1018 successor: SuccessorDescriptor,
1019 stop_slot: LogIndex,
1020 prefix_hash: LogHash,
1021 stop_command_hash: LogHash,
1022 },
1023}
1024
1025impl ConfigChange {
1026 pub const fn stop(config_id: ConfigId, config_digest: LogHash) -> Self {
1027 Self::Stop {
1028 config_id,
1029 config_digest,
1030 }
1031 }
1032
1033 pub const fn activation_barrier(
1034 config_id: ConfigId,
1035 config_digest: LogHash,
1036 stop_slot: LogIndex,
1037 prefix_hash: LogHash,
1038 ) -> Self {
1039 Self::ActivationBarrier {
1040 config_id,
1041 config_digest,
1042 stop_slot,
1043 prefix_hash,
1044 }
1045 }
1046
1047 pub fn bound_stop(
1048 cluster_id: impl Into<ClusterId>,
1049 predecessor_config_id: ConfigId,
1050 predecessor_config_digest: LogHash,
1051 successor_config_id: ConfigId,
1052 successor_members: Vec<NodeId>,
1053 ) -> Result<Self, ConfigChangeDecodeError> {
1054 Ok(Self::BoundStop {
1055 successor: SuccessorDescriptor::new(
1056 cluster_id,
1057 predecessor_config_id,
1058 predecessor_config_digest,
1059 successor_config_id,
1060 successor_members,
1061 )?,
1062 })
1063 }
1064
1065 pub const fn bound_activation_barrier(
1066 successor: SuccessorDescriptor,
1067 stop_slot: LogIndex,
1068 prefix_hash: LogHash,
1069 stop_command_hash: LogHash,
1070 ) -> Self {
1071 Self::BoundActivationBarrier {
1072 successor,
1073 stop_slot,
1074 prefix_hash,
1075 stop_command_hash,
1076 }
1077 }
1078
1079 pub const fn successor(&self) -> Option<&SuccessorDescriptor> {
1080 match self {
1081 Self::BoundStop { successor } | Self::BoundActivationBarrier { successor, .. } => {
1082 Some(successor)
1083 }
1084 _ => None,
1085 }
1086 }
1087
1088 pub fn to_stored_command(&self) -> StoredCommand {
1089 let mut payload = Vec::with_capacity(87);
1090 payload.extend_from_slice(CONFIG_CHANGE_MAGIC);
1091 match self {
1092 Self::Stop {
1093 config_id,
1094 config_digest,
1095 } => {
1096 payload.push(1);
1097 payload.extend_from_slice(&config_id.to_be_bytes());
1098 payload.extend_from_slice(config_digest.as_bytes());
1099 }
1100 Self::ActivationBarrier {
1101 config_id,
1102 config_digest,
1103 stop_slot,
1104 prefix_hash,
1105 } => {
1106 payload.push(2);
1107 payload.extend_from_slice(&config_id.to_be_bytes());
1108 payload.extend_from_slice(config_digest.as_bytes());
1109 payload.extend_from_slice(&stop_slot.to_be_bytes());
1110 payload.extend_from_slice(prefix_hash.as_bytes());
1111 }
1112 Self::BoundStop { successor } => {
1113 payload.push(3);
1114 encode_successor(&mut payload, successor);
1115 }
1116 Self::BoundActivationBarrier {
1117 successor,
1118 stop_slot,
1119 prefix_hash,
1120 stop_command_hash,
1121 } => {
1122 payload.push(4);
1123 encode_successor(&mut payload, successor);
1124 payload.extend_from_slice(&stop_slot.to_be_bytes());
1125 payload.extend_from_slice(prefix_hash.as_bytes());
1126 payload.extend_from_slice(stop_command_hash.as_bytes());
1127 }
1128 }
1129 StoredCommand::new(EntryType::ConfigChange, payload)
1130 }
1131
1132 pub fn recognize(command: &StoredCommand) -> Result<Self, ConfigChangeDecodeError> {
1133 Self::recognize_parts(command.entry_type, &command.payload)
1134 }
1135
1136 pub fn recognize_parts(
1137 entry_type: EntryType,
1138 payload: &[u8],
1139 ) -> Result<Self, ConfigChangeDecodeError> {
1140 if entry_type != EntryType::ConfigChange {
1141 return Err(ConfigChangeDecodeError);
1142 }
1143 let bytes = payload;
1144 if bytes.get(..4) != Some(CONFIG_CHANGE_MAGIC) {
1145 return Err(ConfigChangeDecodeError);
1146 }
1147 let kind = *bytes.get(4).ok_or(ConfigChangeDecodeError)?;
1148 if matches!(kind, 3 | 4) {
1149 let mut cursor = 5;
1150 let successor = decode_successor(bytes, &mut cursor)?;
1151 let change = match kind {
1152 3 => Self::BoundStop { successor },
1153 4 => Self::BoundActivationBarrier {
1154 successor,
1155 stop_slot: read_config_u64_at(bytes, &mut cursor)?,
1156 prefix_hash: read_config_hash_at(bytes, &mut cursor)?,
1157 stop_command_hash: read_config_hash_at(bytes, &mut cursor)?,
1158 },
1159 _ => return Err(ConfigChangeDecodeError),
1160 };
1161 if cursor != bytes.len() {
1162 return Err(ConfigChangeDecodeError);
1163 }
1164 return Ok(change);
1165 }
1166 let config_id = read_config_u64(bytes, 5)?;
1167 let config_digest = read_config_hash(bytes, 13)?;
1168 match kind {
1169 1 if bytes.len() == 45 => Ok(Self::stop(config_id, config_digest)),
1170 2 if bytes.len() == 85 => Ok(Self::activation_barrier(
1171 config_id,
1172 config_digest,
1173 read_config_u64(bytes, 45)?,
1174 read_config_hash(bytes, 53)?,
1175 )),
1176 _ => Err(ConfigChangeDecodeError),
1177 }
1178 }
1179
1180 pub const fn binding(&self) -> (ConfigId, LogHash) {
1181 match self {
1182 Self::Stop {
1183 config_id,
1184 config_digest,
1185 }
1186 | Self::ActivationBarrier {
1187 config_id,
1188 config_digest,
1189 ..
1190 } => (*config_id, *config_digest),
1191 Self::BoundStop { successor } => (
1192 successor.predecessor_config_id,
1193 successor.predecessor_config_digest,
1194 ),
1195 Self::BoundActivationBarrier { successor, .. } => {
1196 (successor.config_id, successor.config_digest)
1197 }
1198 }
1199 }
1200}
1201
1202fn encode_successor(out: &mut Vec<u8>, successor: &SuccessorDescriptor) {
1203 let cluster = successor.cluster_id.as_bytes();
1204 let encoded_len = 83
1205 + cluster.len()
1206 + successor
1207 .members
1208 .iter()
1209 .map(|member| 2 + member.len())
1210 .sum::<usize>();
1211 out.reserve(encoded_len);
1212 let cluster_length =
1213 u16::try_from(cluster.len()).expect("validated successor cluster length fits u16");
1214 out.extend_from_slice(&cluster_length.to_be_bytes());
1215 out.extend_from_slice(cluster);
1216 out.extend_from_slice(&successor.predecessor_config_id.to_be_bytes());
1217 out.extend_from_slice(successor.predecessor_config_digest.as_bytes());
1218 out.extend_from_slice(&successor.config_id.to_be_bytes());
1219 out.extend_from_slice(successor.config_digest.as_bytes());
1220 let member_count =
1221 u8::try_from(successor.members.len()).expect("validated successor member count fits u8");
1222 out.push(member_count);
1223 for member in &successor.members {
1224 let member_length =
1225 u16::try_from(member.len()).expect("validated successor member length fits u16");
1226 out.extend_from_slice(&member_length.to_be_bytes());
1227 out.extend_from_slice(member.as_bytes());
1228 }
1229}
1230
1231fn decode_successor(
1232 bytes: &[u8],
1233 cursor: &mut usize,
1234) -> Result<SuccessorDescriptor, ConfigChangeDecodeError> {
1235 let cluster_id = read_config_string(bytes, cursor)?;
1236 let predecessor_config_id = read_config_u64_at(bytes, cursor)?;
1237 let predecessor_config_digest = read_config_hash_at(bytes, cursor)?;
1238 let config_id = read_config_u64_at(bytes, cursor)?;
1239 let encoded_digest = read_config_hash_at(bytes, cursor)?;
1240 let count = *bytes.get(*cursor).ok_or(ConfigChangeDecodeError)? as usize;
1241 if !(3..=7).contains(&count) {
1242 return Err(ConfigChangeDecodeError);
1243 }
1244 *cursor += 1;
1245 let members = (0..count)
1246 .map(|_| read_config_string(bytes, cursor))
1247 .collect::<Result<Vec<_>, _>>()?;
1248 if !members.windows(2).all(|pair| pair[0] < pair[1]) {
1249 return Err(ConfigChangeDecodeError);
1250 }
1251 let descriptor = SuccessorDescriptor::new(
1252 cluster_id,
1253 predecessor_config_id,
1254 predecessor_config_digest,
1255 config_id,
1256 members,
1257 )?;
1258 if descriptor.config_digest != encoded_digest {
1259 return Err(ConfigChangeDecodeError);
1260 }
1261 Ok(descriptor)
1262}
1263
1264fn read_config_string(bytes: &[u8], cursor: &mut usize) -> Result<String, ConfigChangeDecodeError> {
1265 let length = read_config_u16(bytes, *cursor)? as usize;
1266 let value_start = cursor.checked_add(2).ok_or(ConfigChangeDecodeError)?;
1267 let value_end = value_start
1268 .checked_add(length)
1269 .ok_or(ConfigChangeDecodeError)?;
1270 let value = bytes
1271 .get(value_start..value_end)
1272 .ok_or(ConfigChangeDecodeError)?;
1273 *cursor = value_end;
1274 std::str::from_utf8(value)
1275 .map(str::to_owned)
1276 .map_err(|_| ConfigChangeDecodeError)
1277}
1278
1279fn read_config_u64_at(bytes: &[u8], cursor: &mut usize) -> Result<u64, ConfigChangeDecodeError> {
1280 let value = read_config_u64(bytes, *cursor)?;
1281 *cursor = cursor.checked_add(8).ok_or(ConfigChangeDecodeError)?;
1282 Ok(value)
1283}
1284
1285fn read_config_hash_at(
1286 bytes: &[u8],
1287 cursor: &mut usize,
1288) -> Result<LogHash, ConfigChangeDecodeError> {
1289 let value = read_config_hash(bytes, *cursor)?;
1290 *cursor = cursor.checked_add(32).ok_or(ConfigChangeDecodeError)?;
1291 Ok(value)
1292}
1293
1294#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1295pub struct ConfigChangeDecodeError;
1296
1297impl std::fmt::Display for ConfigChangeDecodeError {
1298 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1299 f.write_str("invalid ConfigChange payload")
1300 }
1301}
1302
1303impl std::error::Error for ConfigChangeDecodeError {}
1304
1305fn read_config_u16(bytes: &[u8], offset: usize) -> Result<u16, ConfigChangeDecodeError> {
1306 let end = offset.checked_add(2).ok_or(ConfigChangeDecodeError)?;
1307 let bytes = bytes.get(offset..end).ok_or(ConfigChangeDecodeError)?;
1308 Ok(u16::from_be_bytes(
1309 bytes.try_into().expect("u16 slice length"),
1310 ))
1311}
1312
1313fn read_config_u64(bytes: &[u8], offset: usize) -> Result<u64, ConfigChangeDecodeError> {
1314 let end = offset.checked_add(8).ok_or(ConfigChangeDecodeError)?;
1315 let bytes = bytes.get(offset..end).ok_or(ConfigChangeDecodeError)?;
1316 Ok(u64::from_be_bytes(
1317 bytes.try_into().expect("u64 slice length"),
1318 ))
1319}
1320
1321fn read_config_hash(bytes: &[u8], offset: usize) -> Result<LogHash, ConfigChangeDecodeError> {
1322 let end = offset.checked_add(32).ok_or(ConfigChangeDecodeError)?;
1323 let bytes: [u8; 32] = bytes
1324 .get(offset..end)
1325 .ok_or(ConfigChangeDecodeError)?
1326 .try_into()
1327 .expect("hash slice length");
1328 Ok(LogHash::from_bytes(bytes))
1329}
1330
1331#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
1332pub struct LogEntry {
1333 pub cluster_id: ClusterId,
1334 pub epoch: Epoch,
1335 pub config_id: ConfigId,
1336 pub index: LogIndex,
1337 pub entry_type: EntryType,
1338 pub payload: Vec<u8>,
1339 pub prev_hash: LogHash,
1340 pub hash: LogHash,
1341}
1342
1343impl LogEntry {
1344 pub fn calculate_hash(
1345 cluster_id: &str,
1346 index: LogIndex,
1347 epoch: Epoch,
1348 config_id: ConfigId,
1349 entry_type: EntryType,
1350 prev_hash: LogHash,
1351 payload: &[u8],
1352 ) -> LogHash {
1353 let cluster_length = (cluster_id.len() as u64).to_be_bytes();
1354 let index = index.to_be_bytes();
1355 let epoch = epoch.to_be_bytes();
1356 let config_id = config_id.to_be_bytes();
1357 let entry_type = [entry_type.as_u8()];
1358 let payload_hash = LogHash::digest(&[payload]);
1359 LogHash::digest(&[
1360 b"rhiza-log-entry-v3\0",
1361 &cluster_length,
1362 cluster_id.as_bytes(),
1363 &index,
1364 &epoch,
1365 &config_id,
1366 &entry_type,
1367 prev_hash.as_bytes(),
1368 payload_hash.as_bytes(),
1369 ])
1370 }
1371
1372 pub fn recompute_hash(&self) -> LogHash {
1373 Self::calculate_hash(
1374 &self.cluster_id,
1375 self.index,
1376 self.epoch,
1377 self.config_id,
1378 self.entry_type,
1379 self.prev_hash,
1380 &self.payload,
1381 )
1382 }
1383}
1384
1385#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
1386#[serde(deny_unknown_fields)]
1387pub struct SnapshotManifest {
1388 snapshot_id: String,
1389 cluster_id: ClusterId,
1390 config_id: ConfigId,
1391 configuration_state: ConfigurationState,
1392 epoch: Epoch,
1393 index: LogIndex,
1394 applied_hash: LogHash,
1395 schema_version: u64,
1396 created_by: NodeId,
1397 executor_fingerprint: LogHash,
1398}
1399
1400impl SnapshotManifest {
1401 #[allow(clippy::too_many_arguments)]
1402 pub fn new(
1403 cluster_id: impl Into<ClusterId>,
1404 configuration_state: ConfigurationState,
1405 epoch: Epoch,
1406 index: LogIndex,
1407 applied_hash: LogHash,
1408 schema_version: u64,
1409 created_by: impl Into<NodeId>,
1410 executor_fingerprint: LogHash,
1411 ) -> Self {
1412 Self {
1413 snapshot_id: format!("snapshot-{index:015}"),
1414 cluster_id: cluster_id.into(),
1415 config_id: configuration_state.config_id(),
1416 configuration_state,
1417 epoch,
1418 index,
1419 applied_hash,
1420 schema_version,
1421 created_by: created_by.into(),
1422 executor_fingerprint,
1423 }
1424 }
1425
1426 pub fn snapshot_id(&self) -> &str {
1427 &self.snapshot_id
1428 }
1429
1430 pub fn cluster_id(&self) -> &str {
1431 &self.cluster_id
1432 }
1433
1434 pub const fn epoch(&self) -> Epoch {
1435 self.epoch
1436 }
1437
1438 pub const fn config_id(&self) -> ConfigId {
1439 self.config_id
1440 }
1441
1442 pub const fn configuration_state(&self) -> &ConfigurationState {
1443 &self.configuration_state
1444 }
1445
1446 pub const fn schema_version(&self) -> u64 {
1447 self.schema_version
1448 }
1449
1450 pub fn created_by(&self) -> &str {
1451 &self.created_by
1452 }
1453
1454 pub const fn index(&self) -> LogIndex {
1455 self.index
1456 }
1457
1458 pub const fn snapshot_index(&self) -> LogIndex {
1459 self.index
1460 }
1461
1462 pub const fn applied_hash(&self) -> LogHash {
1463 self.applied_hash
1464 }
1465
1466 pub const fn executor_fingerprint(&self) -> LogHash {
1467 self.executor_fingerprint
1468 }
1469}
1470
1471impl<'de> serde::Deserialize<'de> for SnapshotManifest {
1472 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
1473 where
1474 D: serde::Deserializer<'de>,
1475 {
1476 #[derive(serde::Deserialize)]
1477 #[serde(deny_unknown_fields)]
1478 struct Wire {
1479 snapshot_id: String,
1480 cluster_id: ClusterId,
1481 config_id: ConfigId,
1482 configuration_state: ConfigurationState,
1483 epoch: Epoch,
1484 index: LogIndex,
1485 applied_hash: LogHash,
1486 schema_version: u64,
1487 created_by: NodeId,
1488 executor_fingerprint: LogHash,
1489 }
1490
1491 let wire = Wire::deserialize(deserializer)?;
1492 if wire.configuration_state.config_id() != wire.config_id {
1493 return Err(serde::de::Error::custom(
1494 "snapshot configuration state does not match config_id",
1495 ));
1496 }
1497 Ok(Self {
1498 snapshot_id: wire.snapshot_id,
1499 cluster_id: wire.cluster_id,
1500 config_id: wire.config_id,
1501 configuration_state: wire.configuration_state,
1502 epoch: wire.epoch,
1503 index: wire.index,
1504 applied_hash: wire.applied_hash,
1505 schema_version: wire.schema_version,
1506 created_by: wire.created_by,
1507 executor_fingerprint: wire.executor_fingerprint,
1508 })
1509 }
1510}
1511
1512#[derive(Clone, Debug, Eq, PartialEq)]
1513pub struct Snapshot {
1514 manifest: SnapshotManifest,
1515 db_bytes: Vec<u8>,
1516}
1517
1518impl Snapshot {
1519 pub fn new(manifest: SnapshotManifest, db_bytes: Vec<u8>) -> Self {
1520 Self { manifest, db_bytes }
1521 }
1522
1523 pub const fn manifest(&self) -> &SnapshotManifest {
1524 &self.manifest
1525 }
1526
1527 pub fn db_bytes(&self) -> &[u8] {
1528 &self.db_bytes
1529 }
1530}
1531
1532#[cfg(test)]
1533mod tests {
1534 use super::{
1535 read_config_hash, read_config_hash_at, read_config_string, read_config_u16,
1536 read_config_u64, read_config_u64_at, CommandEnvelopeError, ConfigChangeDecodeError,
1537 ErrorCategory, ErrorClassification, ExecutionProfile, ExecutionProfileParseError,
1538 ReplicatedCommandEnvelope,
1539 };
1540
1541 #[test]
1542 fn server_error_codes_map_to_categories_and_preserve_unknown_values() {
1543 let cases = [
1544 ("invalid_request", ErrorCategory::InvalidRequest, false),
1545 ("invalid_json", ErrorCategory::InvalidRequest, false),
1546 ("invalid_content_type", ErrorCategory::InvalidRequest, false),
1547 ("unauthorized", ErrorCategory::Authentication, false),
1548 ("request_conflict", ErrorCategory::Conflict, false),
1549 ("precondition_failed", ErrorCategory::Conflict, false),
1550 ("unavailable", ErrorCategory::Unavailable, true),
1551 ("durability_unavailable", ErrorCategory::Unavailable, true),
1552 ("write_timeout", ErrorCategory::Unavailable, true),
1553 ("writes_unavailable", ErrorCategory::Unavailable, true),
1554 ("configuration_transition", ErrorCategory::Unavailable, true),
1555 ("contention", ErrorCategory::Unavailable, true),
1556 ("winner_limit_exceeded", ErrorCategory::Unavailable, true),
1557 ("leader_unavailable", ErrorCategory::Unavailable, true),
1558 ("snapshot_required", ErrorCategory::Unavailable, false),
1559 ("resource_exhausted", ErrorCategory::ResourceExhausted, true),
1560 ("overloaded", ErrorCategory::ResourceExhausted, true),
1561 ("payload_too_large", ErrorCategory::ResourceExhausted, false),
1562 ("data_root_locked", ErrorCategory::Internal, false),
1563 ("unsupported_ack_mode", ErrorCategory::Internal, false),
1564 ("execution_profile_mismatch", ErrorCategory::Internal, false),
1565 ("storage_error", ErrorCategory::Internal, false),
1566 ("reconciliation_error", ErrorCategory::Internal, false),
1567 ("invariant_violation", ErrorCategory::Internal, false),
1568 ("fatal", ErrorCategory::Internal, false),
1569 ("task_failed", ErrorCategory::Internal, false),
1570 ("future_code", ErrorCategory::Unknown, true),
1571 ];
1572
1573 for (code, category, retryable) in cases {
1574 let classification = ErrorClassification::from_server_code(code, retryable);
1575
1576 assert_eq!(classification.code(), code);
1577 assert_eq!(classification.category(), category);
1578 assert_eq!(classification.retryable(), retryable);
1579 }
1580 }
1581
1582 #[test]
1583 fn execution_profile_has_stable_text_and_wire_ids() {
1584 assert_eq!(ExecutionProfile::Sqlite.as_str(), "sql");
1585 assert_eq!(ExecutionProfile::Graph.as_str(), "graph");
1586 assert_eq!(ExecutionProfile::Kv.as_str(), "kv");
1587 assert_eq!(ExecutionProfile::Sqlite.to_string(), "sql");
1588 assert_eq!(ExecutionProfile::Graph.to_string(), "graph");
1589 assert_eq!(ExecutionProfile::Kv.to_string(), "kv");
1590 assert_eq!("sql".parse(), Ok(ExecutionProfile::Sqlite));
1591 assert_eq!("graph".parse(), Ok(ExecutionProfile::Graph));
1592 assert_eq!("kv".parse(), Ok(ExecutionProfile::Kv));
1593 assert_eq!(
1594 "sqlite".parse::<ExecutionProfile>(),
1595 Err(ExecutionProfileParseError)
1596 );
1597 assert_eq!(
1598 serde_json::to_value(ExecutionProfile::Sqlite).unwrap(),
1599 serde_json::json!("sql")
1600 );
1601 assert_eq!(
1602 serde_json::from_value::<ExecutionProfile>(serde_json::json!("sql")).unwrap(),
1603 ExecutionProfile::Sqlite
1604 );
1605 assert!(serde_json::from_value::<ExecutionProfile>(serde_json::json!("sqlite")).is_err());
1606 assert_eq!(
1607 "ladybug".parse::<ExecutionProfile>(),
1608 Err(ExecutionProfileParseError)
1609 );
1610 assert_eq!(ExecutionProfile::Sqlite.wire_id(), 1);
1611 assert_eq!(ExecutionProfile::Graph.wire_id(), 2);
1612 assert_eq!(ExecutionProfile::Kv.wire_id(), 3);
1613 assert_eq!(
1614 ExecutionProfile::from_wire_id(1),
1615 Some(ExecutionProfile::Sqlite)
1616 );
1617 assert_eq!(
1618 ExecutionProfile::from_wire_id(2),
1619 Some(ExecutionProfile::Graph)
1620 );
1621 assert_eq!(
1622 ExecutionProfile::from_wire_id(3),
1623 Some(ExecutionProfile::Kv)
1624 );
1625 assert_eq!(ExecutionProfile::from_wire_id(0), None);
1626 }
1627
1628 #[test]
1629 fn replicated_command_envelope_uses_a_stable_explicit_codec() {
1630 let envelope = ReplicatedCommandEnvelope::new(
1631 ExecutionProfile::Graph,
1632 3,
1633 "req-1",
1634 vec![0xaa, 0xbb, 0xcc],
1635 )
1636 .unwrap();
1637 let encoded = envelope.encode().unwrap();
1638
1639 assert_eq!(
1640 encoded,
1641 [
1642 b'Q', b'C', b'M', b'D', 0, 1, 2, 0, 3, 0, 5, 0, 0, 0, 3, b'r', b'e', b'q', b'-',
1643 b'1', 0xaa, 0xbb, 0xcc,
1644 ]
1645 );
1646 assert_eq!(ReplicatedCommandEnvelope::decode(&encoded), Ok(envelope));
1647 }
1648
1649 #[test]
1650 fn replicated_command_envelope_preserves_backend_bytes_unchanged() {
1651 let opaque_qsql = b"QSQL\0\x02{\"request_id\":\"same\"}".to_vec();
1652 let envelope = ReplicatedCommandEnvelope::new(
1653 ExecutionProfile::Sqlite,
1654 1,
1655 "same",
1656 opaque_qsql.clone(),
1657 )
1658 .unwrap();
1659
1660 let decoded = ReplicatedCommandEnvelope::decode(&envelope.encode().unwrap()).unwrap();
1661 assert_eq!(decoded.body(), opaque_qsql);
1662 }
1663
1664 #[test]
1665 fn replicated_command_envelope_rejects_noncanonical_or_invalid_inputs() {
1666 assert_eq!(
1667 ReplicatedCommandEnvelope::new(ExecutionProfile::Graph, 0, "req", Vec::new()),
1668 Err(CommandEnvelopeError::InvalidCommandVersion)
1669 );
1670 assert_eq!(
1671 ReplicatedCommandEnvelope::new(ExecutionProfile::Graph, 1, "", Vec::new()),
1672 Err(CommandEnvelopeError::EmptyRequestId)
1673 );
1674
1675 let valid =
1676 ReplicatedCommandEnvelope::new(ExecutionProfile::Graph, 1, "req", vec![1, 2, 3])
1677 .unwrap()
1678 .encode()
1679 .unwrap();
1680 let mut unknown_profile = valid.clone();
1681 unknown_profile[6] = 9;
1682 assert_eq!(
1683 ReplicatedCommandEnvelope::decode(&unknown_profile),
1684 Err(CommandEnvelopeError::UnknownExecutionProfile(9))
1685 );
1686
1687 let mut trailing = valid;
1688 trailing.push(0);
1689 assert_eq!(
1690 ReplicatedCommandEnvelope::decode(&trailing),
1691 Err(CommandEnvelopeError::TrailingBytes)
1692 );
1693 }
1694
1695 #[test]
1696 fn replicated_command_envelope_rejects_malformed_wire_boundaries() {
1697 let valid =
1698 ReplicatedCommandEnvelope::new(ExecutionProfile::Graph, 1, "req", vec![1, 2, 3])
1699 .unwrap()
1700 .encode()
1701 .unwrap();
1702
1703 let mut invalid_magic = valid.clone();
1704 invalid_magic[0] = b'X';
1705 assert_eq!(
1706 ReplicatedCommandEnvelope::decode(&invalid_magic),
1707 Err(CommandEnvelopeError::InvalidMagic)
1708 );
1709
1710 let mut unsupported_version = valid.clone();
1711 unsupported_version[4..6].copy_from_slice(&2_u16.to_be_bytes());
1712 assert_eq!(
1713 ReplicatedCommandEnvelope::decode(&unsupported_version),
1714 Err(CommandEnvelopeError::UnsupportedFormatVersion(2))
1715 );
1716
1717 assert_eq!(
1718 ReplicatedCommandEnvelope::decode(&valid[..14]),
1719 Err(CommandEnvelopeError::Truncated)
1720 );
1721
1722 let mut truncated_body = valid.clone();
1723 truncated_body[11..15].copy_from_slice(&4_u32.to_be_bytes());
1724 assert_eq!(
1725 ReplicatedCommandEnvelope::decode(&truncated_body),
1726 Err(CommandEnvelopeError::Truncated)
1727 );
1728
1729 let mut invalid_request_utf8 = valid.clone();
1730 invalid_request_utf8[15] = 0xff;
1731 assert_eq!(
1732 ReplicatedCommandEnvelope::decode(&invalid_request_utf8),
1733 Err(CommandEnvelopeError::InvalidRequestIdUtf8)
1734 );
1735
1736 let mut oversized_request = valid.clone();
1737 oversized_request[9..11].copy_from_slice(&4_u16.to_be_bytes());
1738 assert_eq!(
1739 ReplicatedCommandEnvelope::decode(&oversized_request),
1740 Err(CommandEnvelopeError::Truncated)
1741 );
1742
1743 let mut undersized_body = valid.clone();
1744 undersized_body[11..15].copy_from_slice(&2_u32.to_be_bytes());
1745 assert_eq!(
1746 ReplicatedCommandEnvelope::decode(&undersized_body),
1747 Err(CommandEnvelopeError::TrailingBytes)
1748 );
1749 }
1750
1751 #[test]
1752 fn replicated_command_decoder_never_panics_and_successes_are_canonical() {
1753 let mut state = 0x9e37_79b9_7f4a_7c15_u64;
1754 for length in 0..=256 {
1755 for _ in 0..8 {
1756 let mut bytes = Vec::with_capacity(length);
1757 for _ in 0..length {
1758 state = state
1759 .wrapping_mul(6_364_136_223_846_793_005)
1760 .wrapping_add(1_442_695_040_888_963_407);
1761 bytes.push((state >> 32) as u8);
1762 }
1763 let decoded =
1764 std::panic::catch_unwind(|| ReplicatedCommandEnvelope::decode(&bytes));
1765 let decoded = decoded.expect("decoder must not panic for arbitrary bytes");
1766 if let Ok(envelope) = decoded {
1767 assert_eq!(envelope.encode().unwrap(), bytes);
1768 }
1769 }
1770 }
1771
1772 for sequence in 1_u16..=256 {
1773 state = state
1774 .wrapping_mul(6_364_136_223_846_793_005)
1775 .wrapping_add(1_442_695_040_888_963_407);
1776 let body = state.to_be_bytes()[..usize::from(sequence % 8)].to_vec();
1777 let profile = match sequence % 3 {
1778 0 => ExecutionProfile::Sqlite,
1779 1 => ExecutionProfile::Graph,
1780 _ => ExecutionProfile::Kv,
1781 };
1782 let envelope = ReplicatedCommandEnvelope::new(
1783 profile,
1784 sequence,
1785 format!("request-{sequence}"),
1786 body,
1787 )
1788 .unwrap();
1789 let encoded = envelope.encode().unwrap();
1790 let decoded = ReplicatedCommandEnvelope::decode(&encoded).unwrap();
1791 assert_eq!(decoded.encode().unwrap(), encoded);
1792 }
1793 }
1794
1795 #[test]
1796 fn config_decoder_rejects_overflowing_offsets_without_panicking() {
1797 let bytes = [];
1798 assert_eq!(
1799 read_config_u16(&bytes, usize::MAX),
1800 Err(ConfigChangeDecodeError)
1801 );
1802 assert_eq!(
1803 read_config_u64(&bytes, usize::MAX),
1804 Err(ConfigChangeDecodeError)
1805 );
1806 assert_eq!(
1807 read_config_hash(&bytes, usize::MAX),
1808 Err(ConfigChangeDecodeError)
1809 );
1810
1811 let mut cursor = usize::MAX;
1812 assert_eq!(
1813 read_config_string(&bytes, &mut cursor),
1814 Err(ConfigChangeDecodeError)
1815 );
1816 assert_eq!(
1817 read_config_u64_at(&bytes, &mut cursor),
1818 Err(ConfigChangeDecodeError)
1819 );
1820 assert_eq!(
1821 read_config_hash_at(&bytes, &mut cursor),
1822 Err(ConfigChangeDecodeError)
1823 );
1824 }
1825}