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