1use crate::{
7 db::{
8 codec::{
9 finalize_hash_sha256, new_hash_sha256, write_hash_len_u32, write_hash_str_u32,
10 write_hash_tag_u8, write_hash_u32, write_hash_u64,
11 },
12 commit::CommitSchemaFingerprint,
13 direction::Direction,
14 ordered_overlay::{OrderedOverlayEntry, OrderedOverlayVisit, visit_ordered_overlay},
15 schema::{
16 AcceptedFieldKind, AcceptedSchemaSnapshot, PersistedIndexKeyItemSnapshot,
17 PersistedIndexKeySnapshot, PersistedSchemaSnapshot, SchemaVersion,
18 accepted_schema_cache_fingerprint,
19 accepted_schema_cache_fingerprint_for_persisted_snapshot,
20 accepted_schema_cache_fingerprint_method_version, decode_persisted_schema_snapshot,
21 encode_persisted_schema_snapshot,
22 enum_catalog::{
23 AcceptedEnumCatalogHandle, AcceptedSchemaPublicationError, AcceptedSchemaRevision,
24 AcceptedSchemaRevisionBundle, AcceptedSchemaRootSelection,
25 AcceptedStoreCatalogScope, CandidateSchemaRevision,
26 decode_verified_accepted_schema_revision_bundle,
27 prepare_accepted_schema_root_publication, select_current_accepted_schema_root,
28 },
29 schema_snapshot_integrity_detail,
30 },
31 },
32 error::InternalError,
33 traits::Storable,
34 types::EntityTag,
35};
36use ic_memory::stable_structures::storable::Bound as StorableBound;
37use ic_memory::stable_structures::{
38 BTreeMap as StableBTreeMap, DefaultMemoryImpl, memory_manager::VirtualMemory,
39};
40use sha2::Digest;
41use std::borrow::Cow;
42#[cfg(test)]
43use std::cell::Cell;
44use std::cell::{OnceCell, Ref, RefCell};
45use std::collections::{BTreeMap as StdBTreeMap, BTreeSet};
46use std::convert::Infallible;
47use std::ops::Bound as RangeBound;
48
49const SCHEMA_KEY_BYTES_USIZE: usize = 16;
50const SCHEMA_KEY_BYTES: u32 = 16;
51const SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT: u8 = 0;
52const SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE: u8 = 1;
53const SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT: u8 = 2;
54pub(in crate::db) const MAX_SCHEMA_SNAPSHOT_BYTES: u32 = 512 * 1024;
55const SCHEMA_STORE_CATALOG_FINGERPRINT_VERSION: u8 = 1;
56const SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_VERSION: u8 = 2;
57const SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_VERSION: u8 = 3;
58const RAW_SCHEMA_SNAPSHOT_MAGIC: &[u8; 8] = b"ICYDBSCH";
59const RAW_SCHEMA_SNAPSHOT_VALUE_VERSION: u8 = 1;
60const RAW_SCHEMA_SNAPSHOT_HEADER_BYTES: usize = 25;
61
62#[cfg(test)]
63thread_local! {
64 static LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS: Cell<u64> = const { Cell::new(0) };
65 static ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES: Cell<u64> = const { Cell::new(0) };
66}
67
68#[cfg(test)]
69pub(in crate::db) fn reset_latest_raw_snapshots_by_entity_call_count_for_tests() {
70 LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS.with(|calls| calls.set(0));
71}
72
73#[cfg(test)]
74pub(in crate::db) fn latest_raw_snapshots_by_entity_call_count_for_tests() -> u64 {
75 LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS.with(Cell::get)
76}
77
78#[cfg(test)]
79fn reset_accepted_schema_bundle_cache_miss_count_for_tests() {
80 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(|misses| misses.set(0));
81}
82
83#[cfg(test)]
84fn accepted_schema_bundle_cache_miss_count_for_tests() -> u64 {
85 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(Cell::get)
86}
87
88#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
97struct RawSchemaKey([u8; SCHEMA_KEY_BYTES_USIZE]);
98
99impl RawSchemaKey {
100 #[must_use]
102 fn from_entity_version(entity: EntityTag, version: SchemaVersion) -> Self {
103 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
104 out[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
105 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
106 out[12..].copy_from_slice(&version.get().to_be_bytes());
107
108 Self(out)
109 }
110
111 fn from_accepted_bundle(bundle_key: super::enum_catalog::AcceptedSchemaBundleKey) -> Self {
112 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
113 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE;
114 out[4..12].copy_from_slice(&bundle_key.get().to_be_bytes());
115 Self(out)
116 }
117
118 fn from_accepted_root_slot(slot: usize) -> Result<Self, InternalError> {
119 let slot = u32::try_from(slot).map_err(|_| InternalError::store_invariant())?;
120 if slot > 1 {
121 return Err(InternalError::store_invariant());
122 }
123 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
124 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT;
125 out[12..].copy_from_slice(&slot.to_be_bytes());
126 Ok(Self(out))
127 }
128
129 #[must_use]
131 fn entity_tag(self) -> EntityTag {
132 let mut bytes = [0u8; size_of::<u64>()];
133 bytes.copy_from_slice(&self.0[4..12]);
134
135 EntityTag::new(u64::from_be_bytes(bytes))
136 }
137
138 #[must_use]
140 fn version(self) -> u32 {
141 let mut bytes = [0u8; size_of::<u32>()];
142 bytes.copy_from_slice(&self.0[12..]);
143
144 u32::from_be_bytes(bytes)
145 }
146
147 fn entity_range_bounds(entity: EntityTag) -> (RangeBound<Self>, RangeBound<Self>) {
148 (
149 RangeBound::Included(Self::from_entity_version(entity, SchemaVersion::new(0))),
150 RangeBound::Included(Self::from_entity_version(
151 entity,
152 SchemaVersion::new(u32::MAX),
153 )),
154 )
155 }
156
157 const fn all_entity_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
158 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
159 end[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
160 (
161 RangeBound::Included(Self([0; SCHEMA_KEY_BYTES_USIZE])),
162 RangeBound::Included(Self(end)),
163 )
164 }
165
166 #[cfg(test)]
167 const fn is_entity_snapshot(self) -> bool {
168 self.0[0] == SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT
169 }
170}
171
172impl Storable for RawSchemaKey {
173 fn to_bytes(&self) -> Cow<'_, [u8]> {
174 Cow::Borrowed(&self.0)
175 }
176
177 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
178 debug_assert_eq!(
179 bytes.len(),
180 SCHEMA_KEY_BYTES_USIZE,
181 "RawSchemaKey::from_bytes received unexpected byte length",
182 );
183
184 if bytes.len() != SCHEMA_KEY_BYTES_USIZE {
185 return Self([0u8; SCHEMA_KEY_BYTES_USIZE]);
186 }
187
188 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
189 out.copy_from_slice(bytes.as_ref());
190 Self(out)
191 }
192
193 fn into_bytes(self) -> Vec<u8> {
194 self.0.to_vec()
195 }
196
197 const BOUND: StorableBound = StorableBound::Bounded {
198 max_size: SCHEMA_KEY_BYTES,
199 is_fixed_size: true,
200 };
201}
202
203#[derive(Clone, Debug, Eq, PartialEq)]
214struct RawSchemaSnapshot {
215 payload: Vec<u8>,
216 accepted_schema_fingerprint: Option<CommitSchemaFingerprint>,
217}
218
219impl RawSchemaSnapshot {
220 fn from_persisted_snapshot(snapshot: &PersistedSchemaSnapshot) -> Result<Self, InternalError> {
222 validate_typed_schema_snapshot_for_store(snapshot)?;
223
224 let accepted_schema_fingerprint =
225 accepted_schema_cache_fingerprint_for_persisted_snapshot(snapshot)?;
226 let payload = encode_persisted_schema_snapshot(snapshot)?;
227
228 Ok(Self {
229 payload,
230 accepted_schema_fingerprint: Some(accepted_schema_fingerprint),
231 })
232 }
233
234 #[must_use]
236 const fn from_encoded_control_record(payload: Vec<u8>) -> Self {
237 Self {
238 payload,
239 accepted_schema_fingerprint: None,
240 }
241 }
242
243 #[cfg(test)]
246 #[must_use]
247 const fn from_unchecked_persisted_snapshot_payload(payload: Vec<u8>) -> Self {
248 Self {
249 payload,
250 accepted_schema_fingerprint: Some([0; size_of::<CommitSchemaFingerprint>()]),
251 }
252 }
253
254 #[must_use]
256 const fn as_bytes(&self) -> &[u8] {
257 self.payload.as_slice()
258 }
259
260 #[must_use]
262 fn into_bytes(self) -> Vec<u8> {
263 self.payload
264 }
265
266 fn accepted_schema_fingerprint(&self) -> Result<CommitSchemaFingerprint, InternalError> {
269 self.accepted_schema_fingerprint
270 .ok_or_else(InternalError::store_corruption)
271 }
272
273 fn decode_persisted_snapshot(&self) -> Result<PersistedSchemaSnapshot, InternalError> {
275 let _fingerprint = self.accepted_schema_fingerprint()?;
278 decode_persisted_schema_snapshot(self.as_bytes())
279 }
280}
281
282#[cfg(test)]
283pub(in crate::db::schema) fn validate_raw_schema_snapshot_bytes_for_tests(
284 bytes: Vec<u8>,
285) -> Result<(), InternalError> {
286 let raw = <RawSchemaSnapshot as Storable>::from_bytes(Cow::Owned(bytes));
287 raw.decode_persisted_snapshot().map(drop)
288}
289
290#[derive(Clone, Copy, Debug, Eq, PartialEq)]
291pub(in crate::db) struct AcceptedCatalogIdentity {
292 entity_tag: EntityTag,
293 entity_path: &'static str,
294 store_path: &'static str,
295 accepted_schema_revision: AcceptedSchemaRevision,
296 accepted_schema_version: SchemaVersion,
297 fingerprint_method_version: u8,
298 accepted_schema_fingerprint: CommitSchemaFingerprint,
299}
300
301impl AcceptedCatalogIdentity {
302 #[must_use]
303 pub(in crate::db) const fn new(
304 entity_tag: EntityTag,
305 entity_path: &'static str,
306 store_path: &'static str,
307 accepted_schema_revision: AcceptedSchemaRevision,
308 accepted_schema_version: SchemaVersion,
309 accepted_schema_fingerprint: CommitSchemaFingerprint,
310 ) -> Self {
311 Self {
312 entity_tag,
313 entity_path,
314 store_path,
315 accepted_schema_revision,
316 accepted_schema_version,
317 fingerprint_method_version: accepted_schema_cache_fingerprint_method_version(),
318 accepted_schema_fingerprint,
319 }
320 }
321
322 #[must_use]
323 pub(in crate::db) const fn entity_tag(self) -> EntityTag {
324 self.entity_tag
325 }
326
327 #[must_use]
328 pub(in crate::db) const fn entity_path(self) -> &'static str {
329 self.entity_path
330 }
331
332 #[must_use]
333 pub(in crate::db) const fn store_path(self) -> &'static str {
334 self.store_path
335 }
336
337 #[must_use]
338 pub(in crate::db) const fn accepted_schema_revision(self) -> AcceptedSchemaRevision {
339 self.accepted_schema_revision
340 }
341
342 #[must_use]
343 pub(in crate::db) const fn accepted_schema_version(self) -> SchemaVersion {
344 self.accepted_schema_version
345 }
346
347 #[must_use]
348 pub(in crate::db) const fn fingerprint_method_version(self) -> u8 {
349 self.fingerprint_method_version
350 }
351
352 #[must_use]
353 pub(in crate::db) const fn accepted_schema_fingerprint(self) -> CommitSchemaFingerprint {
354 self.accepted_schema_fingerprint
355 }
356}
357
358#[derive(Clone, Debug, Eq, PartialEq)]
359pub(in crate::db) struct AcceptedCatalogSnapshotSelection {
360 identity: AcceptedCatalogIdentity,
361 enum_catalog: AcceptedEnumCatalogHandle,
362 raw_snapshot: Vec<u8>,
363}
364
365impl AcceptedCatalogSnapshotSelection {
366 #[must_use]
367 const fn new(
368 identity: AcceptedCatalogIdentity,
369 enum_catalog: AcceptedEnumCatalogHandle,
370 raw_snapshot: Vec<u8>,
371 ) -> Self {
372 Self {
373 identity,
374 enum_catalog,
375 raw_snapshot,
376 }
377 }
378
379 #[must_use]
380 pub(in crate::db) const fn identity(&self) -> AcceptedCatalogIdentity {
381 self.identity
382 }
383
384 #[must_use]
385 pub(in crate::db) const fn enum_catalog(&self) -> &AcceptedEnumCatalogHandle {
386 &self.enum_catalog
387 }
388
389 pub(in crate::db) fn from_accepted_snapshot(
391 identity: AcceptedCatalogIdentity,
392 enum_catalog: AcceptedEnumCatalogHandle,
393 snapshot: &AcceptedSchemaSnapshot,
394 ) -> Result<Self, InternalError> {
395 let raw_snapshot =
396 RawSchemaSnapshot::from_persisted_snapshot(snapshot.persisted_snapshot())?;
397 if raw_snapshot.accepted_schema_fingerprint()? != identity.accepted_schema_fingerprint() {
398 return Err(InternalError::store_invariant());
399 }
400
401 Ok(Self::new(identity, enum_catalog, raw_snapshot.into_bytes()))
402 }
403
404 pub(in crate::db) fn decode_verified(&self) -> Result<AcceptedSchemaSnapshot, InternalError> {
405 let snapshot = decode_persisted_schema_snapshot(&self.raw_snapshot)?;
406 let accepted = AcceptedSchemaSnapshot::try_new(snapshot)?;
407 let identity = self.identity();
408
409 if accepted.persisted_snapshot().version() != identity.accepted_schema_version() {
410 return Err(InternalError::store_invariant());
411 }
412 if accepted.entity_path() != identity.entity_path() {
413 return Err(InternalError::store_invariant());
414 }
415
416 let decoded_fingerprint = accepted_schema_cache_fingerprint(&accepted)?;
417 if decoded_fingerprint != identity.accepted_schema_fingerprint() {
418 return Err(InternalError::store_invariant());
419 }
420
421 Ok(accepted)
422 }
423}
424
425impl Storable for RawSchemaSnapshot {
426 fn to_bytes(&self) -> Cow<'_, [u8]> {
427 let Some(fingerprint) = self.accepted_schema_fingerprint else {
428 return Cow::Borrowed(self.as_bytes());
429 };
430
431 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
432 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
433 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
434 bytes.extend_from_slice(&fingerprint);
435 bytes.extend_from_slice(self.as_bytes());
436
437 Cow::Owned(bytes)
438 }
439
440 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
441 let bytes = bytes.into_owned();
442 if bytes.len() >= RAW_SCHEMA_SNAPSHOT_HEADER_BYTES
443 && &bytes[..RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_MAGIC
444 && bytes[RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_VALUE_VERSION
445 {
446 let fingerprint_start = RAW_SCHEMA_SNAPSHOT_MAGIC.len() + size_of::<u8>();
447 let fingerprint_end = fingerprint_start + size_of::<CommitSchemaFingerprint>();
448 let mut fingerprint = [0_u8; size_of::<CommitSchemaFingerprint>()];
449 fingerprint.copy_from_slice(&bytes[fingerprint_start..fingerprint_end]);
450
451 return Self {
452 payload: bytes[fingerprint_end..].to_vec(),
453 accepted_schema_fingerprint: Some(fingerprint),
454 };
455 }
456
457 Self {
458 payload: bytes,
459 accepted_schema_fingerprint: None,
460 }
461 }
462
463 fn into_bytes(self) -> Vec<u8> {
464 let Some(fingerprint) = self.accepted_schema_fingerprint else {
465 return self.payload;
466 };
467
468 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
469 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
470 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
471 bytes.extend_from_slice(&fingerprint);
472 bytes.extend_from_slice(&self.payload);
473
474 bytes
475 }
476
477 const BOUND: StorableBound = StorableBound::Unbounded;
478}
479
480fn validate_typed_schema_snapshot_for_store(
484 snapshot: &PersistedSchemaSnapshot,
485) -> Result<(), InternalError> {
486 if schema_snapshot_integrity_detail(
487 "schema snapshot",
488 snapshot.version(),
489 snapshot.primary_key_field_ids(),
490 snapshot.row_layout(),
491 snapshot.fields(),
492 )
493 .is_some()
494 {
495 return Err(InternalError::store_invariant());
496 }
497
498 Ok(())
499}
500
501#[derive(Clone, Copy, Debug, Eq, PartialEq)]
510pub(in crate::db) struct SchemaStoreFootprint {
511 snapshots: u64,
512 encoded_bytes: u64,
513 latest_snapshot_bytes: u64,
514}
515
516#[derive(Clone, Copy, Debug, Eq, PartialEq)]
524pub(in crate::db) struct SchemaStoreCatalogMetadata {
525 schema_version: SchemaVersion,
526 schema_fingerprint_method_version: u8,
527 schema_fingerprint: CommitSchemaFingerprint,
528 entity_count: u64,
529}
530
531impl SchemaStoreCatalogMetadata {
532 #[must_use]
534 const fn new(
535 schema_version: SchemaVersion,
536 schema_fingerprint_method_version: u8,
537 schema_fingerprint: CommitSchemaFingerprint,
538 entity_count: u64,
539 ) -> Self {
540 Self {
541 schema_version,
542 schema_fingerprint_method_version,
543 schema_fingerprint,
544 entity_count,
545 }
546 }
547
548 #[must_use]
550 pub(in crate::db) const fn schema_version(self) -> SchemaVersion {
551 self.schema_version
552 }
553
554 #[must_use]
556 pub(in crate::db) const fn schema_fingerprint_method_version(self) -> u8 {
557 self.schema_fingerprint_method_version
558 }
559
560 #[must_use]
563 pub(in crate::db) const fn schema_fingerprint(self) -> CommitSchemaFingerprint {
564 self.schema_fingerprint
565 }
566
567 #[must_use]
569 pub(in crate::db) const fn entity_count(self) -> u64 {
570 self.entity_count
571 }
572}
573
574#[derive(Clone, Copy, Debug, Eq, PartialEq)]
583pub(in crate::db) struct SchemaStoreAllocationMetadata {
584 data: SchemaStoreCatalogMetadata,
585 index: SchemaStoreCatalogMetadata,
586 schema: SchemaStoreCatalogMetadata,
587}
588
589impl SchemaStoreAllocationMetadata {
590 #[must_use]
593 const fn new(
594 data: SchemaStoreCatalogMetadata,
595 index: SchemaStoreCatalogMetadata,
596 schema: SchemaStoreCatalogMetadata,
597 ) -> Self {
598 Self {
599 data,
600 index,
601 schema,
602 }
603 }
604
605 #[must_use]
607 pub(in crate::db) const fn data(self) -> SchemaStoreCatalogMetadata {
608 self.data
609 }
610
611 #[must_use]
613 pub(in crate::db) const fn index(self) -> SchemaStoreCatalogMetadata {
614 self.index
615 }
616
617 #[must_use]
620 pub(in crate::db) const fn schema(self) -> SchemaStoreCatalogMetadata {
621 self.schema
622 }
623}
624
625impl SchemaStoreFootprint {
626 #[must_use]
628 const fn new(snapshots: u64, encoded_bytes: u64, latest_snapshot_bytes: u64) -> Self {
629 Self {
630 snapshots,
631 encoded_bytes,
632 latest_snapshot_bytes,
633 }
634 }
635
636 #[must_use]
638 pub(in crate::db) const fn snapshots(self) -> u64 {
639 self.snapshots
640 }
641
642 #[must_use]
644 pub(in crate::db) const fn encoded_bytes(self) -> u64 {
645 self.encoded_bytes
646 }
647
648 #[must_use]
650 pub(in crate::db) const fn latest_snapshot_bytes(self) -> u64 {
651 self.latest_snapshot_bytes
652 }
653}
654
655pub struct SchemaStore {
664 backend: SchemaStoreBackend,
665 accepted_bundle_cache: RefCell<Option<AcceptedSchemaBundleCache>>,
666 accepted_catalog_scope: OnceCell<AcceptedStoreCatalogScope>,
667}
668
669struct AcceptedSchemaBundleCache {
670 selection: AcceptedSchemaRootSelection,
671 bundle: AcceptedSchemaRevisionBundle,
672}
673
674enum SchemaStoreBackend {
675 Heap(StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>),
676 Journaled {
677 canonical:
678 StableBTreeMap<RawSchemaKey, RawSchemaSnapshot, VirtualMemory<DefaultMemoryImpl>>,
679 live: StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
680 tombstones: BTreeSet<RawSchemaKey>,
681 },
682}
683
684#[derive(Clone, Copy, Debug, Eq, PartialEq)]
686enum SchemaStoreVisit {
687 Continue,
688 Stop,
689}
690
691impl SchemaStoreVisit {
692 const fn should_stop(self) -> bool {
693 matches!(self, Self::Stop)
694 }
695}
696
697impl SchemaStore {
698 #[must_use]
700 pub const fn init_heap() -> Self {
701 Self {
702 backend: SchemaStoreBackend::Heap(StdBTreeMap::new()),
703 accepted_bundle_cache: RefCell::new(None),
704 accepted_catalog_scope: OnceCell::new(),
705 }
706 }
707
708 #[must_use]
713 pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
714 Self {
715 backend: SchemaStoreBackend::Journaled {
716 canonical: StableBTreeMap::init(memory),
717 live: StdBTreeMap::new(),
718 tombstones: BTreeSet::new(),
719 },
720 accepted_bundle_cache: RefCell::new(None),
721 accepted_catalog_scope: OnceCell::new(),
722 }
723 }
724
725 pub(in crate::db) fn insert_persisted_snapshot(
727 &mut self,
728 entity: EntityTag,
729 snapshot: &PersistedSchemaSnapshot,
730 ) -> Result<(), InternalError> {
731 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
732 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
733 let _ = self.insert_raw_snapshot(key, raw_snapshot);
734
735 Ok(())
736 }
737
738 pub(in crate::db) fn reset_journaled_live_projection(&mut self) -> Result<(), InternalError> {
741 let SchemaStoreBackend::Journaled {
742 live, tombstones, ..
743 } = &mut self.backend
744 else {
745 return Err(InternalError::store_invariant());
746 };
747
748 live.clear();
749 tombstones.clear();
750 self.accepted_bundle_cache.get_mut().take();
751
752 Ok(())
753 }
754
755 pub(in crate::db) fn fold_persisted_snapshot(
757 &mut self,
758 entity: EntityTag,
759 snapshot: &PersistedSchemaSnapshot,
760 ) -> Result<(), InternalError> {
761 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
762 return Err(InternalError::store_invariant());
763 };
764
765 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
766 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
767 canonical.insert(key, raw_snapshot);
768
769 Ok(())
770 }
771
772 pub(in crate::db) fn current_accepted_schema_root(
774 &self,
775 ) -> Result<Option<AcceptedSchemaRootSelection>, InternalError> {
776 let first = self.accepted_root_slot_bytes(0)?;
777 let second = self.accepted_root_slot_bytes(1)?;
778 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])
779 }
780
781 pub(in crate::db) fn current_accepted_schema_bundle(
783 &self,
784 ) -> Result<Option<AcceptedSchemaRevisionBundle>, InternalError> {
785 let Some(selection) = self.current_accepted_schema_root()? else {
786 return Ok(None);
787 };
788 let key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
789 let raw = self
790 .get_raw_snapshot(&key)
791 .ok_or_else(InternalError::store_corruption)?;
792 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes()).map(Some)
793 }
794
795 pub(in crate::db) fn current_accepted_schema_revision(
797 &self,
798 ) -> Result<Option<AcceptedSchemaRevision>, InternalError> {
799 Ok(self
800 .current_accepted_schema_root()?
801 .map(|selection| selection.root().revision()))
802 }
803
804 pub(in crate::db) fn publish_accepted_schema_candidate(
808 &mut self,
809 expected_revision: AcceptedSchemaRevision,
810 candidate: &CandidateSchemaRevision,
811 ) -> Result<(), InternalError> {
812 let first = self.accepted_root_slot_bytes(0)?;
813 let second = self.accepted_root_slot_bytes(1)?;
814 prepare_accepted_schema_root_publication(
815 [first.as_deref(), second.as_deref()],
816 expected_revision,
817 candidate,
818 )
819 .map_err(map_schema_publication_error)?;
820
821 self.insert_durable_candidate_snapshots(candidate)?;
822 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
823 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
824 let persisted_bundle = self
825 .get_raw_snapshot(&bundle_key)
826 .ok_or_else(InternalError::store_corruption)?;
827 let _verified = decode_verified_accepted_schema_revision_bundle(
828 candidate.root(),
829 persisted_bundle.as_bytes(),
830 )?;
831
832 let first = self.accepted_root_slot_bytes(0)?;
835 let second = self.accepted_root_slot_bytes(1)?;
836 let publication = prepare_accepted_schema_root_publication(
837 [first.as_deref(), second.as_deref()],
838 expected_revision,
839 candidate,
840 )
841 .map_err(map_schema_publication_error)?;
842 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
843 self.insert_durable_raw_value(root_key, publication.encoded_root().to_vec());
844
845 let selected = self
846 .current_accepted_schema_root()?
847 .ok_or_else(InternalError::store_corruption)?;
848 if selected.root() != candidate.root() {
849 return Err(InternalError::store_corruption());
850 }
851 Ok(())
852 }
853
854 pub(in crate::db) fn apply_journaled_accepted_schema_candidate(
856 &mut self,
857 expected_revision: AcceptedSchemaRevision,
858 candidate: &CandidateSchemaRevision,
859 ) -> Result<(), InternalError> {
860 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
861 return Err(InternalError::store_invariant());
862 }
863 if self.current_root_matches_candidate(candidate)? {
864 return Ok(());
865 }
866
867 let first = self.accepted_root_slot_bytes(0)?;
868 let second = self.accepted_root_slot_bytes(1)?;
869 prepare_accepted_schema_root_publication(
870 [first.as_deref(), second.as_deref()],
871 expected_revision,
872 candidate,
873 )
874 .map_err(map_schema_publication_error)?;
875
876 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
877 self.insert_persisted_snapshot(*entity_tag, snapshot)?;
878 }
879 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
880 self.insert_raw_snapshot(
881 bundle_key,
882 RawSchemaSnapshot::from_encoded_control_record(candidate.encoded_bundle().to_vec()),
883 );
884 let persisted_bundle = self
885 .get_raw_snapshot(&bundle_key)
886 .ok_or_else(InternalError::store_corruption)?;
887 let _verified = decode_verified_accepted_schema_revision_bundle(
888 candidate.root(),
889 persisted_bundle.as_bytes(),
890 )?;
891
892 let first = self.accepted_root_slot_bytes(0)?;
893 let second = self.accepted_root_slot_bytes(1)?;
894 let publication = prepare_accepted_schema_root_publication(
895 [first.as_deref(), second.as_deref()],
896 expected_revision,
897 candidate,
898 )
899 .map_err(map_schema_publication_error)?;
900 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
901 self.insert_raw_snapshot(
902 root_key,
903 RawSchemaSnapshot::from_encoded_control_record(publication.encoded_root().to_vec()),
904 );
905
906 if !self.current_root_matches_candidate(candidate)? {
907 return Err(InternalError::store_corruption());
908 }
909 Ok(())
910 }
911
912 pub(in crate::db) fn fold_journaled_accepted_schema_candidate(
914 &mut self,
915 expected_revision: AcceptedSchemaRevision,
916 candidate: &CandidateSchemaRevision,
917 ) -> Result<(), InternalError> {
918 if self.canonical_root_matches_candidate(candidate)? {
919 return Ok(());
920 }
921
922 let first = self.canonical_root_slot_bytes(0)?;
923 let second = self.canonical_root_slot_bytes(1)?;
924 prepare_accepted_schema_root_publication(
925 [first.as_deref(), second.as_deref()],
926 expected_revision,
927 candidate,
928 )
929 .map_err(map_schema_publication_error)?;
930
931 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
932 self.fold_persisted_snapshot(*entity_tag, snapshot)?;
933 }
934 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
935 self.insert_canonical_raw_value(bundle_key, candidate.encoded_bundle().to_vec())?;
936 let persisted_bundle = self
937 .get_canonical_raw_value(&bundle_key)?
938 .ok_or_else(InternalError::store_corruption)?;
939 let _verified = decode_verified_accepted_schema_revision_bundle(
940 candidate.root(),
941 persisted_bundle.as_bytes(),
942 )?;
943
944 let first = self.canonical_root_slot_bytes(0)?;
945 let second = self.canonical_root_slot_bytes(1)?;
946 let publication = prepare_accepted_schema_root_publication(
947 [first.as_deref(), second.as_deref()],
948 expected_revision,
949 candidate,
950 )
951 .map_err(map_schema_publication_error)?;
952 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
953 self.insert_canonical_raw_value(root_key, publication.encoded_root().to_vec())?;
954
955 if !self.canonical_root_matches_candidate(candidate)? {
956 return Err(InternalError::store_corruption());
957 }
958 Ok(())
959 }
960
961 #[cfg(test)]
963 pub(in crate::db) fn get_persisted_snapshot(
964 &self,
965 entity: EntityTag,
966 version: SchemaVersion,
967 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
968 let key = RawSchemaKey::from_entity_version(entity, version);
969 self.get_raw_snapshot(&key)
970 .map(|snapshot| snapshot.decode_persisted_snapshot())
971 .transpose()
972 }
973
974 pub(in crate::db) fn latest_staged_persisted_snapshot(
979 &self,
980 entity: EntityTag,
981 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
982 self.latest_raw_snapshot(entity)
983 .map(|snapshot| snapshot.decode_persisted_snapshot())
984 .transpose()
985 }
986
987 pub(in crate::db) fn current_accepted_persisted_snapshot(
990 &self,
991 entity: EntityTag,
992 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
993 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
994 return Ok(None);
995 };
996
997 Ok(bundle.entity_snapshots().get(&entity).cloned())
998 }
999
1000 pub(in crate::db) fn current_accepted_catalog_selection(
1002 &self,
1003 entity: EntityTag,
1004 entity_path: &'static str,
1005 store_path: &'static str,
1006 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
1007 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1008 return Ok(None);
1009 };
1010 if bundle.store_path() != store_path {
1011 return Err(InternalError::store_corruption());
1012 }
1013 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
1014 return Ok(None);
1015 };
1016 if snapshot.entity_path() != entity_path {
1017 return Err(InternalError::store_corruption());
1018 }
1019
1020 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1021 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
1022 let identity = AcceptedCatalogIdentity::new(
1023 entity,
1024 entity_path,
1025 store_path,
1026 bundle.revision(),
1027 snapshot.version(),
1028 fingerprint,
1029 );
1030
1031 Ok(Some(AcceptedCatalogSnapshotSelection::new(
1032 identity,
1033 AcceptedEnumCatalogHandle::new(
1034 bundle.enum_catalog().clone(),
1035 self.accepted_catalog_scope
1036 .get_or_init(AcceptedStoreCatalogScope::new)
1037 .clone(),
1038 bundle.revision(),
1039 self.current_accepted_schema_root()?
1040 .ok_or_else(InternalError::store_corruption)?
1041 .root()
1042 .fingerprint(),
1043 ),
1044 raw_snapshot.into_bytes(),
1045 )))
1046 }
1047
1048 #[must_use]
1050 pub(in crate::db) fn entity_footprint(&self, entity: EntityTag) -> SchemaStoreFootprint {
1051 let mut snapshots = 0u64;
1052 let mut encoded_bytes = 0u64;
1053 let mut latest = None::<(SchemaVersion, u64)>;
1054
1055 let _: Result<(), std::convert::Infallible> = self.visit_raw_snapshots(|key, snapshot| {
1056 if key.entity_tag() != entity {
1057 return Ok(SchemaStoreVisit::Continue);
1058 }
1059
1060 let snapshot_bytes = u64::try_from(snapshot.as_bytes().len()).unwrap_or(u64::MAX);
1061 snapshots = snapshots.saturating_add(1);
1062 encoded_bytes = encoded_bytes.saturating_add(snapshot_bytes);
1063
1064 let version = SchemaVersion::new(key.version());
1065 if latest
1066 .as_ref()
1067 .is_none_or(|(latest_version, _)| version > *latest_version)
1068 {
1069 latest = Some((version, snapshot_bytes));
1070 }
1071 Ok(SchemaStoreVisit::Continue)
1072 });
1073
1074 SchemaStoreFootprint::new(
1075 snapshots,
1076 encoded_bytes,
1077 latest.map_or(0, |(_, snapshot_bytes)| snapshot_bytes),
1078 )
1079 }
1080
1081 #[cfg(test)]
1087 pub(in crate::db) fn catalog_metadata(
1088 &self,
1089 ) -> Result<Option<SchemaStoreCatalogMetadata>, InternalError> {
1090 Ok(self
1091 .allocation_metadata()?
1092 .map(SchemaStoreAllocationMetadata::schema))
1093 }
1094
1095 pub(in crate::db) fn allocation_metadata(
1102 &self,
1103 ) -> Result<Option<SchemaStoreAllocationMetadata>, InternalError> {
1104 let latest_by_entity = self.latest_raw_snapshots_by_entity();
1105 if latest_by_entity.is_empty() {
1106 return Ok(None);
1107 }
1108
1109 Ok(Some(SchemaStoreAllocationMetadata::new(
1110 derive_data_allocation_metadata(&latest_by_entity)?,
1111 derive_index_allocation_metadata(&latest_by_entity)?,
1112 derive_schema_catalog_metadata(&latest_by_entity)?,
1113 )))
1114 }
1115
1116 fn insert_raw_snapshot(
1118 &mut self,
1119 key: RawSchemaKey,
1120 snapshot: RawSchemaSnapshot,
1121 ) -> Option<RawSchemaSnapshot> {
1122 let previous_journaled = if matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1123 self.get_raw_snapshot_for_backend(&key)
1124 } else {
1125 None
1126 };
1127 match &mut self.backend {
1128 SchemaStoreBackend::Heap(map) => map.insert(key, snapshot),
1129 SchemaStoreBackend::Journaled {
1130 live, tombstones, ..
1131 } => {
1132 tombstones.remove(&key);
1133 live.insert(key, snapshot);
1134 previous_journaled
1135 }
1136 }
1137 }
1138
1139 #[must_use]
1141 fn get_raw_snapshot(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
1142 match &self.backend {
1143 SchemaStoreBackend::Heap(map) => map.get(key).cloned(),
1144 SchemaStoreBackend::Journaled { .. } => self.get_raw_snapshot_for_backend(key),
1145 }
1146 }
1147
1148 fn accepted_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
1149 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
1150 Ok(self
1151 .get_raw_snapshot(&key)
1152 .map(RawSchemaSnapshot::into_bytes))
1153 }
1154
1155 fn canonical_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
1156 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
1157 Ok(self
1158 .get_canonical_raw_value(&key)?
1159 .map(RawSchemaSnapshot::into_bytes))
1160 }
1161
1162 fn current_root_matches_candidate(
1163 &self,
1164 candidate: &CandidateSchemaRevision,
1165 ) -> Result<bool, InternalError> {
1166 let Some(selection) = self.current_accepted_schema_root()? else {
1167 return Ok(false);
1168 };
1169 if selection.root() != candidate.root() {
1170 return Ok(false);
1171 }
1172 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1173 let bundle = self
1174 .get_raw_snapshot(&key)
1175 .ok_or_else(InternalError::store_corruption)?;
1176 let _verified =
1177 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
1178 Ok(true)
1179 }
1180
1181 fn canonical_root_matches_candidate(
1182 &self,
1183 candidate: &CandidateSchemaRevision,
1184 ) -> Result<bool, InternalError> {
1185 let first = self.canonical_root_slot_bytes(0)?;
1186 let second = self.canonical_root_slot_bytes(1)?;
1187 let Some(selection) =
1188 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1189 else {
1190 return Ok(false);
1191 };
1192 if selection.root() != candidate.root() {
1193 return Ok(false);
1194 }
1195 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1196 let bundle = self
1197 .get_canonical_raw_value(&key)?
1198 .ok_or_else(InternalError::store_corruption)?;
1199 let _verified =
1200 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
1201 Ok(true)
1202 }
1203
1204 fn get_canonical_raw_value(
1205 &self,
1206 key: &RawSchemaKey,
1207 ) -> Result<Option<RawSchemaSnapshot>, InternalError> {
1208 match &self.backend {
1209 SchemaStoreBackend::Journaled { canonical, .. } => Ok(canonical.get(key)),
1210 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
1211 }
1212 }
1213
1214 fn insert_canonical_raw_value(
1215 &mut self,
1216 key: RawSchemaKey,
1217 bytes: Vec<u8>,
1218 ) -> Result<(), InternalError> {
1219 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1220 return Err(InternalError::store_invariant());
1221 };
1222 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
1223 Ok(())
1224 }
1225
1226 fn insert_durable_raw_value(&mut self, key: RawSchemaKey, bytes: Vec<u8>) {
1230 let value = RawSchemaSnapshot::from_encoded_control_record(bytes);
1231 match &mut self.backend {
1232 SchemaStoreBackend::Heap(map) => {
1233 map.insert(key, value);
1234 }
1235 SchemaStoreBackend::Journaled {
1236 canonical,
1237 live,
1238 tombstones,
1239 } => {
1240 live.remove(&key);
1241 tombstones.remove(&key);
1242 canonical.insert(key, value);
1243 }
1244 }
1245 }
1246
1247 fn insert_durable_candidate_snapshots(
1248 &mut self,
1249 candidate: &CandidateSchemaRevision,
1250 ) -> Result<(), InternalError> {
1251 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
1252 let key = RawSchemaKey::from_entity_version(*entity_tag, snapshot.version());
1253 let value = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1254 match &mut self.backend {
1255 SchemaStoreBackend::Heap(map) => {
1256 map.insert(key, value);
1257 }
1258 SchemaStoreBackend::Journaled {
1259 canonical,
1260 live,
1261 tombstones,
1262 } => {
1263 live.remove(&key);
1264 tombstones.remove(&key);
1265 canonical.insert(key, value);
1266 }
1267 }
1268 }
1269 Ok(())
1270 }
1271
1272 #[must_use]
1274 #[cfg(test)]
1275 fn contains_raw_snapshot(&self, key: &RawSchemaKey) -> bool {
1276 match &self.backend {
1277 SchemaStoreBackend::Heap(map) => map.contains_key(key),
1278 SchemaStoreBackend::Journaled { .. } => {
1279 self.get_raw_snapshot_for_backend(key).is_some()
1280 }
1281 }
1282 }
1283
1284 #[must_use]
1286 #[cfg(test)]
1287 pub(in crate::db) fn len(&self) -> u64 {
1288 match &self.backend {
1289 SchemaStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
1290 SchemaStoreBackend::Journaled { .. } => {
1291 let mut count = 0_u64;
1292 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
1293 count = count.saturating_add(1);
1294 Ok(SchemaStoreVisit::Continue)
1295 });
1296 count
1297 }
1298 }
1299 }
1300
1301 #[must_use]
1303 #[cfg(test)]
1304 pub(in crate::db) fn is_empty(&self) -> bool {
1305 match &self.backend {
1306 SchemaStoreBackend::Heap(map) => map.is_empty(),
1307 SchemaStoreBackend::Journaled { .. } => {
1308 let mut empty = true;
1309 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
1310 empty = false;
1311 Ok(SchemaStoreVisit::Stop)
1312 });
1313 empty
1314 }
1315 }
1316 }
1317
1318 #[cfg(test)]
1320 pub(in crate::db) fn clear(&mut self) {
1321 self.accepted_bundle_cache.get_mut().take();
1322 match &mut self.backend {
1323 SchemaStoreBackend::Heap(map) => map.clear(),
1324 SchemaStoreBackend::Journaled {
1325 canonical,
1326 live,
1327 tombstones,
1328 } => {
1329 live.clear();
1330 tombstones.clear();
1331 let keys = canonical
1332 .iter()
1333 .map(|entry| *entry.key())
1334 .collect::<Vec<_>>();
1335 for key in keys {
1336 if key.is_entity_snapshot() {
1337 tombstones.insert(key);
1338 } else {
1339 canonical.remove(&key);
1340 }
1341 }
1342 }
1343 }
1344 }
1345
1346 fn current_accepted_schema_bundle_ref(
1347 &self,
1348 ) -> Result<Option<Ref<'_, AcceptedSchemaRevisionBundle>>, InternalError> {
1349 let Some(selection) = self.current_accepted_schema_root()? else {
1350 self.accepted_bundle_cache
1351 .try_borrow_mut()
1352 .map_err(|_| InternalError::store_invariant())?
1353 .take();
1354 return Ok(None);
1355 };
1356
1357 let cache_matches = self
1358 .accepted_bundle_cache
1359 .try_borrow()
1360 .map_err(|_| InternalError::store_invariant())?
1361 .as_ref()
1362 .is_some_and(|cached| cached.selection == selection);
1363 if !cache_matches {
1364 let key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
1365 let raw = self
1366 .get_raw_snapshot(&key)
1367 .ok_or_else(InternalError::store_corruption)?;
1368 let bundle =
1369 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes())?;
1370 #[cfg(test)]
1371 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES
1372 .with(|misses| misses.set(misses.get().saturating_add(1)));
1373 *self
1374 .accepted_bundle_cache
1375 .try_borrow_mut()
1376 .map_err(|_| InternalError::store_invariant())? =
1377 Some(AcceptedSchemaBundleCache { selection, bundle });
1378 }
1379
1380 let cache = self
1381 .accepted_bundle_cache
1382 .try_borrow()
1383 .map_err(|_| InternalError::store_invariant())?;
1384 Ref::filter_map(cache, |cache| {
1385 cache
1386 .as_ref()
1387 .filter(|cached| cached.selection == selection)
1388 .map(|cached| &cached.bundle)
1389 })
1390 .map(Some)
1391 .map_err(|_| InternalError::store_invariant())
1392 }
1393
1394 fn latest_raw_snapshots_by_entity(
1395 &self,
1396 ) -> StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)> {
1397 #[cfg(test)]
1398 LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS.with(|calls| calls.set(calls.get().saturating_add(1)));
1399
1400 let mut latest_by_entity =
1401 StdBTreeMap::<EntityTag, (SchemaVersion, RawSchemaSnapshot)>::new();
1402
1403 let _: Result<(), std::convert::Infallible> = self.visit_raw_snapshots(|key, snapshot| {
1404 let version = SchemaVersion::new(key.version());
1405 match latest_by_entity.get_mut(&key.entity_tag()) {
1406 Some((latest_version, latest_snapshot)) if version > *latest_version => {
1407 *latest_version = version;
1408 *latest_snapshot = snapshot.clone();
1409 }
1410 None => {
1411 latest_by_entity.insert(key.entity_tag(), (version, snapshot.clone()));
1412 }
1413 Some(_) => {}
1414 }
1415 Ok(SchemaStoreVisit::Continue)
1416 });
1417
1418 latest_by_entity
1419 }
1420
1421 fn visit_raw_snapshots<E>(
1424 &self,
1425 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
1426 ) -> Result<(), E> {
1427 let bounds = RawSchemaKey::all_entity_range_bounds();
1428 match &self.backend {
1429 SchemaStoreBackend::Heap(map) => {
1430 let mut visitor = visitor;
1431 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
1432 if visitor(key, snapshot)?.should_stop() {
1433 break;
1434 }
1435 }
1436 }
1437 SchemaStoreBackend::Journaled {
1438 canonical,
1439 live,
1440 tombstones,
1441 } => Self::visit_journaled_raw_snapshot_range(
1442 canonical,
1443 live,
1444 tombstones,
1445 bounds,
1446 Direction::Asc,
1447 visitor,
1448 )?,
1449 }
1450
1451 Ok(())
1452 }
1453
1454 #[cfg(test)]
1455 #[must_use]
1456 pub(in crate::db) fn canonical_len_for_tests(&self) -> u64 {
1457 match &self.backend {
1458 SchemaStoreBackend::Journaled { canonical: map, .. } => map.len(),
1459 SchemaStoreBackend::Heap(_) => 0,
1460 }
1461 }
1462
1463 fn get_raw_snapshot_for_backend(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
1464 let SchemaStoreBackend::Journaled {
1465 canonical,
1466 live,
1467 tombstones,
1468 } = &self.backend
1469 else {
1470 return None;
1471 };
1472
1473 if tombstones.contains(key) {
1474 return None;
1475 }
1476 live.get(key).cloned().or_else(|| canonical.get(key))
1477 }
1478
1479 fn latest_raw_snapshot(&self, entity: EntityTag) -> Option<RawSchemaSnapshot> {
1480 self.latest_raw_snapshot_entry(entity)
1481 .map(|(_, snapshot)| snapshot)
1482 }
1483
1484 fn latest_raw_snapshot_entry(
1485 &self,
1486 entity: EntityTag,
1487 ) -> Option<(SchemaVersion, RawSchemaSnapshot)> {
1488 let bounds = RawSchemaKey::entity_range_bounds(entity);
1489 match &self.backend {
1490 SchemaStoreBackend::Heap(map) => map
1491 .range((bounds.0, bounds.1))
1492 .next_back()
1493 .map(|(key, snapshot)| (SchemaVersion::new(key.version()), snapshot.clone())),
1494 SchemaStoreBackend::Journaled {
1495 canonical,
1496 live,
1497 tombstones,
1498 } => {
1499 let mut latest = None;
1500 let _: Result<(), Infallible> = Self::visit_journaled_raw_snapshot_range(
1501 canonical,
1502 live,
1503 tombstones,
1504 bounds,
1505 Direction::Desc,
1506 |key, snapshot| {
1507 latest = Some((SchemaVersion::new(key.version()), snapshot.clone()));
1508 Ok(SchemaStoreVisit::Stop)
1509 },
1510 );
1511 latest
1512 }
1513 }
1514 }
1515
1516 fn visit_journaled_raw_snapshot_range<E>(
1517 canonical: &StableBTreeMap<
1518 RawSchemaKey,
1519 RawSchemaSnapshot,
1520 VirtualMemory<DefaultMemoryImpl>,
1521 >,
1522 live: &StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
1523 tombstones: &BTreeSet<RawSchemaKey>,
1524 bounds: (RangeBound<RawSchemaKey>, RangeBound<RawSchemaKey>),
1525 direction: Direction,
1526 mut visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
1527 ) -> Result<(), E> {
1528 match direction {
1529 Direction::Asc => visit_ordered_overlay(
1530 canonical.range((bounds.0, bounds.1)),
1531 live.range((bounds.0, bounds.1)),
1532 Direction::Asc,
1533 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
1534 |canonical_entry| !tombstones.contains(canonical_entry.key()),
1535 |live_entry| !tombstones.contains(live_entry.0),
1536 |entry| {
1537 let visit = match entry {
1538 OrderedOverlayEntry::Canonical(canonical_entry) => {
1539 visitor(canonical_entry.key(), &canonical_entry.value())?
1540 }
1541 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
1542 };
1543 Ok(if visit.should_stop() {
1544 OrderedOverlayVisit::Stop
1545 } else {
1546 OrderedOverlayVisit::Continue
1547 })
1548 },
1549 ),
1550 Direction::Desc => visit_ordered_overlay(
1551 canonical.range((bounds.0, bounds.1)).rev(),
1552 live.range((bounds.0, bounds.1)).rev(),
1553 Direction::Desc,
1554 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
1555 |canonical_entry| !tombstones.contains(canonical_entry.key()),
1556 |live_entry| !tombstones.contains(live_entry.0),
1557 |entry| {
1558 let visit = match entry {
1559 OrderedOverlayEntry::Canonical(canonical_entry) => {
1560 visitor(canonical_entry.key(), &canonical_entry.value())?
1561 }
1562 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
1563 };
1564 Ok(if visit.should_stop() {
1565 OrderedOverlayVisit::Stop
1566 } else {
1567 OrderedOverlayVisit::Continue
1568 })
1569 },
1570 ),
1571 }
1572 }
1573}
1574
1575fn map_schema_publication_error(error: AcceptedSchemaPublicationError) -> InternalError {
1576 match error {
1577 AcceptedSchemaPublicationError::StaleSchemaRevision { .. }
1578 | AcceptedSchemaPublicationError::RevisionExhausted => InternalError::store_unsupported(),
1579 AcceptedSchemaPublicationError::InvalidCandidate => InternalError::store_invariant(),
1580 AcceptedSchemaPublicationError::CorruptRootSlots => InternalError::store_corruption(),
1581 }
1582}
1583
1584fn derive_data_allocation_metadata(
1585 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
1586) -> Result<SchemaStoreCatalogMetadata, InternalError> {
1587 let mut max_version = SchemaVersion::initial();
1588 let mut hasher = new_hash_sha256();
1589 write_hash_tag_u8(
1590 &mut hasher,
1591 SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_VERSION,
1592 );
1593
1594 for (entity, (_, snapshot)) in latest_by_entity {
1595 let persisted = snapshot.decode_persisted_snapshot()?;
1596 if persisted.version() > max_version {
1597 max_version = persisted.version();
1598 }
1599
1600 let data_projection = PersistedSchemaSnapshot::new_with_primary_key_fields_and_indexes(
1601 persisted.version(),
1602 persisted.entity_path().to_string(),
1603 persisted.entity_name().to_string(),
1604 persisted.primary_key_field_ids().to_vec(),
1605 persisted.row_layout().clone(),
1606 persisted.fields().to_vec(),
1607 Vec::new(),
1608 );
1609 let encoded = encode_persisted_schema_snapshot(&data_projection)?;
1610
1611 write_hash_u64(&mut hasher, entity.value());
1612 write_hash_u32(&mut hasher, persisted.version().get());
1613 write_hash_len_u32(&mut hasher, encoded.len());
1614 hasher.update(encoded);
1615 }
1616
1617 Ok(finalize_schema_metadata(
1618 max_version,
1619 SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_VERSION,
1620 hasher,
1621 latest_by_entity.len(),
1622 ))
1623}
1624
1625fn derive_index_allocation_metadata(
1626 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
1627) -> Result<SchemaStoreCatalogMetadata, InternalError> {
1628 let mut max_version = SchemaVersion::initial();
1629 let mut hasher = new_hash_sha256();
1630 write_hash_tag_u8(
1631 &mut hasher,
1632 SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_VERSION,
1633 );
1634
1635 for (entity, (_, snapshot)) in latest_by_entity {
1636 let persisted = snapshot.decode_persisted_snapshot()?;
1637 if persisted.version() > max_version {
1638 max_version = persisted.version();
1639 }
1640
1641 write_hash_u64(&mut hasher, entity.value());
1642 write_hash_u32(&mut hasher, persisted.version().get());
1643 write_hash_len_u32(&mut hasher, persisted.indexes().len());
1644 for index in persisted.indexes() {
1645 write_hash_u32(&mut hasher, u32::from(index.ordinal()));
1646 write_hash_str_u32(&mut hasher, index.name());
1647 write_hash_str_u32(&mut hasher, index.store());
1648 write_hash_tag_u8(&mut hasher, u8::from(index.unique()));
1649 write_hash_str_u32(&mut hasher, persisted_index_origin_name(index.origin()));
1650 match index.predicate_sql() {
1651 Some(predicate_sql) => {
1652 write_hash_tag_u8(&mut hasher, 1);
1653 write_hash_str_u32(&mut hasher, predicate_sql);
1654 }
1655 None => write_hash_tag_u8(&mut hasher, 0),
1656 }
1657 hash_persisted_index_key(&mut hasher, index.key());
1658 }
1659 }
1660
1661 Ok(finalize_schema_metadata(
1662 max_version,
1663 SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_VERSION,
1664 hasher,
1665 latest_by_entity.len(),
1666 ))
1667}
1668
1669fn derive_schema_catalog_metadata(
1670 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
1671) -> Result<SchemaStoreCatalogMetadata, InternalError> {
1672 let mut max_version = SchemaVersion::initial();
1673 let mut hasher = new_hash_sha256();
1674 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_CATALOG_FINGERPRINT_VERSION);
1675
1676 for (entity, (version, snapshot)) in latest_by_entity {
1677 let persisted = snapshot.decode_persisted_snapshot()?;
1678 if persisted.version() > max_version {
1679 max_version = persisted.version();
1680 }
1681
1682 write_hash_u64(&mut hasher, entity.value());
1683 write_hash_u32(&mut hasher, version.get());
1684 write_hash_len_u32(&mut hasher, snapshot.as_bytes().len());
1685 hasher.update(snapshot.as_bytes());
1686 }
1687
1688 Ok(finalize_schema_metadata(
1689 max_version,
1690 SCHEMA_STORE_CATALOG_FINGERPRINT_VERSION,
1691 hasher,
1692 latest_by_entity.len(),
1693 ))
1694}
1695
1696fn finalize_schema_metadata(
1697 schema_version: SchemaVersion,
1698 schema_fingerprint_method_version: u8,
1699 hasher: sha2::Sha256,
1700 entity_count: usize,
1701) -> SchemaStoreCatalogMetadata {
1702 let digest = finalize_hash_sha256(hasher);
1703 let mut schema_fingerprint = [0u8; 16];
1704 schema_fingerprint.copy_from_slice(&digest[..16]);
1705
1706 SchemaStoreCatalogMetadata::new(
1707 schema_version,
1708 schema_fingerprint_method_version,
1709 schema_fingerprint,
1710 u64::try_from(entity_count).unwrap_or(u64::MAX),
1711 )
1712}
1713
1714fn hash_persisted_index_key(hasher: &mut sha2::Sha256, key: &PersistedIndexKeySnapshot) {
1715 match key {
1716 PersistedIndexKeySnapshot::FieldPath(paths) => {
1717 write_hash_tag_u8(hasher, 1);
1718 write_hash_len_u32(hasher, paths.len());
1719 for path in paths {
1720 hash_persisted_index_field_path(hasher, path);
1721 }
1722 }
1723 PersistedIndexKeySnapshot::Items(items) => {
1724 write_hash_tag_u8(hasher, 2);
1725 write_hash_len_u32(hasher, items.len());
1726 for item in items {
1727 match item {
1728 PersistedIndexKeyItemSnapshot::FieldPath(path) => {
1729 write_hash_tag_u8(hasher, 1);
1730 hash_persisted_index_field_path(hasher, path);
1731 }
1732 PersistedIndexKeyItemSnapshot::Expression(expression) => {
1733 write_hash_tag_u8(hasher, 2);
1734 write_hash_str_u32(hasher, persisted_expression_op_name(expression.op()));
1735 hash_persisted_index_field_path(hasher, expression.source());
1736 hash_accepted_field_kind(hasher, expression.input_kind());
1737 hash_accepted_field_kind(hasher, expression.output_kind());
1738 write_hash_str_u32(hasher, expression.canonical_text());
1739 }
1740 }
1741 }
1742 }
1743 }
1744}
1745
1746fn hash_persisted_index_field_path(
1747 hasher: &mut sha2::Sha256,
1748 path: &crate::db::schema::PersistedIndexFieldPathSnapshot,
1749) {
1750 write_hash_u32(hasher, path.field_id().get());
1751 write_hash_u32(hasher, u32::from(path.slot().get()));
1752 write_hash_len_u32(hasher, path.path().len());
1753 for segment in path.path() {
1754 write_hash_str_u32(hasher, segment);
1755 }
1756 hash_accepted_field_kind(hasher, path.kind());
1757 write_hash_tag_u8(hasher, u8::from(path.nullable()));
1758}
1759
1760fn hash_accepted_field_kind(hasher: &mut sha2::Sha256, kind: &AcceptedFieldKind) {
1761 match kind {
1762 AcceptedFieldKind::Account => write_hash_tag_u8(hasher, 1),
1763 AcceptedFieldKind::Blob { max_len } => {
1764 write_hash_tag_u8(hasher, 2);
1765 hash_optional_u32(hasher, *max_len);
1766 }
1767 AcceptedFieldKind::Bool => write_hash_tag_u8(hasher, 3),
1768 AcceptedFieldKind::Date => write_hash_tag_u8(hasher, 4),
1769 AcceptedFieldKind::Decimal { scale } => {
1770 write_hash_tag_u8(hasher, 5);
1771 write_hash_u32(hasher, *scale);
1772 }
1773 AcceptedFieldKind::Duration => write_hash_tag_u8(hasher, 6),
1774 AcceptedFieldKind::Enum { type_id } => {
1775 write_hash_tag_u8(hasher, 7);
1776 write_hash_u32(hasher, type_id.get());
1777 }
1778 AcceptedFieldKind::Float32 => write_hash_tag_u8(hasher, 8),
1779 AcceptedFieldKind::Float64 => write_hash_tag_u8(hasher, 9),
1780 AcceptedFieldKind::Int8 => write_hash_tag_u8(hasher, 10),
1781 AcceptedFieldKind::Int16 => write_hash_tag_u8(hasher, 11),
1782 AcceptedFieldKind::Int32 => write_hash_tag_u8(hasher, 12),
1783 AcceptedFieldKind::Int64 => write_hash_tag_u8(hasher, 13),
1784 AcceptedFieldKind::Int128 => write_hash_tag_u8(hasher, 14),
1785 AcceptedFieldKind::IntBig { max_bytes } => {
1786 write_hash_tag_u8(hasher, 15);
1787 write_hash_u32(hasher, *max_bytes);
1788 }
1789 AcceptedFieldKind::Principal => write_hash_tag_u8(hasher, 16),
1790 AcceptedFieldKind::Subaccount => write_hash_tag_u8(hasher, 17),
1791 AcceptedFieldKind::Text { max_len } => {
1792 write_hash_tag_u8(hasher, 18);
1793 hash_optional_u32(hasher, *max_len);
1794 }
1795 AcceptedFieldKind::Timestamp => write_hash_tag_u8(hasher, 19),
1796 AcceptedFieldKind::Nat8 => write_hash_tag_u8(hasher, 20),
1797 AcceptedFieldKind::Nat16 => write_hash_tag_u8(hasher, 21),
1798 AcceptedFieldKind::Nat32 => write_hash_tag_u8(hasher, 22),
1799 AcceptedFieldKind::Nat64 => write_hash_tag_u8(hasher, 23),
1800 AcceptedFieldKind::Nat128 => write_hash_tag_u8(hasher, 24),
1801 AcceptedFieldKind::NatBig { max_bytes } => {
1802 write_hash_tag_u8(hasher, 25);
1803 write_hash_u32(hasher, *max_bytes);
1804 }
1805 AcceptedFieldKind::Ulid => write_hash_tag_u8(hasher, 26),
1806 AcceptedFieldKind::Unit => write_hash_tag_u8(hasher, 27),
1807 AcceptedFieldKind::Relation {
1808 target_path,
1809 target_entity_name,
1810 target_entity_tag,
1811 target_store_path,
1812 key_kind,
1813 strength,
1814 } => {
1815 write_hash_tag_u8(hasher, 28);
1816 write_hash_str_u32(hasher, target_path);
1817 write_hash_str_u32(hasher, target_entity_name);
1818 write_hash_u64(hasher, target_entity_tag.value());
1819 write_hash_str_u32(hasher, target_store_path);
1820 hash_accepted_field_kind(hasher, key_kind);
1821 write_hash_str_u32(hasher, accepted_relation_strength_name(*strength));
1822 }
1823 AcceptedFieldKind::List(inner) => {
1824 write_hash_tag_u8(hasher, 29);
1825 hash_accepted_field_kind(hasher, inner);
1826 }
1827 AcceptedFieldKind::Set(inner) => {
1828 write_hash_tag_u8(hasher, 30);
1829 hash_accepted_field_kind(hasher, inner);
1830 }
1831 AcceptedFieldKind::Map { key, value } => {
1832 write_hash_tag_u8(hasher, 31);
1833 hash_accepted_field_kind(hasher, key);
1834 hash_accepted_field_kind(hasher, value);
1835 }
1836 AcceptedFieldKind::Structured { queryable } => {
1837 write_hash_tag_u8(hasher, 32);
1838 write_hash_tag_u8(hasher, u8::from(*queryable));
1839 }
1840 }
1841}
1842
1843fn hash_optional_u32(hasher: &mut sha2::Sha256, value: Option<u32>) {
1844 match value {
1845 Some(value) => {
1846 write_hash_tag_u8(hasher, 1);
1847 write_hash_u32(hasher, value);
1848 }
1849 None => write_hash_tag_u8(hasher, 0),
1850 }
1851}
1852
1853const fn persisted_index_origin_name(
1854 origin: crate::db::schema::PersistedIndexOrigin,
1855) -> &'static str {
1856 match origin {
1857 crate::db::schema::PersistedIndexOrigin::Generated => "generated",
1858 crate::db::schema::PersistedIndexOrigin::SqlDdl => "sql_ddl",
1859 }
1860}
1861
1862const fn persisted_expression_op_name(
1863 op: crate::db::schema::PersistedIndexExpressionOp,
1864) -> &'static str {
1865 match op {
1866 crate::db::schema::PersistedIndexExpressionOp::Lower => "lower",
1867 crate::db::schema::PersistedIndexExpressionOp::Upper => "upper",
1868 crate::db::schema::PersistedIndexExpressionOp::Trim => "trim",
1869 crate::db::schema::PersistedIndexExpressionOp::LowerTrim => "lower_trim",
1870 crate::db::schema::PersistedIndexExpressionOp::Date => "date",
1871 crate::db::schema::PersistedIndexExpressionOp::Year => "year",
1872 crate::db::schema::PersistedIndexExpressionOp::Month => "month",
1873 crate::db::schema::PersistedIndexExpressionOp::Day => "day",
1874 }
1875}
1876
1877const fn accepted_relation_strength_name(
1878 strength: crate::db::schema::AcceptedRelationStrength,
1879) -> &'static str {
1880 match strength {
1881 crate::db::schema::AcceptedRelationStrength::Strong => "strong",
1882 crate::db::schema::AcceptedRelationStrength::Weak => "weak",
1883 }
1884}
1885
1886#[cfg(test)]
1891mod tests;