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