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