1use std::error::Error;
15use std::fmt;
16use std::marker::PhantomData;
17use std::num::NonZeroU32;
18use std::sync::Arc;
19
20use serde_json::Value;
21use sha2::{Digest, Sha256};
22
23use crate::definition::ComponentStreamIdentity;
24use crate::state::{json_depth, upgrade_schema_chain};
25use crate::{
26 DurableStateKind, StateCodecError, StateError, StateLimits, StateSchemaId, StateSchemaVersion,
27 VersionedStateCodec,
28};
29
30const MAX_TOKEN_BYTES: usize = 128;
31const MAX_UPGRADE_CHAIN: usize = 64;
36
37#[derive(Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
45pub struct CodecId(String);
46
47impl CodecId {
48 pub fn new(value: impl Into<String>) -> Result<Self, ComponentStateError> {
56 let value = value.into();
57 if value.is_empty() {
58 return Err(ComponentStateError::EmptyCodecId);
59 }
60 if value.len() > MAX_TOKEN_BYTES {
61 return Err(ComponentStateError::CodecIdTooLong {
62 max_bytes: MAX_TOKEN_BYTES,
63 });
64 }
65 if value.trim() != value || value.chars().any(char::is_control) {
66 return Err(ComponentStateError::Malformed);
67 }
68 Ok(Self(value))
69 }
70
71 #[must_use]
73 pub fn as_str(&self) -> &str {
74 &self.0
75 }
76}
77
78impl fmt::Debug for CodecId {
79 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
80 formatter.debug_tuple("CodecId").field(&self.0).finish()
84 }
85}
86
87impl fmt::Display for CodecId {
88 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
89 formatter.write_str(self.as_str())
90 }
91}
92
93#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
95pub struct CodecVersion(NonZeroU32);
96
97impl CodecVersion {
98 pub fn new(value: u32) -> Result<Self, ComponentStateError> {
104 NonZeroU32::new(value)
105 .map(Self)
106 .ok_or(ComponentStateError::ZeroCodecVersion)
107 }
108
109 #[must_use]
111 pub const fn get(self) -> u32 {
112 self.0.get()
113 }
114}
115
116#[derive(Clone, Copy)]
122pub struct CodecVersionUpgrade {
123 from: CodecVersion,
124 to: CodecVersion,
125 apply: fn(&[u8]) -> Result<Vec<u8>, StateCodecError>,
126}
127
128impl CodecVersionUpgrade {
129 pub fn new(
136 from: CodecVersion,
137 to: CodecVersion,
138 apply: fn(&[u8]) -> Result<Vec<u8>, StateCodecError>,
139 ) -> Result<Self, ComponentStateError> {
140 if to <= from {
141 return Err(ComponentStateError::NonIncreasingCodecUpgrade);
142 }
143 Ok(Self { from, to, apply })
144 }
145
146 #[must_use]
148 pub const fn from(&self) -> CodecVersion {
149 self.from
150 }
151
152 #[must_use]
154 pub const fn to(&self) -> CodecVersion {
155 self.to
156 }
157}
158
159impl fmt::Debug for CodecVersionUpgrade {
160 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
161 formatter
162 .debug_struct("CodecVersionUpgrade")
163 .field("from", &self.from)
164 .field("to", &self.to)
165 .finish_non_exhaustive()
166 }
167}
168
169#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
175#[non_exhaustive]
176pub enum ChecksumAlgorithm {
177 Sha256,
179}
180
181impl ChecksumAlgorithm {
182 pub const CURRENT: Self = Self::Sha256;
184
185 const fn identifier(self) -> u16 {
186 match self {
187 Self::Sha256 => 1,
188 }
189 }
190
191 const fn algorithm_version(self) -> u16 {
192 match self {
193 Self::Sha256 => 1,
194 }
195 }
196
197 const fn resolve(identifier: u16, version: u16) -> Option<Self> {
198 match (identifier, version) {
199 (1, 1) => Some(Self::Sha256),
200 _ => None,
201 }
202 }
203
204 fn digest(self, input: &[u8]) -> [u8; 32] {
205 match self {
206 Self::Sha256 => Sha256::digest(input).into(),
207 }
208 }
209}
210
211#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
218#[non_exhaustive]
219pub enum StateSensitivity {
220 #[default]
222 Sensitive,
223 NonSensitive,
226}
227
228#[derive(Clone, Copy, Debug, Eq, PartialEq)]
234#[non_exhaustive]
235pub enum RestartabilityDeclaration {
236 Restartable,
239 NotRestartable,
242}
243
244#[derive(Clone, Copy, Eq, Hash, PartialEq)]
249pub struct ContentIdentity([u8; 32]);
250
251impl ContentIdentity {
252 #[must_use]
254 pub const fn from_bytes(bytes: [u8; 32]) -> Self {
255 Self(bytes)
256 }
257
258 #[must_use]
260 pub fn of(bytes: &[u8]) -> Self {
261 Self(Sha256::digest(bytes).into())
262 }
263
264 #[must_use]
266 pub const fn as_bytes(&self) -> &[u8; 32] {
267 &self.0
268 }
269}
270
271impl fmt::Debug for ContentIdentity {
272 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
273 write!(formatter, "ContentIdentity({})", hex_encode(&self.0))
274 }
275}
276
277fn hex_encode(bytes: &[u8]) -> String {
278 use fmt::Write as _;
279 let mut out = String::with_capacity(bytes.len() * 2);
280 for byte in bytes {
281 let _ = write!(out, "{byte:02x}");
282 }
283 out
284}
285
286#[derive(Clone, Copy, Debug, Eq, PartialEq)]
293pub struct ExternalStateReference {
294 content_id: ContentIdentity,
295 encoded_len: u64,
296}
297
298impl ExternalStateReference {
299 #[must_use]
301 pub const fn new(content_id: ContentIdentity, encoded_len: u64) -> Self {
302 Self {
303 content_id,
304 encoded_len,
305 }
306 }
307
308 #[must_use]
310 pub const fn content_id(&self) -> ContentIdentity {
311 self.content_id
312 }
313
314 #[must_use]
316 pub const fn encoded_len(&self) -> u64 {
317 self.encoded_len
318 }
319
320 pub fn verify(&self, bytes: &[u8]) -> Result<(), ComponentStateError> {
334 if ContentIdentity::of(bytes) == self.content_id {
335 Ok(())
336 } else {
337 Err(ComponentStateError::ExternalReferenceContentMismatch)
338 }
339 }
340}
341
342pub trait ExternalStateStore: Send + Sync {
349 fn resolve<'a>(
358 &'a self,
359 reference: &'a ExternalStateReference,
360 ) -> impl Future<Output = Result<Vec<u8>, ExternalStateError>> + Send + 'a;
361
362 fn store<'a>(
369 &'a self,
370 bytes: &'a [u8],
371 ) -> impl Future<Output = Result<ExternalStateReference, ExternalStateError>> + Send + 'a;
372}
373
374#[derive(Clone, Copy, Debug, Eq, PartialEq)]
376#[non_exhaustive]
377pub enum ExternalStateError {
378 ContentMissing,
380 ContentMismatch,
382 StoreFailed,
384}
385
386impl fmt::Display for ExternalStateError {
387 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
388 formatter.write_str(match self {
389 Self::ContentMissing => "external component state content is missing",
390 Self::ContentMismatch => "external component state content identity mismatch",
391 Self::StoreFailed => "external component state store failed",
392 })
393 }
394}
395
396impl Error for ExternalStateError {}
397
398#[derive(Clone, Debug, Eq, PartialEq)]
404pub enum ComponentStatePayload {
405 Inline(Vec<u8>),
407 External(ExternalStateReference),
409}
410
411#[derive(Clone, Eq, PartialEq)]
412enum StoredPayload {
413 Inline(Vec<u8>),
424 External(ExternalStateReference),
425}
426
427#[derive(Clone, Eq, PartialEq)]
433pub struct ComponentStateEnvelope {
434 namespace: ComponentStreamIdentity,
435 schema_id: StateSchemaId,
436 schema_version: StateSchemaVersion,
437 codec_id: CodecId,
438 codec_version: CodecVersion,
439 checksum_algorithm: ChecksumAlgorithm,
440 checksum: [u8; 32],
441 payload: StoredPayload,
442 encoded_len: usize,
443}
444
445fn canonical_checksum_input(
446 namespace: &ComponentStreamIdentity,
447 schema_id: &StateSchemaId,
448 schema_version: StateSchemaVersion,
449 codec_id: &CodecId,
450 codec_version: CodecVersion,
451 payload_marker: &[u8],
452) -> Vec<u8> {
453 let mut input = Vec::with_capacity(64 + payload_marker.len());
454 input.extend_from_slice(b"oxide-batch.component-state\0");
455 input.extend_from_slice(namespace.as_str().as_bytes());
456 input.push(0);
457 input.extend_from_slice(schema_id.as_str().as_bytes());
458 input.push(0);
459 input.extend_from_slice(&schema_version.get().to_be_bytes());
460 input.extend_from_slice(codec_id.as_str().as_bytes());
461 input.push(0);
462 input.extend_from_slice(&codec_version.get().to_be_bytes());
463 input.extend_from_slice(payload_marker);
464 input
465}
466
467fn inline_payload_marker(bytes: &[u8]) -> Vec<u8> {
468 let mut marker = Vec::with_capacity(1 + bytes.len());
469 marker.push(0);
470 marker.extend_from_slice(bytes);
471 marker
472}
473
474fn external_payload_marker(reference: &ExternalStateReference) -> Vec<u8> {
475 let mut marker = Vec::with_capacity(1 + 32 + 8);
476 marker.push(1);
477 marker.extend_from_slice(reference.content_id().as_bytes());
478 marker.extend_from_slice(&reference.encoded_len().to_be_bytes());
479 marker
480}
481
482fn check_upgraded_component_state(payload: &[u8]) -> Result<(), ComponentStateError> {
490 const HARD_MAXIMUM_BYTES: usize = 1024 * 1024;
491 const HARD_MAXIMUM_DEPTH: usize = 64;
492 if payload.len() > HARD_MAXIMUM_BYTES {
493 return Err(ComponentStateError::TooLarge {
494 max_bytes: HARD_MAXIMUM_BYTES,
495 });
496 }
497 let value: Value =
498 serde_json::from_slice(payload).map_err(|_| ComponentStateError::PayloadNotObject)?;
499 if !value.is_object() {
500 return Err(ComponentStateError::PayloadNotObject);
501 }
502 if json_depth(&value) > HARD_MAXIMUM_DEPTH {
503 return Err(ComponentStateError::TooDeep {
504 max_depth: HARD_MAXIMUM_DEPTH,
505 });
506 }
507 Ok(())
508}
509
510fn upgrade_codec_chain<T>(
511 recorded: CodecVersion,
512 codec: &(impl ComponentStateCodec<T> + ?Sized),
513 mut payload: Vec<u8>,
514) -> Result<Vec<u8>, ComponentStateError> {
515 let current = codec.codec_version();
516 let upgrades = codec.codec_upgrades();
517 let mut version = recorded;
518 let mut applied = 0_usize;
519 while version < current {
520 let mut edges = upgrades.iter().filter(|upgrade| upgrade.from == version);
521 let edge = edges
522 .next()
523 .ok_or(ComponentStateError::NoCodecUpgradePath {
524 found: version.get(),
525 current: current.get(),
526 })?;
527 if edges.next().is_some() {
528 return Err(ComponentStateError::AmbiguousCodecUpgrade {
529 from: version.get(),
530 });
531 }
532 if edge.to > current {
533 return Err(ComponentStateError::CodecUpgradeOvershootsCurrent {
534 to: edge.to.get(),
535 current: current.get(),
536 });
537 }
538 applied += 1;
539 if applied > MAX_UPGRADE_CHAIN {
540 return Err(ComponentStateError::CodecUpgradeChainTooLong {
541 max_upgrades: MAX_UPGRADE_CHAIN,
542 });
543 }
544 payload = (edge.apply)(&payload).map_err(ComponentStateError::Codec)?;
545 check_upgraded_component_state(&payload)?;
546 version = edge.to;
547 }
548 Ok(payload)
549}
550
551fn map_schema_chain_error(error: &StateError) -> ComponentStateError {
552 match error {
553 StateError::SchemaMismatch { .. } => ComponentStateError::SchemaMismatch,
554 StateError::UnsupportedSchemaVersion { found, current, .. } => {
555 ComponentStateError::UnsupportedSchemaVersion {
556 found: found.get(),
557 current: current.get(),
558 }
559 }
560 StateError::NoUpgradePath { found, current, .. } => {
561 ComponentStateError::NoSchemaUpgradePath {
562 found: found.get(),
563 current: current.get(),
564 }
565 }
566 StateError::AmbiguousUpgrade { from, .. } => {
567 ComponentStateError::AmbiguousSchemaUpgrade { from: from.get() }
568 }
569 StateError::UpgradeOvershootsCurrent { to, current, .. } => {
570 ComponentStateError::SchemaUpgradeOvershootsCurrent {
571 to: to.get(),
572 current: current.get(),
573 }
574 }
575 StateError::UpgradeChainTooLong { max_upgrades, .. } => {
576 ComponentStateError::SchemaUpgradeChainTooLong {
577 max_upgrades: *max_upgrades,
578 }
579 }
580 StateError::TooLarge { max_bytes, .. } => ComponentStateError::TooLarge {
581 max_bytes: *max_bytes,
582 },
583 StateError::TooDeep { max_depth, .. } => ComponentStateError::TooDeep {
584 max_depth: *max_depth,
585 },
586 StateError::Codec(inner) => ComponentStateError::Codec(*inner),
587 _ => ComponentStateError::Malformed,
588 }
589}
590
591impl ComponentStateEnvelope {
592 pub fn encode<T>(
603 namespace: ComponentStreamIdentity,
604 value: &T,
605 codec: &(impl ComponentStateCodec<T> + ?Sized),
606 limits: StateLimits,
607 ) -> Result<Self, ComponentStateError> {
608 let payload_bytes = codec.encode(value).map_err(ComponentStateError::Codec)?;
609 if payload_bytes.len() > limits.maximum_bytes() {
610 return Err(ComponentStateError::TooLarge {
611 max_bytes: limits.maximum_bytes(),
612 });
613 }
614 let schema_id = codec.schema_id().clone();
615 let schema_version = codec.current_version();
616 let codec_id = codec.codec_id().clone();
617 let codec_version = codec.codec_version();
618 let checksum_algorithm = ChecksumAlgorithm::CURRENT;
619 let checksum = checksum_algorithm.digest(&canonical_checksum_input(
620 &namespace,
621 &schema_id,
622 schema_version,
623 &codec_id,
624 codec_version,
625 &inline_payload_marker(&payload_bytes),
626 ));
627 let payload_value: Value = serde_json::from_slice(&payload_bytes)
628 .map_err(|_| ComponentStateError::InvalidPayload)?;
629 if !payload_value.is_object() {
630 return Err(ComponentStateError::PayloadNotObject);
631 }
632 if json_depth(&payload_value) > limits.maximum_depth() {
633 return Err(ComponentStateError::TooDeep {
634 max_depth: limits.maximum_depth(),
635 });
636 }
637 let encoded_len = payload_bytes.len();
638 Ok(Self {
639 namespace,
640 schema_id,
641 schema_version,
642 codec_id,
643 codec_version,
644 checksum_algorithm,
645 checksum,
646 payload: StoredPayload::Inline(payload_bytes),
647 encoded_len,
648 })
649 }
650
651 #[must_use]
657 pub fn external(
658 namespace: ComponentStreamIdentity,
659 schema_id: StateSchemaId,
660 schema_version: StateSchemaVersion,
661 codec_id: CodecId,
662 codec_version: CodecVersion,
663 reference: ExternalStateReference,
664 ) -> Self {
665 let checksum_algorithm = ChecksumAlgorithm::CURRENT;
666 let checksum = checksum_algorithm.digest(&canonical_checksum_input(
667 &namespace,
668 &schema_id,
669 schema_version,
670 &codec_id,
671 codec_version,
672 &external_payload_marker(&reference),
673 ));
674 Self {
675 namespace,
676 schema_id,
677 schema_version,
678 codec_id,
679 codec_version,
680 checksum_algorithm,
681 checksum,
682 payload: StoredPayload::External(reference),
683 encoded_len: 0,
684 }
685 }
686
687 #[allow(clippy::too_many_arguments)]
698 pub fn from_durable(
699 namespace: ComponentStreamIdentity,
700 schema_id: &str,
701 schema_version: u32,
702 codec_id: &str,
703 codec_version: u32,
704 checksum_algorithm: u16,
705 checksum_algorithm_version: u16,
706 checksum: [u8; 32],
707 payload: ComponentStatePayload,
708 limits: StateLimits,
709 ) -> Result<Self, ComponentStateError> {
710 let schema_id =
711 StateSchemaId::new(schema_id).map_err(|_| ComponentStateError::Malformed)?;
712 let schema_version =
713 StateSchemaVersion::new(schema_version).map_err(|_| ComponentStateError::Malformed)?;
714 let codec_id = CodecId::new(codec_id)?;
715 let codec_version = CodecVersion::new(codec_version)?;
716 let algorithm = ChecksumAlgorithm::resolve(checksum_algorithm, checksum_algorithm_version)
717 .ok_or(ComponentStateError::ChecksumAlgorithmUnsupported {
718 algorithm: checksum_algorithm,
719 version: checksum_algorithm_version,
720 })?;
721
722 let (marker, size_check) = match &payload {
723 ComponentStatePayload::Inline(bytes) => {
724 (inline_payload_marker(bytes), Some(bytes.len()))
725 }
726 ComponentStatePayload::External(reference) => {
727 (external_payload_marker(reference), None)
728 }
729 };
730 if let Some(len) = size_check
731 && len > limits.maximum_bytes()
732 {
733 return Err(ComponentStateError::TooLarge {
734 max_bytes: limits.maximum_bytes(),
735 });
736 }
737
738 let expected = algorithm.digest(&canonical_checksum_input(
741 &namespace,
742 &schema_id,
743 schema_version,
744 &codec_id,
745 codec_version,
746 &marker,
747 ));
748 if expected != checksum {
749 return Err(ComponentStateError::ChecksumMismatch);
750 }
751
752 let (stored, encoded_len) = match payload {
753 ComponentStatePayload::Inline(bytes) => {
754 let value: Value =
755 serde_json::from_slice(&bytes).map_err(|_| ComponentStateError::Malformed)?;
756 if !value.is_object() {
757 return Err(ComponentStateError::PayloadNotObject);
758 }
759 if json_depth(&value) > limits.maximum_depth() {
760 return Err(ComponentStateError::TooDeep {
761 max_depth: limits.maximum_depth(),
762 });
763 }
764 let encoded_len = bytes.len();
765 (StoredPayload::Inline(bytes), encoded_len)
766 }
767 ComponentStatePayload::External(reference) => (StoredPayload::External(reference), 0),
768 };
769
770 Ok(Self {
771 namespace,
772 schema_id,
773 schema_version,
774 codec_id,
775 codec_version,
776 checksum_algorithm: algorithm,
777 checksum,
778 payload: stored,
779 encoded_len,
780 })
781 }
782
783 pub fn decode<T>(
798 &self,
799 codec: &(impl ComponentStateCodec<T> + ?Sized),
800 ) -> Result<T, ComponentStateError> {
801 if self.codec_id != *codec.codec_id() {
802 return Err(ComponentStateError::UnknownCodec);
803 }
804 if self.schema_id != *codec.schema_id() {
805 return Err(ComponentStateError::SchemaMismatch);
806 }
807 let current_schema = codec.current_version();
808 if self.schema_version > current_schema {
809 return Err(ComponentStateError::UnsupportedSchemaVersion {
810 found: self.schema_version.get(),
811 current: current_schema.get(),
812 });
813 }
814 let current_codec = codec.codec_version();
815 if self.codec_version > current_codec {
816 return Err(ComponentStateError::UnsupportedCodecVersion {
817 found: self.codec_version.get(),
818 current: current_codec.get(),
819 });
820 }
821 let Self {
822 payload: StoredPayload::Inline(bytes),
823 ..
824 } = self
825 else {
826 return Err(ComponentStateError::ExternalReferenceMissing);
827 };
828 let payload_bytes = bytes.clone();
829 let payload_bytes = upgrade_schema_chain(
830 DurableStateKind::ComponentState,
831 self.schema_version,
832 codec,
833 payload_bytes,
834 )
835 .map_err(|error| map_schema_chain_error(&error))?;
836 let payload_bytes = upgrade_codec_chain(self.codec_version, codec, payload_bytes)?;
837 codec
838 .decode(&payload_bytes)
839 .map_err(ComponentStateError::Codec)
840 }
841
842 pub fn validated_for_open<T>(
861 &self,
862 codec: &(impl ComponentStateCodec<T> + ?Sized),
863 ) -> Result<Self, ComponentStateError> {
864 if self.codec_id != *codec.codec_id() {
865 return Err(ComponentStateError::UnknownCodec);
866 }
867 if self.schema_id != *codec.schema_id() {
868 return Err(ComponentStateError::SchemaMismatch);
869 }
870 let current_schema = codec.current_version();
871 if self.schema_version > current_schema {
872 return Err(ComponentStateError::UnsupportedSchemaVersion {
873 found: self.schema_version.get(),
874 current: current_schema.get(),
875 });
876 }
877 let current_codec = codec.codec_version();
878 if self.codec_version > current_codec {
879 return Err(ComponentStateError::UnsupportedCodecVersion {
880 found: self.codec_version.get(),
881 current: current_codec.get(),
882 });
883 }
884 if matches!(self.payload, StoredPayload::External(_)) {
885 if self.schema_version < current_schema {
886 return Err(ComponentStateError::ExternalSchemaMigrationUnsupported {
887 found: self.schema_version.get(),
888 current: current_schema.get(),
889 });
890 }
891 if self.codec_version < current_codec {
892 return Err(ComponentStateError::ExternalCodecMigrationUnsupported {
893 found: self.codec_version.get(),
894 current: current_codec.get(),
895 });
896 }
897 return Ok(self.clone());
898 }
899 let Self {
900 payload: StoredPayload::Inline(bytes),
901 ..
902 } = self
903 else {
904 return Ok(self.clone());
905 };
906 let migrated = upgrade_schema_chain(
907 DurableStateKind::ComponentState,
908 self.schema_version,
909 codec,
910 bytes.clone(),
911 )
912 .map_err(|error| map_schema_chain_error(&error))?;
913 let migrated = upgrade_codec_chain(self.codec_version, codec, migrated)?;
914 if migrated == *bytes {
915 return Ok(self.clone());
916 }
917 let checksum_algorithm = ChecksumAlgorithm::CURRENT;
918 let checksum = checksum_algorithm.digest(&canonical_checksum_input(
919 &self.namespace,
920 &self.schema_id,
921 current_schema,
922 &self.codec_id,
923 current_codec,
924 &inline_payload_marker(&migrated),
925 ));
926 let encoded_len = migrated.len();
927 Ok(Self {
928 namespace: self.namespace.clone(),
929 schema_id: self.schema_id.clone(),
930 schema_version: current_schema,
931 codec_id: self.codec_id.clone(),
932 codec_version: current_codec,
933 checksum_algorithm,
934 checksum,
935 payload: StoredPayload::Inline(migrated),
936 encoded_len,
937 })
938 }
939
940 #[must_use]
942 pub const fn namespace(&self) -> &ComponentStreamIdentity {
943 &self.namespace
944 }
945
946 #[must_use]
948 pub const fn schema_id(&self) -> &StateSchemaId {
949 &self.schema_id
950 }
951
952 #[must_use]
954 pub const fn schema_version(&self) -> StateSchemaVersion {
955 self.schema_version
956 }
957
958 #[must_use]
960 pub const fn codec_id(&self) -> &CodecId {
961 &self.codec_id
962 }
963
964 #[must_use]
966 pub const fn codec_version(&self) -> CodecVersion {
967 self.codec_version
968 }
969
970 #[must_use]
972 pub const fn is_external(&self) -> bool {
973 matches!(self.payload, StoredPayload::External(_))
974 }
975
976 #[must_use]
978 pub const fn checksum_algorithm(&self) -> u16 {
979 self.checksum_algorithm.identifier()
980 }
981
982 #[must_use]
984 pub const fn checksum_algorithm_version(&self) -> u16 {
985 self.checksum_algorithm.algorithm_version()
986 }
987
988 #[must_use]
995 pub const fn checksum(&self) -> [u8; 32] {
996 self.checksum
997 }
998
999 pub fn payload(&self) -> Result<ComponentStatePayload, ComponentStateError> {
1008 match &self.payload {
1009 StoredPayload::Inline(bytes) => Ok(ComponentStatePayload::Inline(bytes.clone())),
1010 StoredPayload::External(reference) => Ok(ComponentStatePayload::External(*reference)),
1011 }
1012 }
1013
1014 #[must_use]
1017 pub fn encoded_len(&self) -> usize {
1018 match &self.payload {
1019 StoredPayload::External(reference) => {
1020 usize::try_from(reference.encoded_len()).unwrap_or(usize::MAX)
1021 }
1022 StoredPayload::Inline(_) => self.encoded_len,
1023 }
1024 }
1025}
1026
1027impl fmt::Debug for ComponentStateEnvelope {
1028 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1029 formatter
1030 .debug_struct("ComponentStateEnvelope")
1031 .field("namespace", &self.namespace)
1032 .field("schema_id", &self.schema_id)
1033 .field("schema_version", &self.schema_version)
1034 .field("codec_id", &self.codec_id)
1035 .field("codec_version", &self.codec_version)
1036 .field("checksum_algorithm", &self.checksum_algorithm)
1037 .field("checksum", &"<redacted>")
1038 .field("is_external", &self.is_external())
1039 .field("encoded_len", &self.encoded_len())
1040 .field("payload", &"<redacted>")
1041 .finish()
1042 }
1043}
1044
1045pub trait ComponentStateCodec<T>: VersionedStateCodec<T> {
1053 fn codec_id(&self) -> &CodecId;
1055
1056 fn codec_version(&self) -> CodecVersion;
1058
1059 fn codec_upgrades(&self) -> &[CodecVersionUpgrade] {
1061 &[]
1062 }
1063
1064 fn sensitivity(&self) -> StateSensitivity {
1068 StateSensitivity::Sensitive
1069 }
1070
1071 fn restartability(&self) -> RestartabilityDeclaration;
1073}
1074
1075pub struct DefaultComponentCodec<C> {
1078 schema: C,
1079 codec_id: CodecId,
1080 codec_version: CodecVersion,
1081 codec_upgrades: Vec<CodecVersionUpgrade>,
1082 sensitivity: StateSensitivity,
1083 restartability: RestartabilityDeclaration,
1084}
1085
1086impl<C> DefaultComponentCodec<C> {
1087 #[must_use]
1090 pub const fn new(
1091 schema: C,
1092 codec_id: CodecId,
1093 codec_version: CodecVersion,
1094 restartability: RestartabilityDeclaration,
1095 ) -> Self {
1096 Self {
1097 schema,
1098 codec_id,
1099 codec_version,
1100 codec_upgrades: Vec::new(),
1101 sensitivity: StateSensitivity::Sensitive,
1102 restartability,
1103 }
1104 }
1105
1106 #[must_use]
1108 pub fn with_codec_upgrades(mut self, upgrades: Vec<CodecVersionUpgrade>) -> Self {
1109 self.codec_upgrades = upgrades;
1110 self
1111 }
1112
1113 #[must_use]
1115 pub const fn with_sensitivity(mut self, sensitivity: StateSensitivity) -> Self {
1116 self.sensitivity = sensitivity;
1117 self
1118 }
1119}
1120
1121impl<T, C: VersionedStateCodec<T>> VersionedStateCodec<T> for DefaultComponentCodec<C> {
1122 fn schema_id(&self) -> &StateSchemaId {
1123 self.schema.schema_id()
1124 }
1125
1126 fn current_version(&self) -> StateSchemaVersion {
1127 self.schema.current_version()
1128 }
1129
1130 fn upgrades(&self) -> &[crate::StateSchemaUpgrade] {
1131 self.schema.upgrades()
1132 }
1133
1134 fn encode(&self, value: &T) -> Result<Vec<u8>, StateCodecError> {
1135 self.schema.encode(value)
1136 }
1137
1138 fn decode(&self, payload: &[u8]) -> Result<T, StateCodecError> {
1139 self.schema.decode(payload)
1140 }
1141}
1142
1143impl<T, C: VersionedStateCodec<T>> ComponentStateCodec<T> for DefaultComponentCodec<C> {
1144 fn codec_id(&self) -> &CodecId {
1145 &self.codec_id
1146 }
1147
1148 fn codec_version(&self) -> CodecVersion {
1149 self.codec_version
1150 }
1151
1152 fn codec_upgrades(&self) -> &[CodecVersionUpgrade] {
1153 &self.codec_upgrades
1154 }
1155
1156 fn sensitivity(&self) -> StateSensitivity {
1157 self.sensitivity
1158 }
1159
1160 fn restartability(&self) -> RestartabilityDeclaration {
1161 self.restartability
1162 }
1163}
1164
1165mod stream_contract_sealed {
1166 use super::{
1167 CodecId, ComponentStateEnvelope, ComponentStateError, RestartabilityDeclaration,
1168 StateSchemaId,
1169 };
1170
1171 pub trait StreamStateContractObject: Send + Sync {
1177 fn schema_id(&self) -> &StateSchemaId;
1178 fn codec_id(&self) -> &CodecId;
1179 fn restartability(&self) -> RestartabilityDeclaration;
1180 fn validate_for_open(
1181 &self,
1182 envelope: &ComponentStateEnvelope,
1183 ) -> Result<ComponentStateEnvelope, ComponentStateError>;
1184 }
1185}
1186
1187struct CodecContract<T, C> {
1188 codec: C,
1189 marker: PhantomData<fn() -> T>,
1193}
1194
1195impl<T, C: ComponentStateCodec<T> + Send + Sync> stream_contract_sealed::StreamStateContractObject
1196 for CodecContract<T, C>
1197{
1198 fn schema_id(&self) -> &StateSchemaId {
1199 self.codec.schema_id()
1200 }
1201
1202 fn codec_id(&self) -> &CodecId {
1203 self.codec.codec_id()
1204 }
1205
1206 fn restartability(&self) -> RestartabilityDeclaration {
1207 self.codec.restartability()
1208 }
1209
1210 fn validate_for_open(
1211 &self,
1212 envelope: &ComponentStateEnvelope,
1213 ) -> Result<ComponentStateEnvelope, ComponentStateError> {
1214 envelope.validated_for_open(&self.codec)
1215 }
1216}
1217
1218pub struct StreamStateContract(Arc<dyn stream_contract_sealed::StreamStateContractObject>);
1231
1232impl StreamStateContract {
1233 pub fn new<T, C>(codec: C) -> Self
1236 where
1237 C: ComponentStateCodec<T> + Send + Sync + 'static,
1238 T: 'static,
1239 {
1240 Self(Arc::new(CodecContract {
1241 codec,
1242 marker: PhantomData,
1243 }))
1244 }
1245
1246 #[must_use]
1248 pub fn schema_id(&self) -> &StateSchemaId {
1249 self.0.schema_id()
1250 }
1251
1252 #[must_use]
1254 pub fn codec_id(&self) -> &CodecId {
1255 self.0.codec_id()
1256 }
1257
1258 #[must_use]
1261 pub fn restartability(&self) -> RestartabilityDeclaration {
1262 self.0.restartability()
1263 }
1264
1265 pub fn validate_for_open(
1273 &self,
1274 envelope: &ComponentStateEnvelope,
1275 ) -> Result<ComponentStateEnvelope, ComponentStateError> {
1276 self.0.validate_for_open(envelope)
1277 }
1278}
1279
1280impl fmt::Debug for StreamStateContract {
1281 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1282 formatter
1283 .debug_struct("StreamStateContract")
1284 .field("schema_id", &self.schema_id())
1285 .field("codec_id", &self.codec_id())
1286 .field("restartability", &self.restartability())
1287 .finish()
1288 }
1289}
1290
1291#[derive(Clone, Debug, Eq, PartialEq)]
1296#[non_exhaustive]
1297pub enum ComponentStateError {
1298 EmptyCodecId,
1300 CodecIdTooLong {
1302 max_bytes: usize,
1304 },
1305 ZeroCodecVersion,
1307 NonIncreasingCodecUpgrade,
1310 TooLarge {
1312 max_bytes: usize,
1314 },
1315 TooDeep {
1317 max_depth: usize,
1319 },
1320 Malformed,
1322 SchemaMismatch,
1324 UnsupportedSchemaVersion {
1326 found: u32,
1328 current: u32,
1330 },
1331 NoSchemaUpgradePath {
1334 found: u32,
1336 current: u32,
1338 },
1339 AmbiguousSchemaUpgrade {
1341 from: u32,
1343 },
1344 SchemaUpgradeOvershootsCurrent {
1347 to: u32,
1349 current: u32,
1351 },
1352 SchemaUpgradeChainTooLong {
1354 max_upgrades: usize,
1356 },
1357 UnknownCodec,
1359 UnsupportedCodecVersion {
1361 found: u32,
1363 current: u32,
1365 },
1366 ExternalSchemaMigrationUnsupported {
1369 found: u32,
1371 current: u32,
1373 },
1374 ExternalCodecMigrationUnsupported {
1377 found: u32,
1379 current: u32,
1381 },
1382 NoCodecUpgradePath {
1385 found: u32,
1387 current: u32,
1389 },
1390 AmbiguousCodecUpgrade {
1392 from: u32,
1394 },
1395 CodecUpgradeOvershootsCurrent {
1398 to: u32,
1400 current: u32,
1402 },
1403 CodecUpgradeChainTooLong {
1405 max_upgrades: usize,
1407 },
1408 ChecksumAlgorithmUnsupported {
1410 algorithm: u16,
1412 version: u16,
1414 },
1415 ChecksumMismatch,
1419 ExternalReferenceMissing,
1421 ExternalReferenceContentMismatch,
1424 InvalidPayload,
1426 PayloadNotObject,
1428 Codec(StateCodecError),
1431}
1432
1433impl fmt::Display for ComponentStateError {
1434 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1435 match self {
1436 Self::EmptyCodecId => {
1437 formatter.write_str("component state codec identifier must not be empty")
1438 }
1439 Self::CodecIdTooLong { max_bytes } => {
1440 write!(
1441 formatter,
1442 "component state codec identifier exceeds {max_bytes} UTF-8 bytes"
1443 )
1444 }
1445 Self::ZeroCodecVersion => {
1446 formatter.write_str("component state codec version must be nonzero")
1447 }
1448 Self::NonIncreasingCodecUpgrade => {
1449 formatter.write_str("component state codec upgrade must increase the version")
1450 }
1451 Self::TooLarge { max_bytes } => {
1452 write!(formatter, "component state exceeds {max_bytes} bytes")
1453 }
1454 Self::TooDeep { max_depth } => {
1455 write!(formatter, "component state exceeds JSON depth {max_depth}")
1456 }
1457 Self::Malformed => formatter.write_str("component state is malformed"),
1458 Self::SchemaMismatch => {
1459 formatter.write_str("component state schema does not match the component")
1460 }
1461 Self::UnsupportedSchemaVersion { .. } => {
1462 formatter.write_str("component state schema version is unsupported")
1463 }
1464 Self::NoSchemaUpgradePath { .. } => {
1465 formatter.write_str("component state schema version has no upgrade path")
1466 }
1467 Self::AmbiguousSchemaUpgrade { .. } => {
1468 formatter.write_str("component state schema upgrade is ambiguous")
1469 }
1470 Self::SchemaUpgradeOvershootsCurrent { .. } => {
1471 formatter.write_str("component state schema upgrade passes the current version")
1472 }
1473 Self::SchemaUpgradeChainTooLong { max_upgrades } => {
1474 write!(
1475 formatter,
1476 "component state schema upgrade chain exceeds {max_upgrades} upgrades"
1477 )
1478 }
1479 Self::UnknownCodec => formatter.write_str("component state codec is unknown"),
1480 Self::UnsupportedCodecVersion { .. } => {
1481 formatter.write_str("component state codec version is unsupported")
1482 }
1483 Self::ExternalSchemaMigrationUnsupported { .. } => {
1484 formatter.write_str("external component state schema migration is unsupported")
1485 }
1486 Self::ExternalCodecMigrationUnsupported { .. } => {
1487 formatter.write_str("external component state codec migration is unsupported")
1488 }
1489 Self::NoCodecUpgradePath { .. } => {
1490 formatter.write_str("component state codec version has no upgrade path")
1491 }
1492 Self::AmbiguousCodecUpgrade { .. } => {
1493 formatter.write_str("component state codec upgrade is ambiguous")
1494 }
1495 Self::CodecUpgradeOvershootsCurrent { .. } => {
1496 formatter.write_str("component state codec upgrade passes the current version")
1497 }
1498 Self::CodecUpgradeChainTooLong { max_upgrades } => {
1499 write!(
1500 formatter,
1501 "component state codec upgrade chain exceeds {max_upgrades} upgrades"
1502 )
1503 }
1504 Self::ChecksumAlgorithmUnsupported { .. } => {
1505 formatter.write_str("component state checksum algorithm is unsupported")
1506 }
1507 Self::ChecksumMismatch => {
1508 formatter.write_str("component state checksum does not match")
1509 }
1510 Self::ExternalReferenceMissing => {
1511 formatter.write_str("component state external reference is missing")
1512 }
1513 Self::ExternalReferenceContentMismatch => {
1514 formatter.write_str("component state external content identity mismatch")
1515 }
1516 Self::InvalidPayload => {
1517 formatter.write_str("component state payload is not valid JSON")
1518 }
1519 Self::PayloadNotObject => {
1520 formatter.write_str("component state payload must be a JSON object")
1521 }
1522 Self::Codec(error) => error.fmt(formatter),
1523 }
1524 }
1525}
1526
1527impl Error for ComponentStateError {
1528 fn source(&self) -> Option<&(dyn Error + 'static)> {
1529 match self {
1530 Self::Codec(error) => Some(error),
1531 _ => None,
1532 }
1533 }
1534}