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 pub(in crate::db) fn current_accepted_runtime_entity_for_name(
1407 &self,
1408 registered_store_path: &'static str,
1409 entity_name: &str,
1410 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1411 self.current_accepted_runtime_entity_matching(registered_store_path, |_, snapshot_name| {
1412 snapshot_name == entity_name
1413 })
1414 }
1415
1416 fn current_accepted_runtime_entity_matching(
1417 &self,
1418 registered_store_path: &'static str,
1419 mut predicate: impl FnMut(&str, &str) -> bool,
1420 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1421 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1422 return Ok(None);
1423 };
1424 if bundle.store_path() != registered_store_path {
1425 return Err(InternalError::store_corruption());
1426 }
1427
1428 let mut matched = None;
1429 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1430 if !predicate(snapshot.entity_path(), snapshot.entity_name()) {
1431 continue;
1432 }
1433 let entity = AcceptedRuntimeEntity::from_accepted_snapshot(
1434 &bundle,
1435 *entity_tag,
1436 snapshot,
1437 registered_store_path,
1438 )?;
1439 if matched.replace(entity).is_some() {
1440 return Err(InternalError::store_corruption());
1441 }
1442 }
1443
1444 Ok(matched)
1445 }
1446
1447 pub(in crate::db) fn current_accepted_schema_revision(
1449 &self,
1450 ) -> Result<Option<AcceptedSchemaRevision>, InternalError> {
1451 Ok(self
1452 .current_accepted_schema_root()?
1453 .map(|selection| selection.root().revision()))
1454 }
1455
1456 pub(in crate::db) fn pending_relation_activation_for_target(
1462 &self,
1463 target_path: &str,
1464 ) -> Result<Option<PendingRelationActivationDeleteBarrier>, InternalError> {
1465 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1466 return Ok(None);
1467 };
1468 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1469 let Some(candidate) = snapshot
1470 .candidate_relations()
1471 .iter()
1472 .find(|candidate| candidate.target_path() == target_path)
1473 else {
1474 continue;
1475 };
1476 let activation = snapshot
1477 .constraint_activations()
1478 .iter()
1479 .find(|activation| {
1480 matches!(
1481 activation.kind(),
1482 ConstraintActivationKind::Relation { relation_id }
1483 if *relation_id == candidate.id()
1484 )
1485 })
1486 .ok_or_else(InternalError::store_corruption)?;
1487 return Ok(Some(PendingRelationActivationDeleteBarrier {
1488 accepted_schema_fingerprint:
1489 accepted_schema_cache_fingerprint_for_persisted_snapshot(snapshot)?,
1490 source_entity_tag: *entity_tag,
1491 constraint_id: activation.id(),
1492 }));
1493 }
1494
1495 Ok(None)
1496 }
1497
1498 pub(in crate::db) fn entity_has_relation_to_target(
1500 &self,
1501 source_entity: EntityTag,
1502 target_path: &str,
1503 ) -> Result<bool, InternalError> {
1504 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1505 return Ok(false);
1506 };
1507 let Some(snapshot) = bundle.entity_snapshots().get(&source_entity) else {
1508 return Ok(false);
1509 };
1510
1511 Ok(snapshot
1512 .relations()
1513 .iter()
1514 .any(|relation| relation.target_path() == target_path))
1515 }
1516
1517 pub(in crate::db) fn validate_live_activation_transition(
1519 &self,
1520 candidate: &AcceptedSchemaRevisionBundle,
1521 ) -> Result<(), InternalError> {
1522 let Some(current) = self.current_accepted_schema_bundle()? else {
1523 return Ok(());
1524 };
1525 for (entity_tag, before) in current.entity_snapshots() {
1526 if before.constraint_activations().is_empty() {
1527 continue;
1528 }
1529 let after = candidate
1530 .entity_snapshots()
1531 .get(entity_tag)
1532 .ok_or_else(InternalError::store_invariant)?;
1533 if before == after {
1534 continue;
1535 }
1536 let expected_shape = before
1537 .clone()
1538 .with_constraint_catalog(after.constraint_catalog().clone());
1539 let catalog_only_transition = expected_shape == *after
1540 && before
1541 .constraint_catalog()
1542 .permits_live_activation_transition_to(after.constraint_catalog());
1543 let sql_row_local_abort_with_version =
1544 before.constraint_activations().iter().any(|activation| {
1545 activation.origin() == ConstraintOrigin::SqlDdl
1546 && matches!(
1547 activation.kind(),
1548 ConstraintActivationKind::Check { .. }
1549 | ConstraintActivationKind::NotNull { .. }
1550 )
1551 && before.version().get().checked_add(1) == Some(after.version().get())
1552 && before
1553 .constraint_catalog()
1554 .clone()
1555 .with_aborted_activation(activation.id())
1556 .is_ok_and(|catalog| catalog == *after.constraint_catalog())
1557 && before
1558 .clone()
1559 .with_constraint_catalog(after.constraint_catalog().clone())
1560 .with_schema_version(after.version())
1561 == *after
1562 });
1563 let sql_unique_abort_with_version =
1564 before.constraint_activations().iter().any(|activation| {
1565 activation.origin() == ConstraintOrigin::SqlDdl
1566 && matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
1567 && before.version().get().checked_add(1) == Some(after.version().get())
1568 && before
1569 .with_aborted_unique_activation(activation.id(), after.version())
1570 .is_ok_and(|expected| expected == *after)
1571 });
1572 let not_null_promotion = before.constraint_activations().iter().any(|activation| {
1573 matches!(activation.kind(), ConstraintActivationKind::NotNull { .. })
1574 && before
1575 .with_promoted_not_null_activation(activation.id(), after.version())
1576 .is_ok_and(|expected| expected == *after)
1577 });
1578 let unique_promotion = before.constraint_activations().iter().any(|activation| {
1579 matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
1580 && before
1581 .with_promoted_unique_activation(activation.id(), after.version())
1582 .is_ok_and(|expected| expected == *after)
1583 });
1584 let relation_promotion = before.constraint_activations().iter().any(|activation| {
1585 matches!(activation.kind(), ConstraintActivationKind::Relation { .. })
1586 && before
1587 .with_promoted_relation_activation(activation.id(), after.version())
1588 .is_ok_and(|expected| expected == *after)
1589 });
1590 if !catalog_only_transition
1591 && !sql_row_local_abort_with_version
1592 && !sql_unique_abort_with_version
1593 && !not_null_promotion
1594 && !unique_promotion
1595 && !relation_promotion
1596 {
1597 return Err(InternalError::store_invariant());
1598 }
1599 }
1600 Ok(())
1601 }
1602
1603 pub(in crate::db) fn validate_constraint_validation_job_closure(
1605 &self,
1606 bundle: &AcceptedSchemaRevisionBundle,
1607 ) -> Result<(), InternalError> {
1608 self.validate_constraint_validation_job_closure_with_change(bundle, None, None)
1609 }
1610
1611 pub(in crate::db) fn validate_constraint_validation_job_closure_with_change(
1614 &self,
1615 bundle: &AcceptedSchemaRevisionBundle,
1616 replacement: Option<&ConstraintValidationJob>,
1617 removal: Option<(EntityTag, ConstraintId)>,
1618 ) -> Result<(), InternalError> {
1619 if replacement.is_some() && removal.is_some() {
1620 return Err(InternalError::store_invariant());
1621 }
1622 let replacement_key = replacement.map(|job| {
1623 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id())
1624 });
1625 let removal_key = removal.map(|(entity_tag, constraint_id)| {
1626 RawSchemaKey::from_constraint_validation_job(entity_tag, constraint_id)
1627 });
1628 let mut expected = BTreeSet::new();
1629 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1630 for activation in snapshot.constraint_activations() {
1631 let key =
1632 RawSchemaKey::from_constraint_validation_job(*entity_tag, activation.id());
1633 match activation.state() {
1634 ConstraintActivationState::EnforcingNewWrites => {
1635 if self
1636 .constraint_validation_job_after_change(
1637 key,
1638 replacement,
1639 replacement_key,
1640 removal_key,
1641 )?
1642 .is_some()
1643 {
1644 return Err(InternalError::store_corruption());
1645 }
1646 }
1647 ConstraintActivationState::Validating => {
1648 let job = self
1649 .constraint_validation_job_after_change(
1650 key,
1651 replacement,
1652 replacement_key,
1653 removal_key,
1654 )?
1655 .ok_or_else(InternalError::store_corruption)?;
1656 if job.entity_tag() != *entity_tag
1657 || job.entity_path() != snapshot.entity_path()
1658 {
1659 return Err(InternalError::store_corruption());
1660 }
1661 job.validate(Some(activation))?;
1662 expected.insert(key);
1663 }
1664 }
1665 }
1666 }
1667
1668 self.visit_constraint_validation_jobs(|key, raw| {
1669 if removal_key == Some(*key) || replacement_key == Some(*key) {
1670 return Ok(SchemaStoreVisit::Continue);
1671 }
1672 if !expected.contains(key) {
1673 return Err(InternalError::store_corruption());
1674 }
1675 let job = decode_constraint_validation_job(raw.as_bytes())?;
1676 if job.entity_tag() != key.entity_tag()
1677 || key.constraint_id() != Some(job.constraint_id())
1678 {
1679 return Err(InternalError::store_corruption());
1680 }
1681 Ok(SchemaStoreVisit::Continue)
1682 })?;
1683
1684 if let Some(key) = replacement_key
1685 && !expected.contains(&key)
1686 {
1687 return Err(InternalError::store_corruption());
1688 }
1689 if let Some(key) = removal_key
1690 && expected.contains(&key)
1691 {
1692 return Err(InternalError::store_corruption());
1693 }
1694
1695 Ok(())
1696 }
1697
1698 fn constraint_validation_job_after_change(
1699 &self,
1700 key: RawSchemaKey,
1701 replacement: Option<&ConstraintValidationJob>,
1702 replacement_key: Option<RawSchemaKey>,
1703 removal_key: Option<RawSchemaKey>,
1704 ) -> Result<Option<ConstraintValidationJob>, InternalError> {
1705 if removal_key == Some(key) {
1706 return Ok(None);
1707 }
1708 if replacement_key == Some(key) {
1709 return Ok(replacement.cloned());
1710 }
1711 self.get_raw_snapshot(&key)
1712 .map(|raw| decode_constraint_validation_job(raw.as_bytes()))
1713 .transpose()
1714 }
1715
1716 pub(in crate::db) fn current_accepted_schema_authority_matches(
1719 &self,
1720 expected: &AcceptedSchemaAuthority,
1721 ) -> Result<bool, InternalError> {
1722 let Some(store_scope) = self.accepted_catalog_scope.get() else {
1723 return Ok(false);
1724 };
1725
1726 if let Some(cached) = self
1729 .accepted_bundle_cache
1730 .try_borrow()
1731 .map_err(|_| InternalError::store_invariant())?
1732 .as_ref()
1733 {
1734 let root = cached.selection.root();
1735 return Ok(expected.matches_store_root(
1736 store_scope,
1737 root.revision(),
1738 root.fingerprint(),
1739 ));
1740 }
1741
1742 let Some(selection) = self.current_accepted_schema_root()? else {
1743 return Ok(false);
1744 };
1745 let root = selection.root();
1746
1747 Ok(expected.matches_store_root(store_scope, root.revision(), root.fingerprint()))
1748 }
1749
1750 pub(in crate::db) fn publish_accepted_schema_candidate(
1756 &mut self,
1757 incarnation: DatabaseIncarnationId,
1758 expected_revision: AcceptedSchemaRevision,
1759 candidate: &CandidateSchemaRevision,
1760 ) -> Result<(), InternalError> {
1761 let identity_transition = self.prepare_identity_state_transition(
1762 incarnation,
1763 candidate,
1764 IdentityStateStorageView::Effective,
1765 )?;
1766 if self.current_root_matches_candidate(candidate)? {
1767 if !identity_transition.is_empty() {
1768 return Err(InternalError::identity_state_corruption());
1769 }
1770 let selection = self
1771 .current_accepted_schema_root()?
1772 .ok_or_else(InternalError::store_corruption)?;
1773 self.retain_durable_candidate_entries(candidate, selection.slot())?;
1774 return Ok(());
1775 }
1776 let first = self.accepted_root_slot_bytes(0)?;
1777 let second = self.accepted_root_slot_bytes(1)?;
1778 prepare_accepted_schema_root_publication(
1779 [first.as_deref(), second.as_deref()],
1780 expected_revision,
1781 candidate,
1782 )
1783 .map_err(map_schema_publication_error)?;
1784
1785 self.insert_durable_candidate_snapshots(candidate)?;
1786 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1787 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
1788 let persisted_bundle = self
1789 .get_raw_snapshot(&bundle_key)
1790 .ok_or_else(InternalError::store_corruption)?;
1791 let _verified = decode_verified_accepted_schema_revision_bundle(
1792 candidate.root(),
1793 persisted_bundle.as_bytes(),
1794 )?;
1795 self.apply_identity_state_transition(
1796 identity_transition,
1797 IdentityStateWriteTarget::Durable,
1798 )?;
1799
1800 let first = self.accepted_root_slot_bytes(0)?;
1803 let second = self.accepted_root_slot_bytes(1)?;
1804 let publication = prepare_accepted_schema_root_publication(
1805 [first.as_deref(), second.as_deref()],
1806 expected_revision,
1807 candidate,
1808 )
1809 .map_err(map_schema_publication_error)?;
1810 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
1811 self.insert_durable_raw_value(root_key, publication.encoded_root().to_vec());
1812
1813 let selected = self
1814 .current_accepted_schema_root()?
1815 .ok_or_else(InternalError::store_corruption)?;
1816 if selected.root() != candidate.root() {
1817 return Err(InternalError::store_corruption());
1818 }
1819 self.retain_durable_candidate_entries(candidate, selected.slot())?;
1820 Ok(())
1821 }
1822
1823 pub(in crate::db) fn restore_live_accepted_schema_checkpoint(
1826 &mut self,
1827 incarnation: DatabaseIncarnationId,
1828 candidate: &CandidateSchemaRevision,
1829 checkpoint_identity_states: &IdentityStateInventory,
1830 ) -> Result<(), InternalError> {
1831 if !matches!(self.backend, SchemaStoreBackend::Heap(_)) {
1832 return Err(InternalError::store_invariant());
1833 }
1834 let checkpoint_validation = prepare_identity_state_transition(
1835 incarnation,
1836 Some(candidate.bundle()),
1837 candidate.bundle(),
1838 checkpoint_identity_states.clone(),
1839 )?;
1840 if !checkpoint_validation.is_empty() {
1841 return Err(InternalError::identity_state_corruption());
1842 }
1843 if self.current_root_matches_candidate(candidate)? {
1844 for state in checkpoint_identity_states.values() {
1845 let key = RawSchemaKey::from_identity_state(
1846 state.owner().entity_tag(),
1847 state.owner().field_id(),
1848 );
1849 self.insert_durable_raw_value(key, encode_identity_state(state)?);
1850 }
1851 if self.identity_state_inventory(IdentityStateStorageView::Effective)?
1852 != *checkpoint_identity_states
1853 {
1854 return Err(InternalError::identity_state_corruption());
1855 }
1856 let selection = self
1857 .current_accepted_schema_root()?
1858 .ok_or_else(InternalError::store_corruption)?;
1859 self.retain_durable_candidate_entries(candidate, selection.slot())?;
1860 return Ok(());
1861 }
1862 if self.current_accepted_schema_root()?.is_some()
1863 || !self
1864 .identity_state_inventory(IdentityStateStorageView::Effective)?
1865 .is_empty()
1866 {
1867 return Err(InternalError::store_corruption());
1868 }
1869
1870 self.insert_durable_candidate_snapshots(candidate)?;
1871 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1872 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
1873 for state in checkpoint_identity_states.values() {
1874 let key = RawSchemaKey::from_identity_state(
1875 state.owner().entity_tag(),
1876 state.owner().field_id(),
1877 );
1878 self.insert_durable_raw_value(key, encode_identity_state(state)?);
1879 }
1880 let root_key = RawSchemaKey::from_accepted_root_slot(0)?;
1881 self.insert_durable_raw_value(root_key, candidate.encoded_root().to_vec());
1882
1883 let selected = self
1884 .current_accepted_schema_root()?
1885 .ok_or_else(InternalError::store_corruption)?;
1886 if selected.root() != candidate.root() {
1887 return Err(InternalError::store_corruption());
1888 }
1889 self.retain_durable_candidate_entries(candidate, selected.slot())?;
1890 Ok(())
1891 }
1892
1893 pub(in crate::db) fn preflight_accepted_schema_candidate(
1900 &self,
1901 incarnation: DatabaseIncarnationId,
1902 expected_revision: AcceptedSchemaRevision,
1903 candidate: &CandidateSchemaRevision,
1904 ) -> Result<bool, InternalError> {
1905 let identity_transition = self.prepare_identity_state_transition(
1906 incarnation,
1907 candidate,
1908 IdentityStateStorageView::Effective,
1909 )?;
1910 if self.current_root_matches_candidate(candidate)? {
1911 if !identity_transition.is_empty() {
1912 return Err(InternalError::identity_state_corruption());
1913 }
1914 return Ok(true);
1915 }
1916 let first = self.accepted_root_slot_bytes(0)?;
1917 let second = self.accepted_root_slot_bytes(1)?;
1918 prepare_accepted_schema_root_publication(
1919 [first.as_deref(), second.as_deref()],
1920 expected_revision,
1921 candidate,
1922 )
1923 .map_err(map_schema_publication_error)?;
1924
1925 Ok(false)
1926 }
1927
1928 pub(in crate::db) fn projected_identity_state_count(
1930 &self,
1931 incarnation: DatabaseIncarnationId,
1932 candidate: &CandidateSchemaRevision,
1933 ) -> Result<usize, InternalError> {
1934 Ok(self
1935 .prepare_identity_state_transition(
1936 incarnation,
1937 candidate,
1938 IdentityStateStorageView::Effective,
1939 )?
1940 .projected_inventory_len())
1941 }
1942
1943 pub(in crate::db) fn apply_journaled_accepted_schema_candidate(
1945 &mut self,
1946 incarnation: DatabaseIncarnationId,
1947 expected_revision: AcceptedSchemaRevision,
1948 candidate: &CandidateSchemaRevision,
1949 ) -> Result<(), InternalError> {
1950 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1951 return Err(InternalError::store_invariant());
1952 }
1953 let identity_transition = self.prepare_identity_state_transition(
1954 incarnation,
1955 candidate,
1956 IdentityStateStorageView::Effective,
1957 )?;
1958 if self.current_root_matches_candidate(candidate)? {
1959 if !identity_transition.is_empty() {
1960 return Err(InternalError::identity_state_corruption());
1961 }
1962 let selection = self
1963 .current_accepted_schema_root()?
1964 .ok_or_else(InternalError::store_corruption)?;
1965 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
1966 return Ok(());
1967 }
1968
1969 let first = self.accepted_root_slot_bytes(0)?;
1970 let second = self.accepted_root_slot_bytes(1)?;
1971 prepare_accepted_schema_root_publication(
1972 [first.as_deref(), second.as_deref()],
1973 expected_revision,
1974 candidate,
1975 )
1976 .map_err(map_schema_publication_error)?;
1977
1978 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
1979 self.insert_persisted_snapshot(*entity_tag, snapshot)?;
1980 }
1981 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1982 self.insert_raw_snapshot(
1983 bundle_key,
1984 RawSchemaSnapshot::from_encoded_control_record(candidate.encoded_bundle().to_vec()),
1985 );
1986 let persisted_bundle = self
1987 .get_raw_snapshot(&bundle_key)
1988 .ok_or_else(InternalError::store_corruption)?;
1989 let _verified = decode_verified_accepted_schema_revision_bundle(
1990 candidate.root(),
1991 persisted_bundle.as_bytes(),
1992 )?;
1993 self.apply_identity_state_transition(
1994 identity_transition,
1995 IdentityStateWriteTarget::Materialized,
1996 )?;
1997
1998 let first = self.accepted_root_slot_bytes(0)?;
1999 let second = self.accepted_root_slot_bytes(1)?;
2000 let publication = prepare_accepted_schema_root_publication(
2001 [first.as_deref(), second.as_deref()],
2002 expected_revision,
2003 candidate,
2004 )
2005 .map_err(map_schema_publication_error)?;
2006 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
2007 self.insert_raw_snapshot(
2008 root_key,
2009 RawSchemaSnapshot::from_encoded_control_record(publication.encoded_root().to_vec()),
2010 );
2011
2012 if !self.current_root_matches_candidate(candidate)? {
2013 return Err(InternalError::store_corruption());
2014 }
2015 let selection = self
2016 .current_accepted_schema_root()?
2017 .ok_or_else(InternalError::store_corruption)?;
2018 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
2019 Ok(())
2020 }
2021
2022 pub(in crate::db) fn fold_journaled_accepted_schema_candidate(
2024 &mut self,
2025 incarnation: DatabaseIncarnationId,
2026 expected_revision: AcceptedSchemaRevision,
2027 candidate: &CandidateSchemaRevision,
2028 ) -> Result<(), InternalError> {
2029 let identity_transition = self.prepare_identity_state_transition(
2030 incarnation,
2031 candidate,
2032 IdentityStateStorageView::Canonical,
2033 )?;
2034 if self.canonical_root_matches_candidate(candidate)? {
2035 if !identity_transition.is_empty() {
2036 return Err(InternalError::identity_state_corruption());
2037 }
2038 let first = self.canonical_root_slot_bytes(0)?;
2039 let second = self.canonical_root_slot_bytes(1)?;
2040 let selection =
2041 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2042 .ok_or_else(InternalError::store_corruption)?;
2043 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
2044 return Ok(());
2045 }
2046
2047 let first = self.canonical_root_slot_bytes(0)?;
2048 let second = self.canonical_root_slot_bytes(1)?;
2049 prepare_accepted_schema_root_publication(
2050 [first.as_deref(), second.as_deref()],
2051 expected_revision,
2052 candidate,
2053 )
2054 .map_err(map_schema_publication_error)?;
2055
2056 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2057 self.fold_persisted_snapshot(*entity_tag, snapshot)?;
2058 }
2059 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2060 self.insert_canonical_raw_value(bundle_key, candidate.encoded_bundle().to_vec())?;
2061 let persisted_bundle = self
2062 .get_canonical_raw_value(&bundle_key)?
2063 .ok_or_else(InternalError::store_corruption)?;
2064 let _verified = decode_verified_accepted_schema_revision_bundle(
2065 candidate.root(),
2066 persisted_bundle.as_bytes(),
2067 )?;
2068 self.apply_identity_state_transition(
2069 identity_transition,
2070 IdentityStateWriteTarget::Canonical,
2071 )?;
2072
2073 let first = self.canonical_root_slot_bytes(0)?;
2074 let second = self.canonical_root_slot_bytes(1)?;
2075 let publication = prepare_accepted_schema_root_publication(
2076 [first.as_deref(), second.as_deref()],
2077 expected_revision,
2078 candidate,
2079 )
2080 .map_err(map_schema_publication_error)?;
2081 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
2082 self.insert_canonical_raw_value(root_key, publication.encoded_root().to_vec())?;
2083
2084 if !self.canonical_root_matches_candidate(candidate)? {
2085 return Err(InternalError::store_corruption());
2086 }
2087 let first = self.canonical_root_slot_bytes(0)?;
2088 let second = self.canonical_root_slot_bytes(1)?;
2089 let selection = select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2090 .ok_or_else(InternalError::store_corruption)?;
2091 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
2092 Ok(())
2093 }
2094
2095 pub(in crate::db) fn get_persisted_snapshot(
2097 &self,
2098 entity: EntityTag,
2099 version: SchemaVersion,
2100 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2101 let key = RawSchemaKey::from_entity_version(entity, version);
2102 self.get_raw_snapshot(&key)
2103 .map(|snapshot| snapshot.decode_persisted_snapshot())
2104 .transpose()
2105 }
2106
2107 #[cfg(test)]
2108 fn latest_staged_persisted_snapshot(
2109 &self,
2110 entity: EntityTag,
2111 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2112 self.latest_raw_snapshots_by_entity()
2113 .remove(&entity)
2114 .map(|(_, snapshot)| snapshot.decode_persisted_snapshot())
2115 .transpose()
2116 }
2117
2118 pub(in crate::db) fn current_accepted_persisted_snapshot(
2121 &self,
2122 entity: EntityTag,
2123 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2124 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2125 return Ok(None);
2126 };
2127
2128 Ok(bundle.entity_snapshots().get(&entity).cloned())
2129 }
2130
2131 pub(in crate::db) fn current_accepted_catalog_selection(
2133 &self,
2134 entity: EntityTag,
2135 entity_path: &str,
2136 store_path: &'static str,
2137 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
2138 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2139 return Ok(None);
2140 };
2141 if bundle.store_path() != store_path {
2142 return Err(InternalError::store_corruption());
2143 }
2144 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
2145 return Ok(None);
2146 };
2147 if snapshot.entity_path() != entity_path {
2148 return Err(InternalError::store_corruption());
2149 }
2150
2151 let cache = self
2152 .accepted_bundle_cache
2153 .try_borrow()
2154 .map_err(|_| InternalError::store_invariant())?;
2155 let cached = cache.as_ref().ok_or_else(InternalError::store_invariant)?;
2156 if let Some(selection) = cached
2157 .entity_selections
2158 .try_borrow()
2159 .map_err(|_| InternalError::store_invariant())?
2160 .get(&entity)
2161 .cloned()
2162 {
2163 return Ok(Some(selection));
2164 }
2165
2166 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2167 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
2168 let identity = AcceptedCatalogIdentity::new(
2169 entity,
2170 entity_path,
2171 store_path,
2172 bundle.revision(),
2173 snapshot.version(),
2174 fingerprint,
2175 );
2176
2177 let selected = AcceptedCatalogSnapshotSelection::new(
2178 identity,
2179 cached.value_catalog.clone(),
2180 Rc::from(raw_snapshot.into_bytes()),
2181 );
2182 cached
2183 .entity_selections
2184 .try_borrow_mut()
2185 .map_err(|_| InternalError::store_invariant())?
2186 .insert(entity, selected.clone());
2187
2188 Ok(Some(selected))
2189 }
2190
2191 pub(in crate::db) fn current_canonical_accepted_catalog_selection(
2195 &self,
2196 entity: EntityTag,
2197 entity_path: &str,
2198 store_path: &'static str,
2199 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
2200 let first = self.canonical_root_slot_bytes(0)?;
2201 let second = self.canonical_root_slot_bytes(1)?;
2202 let Some(selection) =
2203 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2204 else {
2205 return Ok(None);
2206 };
2207 let bundle_key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
2208 let raw_bundle = self
2209 .get_canonical_raw_value(&bundle_key)?
2210 .ok_or_else(InternalError::store_corruption)?;
2211 let bundle = decode_verified_accepted_schema_revision_bundle(
2212 selection.root(),
2213 raw_bundle.as_bytes(),
2214 )?;
2215 if bundle.store_path() != store_path {
2216 return Err(InternalError::store_corruption());
2217 }
2218 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
2219 return Ok(None);
2220 };
2221 if snapshot.entity_path() != entity_path {
2222 return Err(InternalError::store_corruption());
2223 }
2224
2225 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2226 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
2227 let identity = AcceptedCatalogIdentity::new(
2228 entity,
2229 entity_path,
2230 store_path,
2231 bundle.revision(),
2232 snapshot.version(),
2233 fingerprint,
2234 );
2235
2236 Ok(Some(AcceptedCatalogSnapshotSelection::new(
2237 identity,
2238 AcceptedValueCatalogHandle::new(
2239 bundle.enum_catalog().clone(),
2240 bundle.composite_catalog().clone(),
2241 self.accepted_catalog_scope
2242 .get_or_init(AcceptedStoreCatalogScope::new)
2243 .clone(),
2244 bundle.revision(),
2245 selection.root().fingerprint(),
2246 ),
2247 Rc::from(raw_snapshot.into_bytes()),
2248 )))
2249 }
2250
2251 #[cfg(test)]
2257 pub(in crate::db) fn catalog_metadata(
2258 &self,
2259 ) -> Result<Option<SchemaStoreCatalogMetadata>, InternalError> {
2260 Ok(self
2261 .allocation_metadata()?
2262 .map(SchemaStoreAllocationMetadata::schema))
2263 }
2264
2265 pub(in crate::db) fn allocation_metadata(
2272 &self,
2273 ) -> Result<Option<SchemaStoreAllocationMetadata>, InternalError> {
2274 let latest_by_entity = self.latest_raw_snapshots_by_entity();
2275 if latest_by_entity.is_empty() {
2276 return Ok(None);
2277 }
2278
2279 Ok(Some(SchemaStoreAllocationMetadata::new(
2280 derive_data_allocation_metadata(&latest_by_entity)?,
2281 derive_index_allocation_metadata(&latest_by_entity)?,
2282 derive_schema_catalog_metadata(&latest_by_entity)?,
2283 )))
2284 }
2285
2286 fn insert_raw_snapshot(
2288 &mut self,
2289 key: RawSchemaKey,
2290 snapshot: RawSchemaSnapshot,
2291 ) -> Option<RawSchemaSnapshot> {
2292 self.invalidate_accepted_bundle_cache_for_key(key);
2293 let previous_journaled = if matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
2294 self.get_raw_snapshot_for_backend(&key)
2295 } else {
2296 None
2297 };
2298 match &mut self.backend {
2299 SchemaStoreBackend::Heap(map) => map.insert(key, snapshot),
2300 SchemaStoreBackend::Journaled {
2301 live, tombstones, ..
2302 } => {
2303 tombstones.remove(&key);
2304 live.insert(key, snapshot);
2305 previous_journaled
2306 }
2307 }
2308 }
2309
2310 #[must_use]
2312 fn get_raw_snapshot(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
2313 match &self.backend {
2314 SchemaStoreBackend::Heap(map) => map.get(key).cloned(),
2315 SchemaStoreBackend::Journaled { .. } => self.get_raw_snapshot_for_backend(key),
2316 }
2317 }
2318
2319 fn accepted_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
2320 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
2321 Ok(self
2322 .get_raw_snapshot(&key)
2323 .map(RawSchemaSnapshot::into_bytes))
2324 }
2325
2326 fn canonical_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
2327 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
2328 Ok(self
2329 .get_canonical_raw_value(&key)?
2330 .map(RawSchemaSnapshot::into_bytes))
2331 }
2332
2333 fn current_root_matches_candidate(
2334 &self,
2335 candidate: &CandidateSchemaRevision,
2336 ) -> Result<bool, InternalError> {
2337 let Some(selection) = self.current_accepted_schema_root()? else {
2338 return Ok(false);
2339 };
2340 if selection.root() != candidate.root() {
2341 return Ok(false);
2342 }
2343 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2344 let bundle = self
2345 .get_raw_snapshot(&key)
2346 .ok_or_else(InternalError::store_corruption)?;
2347 let _verified =
2348 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
2349 Ok(true)
2350 }
2351
2352 fn canonical_root_matches_candidate(
2353 &self,
2354 candidate: &CandidateSchemaRevision,
2355 ) -> Result<bool, InternalError> {
2356 let first = self.canonical_root_slot_bytes(0)?;
2357 let second = self.canonical_root_slot_bytes(1)?;
2358 let Some(selection) =
2359 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2360 else {
2361 return Ok(false);
2362 };
2363 if selection.root() != candidate.root() {
2364 return Ok(false);
2365 }
2366 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2367 let bundle = self
2368 .get_canonical_raw_value(&key)?
2369 .ok_or_else(InternalError::store_corruption)?;
2370 let _verified =
2371 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
2372 Ok(true)
2373 }
2374
2375 fn get_canonical_raw_value(
2376 &self,
2377 key: &RawSchemaKey,
2378 ) -> Result<Option<RawSchemaSnapshot>, InternalError> {
2379 match &self.backend {
2380 SchemaStoreBackend::Journaled { canonical, .. } => Ok(canonical.get(key)),
2381 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
2382 }
2383 }
2384
2385 fn insert_canonical_raw_value(
2386 &mut self,
2387 key: RawSchemaKey,
2388 bytes: Vec<u8>,
2389 ) -> Result<(), InternalError> {
2390 self.invalidate_accepted_bundle_cache_for_key(key);
2391 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
2392 return Err(InternalError::store_invariant());
2393 };
2394 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
2395 Ok(())
2396 }
2397
2398 fn insert_durable_raw_value(&mut self, key: RawSchemaKey, bytes: Vec<u8>) {
2402 self.invalidate_accepted_bundle_cache_for_key(key);
2403 let value = RawSchemaSnapshot::from_encoded_control_record(bytes);
2404 match &mut self.backend {
2405 SchemaStoreBackend::Heap(map) => {
2406 map.insert(key, value);
2407 }
2408 SchemaStoreBackend::Journaled {
2409 canonical,
2410 live,
2411 tombstones,
2412 } => {
2413 live.remove(&key);
2414 tombstones.remove(&key);
2415 canonical.insert(key, value);
2416 }
2417 }
2418 }
2419
2420 fn invalidate_accepted_bundle_cache_for_key(&mut self, key: RawSchemaKey) {
2421 if key.is_accepted_root() {
2422 self.accepted_bundle_cache.get_mut().take();
2423 }
2424 }
2425
2426 fn insert_durable_candidate_snapshots(
2427 &mut self,
2428 candidate: &CandidateSchemaRevision,
2429 ) -> Result<(), InternalError> {
2430 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2431 let key = RawSchemaKey::from_entity_version(*entity_tag, snapshot.version());
2432 let value = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2433 match &mut self.backend {
2434 SchemaStoreBackend::Heap(map) => {
2435 map.insert(key, value);
2436 }
2437 SchemaStoreBackend::Journaled {
2438 canonical,
2439 live,
2440 tombstones,
2441 } => {
2442 live.remove(&key);
2443 tombstones.remove(&key);
2444 canonical.insert(key, value);
2445 }
2446 }
2447 }
2448 Ok(())
2449 }
2450
2451 fn candidate_entry_keys(
2452 candidate: &CandidateSchemaRevision,
2453 root_slot: usize,
2454 ) -> Result<BTreeSet<RawSchemaKey>, InternalError> {
2455 let mut keys = candidate
2456 .bundle()
2457 .entity_snapshots()
2458 .iter()
2459 .map(|(entity_tag, snapshot)| {
2460 RawSchemaKey::from_entity_version(*entity_tag, snapshot.version())
2461 })
2462 .collect::<BTreeSet<_>>();
2463 keys.insert(RawSchemaKey::from_accepted_bundle(
2464 candidate.root().bundle_key(),
2465 ));
2466 keys.insert(RawSchemaKey::from_accepted_root_slot(root_slot)?);
2467 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2468 for activation in snapshot
2469 .constraint_activations()
2470 .iter()
2471 .filter(|activation| activation.state() == ConstraintActivationState::Validating)
2472 {
2473 keys.insert(RawSchemaKey::from_constraint_validation_job(
2474 *entity_tag,
2475 activation.id(),
2476 ));
2477 }
2478 }
2479 Ok(keys)
2480 }
2481
2482 fn retain_durable_candidate_entries(
2486 &mut self,
2487 candidate: &CandidateSchemaRevision,
2488 root_slot: usize,
2489 ) -> Result<(), InternalError> {
2490 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
2491 self.accepted_bundle_cache.get_mut().take();
2492 match &mut self.backend {
2493 SchemaStoreBackend::Heap(map) => {
2494 map.retain(|key, _| keep.contains(key) || key.is_identity_state());
2495 }
2496 SchemaStoreBackend::Journaled {
2497 canonical,
2498 live,
2499 tombstones,
2500 } => {
2501 let stale = canonical
2502 .iter()
2503 .filter_map(|entry| {
2504 (!keep.contains(entry.key()) && !entry.key().is_identity_state())
2505 .then_some(*entry.key())
2506 })
2507 .collect::<Vec<_>>();
2508 for key in stale {
2509 canonical.remove(&key);
2510 }
2511 live.retain(|key, _| keep.contains(key) || key.is_identity_state());
2512 tombstones.clear();
2513 }
2514 }
2515 Ok(())
2516 }
2517
2518 fn retain_materialized_candidate_entries(
2519 &mut self,
2520 candidate: &CandidateSchemaRevision,
2521 root_slot: usize,
2522 ) -> Result<(), InternalError> {
2523 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
2524 self.accepted_bundle_cache.get_mut().take();
2525 let SchemaStoreBackend::Journaled {
2526 canonical,
2527 live,
2528 tombstones,
2529 } = &mut self.backend
2530 else {
2531 return Err(InternalError::store_invariant());
2532 };
2533 live.retain(|key, _| keep.contains(key) || key.is_identity_state());
2534 let canonical_keys = canonical
2535 .iter()
2536 .map(|entry| *entry.key())
2537 .collect::<Vec<_>>();
2538 for key in canonical_keys {
2539 if keep.contains(&key) || key.is_identity_state() {
2540 tombstones.remove(&key);
2541 } else {
2542 tombstones.insert(key);
2543 }
2544 }
2545 Ok(())
2546 }
2547
2548 fn retain_canonical_candidate_entries(
2549 &mut self,
2550 candidate: &CandidateSchemaRevision,
2551 root_slot: usize,
2552 ) -> Result<(), InternalError> {
2553 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
2554 self.accepted_bundle_cache.get_mut().take();
2555 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
2556 return Err(InternalError::store_invariant());
2557 };
2558 let stale = canonical
2559 .iter()
2560 .filter_map(|entry| {
2561 (!keep.contains(entry.key()) && !entry.key().is_identity_state())
2562 .then_some(*entry.key())
2563 })
2564 .collect::<Vec<_>>();
2565 for key in stale {
2566 canonical.remove(&key);
2567 }
2568 Ok(())
2569 }
2570
2571 #[must_use]
2573 #[cfg(test)]
2574 fn contains_raw_snapshot(&self, key: &RawSchemaKey) -> bool {
2575 match &self.backend {
2576 SchemaStoreBackend::Heap(map) => map.contains_key(key),
2577 SchemaStoreBackend::Journaled { .. } => {
2578 self.get_raw_snapshot_for_backend(key).is_some()
2579 }
2580 }
2581 }
2582
2583 #[must_use]
2585 #[cfg(test)]
2586 pub(in crate::db) fn len(&self) -> u64 {
2587 match &self.backend {
2588 SchemaStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
2589 SchemaStoreBackend::Journaled { .. } => {
2590 let mut count = 0_u64;
2591 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
2592 count = count.saturating_add(1);
2593 Ok(SchemaStoreVisit::Continue)
2594 });
2595 count
2596 }
2597 }
2598 }
2599
2600 #[must_use]
2602 #[cfg(test)]
2603 pub(in crate::db) fn is_empty(&self) -> bool {
2604 match &self.backend {
2605 SchemaStoreBackend::Heap(map) => map.is_empty(),
2606 SchemaStoreBackend::Journaled { .. } => {
2607 let mut empty = true;
2608 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
2609 empty = false;
2610 Ok(SchemaStoreVisit::Stop)
2611 });
2612 empty
2613 }
2614 }
2615 }
2616
2617 #[cfg(test)]
2619 pub(in crate::db) fn clear(&mut self) {
2620 self.accepted_bundle_cache.get_mut().take();
2621 match &mut self.backend {
2622 SchemaStoreBackend::Heap(map) => map.clear(),
2623 SchemaStoreBackend::Journaled {
2624 canonical,
2625 live,
2626 tombstones,
2627 } => {
2628 live.clear();
2629 tombstones.clear();
2630 let keys = canonical
2631 .iter()
2632 .map(|entry| *entry.key())
2633 .collect::<Vec<_>>();
2634 for key in keys {
2635 if key.is_entity_snapshot() {
2636 tombstones.insert(key);
2637 } else {
2638 canonical.remove(&key);
2639 }
2640 }
2641 }
2642 }
2643 }
2644
2645 fn current_accepted_schema_bundle_ref(
2646 &self,
2647 ) -> Result<Option<Ref<'_, AcceptedSchemaRevisionBundle>>, InternalError> {
2648 let Some(selection) = self.current_accepted_schema_root()? else {
2649 self.accepted_bundle_cache
2650 .try_borrow_mut()
2651 .map_err(|_| InternalError::store_invariant())?
2652 .take();
2653 return Ok(None);
2654 };
2655
2656 let cache_matches = self
2657 .accepted_bundle_cache
2658 .try_borrow()
2659 .map_err(|_| InternalError::store_invariant())?
2660 .as_ref()
2661 .is_some_and(|cached| cached.selection == selection);
2662 if !cache_matches {
2663 let key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
2664 let raw = self
2665 .get_raw_snapshot(&key)
2666 .ok_or_else(InternalError::store_corruption)?;
2667 let bundle =
2668 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes())?;
2669 self.validate_constraint_validation_job_closure(&bundle)?;
2670 #[cfg(test)]
2671 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES
2672 .with(|misses| misses.set(misses.get().saturating_add(1)));
2673 let value_catalog = AcceptedValueCatalogHandle::new(
2674 bundle.enum_catalog().clone(),
2675 bundle.composite_catalog().clone(),
2676 self.accepted_catalog_scope
2677 .get_or_init(AcceptedStoreCatalogScope::new)
2678 .clone(),
2679 bundle.revision(),
2680 selection.root().fingerprint(),
2681 );
2682 *self
2683 .accepted_bundle_cache
2684 .try_borrow_mut()
2685 .map_err(|_| InternalError::store_invariant())? = Some(AcceptedSchemaBundleCache {
2686 selection,
2687 bundle,
2688 value_catalog,
2689 entity_selections: RefCell::new(StdBTreeMap::new()),
2690 });
2691 }
2692
2693 let cache = self
2694 .accepted_bundle_cache
2695 .try_borrow()
2696 .map_err(|_| InternalError::store_invariant())?;
2697 let bundle = Ref::filter_map(cache, |cache| {
2698 cache
2699 .as_ref()
2700 .filter(|cached| cached.selection == selection)
2701 .map(|cached| &cached.bundle)
2702 })
2703 .map_err(|_| InternalError::store_invariant())?;
2704 self.validate_identity_state_closure(&bundle)?;
2705 Ok(Some(bundle))
2706 }
2707
2708 fn latest_raw_snapshots_by_entity(
2709 &self,
2710 ) -> StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)> {
2711 let mut latest_by_entity =
2712 StdBTreeMap::<EntityTag, (SchemaVersion, RawSchemaSnapshot)>::new();
2713
2714 let _: Result<(), std::convert::Infallible> = self.visit_raw_snapshots(|key, snapshot| {
2715 let version = SchemaVersion::new(key.version());
2716 match latest_by_entity.get_mut(&key.entity_tag()) {
2717 Some((latest_version, latest_snapshot)) if version > *latest_version => {
2718 *latest_version = version;
2719 *latest_snapshot = snapshot.clone();
2720 }
2721 None => {
2722 latest_by_entity.insert(key.entity_tag(), (version, snapshot.clone()));
2723 }
2724 Some(_) => {}
2725 }
2726 Ok(SchemaStoreVisit::Continue)
2727 });
2728
2729 latest_by_entity
2730 }
2731
2732 fn visit_raw_snapshots<E>(
2735 &self,
2736 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2737 ) -> Result<(), E> {
2738 let bounds = RawSchemaKey::all_entity_range_bounds();
2739 match &self.backend {
2740 SchemaStoreBackend::Heap(map) => {
2741 let mut visitor = visitor;
2742 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
2743 if visitor(key, snapshot)?.should_stop() {
2744 break;
2745 }
2746 }
2747 }
2748 SchemaStoreBackend::Journaled {
2749 canonical,
2750 live,
2751 tombstones,
2752 } => Self::visit_journaled_raw_snapshot_range(
2753 canonical,
2754 live,
2755 tombstones,
2756 bounds,
2757 Direction::Asc,
2758 visitor,
2759 )?,
2760 }
2761
2762 Ok(())
2763 }
2764
2765 fn visit_constraint_validation_jobs<E>(
2766 &self,
2767 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2768 ) -> Result<(), E> {
2769 let bounds = RawSchemaKey::all_constraint_validation_job_range_bounds();
2770 match &self.backend {
2771 SchemaStoreBackend::Heap(map) => {
2772 let mut visitor = visitor;
2773 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
2774 if visitor(key, snapshot)?.should_stop() {
2775 break;
2776 }
2777 }
2778 }
2779 SchemaStoreBackend::Journaled {
2780 canonical,
2781 live,
2782 tombstones,
2783 } => Self::visit_journaled_raw_snapshot_range(
2784 canonical,
2785 live,
2786 tombstones,
2787 bounds,
2788 Direction::Asc,
2789 visitor,
2790 )?,
2791 }
2792 Ok(())
2793 }
2794
2795 #[cfg(test)]
2796 #[must_use]
2797 pub(in crate::db) fn canonical_len_for_tests(&self) -> u64 {
2798 match &self.backend {
2799 SchemaStoreBackend::Journaled { canonical: map, .. } => map.len(),
2800 SchemaStoreBackend::Heap(_) => 0,
2801 }
2802 }
2803
2804 fn get_raw_snapshot_for_backend(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
2805 let SchemaStoreBackend::Journaled {
2806 canonical,
2807 live,
2808 tombstones,
2809 } = &self.backend
2810 else {
2811 return None;
2812 };
2813
2814 if tombstones.contains(key) {
2815 return None;
2816 }
2817 live.get(key).cloned().or_else(|| canonical.get(key))
2818 }
2819
2820 fn visit_journaled_raw_snapshot_range<E>(
2821 canonical: &StableBTreeMap<
2822 RawSchemaKey,
2823 RawSchemaSnapshot,
2824 VirtualMemory<DefaultMemoryImpl>,
2825 >,
2826 live: &StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
2827 tombstones: &BTreeSet<RawSchemaKey>,
2828 bounds: (RangeBound<RawSchemaKey>, RangeBound<RawSchemaKey>),
2829 direction: Direction,
2830 mut visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2831 ) -> Result<(), E> {
2832 match direction {
2833 Direction::Asc => visit_ordered_overlay(
2834 canonical.range((bounds.0, bounds.1)),
2835 live.range((bounds.0, bounds.1)),
2836 Direction::Asc,
2837 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
2838 |canonical_entry| !tombstones.contains(canonical_entry.key()),
2839 |live_entry| !tombstones.contains(live_entry.0),
2840 |entry| {
2841 let visit = match entry {
2842 OrderedOverlayEntry::Canonical(canonical_entry) => {
2843 visitor(canonical_entry.key(), &canonical_entry.value())?
2844 }
2845 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
2846 };
2847 Ok(if visit.should_stop() {
2848 OrderedOverlayVisit::Stop
2849 } else {
2850 OrderedOverlayVisit::Continue
2851 })
2852 },
2853 ),
2854 Direction::Desc => visit_ordered_overlay(
2855 canonical.range((bounds.0, bounds.1)).rev(),
2856 live.range((bounds.0, bounds.1)).rev(),
2857 Direction::Desc,
2858 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
2859 |canonical_entry| !tombstones.contains(canonical_entry.key()),
2860 |live_entry| !tombstones.contains(live_entry.0),
2861 |entry| {
2862 let visit = match entry {
2863 OrderedOverlayEntry::Canonical(canonical_entry) => {
2864 visitor(canonical_entry.key(), &canonical_entry.value())?
2865 }
2866 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
2867 };
2868 Ok(if visit.should_stop() {
2869 OrderedOverlayVisit::Stop
2870 } else {
2871 OrderedOverlayVisit::Continue
2872 })
2873 },
2874 ),
2875 }
2876 }
2877}
2878
2879fn map_schema_publication_error(error: AcceptedSchemaPublicationError) -> InternalError {
2880 match error {
2881 AcceptedSchemaPublicationError::StaleSchemaRevision { .. }
2882 | AcceptedSchemaPublicationError::RevisionExhausted => InternalError::store_unsupported(),
2883 AcceptedSchemaPublicationError::InvalidCandidate => InternalError::store_invariant(),
2884 AcceptedSchemaPublicationError::CorruptRootSlots => InternalError::store_corruption(),
2885 }
2886}
2887
2888fn derive_data_allocation_metadata(
2889 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
2890) -> Result<SchemaStoreCatalogMetadata, InternalError> {
2891 let mut max_version = SchemaVersion::initial();
2892 let mut hasher = new_hash_sha256();
2893 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN);
2894
2895 for (entity, (_, snapshot)) in latest_by_entity {
2896 let persisted = snapshot.decode_persisted_snapshot()?;
2897 if persisted.version() > max_version {
2898 max_version = persisted.version();
2899 }
2900
2901 let data_projection = PersistedSchemaSnapshot::new_with_primary_key_fields_and_indexes(
2902 persisted.version(),
2903 persisted.entity_path().to_string(),
2904 persisted.entity_name().to_string(),
2905 persisted.primary_key_field_ids().to_vec(),
2906 persisted.row_layout().clone(),
2907 persisted.fields().to_vec(),
2908 Vec::new(),
2909 );
2910 let constraint_catalog = crate::db::schema::AcceptedConstraintCatalog::initial(
2911 data_projection.fields(),
2912 data_projection.indexes(),
2913 data_projection.relations(),
2914 )
2915 .map_err(|_| InternalError::store_invariant())?;
2916 let data_projection = data_projection.with_constraint_catalog(constraint_catalog);
2917 let encoded = encode_persisted_schema_snapshot(&data_projection)?;
2918
2919 write_hash_u64(&mut hasher, entity.value());
2920 write_hash_u32(&mut hasher, persisted.version().get());
2921 write_hash_len_u32(&mut hasher, encoded.len());
2922 hasher.update(encoded);
2923 }
2924
2925 Ok(finalize_schema_metadata(
2926 max_version,
2927 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
2928 hasher,
2929 latest_by_entity.len(),
2930 ))
2931}
2932
2933fn derive_index_allocation_metadata(
2934 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
2935) -> Result<SchemaStoreCatalogMetadata, InternalError> {
2936 let mut max_version = SchemaVersion::initial();
2937 let mut hasher = new_hash_sha256();
2938 write_hash_tag_u8(
2939 &mut hasher,
2940 SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN,
2941 );
2942
2943 for (entity, (_, snapshot)) in latest_by_entity {
2944 let persisted = snapshot.decode_persisted_snapshot()?;
2945 if persisted.version() > max_version {
2946 max_version = persisted.version();
2947 }
2948
2949 write_hash_u64(&mut hasher, entity.value());
2950 write_hash_u32(&mut hasher, persisted.version().get());
2951 write_hash_len_u32(&mut hasher, persisted.indexes().len());
2952 for index in persisted.indexes() {
2953 write_hash_u32(&mut hasher, u32::from(index.ordinal()));
2954 write_hash_str_u32(&mut hasher, index.name());
2955 write_hash_str_u32(&mut hasher, index.store());
2956 write_hash_tag_u8(&mut hasher, u8::from(index.unique()));
2957 write_hash_str_u32(&mut hasher, persisted_index_origin_name(index.origin()));
2958 match index.predicate_sql() {
2959 Some(predicate_sql) => {
2960 write_hash_tag_u8(&mut hasher, 1);
2961 write_hash_str_u32(&mut hasher, predicate_sql);
2962 }
2963 None => write_hash_tag_u8(&mut hasher, 0),
2964 }
2965 hash_persisted_index_key(&mut hasher, index.key());
2966 }
2967 }
2968
2969 Ok(finalize_schema_metadata(
2970 max_version,
2971 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
2972 hasher,
2973 latest_by_entity.len(),
2974 ))
2975}
2976
2977fn derive_schema_catalog_metadata(
2978 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
2979) -> Result<SchemaStoreCatalogMetadata, InternalError> {
2980 let mut max_version = SchemaVersion::initial();
2981 let mut hasher = new_hash_sha256();
2982 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN);
2983
2984 for (entity, (version, snapshot)) in latest_by_entity {
2985 let persisted = snapshot.decode_persisted_snapshot()?;
2986 if persisted.version() > max_version {
2987 max_version = persisted.version();
2988 }
2989
2990 write_hash_u64(&mut hasher, entity.value());
2991 write_hash_u32(&mut hasher, version.get());
2992 write_hash_len_u32(&mut hasher, snapshot.as_bytes().len());
2993 hasher.update(snapshot.as_bytes());
2994 }
2995
2996 Ok(finalize_schema_metadata(
2997 max_version,
2998 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
2999 hasher,
3000 latest_by_entity.len(),
3001 ))
3002}
3003
3004fn finalize_schema_metadata(
3005 schema_version: SchemaVersion,
3006 schema_fingerprint_method_version: u8,
3007 hasher: sha2::Sha256,
3008 entity_count: usize,
3009) -> SchemaStoreCatalogMetadata {
3010 let digest = finalize_hash_sha256(hasher);
3011 let mut schema_fingerprint = [0u8; 16];
3012 schema_fingerprint.copy_from_slice(&digest[..16]);
3013
3014 SchemaStoreCatalogMetadata::new(
3015 schema_version,
3016 schema_fingerprint_method_version,
3017 schema_fingerprint,
3018 u64::try_from(entity_count).unwrap_or(u64::MAX),
3019 )
3020}
3021
3022fn hash_persisted_index_key(hasher: &mut sha2::Sha256, key: &PersistedIndexKeySnapshot) {
3023 match key {
3024 PersistedIndexKeySnapshot::FieldPath(paths) => {
3025 write_hash_tag_u8(hasher, 1);
3026 write_hash_len_u32(hasher, paths.len());
3027 for path in paths {
3028 hash_persisted_index_field_path(hasher, path);
3029 }
3030 }
3031 PersistedIndexKeySnapshot::Items(items) => {
3032 write_hash_tag_u8(hasher, 2);
3033 write_hash_len_u32(hasher, items.len());
3034 for item in items {
3035 match item {
3036 PersistedIndexKeyItemSnapshot::FieldPath(path) => {
3037 write_hash_tag_u8(hasher, 1);
3038 hash_persisted_index_field_path(hasher, path);
3039 }
3040 PersistedIndexKeyItemSnapshot::Expression(expression) => {
3041 write_hash_tag_u8(hasher, 2);
3042 write_hash_str_u32(hasher, persisted_expression_op_name(expression.op()));
3043 hash_persisted_index_field_path(hasher, expression.source());
3044 hash_accepted_field_kind(hasher, expression.input_kind());
3045 hash_accepted_field_kind(hasher, expression.output_kind());
3046 write_hash_str_u32(hasher, expression.canonical_text());
3047 }
3048 }
3049 }
3050 }
3051 }
3052}
3053
3054fn hash_persisted_index_field_path(
3055 hasher: &mut sha2::Sha256,
3056 path: &crate::db::schema::PersistedIndexFieldPathSnapshot,
3057) {
3058 write_hash_u32(hasher, path.field_id().get());
3059 write_hash_u32(hasher, u32::from(path.slot().get()));
3060 write_hash_len_u32(hasher, path.path().len());
3061 for segment in path.path() {
3062 write_hash_str_u32(hasher, segment);
3063 }
3064 hash_accepted_field_kind(hasher, path.kind());
3065 write_hash_tag_u8(hasher, u8::from(path.nullable()));
3066}
3067
3068fn hash_accepted_field_kind(hasher: &mut sha2::Sha256, kind: &AcceptedFieldKind) {
3069 match kind {
3070 AcceptedFieldKind::Account => write_hash_tag_u8(hasher, 1),
3071 AcceptedFieldKind::Blob { max_len } => {
3072 write_hash_tag_u8(hasher, 2);
3073 hash_optional_u32(hasher, *max_len);
3074 }
3075 AcceptedFieldKind::Bool => {
3076 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL);
3077 }
3078 AcceptedFieldKind::Date => write_hash_tag_u8(hasher, 4),
3079 AcceptedFieldKind::Decimal { scale } => {
3080 write_hash_tag_u8(hasher, 5);
3081 write_hash_u32(hasher, *scale);
3082 }
3083 AcceptedFieldKind::Duration => write_hash_tag_u8(hasher, 6),
3084 AcceptedFieldKind::Enum { type_id } => {
3085 write_hash_tag_u8(hasher, 7);
3086 write_hash_u32(hasher, type_id.get());
3087 }
3088 AcceptedFieldKind::Float32 => write_hash_tag_u8(hasher, 8),
3089 AcceptedFieldKind::Float64 => write_hash_tag_u8(hasher, 9),
3090 AcceptedFieldKind::Int8 => write_hash_tag_u8(hasher, 10),
3091 AcceptedFieldKind::Int16 => write_hash_tag_u8(hasher, 11),
3092 AcceptedFieldKind::Int32 => write_hash_tag_u8(hasher, 12),
3093 AcceptedFieldKind::Int64 => write_hash_tag_u8(hasher, 13),
3094 AcceptedFieldKind::Int128 => write_hash_tag_u8(hasher, 14),
3095 AcceptedFieldKind::IntBig { max_bytes } => {
3096 write_hash_tag_u8(hasher, 15);
3097 write_hash_u32(hasher, *max_bytes);
3098 }
3099 AcceptedFieldKind::Principal => write_hash_tag_u8(hasher, 16),
3100 AcceptedFieldKind::Subaccount => write_hash_tag_u8(hasher, 17),
3101 AcceptedFieldKind::Text { max_len } => {
3102 write_hash_tag_u8(hasher, 18);
3103 hash_optional_u32(hasher, *max_len);
3104 }
3105 AcceptedFieldKind::Timestamp => write_hash_tag_u8(hasher, 19),
3106 AcceptedFieldKind::Nat8 => write_hash_tag_u8(hasher, 20),
3107 AcceptedFieldKind::Nat16 => write_hash_tag_u8(hasher, 21),
3108 AcceptedFieldKind::Nat32 => write_hash_tag_u8(hasher, 22),
3109 AcceptedFieldKind::Nat64 => write_hash_tag_u8(hasher, 23),
3110 AcceptedFieldKind::Nat128 => write_hash_tag_u8(hasher, 24),
3111 AcceptedFieldKind::NatBig { max_bytes } => {
3112 write_hash_tag_u8(hasher, 25);
3113 write_hash_u32(hasher, *max_bytes);
3114 }
3115 AcceptedFieldKind::Ulid => write_hash_tag_u8(hasher, 26),
3116 AcceptedFieldKind::Unit => write_hash_tag_u8(hasher, 27),
3117 AcceptedFieldKind::Relation {
3118 target_path,
3119 target_entity_name,
3120 target_entity_tag,
3121 target_store_path,
3122 key_kind,
3123 } => {
3124 write_hash_tag_u8(hasher, 28);
3125 write_hash_str_u32(hasher, target_path);
3126 write_hash_str_u32(hasher, target_entity_name);
3127 write_hash_u64(hasher, target_entity_tag.value());
3128 write_hash_str_u32(hasher, target_store_path);
3129 hash_accepted_field_kind(hasher, key_kind);
3130 }
3131 AcceptedFieldKind::List(inner) => {
3132 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST);
3133 hash_accepted_field_kind(hasher, inner);
3134 }
3135 AcceptedFieldKind::Set(inner) => {
3136 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET);
3137 hash_accepted_field_kind(hasher, inner);
3138 }
3139 AcceptedFieldKind::Map { key, value } => {
3140 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP);
3141 hash_accepted_field_kind(hasher, key);
3142 hash_accepted_field_kind(hasher, value);
3143 }
3144 AcceptedFieldKind::Composite { type_id } => {
3145 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE);
3146 write_hash_u32(hasher, type_id.get());
3147 }
3148 }
3149}
3150
3151fn hash_optional_u32(hasher: &mut sha2::Sha256, value: Option<u32>) {
3152 match value {
3153 Some(value) => {
3154 write_hash_tag_u8(hasher, 1);
3155 write_hash_u32(hasher, value);
3156 }
3157 None => write_hash_tag_u8(hasher, 0),
3158 }
3159}
3160
3161const fn persisted_index_origin_name(
3162 origin: crate::db::schema::PersistedIndexOrigin,
3163) -> &'static str {
3164 match origin {
3165 crate::db::schema::PersistedIndexOrigin::Generated => "generated",
3166 crate::db::schema::PersistedIndexOrigin::SqlDdl => "sql_ddl",
3167 }
3168}
3169
3170const fn persisted_expression_op_name(
3171 op: crate::db::schema::PersistedIndexExpressionOp,
3172) -> &'static str {
3173 match op {
3174 crate::db::schema::PersistedIndexExpressionOp::Lower => "lower",
3175 crate::db::schema::PersistedIndexExpressionOp::Upper => "upper",
3176 crate::db::schema::PersistedIndexExpressionOp::Trim => "trim",
3177 crate::db::schema::PersistedIndexExpressionOp::LowerTrim => "lower_trim",
3178 crate::db::schema::PersistedIndexExpressionOp::Date => "date",
3179 crate::db::schema::PersistedIndexExpressionOp::Year => "year",
3180 crate::db::schema::PersistedIndexExpressionOp::Month => "month",
3181 crate::db::schema::PersistedIndexExpressionOp::Day => "day",
3182 }
3183}
3184
3185#[cfg(test)]
3190mod tests;