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";
868const CONFIG_CHANGE_VERSION: u16 = 1;
869const BOUND_CONFIG_CHANGE_VERSION: u16 = 2;
870
871#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
872pub struct SuccessorDescriptor {
873 cluster_id: ClusterId,
874 predecessor_config_id: ConfigId,
875 predecessor_config_digest: LogHash,
876 config_id: ConfigId,
877 config_digest: LogHash,
878 members: Vec<NodeId>,
879}
880
881impl<'de> serde::Deserialize<'de> for SuccessorDescriptor {
882 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
883 where
884 D: serde::Deserializer<'de>,
885 {
886 #[derive(serde::Deserialize)]
887 #[serde(deny_unknown_fields)]
888 struct Wire {
889 cluster_id: ClusterId,
890 predecessor_config_id: ConfigId,
891 predecessor_config_digest: LogHash,
892 config_id: ConfigId,
893 config_digest: LogHash,
894 members: Vec<NodeId>,
895 }
896
897 let wire = Wire::deserialize(deserializer)?;
898 if !wire.members.windows(2).all(|pair| pair[0] < pair[1]) {
899 return Err(serde::de::Error::custom(
900 "successor members are not canonical",
901 ));
902 }
903 let encoded_digest = wire.config_digest;
904 let descriptor = Self::new(
905 wire.cluster_id,
906 wire.predecessor_config_id,
907 wire.predecessor_config_digest,
908 wire.config_id,
909 wire.members,
910 )
911 .map_err(serde::de::Error::custom)?;
912 if descriptor.config_digest != encoded_digest {
913 return Err(serde::de::Error::custom(
914 "successor membership digest mismatch",
915 ));
916 }
917 Ok(descriptor)
918 }
919}
920
921impl SuccessorDescriptor {
922 pub fn new(
923 cluster_id: impl Into<ClusterId>,
924 predecessor_config_id: ConfigId,
925 predecessor_config_digest: LogHash,
926 config_id: ConfigId,
927 members: Vec<NodeId>,
928 ) -> Result<Self, ConfigChangeDecodeError> {
929 let cluster_id = cluster_id.into();
930 if cluster_id.is_empty() || cluster_id.len() > usize::from(u16::MAX) {
931 return Err(ConfigChangeDecodeError);
932 }
933 if predecessor_config_id.checked_add(1) != Some(config_id) {
934 return Err(ConfigChangeDecodeError);
935 }
936 let mut canonical = members;
937 let original_len = canonical.len();
938 canonical.sort();
939 canonical.dedup();
940 if canonical.len() != original_len
941 || !(3..=7).contains(&canonical.len())
942 || canonical
943 .iter()
944 .any(|member| member.is_empty() || member.len() > usize::from(u16::MAX))
945 {
946 return Err(ConfigChangeDecodeError);
947 }
948 let config_digest = canonical_membership_digest(&canonical)?;
949 Ok(Self {
950 cluster_id,
951 predecessor_config_id,
952 predecessor_config_digest,
953 config_id,
954 config_digest,
955 members: canonical,
956 })
957 }
958
959 pub fn cluster_id(&self) -> &str {
960 &self.cluster_id
961 }
962
963 pub const fn predecessor_config_id(&self) -> ConfigId {
964 self.predecessor_config_id
965 }
966
967 pub const fn predecessor_config_digest(&self) -> LogHash {
968 self.predecessor_config_digest
969 }
970
971 pub const fn config_id(&self) -> ConfigId {
972 self.config_id
973 }
974
975 pub const fn digest(&self) -> LogHash {
976 self.config_digest
977 }
978
979 pub fn members(&self) -> &[NodeId] {
980 &self.members
981 }
982}
983
984pub fn canonical_membership_digest(members: &[NodeId]) -> Result<LogHash, ConfigChangeDecodeError> {
985 if !(3..=7).contains(&members.len())
986 || members
987 .iter()
988 .any(|member| member.is_empty() || member.len() > usize::from(u16::MAX))
989 || !members.windows(2).all(|pair| pair[0] < pair[1])
990 {
991 return Err(ConfigChangeDecodeError);
992 }
993 let encoded_len = 14 + members.len() * 8 + members.iter().map(String::len).sum::<usize>();
994 let mut encoded = Vec::with_capacity(encoded_len);
995 encoded.extend_from_slice(b"QMEM\0\x01");
996 encoded.extend_from_slice(&(members.len() as u64).to_be_bytes());
997 for member in members {
998 encoded.extend_from_slice(&(member.len() as u64).to_be_bytes());
999 encoded.extend_from_slice(member.as_bytes());
1000 }
1001 Ok(LogHash::digest(&[&encoded]))
1002}
1003
1004#[derive(Clone, Debug, Eq, PartialEq)]
1005pub enum ConfigChange {
1006 Stop {
1007 config_id: ConfigId,
1008 config_digest: LogHash,
1009 },
1010 ActivationBarrier {
1011 config_id: ConfigId,
1012 config_digest: LogHash,
1013 stop_slot: LogIndex,
1014 prefix_hash: LogHash,
1015 },
1016 BoundStop {
1017 successor: SuccessorDescriptor,
1018 },
1019 BoundActivationBarrier {
1020 successor: SuccessorDescriptor,
1021 stop_slot: LogIndex,
1022 prefix_hash: LogHash,
1023 stop_command_hash: LogHash,
1024 },
1025}
1026
1027impl ConfigChange {
1028 pub const fn stop(config_id: ConfigId, config_digest: LogHash) -> Self {
1029 Self::Stop {
1030 config_id,
1031 config_digest,
1032 }
1033 }
1034
1035 pub const fn activation_barrier(
1036 config_id: ConfigId,
1037 config_digest: LogHash,
1038 stop_slot: LogIndex,
1039 prefix_hash: LogHash,
1040 ) -> Self {
1041 Self::ActivationBarrier {
1042 config_id,
1043 config_digest,
1044 stop_slot,
1045 prefix_hash,
1046 }
1047 }
1048
1049 pub fn bound_stop(
1050 cluster_id: impl Into<ClusterId>,
1051 predecessor_config_id: ConfigId,
1052 predecessor_config_digest: LogHash,
1053 successor_config_id: ConfigId,
1054 successor_members: Vec<NodeId>,
1055 ) -> Result<Self, ConfigChangeDecodeError> {
1056 Ok(Self::BoundStop {
1057 successor: SuccessorDescriptor::new(
1058 cluster_id,
1059 predecessor_config_id,
1060 predecessor_config_digest,
1061 successor_config_id,
1062 successor_members,
1063 )?,
1064 })
1065 }
1066
1067 pub const fn bound_activation_barrier(
1068 successor: SuccessorDescriptor,
1069 stop_slot: LogIndex,
1070 prefix_hash: LogHash,
1071 stop_command_hash: LogHash,
1072 ) -> Self {
1073 Self::BoundActivationBarrier {
1074 successor,
1075 stop_slot,
1076 prefix_hash,
1077 stop_command_hash,
1078 }
1079 }
1080
1081 pub const fn successor(&self) -> Option<&SuccessorDescriptor> {
1082 match self {
1083 Self::BoundStop { successor } | Self::BoundActivationBarrier { successor, .. } => {
1084 Some(successor)
1085 }
1086 _ => None,
1087 }
1088 }
1089
1090 pub fn to_stored_command(&self) -> StoredCommand {
1091 let mut payload = Vec::with_capacity(87);
1092 payload.extend_from_slice(CONFIG_CHANGE_MAGIC);
1093 let version = if matches!(
1094 self,
1095 Self::BoundStop { .. } | Self::BoundActivationBarrier { .. }
1096 ) {
1097 BOUND_CONFIG_CHANGE_VERSION
1098 } else {
1099 CONFIG_CHANGE_VERSION
1100 };
1101 payload.extend_from_slice(&version.to_be_bytes());
1102 match self {
1103 Self::Stop {
1104 config_id,
1105 config_digest,
1106 } => {
1107 payload.push(1);
1108 payload.extend_from_slice(&config_id.to_be_bytes());
1109 payload.extend_from_slice(config_digest.as_bytes());
1110 }
1111 Self::ActivationBarrier {
1112 config_id,
1113 config_digest,
1114 stop_slot,
1115 prefix_hash,
1116 } => {
1117 payload.push(2);
1118 payload.extend_from_slice(&config_id.to_be_bytes());
1119 payload.extend_from_slice(config_digest.as_bytes());
1120 payload.extend_from_slice(&stop_slot.to_be_bytes());
1121 payload.extend_from_slice(prefix_hash.as_bytes());
1122 }
1123 Self::BoundStop { successor } => {
1124 payload.push(1);
1125 encode_successor(&mut payload, successor);
1126 }
1127 Self::BoundActivationBarrier {
1128 successor,
1129 stop_slot,
1130 prefix_hash,
1131 stop_command_hash,
1132 } => {
1133 payload.push(2);
1134 encode_successor(&mut payload, successor);
1135 payload.extend_from_slice(&stop_slot.to_be_bytes());
1136 payload.extend_from_slice(prefix_hash.as_bytes());
1137 payload.extend_from_slice(stop_command_hash.as_bytes());
1138 }
1139 }
1140 StoredCommand::new(EntryType::ConfigChange, payload)
1141 }
1142
1143 pub fn recognize(command: &StoredCommand) -> Result<Self, ConfigChangeDecodeError> {
1144 Self::recognize_parts(command.entry_type, &command.payload)
1145 }
1146
1147 pub fn recognize_parts(
1148 entry_type: EntryType,
1149 payload: &[u8],
1150 ) -> Result<Self, ConfigChangeDecodeError> {
1151 if entry_type != EntryType::ConfigChange {
1152 return Err(ConfigChangeDecodeError);
1153 }
1154 let bytes = payload;
1155 if bytes.get(..4) != Some(CONFIG_CHANGE_MAGIC) {
1156 return Err(ConfigChangeDecodeError);
1157 }
1158 let version = read_config_u16(bytes, 4)?;
1159 if version == BOUND_CONFIG_CHANGE_VERSION {
1160 let kind = *bytes.get(6).ok_or(ConfigChangeDecodeError)?;
1161 let mut cursor = 7;
1162 let successor = decode_successor(bytes, &mut cursor)?;
1163 let change = match kind {
1164 1 => Self::BoundStop { successor },
1165 2 => Self::BoundActivationBarrier {
1166 successor,
1167 stop_slot: read_config_u64_at(bytes, &mut cursor)?,
1168 prefix_hash: read_config_hash_at(bytes, &mut cursor)?,
1169 stop_command_hash: read_config_hash_at(bytes, &mut cursor)?,
1170 },
1171 _ => return Err(ConfigChangeDecodeError),
1172 };
1173 if cursor != bytes.len() {
1174 return Err(ConfigChangeDecodeError);
1175 }
1176 return Ok(change);
1177 }
1178 if version != CONFIG_CHANGE_VERSION {
1179 return Err(ConfigChangeDecodeError);
1180 }
1181 let kind = *bytes.get(6).ok_or(ConfigChangeDecodeError)?;
1182 let config_id = read_config_u64(bytes, 7)?;
1183 let config_digest = read_config_hash(bytes, 15)?;
1184 match kind {
1185 1 if bytes.len() == 47 => Ok(Self::stop(config_id, config_digest)),
1186 2 if bytes.len() == 87 => Ok(Self::activation_barrier(
1187 config_id,
1188 config_digest,
1189 read_config_u64(bytes, 47)?,
1190 read_config_hash(bytes, 55)?,
1191 )),
1192 _ => Err(ConfigChangeDecodeError),
1193 }
1194 }
1195
1196 pub const fn binding(&self) -> (ConfigId, LogHash) {
1197 match self {
1198 Self::Stop {
1199 config_id,
1200 config_digest,
1201 }
1202 | Self::ActivationBarrier {
1203 config_id,
1204 config_digest,
1205 ..
1206 } => (*config_id, *config_digest),
1207 Self::BoundStop { successor } => (
1208 successor.predecessor_config_id,
1209 successor.predecessor_config_digest,
1210 ),
1211 Self::BoundActivationBarrier { successor, .. } => {
1212 (successor.config_id, successor.config_digest)
1213 }
1214 }
1215 }
1216}
1217
1218fn encode_successor(out: &mut Vec<u8>, successor: &SuccessorDescriptor) {
1219 let cluster = successor.cluster_id.as_bytes();
1220 let encoded_len = 83
1221 + cluster.len()
1222 + successor
1223 .members
1224 .iter()
1225 .map(|member| 2 + member.len())
1226 .sum::<usize>();
1227 out.reserve(encoded_len);
1228 let cluster_length =
1229 u16::try_from(cluster.len()).expect("validated successor cluster length fits u16");
1230 out.extend_from_slice(&cluster_length.to_be_bytes());
1231 out.extend_from_slice(cluster);
1232 out.extend_from_slice(&successor.predecessor_config_id.to_be_bytes());
1233 out.extend_from_slice(successor.predecessor_config_digest.as_bytes());
1234 out.extend_from_slice(&successor.config_id.to_be_bytes());
1235 out.extend_from_slice(successor.config_digest.as_bytes());
1236 let member_count =
1237 u8::try_from(successor.members.len()).expect("validated successor member count fits u8");
1238 out.push(member_count);
1239 for member in &successor.members {
1240 let member_length =
1241 u16::try_from(member.len()).expect("validated successor member length fits u16");
1242 out.extend_from_slice(&member_length.to_be_bytes());
1243 out.extend_from_slice(member.as_bytes());
1244 }
1245}
1246
1247fn decode_successor(
1248 bytes: &[u8],
1249 cursor: &mut usize,
1250) -> Result<SuccessorDescriptor, ConfigChangeDecodeError> {
1251 let cluster_id = read_config_string(bytes, cursor)?;
1252 let predecessor_config_id = read_config_u64_at(bytes, cursor)?;
1253 let predecessor_config_digest = read_config_hash_at(bytes, cursor)?;
1254 let config_id = read_config_u64_at(bytes, cursor)?;
1255 let encoded_digest = read_config_hash_at(bytes, cursor)?;
1256 let count = *bytes.get(*cursor).ok_or(ConfigChangeDecodeError)? as usize;
1257 if !(3..=7).contains(&count) {
1258 return Err(ConfigChangeDecodeError);
1259 }
1260 *cursor += 1;
1261 let members = (0..count)
1262 .map(|_| read_config_string(bytes, cursor))
1263 .collect::<Result<Vec<_>, _>>()?;
1264 if !members.windows(2).all(|pair| pair[0] < pair[1]) {
1265 return Err(ConfigChangeDecodeError);
1266 }
1267 let descriptor = SuccessorDescriptor::new(
1268 cluster_id,
1269 predecessor_config_id,
1270 predecessor_config_digest,
1271 config_id,
1272 members,
1273 )?;
1274 if descriptor.config_digest != encoded_digest {
1275 return Err(ConfigChangeDecodeError);
1276 }
1277 Ok(descriptor)
1278}
1279
1280fn read_config_string(bytes: &[u8], cursor: &mut usize) -> Result<String, ConfigChangeDecodeError> {
1281 let length = read_config_u16(bytes, *cursor)? as usize;
1282 let value_start = cursor.checked_add(2).ok_or(ConfigChangeDecodeError)?;
1283 let value_end = value_start
1284 .checked_add(length)
1285 .ok_or(ConfigChangeDecodeError)?;
1286 let value = bytes
1287 .get(value_start..value_end)
1288 .ok_or(ConfigChangeDecodeError)?;
1289 *cursor = value_end;
1290 std::str::from_utf8(value)
1291 .map(str::to_owned)
1292 .map_err(|_| ConfigChangeDecodeError)
1293}
1294
1295fn read_config_u64_at(bytes: &[u8], cursor: &mut usize) -> Result<u64, ConfigChangeDecodeError> {
1296 let value = read_config_u64(bytes, *cursor)?;
1297 *cursor = cursor.checked_add(8).ok_or(ConfigChangeDecodeError)?;
1298 Ok(value)
1299}
1300
1301fn read_config_hash_at(
1302 bytes: &[u8],
1303 cursor: &mut usize,
1304) -> Result<LogHash, ConfigChangeDecodeError> {
1305 let value = read_config_hash(bytes, *cursor)?;
1306 *cursor = cursor.checked_add(32).ok_or(ConfigChangeDecodeError)?;
1307 Ok(value)
1308}
1309
1310#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1311pub struct ConfigChangeDecodeError;
1312
1313impl std::fmt::Display for ConfigChangeDecodeError {
1314 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1315 f.write_str("invalid ConfigChange payload")
1316 }
1317}
1318
1319impl std::error::Error for ConfigChangeDecodeError {}
1320
1321fn read_config_u16(bytes: &[u8], offset: usize) -> Result<u16, ConfigChangeDecodeError> {
1322 let end = offset.checked_add(2).ok_or(ConfigChangeDecodeError)?;
1323 let bytes = bytes.get(offset..end).ok_or(ConfigChangeDecodeError)?;
1324 Ok(u16::from_be_bytes(
1325 bytes.try_into().expect("u16 slice length"),
1326 ))
1327}
1328
1329fn read_config_u64(bytes: &[u8], offset: usize) -> Result<u64, ConfigChangeDecodeError> {
1330 let end = offset.checked_add(8).ok_or(ConfigChangeDecodeError)?;
1331 let bytes = bytes.get(offset..end).ok_or(ConfigChangeDecodeError)?;
1332 Ok(u64::from_be_bytes(
1333 bytes.try_into().expect("u64 slice length"),
1334 ))
1335}
1336
1337fn read_config_hash(bytes: &[u8], offset: usize) -> Result<LogHash, ConfigChangeDecodeError> {
1338 let end = offset.checked_add(32).ok_or(ConfigChangeDecodeError)?;
1339 let bytes: [u8; 32] = bytes
1340 .get(offset..end)
1341 .ok_or(ConfigChangeDecodeError)?
1342 .try_into()
1343 .expect("hash slice length");
1344 Ok(LogHash::from_bytes(bytes))
1345}
1346
1347#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
1348pub struct LogEntry {
1349 pub cluster_id: ClusterId,
1350 pub epoch: Epoch,
1351 pub config_id: ConfigId,
1352 pub index: LogIndex,
1353 pub entry_type: EntryType,
1354 pub payload: Vec<u8>,
1355 pub prev_hash: LogHash,
1356 pub hash: LogHash,
1357}
1358
1359impl LogEntry {
1360 pub fn calculate_hash(
1361 cluster_id: &str,
1362 index: LogIndex,
1363 epoch: Epoch,
1364 config_id: ConfigId,
1365 entry_type: EntryType,
1366 prev_hash: LogHash,
1367 payload: &[u8],
1368 ) -> LogHash {
1369 let cluster_length = (cluster_id.len() as u64).to_be_bytes();
1370 let index = index.to_be_bytes();
1371 let epoch = epoch.to_be_bytes();
1372 let config_id = config_id.to_be_bytes();
1373 let entry_type = [entry_type.as_u8()];
1374 let payload_hash = LogHash::digest(&[payload]);
1375 LogHash::digest(&[
1376 b"rhiza-log-entry-v3\0",
1377 &cluster_length,
1378 cluster_id.as_bytes(),
1379 &index,
1380 &epoch,
1381 &config_id,
1382 &entry_type,
1383 prev_hash.as_bytes(),
1384 payload_hash.as_bytes(),
1385 ])
1386 }
1387
1388 pub fn recompute_hash(&self) -> LogHash {
1389 Self::calculate_hash(
1390 &self.cluster_id,
1391 self.index,
1392 self.epoch,
1393 self.config_id,
1394 self.entry_type,
1395 self.prev_hash,
1396 &self.payload,
1397 )
1398 }
1399}
1400
1401#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
1402#[serde(deny_unknown_fields)]
1403pub struct SnapshotManifest {
1404 snapshot_id: String,
1405 cluster_id: ClusterId,
1406 config_id: ConfigId,
1407 configuration_state: ConfigurationState,
1408 epoch: Epoch,
1409 index: LogIndex,
1410 applied_hash: LogHash,
1411 schema_version: u64,
1412 created_by: NodeId,
1413 executor_fingerprint: LogHash,
1414}
1415
1416impl SnapshotManifest {
1417 #[allow(clippy::too_many_arguments)]
1418 pub fn new(
1419 cluster_id: impl Into<ClusterId>,
1420 configuration_state: ConfigurationState,
1421 epoch: Epoch,
1422 index: LogIndex,
1423 applied_hash: LogHash,
1424 schema_version: u64,
1425 created_by: impl Into<NodeId>,
1426 executor_fingerprint: LogHash,
1427 ) -> Self {
1428 Self {
1429 snapshot_id: format!("snapshot-{index:015}"),
1430 cluster_id: cluster_id.into(),
1431 config_id: configuration_state.config_id(),
1432 configuration_state,
1433 epoch,
1434 index,
1435 applied_hash,
1436 schema_version,
1437 created_by: created_by.into(),
1438 executor_fingerprint,
1439 }
1440 }
1441
1442 pub fn snapshot_id(&self) -> &str {
1443 &self.snapshot_id
1444 }
1445
1446 pub fn cluster_id(&self) -> &str {
1447 &self.cluster_id
1448 }
1449
1450 pub const fn epoch(&self) -> Epoch {
1451 self.epoch
1452 }
1453
1454 pub const fn config_id(&self) -> ConfigId {
1455 self.config_id
1456 }
1457
1458 pub const fn configuration_state(&self) -> &ConfigurationState {
1459 &self.configuration_state
1460 }
1461
1462 pub const fn schema_version(&self) -> u64 {
1463 self.schema_version
1464 }
1465
1466 pub fn created_by(&self) -> &str {
1467 &self.created_by
1468 }
1469
1470 pub const fn index(&self) -> LogIndex {
1471 self.index
1472 }
1473
1474 pub const fn snapshot_index(&self) -> LogIndex {
1475 self.index
1476 }
1477
1478 pub const fn applied_hash(&self) -> LogHash {
1479 self.applied_hash
1480 }
1481
1482 pub const fn executor_fingerprint(&self) -> LogHash {
1483 self.executor_fingerprint
1484 }
1485}
1486
1487impl<'de> serde::Deserialize<'de> for SnapshotManifest {
1488 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
1489 where
1490 D: serde::Deserializer<'de>,
1491 {
1492 #[derive(serde::Deserialize)]
1493 #[serde(deny_unknown_fields)]
1494 struct Wire {
1495 snapshot_id: String,
1496 cluster_id: ClusterId,
1497 config_id: ConfigId,
1498 configuration_state: ConfigurationState,
1499 epoch: Epoch,
1500 index: LogIndex,
1501 applied_hash: LogHash,
1502 schema_version: u64,
1503 created_by: NodeId,
1504 executor_fingerprint: LogHash,
1505 }
1506
1507 let wire = Wire::deserialize(deserializer)?;
1508 if wire.configuration_state.config_id() != wire.config_id {
1509 return Err(serde::de::Error::custom(
1510 "snapshot configuration state does not match config_id",
1511 ));
1512 }
1513 Ok(Self {
1514 snapshot_id: wire.snapshot_id,
1515 cluster_id: wire.cluster_id,
1516 config_id: wire.config_id,
1517 configuration_state: wire.configuration_state,
1518 epoch: wire.epoch,
1519 index: wire.index,
1520 applied_hash: wire.applied_hash,
1521 schema_version: wire.schema_version,
1522 created_by: wire.created_by,
1523 executor_fingerprint: wire.executor_fingerprint,
1524 })
1525 }
1526}
1527
1528#[derive(Clone, Debug, Eq, PartialEq)]
1529pub struct Snapshot {
1530 manifest: SnapshotManifest,
1531 db_bytes: Vec<u8>,
1532}
1533
1534impl Snapshot {
1535 pub fn new(manifest: SnapshotManifest, db_bytes: Vec<u8>) -> Self {
1536 Self { manifest, db_bytes }
1537 }
1538
1539 pub const fn manifest(&self) -> &SnapshotManifest {
1540 &self.manifest
1541 }
1542
1543 pub fn db_bytes(&self) -> &[u8] {
1544 &self.db_bytes
1545 }
1546}
1547
1548#[cfg(test)]
1549mod tests {
1550 use super::{
1551 read_config_hash, read_config_hash_at, read_config_string, read_config_u16,
1552 read_config_u64, read_config_u64_at, CommandEnvelopeError, ConfigChangeDecodeError,
1553 ErrorCategory, ErrorClassification, ExecutionProfile, ExecutionProfileParseError,
1554 ReplicatedCommandEnvelope,
1555 };
1556
1557 #[test]
1558 fn server_error_codes_map_to_categories_and_preserve_unknown_values() {
1559 let cases = [
1560 ("invalid_request", ErrorCategory::InvalidRequest, false),
1561 ("invalid_json", ErrorCategory::InvalidRequest, false),
1562 ("invalid_content_type", ErrorCategory::InvalidRequest, false),
1563 ("unauthorized", ErrorCategory::Authentication, false),
1564 ("request_conflict", ErrorCategory::Conflict, false),
1565 ("precondition_failed", ErrorCategory::Conflict, false),
1566 ("unavailable", ErrorCategory::Unavailable, true),
1567 ("durability_unavailable", ErrorCategory::Unavailable, true),
1568 ("write_timeout", ErrorCategory::Unavailable, true),
1569 ("writes_unavailable", ErrorCategory::Unavailable, true),
1570 ("configuration_transition", ErrorCategory::Unavailable, true),
1571 ("contention", ErrorCategory::Unavailable, true),
1572 ("winner_limit_exceeded", ErrorCategory::Unavailable, true),
1573 ("leader_unavailable", ErrorCategory::Unavailable, true),
1574 ("snapshot_required", ErrorCategory::Unavailable, false),
1575 ("resource_exhausted", ErrorCategory::ResourceExhausted, true),
1576 ("overloaded", ErrorCategory::ResourceExhausted, true),
1577 ("payload_too_large", ErrorCategory::ResourceExhausted, false),
1578 ("data_root_locked", ErrorCategory::Internal, false),
1579 ("unsupported_ack_mode", ErrorCategory::Internal, false),
1580 ("execution_profile_mismatch", ErrorCategory::Internal, false),
1581 ("storage_error", ErrorCategory::Internal, false),
1582 ("reconciliation_error", ErrorCategory::Internal, false),
1583 ("invariant_violation", ErrorCategory::Internal, false),
1584 ("fatal", ErrorCategory::Internal, false),
1585 ("task_failed", ErrorCategory::Internal, false),
1586 ("future_code", ErrorCategory::Unknown, true),
1587 ];
1588
1589 for (code, category, retryable) in cases {
1590 let classification = ErrorClassification::from_server_code(code, retryable);
1591
1592 assert_eq!(classification.code(), code);
1593 assert_eq!(classification.category(), category);
1594 assert_eq!(classification.retryable(), retryable);
1595 }
1596 }
1597
1598 #[test]
1599 fn execution_profile_has_stable_text_and_wire_ids() {
1600 assert_eq!(ExecutionProfile::Sqlite.as_str(), "sql");
1601 assert_eq!(ExecutionProfile::Graph.as_str(), "graph");
1602 assert_eq!(ExecutionProfile::Kv.as_str(), "kv");
1603 assert_eq!(ExecutionProfile::Sqlite.to_string(), "sql");
1604 assert_eq!(ExecutionProfile::Graph.to_string(), "graph");
1605 assert_eq!(ExecutionProfile::Kv.to_string(), "kv");
1606 assert_eq!("sql".parse(), Ok(ExecutionProfile::Sqlite));
1607 assert_eq!("graph".parse(), Ok(ExecutionProfile::Graph));
1608 assert_eq!("kv".parse(), Ok(ExecutionProfile::Kv));
1609 assert_eq!(
1610 "sqlite".parse::<ExecutionProfile>(),
1611 Err(ExecutionProfileParseError)
1612 );
1613 assert_eq!(
1614 serde_json::to_value(ExecutionProfile::Sqlite).unwrap(),
1615 serde_json::json!("sql")
1616 );
1617 assert_eq!(
1618 serde_json::from_value::<ExecutionProfile>(serde_json::json!("sql")).unwrap(),
1619 ExecutionProfile::Sqlite
1620 );
1621 assert!(serde_json::from_value::<ExecutionProfile>(serde_json::json!("sqlite")).is_err());
1622 assert_eq!(
1623 "ladybug".parse::<ExecutionProfile>(),
1624 Err(ExecutionProfileParseError)
1625 );
1626 assert_eq!(ExecutionProfile::Sqlite.wire_id(), 1);
1627 assert_eq!(ExecutionProfile::Graph.wire_id(), 2);
1628 assert_eq!(ExecutionProfile::Kv.wire_id(), 3);
1629 assert_eq!(
1630 ExecutionProfile::from_wire_id(1),
1631 Some(ExecutionProfile::Sqlite)
1632 );
1633 assert_eq!(
1634 ExecutionProfile::from_wire_id(2),
1635 Some(ExecutionProfile::Graph)
1636 );
1637 assert_eq!(
1638 ExecutionProfile::from_wire_id(3),
1639 Some(ExecutionProfile::Kv)
1640 );
1641 assert_eq!(ExecutionProfile::from_wire_id(0), None);
1642 }
1643
1644 #[test]
1645 fn replicated_command_envelope_uses_a_stable_explicit_codec() {
1646 let envelope = ReplicatedCommandEnvelope::new(
1647 ExecutionProfile::Graph,
1648 3,
1649 "req-1",
1650 vec![0xaa, 0xbb, 0xcc],
1651 )
1652 .unwrap();
1653 let encoded = envelope.encode().unwrap();
1654
1655 assert_eq!(
1656 encoded,
1657 [
1658 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'-',
1659 b'1', 0xaa, 0xbb, 0xcc,
1660 ]
1661 );
1662 assert_eq!(ReplicatedCommandEnvelope::decode(&encoded), Ok(envelope));
1663 }
1664
1665 #[test]
1666 fn replicated_command_envelope_preserves_backend_bytes_unchanged() {
1667 let legacy_qsql = b"QSQL\0\x02{\"request_id\":\"same\"}".to_vec();
1668 let envelope = ReplicatedCommandEnvelope::new(
1669 ExecutionProfile::Sqlite,
1670 1,
1671 "same",
1672 legacy_qsql.clone(),
1673 )
1674 .unwrap();
1675
1676 let decoded = ReplicatedCommandEnvelope::decode(&envelope.encode().unwrap()).unwrap();
1677 assert_eq!(decoded.body(), legacy_qsql);
1678 }
1679
1680 #[test]
1681 fn replicated_command_envelope_rejects_noncanonical_or_invalid_inputs() {
1682 assert_eq!(
1683 ReplicatedCommandEnvelope::new(ExecutionProfile::Graph, 0, "req", Vec::new()),
1684 Err(CommandEnvelopeError::InvalidCommandVersion)
1685 );
1686 assert_eq!(
1687 ReplicatedCommandEnvelope::new(ExecutionProfile::Graph, 1, "", Vec::new()),
1688 Err(CommandEnvelopeError::EmptyRequestId)
1689 );
1690
1691 let valid =
1692 ReplicatedCommandEnvelope::new(ExecutionProfile::Graph, 1, "req", vec![1, 2, 3])
1693 .unwrap()
1694 .encode()
1695 .unwrap();
1696 let mut unknown_profile = valid.clone();
1697 unknown_profile[6] = 9;
1698 assert_eq!(
1699 ReplicatedCommandEnvelope::decode(&unknown_profile),
1700 Err(CommandEnvelopeError::UnknownExecutionProfile(9))
1701 );
1702
1703 let mut trailing = valid;
1704 trailing.push(0);
1705 assert_eq!(
1706 ReplicatedCommandEnvelope::decode(&trailing),
1707 Err(CommandEnvelopeError::TrailingBytes)
1708 );
1709 }
1710
1711 #[test]
1712 fn replicated_command_envelope_rejects_malformed_wire_boundaries() {
1713 let valid =
1714 ReplicatedCommandEnvelope::new(ExecutionProfile::Graph, 1, "req", vec![1, 2, 3])
1715 .unwrap()
1716 .encode()
1717 .unwrap();
1718
1719 let mut invalid_magic = valid.clone();
1720 invalid_magic[0] = b'X';
1721 assert_eq!(
1722 ReplicatedCommandEnvelope::decode(&invalid_magic),
1723 Err(CommandEnvelopeError::InvalidMagic)
1724 );
1725
1726 let mut unsupported_version = valid.clone();
1727 unsupported_version[4..6].copy_from_slice(&2_u16.to_be_bytes());
1728 assert_eq!(
1729 ReplicatedCommandEnvelope::decode(&unsupported_version),
1730 Err(CommandEnvelopeError::UnsupportedFormatVersion(2))
1731 );
1732
1733 assert_eq!(
1734 ReplicatedCommandEnvelope::decode(&valid[..14]),
1735 Err(CommandEnvelopeError::Truncated)
1736 );
1737
1738 let mut truncated_body = valid.clone();
1739 truncated_body[11..15].copy_from_slice(&4_u32.to_be_bytes());
1740 assert_eq!(
1741 ReplicatedCommandEnvelope::decode(&truncated_body),
1742 Err(CommandEnvelopeError::Truncated)
1743 );
1744
1745 let mut invalid_request_utf8 = valid.clone();
1746 invalid_request_utf8[15] = 0xff;
1747 assert_eq!(
1748 ReplicatedCommandEnvelope::decode(&invalid_request_utf8),
1749 Err(CommandEnvelopeError::InvalidRequestIdUtf8)
1750 );
1751
1752 let mut oversized_request = valid.clone();
1753 oversized_request[9..11].copy_from_slice(&4_u16.to_be_bytes());
1754 assert_eq!(
1755 ReplicatedCommandEnvelope::decode(&oversized_request),
1756 Err(CommandEnvelopeError::Truncated)
1757 );
1758
1759 let mut undersized_body = valid.clone();
1760 undersized_body[11..15].copy_from_slice(&2_u32.to_be_bytes());
1761 assert_eq!(
1762 ReplicatedCommandEnvelope::decode(&undersized_body),
1763 Err(CommandEnvelopeError::TrailingBytes)
1764 );
1765 }
1766
1767 #[test]
1768 fn replicated_command_decoder_never_panics_and_successes_are_canonical() {
1769 let mut state = 0x9e37_79b9_7f4a_7c15_u64;
1770 for length in 0..=256 {
1771 for _ in 0..8 {
1772 let mut bytes = Vec::with_capacity(length);
1773 for _ in 0..length {
1774 state = state
1775 .wrapping_mul(6_364_136_223_846_793_005)
1776 .wrapping_add(1_442_695_040_888_963_407);
1777 bytes.push((state >> 32) as u8);
1778 }
1779 let decoded =
1780 std::panic::catch_unwind(|| ReplicatedCommandEnvelope::decode(&bytes));
1781 let decoded = decoded.expect("decoder must not panic for arbitrary bytes");
1782 if let Ok(envelope) = decoded {
1783 assert_eq!(envelope.encode().unwrap(), bytes);
1784 }
1785 }
1786 }
1787
1788 for sequence in 1_u16..=256 {
1789 state = state
1790 .wrapping_mul(6_364_136_223_846_793_005)
1791 .wrapping_add(1_442_695_040_888_963_407);
1792 let body = state.to_be_bytes()[..usize::from(sequence % 8)].to_vec();
1793 let profile = match sequence % 3 {
1794 0 => ExecutionProfile::Sqlite,
1795 1 => ExecutionProfile::Graph,
1796 _ => ExecutionProfile::Kv,
1797 };
1798 let envelope = ReplicatedCommandEnvelope::new(
1799 profile,
1800 sequence,
1801 format!("request-{sequence}"),
1802 body,
1803 )
1804 .unwrap();
1805 let encoded = envelope.encode().unwrap();
1806 let decoded = ReplicatedCommandEnvelope::decode(&encoded).unwrap();
1807 assert_eq!(decoded.encode().unwrap(), encoded);
1808 }
1809 }
1810
1811 #[test]
1812 fn config_decoder_rejects_overflowing_offsets_without_panicking() {
1813 let bytes = [];
1814 assert_eq!(
1815 read_config_u16(&bytes, usize::MAX),
1816 Err(ConfigChangeDecodeError)
1817 );
1818 assert_eq!(
1819 read_config_u64(&bytes, usize::MAX),
1820 Err(ConfigChangeDecodeError)
1821 );
1822 assert_eq!(
1823 read_config_hash(&bytes, usize::MAX),
1824 Err(ConfigChangeDecodeError)
1825 );
1826
1827 let mut cursor = usize::MAX;
1828 assert_eq!(
1829 read_config_string(&bytes, &mut cursor),
1830 Err(ConfigChangeDecodeError)
1831 );
1832 assert_eq!(
1833 read_config_u64_at(&bytes, &mut cursor),
1834 Err(ConfigChangeDecodeError)
1835 );
1836 assert_eq!(
1837 read_config_hash_at(&bytes, &mut cursor),
1838 Err(ConfigChangeDecodeError)
1839 );
1840 }
1841}