1use crate::db::schema::identity_state::{
7 IdentityAdvanceId, IdentityRangeAdvance, IdentityRangeCommitState, IdentityState,
8 IdentityStateInventory, IdentityStateLifecycle, IdentityStateTransition,
9 IdentityStatementCursor, MAX_IDENTITY_STATE_RECORDS_PER_DATABASE, decode_identity_state,
10 encode_identity_state, identity_kind_maximum, prepare_identity_state_transition,
11 validate_identity_state_closure,
12};
13use crate::{
14 db::{
15 codec::{
16 finalize_hash_sha256, new_hash_sha256, write_hash_len_u32, write_hash_str_u32,
17 write_hash_tag_u8, write_hash_u32, write_hash_u64,
18 },
19 commit::CommitSchemaFingerprint,
20 direction::Direction,
21 integrity::DatabaseIncarnationId,
22 ordered_overlay::{OrderedOverlayEntry, OrderedOverlayVisit, visit_ordered_overlay},
23 runtime_entity_catalog::AcceptedRuntimeEntity,
24 schema::{
25 AcceptedFieldKind, AcceptedRowLayoutRuntimeContract, AcceptedSchemaSnapshot,
26 ConstraintActivationKind, ConstraintActivationState, ConstraintId, ConstraintOrigin,
27 ConstraintValidationJob, FieldId, PersistedIndexKeyItemSnapshot,
28 PersistedIndexKeySnapshot, PersistedSchemaSnapshot, SchemaVersion,
29 accepted_schema_cache_fingerprint,
30 accepted_schema_cache_fingerprint_for_persisted_snapshot,
31 accepted_schema_cache_fingerprint_method_version, decode_constraint_validation_job,
32 decode_persisted_schema_snapshot, encode_constraint_validation_job,
33 encode_persisted_schema_snapshot,
34 enum_catalog::{
35 AcceptedSchemaAuthority, AcceptedSchemaPublicationError, AcceptedSchemaRevision,
36 AcceptedSchemaRevisionBundle, AcceptedSchemaRootSelection,
37 AcceptedStoreCatalogScope, AcceptedValueCatalogHandle, CandidateSchemaRevision,
38 decode_verified_accepted_schema_revision_bundle,
39 prepare_accepted_schema_root_publication, select_current_accepted_schema_root,
40 },
41 schema_snapshot_integrity_detail,
42 },
43 },
44 error::InternalError,
45 types::EntityTag,
46};
47use ic_stable_structures::{
48 BTreeMap as StableBTreeMap, DefaultMemoryImpl, Storable, memory_manager::VirtualMemory,
49 storable::Bound as StorableBound,
50};
51use sha2::Digest;
52use std::borrow::Cow;
53#[cfg(test)]
54use std::cell::Cell;
55use std::cell::{OnceCell, Ref, RefCell};
56use std::collections::{BTreeMap as StdBTreeMap, BTreeSet};
57#[cfg(test)]
58use std::convert::Infallible;
59use std::ops::Bound as RangeBound;
60use std::rc::Rc;
61
62const SCHEMA_KEY_BYTES_USIZE: usize = 16;
63const SCHEMA_KEY_BYTES: u32 = 16;
64const SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT: u8 = 0;
65const SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE: u8 = 1;
66const SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT: u8 = 2;
67const SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB: u8 = 3;
68const SCHEMA_KEY_NAMESPACE_IDENTITY_STATE: u8 = 4;
69const SCHEMA_STORE_FINGERPRINT_METHOD_VERSION: u8 = 1;
72const SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN: u8 = 1;
73const SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN: u8 = 2;
74const SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN: u8 = 3;
75const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL: u8 = 3;
76const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST: u8 = 29;
77const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET: u8 = 30;
78const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP: u8 = 31;
79const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE: u8 = 32;
80const RAW_SCHEMA_SNAPSHOT_MAGIC: &[u8; 8] = b"ICYDBCAT";
81const RAW_SCHEMA_SNAPSHOT_VALUE_VERSION: u8 = 1;
82const RAW_SCHEMA_SNAPSHOT_HEADER_BYTES: usize = 25;
83
84pub(in crate::db) fn load_accepted_schema_snapshot(
89 schema_store: &SchemaStore,
90 entity_tag: EntityTag,
91 entity_path: &str,
92) -> Result<AcceptedSchemaSnapshot, InternalError> {
93 let bundle = schema_store
94 .current_accepted_schema_bundle()?
95 .ok_or_else(InternalError::store_corruption)?;
96 let snapshot = bundle
97 .entity_snapshots()
98 .get(&entity_tag)
99 .cloned()
100 .ok_or_else(InternalError::store_corruption)?;
101 if snapshot.entity_path() != entity_path {
102 return Err(InternalError::store_corruption());
103 }
104 let accepted = AcceptedSchemaSnapshot::try_new(snapshot)?;
105 let _runtime_contract = AcceptedRowLayoutRuntimeContract::from_accepted_schema(&accepted)?;
106
107 Ok(accepted)
108}
109
110#[cfg(test)]
111thread_local! {
112 static ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES: Cell<u64> = const { Cell::new(0) };
113}
114
115#[cfg(test)]
116fn reset_accepted_schema_bundle_cache_miss_count_for_tests() {
117 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(|misses| misses.set(0));
118}
119
120#[cfg(test)]
121fn accepted_schema_bundle_cache_miss_count_for_tests() -> u64 {
122 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(Cell::get)
123}
124
125#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
134struct RawSchemaKey([u8; SCHEMA_KEY_BYTES_USIZE]);
135
136impl RawSchemaKey {
137 #[must_use]
139 fn from_entity_version(entity: EntityTag, version: SchemaVersion) -> Self {
140 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
141 out[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
142 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
143 out[12..].copy_from_slice(&version.get().to_be_bytes());
144
145 Self(out)
146 }
147
148 fn from_accepted_bundle(bundle_key: super::enum_catalog::AcceptedSchemaBundleKey) -> Self {
149 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
150 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE;
151 out[4..12].copy_from_slice(&bundle_key.get().to_be_bytes());
152 Self(out)
153 }
154
155 fn from_accepted_root_slot(slot: usize) -> Result<Self, InternalError> {
156 let slot = u32::try_from(slot).map_err(|_| InternalError::store_invariant())?;
157 if slot > 1 {
158 return Err(InternalError::store_invariant());
159 }
160 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
161 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT;
162 out[12..].copy_from_slice(&slot.to_be_bytes());
163 Ok(Self(out))
164 }
165
166 fn from_constraint_validation_job(entity: EntityTag, constraint_id: ConstraintId) -> Self {
167 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
168 out[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
169 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
170 out[12..].copy_from_slice(&constraint_id.get().to_be_bytes());
171 Self(out)
172 }
173
174 fn from_identity_state(entity: EntityTag, field_id: FieldId) -> Self {
175 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
176 out[0] = SCHEMA_KEY_NAMESPACE_IDENTITY_STATE;
177 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
178 out[12..].copy_from_slice(&field_id.get().to_be_bytes());
179 Self(out)
180 }
181
182 #[must_use]
184 fn entity_tag(self) -> EntityTag {
185 let mut bytes = [0u8; size_of::<u64>()];
186 bytes.copy_from_slice(&self.0[4..12]);
187
188 EntityTag::new(u64::from_be_bytes(bytes))
189 }
190
191 #[must_use]
193 fn version(self) -> u32 {
194 let mut bytes = [0u8; size_of::<u32>()];
195 bytes.copy_from_slice(&self.0[12..]);
196
197 u32::from_be_bytes(bytes)
198 }
199
200 const fn all_entity_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
201 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
202 end[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
203 (
204 RangeBound::Included(Self([0; SCHEMA_KEY_BYTES_USIZE])),
205 RangeBound::Included(Self(end)),
206 )
207 }
208
209 #[cfg(test)]
210 fn entity_range_bounds(entity: EntityTag) -> (RangeBound<Self>, RangeBound<Self>) {
211 (
212 RangeBound::Included(Self::from_entity_version(entity, SchemaVersion::initial())),
213 RangeBound::Included(Self::from_entity_version(
214 entity,
215 SchemaVersion::new(u32::MAX),
216 )),
217 )
218 }
219
220 const fn all_constraint_validation_job_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
221 let mut start = [0u8; SCHEMA_KEY_BYTES_USIZE];
222 start[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
223 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
224 end[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
225 (
226 RangeBound::Included(Self(start)),
227 RangeBound::Included(Self(end)),
228 )
229 }
230
231 const fn all_identity_state_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
232 let mut start = [0u8; SCHEMA_KEY_BYTES_USIZE];
233 start[0] = SCHEMA_KEY_NAMESPACE_IDENTITY_STATE;
234 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
235 end[0] = SCHEMA_KEY_NAMESPACE_IDENTITY_STATE;
236 (
237 RangeBound::Included(Self(start)),
238 RangeBound::Included(Self(end)),
239 )
240 }
241
242 #[cfg(test)]
243 const fn is_entity_snapshot(self) -> bool {
244 self.0[0] == SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT
245 }
246
247 const fn is_accepted_root(self) -> bool {
248 self.0[0] == SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT
249 }
250
251 const fn is_constraint_validation_job(self) -> bool {
252 self.0[0] == SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB
253 }
254
255 const fn is_identity_state(self) -> bool {
256 self.0[0] == SCHEMA_KEY_NAMESPACE_IDENTITY_STATE
257 }
258
259 fn constraint_id(self) -> Option<ConstraintId> {
260 self.is_constraint_validation_job()
261 .then(|| ConstraintId::new(self.version()))
262 .flatten()
263 }
264}
265
266impl Storable for RawSchemaKey {
267 fn to_bytes(&self) -> Cow<'_, [u8]> {
268 Cow::Borrowed(&self.0)
269 }
270
271 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
272 debug_assert_eq!(
273 bytes.len(),
274 SCHEMA_KEY_BYTES_USIZE,
275 "RawSchemaKey::from_bytes received unexpected byte length",
276 );
277
278 if bytes.len() != SCHEMA_KEY_BYTES_USIZE {
279 return Self([0u8; SCHEMA_KEY_BYTES_USIZE]);
280 }
281
282 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
283 out.copy_from_slice(bytes.as_ref());
284 Self(out)
285 }
286
287 fn into_bytes(self) -> Vec<u8> {
288 self.0.to_vec()
289 }
290
291 const BOUND: StorableBound = StorableBound::Bounded {
292 max_size: SCHEMA_KEY_BYTES,
293 is_fixed_size: true,
294 };
295}
296
297#[derive(Clone, Debug, Eq, PartialEq)]
308struct RawSchemaSnapshot {
309 payload: Vec<u8>,
310 accepted_schema_fingerprint: Option<CommitSchemaFingerprint>,
311}
312
313impl RawSchemaSnapshot {
314 fn from_persisted_snapshot(snapshot: &PersistedSchemaSnapshot) -> Result<Self, InternalError> {
316 validate_typed_schema_snapshot_for_store(snapshot)?;
317
318 let accepted_schema_fingerprint =
319 accepted_schema_cache_fingerprint_for_persisted_snapshot(snapshot)?;
320 let payload = encode_persisted_schema_snapshot(snapshot)?;
321
322 Ok(Self {
323 payload,
324 accepted_schema_fingerprint: Some(accepted_schema_fingerprint),
325 })
326 }
327
328 #[must_use]
330 const fn from_encoded_control_record(payload: Vec<u8>) -> Self {
331 Self {
332 payload,
333 accepted_schema_fingerprint: None,
334 }
335 }
336
337 #[cfg(test)]
340 #[must_use]
341 const fn from_unchecked_persisted_snapshot_payload(payload: Vec<u8>) -> Self {
342 Self {
343 payload,
344 accepted_schema_fingerprint: Some([0; size_of::<CommitSchemaFingerprint>()]),
345 }
346 }
347
348 #[must_use]
350 const fn as_bytes(&self) -> &[u8] {
351 self.payload.as_slice()
352 }
353
354 #[must_use]
356 fn into_bytes(self) -> Vec<u8> {
357 self.payload
358 }
359
360 fn accepted_schema_fingerprint(&self) -> Result<CommitSchemaFingerprint, InternalError> {
363 self.accepted_schema_fingerprint
364 .ok_or_else(InternalError::store_corruption)
365 }
366
367 fn decode_persisted_snapshot(&self) -> Result<PersistedSchemaSnapshot, InternalError> {
369 let _fingerprint = self.accepted_schema_fingerprint()?;
372 decode_persisted_schema_snapshot(self.as_bytes())
373 }
374}
375
376#[cfg(test)]
377pub(in crate::db::schema) fn validate_raw_schema_snapshot_bytes_for_tests(
378 bytes: Vec<u8>,
379) -> Result<(), InternalError> {
380 let raw = <RawSchemaSnapshot as Storable>::from_bytes(Cow::Owned(bytes));
381 raw.decode_persisted_snapshot().map(drop)
382}
383
384#[derive(Clone, Debug, Eq, PartialEq)]
385pub(in crate::db) struct AcceptedCatalogIdentity {
386 entity_tag: EntityTag,
387 entity_path: Rc<str>,
388 store_path: &'static str,
389 accepted_schema_revision: AcceptedSchemaRevision,
390 accepted_schema_version: SchemaVersion,
391 fingerprint_method_version: u8,
392 accepted_schema_fingerprint: CommitSchemaFingerprint,
393}
394
395impl AcceptedCatalogIdentity {
396 #[must_use]
397 pub(in crate::db) fn new(
398 entity_tag: EntityTag,
399 entity_path: impl Into<Rc<str>>,
400 store_path: &'static str,
401 accepted_schema_revision: AcceptedSchemaRevision,
402 accepted_schema_version: SchemaVersion,
403 accepted_schema_fingerprint: CommitSchemaFingerprint,
404 ) -> Self {
405 Self {
406 entity_tag,
407 entity_path: entity_path.into(),
408 store_path,
409 accepted_schema_revision,
410 accepted_schema_version,
411 fingerprint_method_version: accepted_schema_cache_fingerprint_method_version(),
412 accepted_schema_fingerprint,
413 }
414 }
415
416 #[must_use]
417 pub(in crate::db) const fn entity_tag(&self) -> EntityTag {
418 self.entity_tag
419 }
420
421 #[must_use]
422 pub(in crate::db) fn entity_path(&self) -> &str {
423 self.entity_path.as_ref()
424 }
425
426 #[must_use]
427 pub(in crate::db) fn entity_path_handle(&self) -> Rc<str> {
428 self.entity_path.clone()
429 }
430
431 #[must_use]
432 pub(in crate::db) const fn store_path(&self) -> &'static str {
433 self.store_path
434 }
435
436 #[must_use]
437 pub(in crate::db) const fn accepted_schema_revision(&self) -> AcceptedSchemaRevision {
438 self.accepted_schema_revision
439 }
440
441 #[must_use]
442 pub(in crate::db) const fn accepted_schema_version(&self) -> SchemaVersion {
443 self.accepted_schema_version
444 }
445
446 #[must_use]
447 pub(in crate::db) const fn fingerprint_method_version(&self) -> u8 {
448 self.fingerprint_method_version
449 }
450
451 #[must_use]
452 pub(in crate::db) const fn accepted_schema_fingerprint(&self) -> CommitSchemaFingerprint {
453 self.accepted_schema_fingerprint
454 }
455}
456
457#[derive(Clone, Debug, Eq, PartialEq)]
458pub(in crate::db) struct AcceptedCatalogSnapshotSelection {
459 identity: AcceptedCatalogIdentity,
460 value_catalog: AcceptedValueCatalogHandle,
461 raw_snapshot: Rc<[u8]>,
462}
463
464impl AcceptedCatalogSnapshotSelection {
465 #[must_use]
466 const fn new(
467 identity: AcceptedCatalogIdentity,
468 value_catalog: AcceptedValueCatalogHandle,
469 raw_snapshot: Rc<[u8]>,
470 ) -> Self {
471 Self {
472 identity,
473 value_catalog,
474 raw_snapshot,
475 }
476 }
477
478 #[must_use]
479 pub(in crate::db) fn identity(&self) -> AcceptedCatalogIdentity {
480 self.identity.clone()
481 }
482
483 #[must_use]
484 pub(in crate::db) const fn value_catalog_handle(&self) -> &AcceptedValueCatalogHandle {
485 &self.value_catalog
486 }
487
488 pub(in crate::db) fn from_candidate(
491 candidate: &CandidateSchemaRevision,
492 entity_tag: EntityTag,
493 entity_path: &str,
494 store_path: &'static str,
495 ) -> Result<Option<Self>, InternalError> {
496 if candidate.store_path() != store_path {
497 return Err(InternalError::store_corruption());
498 }
499 let Some(snapshot) = candidate.bundle().entity_snapshots().get(&entity_tag) else {
500 return Ok(None);
501 };
502 if snapshot.entity_path() != entity_path {
503 return Err(InternalError::store_corruption());
504 }
505
506 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
507 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
508 let identity = AcceptedCatalogIdentity::new(
509 entity_tag,
510 entity_path,
511 store_path,
512 candidate.revision(),
513 snapshot.version(),
514 fingerprint,
515 );
516
517 Ok(Some(Self::new(
518 identity,
519 AcceptedValueCatalogHandle::new(
520 candidate.bundle().enum_catalog().clone(),
521 candidate.bundle().composite_catalog().clone(),
522 AcceptedStoreCatalogScope::new(),
523 candidate.revision(),
524 candidate.root().fingerprint(),
525 ),
526 Rc::from(raw_snapshot.into_bytes()),
527 )))
528 }
529
530 pub(in crate::db) fn decode_verified(&self) -> Result<AcceptedSchemaSnapshot, InternalError> {
531 let snapshot = decode_persisted_schema_snapshot(self.raw_snapshot.as_ref())?;
532 let accepted = AcceptedSchemaSnapshot::try_new(snapshot)?;
533 let identity = self.identity();
534
535 if accepted.persisted_snapshot().version() != identity.accepted_schema_version() {
536 return Err(InternalError::store_invariant());
537 }
538 if accepted.entity_path() != identity.entity_path() {
539 return Err(InternalError::store_invariant());
540 }
541
542 let decoded_fingerprint = accepted_schema_cache_fingerprint(&accepted)?;
543 if decoded_fingerprint != identity.accepted_schema_fingerprint() {
544 return Err(InternalError::store_invariant());
545 }
546
547 Ok(accepted)
548 }
549}
550
551impl Storable for RawSchemaSnapshot {
552 fn to_bytes(&self) -> Cow<'_, [u8]> {
553 let Some(fingerprint) = self.accepted_schema_fingerprint else {
554 return Cow::Borrowed(self.as_bytes());
555 };
556
557 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
558 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
559 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
560 bytes.extend_from_slice(&fingerprint);
561 bytes.extend_from_slice(self.as_bytes());
562
563 Cow::Owned(bytes)
564 }
565
566 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
567 let bytes = bytes.into_owned();
568 if bytes.len() >= RAW_SCHEMA_SNAPSHOT_HEADER_BYTES
569 && &bytes[..RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_MAGIC
570 && bytes[RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_VALUE_VERSION
571 {
572 let fingerprint_start = RAW_SCHEMA_SNAPSHOT_MAGIC.len() + size_of::<u8>();
573 let fingerprint_end = fingerprint_start + size_of::<CommitSchemaFingerprint>();
574 let mut fingerprint = [0_u8; size_of::<CommitSchemaFingerprint>()];
575 fingerprint.copy_from_slice(&bytes[fingerprint_start..fingerprint_end]);
576
577 return Self {
578 payload: bytes[fingerprint_end..].to_vec(),
579 accepted_schema_fingerprint: Some(fingerprint),
580 };
581 }
582
583 Self {
584 payload: bytes,
585 accepted_schema_fingerprint: None,
586 }
587 }
588
589 fn into_bytes(self) -> Vec<u8> {
590 let Some(fingerprint) = self.accepted_schema_fingerprint else {
591 return self.payload;
592 };
593
594 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
595 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
596 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
597 bytes.extend_from_slice(&fingerprint);
598 bytes.extend_from_slice(&self.payload);
599
600 bytes
601 }
602
603 const BOUND: StorableBound = StorableBound::Unbounded;
604}
605
606fn validate_typed_schema_snapshot_for_store(
610 snapshot: &PersistedSchemaSnapshot,
611) -> Result<(), InternalError> {
612 if schema_snapshot_integrity_detail(
613 "schema snapshot",
614 snapshot.version(),
615 snapshot.primary_key_field_ids(),
616 snapshot.row_layout(),
617 snapshot.fields(),
618 )
619 .is_some()
620 {
621 return Err(InternalError::store_invariant());
622 }
623
624 Ok(())
625}
626
627#[derive(Clone, Copy, Debug, Eq, PartialEq)]
635pub(in crate::db) struct SchemaStoreCatalogMetadata {
636 schema_version: SchemaVersion,
637 schema_fingerprint_method_version: u8,
638 schema_fingerprint: CommitSchemaFingerprint,
639 entity_count: u64,
640}
641
642impl SchemaStoreCatalogMetadata {
643 #[must_use]
645 const fn new(
646 schema_version: SchemaVersion,
647 schema_fingerprint_method_version: u8,
648 schema_fingerprint: CommitSchemaFingerprint,
649 entity_count: u64,
650 ) -> Self {
651 Self {
652 schema_version,
653 schema_fingerprint_method_version,
654 schema_fingerprint,
655 entity_count,
656 }
657 }
658
659 #[must_use]
661 pub(in crate::db) const fn schema_version(self) -> SchemaVersion {
662 self.schema_version
663 }
664
665 #[must_use]
667 pub(in crate::db) const fn schema_fingerprint_method_version(self) -> u8 {
668 self.schema_fingerprint_method_version
669 }
670
671 #[must_use]
674 pub(in crate::db) const fn schema_fingerprint(self) -> CommitSchemaFingerprint {
675 self.schema_fingerprint
676 }
677
678 #[must_use]
680 pub(in crate::db) const fn entity_count(self) -> u64 {
681 self.entity_count
682 }
683}
684
685#[derive(Clone, Copy, Debug, Eq, PartialEq)]
694pub(in crate::db) struct SchemaStoreAllocationMetadata {
695 data: SchemaStoreCatalogMetadata,
696 index: SchemaStoreCatalogMetadata,
697 schema: SchemaStoreCatalogMetadata,
698}
699
700impl SchemaStoreAllocationMetadata {
701 #[must_use]
704 const fn new(
705 data: SchemaStoreCatalogMetadata,
706 index: SchemaStoreCatalogMetadata,
707 schema: SchemaStoreCatalogMetadata,
708 ) -> Self {
709 Self {
710 data,
711 index,
712 schema,
713 }
714 }
715
716 #[must_use]
718 pub(in crate::db) const fn data(self) -> SchemaStoreCatalogMetadata {
719 self.data
720 }
721
722 #[must_use]
724 pub(in crate::db) const fn index(self) -> SchemaStoreCatalogMetadata {
725 self.index
726 }
727
728 #[must_use]
731 pub(in crate::db) const fn schema(self) -> SchemaStoreCatalogMetadata {
732 self.schema
733 }
734}
735
736pub(in crate::db) struct PendingRelationActivationDeleteBarrier {
744 accepted_schema_fingerprint: CommitSchemaFingerprint,
745 source_entity_tag: EntityTag,
746 constraint_id: ConstraintId,
747}
748
749impl PendingRelationActivationDeleteBarrier {
750 #[must_use]
751 pub(in crate::db) const fn accepted_schema_fingerprint(&self) -> CommitSchemaFingerprint {
752 self.accepted_schema_fingerprint
753 }
754
755 #[must_use]
756 pub(in crate::db) const fn source_entity_tag(&self) -> EntityTag {
757 self.source_entity_tag
758 }
759
760 #[must_use]
762 pub(in crate::db) const fn constraint_id(&self) -> ConstraintId {
763 self.constraint_id
764 }
765}
766
767pub struct SchemaStore {
776 backend: SchemaStoreBackend,
777 accepted_bundle_cache: RefCell<Option<AcceptedSchemaBundleCache>>,
778 accepted_catalog_scope: OnceCell<AcceptedStoreCatalogScope>,
779}
780
781struct AcceptedSchemaBundleCache {
782 selection: AcceptedSchemaRootSelection,
783 bundle: AcceptedSchemaRevisionBundle,
784 value_catalog: AcceptedValueCatalogHandle,
785 entity_selections: RefCell<StdBTreeMap<EntityTag, AcceptedCatalogSnapshotSelection>>,
786}
787
788enum SchemaStoreBackend {
789 Heap(StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>),
790 Journaled {
791 canonical:
792 StableBTreeMap<RawSchemaKey, RawSchemaSnapshot, VirtualMemory<DefaultMemoryImpl>>,
793 live: StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
794 tombstones: BTreeSet<RawSchemaKey>,
795 },
796}
797
798#[derive(Clone, Copy, Debug, Eq, PartialEq)]
800enum SchemaStoreVisit {
801 Continue,
802 #[cfg(test)]
803 Stop,
804}
805
806impl SchemaStoreVisit {
807 const fn should_stop(self) -> bool {
808 match self {
809 Self::Continue => false,
810 #[cfg(test)]
811 Self::Stop => true,
812 }
813 }
814}
815
816#[derive(Clone, Copy)]
817enum IdentityStateStorageView {
818 Effective,
819 Canonical,
820}
821
822#[derive(Clone, Copy)]
823enum IdentityStateWriteTarget {
824 Durable,
825 Materialized,
826 Canonical,
827}
828
829impl SchemaStore {
830 #[must_use]
832 pub const fn init_heap() -> Self {
833 Self {
834 backend: SchemaStoreBackend::Heap(StdBTreeMap::new()),
835 accepted_bundle_cache: RefCell::new(None),
836 accepted_catalog_scope: OnceCell::new(),
837 }
838 }
839
840 #[must_use]
845 pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
846 Self {
847 backend: SchemaStoreBackend::Journaled {
848 canonical: StableBTreeMap::init(memory),
849 live: StdBTreeMap::new(),
850 tombstones: BTreeSet::new(),
851 },
852 accepted_bundle_cache: RefCell::new(None),
853 accepted_catalog_scope: OnceCell::new(),
854 }
855 }
856
857 fn prepare_identity_state_transition(
858 &self,
859 incarnation: DatabaseIncarnationId,
860 candidate: &CandidateSchemaRevision,
861 view: IdentityStateStorageView,
862 ) -> Result<IdentityStateTransition, InternalError> {
863 let current = match view {
864 IdentityStateStorageView::Effective => self
865 .current_accepted_schema_bundle_ref()?
866 .as_ref()
867 .map(|bundle| (*bundle).clone()),
868 IdentityStateStorageView::Canonical => {
869 self.current_canonical_accepted_schema_bundle()?
870 }
871 };
872 let inventory = self.identity_state_inventory(view)?;
873 prepare_identity_state_transition(
874 incarnation,
875 current.as_ref(),
876 candidate.bundle(),
877 inventory,
878 )
879 }
880
881 fn validate_identity_state_closure(
882 &self,
883 bundle: &AcceptedSchemaRevisionBundle,
884 ) -> Result<(), InternalError> {
885 let inventory = self.identity_state_inventory(IdentityStateStorageView::Effective)?;
886 validate_identity_state_closure(bundle, &inventory)
887 }
888
889 pub(in crate::db) fn identity_statement_cursor(
891 &self,
892 database_incarnation_id: DatabaseIncarnationId,
893 entity_tag: EntityTag,
894 field_id: FieldId,
895 accepted_kind: &AcceptedFieldKind,
896 ) -> Result<IdentityStatementCursor, InternalError> {
897 let key = RawSchemaKey::from_identity_state(entity_tag, field_id);
898 let raw = self
899 .get_raw_snapshot(&key)
900 .ok_or_else(InternalError::identity_state_corruption)?;
901 let state = decode_identity_state(raw.as_bytes())?;
902 let owner = state.owner();
903 if owner.database_incarnation_id() != database_incarnation_id
904 || owner.entity_tag() != entity_tag
905 || owner.field_id() != field_id
906 || state.accepted_kind() != accepted_kind
907 || state.lifecycle() != IdentityStateLifecycle::Active
908 {
909 return Err(InternalError::identity_state_corruption());
910 }
911 IdentityStatementCursor::from_active_state(&state)
912 }
913
914 pub(in crate::db) fn identity_high_water_for_integrity(
916 &self,
917 database_incarnation_id: DatabaseIncarnationId,
918 entity_tag: EntityTag,
919 field_id: FieldId,
920 accepted_kind: &AcceptedFieldKind,
921 ) -> Result<u128, InternalError> {
922 let key = RawSchemaKey::from_identity_state(entity_tag, field_id);
923 let raw = self
924 .get_raw_snapshot(&key)
925 .ok_or_else(InternalError::identity_state_corruption)?;
926 let state = decode_identity_state(raw.as_bytes())?;
927 let owner = state.owner();
928 if owner.database_incarnation_id() != database_incarnation_id
929 || owner.entity_tag() != entity_tag
930 || owner.field_id() != field_id
931 || state.accepted_kind() != accepted_kind
932 || state.lifecycle() != IdentityStateLifecycle::Active
933 {
934 return Err(InternalError::identity_state_corruption());
935 }
936 Ok(state.materialized_high_water())
937 }
938
939 pub(in crate::db) fn preflight_identity_range_advance(
941 &self,
942 range: IdentityRangeAdvance,
943 ) -> Result<(), InternalError> {
944 let state =
945 self.identity_state_for_owner(range.owner(), IdentityStateStorageView::Effective)?;
946 if state.lifecycle() != IdentityStateLifecycle::Active
947 || range.new_high_water()
948 > identity_kind_maximum(state.accepted_kind())
949 .ok_or_else(InternalError::identity_state_corruption)?
950 {
951 return Err(InternalError::identity_state_corruption());
952 }
953 if state.materialized_high_water() != range.expected_high_water() {
954 return Err(InternalError::identity_state_conflict());
955 }
956 Ok(())
957 }
958
959 pub(in crate::db) fn apply_identity_range_advance(
961 &mut self,
962 range: IdentityRangeAdvance,
963 advance_id: IdentityAdvanceId,
964 ) -> Result<(), InternalError> {
965 self.apply_identity_range_advance_to(
966 range,
967 advance_id,
968 IdentityStateStorageView::Effective,
969 IdentityStateWriteTarget::Materialized,
970 )
971 }
972
973 pub(in crate::db) fn fold_identity_range_advance(
975 &mut self,
976 range: IdentityRangeAdvance,
977 advance_id: IdentityAdvanceId,
978 ) -> Result<(), InternalError> {
979 self.apply_identity_range_advance_to(
980 range,
981 advance_id,
982 IdentityStateStorageView::Canonical,
983 IdentityStateWriteTarget::Canonical,
984 )
985 }
986
987 pub(in crate::db) fn verify_identity_range_advance(
989 &self,
990 range: IdentityRangeAdvance,
991 advance_id: IdentityAdvanceId,
992 ) -> Result<(), InternalError> {
993 let state =
994 self.identity_state_for_owner(range.owner(), IdentityStateStorageView::Effective)?;
995 if state.materialized_high_water() != range.new_high_water()
996 || state.last_applied_advance() != Some(advance_id)
997 {
998 return Err(InternalError::recovery_effect_verification_failed());
999 }
1000 Ok(())
1001 }
1002
1003 pub(in crate::db) fn identity_range_commit_state(
1005 &self,
1006 range: IdentityRangeAdvance,
1007 advance_id: IdentityAdvanceId,
1008 canonical: bool,
1009 ) -> Result<IdentityRangeCommitState, InternalError> {
1010 let view = if canonical {
1011 IdentityStateStorageView::Canonical
1012 } else {
1013 IdentityStateStorageView::Effective
1014 };
1015 self.identity_state_for_owner(range.owner(), view)?
1016 .range_commit_state(range, advance_id)
1017 }
1018
1019 pub(in crate::db) fn identity_state_inventory_for_integrity(
1022 &self,
1023 incarnation: DatabaseIncarnationId,
1024 ) -> Result<Vec<IdentityState>, InternalError> {
1025 let has_accepted_bundle = self.current_accepted_schema_bundle_ref()?.is_some();
1026 let inventory = self.identity_state_inventory(IdentityStateStorageView::Effective)?;
1027 if !has_accepted_bundle && !inventory.is_empty() {
1028 return Err(InternalError::identity_state_corruption());
1029 }
1030 if inventory
1031 .values()
1032 .any(|state| state.owner().database_incarnation_id() != incarnation)
1033 {
1034 return Err(InternalError::identity_state_corruption());
1035 }
1036 Ok(inventory.into_values().collect())
1037 }
1038
1039 fn identity_state_for_owner(
1040 &self,
1041 owner: crate::db::schema::identity_state::IdentityStateOwner,
1042 view: IdentityStateStorageView,
1043 ) -> Result<IdentityState, InternalError> {
1044 let key = RawSchemaKey::from_identity_state(owner.entity_tag(), owner.field_id());
1045 let raw = match view {
1046 IdentityStateStorageView::Effective => self.get_raw_snapshot(&key),
1047 IdentityStateStorageView::Canonical => self.get_canonical_raw_value(&key)?,
1048 }
1049 .ok_or_else(InternalError::identity_state_corruption)?;
1050 let state = decode_identity_state(raw.as_bytes())?;
1051 if state.owner() != owner {
1052 return Err(InternalError::identity_state_corruption());
1053 }
1054 Ok(state)
1055 }
1056
1057 fn apply_identity_range_advance_to(
1058 &mut self,
1059 range: IdentityRangeAdvance,
1060 advance_id: IdentityAdvanceId,
1061 view: IdentityStateStorageView,
1062 target: IdentityStateWriteTarget,
1063 ) -> Result<(), InternalError> {
1064 let state = self.identity_state_for_owner(range.owner(), view)?;
1065 let advanced = state.apply_range_advance(range, advance_id)?;
1066 let key = RawSchemaKey::from_identity_state(
1067 advanced.owner().entity_tag(),
1068 advanced.owner().field_id(),
1069 );
1070 let bytes = encode_identity_state(&advanced)?;
1071 match target {
1072 IdentityStateWriteTarget::Materialized => {
1073 self.insert_raw_snapshot(
1074 key,
1075 RawSchemaSnapshot::from_encoded_control_record(bytes),
1076 );
1077 }
1078 IdentityStateWriteTarget::Canonical => {
1079 self.insert_canonical_raw_value(key, bytes)?;
1080 }
1081 IdentityStateWriteTarget::Durable => {
1082 return Err(InternalError::store_invariant());
1083 }
1084 }
1085 Ok(())
1086 }
1087
1088 fn identity_state_inventory(
1089 &self,
1090 view: IdentityStateStorageView,
1091 ) -> Result<IdentityStateInventory, InternalError> {
1092 let bounds = RawSchemaKey::all_identity_state_range_bounds();
1093 let mut inventory = StdBTreeMap::new();
1094 let mut collect = |key: &RawSchemaKey,
1095 raw: &RawSchemaSnapshot|
1096 -> Result<SchemaStoreVisit, InternalError> {
1097 if inventory.len() >= MAX_IDENTITY_STATE_RECORDS_PER_DATABASE {
1098 return Err(InternalError::identity_state_corruption());
1099 }
1100 let state = decode_identity_state(raw.as_bytes())?;
1101 let state_key = (key.entity_tag(), FieldId::new(key.version()));
1102 if !key.is_identity_state()
1103 || state.owner().entity_tag() != state_key.0
1104 || state.owner().field_id() != state_key.1
1105 || inventory.insert(state_key, state).is_some()
1106 {
1107 return Err(InternalError::identity_state_corruption());
1108 }
1109 Ok(SchemaStoreVisit::Continue)
1110 };
1111
1112 match (&self.backend, view) {
1113 (SchemaStoreBackend::Heap(map), IdentityStateStorageView::Effective) => {
1114 for (key, raw) in map.range((bounds.0, bounds.1)) {
1115 collect(key, raw)?;
1116 }
1117 }
1118 (
1119 SchemaStoreBackend::Journaled {
1120 canonical,
1121 live,
1122 tombstones,
1123 },
1124 IdentityStateStorageView::Effective,
1125 ) => Self::visit_journaled_raw_snapshot_range(
1126 canonical,
1127 live,
1128 tombstones,
1129 bounds,
1130 Direction::Asc,
1131 &mut collect,
1132 )?,
1133 (
1134 SchemaStoreBackend::Journaled { canonical, .. },
1135 IdentityStateStorageView::Canonical,
1136 ) => {
1137 for entry in canonical.range((bounds.0, bounds.1)) {
1138 collect(entry.key(), &entry.value())?;
1139 }
1140 }
1141 (SchemaStoreBackend::Heap(_), IdentityStateStorageView::Canonical) => {
1142 return Err(InternalError::store_invariant());
1143 }
1144 }
1145
1146 Ok(inventory)
1147 }
1148
1149 fn apply_identity_state_transition(
1150 &mut self,
1151 transition: IdentityStateTransition,
1152 target: IdentityStateWriteTarget,
1153 ) -> Result<(), InternalError> {
1154 for state in transition.into_updates() {
1155 let key = RawSchemaKey::from_identity_state(
1156 state.owner().entity_tag(),
1157 state.owner().field_id(),
1158 );
1159 let bytes = encode_identity_state(&state)?;
1160 match target {
1161 IdentityStateWriteTarget::Durable => {
1162 self.insert_durable_raw_value(key, bytes);
1163 }
1164 IdentityStateWriteTarget::Materialized => {
1165 self.insert_raw_snapshot(
1166 key,
1167 RawSchemaSnapshot::from_encoded_control_record(bytes),
1168 );
1169 }
1170 IdentityStateWriteTarget::Canonical => {
1171 self.insert_canonical_raw_value(key, bytes)?;
1172 }
1173 }
1174 }
1175 Ok(())
1176 }
1177
1178 fn current_canonical_accepted_schema_bundle(
1179 &self,
1180 ) -> Result<Option<AcceptedSchemaRevisionBundle>, InternalError> {
1181 let first = self.canonical_root_slot_bytes(0)?;
1182 let second = self.canonical_root_slot_bytes(1)?;
1183 let Some(selection) =
1184 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1185 else {
1186 return Ok(None);
1187 };
1188 let bundle_key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
1189 let raw = self
1190 .get_canonical_raw_value(&bundle_key)?
1191 .ok_or_else(InternalError::store_corruption)?;
1192 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes()).map(Some)
1193 }
1194
1195 pub(in crate::db) fn insert_persisted_snapshot(
1197 &mut self,
1198 entity: EntityTag,
1199 snapshot: &PersistedSchemaSnapshot,
1200 ) -> Result<(), InternalError> {
1201 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
1202 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1203 let _ = self.insert_raw_snapshot(key, raw_snapshot);
1204
1205 Ok(())
1206 }
1207
1208 pub(in crate::db) fn constraint_validation_job(
1210 &self,
1211 entity: EntityTag,
1212 constraint_id: ConstraintId,
1213 ) -> Result<Option<ConstraintValidationJob>, InternalError> {
1214 let key = RawSchemaKey::from_constraint_validation_job(entity, constraint_id);
1215 self.get_raw_snapshot(&key)
1216 .map(|raw| decode_constraint_validation_job(raw.as_bytes()))
1217 .transpose()
1218 }
1219
1220 pub(in crate::db) fn apply_constraint_validation_job(
1222 &mut self,
1223 job: &ConstraintValidationJob,
1224 ) -> Result<(), InternalError> {
1225 let key =
1226 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id());
1227 let bytes = encode_constraint_validation_job(job)?;
1228 let _ =
1229 self.insert_raw_snapshot(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
1230 Ok(())
1231 }
1232
1233 #[expect(
1235 clippy::unnecessary_wraps,
1236 reason = "marker apply operations share one fallible callback contract"
1237 )]
1238 pub(in crate::db) fn apply_constraint_validation_job_removal(
1239 &mut self,
1240 entity: EntityTag,
1241 constraint_id: ConstraintId,
1242 ) -> Result<(), InternalError> {
1243 let key = RawSchemaKey::from_constraint_validation_job(entity, constraint_id);
1244 match &mut self.backend {
1245 SchemaStoreBackend::Heap(map) => {
1246 map.remove(&key);
1247 }
1248 SchemaStoreBackend::Journaled {
1249 live, tombstones, ..
1250 } => {
1251 live.remove(&key);
1252 tombstones.insert(key);
1253 }
1254 }
1255 Ok(())
1256 }
1257
1258 pub(in crate::db) fn fold_constraint_validation_job(
1260 &mut self,
1261 job: &ConstraintValidationJob,
1262 ) -> Result<(), InternalError> {
1263 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1264 return Err(InternalError::store_invariant());
1265 };
1266 let key =
1267 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id());
1268 let bytes = encode_constraint_validation_job(job)?;
1269 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
1270 Ok(())
1271 }
1272
1273 pub(in crate::db) fn fold_constraint_validation_job_removal(
1275 &mut self,
1276 entity: EntityTag,
1277 constraint_id: ConstraintId,
1278 ) -> Result<(), InternalError> {
1279 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1280 return Err(InternalError::store_invariant());
1281 };
1282 canonical.remove(&RawSchemaKey::from_constraint_validation_job(
1283 entity,
1284 constraint_id,
1285 ));
1286 Ok(())
1287 }
1288
1289 pub(in crate::db) fn reset_journaled_live_projection(&mut self) -> Result<(), InternalError> {
1292 let SchemaStoreBackend::Journaled {
1293 live, tombstones, ..
1294 } = &mut self.backend
1295 else {
1296 return Err(InternalError::store_invariant());
1297 };
1298
1299 live.clear();
1300 tombstones.clear();
1301 self.accepted_bundle_cache.get_mut().take();
1302
1303 Ok(())
1304 }
1305
1306 pub(in crate::db) fn fold_persisted_snapshot(
1308 &mut self,
1309 entity: EntityTag,
1310 snapshot: &PersistedSchemaSnapshot,
1311 ) -> Result<(), InternalError> {
1312 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1313 return Err(InternalError::store_invariant());
1314 };
1315
1316 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
1317 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1318 canonical.insert(key, raw_snapshot);
1319
1320 Ok(())
1321 }
1322
1323 pub(in crate::db) fn current_accepted_schema_root(
1325 &self,
1326 ) -> Result<Option<AcceptedSchemaRootSelection>, InternalError> {
1327 let first = self.accepted_root_slot_bytes(0)?;
1328 let second = self.accepted_root_slot_bytes(1)?;
1329 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])
1330 }
1331
1332 pub(in crate::db) fn current_accepted_schema_bundle(
1334 &self,
1335 ) -> Result<Option<AcceptedSchemaRevisionBundle>, InternalError> {
1336 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1337 return Ok(None);
1338 };
1339 self.validate_constraint_validation_job_closure(&bundle)?;
1340 Ok(Some(bundle.clone()))
1341 }
1342
1343 pub(in crate::db) fn current_accepted_runtime_entities(
1345 &self,
1346 registered_store_path: &'static str,
1347 ) -> Result<Vec<AcceptedRuntimeEntity>, InternalError> {
1348 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1349 return Ok(Vec::new());
1350 };
1351 if bundle.store_path() != registered_store_path {
1352 return Err(InternalError::store_corruption());
1353 }
1354
1355 bundle
1356 .entity_snapshots()
1357 .iter()
1358 .map(|(entity_tag, snapshot)| {
1359 AcceptedRuntimeEntity::from_accepted_snapshot(
1360 &bundle,
1361 *entity_tag,
1362 snapshot,
1363 registered_store_path,
1364 )
1365 })
1366 .collect()
1367 }
1368
1369 pub(in crate::db) fn current_accepted_runtime_entity_for_tag(
1371 &self,
1372 registered_store_path: &'static str,
1373 entity_tag: EntityTag,
1374 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1375 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1376 return Ok(None);
1377 };
1378 if bundle.store_path() != registered_store_path {
1379 return Err(InternalError::store_corruption());
1380 }
1381 let Some(snapshot) = bundle.entity_snapshots().get(&entity_tag) else {
1382 return Ok(None);
1383 };
1384
1385 AcceptedRuntimeEntity::from_accepted_snapshot(
1386 &bundle,
1387 entity_tag,
1388 snapshot,
1389 registered_store_path,
1390 )
1391 .map(Some)
1392 }
1393
1394 pub(in crate::db) fn current_accepted_runtime_entity_for_path(
1396 &self,
1397 registered_store_path: &'static str,
1398 entity_path: &str,
1399 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1400 self.current_accepted_runtime_entity_matching(registered_store_path, |snapshot_path, _| {
1401 snapshot_path == entity_path
1402 })
1403 }
1404
1405 #[cfg(test)]
1407 pub(in crate::db) fn current_accepted_runtime_entity_for_name(
1408 &self,
1409 registered_store_path: &'static str,
1410 entity_name: &str,
1411 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1412 self.current_accepted_runtime_entity_matching(registered_store_path, |_, snapshot_name| {
1413 snapshot_name == entity_name
1414 })
1415 }
1416
1417 fn current_accepted_runtime_entity_matching(
1418 &self,
1419 registered_store_path: &'static str,
1420 mut predicate: impl FnMut(&str, &str) -> bool,
1421 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1422 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1423 return Ok(None);
1424 };
1425 if bundle.store_path() != registered_store_path {
1426 return Err(InternalError::store_corruption());
1427 }
1428
1429 let mut matched = None;
1430 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1431 if !predicate(snapshot.entity_path(), snapshot.entity_name()) {
1432 continue;
1433 }
1434 let entity = AcceptedRuntimeEntity::from_accepted_snapshot(
1435 &bundle,
1436 *entity_tag,
1437 snapshot,
1438 registered_store_path,
1439 )?;
1440 if matched.replace(entity).is_some() {
1441 return Err(InternalError::store_corruption());
1442 }
1443 }
1444
1445 Ok(matched)
1446 }
1447
1448 pub(in crate::db) fn current_accepted_schema_revision(
1450 &self,
1451 ) -> Result<Option<AcceptedSchemaRevision>, InternalError> {
1452 Ok(self
1453 .current_accepted_schema_root()?
1454 .map(|selection| selection.root().revision()))
1455 }
1456
1457 pub(in crate::db) fn pending_relation_activation_for_target(
1463 &self,
1464 target_path: &str,
1465 ) -> Result<Option<PendingRelationActivationDeleteBarrier>, InternalError> {
1466 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1467 return Ok(None);
1468 };
1469 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1470 let Some(candidate) = snapshot
1471 .candidate_relations()
1472 .iter()
1473 .find(|candidate| candidate.target_path() == target_path)
1474 else {
1475 continue;
1476 };
1477 let activation = snapshot
1478 .constraint_activations()
1479 .iter()
1480 .find(|activation| {
1481 matches!(
1482 activation.kind(),
1483 ConstraintActivationKind::Relation { relation_id }
1484 if *relation_id == candidate.id()
1485 )
1486 })
1487 .ok_or_else(InternalError::store_corruption)?;
1488 return Ok(Some(PendingRelationActivationDeleteBarrier {
1489 accepted_schema_fingerprint:
1490 accepted_schema_cache_fingerprint_for_persisted_snapshot(snapshot)?,
1491 source_entity_tag: *entity_tag,
1492 constraint_id: activation.id(),
1493 }));
1494 }
1495
1496 Ok(None)
1497 }
1498
1499 pub(in crate::db) fn entity_has_relation_to_target(
1501 &self,
1502 source_entity: EntityTag,
1503 target_path: &str,
1504 ) -> Result<bool, InternalError> {
1505 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1506 return Ok(false);
1507 };
1508 let Some(snapshot) = bundle.entity_snapshots().get(&source_entity) else {
1509 return Ok(false);
1510 };
1511
1512 Ok(snapshot
1513 .relations()
1514 .iter()
1515 .any(|relation| relation.target_path() == target_path))
1516 }
1517
1518 pub(in crate::db) fn validate_live_activation_transition(
1520 &self,
1521 candidate: &AcceptedSchemaRevisionBundle,
1522 ) -> Result<(), InternalError> {
1523 let Some(current) = self.current_accepted_schema_bundle()? else {
1524 return Ok(());
1525 };
1526 for (entity_tag, before) in current.entity_snapshots() {
1527 if before.constraint_activations().is_empty() {
1528 continue;
1529 }
1530 let after = candidate
1531 .entity_snapshots()
1532 .get(entity_tag)
1533 .ok_or_else(InternalError::store_invariant)?;
1534 if before == after {
1535 continue;
1536 }
1537 let expected_shape = before
1538 .clone()
1539 .with_constraint_catalog(after.constraint_catalog().clone());
1540 let catalog_only_transition = expected_shape == *after
1541 && before
1542 .constraint_catalog()
1543 .permits_live_activation_transition_to(after.constraint_catalog());
1544 let sql_row_local_abort_with_version =
1545 before.constraint_activations().iter().any(|activation| {
1546 activation.origin() == ConstraintOrigin::SqlDdl
1547 && matches!(
1548 activation.kind(),
1549 ConstraintActivationKind::Check { .. }
1550 | ConstraintActivationKind::NotNull { .. }
1551 )
1552 && before.version().get().checked_add(1) == Some(after.version().get())
1553 && before
1554 .constraint_catalog()
1555 .clone()
1556 .with_aborted_activation(activation.id())
1557 .is_ok_and(|catalog| catalog == *after.constraint_catalog())
1558 && before
1559 .clone()
1560 .with_constraint_catalog(after.constraint_catalog().clone())
1561 .with_schema_version(after.version())
1562 == *after
1563 });
1564 let sql_unique_abort_with_version =
1565 before.constraint_activations().iter().any(|activation| {
1566 activation.origin() == ConstraintOrigin::SqlDdl
1567 && matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
1568 && before.version().get().checked_add(1) == Some(after.version().get())
1569 && before
1570 .with_aborted_unique_activation(activation.id(), after.version())
1571 .is_ok_and(|expected| expected == *after)
1572 });
1573 let not_null_promotion = before.constraint_activations().iter().any(|activation| {
1574 matches!(activation.kind(), ConstraintActivationKind::NotNull { .. })
1575 && before
1576 .with_promoted_not_null_activation(activation.id(), after.version())
1577 .is_ok_and(|expected| expected == *after)
1578 });
1579 let unique_promotion = before.constraint_activations().iter().any(|activation| {
1580 matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
1581 && before
1582 .with_promoted_unique_activation(activation.id(), after.version())
1583 .is_ok_and(|expected| expected == *after)
1584 });
1585 let relation_promotion = before.constraint_activations().iter().any(|activation| {
1586 matches!(activation.kind(), ConstraintActivationKind::Relation { .. })
1587 && before
1588 .with_promoted_relation_activation(activation.id(), after.version())
1589 .is_ok_and(|expected| expected == *after)
1590 });
1591 if !catalog_only_transition
1592 && !sql_row_local_abort_with_version
1593 && !sql_unique_abort_with_version
1594 && !not_null_promotion
1595 && !unique_promotion
1596 && !relation_promotion
1597 {
1598 return Err(InternalError::store_invariant());
1599 }
1600 }
1601 Ok(())
1602 }
1603
1604 pub(in crate::db) fn validate_constraint_validation_job_closure(
1606 &self,
1607 bundle: &AcceptedSchemaRevisionBundle,
1608 ) -> Result<(), InternalError> {
1609 self.validate_constraint_validation_job_closure_with_change(bundle, None, None)
1610 }
1611
1612 pub(in crate::db) fn validate_constraint_validation_job_closure_with_change(
1615 &self,
1616 bundle: &AcceptedSchemaRevisionBundle,
1617 replacement: Option<&ConstraintValidationJob>,
1618 removal: Option<(EntityTag, ConstraintId)>,
1619 ) -> Result<(), InternalError> {
1620 if replacement.is_some() && removal.is_some() {
1621 return Err(InternalError::store_invariant());
1622 }
1623 let replacement_key = replacement.map(|job| {
1624 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id())
1625 });
1626 let removal_key = removal.map(|(entity_tag, constraint_id)| {
1627 RawSchemaKey::from_constraint_validation_job(entity_tag, constraint_id)
1628 });
1629 let mut expected = BTreeSet::new();
1630 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1631 for activation in snapshot.constraint_activations() {
1632 let key =
1633 RawSchemaKey::from_constraint_validation_job(*entity_tag, activation.id());
1634 match activation.state() {
1635 ConstraintActivationState::EnforcingNewWrites => {
1636 if self
1637 .constraint_validation_job_after_change(
1638 key,
1639 replacement,
1640 replacement_key,
1641 removal_key,
1642 )?
1643 .is_some()
1644 {
1645 return Err(InternalError::store_corruption());
1646 }
1647 }
1648 ConstraintActivationState::Validating => {
1649 let job = self
1650 .constraint_validation_job_after_change(
1651 key,
1652 replacement,
1653 replacement_key,
1654 removal_key,
1655 )?
1656 .ok_or_else(InternalError::store_corruption)?;
1657 if job.entity_tag() != *entity_tag
1658 || job.entity_path() != snapshot.entity_path()
1659 {
1660 return Err(InternalError::store_corruption());
1661 }
1662 job.validate(Some(activation))?;
1663 expected.insert(key);
1664 }
1665 }
1666 }
1667 }
1668
1669 self.visit_constraint_validation_jobs(|key, raw| {
1670 if removal_key == Some(*key) || replacement_key == Some(*key) {
1671 return Ok(SchemaStoreVisit::Continue);
1672 }
1673 if !expected.contains(key) {
1674 return Err(InternalError::store_corruption());
1675 }
1676 let job = decode_constraint_validation_job(raw.as_bytes())?;
1677 if job.entity_tag() != key.entity_tag()
1678 || key.constraint_id() != Some(job.constraint_id())
1679 {
1680 return Err(InternalError::store_corruption());
1681 }
1682 Ok(SchemaStoreVisit::Continue)
1683 })?;
1684
1685 if let Some(key) = replacement_key
1686 && !expected.contains(&key)
1687 {
1688 return Err(InternalError::store_corruption());
1689 }
1690 if let Some(key) = removal_key
1691 && expected.contains(&key)
1692 {
1693 return Err(InternalError::store_corruption());
1694 }
1695
1696 Ok(())
1697 }
1698
1699 fn constraint_validation_job_after_change(
1700 &self,
1701 key: RawSchemaKey,
1702 replacement: Option<&ConstraintValidationJob>,
1703 replacement_key: Option<RawSchemaKey>,
1704 removal_key: Option<RawSchemaKey>,
1705 ) -> Result<Option<ConstraintValidationJob>, InternalError> {
1706 if removal_key == Some(key) {
1707 return Ok(None);
1708 }
1709 if replacement_key == Some(key) {
1710 return Ok(replacement.cloned());
1711 }
1712 self.get_raw_snapshot(&key)
1713 .map(|raw| decode_constraint_validation_job(raw.as_bytes()))
1714 .transpose()
1715 }
1716
1717 pub(in crate::db) fn current_accepted_schema_authority_matches(
1720 &self,
1721 expected: &AcceptedSchemaAuthority,
1722 ) -> Result<bool, InternalError> {
1723 let Some(store_scope) = self.accepted_catalog_scope.get() else {
1724 return Ok(false);
1725 };
1726
1727 if let Some(cached) = self
1730 .accepted_bundle_cache
1731 .try_borrow()
1732 .map_err(|_| InternalError::store_invariant())?
1733 .as_ref()
1734 {
1735 let root = cached.selection.root();
1736 return Ok(expected.matches_store_root(
1737 store_scope,
1738 root.revision(),
1739 root.fingerprint(),
1740 ));
1741 }
1742
1743 let Some(selection) = self.current_accepted_schema_root()? else {
1744 return Ok(false);
1745 };
1746 let root = selection.root();
1747
1748 Ok(expected.matches_store_root(store_scope, root.revision(), root.fingerprint()))
1749 }
1750
1751 pub(in crate::db) fn publish_accepted_schema_candidate(
1757 &mut self,
1758 incarnation: DatabaseIncarnationId,
1759 expected_revision: AcceptedSchemaRevision,
1760 candidate: &CandidateSchemaRevision,
1761 ) -> Result<(), InternalError> {
1762 let identity_transition = self.prepare_identity_state_transition(
1763 incarnation,
1764 candidate,
1765 IdentityStateStorageView::Effective,
1766 )?;
1767 if self.current_root_matches_candidate(candidate)? {
1768 if !identity_transition.is_empty() {
1769 return Err(InternalError::identity_state_corruption());
1770 }
1771 let selection = self
1772 .current_accepted_schema_root()?
1773 .ok_or_else(InternalError::store_corruption)?;
1774 self.retain_durable_candidate_entries(candidate, selection.slot())?;
1775 return Ok(());
1776 }
1777 let first = self.accepted_root_slot_bytes(0)?;
1778 let second = self.accepted_root_slot_bytes(1)?;
1779 prepare_accepted_schema_root_publication(
1780 [first.as_deref(), second.as_deref()],
1781 expected_revision,
1782 candidate,
1783 )
1784 .map_err(map_schema_publication_error)?;
1785
1786 self.insert_durable_candidate_snapshots(candidate)?;
1787 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1788 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
1789 let persisted_bundle = self
1790 .get_raw_snapshot(&bundle_key)
1791 .ok_or_else(InternalError::store_corruption)?;
1792 let _verified = decode_verified_accepted_schema_revision_bundle(
1793 candidate.root(),
1794 persisted_bundle.as_bytes(),
1795 )?;
1796 self.apply_identity_state_transition(
1797 identity_transition,
1798 IdentityStateWriteTarget::Durable,
1799 )?;
1800
1801 let first = self.accepted_root_slot_bytes(0)?;
1804 let second = self.accepted_root_slot_bytes(1)?;
1805 let publication = prepare_accepted_schema_root_publication(
1806 [first.as_deref(), second.as_deref()],
1807 expected_revision,
1808 candidate,
1809 )
1810 .map_err(map_schema_publication_error)?;
1811 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
1812 self.insert_durable_raw_value(root_key, publication.encoded_root().to_vec());
1813
1814 let selected = self
1815 .current_accepted_schema_root()?
1816 .ok_or_else(InternalError::store_corruption)?;
1817 if selected.root() != candidate.root() {
1818 return Err(InternalError::store_corruption());
1819 }
1820 self.retain_durable_candidate_entries(candidate, selected.slot())?;
1821 Ok(())
1822 }
1823
1824 pub(in crate::db) fn restore_live_accepted_schema_checkpoint(
1827 &mut self,
1828 incarnation: DatabaseIncarnationId,
1829 candidate: &CandidateSchemaRevision,
1830 checkpoint_identity_states: &IdentityStateInventory,
1831 ) -> Result<(), InternalError> {
1832 if !matches!(self.backend, SchemaStoreBackend::Heap(_)) {
1833 return Err(InternalError::store_invariant());
1834 }
1835 let checkpoint_validation = prepare_identity_state_transition(
1836 incarnation,
1837 Some(candidate.bundle()),
1838 candidate.bundle(),
1839 checkpoint_identity_states.clone(),
1840 )?;
1841 if !checkpoint_validation.is_empty() {
1842 return Err(InternalError::identity_state_corruption());
1843 }
1844 if self.current_root_matches_candidate(candidate)? {
1845 for state in checkpoint_identity_states.values() {
1846 let key = RawSchemaKey::from_identity_state(
1847 state.owner().entity_tag(),
1848 state.owner().field_id(),
1849 );
1850 self.insert_durable_raw_value(key, encode_identity_state(state)?);
1851 }
1852 if self.identity_state_inventory(IdentityStateStorageView::Effective)?
1853 != *checkpoint_identity_states
1854 {
1855 return Err(InternalError::identity_state_corruption());
1856 }
1857 let selection = self
1858 .current_accepted_schema_root()?
1859 .ok_or_else(InternalError::store_corruption)?;
1860 self.retain_durable_candidate_entries(candidate, selection.slot())?;
1861 return Ok(());
1862 }
1863 if self.current_accepted_schema_root()?.is_some()
1864 || !self
1865 .identity_state_inventory(IdentityStateStorageView::Effective)?
1866 .is_empty()
1867 {
1868 return Err(InternalError::store_corruption());
1869 }
1870
1871 self.insert_durable_candidate_snapshots(candidate)?;
1872 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1873 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
1874 for state in checkpoint_identity_states.values() {
1875 let key = RawSchemaKey::from_identity_state(
1876 state.owner().entity_tag(),
1877 state.owner().field_id(),
1878 );
1879 self.insert_durable_raw_value(key, encode_identity_state(state)?);
1880 }
1881 let root_key = RawSchemaKey::from_accepted_root_slot(0)?;
1882 self.insert_durable_raw_value(root_key, candidate.encoded_root().to_vec());
1883
1884 let selected = self
1885 .current_accepted_schema_root()?
1886 .ok_or_else(InternalError::store_corruption)?;
1887 if selected.root() != candidate.root() {
1888 return Err(InternalError::store_corruption());
1889 }
1890 self.retain_durable_candidate_entries(candidate, selected.slot())?;
1891 Ok(())
1892 }
1893
1894 pub(in crate::db) fn preflight_accepted_schema_candidate(
1901 &self,
1902 incarnation: DatabaseIncarnationId,
1903 expected_revision: AcceptedSchemaRevision,
1904 candidate: &CandidateSchemaRevision,
1905 ) -> Result<bool, InternalError> {
1906 let identity_transition = self.prepare_identity_state_transition(
1907 incarnation,
1908 candidate,
1909 IdentityStateStorageView::Effective,
1910 )?;
1911 if self.current_root_matches_candidate(candidate)? {
1912 if !identity_transition.is_empty() {
1913 return Err(InternalError::identity_state_corruption());
1914 }
1915 return Ok(true);
1916 }
1917 let first = self.accepted_root_slot_bytes(0)?;
1918 let second = self.accepted_root_slot_bytes(1)?;
1919 prepare_accepted_schema_root_publication(
1920 [first.as_deref(), second.as_deref()],
1921 expected_revision,
1922 candidate,
1923 )
1924 .map_err(map_schema_publication_error)?;
1925
1926 Ok(false)
1927 }
1928
1929 pub(in crate::db) fn projected_identity_state_count(
1931 &self,
1932 incarnation: DatabaseIncarnationId,
1933 candidate: &CandidateSchemaRevision,
1934 ) -> Result<usize, InternalError> {
1935 Ok(self
1936 .prepare_identity_state_transition(
1937 incarnation,
1938 candidate,
1939 IdentityStateStorageView::Effective,
1940 )?
1941 .projected_inventory_len())
1942 }
1943
1944 pub(in crate::db) fn apply_journaled_accepted_schema_candidate(
1946 &mut self,
1947 incarnation: DatabaseIncarnationId,
1948 expected_revision: AcceptedSchemaRevision,
1949 candidate: &CandidateSchemaRevision,
1950 ) -> Result<(), InternalError> {
1951 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1952 return Err(InternalError::store_invariant());
1953 }
1954 let identity_transition = self.prepare_identity_state_transition(
1955 incarnation,
1956 candidate,
1957 IdentityStateStorageView::Effective,
1958 )?;
1959 if self.current_root_matches_candidate(candidate)? {
1960 if !identity_transition.is_empty() {
1961 return Err(InternalError::identity_state_corruption());
1962 }
1963 let selection = self
1964 .current_accepted_schema_root()?
1965 .ok_or_else(InternalError::store_corruption)?;
1966 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
1967 return Ok(());
1968 }
1969
1970 let first = self.accepted_root_slot_bytes(0)?;
1971 let second = self.accepted_root_slot_bytes(1)?;
1972 prepare_accepted_schema_root_publication(
1973 [first.as_deref(), second.as_deref()],
1974 expected_revision,
1975 candidate,
1976 )
1977 .map_err(map_schema_publication_error)?;
1978
1979 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
1980 self.insert_persisted_snapshot(*entity_tag, snapshot)?;
1981 }
1982 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1983 self.insert_raw_snapshot(
1984 bundle_key,
1985 RawSchemaSnapshot::from_encoded_control_record(candidate.encoded_bundle().to_vec()),
1986 );
1987 let persisted_bundle = self
1988 .get_raw_snapshot(&bundle_key)
1989 .ok_or_else(InternalError::store_corruption)?;
1990 let _verified = decode_verified_accepted_schema_revision_bundle(
1991 candidate.root(),
1992 persisted_bundle.as_bytes(),
1993 )?;
1994 self.apply_identity_state_transition(
1995 identity_transition,
1996 IdentityStateWriteTarget::Materialized,
1997 )?;
1998
1999 let first = self.accepted_root_slot_bytes(0)?;
2000 let second = self.accepted_root_slot_bytes(1)?;
2001 let publication = prepare_accepted_schema_root_publication(
2002 [first.as_deref(), second.as_deref()],
2003 expected_revision,
2004 candidate,
2005 )
2006 .map_err(map_schema_publication_error)?;
2007 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
2008 self.insert_raw_snapshot(
2009 root_key,
2010 RawSchemaSnapshot::from_encoded_control_record(publication.encoded_root().to_vec()),
2011 );
2012
2013 if !self.current_root_matches_candidate(candidate)? {
2014 return Err(InternalError::store_corruption());
2015 }
2016 let selection = self
2017 .current_accepted_schema_root()?
2018 .ok_or_else(InternalError::store_corruption)?;
2019 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
2020 Ok(())
2021 }
2022
2023 pub(in crate::db) fn fold_journaled_accepted_schema_candidate(
2025 &mut self,
2026 incarnation: DatabaseIncarnationId,
2027 expected_revision: AcceptedSchemaRevision,
2028 candidate: &CandidateSchemaRevision,
2029 ) -> Result<(), InternalError> {
2030 let identity_transition = self.prepare_identity_state_transition(
2031 incarnation,
2032 candidate,
2033 IdentityStateStorageView::Canonical,
2034 )?;
2035 if self.canonical_root_matches_candidate(candidate)? {
2036 if !identity_transition.is_empty() {
2037 return Err(InternalError::identity_state_corruption());
2038 }
2039 let first = self.canonical_root_slot_bytes(0)?;
2040 let second = self.canonical_root_slot_bytes(1)?;
2041 let selection =
2042 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2043 .ok_or_else(InternalError::store_corruption)?;
2044 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
2045 return Ok(());
2046 }
2047
2048 let first = self.canonical_root_slot_bytes(0)?;
2049 let second = self.canonical_root_slot_bytes(1)?;
2050 prepare_accepted_schema_root_publication(
2051 [first.as_deref(), second.as_deref()],
2052 expected_revision,
2053 candidate,
2054 )
2055 .map_err(map_schema_publication_error)?;
2056
2057 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2058 self.fold_persisted_snapshot(*entity_tag, snapshot)?;
2059 }
2060 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2061 self.insert_canonical_raw_value(bundle_key, candidate.encoded_bundle().to_vec())?;
2062 let persisted_bundle = self
2063 .get_canonical_raw_value(&bundle_key)?
2064 .ok_or_else(InternalError::store_corruption)?;
2065 let _verified = decode_verified_accepted_schema_revision_bundle(
2066 candidate.root(),
2067 persisted_bundle.as_bytes(),
2068 )?;
2069 self.apply_identity_state_transition(
2070 identity_transition,
2071 IdentityStateWriteTarget::Canonical,
2072 )?;
2073
2074 let first = self.canonical_root_slot_bytes(0)?;
2075 let second = self.canonical_root_slot_bytes(1)?;
2076 let publication = prepare_accepted_schema_root_publication(
2077 [first.as_deref(), second.as_deref()],
2078 expected_revision,
2079 candidate,
2080 )
2081 .map_err(map_schema_publication_error)?;
2082 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
2083 self.insert_canonical_raw_value(root_key, publication.encoded_root().to_vec())?;
2084
2085 if !self.canonical_root_matches_candidate(candidate)? {
2086 return Err(InternalError::store_corruption());
2087 }
2088 let first = self.canonical_root_slot_bytes(0)?;
2089 let second = self.canonical_root_slot_bytes(1)?;
2090 let selection = select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2091 .ok_or_else(InternalError::store_corruption)?;
2092 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
2093 Ok(())
2094 }
2095
2096 pub(in crate::db) fn get_persisted_snapshot(
2098 &self,
2099 entity: EntityTag,
2100 version: SchemaVersion,
2101 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2102 let key = RawSchemaKey::from_entity_version(entity, version);
2103 self.get_raw_snapshot(&key)
2104 .map(|snapshot| snapshot.decode_persisted_snapshot())
2105 .transpose()
2106 }
2107
2108 #[cfg(test)]
2109 fn latest_staged_persisted_snapshot(
2110 &self,
2111 entity: EntityTag,
2112 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2113 self.latest_raw_snapshots_by_entity()
2114 .remove(&entity)
2115 .map(|(_, snapshot)| snapshot.decode_persisted_snapshot())
2116 .transpose()
2117 }
2118
2119 pub(in crate::db) fn current_accepted_persisted_snapshot(
2122 &self,
2123 entity: EntityTag,
2124 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2125 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2126 return Ok(None);
2127 };
2128
2129 Ok(bundle.entity_snapshots().get(&entity).cloned())
2130 }
2131
2132 pub(in crate::db) fn current_accepted_catalog_selection(
2134 &self,
2135 entity: EntityTag,
2136 entity_path: &str,
2137 store_path: &'static str,
2138 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
2139 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2140 return Ok(None);
2141 };
2142 if bundle.store_path() != store_path {
2143 return Err(InternalError::store_corruption());
2144 }
2145 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
2146 return Ok(None);
2147 };
2148 if snapshot.entity_path() != entity_path {
2149 return Err(InternalError::store_corruption());
2150 }
2151
2152 let cache = self
2153 .accepted_bundle_cache
2154 .try_borrow()
2155 .map_err(|_| InternalError::store_invariant())?;
2156 let cached = cache.as_ref().ok_or_else(InternalError::store_invariant)?;
2157 if let Some(selection) = cached
2158 .entity_selections
2159 .try_borrow()
2160 .map_err(|_| InternalError::store_invariant())?
2161 .get(&entity)
2162 .cloned()
2163 {
2164 return Ok(Some(selection));
2165 }
2166
2167 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2168 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
2169 let identity = AcceptedCatalogIdentity::new(
2170 entity,
2171 entity_path,
2172 store_path,
2173 bundle.revision(),
2174 snapshot.version(),
2175 fingerprint,
2176 );
2177
2178 let selected = AcceptedCatalogSnapshotSelection::new(
2179 identity,
2180 cached.value_catalog.clone(),
2181 Rc::from(raw_snapshot.into_bytes()),
2182 );
2183 cached
2184 .entity_selections
2185 .try_borrow_mut()
2186 .map_err(|_| InternalError::store_invariant())?
2187 .insert(entity, selected.clone());
2188
2189 Ok(Some(selected))
2190 }
2191
2192 pub(in crate::db) fn current_canonical_accepted_catalog_selection(
2196 &self,
2197 entity: EntityTag,
2198 entity_path: &str,
2199 store_path: &'static str,
2200 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
2201 let first = self.canonical_root_slot_bytes(0)?;
2202 let second = self.canonical_root_slot_bytes(1)?;
2203 let Some(selection) =
2204 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2205 else {
2206 return Ok(None);
2207 };
2208 let bundle_key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
2209 let raw_bundle = self
2210 .get_canonical_raw_value(&bundle_key)?
2211 .ok_or_else(InternalError::store_corruption)?;
2212 let bundle = decode_verified_accepted_schema_revision_bundle(
2213 selection.root(),
2214 raw_bundle.as_bytes(),
2215 )?;
2216 if bundle.store_path() != store_path {
2217 return Err(InternalError::store_corruption());
2218 }
2219 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
2220 return Ok(None);
2221 };
2222 if snapshot.entity_path() != entity_path {
2223 return Err(InternalError::store_corruption());
2224 }
2225
2226 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2227 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
2228 let identity = AcceptedCatalogIdentity::new(
2229 entity,
2230 entity_path,
2231 store_path,
2232 bundle.revision(),
2233 snapshot.version(),
2234 fingerprint,
2235 );
2236
2237 Ok(Some(AcceptedCatalogSnapshotSelection::new(
2238 identity,
2239 AcceptedValueCatalogHandle::new(
2240 bundle.enum_catalog().clone(),
2241 bundle.composite_catalog().clone(),
2242 self.accepted_catalog_scope
2243 .get_or_init(AcceptedStoreCatalogScope::new)
2244 .clone(),
2245 bundle.revision(),
2246 selection.root().fingerprint(),
2247 ),
2248 Rc::from(raw_snapshot.into_bytes()),
2249 )))
2250 }
2251
2252 #[cfg(test)]
2258 pub(in crate::db) fn catalog_metadata(
2259 &self,
2260 ) -> Result<Option<SchemaStoreCatalogMetadata>, InternalError> {
2261 Ok(self
2262 .allocation_metadata()?
2263 .map(SchemaStoreAllocationMetadata::schema))
2264 }
2265
2266 pub(in crate::db) fn allocation_metadata(
2273 &self,
2274 ) -> Result<Option<SchemaStoreAllocationMetadata>, InternalError> {
2275 let latest_by_entity = self.latest_raw_snapshots_by_entity();
2276 if latest_by_entity.is_empty() {
2277 return Ok(None);
2278 }
2279
2280 Ok(Some(SchemaStoreAllocationMetadata::new(
2281 derive_data_allocation_metadata(&latest_by_entity)?,
2282 derive_index_allocation_metadata(&latest_by_entity)?,
2283 derive_schema_catalog_metadata(&latest_by_entity)?,
2284 )))
2285 }
2286
2287 fn insert_raw_snapshot(
2289 &mut self,
2290 key: RawSchemaKey,
2291 snapshot: RawSchemaSnapshot,
2292 ) -> Option<RawSchemaSnapshot> {
2293 self.invalidate_accepted_bundle_cache_for_key(key);
2294 let previous_journaled = if matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
2295 self.get_raw_snapshot_for_backend(&key)
2296 } else {
2297 None
2298 };
2299 match &mut self.backend {
2300 SchemaStoreBackend::Heap(map) => map.insert(key, snapshot),
2301 SchemaStoreBackend::Journaled {
2302 live, tombstones, ..
2303 } => {
2304 tombstones.remove(&key);
2305 live.insert(key, snapshot);
2306 previous_journaled
2307 }
2308 }
2309 }
2310
2311 #[must_use]
2313 fn get_raw_snapshot(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
2314 match &self.backend {
2315 SchemaStoreBackend::Heap(map) => map.get(key).cloned(),
2316 SchemaStoreBackend::Journaled { .. } => self.get_raw_snapshot_for_backend(key),
2317 }
2318 }
2319
2320 fn accepted_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
2321 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
2322 Ok(self
2323 .get_raw_snapshot(&key)
2324 .map(RawSchemaSnapshot::into_bytes))
2325 }
2326
2327 fn canonical_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
2328 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
2329 Ok(self
2330 .get_canonical_raw_value(&key)?
2331 .map(RawSchemaSnapshot::into_bytes))
2332 }
2333
2334 fn current_root_matches_candidate(
2335 &self,
2336 candidate: &CandidateSchemaRevision,
2337 ) -> Result<bool, InternalError> {
2338 let Some(selection) = self.current_accepted_schema_root()? else {
2339 return Ok(false);
2340 };
2341 if selection.root() != candidate.root() {
2342 return Ok(false);
2343 }
2344 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2345 let bundle = self
2346 .get_raw_snapshot(&key)
2347 .ok_or_else(InternalError::store_corruption)?;
2348 let _verified =
2349 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
2350 Ok(true)
2351 }
2352
2353 fn canonical_root_matches_candidate(
2354 &self,
2355 candidate: &CandidateSchemaRevision,
2356 ) -> Result<bool, InternalError> {
2357 let first = self.canonical_root_slot_bytes(0)?;
2358 let second = self.canonical_root_slot_bytes(1)?;
2359 let Some(selection) =
2360 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2361 else {
2362 return Ok(false);
2363 };
2364 if selection.root() != candidate.root() {
2365 return Ok(false);
2366 }
2367 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2368 let bundle = self
2369 .get_canonical_raw_value(&key)?
2370 .ok_or_else(InternalError::store_corruption)?;
2371 let _verified =
2372 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
2373 Ok(true)
2374 }
2375
2376 fn get_canonical_raw_value(
2377 &self,
2378 key: &RawSchemaKey,
2379 ) -> Result<Option<RawSchemaSnapshot>, InternalError> {
2380 match &self.backend {
2381 SchemaStoreBackend::Journaled { canonical, .. } => Ok(canonical.get(key)),
2382 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
2383 }
2384 }
2385
2386 fn insert_canonical_raw_value(
2387 &mut self,
2388 key: RawSchemaKey,
2389 bytes: Vec<u8>,
2390 ) -> Result<(), InternalError> {
2391 self.invalidate_accepted_bundle_cache_for_key(key);
2392 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
2393 return Err(InternalError::store_invariant());
2394 };
2395 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
2396 Ok(())
2397 }
2398
2399 fn insert_durable_raw_value(&mut self, key: RawSchemaKey, bytes: Vec<u8>) {
2403 self.invalidate_accepted_bundle_cache_for_key(key);
2404 let value = RawSchemaSnapshot::from_encoded_control_record(bytes);
2405 match &mut self.backend {
2406 SchemaStoreBackend::Heap(map) => {
2407 map.insert(key, value);
2408 }
2409 SchemaStoreBackend::Journaled {
2410 canonical,
2411 live,
2412 tombstones,
2413 } => {
2414 live.remove(&key);
2415 tombstones.remove(&key);
2416 canonical.insert(key, value);
2417 }
2418 }
2419 }
2420
2421 fn invalidate_accepted_bundle_cache_for_key(&mut self, key: RawSchemaKey) {
2422 if key.is_accepted_root() {
2423 self.accepted_bundle_cache.get_mut().take();
2424 }
2425 }
2426
2427 fn insert_durable_candidate_snapshots(
2428 &mut self,
2429 candidate: &CandidateSchemaRevision,
2430 ) -> Result<(), InternalError> {
2431 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2432 let key = RawSchemaKey::from_entity_version(*entity_tag, snapshot.version());
2433 let value = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2434 match &mut self.backend {
2435 SchemaStoreBackend::Heap(map) => {
2436 map.insert(key, value);
2437 }
2438 SchemaStoreBackend::Journaled {
2439 canonical,
2440 live,
2441 tombstones,
2442 } => {
2443 live.remove(&key);
2444 tombstones.remove(&key);
2445 canonical.insert(key, value);
2446 }
2447 }
2448 }
2449 Ok(())
2450 }
2451
2452 fn candidate_entry_keys(
2453 candidate: &CandidateSchemaRevision,
2454 root_slot: usize,
2455 ) -> Result<BTreeSet<RawSchemaKey>, InternalError> {
2456 let mut keys = candidate
2457 .bundle()
2458 .entity_snapshots()
2459 .iter()
2460 .map(|(entity_tag, snapshot)| {
2461 RawSchemaKey::from_entity_version(*entity_tag, snapshot.version())
2462 })
2463 .collect::<BTreeSet<_>>();
2464 keys.insert(RawSchemaKey::from_accepted_bundle(
2465 candidate.root().bundle_key(),
2466 ));
2467 keys.insert(RawSchemaKey::from_accepted_root_slot(root_slot)?);
2468 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2469 for activation in snapshot
2470 .constraint_activations()
2471 .iter()
2472 .filter(|activation| activation.state() == ConstraintActivationState::Validating)
2473 {
2474 keys.insert(RawSchemaKey::from_constraint_validation_job(
2475 *entity_tag,
2476 activation.id(),
2477 ));
2478 }
2479 }
2480 Ok(keys)
2481 }
2482
2483 fn retain_durable_candidate_entries(
2487 &mut self,
2488 candidate: &CandidateSchemaRevision,
2489 root_slot: usize,
2490 ) -> Result<(), InternalError> {
2491 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
2492 self.accepted_bundle_cache.get_mut().take();
2493 match &mut self.backend {
2494 SchemaStoreBackend::Heap(map) => {
2495 map.retain(|key, _| keep.contains(key) || key.is_identity_state());
2496 }
2497 SchemaStoreBackend::Journaled {
2498 canonical,
2499 live,
2500 tombstones,
2501 } => {
2502 let stale = canonical
2503 .iter()
2504 .filter_map(|entry| {
2505 (!keep.contains(entry.key()) && !entry.key().is_identity_state())
2506 .then_some(*entry.key())
2507 })
2508 .collect::<Vec<_>>();
2509 for key in stale {
2510 canonical.remove(&key);
2511 }
2512 live.retain(|key, _| keep.contains(key) || key.is_identity_state());
2513 tombstones.clear();
2514 }
2515 }
2516 Ok(())
2517 }
2518
2519 fn retain_materialized_candidate_entries(
2520 &mut self,
2521 candidate: &CandidateSchemaRevision,
2522 root_slot: usize,
2523 ) -> Result<(), InternalError> {
2524 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
2525 self.accepted_bundle_cache.get_mut().take();
2526 let SchemaStoreBackend::Journaled {
2527 canonical,
2528 live,
2529 tombstones,
2530 } = &mut self.backend
2531 else {
2532 return Err(InternalError::store_invariant());
2533 };
2534 live.retain(|key, _| keep.contains(key) || key.is_identity_state());
2535 let canonical_keys = canonical
2536 .iter()
2537 .map(|entry| *entry.key())
2538 .collect::<Vec<_>>();
2539 for key in canonical_keys {
2540 if keep.contains(&key) || key.is_identity_state() {
2541 tombstones.remove(&key);
2542 } else {
2543 tombstones.insert(key);
2544 }
2545 }
2546 Ok(())
2547 }
2548
2549 fn retain_canonical_candidate_entries(
2550 &mut self,
2551 candidate: &CandidateSchemaRevision,
2552 root_slot: usize,
2553 ) -> Result<(), InternalError> {
2554 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
2555 self.accepted_bundle_cache.get_mut().take();
2556 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
2557 return Err(InternalError::store_invariant());
2558 };
2559 let stale = canonical
2560 .iter()
2561 .filter_map(|entry| {
2562 (!keep.contains(entry.key()) && !entry.key().is_identity_state())
2563 .then_some(*entry.key())
2564 })
2565 .collect::<Vec<_>>();
2566 for key in stale {
2567 canonical.remove(&key);
2568 }
2569 Ok(())
2570 }
2571
2572 #[must_use]
2574 #[cfg(test)]
2575 fn contains_raw_snapshot(&self, key: &RawSchemaKey) -> bool {
2576 match &self.backend {
2577 SchemaStoreBackend::Heap(map) => map.contains_key(key),
2578 SchemaStoreBackend::Journaled { .. } => {
2579 self.get_raw_snapshot_for_backend(key).is_some()
2580 }
2581 }
2582 }
2583
2584 #[must_use]
2586 #[cfg(test)]
2587 pub(in crate::db) fn len(&self) -> u64 {
2588 match &self.backend {
2589 SchemaStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
2590 SchemaStoreBackend::Journaled { .. } => {
2591 let mut count = 0_u64;
2592 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
2593 count = count.saturating_add(1);
2594 Ok(SchemaStoreVisit::Continue)
2595 });
2596 count
2597 }
2598 }
2599 }
2600
2601 #[must_use]
2603 #[cfg(test)]
2604 pub(in crate::db) fn is_empty(&self) -> bool {
2605 match &self.backend {
2606 SchemaStoreBackend::Heap(map) => map.is_empty(),
2607 SchemaStoreBackend::Journaled { .. } => {
2608 let mut empty = true;
2609 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
2610 empty = false;
2611 Ok(SchemaStoreVisit::Stop)
2612 });
2613 empty
2614 }
2615 }
2616 }
2617
2618 #[cfg(test)]
2620 pub(in crate::db) fn clear(&mut self) {
2621 self.accepted_bundle_cache.get_mut().take();
2622 match &mut self.backend {
2623 SchemaStoreBackend::Heap(map) => map.clear(),
2624 SchemaStoreBackend::Journaled {
2625 canonical,
2626 live,
2627 tombstones,
2628 } => {
2629 live.clear();
2630 tombstones.clear();
2631 let keys = canonical
2632 .iter()
2633 .map(|entry| *entry.key())
2634 .collect::<Vec<_>>();
2635 for key in keys {
2636 if key.is_entity_snapshot() {
2637 tombstones.insert(key);
2638 } else {
2639 canonical.remove(&key);
2640 }
2641 }
2642 }
2643 }
2644 }
2645
2646 fn current_accepted_schema_bundle_ref(
2647 &self,
2648 ) -> Result<Option<Ref<'_, AcceptedSchemaRevisionBundle>>, InternalError> {
2649 let Some(selection) = self.current_accepted_schema_root()? else {
2650 self.accepted_bundle_cache
2651 .try_borrow_mut()
2652 .map_err(|_| InternalError::store_invariant())?
2653 .take();
2654 return Ok(None);
2655 };
2656
2657 let cache_matches = self
2658 .accepted_bundle_cache
2659 .try_borrow()
2660 .map_err(|_| InternalError::store_invariant())?
2661 .as_ref()
2662 .is_some_and(|cached| cached.selection == selection);
2663 if !cache_matches {
2664 let key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
2665 let raw = self
2666 .get_raw_snapshot(&key)
2667 .ok_or_else(InternalError::store_corruption)?;
2668 let bundle =
2669 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes())?;
2670 self.validate_constraint_validation_job_closure(&bundle)?;
2671 #[cfg(test)]
2672 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES
2673 .with(|misses| misses.set(misses.get().saturating_add(1)));
2674 let value_catalog = AcceptedValueCatalogHandle::new(
2675 bundle.enum_catalog().clone(),
2676 bundle.composite_catalog().clone(),
2677 self.accepted_catalog_scope
2678 .get_or_init(AcceptedStoreCatalogScope::new)
2679 .clone(),
2680 bundle.revision(),
2681 selection.root().fingerprint(),
2682 );
2683 *self
2684 .accepted_bundle_cache
2685 .try_borrow_mut()
2686 .map_err(|_| InternalError::store_invariant())? = Some(AcceptedSchemaBundleCache {
2687 selection,
2688 bundle,
2689 value_catalog,
2690 entity_selections: RefCell::new(StdBTreeMap::new()),
2691 });
2692 }
2693
2694 let cache = self
2695 .accepted_bundle_cache
2696 .try_borrow()
2697 .map_err(|_| InternalError::store_invariant())?;
2698 let bundle = Ref::filter_map(cache, |cache| {
2699 cache
2700 .as_ref()
2701 .filter(|cached| cached.selection == selection)
2702 .map(|cached| &cached.bundle)
2703 })
2704 .map_err(|_| InternalError::store_invariant())?;
2705 self.validate_identity_state_closure(&bundle)?;
2706 Ok(Some(bundle))
2707 }
2708
2709 fn latest_raw_snapshots_by_entity(
2710 &self,
2711 ) -> StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)> {
2712 let mut latest_by_entity =
2713 StdBTreeMap::<EntityTag, (SchemaVersion, RawSchemaSnapshot)>::new();
2714
2715 let _: Result<(), std::convert::Infallible> = self.visit_raw_snapshots(|key, snapshot| {
2716 let version = SchemaVersion::new(key.version());
2717 match latest_by_entity.get_mut(&key.entity_tag()) {
2718 Some((latest_version, latest_snapshot)) if version > *latest_version => {
2719 *latest_version = version;
2720 *latest_snapshot = snapshot.clone();
2721 }
2722 None => {
2723 latest_by_entity.insert(key.entity_tag(), (version, snapshot.clone()));
2724 }
2725 Some(_) => {}
2726 }
2727 Ok(SchemaStoreVisit::Continue)
2728 });
2729
2730 latest_by_entity
2731 }
2732
2733 fn visit_raw_snapshots<E>(
2736 &self,
2737 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2738 ) -> Result<(), E> {
2739 let bounds = RawSchemaKey::all_entity_range_bounds();
2740 match &self.backend {
2741 SchemaStoreBackend::Heap(map) => {
2742 let mut visitor = visitor;
2743 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
2744 if visitor(key, snapshot)?.should_stop() {
2745 break;
2746 }
2747 }
2748 }
2749 SchemaStoreBackend::Journaled {
2750 canonical,
2751 live,
2752 tombstones,
2753 } => Self::visit_journaled_raw_snapshot_range(
2754 canonical,
2755 live,
2756 tombstones,
2757 bounds,
2758 Direction::Asc,
2759 visitor,
2760 )?,
2761 }
2762
2763 Ok(())
2764 }
2765
2766 fn visit_constraint_validation_jobs<E>(
2767 &self,
2768 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2769 ) -> Result<(), E> {
2770 let bounds = RawSchemaKey::all_constraint_validation_job_range_bounds();
2771 match &self.backend {
2772 SchemaStoreBackend::Heap(map) => {
2773 let mut visitor = visitor;
2774 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
2775 if visitor(key, snapshot)?.should_stop() {
2776 break;
2777 }
2778 }
2779 }
2780 SchemaStoreBackend::Journaled {
2781 canonical,
2782 live,
2783 tombstones,
2784 } => Self::visit_journaled_raw_snapshot_range(
2785 canonical,
2786 live,
2787 tombstones,
2788 bounds,
2789 Direction::Asc,
2790 visitor,
2791 )?,
2792 }
2793 Ok(())
2794 }
2795
2796 #[cfg(test)]
2797 #[must_use]
2798 pub(in crate::db) fn canonical_len_for_tests(&self) -> u64 {
2799 match &self.backend {
2800 SchemaStoreBackend::Journaled { canonical: map, .. } => map.len(),
2801 SchemaStoreBackend::Heap(_) => 0,
2802 }
2803 }
2804
2805 fn get_raw_snapshot_for_backend(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
2806 let SchemaStoreBackend::Journaled {
2807 canonical,
2808 live,
2809 tombstones,
2810 } = &self.backend
2811 else {
2812 return None;
2813 };
2814
2815 if tombstones.contains(key) {
2816 return None;
2817 }
2818 live.get(key).cloned().or_else(|| canonical.get(key))
2819 }
2820
2821 fn visit_journaled_raw_snapshot_range<E>(
2822 canonical: &StableBTreeMap<
2823 RawSchemaKey,
2824 RawSchemaSnapshot,
2825 VirtualMemory<DefaultMemoryImpl>,
2826 >,
2827 live: &StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
2828 tombstones: &BTreeSet<RawSchemaKey>,
2829 bounds: (RangeBound<RawSchemaKey>, RangeBound<RawSchemaKey>),
2830 direction: Direction,
2831 mut visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2832 ) -> Result<(), E> {
2833 match direction {
2834 Direction::Asc => visit_ordered_overlay(
2835 canonical.range((bounds.0, bounds.1)),
2836 live.range((bounds.0, bounds.1)),
2837 Direction::Asc,
2838 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
2839 |canonical_entry| !tombstones.contains(canonical_entry.key()),
2840 |live_entry| !tombstones.contains(live_entry.0),
2841 |entry| {
2842 let visit = match entry {
2843 OrderedOverlayEntry::Canonical(canonical_entry) => {
2844 visitor(canonical_entry.key(), &canonical_entry.value())?
2845 }
2846 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
2847 };
2848 Ok(if visit.should_stop() {
2849 OrderedOverlayVisit::Stop
2850 } else {
2851 OrderedOverlayVisit::Continue
2852 })
2853 },
2854 ),
2855 Direction::Desc => visit_ordered_overlay(
2856 canonical.range((bounds.0, bounds.1)).rev(),
2857 live.range((bounds.0, bounds.1)).rev(),
2858 Direction::Desc,
2859 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
2860 |canonical_entry| !tombstones.contains(canonical_entry.key()),
2861 |live_entry| !tombstones.contains(live_entry.0),
2862 |entry| {
2863 let visit = match entry {
2864 OrderedOverlayEntry::Canonical(canonical_entry) => {
2865 visitor(canonical_entry.key(), &canonical_entry.value())?
2866 }
2867 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
2868 };
2869 Ok(if visit.should_stop() {
2870 OrderedOverlayVisit::Stop
2871 } else {
2872 OrderedOverlayVisit::Continue
2873 })
2874 },
2875 ),
2876 }
2877 }
2878}
2879
2880fn map_schema_publication_error(error: AcceptedSchemaPublicationError) -> InternalError {
2881 match error {
2882 AcceptedSchemaPublicationError::StaleSchemaRevision { .. }
2883 | AcceptedSchemaPublicationError::RevisionExhausted => InternalError::store_unsupported(),
2884 AcceptedSchemaPublicationError::InvalidCandidate => InternalError::store_invariant(),
2885 AcceptedSchemaPublicationError::CorruptRootSlots => InternalError::store_corruption(),
2886 }
2887}
2888
2889fn derive_data_allocation_metadata(
2890 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
2891) -> Result<SchemaStoreCatalogMetadata, InternalError> {
2892 let mut max_version = SchemaVersion::initial();
2893 let mut hasher = new_hash_sha256();
2894 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN);
2895
2896 for (entity, (_, snapshot)) in latest_by_entity {
2897 let persisted = snapshot.decode_persisted_snapshot()?;
2898 if persisted.version() > max_version {
2899 max_version = persisted.version();
2900 }
2901
2902 let data_projection = PersistedSchemaSnapshot::new_with_primary_key_fields_and_indexes(
2903 persisted.version(),
2904 persisted.entity_path().to_string(),
2905 persisted.entity_name().to_string(),
2906 persisted.primary_key_field_ids().to_vec(),
2907 persisted.row_layout().clone(),
2908 persisted.fields().to_vec(),
2909 Vec::new(),
2910 );
2911 let constraint_catalog = crate::db::schema::AcceptedConstraintCatalog::initial(
2912 data_projection.fields(),
2913 data_projection.indexes(),
2914 data_projection.relations(),
2915 )
2916 .map_err(|_| InternalError::store_invariant())?;
2917 let data_projection = data_projection.with_constraint_catalog(constraint_catalog);
2918 let encoded = encode_persisted_schema_snapshot(&data_projection)?;
2919
2920 write_hash_u64(&mut hasher, entity.value());
2921 write_hash_u32(&mut hasher, persisted.version().get());
2922 write_hash_len_u32(&mut hasher, encoded.len());
2923 hasher.update(encoded);
2924 }
2925
2926 Ok(finalize_schema_metadata(
2927 max_version,
2928 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
2929 hasher,
2930 latest_by_entity.len(),
2931 ))
2932}
2933
2934fn derive_index_allocation_metadata(
2935 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
2936) -> Result<SchemaStoreCatalogMetadata, InternalError> {
2937 let mut max_version = SchemaVersion::initial();
2938 let mut hasher = new_hash_sha256();
2939 write_hash_tag_u8(
2940 &mut hasher,
2941 SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN,
2942 );
2943
2944 for (entity, (_, snapshot)) in latest_by_entity {
2945 let persisted = snapshot.decode_persisted_snapshot()?;
2946 if persisted.version() > max_version {
2947 max_version = persisted.version();
2948 }
2949
2950 write_hash_u64(&mut hasher, entity.value());
2951 write_hash_u32(&mut hasher, persisted.version().get());
2952 write_hash_len_u32(&mut hasher, persisted.indexes().len());
2953 for index in persisted.indexes() {
2954 write_hash_u32(&mut hasher, u32::from(index.ordinal()));
2955 write_hash_str_u32(&mut hasher, index.name());
2956 write_hash_str_u32(&mut hasher, index.store());
2957 write_hash_tag_u8(&mut hasher, u8::from(index.unique()));
2958 write_hash_str_u32(&mut hasher, persisted_index_origin_name(index.origin()));
2959 match index.predicate_sql() {
2960 Some(predicate_sql) => {
2961 write_hash_tag_u8(&mut hasher, 1);
2962 write_hash_str_u32(&mut hasher, predicate_sql);
2963 }
2964 None => write_hash_tag_u8(&mut hasher, 0),
2965 }
2966 hash_persisted_index_key(&mut hasher, index.key());
2967 }
2968 }
2969
2970 Ok(finalize_schema_metadata(
2971 max_version,
2972 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
2973 hasher,
2974 latest_by_entity.len(),
2975 ))
2976}
2977
2978fn derive_schema_catalog_metadata(
2979 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
2980) -> Result<SchemaStoreCatalogMetadata, InternalError> {
2981 let mut max_version = SchemaVersion::initial();
2982 let mut hasher = new_hash_sha256();
2983 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN);
2984
2985 for (entity, (version, snapshot)) in latest_by_entity {
2986 let persisted = snapshot.decode_persisted_snapshot()?;
2987 if persisted.version() > max_version {
2988 max_version = persisted.version();
2989 }
2990
2991 write_hash_u64(&mut hasher, entity.value());
2992 write_hash_u32(&mut hasher, version.get());
2993 write_hash_len_u32(&mut hasher, snapshot.as_bytes().len());
2994 hasher.update(snapshot.as_bytes());
2995 }
2996
2997 Ok(finalize_schema_metadata(
2998 max_version,
2999 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
3000 hasher,
3001 latest_by_entity.len(),
3002 ))
3003}
3004
3005fn finalize_schema_metadata(
3006 schema_version: SchemaVersion,
3007 schema_fingerprint_method_version: u8,
3008 hasher: sha2::Sha256,
3009 entity_count: usize,
3010) -> SchemaStoreCatalogMetadata {
3011 let digest = finalize_hash_sha256(hasher);
3012 let mut schema_fingerprint = [0u8; 16];
3013 schema_fingerprint.copy_from_slice(&digest[..16]);
3014
3015 SchemaStoreCatalogMetadata::new(
3016 schema_version,
3017 schema_fingerprint_method_version,
3018 schema_fingerprint,
3019 u64::try_from(entity_count).unwrap_or(u64::MAX),
3020 )
3021}
3022
3023fn hash_persisted_index_key(hasher: &mut sha2::Sha256, key: &PersistedIndexKeySnapshot) {
3024 match key {
3025 PersistedIndexKeySnapshot::FieldPath(paths) => {
3026 write_hash_tag_u8(hasher, 1);
3027 write_hash_len_u32(hasher, paths.len());
3028 for path in paths {
3029 hash_persisted_index_field_path(hasher, path);
3030 }
3031 }
3032 PersistedIndexKeySnapshot::Items(items) => {
3033 write_hash_tag_u8(hasher, 2);
3034 write_hash_len_u32(hasher, items.len());
3035 for item in items {
3036 match item {
3037 PersistedIndexKeyItemSnapshot::FieldPath(path) => {
3038 write_hash_tag_u8(hasher, 1);
3039 hash_persisted_index_field_path(hasher, path);
3040 }
3041 PersistedIndexKeyItemSnapshot::Expression(expression) => {
3042 write_hash_tag_u8(hasher, 2);
3043 write_hash_str_u32(hasher, persisted_expression_op_name(expression.op()));
3044 hash_persisted_index_field_path(hasher, expression.source());
3045 hash_accepted_field_kind(hasher, expression.input_kind());
3046 hash_accepted_field_kind(hasher, expression.output_kind());
3047 write_hash_str_u32(hasher, expression.canonical_text());
3048 }
3049 }
3050 }
3051 }
3052 }
3053}
3054
3055fn hash_persisted_index_field_path(
3056 hasher: &mut sha2::Sha256,
3057 path: &crate::db::schema::PersistedIndexFieldPathSnapshot,
3058) {
3059 write_hash_u32(hasher, path.field_id().get());
3060 write_hash_u32(hasher, u32::from(path.slot().get()));
3061 write_hash_len_u32(hasher, path.path().len());
3062 for segment in path.path() {
3063 write_hash_str_u32(hasher, segment);
3064 }
3065 hash_accepted_field_kind(hasher, path.kind());
3066 write_hash_tag_u8(hasher, u8::from(path.nullable()));
3067}
3068
3069fn hash_accepted_field_kind(hasher: &mut sha2::Sha256, kind: &AcceptedFieldKind) {
3070 match kind {
3071 AcceptedFieldKind::Account => write_hash_tag_u8(hasher, 1),
3072 AcceptedFieldKind::Blob { max_len } => {
3073 write_hash_tag_u8(hasher, 2);
3074 hash_optional_u32(hasher, *max_len);
3075 }
3076 AcceptedFieldKind::Bool => {
3077 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL);
3078 }
3079 AcceptedFieldKind::Date => write_hash_tag_u8(hasher, 4),
3080 AcceptedFieldKind::Decimal { scale } => {
3081 write_hash_tag_u8(hasher, 5);
3082 write_hash_u32(hasher, *scale);
3083 }
3084 AcceptedFieldKind::Duration => write_hash_tag_u8(hasher, 6),
3085 AcceptedFieldKind::Enum { type_id } => {
3086 write_hash_tag_u8(hasher, 7);
3087 write_hash_u32(hasher, type_id.get());
3088 }
3089 AcceptedFieldKind::Float32 => write_hash_tag_u8(hasher, 8),
3090 AcceptedFieldKind::Float64 => write_hash_tag_u8(hasher, 9),
3091 AcceptedFieldKind::Int8 => write_hash_tag_u8(hasher, 10),
3092 AcceptedFieldKind::Int16 => write_hash_tag_u8(hasher, 11),
3093 AcceptedFieldKind::Int32 => write_hash_tag_u8(hasher, 12),
3094 AcceptedFieldKind::Int64 => write_hash_tag_u8(hasher, 13),
3095 AcceptedFieldKind::Int128 => write_hash_tag_u8(hasher, 14),
3096 AcceptedFieldKind::IntBig { max_bytes } => {
3097 write_hash_tag_u8(hasher, 15);
3098 write_hash_u32(hasher, *max_bytes);
3099 }
3100 AcceptedFieldKind::Principal => write_hash_tag_u8(hasher, 16),
3101 AcceptedFieldKind::Subaccount => write_hash_tag_u8(hasher, 17),
3102 AcceptedFieldKind::Text { max_len } => {
3103 write_hash_tag_u8(hasher, 18);
3104 hash_optional_u32(hasher, *max_len);
3105 }
3106 AcceptedFieldKind::Timestamp => write_hash_tag_u8(hasher, 19),
3107 AcceptedFieldKind::Nat8 => write_hash_tag_u8(hasher, 20),
3108 AcceptedFieldKind::Nat16 => write_hash_tag_u8(hasher, 21),
3109 AcceptedFieldKind::Nat32 => write_hash_tag_u8(hasher, 22),
3110 AcceptedFieldKind::Nat64 => write_hash_tag_u8(hasher, 23),
3111 AcceptedFieldKind::Nat128 => write_hash_tag_u8(hasher, 24),
3112 AcceptedFieldKind::NatBig { max_bytes } => {
3113 write_hash_tag_u8(hasher, 25);
3114 write_hash_u32(hasher, *max_bytes);
3115 }
3116 AcceptedFieldKind::Ulid => write_hash_tag_u8(hasher, 26),
3117 AcceptedFieldKind::Unit => write_hash_tag_u8(hasher, 27),
3118 AcceptedFieldKind::Relation {
3119 target_path,
3120 target_entity_name,
3121 target_entity_tag,
3122 target_store_path,
3123 key_kind,
3124 } => {
3125 write_hash_tag_u8(hasher, 28);
3126 write_hash_str_u32(hasher, target_path);
3127 write_hash_str_u32(hasher, target_entity_name);
3128 write_hash_u64(hasher, target_entity_tag.value());
3129 write_hash_str_u32(hasher, target_store_path);
3130 hash_accepted_field_kind(hasher, key_kind);
3131 }
3132 AcceptedFieldKind::List(inner) => {
3133 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST);
3134 hash_accepted_field_kind(hasher, inner);
3135 }
3136 AcceptedFieldKind::Set(inner) => {
3137 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET);
3138 hash_accepted_field_kind(hasher, inner);
3139 }
3140 AcceptedFieldKind::Map { key, value } => {
3141 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP);
3142 hash_accepted_field_kind(hasher, key);
3143 hash_accepted_field_kind(hasher, value);
3144 }
3145 AcceptedFieldKind::Composite { type_id } => {
3146 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE);
3147 write_hash_u32(hasher, type_id.get());
3148 }
3149 }
3150}
3151
3152fn hash_optional_u32(hasher: &mut sha2::Sha256, value: Option<u32>) {
3153 match value {
3154 Some(value) => {
3155 write_hash_tag_u8(hasher, 1);
3156 write_hash_u32(hasher, value);
3157 }
3158 None => write_hash_tag_u8(hasher, 0),
3159 }
3160}
3161
3162const fn persisted_index_origin_name(
3163 origin: crate::db::schema::PersistedIndexOrigin,
3164) -> &'static str {
3165 match origin {
3166 crate::db::schema::PersistedIndexOrigin::Generated => "generated",
3167 crate::db::schema::PersistedIndexOrigin::SqlDdl => "sql_ddl",
3168 }
3169}
3170
3171const fn persisted_expression_op_name(
3172 op: crate::db::schema::PersistedIndexExpressionOp,
3173) -> &'static str {
3174 match op {
3175 crate::db::schema::PersistedIndexExpressionOp::Lower => "lower",
3176 crate::db::schema::PersistedIndexExpressionOp::Upper => "upper",
3177 crate::db::schema::PersistedIndexExpressionOp::Trim => "trim",
3178 crate::db::schema::PersistedIndexExpressionOp::LowerTrim => "lower_trim",
3179 crate::db::schema::PersistedIndexExpressionOp::Date => "date",
3180 crate::db::schema::PersistedIndexExpressionOp::Year => "year",
3181 crate::db::schema::PersistedIndexExpressionOp::Month => "month",
3182 crate::db::schema::PersistedIndexExpressionOp::Day => "day",
3183 }
3184}
3185
3186#[cfg(test)]
3191mod tests;