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 pub(in crate::db) fn preflight_fold_recovered_journal(&self) -> Result<(), InternalError> {
859 match self.backend {
860 SchemaStoreBackend::Journaled { .. } => Ok(()),
861 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
862 }
863 }
864
865 fn prepare_identity_state_transition(
866 &self,
867 incarnation: DatabaseIncarnationId,
868 candidate: &CandidateSchemaRevision,
869 view: IdentityStateStorageView,
870 ) -> Result<IdentityStateTransition, InternalError> {
871 let current = match view {
872 IdentityStateStorageView::Effective => self
873 .current_accepted_schema_bundle_ref()?
874 .as_ref()
875 .map(|bundle| (*bundle).clone()),
876 IdentityStateStorageView::Canonical => {
877 self.current_canonical_accepted_schema_bundle()?
878 }
879 };
880 let inventory = self.identity_state_inventory(view)?;
881 prepare_identity_state_transition(
882 incarnation,
883 current.as_ref(),
884 candidate.bundle(),
885 inventory,
886 )
887 }
888
889 fn validate_identity_state_closure(
890 &self,
891 bundle: &AcceptedSchemaRevisionBundle,
892 ) -> Result<(), InternalError> {
893 let inventory = self.identity_state_inventory(IdentityStateStorageView::Effective)?;
894 validate_identity_state_closure(bundle, &inventory)
895 }
896
897 pub(in crate::db) fn identity_statement_cursor(
899 &self,
900 database_incarnation_id: DatabaseIncarnationId,
901 entity_tag: EntityTag,
902 field_id: FieldId,
903 accepted_kind: &AcceptedFieldKind,
904 ) -> Result<IdentityStatementCursor, InternalError> {
905 let key = RawSchemaKey::from_identity_state(entity_tag, field_id);
906 let raw = self
907 .get_raw_snapshot(&key)
908 .ok_or_else(InternalError::identity_state_corruption)?;
909 let state = decode_identity_state(raw.as_bytes())?;
910 let owner = state.owner();
911 if owner.database_incarnation_id() != database_incarnation_id
912 || owner.entity_tag() != entity_tag
913 || owner.field_id() != field_id
914 || state.accepted_kind() != accepted_kind
915 || state.lifecycle() != IdentityStateLifecycle::Active
916 {
917 return Err(InternalError::identity_state_corruption());
918 }
919 IdentityStatementCursor::from_active_state(&state)
920 }
921
922 pub(in crate::db) fn identity_high_water_for_integrity(
924 &self,
925 database_incarnation_id: DatabaseIncarnationId,
926 entity_tag: EntityTag,
927 field_id: FieldId,
928 accepted_kind: &AcceptedFieldKind,
929 ) -> Result<u128, InternalError> {
930 let key = RawSchemaKey::from_identity_state(entity_tag, field_id);
931 let raw = self
932 .get_raw_snapshot(&key)
933 .ok_or_else(InternalError::identity_state_corruption)?;
934 let state = decode_identity_state(raw.as_bytes())?;
935 let owner = state.owner();
936 if owner.database_incarnation_id() != database_incarnation_id
937 || owner.entity_tag() != entity_tag
938 || owner.field_id() != field_id
939 || state.accepted_kind() != accepted_kind
940 || state.lifecycle() != IdentityStateLifecycle::Active
941 {
942 return Err(InternalError::identity_state_corruption());
943 }
944 Ok(state.materialized_high_water())
945 }
946
947 pub(in crate::db) fn preflight_identity_range_advance(
949 &self,
950 range: IdentityRangeAdvance,
951 ) -> Result<(), InternalError> {
952 let state =
953 self.identity_state_for_owner(range.owner(), IdentityStateStorageView::Effective)?;
954 if state.lifecycle() != IdentityStateLifecycle::Active
955 || range.new_high_water()
956 > identity_kind_maximum(state.accepted_kind())
957 .ok_or_else(InternalError::identity_state_corruption)?
958 {
959 return Err(InternalError::identity_state_corruption());
960 }
961 if state.materialized_high_water() != range.expected_high_water() {
962 return Err(InternalError::identity_state_conflict());
963 }
964 Ok(())
965 }
966
967 pub(in crate::db) fn apply_identity_range_advance(
969 &mut self,
970 range: IdentityRangeAdvance,
971 advance_id: IdentityAdvanceId,
972 ) -> Result<(), InternalError> {
973 self.apply_identity_range_advance_to(
974 range,
975 advance_id,
976 IdentityStateStorageView::Effective,
977 IdentityStateWriteTarget::Materialized,
978 )
979 }
980
981 pub(in crate::db) fn fold_identity_range_advance(
983 &mut self,
984 range: IdentityRangeAdvance,
985 advance_id: IdentityAdvanceId,
986 ) -> Result<(), InternalError> {
987 self.apply_identity_range_advance_to(
988 range,
989 advance_id,
990 IdentityStateStorageView::Canonical,
991 IdentityStateWriteTarget::Canonical,
992 )
993 }
994
995 pub(in crate::db) fn preflight_fold_identity_range_advance(
997 &self,
998 range: IdentityRangeAdvance,
999 advance_id: IdentityAdvanceId,
1000 ) -> Result<(), InternalError> {
1001 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1002 return Err(InternalError::store_invariant());
1003 }
1004 let state =
1005 self.identity_state_for_owner(range.owner(), IdentityStateStorageView::Canonical)?;
1006 let advanced = state.apply_range_advance(range, advance_id)?;
1007 let _encoded = encode_identity_state(&advanced)?;
1008 Ok(())
1009 }
1010
1011 pub(in crate::db) fn verify_identity_range_advance(
1013 &self,
1014 range: IdentityRangeAdvance,
1015 advance_id: IdentityAdvanceId,
1016 ) -> Result<(), InternalError> {
1017 let state =
1018 self.identity_state_for_owner(range.owner(), IdentityStateStorageView::Effective)?;
1019 if state.materialized_high_water() != range.new_high_water()
1020 || state.last_applied_advance() != Some(advance_id)
1021 {
1022 return Err(InternalError::recovery_effect_verification_failed());
1023 }
1024 Ok(())
1025 }
1026
1027 pub(in crate::db) fn identity_range_commit_state(
1029 &self,
1030 range: IdentityRangeAdvance,
1031 advance_id: IdentityAdvanceId,
1032 canonical: bool,
1033 ) -> Result<IdentityRangeCommitState, InternalError> {
1034 let view = if canonical {
1035 IdentityStateStorageView::Canonical
1036 } else {
1037 IdentityStateStorageView::Effective
1038 };
1039 self.identity_state_for_owner(range.owner(), view)?
1040 .range_commit_state(range, advance_id)
1041 }
1042
1043 pub(in crate::db) fn identity_state_inventory_for_integrity(
1046 &self,
1047 incarnation: DatabaseIncarnationId,
1048 ) -> Result<Vec<IdentityState>, InternalError> {
1049 let has_accepted_bundle = self.current_accepted_schema_bundle_ref()?.is_some();
1050 let inventory = self.identity_state_inventory(IdentityStateStorageView::Effective)?;
1051 if !has_accepted_bundle && !inventory.is_empty() {
1052 return Err(InternalError::identity_state_corruption());
1053 }
1054 if inventory
1055 .values()
1056 .any(|state| state.owner().database_incarnation_id() != incarnation)
1057 {
1058 return Err(InternalError::identity_state_corruption());
1059 }
1060 Ok(inventory.into_values().collect())
1061 }
1062
1063 fn identity_state_for_owner(
1064 &self,
1065 owner: crate::db::schema::identity_state::IdentityStateOwner,
1066 view: IdentityStateStorageView,
1067 ) -> Result<IdentityState, InternalError> {
1068 let key = RawSchemaKey::from_identity_state(owner.entity_tag(), owner.field_id());
1069 let raw = match view {
1070 IdentityStateStorageView::Effective => self.get_raw_snapshot(&key),
1071 IdentityStateStorageView::Canonical => self.get_canonical_raw_value(&key)?,
1072 }
1073 .ok_or_else(InternalError::identity_state_corruption)?;
1074 let state = decode_identity_state(raw.as_bytes())?;
1075 if state.owner() != owner {
1076 return Err(InternalError::identity_state_corruption());
1077 }
1078 Ok(state)
1079 }
1080
1081 fn apply_identity_range_advance_to(
1082 &mut self,
1083 range: IdentityRangeAdvance,
1084 advance_id: IdentityAdvanceId,
1085 view: IdentityStateStorageView,
1086 target: IdentityStateWriteTarget,
1087 ) -> Result<(), InternalError> {
1088 let state = self.identity_state_for_owner(range.owner(), view)?;
1089 let advanced = state.apply_range_advance(range, advance_id)?;
1090 let key = RawSchemaKey::from_identity_state(
1091 advanced.owner().entity_tag(),
1092 advanced.owner().field_id(),
1093 );
1094 let bytes = encode_identity_state(&advanced)?;
1095 match target {
1096 IdentityStateWriteTarget::Materialized => {
1097 self.insert_raw_snapshot(
1098 key,
1099 RawSchemaSnapshot::from_encoded_control_record(bytes),
1100 );
1101 }
1102 IdentityStateWriteTarget::Canonical => {
1103 self.insert_canonical_raw_value(key, bytes)?;
1104 }
1105 IdentityStateWriteTarget::Durable => {
1106 return Err(InternalError::store_invariant());
1107 }
1108 }
1109 Ok(())
1110 }
1111
1112 fn identity_state_inventory(
1113 &self,
1114 view: IdentityStateStorageView,
1115 ) -> Result<IdentityStateInventory, InternalError> {
1116 let bounds = RawSchemaKey::all_identity_state_range_bounds();
1117 let mut inventory = StdBTreeMap::new();
1118 let mut collect = |key: &RawSchemaKey,
1119 raw: &RawSchemaSnapshot|
1120 -> Result<SchemaStoreVisit, InternalError> {
1121 if inventory.len() >= MAX_IDENTITY_STATE_RECORDS_PER_DATABASE {
1122 return Err(InternalError::identity_state_corruption());
1123 }
1124 let state = decode_identity_state(raw.as_bytes())?;
1125 let state_key = (key.entity_tag(), FieldId::new(key.version()));
1126 if !key.is_identity_state()
1127 || state.owner().entity_tag() != state_key.0
1128 || state.owner().field_id() != state_key.1
1129 || inventory.insert(state_key, state).is_some()
1130 {
1131 return Err(InternalError::identity_state_corruption());
1132 }
1133 Ok(SchemaStoreVisit::Continue)
1134 };
1135
1136 match (&self.backend, view) {
1137 (SchemaStoreBackend::Heap(map), IdentityStateStorageView::Effective) => {
1138 for (key, raw) in map.range((bounds.0, bounds.1)) {
1139 collect(key, raw)?;
1140 }
1141 }
1142 (
1143 SchemaStoreBackend::Journaled {
1144 canonical,
1145 live,
1146 tombstones,
1147 },
1148 IdentityStateStorageView::Effective,
1149 ) => Self::visit_journaled_raw_snapshot_range(
1150 canonical,
1151 live,
1152 tombstones,
1153 bounds,
1154 Direction::Asc,
1155 &mut collect,
1156 )?,
1157 (
1158 SchemaStoreBackend::Journaled { canonical, .. },
1159 IdentityStateStorageView::Canonical,
1160 ) => {
1161 for entry in canonical.range((bounds.0, bounds.1)) {
1162 collect(entry.key(), &entry.value())?;
1163 }
1164 }
1165 (SchemaStoreBackend::Heap(_), IdentityStateStorageView::Canonical) => {
1166 return Err(InternalError::store_invariant());
1167 }
1168 }
1169
1170 Ok(inventory)
1171 }
1172
1173 fn apply_identity_state_transition(
1174 &mut self,
1175 transition: IdentityStateTransition,
1176 target: IdentityStateWriteTarget,
1177 ) -> Result<(), InternalError> {
1178 for state in transition.into_updates() {
1179 let key = RawSchemaKey::from_identity_state(
1180 state.owner().entity_tag(),
1181 state.owner().field_id(),
1182 );
1183 let bytes = encode_identity_state(&state)?;
1184 match target {
1185 IdentityStateWriteTarget::Durable => {
1186 self.insert_durable_raw_value(key, bytes);
1187 }
1188 IdentityStateWriteTarget::Materialized => {
1189 self.insert_raw_snapshot(
1190 key,
1191 RawSchemaSnapshot::from_encoded_control_record(bytes),
1192 );
1193 }
1194 IdentityStateWriteTarget::Canonical => {
1195 self.insert_canonical_raw_value(key, bytes)?;
1196 }
1197 }
1198 }
1199 Ok(())
1200 }
1201
1202 fn current_canonical_accepted_schema_bundle(
1203 &self,
1204 ) -> Result<Option<AcceptedSchemaRevisionBundle>, InternalError> {
1205 let first = self.canonical_root_slot_bytes(0)?;
1206 let second = self.canonical_root_slot_bytes(1)?;
1207 let Some(selection) =
1208 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1209 else {
1210 return Ok(None);
1211 };
1212 let bundle_key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
1213 let raw = self
1214 .get_canonical_raw_value(&bundle_key)?
1215 .ok_or_else(InternalError::store_corruption)?;
1216 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes()).map(Some)
1217 }
1218
1219 pub(in crate::db) fn insert_persisted_snapshot(
1221 &mut self,
1222 entity: EntityTag,
1223 snapshot: &PersistedSchemaSnapshot,
1224 ) -> Result<(), InternalError> {
1225 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
1226 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1227 let _ = self.insert_raw_snapshot(key, raw_snapshot);
1228
1229 Ok(())
1230 }
1231
1232 pub(in crate::db) fn constraint_validation_job(
1234 &self,
1235 entity: EntityTag,
1236 constraint_id: ConstraintId,
1237 ) -> Result<Option<ConstraintValidationJob>, InternalError> {
1238 let key = RawSchemaKey::from_constraint_validation_job(entity, constraint_id);
1239 self.get_raw_snapshot(&key)
1240 .map(|raw| decode_constraint_validation_job(raw.as_bytes()))
1241 .transpose()
1242 }
1243
1244 pub(in crate::db) fn apply_constraint_validation_job(
1246 &mut self,
1247 job: &ConstraintValidationJob,
1248 ) -> Result<(), InternalError> {
1249 let key =
1250 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id());
1251 let bytes = encode_constraint_validation_job(job)?;
1252 let _ =
1253 self.insert_raw_snapshot(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
1254 Ok(())
1255 }
1256
1257 #[expect(
1259 clippy::unnecessary_wraps,
1260 reason = "marker apply operations share one fallible callback contract"
1261 )]
1262 pub(in crate::db) fn apply_constraint_validation_job_removal(
1263 &mut self,
1264 entity: EntityTag,
1265 constraint_id: ConstraintId,
1266 ) -> Result<(), InternalError> {
1267 let key = RawSchemaKey::from_constraint_validation_job(entity, constraint_id);
1268 match &mut self.backend {
1269 SchemaStoreBackend::Heap(map) => {
1270 map.remove(&key);
1271 }
1272 SchemaStoreBackend::Journaled {
1273 live, tombstones, ..
1274 } => {
1275 live.remove(&key);
1276 tombstones.insert(key);
1277 }
1278 }
1279 Ok(())
1280 }
1281
1282 pub(in crate::db) fn fold_constraint_validation_job(
1284 &mut self,
1285 job: &ConstraintValidationJob,
1286 ) -> Result<(), InternalError> {
1287 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1288 return Err(InternalError::store_invariant());
1289 };
1290 let key =
1291 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id());
1292 let bytes = encode_constraint_validation_job(job)?;
1293 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
1294 Ok(())
1295 }
1296
1297 pub(in crate::db) fn preflight_fold_constraint_validation_job(
1299 &self,
1300 job: &ConstraintValidationJob,
1301 ) -> Result<(), InternalError> {
1302 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1303 return Err(InternalError::store_invariant());
1304 }
1305 let _encoded = encode_constraint_validation_job(job)?;
1306 Ok(())
1307 }
1308
1309 pub(in crate::db) fn fold_constraint_validation_job_removal(
1311 &mut self,
1312 entity: EntityTag,
1313 constraint_id: ConstraintId,
1314 ) -> Result<(), InternalError> {
1315 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1316 return Err(InternalError::store_invariant());
1317 };
1318 canonical.remove(&RawSchemaKey::from_constraint_validation_job(
1319 entity,
1320 constraint_id,
1321 ));
1322 Ok(())
1323 }
1324
1325 pub(in crate::db) fn preflight_fold_constraint_validation_job_removal(
1327 &self,
1328 ) -> Result<(), InternalError> {
1329 match self.backend {
1330 SchemaStoreBackend::Journaled { .. } => Ok(()),
1331 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
1332 }
1333 }
1334
1335 pub(in crate::db) fn reset_journaled_live_projection(&mut self) -> Result<(), InternalError> {
1338 let SchemaStoreBackend::Journaled {
1339 live, tombstones, ..
1340 } = &mut self.backend
1341 else {
1342 return Err(InternalError::store_invariant());
1343 };
1344
1345 live.clear();
1346 tombstones.clear();
1347 self.accepted_bundle_cache.get_mut().take();
1348
1349 Ok(())
1350 }
1351
1352 pub(in crate::db) fn fold_persisted_snapshot(
1354 &mut self,
1355 entity: EntityTag,
1356 snapshot: &PersistedSchemaSnapshot,
1357 ) -> Result<(), InternalError> {
1358 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1359 return Err(InternalError::store_invariant());
1360 };
1361
1362 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
1363 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1364 canonical.insert(key, raw_snapshot);
1365
1366 Ok(())
1367 }
1368
1369 pub(in crate::db) fn preflight_fold_persisted_snapshot(
1371 &self,
1372 snapshot: &PersistedSchemaSnapshot,
1373 ) -> Result<(), InternalError> {
1374 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1375 return Err(InternalError::store_invariant());
1376 }
1377 let _encoded = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1378 Ok(())
1379 }
1380
1381 pub(in crate::db) fn current_accepted_schema_root(
1383 &self,
1384 ) -> Result<Option<AcceptedSchemaRootSelection>, InternalError> {
1385 let first = self.accepted_root_slot_bytes(0)?;
1386 let second = self.accepted_root_slot_bytes(1)?;
1387 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])
1388 }
1389
1390 pub(in crate::db) fn current_accepted_schema_bundle(
1392 &self,
1393 ) -> Result<Option<AcceptedSchemaRevisionBundle>, InternalError> {
1394 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1395 return Ok(None);
1396 };
1397 self.validate_constraint_validation_job_closure(&bundle)?;
1398 Ok(Some(bundle.clone()))
1399 }
1400
1401 pub(in crate::db) fn current_accepted_runtime_entities(
1403 &self,
1404 registered_store_path: &'static str,
1405 ) -> Result<Vec<AcceptedRuntimeEntity>, InternalError> {
1406 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1407 return Ok(Vec::new());
1408 };
1409 if bundle.store_path() != registered_store_path {
1410 return Err(InternalError::store_corruption());
1411 }
1412
1413 bundle
1414 .entity_snapshots()
1415 .iter()
1416 .map(|(entity_tag, snapshot)| {
1417 AcceptedRuntimeEntity::from_accepted_snapshot(
1418 &bundle,
1419 *entity_tag,
1420 snapshot,
1421 registered_store_path,
1422 )
1423 })
1424 .collect()
1425 }
1426
1427 pub(in crate::db) fn current_accepted_runtime_entity_for_tag(
1429 &self,
1430 registered_store_path: &'static str,
1431 entity_tag: EntityTag,
1432 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1433 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1434 return Ok(None);
1435 };
1436 if bundle.store_path() != registered_store_path {
1437 return Err(InternalError::store_corruption());
1438 }
1439 let Some(snapshot) = bundle.entity_snapshots().get(&entity_tag) else {
1440 return Ok(None);
1441 };
1442
1443 AcceptedRuntimeEntity::from_accepted_snapshot(
1444 &bundle,
1445 entity_tag,
1446 snapshot,
1447 registered_store_path,
1448 )
1449 .map(Some)
1450 }
1451
1452 pub(in crate::db) fn current_accepted_runtime_entity_for_path(
1454 &self,
1455 registered_store_path: &'static str,
1456 entity_path: &str,
1457 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1458 self.current_accepted_runtime_entity_matching(registered_store_path, |snapshot_path, _| {
1459 snapshot_path == entity_path
1460 })
1461 }
1462
1463 #[cfg(test)]
1465 pub(in crate::db) fn current_accepted_runtime_entity_for_name(
1466 &self,
1467 registered_store_path: &'static str,
1468 entity_name: &str,
1469 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1470 self.current_accepted_runtime_entity_matching(registered_store_path, |_, snapshot_name| {
1471 snapshot_name == entity_name
1472 })
1473 }
1474
1475 fn current_accepted_runtime_entity_matching(
1476 &self,
1477 registered_store_path: &'static str,
1478 mut predicate: impl FnMut(&str, &str) -> bool,
1479 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1480 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1481 return Ok(None);
1482 };
1483 if bundle.store_path() != registered_store_path {
1484 return Err(InternalError::store_corruption());
1485 }
1486
1487 let mut matched = None;
1488 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1489 if !predicate(snapshot.entity_path(), snapshot.entity_name()) {
1490 continue;
1491 }
1492 let entity = AcceptedRuntimeEntity::from_accepted_snapshot(
1493 &bundle,
1494 *entity_tag,
1495 snapshot,
1496 registered_store_path,
1497 )?;
1498 if matched.replace(entity).is_some() {
1499 return Err(InternalError::store_corruption());
1500 }
1501 }
1502
1503 Ok(matched)
1504 }
1505
1506 pub(in crate::db) fn current_accepted_schema_revision(
1508 &self,
1509 ) -> Result<Option<AcceptedSchemaRevision>, InternalError> {
1510 Ok(self
1511 .current_accepted_schema_root()?
1512 .map(|selection| selection.root().revision()))
1513 }
1514
1515 pub(in crate::db) fn pending_relation_activation_for_target(
1521 &self,
1522 target_path: &str,
1523 ) -> Result<Option<PendingRelationActivationDeleteBarrier>, InternalError> {
1524 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1525 return Ok(None);
1526 };
1527 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1528 let Some(candidate) = snapshot
1529 .candidate_relations()
1530 .iter()
1531 .find(|candidate| candidate.target_path() == target_path)
1532 else {
1533 continue;
1534 };
1535 let activation = snapshot
1536 .constraint_activations()
1537 .iter()
1538 .find(|activation| {
1539 matches!(
1540 activation.kind(),
1541 ConstraintActivationKind::Relation { relation_id }
1542 if *relation_id == candidate.id()
1543 )
1544 })
1545 .ok_or_else(InternalError::store_corruption)?;
1546 return Ok(Some(PendingRelationActivationDeleteBarrier {
1547 accepted_schema_fingerprint:
1548 accepted_schema_cache_fingerprint_for_persisted_snapshot(snapshot)?,
1549 source_entity_tag: *entity_tag,
1550 constraint_id: activation.id(),
1551 }));
1552 }
1553
1554 Ok(None)
1555 }
1556
1557 pub(in crate::db) fn entity_has_relation_to_target(
1559 &self,
1560 source_entity: EntityTag,
1561 target_path: &str,
1562 ) -> Result<bool, InternalError> {
1563 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1564 return Ok(false);
1565 };
1566 let Some(snapshot) = bundle.entity_snapshots().get(&source_entity) else {
1567 return Ok(false);
1568 };
1569
1570 Ok(snapshot
1571 .relations()
1572 .iter()
1573 .any(|relation| relation.target_path() == target_path))
1574 }
1575
1576 pub(in crate::db) fn validate_live_activation_transition(
1578 &self,
1579 candidate: &AcceptedSchemaRevisionBundle,
1580 ) -> Result<(), InternalError> {
1581 let Some(current) = self.current_accepted_schema_bundle()? else {
1582 return Ok(());
1583 };
1584 for (entity_tag, before) in current.entity_snapshots() {
1585 if before.constraint_activations().is_empty() {
1586 continue;
1587 }
1588 let after = candidate
1589 .entity_snapshots()
1590 .get(entity_tag)
1591 .ok_or_else(InternalError::store_invariant)?;
1592 if before == after {
1593 continue;
1594 }
1595 let expected_shape = before
1596 .clone()
1597 .with_constraint_catalog(after.constraint_catalog().clone());
1598 let catalog_only_transition = expected_shape == *after
1599 && before
1600 .constraint_catalog()
1601 .permits_live_activation_transition_to(after.constraint_catalog());
1602 let sql_row_local_abort_with_version =
1603 before.constraint_activations().iter().any(|activation| {
1604 activation.origin() == ConstraintOrigin::SqlDdl
1605 && matches!(
1606 activation.kind(),
1607 ConstraintActivationKind::Check { .. }
1608 | ConstraintActivationKind::NotNull { .. }
1609 )
1610 && before.version().get().checked_add(1) == Some(after.version().get())
1611 && before
1612 .constraint_catalog()
1613 .clone()
1614 .with_aborted_activation(activation.id())
1615 .is_ok_and(|catalog| catalog == *after.constraint_catalog())
1616 && before
1617 .clone()
1618 .with_constraint_catalog(after.constraint_catalog().clone())
1619 .with_schema_version(after.version())
1620 == *after
1621 });
1622 let sql_unique_abort_with_version =
1623 before.constraint_activations().iter().any(|activation| {
1624 activation.origin() == ConstraintOrigin::SqlDdl
1625 && matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
1626 && before.version().get().checked_add(1) == Some(after.version().get())
1627 && before
1628 .with_aborted_unique_activation(activation.id(), after.version())
1629 .is_ok_and(|expected| expected == *after)
1630 });
1631 let not_null_promotion = before.constraint_activations().iter().any(|activation| {
1632 matches!(activation.kind(), ConstraintActivationKind::NotNull { .. })
1633 && before
1634 .with_promoted_not_null_activation(activation.id(), after.version())
1635 .is_ok_and(|expected| expected == *after)
1636 });
1637 let unique_promotion = before.constraint_activations().iter().any(|activation| {
1638 matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
1639 && before
1640 .with_promoted_unique_activation(activation.id(), after.version())
1641 .is_ok_and(|expected| expected == *after)
1642 });
1643 let relation_promotion = before.constraint_activations().iter().any(|activation| {
1644 matches!(activation.kind(), ConstraintActivationKind::Relation { .. })
1645 && before
1646 .with_promoted_relation_activation(activation.id(), after.version())
1647 .is_ok_and(|expected| expected == *after)
1648 });
1649 if !catalog_only_transition
1650 && !sql_row_local_abort_with_version
1651 && !sql_unique_abort_with_version
1652 && !not_null_promotion
1653 && !unique_promotion
1654 && !relation_promotion
1655 {
1656 return Err(InternalError::store_invariant());
1657 }
1658 }
1659 Ok(())
1660 }
1661
1662 pub(in crate::db) fn validate_constraint_validation_job_closure(
1664 &self,
1665 bundle: &AcceptedSchemaRevisionBundle,
1666 ) -> Result<(), InternalError> {
1667 self.validate_constraint_validation_job_closure_with_change(bundle, None, None)
1668 }
1669
1670 pub(in crate::db) fn validate_constraint_validation_job_closure_with_change(
1673 &self,
1674 bundle: &AcceptedSchemaRevisionBundle,
1675 replacement: Option<&ConstraintValidationJob>,
1676 removal: Option<(EntityTag, ConstraintId)>,
1677 ) -> Result<(), InternalError> {
1678 if replacement.is_some() && removal.is_some() {
1679 return Err(InternalError::store_invariant());
1680 }
1681 let replacement_key = replacement.map(|job| {
1682 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id())
1683 });
1684 let removal_key = removal.map(|(entity_tag, constraint_id)| {
1685 RawSchemaKey::from_constraint_validation_job(entity_tag, constraint_id)
1686 });
1687 let mut expected = BTreeSet::new();
1688 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1689 for activation in snapshot.constraint_activations() {
1690 let key =
1691 RawSchemaKey::from_constraint_validation_job(*entity_tag, activation.id());
1692 match activation.state() {
1693 ConstraintActivationState::EnforcingNewWrites => {
1694 if self
1695 .constraint_validation_job_after_change(
1696 key,
1697 replacement,
1698 replacement_key,
1699 removal_key,
1700 )?
1701 .is_some()
1702 {
1703 return Err(InternalError::store_corruption());
1704 }
1705 }
1706 ConstraintActivationState::Validating => {
1707 let job = self
1708 .constraint_validation_job_after_change(
1709 key,
1710 replacement,
1711 replacement_key,
1712 removal_key,
1713 )?
1714 .ok_or_else(InternalError::store_corruption)?;
1715 if job.entity_tag() != *entity_tag
1716 || job.entity_path() != snapshot.entity_path()
1717 {
1718 return Err(InternalError::store_corruption());
1719 }
1720 job.validate(Some(activation))?;
1721 expected.insert(key);
1722 }
1723 }
1724 }
1725 }
1726
1727 self.visit_constraint_validation_jobs(|key, raw| {
1728 if removal_key == Some(*key) || replacement_key == Some(*key) {
1729 return Ok(SchemaStoreVisit::Continue);
1730 }
1731 if !expected.contains(key) {
1732 return Err(InternalError::store_corruption());
1733 }
1734 let job = decode_constraint_validation_job(raw.as_bytes())?;
1735 if job.entity_tag() != key.entity_tag()
1736 || key.constraint_id() != Some(job.constraint_id())
1737 {
1738 return Err(InternalError::store_corruption());
1739 }
1740 Ok(SchemaStoreVisit::Continue)
1741 })?;
1742
1743 if let Some(key) = replacement_key
1744 && !expected.contains(&key)
1745 {
1746 return Err(InternalError::store_corruption());
1747 }
1748 if let Some(key) = removal_key
1749 && expected.contains(&key)
1750 {
1751 return Err(InternalError::store_corruption());
1752 }
1753
1754 Ok(())
1755 }
1756
1757 fn constraint_validation_job_after_change(
1758 &self,
1759 key: RawSchemaKey,
1760 replacement: Option<&ConstraintValidationJob>,
1761 replacement_key: Option<RawSchemaKey>,
1762 removal_key: Option<RawSchemaKey>,
1763 ) -> Result<Option<ConstraintValidationJob>, InternalError> {
1764 if removal_key == Some(key) {
1765 return Ok(None);
1766 }
1767 if replacement_key == Some(key) {
1768 return Ok(replacement.cloned());
1769 }
1770 self.get_raw_snapshot(&key)
1771 .map(|raw| decode_constraint_validation_job(raw.as_bytes()))
1772 .transpose()
1773 }
1774
1775 pub(in crate::db) fn current_accepted_schema_authority_matches(
1778 &self,
1779 expected: &AcceptedSchemaAuthority,
1780 ) -> Result<bool, InternalError> {
1781 let Some(store_scope) = self.accepted_catalog_scope.get() else {
1782 return Ok(false);
1783 };
1784
1785 if let Some(cached) = self
1788 .accepted_bundle_cache
1789 .try_borrow()
1790 .map_err(|_| InternalError::store_invariant())?
1791 .as_ref()
1792 {
1793 let root = cached.selection.root();
1794 return Ok(expected.matches_store_root(
1795 store_scope,
1796 root.revision(),
1797 root.fingerprint(),
1798 ));
1799 }
1800
1801 let Some(selection) = self.current_accepted_schema_root()? else {
1802 return Ok(false);
1803 };
1804 let root = selection.root();
1805
1806 Ok(expected.matches_store_root(store_scope, root.revision(), root.fingerprint()))
1807 }
1808
1809 pub(in crate::db) fn publish_accepted_schema_candidate(
1815 &mut self,
1816 incarnation: DatabaseIncarnationId,
1817 expected_revision: AcceptedSchemaRevision,
1818 candidate: &CandidateSchemaRevision,
1819 ) -> Result<(), InternalError> {
1820 let identity_transition = self.prepare_identity_state_transition(
1821 incarnation,
1822 candidate,
1823 IdentityStateStorageView::Effective,
1824 )?;
1825 if self.current_root_matches_candidate(candidate)? {
1826 if !identity_transition.is_empty() {
1827 return Err(InternalError::identity_state_corruption());
1828 }
1829 let selection = self
1830 .current_accepted_schema_root()?
1831 .ok_or_else(InternalError::store_corruption)?;
1832 self.retain_durable_candidate_entries(candidate, selection.slot())?;
1833 return Ok(());
1834 }
1835 let first = self.accepted_root_slot_bytes(0)?;
1836 let second = self.accepted_root_slot_bytes(1)?;
1837 prepare_accepted_schema_root_publication(
1838 [first.as_deref(), second.as_deref()],
1839 expected_revision,
1840 candidate,
1841 )
1842 .map_err(map_schema_publication_error)?;
1843
1844 self.insert_durable_candidate_snapshots(candidate)?;
1845 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1846 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
1847 let persisted_bundle = self
1848 .get_raw_snapshot(&bundle_key)
1849 .ok_or_else(InternalError::store_corruption)?;
1850 let _verified = decode_verified_accepted_schema_revision_bundle(
1851 candidate.root(),
1852 persisted_bundle.as_bytes(),
1853 )?;
1854 self.apply_identity_state_transition(
1855 identity_transition,
1856 IdentityStateWriteTarget::Durable,
1857 )?;
1858
1859 let first = self.accepted_root_slot_bytes(0)?;
1862 let second = self.accepted_root_slot_bytes(1)?;
1863 let publication = prepare_accepted_schema_root_publication(
1864 [first.as_deref(), second.as_deref()],
1865 expected_revision,
1866 candidate,
1867 )
1868 .map_err(map_schema_publication_error)?;
1869 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
1870 self.insert_durable_raw_value(root_key, publication.encoded_root().to_vec());
1871
1872 let selected = self
1873 .current_accepted_schema_root()?
1874 .ok_or_else(InternalError::store_corruption)?;
1875 if selected.root() != candidate.root() {
1876 return Err(InternalError::store_corruption());
1877 }
1878 self.retain_durable_candidate_entries(candidate, selected.slot())?;
1879 Ok(())
1880 }
1881
1882 pub(in crate::db) fn restore_live_accepted_schema_checkpoint(
1885 &mut self,
1886 incarnation: DatabaseIncarnationId,
1887 candidate: &CandidateSchemaRevision,
1888 checkpoint_identity_states: &IdentityStateInventory,
1889 ) -> Result<(), InternalError> {
1890 if !matches!(self.backend, SchemaStoreBackend::Heap(_)) {
1891 return Err(InternalError::store_invariant());
1892 }
1893 let checkpoint_validation = prepare_identity_state_transition(
1894 incarnation,
1895 Some(candidate.bundle()),
1896 candidate.bundle(),
1897 checkpoint_identity_states.clone(),
1898 )?;
1899 if !checkpoint_validation.is_empty() {
1900 return Err(InternalError::identity_state_corruption());
1901 }
1902 if self.current_root_matches_candidate(candidate)? {
1903 for state in checkpoint_identity_states.values() {
1904 let key = RawSchemaKey::from_identity_state(
1905 state.owner().entity_tag(),
1906 state.owner().field_id(),
1907 );
1908 self.insert_durable_raw_value(key, encode_identity_state(state)?);
1909 }
1910 if self.identity_state_inventory(IdentityStateStorageView::Effective)?
1911 != *checkpoint_identity_states
1912 {
1913 return Err(InternalError::identity_state_corruption());
1914 }
1915 let selection = self
1916 .current_accepted_schema_root()?
1917 .ok_or_else(InternalError::store_corruption)?;
1918 self.retain_durable_candidate_entries(candidate, selection.slot())?;
1919 return Ok(());
1920 }
1921 if self.current_accepted_schema_root()?.is_some()
1922 || !self
1923 .identity_state_inventory(IdentityStateStorageView::Effective)?
1924 .is_empty()
1925 {
1926 return Err(InternalError::store_corruption());
1927 }
1928
1929 self.insert_durable_candidate_snapshots(candidate)?;
1930 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1931 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
1932 for state in checkpoint_identity_states.values() {
1933 let key = RawSchemaKey::from_identity_state(
1934 state.owner().entity_tag(),
1935 state.owner().field_id(),
1936 );
1937 self.insert_durable_raw_value(key, encode_identity_state(state)?);
1938 }
1939 let root_key = RawSchemaKey::from_accepted_root_slot(0)?;
1940 self.insert_durable_raw_value(root_key, candidate.encoded_root().to_vec());
1941
1942 let selected = self
1943 .current_accepted_schema_root()?
1944 .ok_or_else(InternalError::store_corruption)?;
1945 if selected.root() != candidate.root() {
1946 return Err(InternalError::store_corruption());
1947 }
1948 self.retain_durable_candidate_entries(candidate, selected.slot())?;
1949 Ok(())
1950 }
1951
1952 pub(in crate::db) fn preflight_accepted_schema_candidate(
1959 &self,
1960 incarnation: DatabaseIncarnationId,
1961 expected_revision: AcceptedSchemaRevision,
1962 candidate: &CandidateSchemaRevision,
1963 ) -> Result<bool, InternalError> {
1964 let identity_transition = self.prepare_identity_state_transition(
1965 incarnation,
1966 candidate,
1967 IdentityStateStorageView::Effective,
1968 )?;
1969 if self.current_root_matches_candidate(candidate)? {
1970 if !identity_transition.is_empty() {
1971 return Err(InternalError::identity_state_corruption());
1972 }
1973 return Ok(true);
1974 }
1975 let first = self.accepted_root_slot_bytes(0)?;
1976 let second = self.accepted_root_slot_bytes(1)?;
1977 prepare_accepted_schema_root_publication(
1978 [first.as_deref(), second.as_deref()],
1979 expected_revision,
1980 candidate,
1981 )
1982 .map_err(map_schema_publication_error)?;
1983
1984 Ok(false)
1985 }
1986
1987 pub(in crate::db) fn preflight_fold_journaled_accepted_schema_candidate(
1989 &self,
1990 incarnation: DatabaseIncarnationId,
1991 expected_revision: AcceptedSchemaRevision,
1992 candidate: &CandidateSchemaRevision,
1993 ) -> Result<(), InternalError> {
1994 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1995 return Err(InternalError::store_invariant());
1996 }
1997 let identity_transition = self.prepare_identity_state_transition(
1998 incarnation,
1999 candidate,
2000 IdentityStateStorageView::Canonical,
2001 )?;
2002 let candidate_is_current = self.canonical_root_matches_candidate(candidate)?;
2003 if candidate_is_current && !identity_transition.is_empty() {
2004 return Err(InternalError::identity_state_corruption());
2005 }
2006 for state in identity_transition.into_updates() {
2007 let _encoded = encode_identity_state(&state)?;
2008 }
2009 for snapshot in candidate.bundle().entity_snapshots().values() {
2010 let _encoded = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2011 }
2012
2013 let first = self.canonical_root_slot_bytes(0)?;
2014 let second = self.canonical_root_slot_bytes(1)?;
2015 let root_slot = if candidate_is_current {
2016 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2017 .ok_or_else(InternalError::store_corruption)?
2018 .slot()
2019 } else {
2020 prepare_accepted_schema_root_publication(
2021 [first.as_deref(), second.as_deref()],
2022 expected_revision,
2023 candidate,
2024 )
2025 .map_err(map_schema_publication_error)?
2026 .target_slot()
2027 };
2028 let _retained = Self::candidate_entry_keys(candidate, root_slot)?;
2029 Ok(())
2030 }
2031
2032 pub(in crate::db) fn projected_identity_state_count(
2034 &self,
2035 incarnation: DatabaseIncarnationId,
2036 candidate: &CandidateSchemaRevision,
2037 ) -> Result<usize, InternalError> {
2038 Ok(self
2039 .prepare_identity_state_transition(
2040 incarnation,
2041 candidate,
2042 IdentityStateStorageView::Effective,
2043 )?
2044 .projected_inventory_len())
2045 }
2046
2047 pub(in crate::db) fn apply_journaled_accepted_schema_candidate(
2049 &mut self,
2050 incarnation: DatabaseIncarnationId,
2051 expected_revision: AcceptedSchemaRevision,
2052 candidate: &CandidateSchemaRevision,
2053 ) -> Result<(), InternalError> {
2054 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
2055 return Err(InternalError::store_invariant());
2056 }
2057 let identity_transition = self.prepare_identity_state_transition(
2058 incarnation,
2059 candidate,
2060 IdentityStateStorageView::Effective,
2061 )?;
2062 if self.current_root_matches_candidate(candidate)? {
2063 if !identity_transition.is_empty() {
2064 return Err(InternalError::identity_state_corruption());
2065 }
2066 let selection = self
2067 .current_accepted_schema_root()?
2068 .ok_or_else(InternalError::store_corruption)?;
2069 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
2070 return Ok(());
2071 }
2072
2073 let first = self.accepted_root_slot_bytes(0)?;
2074 let second = self.accepted_root_slot_bytes(1)?;
2075 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
2082 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2083 self.insert_persisted_snapshot(*entity_tag, snapshot)?;
2084 }
2085 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2086 self.insert_raw_snapshot(
2087 bundle_key,
2088 RawSchemaSnapshot::from_encoded_control_record(candidate.encoded_bundle().to_vec()),
2089 );
2090 let persisted_bundle = self
2091 .get_raw_snapshot(&bundle_key)
2092 .ok_or_else(InternalError::store_corruption)?;
2093 let _verified = decode_verified_accepted_schema_revision_bundle(
2094 candidate.root(),
2095 persisted_bundle.as_bytes(),
2096 )?;
2097 self.apply_identity_state_transition(
2098 identity_transition,
2099 IdentityStateWriteTarget::Materialized,
2100 )?;
2101
2102 let first = self.accepted_root_slot_bytes(0)?;
2103 let second = self.accepted_root_slot_bytes(1)?;
2104 let publication = prepare_accepted_schema_root_publication(
2105 [first.as_deref(), second.as_deref()],
2106 expected_revision,
2107 candidate,
2108 )
2109 .map_err(map_schema_publication_error)?;
2110 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
2111 self.insert_raw_snapshot(
2112 root_key,
2113 RawSchemaSnapshot::from_encoded_control_record(publication.encoded_root().to_vec()),
2114 );
2115
2116 if !self.current_root_matches_candidate(candidate)? {
2117 return Err(InternalError::store_corruption());
2118 }
2119 let selection = self
2120 .current_accepted_schema_root()?
2121 .ok_or_else(InternalError::store_corruption)?;
2122 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
2123 Ok(())
2124 }
2125
2126 pub(in crate::db) fn fold_journaled_accepted_schema_candidate(
2128 &mut self,
2129 incarnation: DatabaseIncarnationId,
2130 expected_revision: AcceptedSchemaRevision,
2131 candidate: &CandidateSchemaRevision,
2132 ) -> Result<(), InternalError> {
2133 let identity_transition = self.prepare_identity_state_transition(
2134 incarnation,
2135 candidate,
2136 IdentityStateStorageView::Canonical,
2137 )?;
2138 if self.canonical_root_matches_candidate(candidate)? {
2139 if !identity_transition.is_empty() {
2140 return Err(InternalError::identity_state_corruption());
2141 }
2142 let first = self.canonical_root_slot_bytes(0)?;
2143 let second = self.canonical_root_slot_bytes(1)?;
2144 let selection =
2145 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2146 .ok_or_else(InternalError::store_corruption)?;
2147 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
2148 return Ok(());
2149 }
2150
2151 let first = self.canonical_root_slot_bytes(0)?;
2152 let second = self.canonical_root_slot_bytes(1)?;
2153 prepare_accepted_schema_root_publication(
2154 [first.as_deref(), second.as_deref()],
2155 expected_revision,
2156 candidate,
2157 )
2158 .map_err(map_schema_publication_error)?;
2159
2160 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2161 self.fold_persisted_snapshot(*entity_tag, snapshot)?;
2162 }
2163 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2164 self.insert_canonical_raw_value(bundle_key, candidate.encoded_bundle().to_vec())?;
2165 let persisted_bundle = self
2166 .get_canonical_raw_value(&bundle_key)?
2167 .ok_or_else(InternalError::store_corruption)?;
2168 let _verified = decode_verified_accepted_schema_revision_bundle(
2169 candidate.root(),
2170 persisted_bundle.as_bytes(),
2171 )?;
2172 self.apply_identity_state_transition(
2173 identity_transition,
2174 IdentityStateWriteTarget::Canonical,
2175 )?;
2176
2177 let first = self.canonical_root_slot_bytes(0)?;
2178 let second = self.canonical_root_slot_bytes(1)?;
2179 let publication = prepare_accepted_schema_root_publication(
2180 [first.as_deref(), second.as_deref()],
2181 expected_revision,
2182 candidate,
2183 )
2184 .map_err(map_schema_publication_error)?;
2185 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
2186 self.insert_canonical_raw_value(root_key, publication.encoded_root().to_vec())?;
2187
2188 if !self.canonical_root_matches_candidate(candidate)? {
2189 return Err(InternalError::store_corruption());
2190 }
2191 let first = self.canonical_root_slot_bytes(0)?;
2192 let second = self.canonical_root_slot_bytes(1)?;
2193 let selection = select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2194 .ok_or_else(InternalError::store_corruption)?;
2195 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
2196 Ok(())
2197 }
2198
2199 pub(in crate::db) fn get_persisted_snapshot(
2201 &self,
2202 entity: EntityTag,
2203 version: SchemaVersion,
2204 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2205 let key = RawSchemaKey::from_entity_version(entity, version);
2206 self.get_raw_snapshot(&key)
2207 .map(|snapshot| snapshot.decode_persisted_snapshot())
2208 .transpose()
2209 }
2210
2211 #[cfg(test)]
2212 fn latest_staged_persisted_snapshot(
2213 &self,
2214 entity: EntityTag,
2215 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2216 self.latest_raw_snapshots_by_entity()
2217 .remove(&entity)
2218 .map(|(_, snapshot)| snapshot.decode_persisted_snapshot())
2219 .transpose()
2220 }
2221
2222 pub(in crate::db) fn current_accepted_persisted_snapshot(
2225 &self,
2226 entity: EntityTag,
2227 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2228 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2229 return Ok(None);
2230 };
2231
2232 Ok(bundle.entity_snapshots().get(&entity).cloned())
2233 }
2234
2235 pub(in crate::db) fn current_accepted_catalog_selection(
2237 &self,
2238 entity: EntityTag,
2239 entity_path: &str,
2240 store_path: &'static str,
2241 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
2242 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2243 return Ok(None);
2244 };
2245 if bundle.store_path() != store_path {
2246 return Err(InternalError::store_corruption());
2247 }
2248 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
2249 return Ok(None);
2250 };
2251 if snapshot.entity_path() != entity_path {
2252 return Err(InternalError::store_corruption());
2253 }
2254
2255 let cache = self
2256 .accepted_bundle_cache
2257 .try_borrow()
2258 .map_err(|_| InternalError::store_invariant())?;
2259 let cached = cache.as_ref().ok_or_else(InternalError::store_invariant)?;
2260 if let Some(selection) = cached
2261 .entity_selections
2262 .try_borrow()
2263 .map_err(|_| InternalError::store_invariant())?
2264 .get(&entity)
2265 .cloned()
2266 {
2267 return Ok(Some(selection));
2268 }
2269
2270 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2271 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
2272 let identity = AcceptedCatalogIdentity::new(
2273 entity,
2274 entity_path,
2275 store_path,
2276 bundle.revision(),
2277 snapshot.version(),
2278 fingerprint,
2279 );
2280
2281 let selected = AcceptedCatalogSnapshotSelection::new(
2282 identity,
2283 cached.value_catalog.clone(),
2284 Rc::from(raw_snapshot.into_bytes()),
2285 );
2286 cached
2287 .entity_selections
2288 .try_borrow_mut()
2289 .map_err(|_| InternalError::store_invariant())?
2290 .insert(entity, selected.clone());
2291
2292 Ok(Some(selected))
2293 }
2294
2295 pub(in crate::db) fn current_canonical_accepted_catalog_selection(
2299 &self,
2300 entity: EntityTag,
2301 entity_path: &str,
2302 store_path: &'static str,
2303 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
2304 let first = self.canonical_root_slot_bytes(0)?;
2305 let second = self.canonical_root_slot_bytes(1)?;
2306 let Some(selection) =
2307 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2308 else {
2309 return Ok(None);
2310 };
2311 let bundle_key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
2312 let raw_bundle = self
2313 .get_canonical_raw_value(&bundle_key)?
2314 .ok_or_else(InternalError::store_corruption)?;
2315 let bundle = decode_verified_accepted_schema_revision_bundle(
2316 selection.root(),
2317 raw_bundle.as_bytes(),
2318 )?;
2319 if bundle.store_path() != store_path {
2320 return Err(InternalError::store_corruption());
2321 }
2322 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
2323 return Ok(None);
2324 };
2325 if snapshot.entity_path() != entity_path {
2326 return Err(InternalError::store_corruption());
2327 }
2328
2329 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2330 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
2331 let identity = AcceptedCatalogIdentity::new(
2332 entity,
2333 entity_path,
2334 store_path,
2335 bundle.revision(),
2336 snapshot.version(),
2337 fingerprint,
2338 );
2339
2340 Ok(Some(AcceptedCatalogSnapshotSelection::new(
2341 identity,
2342 AcceptedValueCatalogHandle::new(
2343 bundle.enum_catalog().clone(),
2344 bundle.composite_catalog().clone(),
2345 self.accepted_catalog_scope
2346 .get_or_init(AcceptedStoreCatalogScope::new)
2347 .clone(),
2348 bundle.revision(),
2349 selection.root().fingerprint(),
2350 ),
2351 Rc::from(raw_snapshot.into_bytes()),
2352 )))
2353 }
2354
2355 #[cfg(test)]
2361 pub(in crate::db) fn catalog_metadata(
2362 &self,
2363 ) -> Result<Option<SchemaStoreCatalogMetadata>, InternalError> {
2364 Ok(self
2365 .allocation_metadata()?
2366 .map(SchemaStoreAllocationMetadata::schema))
2367 }
2368
2369 pub(in crate::db) fn allocation_metadata(
2376 &self,
2377 ) -> Result<Option<SchemaStoreAllocationMetadata>, InternalError> {
2378 let latest_by_entity = self.latest_raw_snapshots_by_entity();
2379 if latest_by_entity.is_empty() {
2380 return Ok(None);
2381 }
2382
2383 Ok(Some(SchemaStoreAllocationMetadata::new(
2384 derive_data_allocation_metadata(&latest_by_entity)?,
2385 derive_index_allocation_metadata(&latest_by_entity)?,
2386 derive_schema_catalog_metadata(&latest_by_entity)?,
2387 )))
2388 }
2389
2390 fn insert_raw_snapshot(
2392 &mut self,
2393 key: RawSchemaKey,
2394 snapshot: RawSchemaSnapshot,
2395 ) -> Option<RawSchemaSnapshot> {
2396 self.invalidate_accepted_bundle_cache_for_key(key);
2397 let previous_journaled = if matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
2398 self.get_raw_snapshot_for_backend(&key)
2399 } else {
2400 None
2401 };
2402 match &mut self.backend {
2403 SchemaStoreBackend::Heap(map) => map.insert(key, snapshot),
2404 SchemaStoreBackend::Journaled {
2405 live, tombstones, ..
2406 } => {
2407 tombstones.remove(&key);
2408 live.insert(key, snapshot);
2409 previous_journaled
2410 }
2411 }
2412 }
2413
2414 #[must_use]
2416 fn get_raw_snapshot(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
2417 match &self.backend {
2418 SchemaStoreBackend::Heap(map) => map.get(key).cloned(),
2419 SchemaStoreBackend::Journaled { .. } => self.get_raw_snapshot_for_backend(key),
2420 }
2421 }
2422
2423 fn accepted_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
2424 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
2425 Ok(self
2426 .get_raw_snapshot(&key)
2427 .map(RawSchemaSnapshot::into_bytes))
2428 }
2429
2430 fn canonical_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
2431 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
2432 Ok(self
2433 .get_canonical_raw_value(&key)?
2434 .map(RawSchemaSnapshot::into_bytes))
2435 }
2436
2437 fn current_root_matches_candidate(
2438 &self,
2439 candidate: &CandidateSchemaRevision,
2440 ) -> Result<bool, InternalError> {
2441 let Some(selection) = self.current_accepted_schema_root()? else {
2442 return Ok(false);
2443 };
2444 if selection.root() != candidate.root() {
2445 return Ok(false);
2446 }
2447 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2448 let bundle = self
2449 .get_raw_snapshot(&key)
2450 .ok_or_else(InternalError::store_corruption)?;
2451 let _verified =
2452 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
2453 Ok(true)
2454 }
2455
2456 fn canonical_root_matches_candidate(
2457 &self,
2458 candidate: &CandidateSchemaRevision,
2459 ) -> Result<bool, InternalError> {
2460 let first = self.canonical_root_slot_bytes(0)?;
2461 let second = self.canonical_root_slot_bytes(1)?;
2462 let Some(selection) =
2463 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2464 else {
2465 return Ok(false);
2466 };
2467 if selection.root() != candidate.root() {
2468 return Ok(false);
2469 }
2470 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2471 let bundle = self
2472 .get_canonical_raw_value(&key)?
2473 .ok_or_else(InternalError::store_corruption)?;
2474 let _verified =
2475 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
2476 Ok(true)
2477 }
2478
2479 fn get_canonical_raw_value(
2480 &self,
2481 key: &RawSchemaKey,
2482 ) -> Result<Option<RawSchemaSnapshot>, InternalError> {
2483 match &self.backend {
2484 SchemaStoreBackend::Journaled { canonical, .. } => Ok(canonical.get(key)),
2485 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
2486 }
2487 }
2488
2489 fn insert_canonical_raw_value(
2490 &mut self,
2491 key: RawSchemaKey,
2492 bytes: Vec<u8>,
2493 ) -> Result<(), InternalError> {
2494 self.invalidate_accepted_bundle_cache_for_key(key);
2495 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
2496 return Err(InternalError::store_invariant());
2497 };
2498 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
2499 Ok(())
2500 }
2501
2502 fn insert_durable_raw_value(&mut self, key: RawSchemaKey, bytes: Vec<u8>) {
2506 self.invalidate_accepted_bundle_cache_for_key(key);
2507 let value = RawSchemaSnapshot::from_encoded_control_record(bytes);
2508 match &mut self.backend {
2509 SchemaStoreBackend::Heap(map) => {
2510 map.insert(key, value);
2511 }
2512 SchemaStoreBackend::Journaled {
2513 canonical,
2514 live,
2515 tombstones,
2516 } => {
2517 live.remove(&key);
2518 tombstones.remove(&key);
2519 canonical.insert(key, value);
2520 }
2521 }
2522 }
2523
2524 fn invalidate_accepted_bundle_cache_for_key(&mut self, key: RawSchemaKey) {
2525 if key.is_accepted_root() {
2526 self.accepted_bundle_cache.get_mut().take();
2527 }
2528 }
2529
2530 fn insert_durable_candidate_snapshots(
2531 &mut self,
2532 candidate: &CandidateSchemaRevision,
2533 ) -> Result<(), InternalError> {
2534 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2535 let key = RawSchemaKey::from_entity_version(*entity_tag, snapshot.version());
2536 let value = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2537 match &mut self.backend {
2538 SchemaStoreBackend::Heap(map) => {
2539 map.insert(key, value);
2540 }
2541 SchemaStoreBackend::Journaled {
2542 canonical,
2543 live,
2544 tombstones,
2545 } => {
2546 live.remove(&key);
2547 tombstones.remove(&key);
2548 canonical.insert(key, value);
2549 }
2550 }
2551 }
2552 Ok(())
2553 }
2554
2555 fn candidate_entry_keys(
2556 candidate: &CandidateSchemaRevision,
2557 root_slot: usize,
2558 ) -> Result<BTreeSet<RawSchemaKey>, InternalError> {
2559 let mut keys = candidate
2560 .bundle()
2561 .entity_snapshots()
2562 .iter()
2563 .map(|(entity_tag, snapshot)| {
2564 RawSchemaKey::from_entity_version(*entity_tag, snapshot.version())
2565 })
2566 .collect::<BTreeSet<_>>();
2567 keys.insert(RawSchemaKey::from_accepted_bundle(
2568 candidate.root().bundle_key(),
2569 ));
2570 keys.insert(RawSchemaKey::from_accepted_root_slot(root_slot)?);
2571 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2572 for activation in snapshot
2573 .constraint_activations()
2574 .iter()
2575 .filter(|activation| activation.state() == ConstraintActivationState::Validating)
2576 {
2577 keys.insert(RawSchemaKey::from_constraint_validation_job(
2578 *entity_tag,
2579 activation.id(),
2580 ));
2581 }
2582 }
2583 Ok(keys)
2584 }
2585
2586 fn retain_durable_candidate_entries(
2590 &mut self,
2591 candidate: &CandidateSchemaRevision,
2592 root_slot: usize,
2593 ) -> Result<(), InternalError> {
2594 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
2595 self.accepted_bundle_cache.get_mut().take();
2596 match &mut self.backend {
2597 SchemaStoreBackend::Heap(map) => {
2598 map.retain(|key, _| keep.contains(key) || key.is_identity_state());
2599 }
2600 SchemaStoreBackend::Journaled {
2601 canonical,
2602 live,
2603 tombstones,
2604 } => {
2605 let stale = canonical
2606 .iter()
2607 .filter_map(|entry| {
2608 (!keep.contains(entry.key()) && !entry.key().is_identity_state())
2609 .then_some(*entry.key())
2610 })
2611 .collect::<Vec<_>>();
2612 for key in stale {
2613 canonical.remove(&key);
2614 }
2615 live.retain(|key, _| keep.contains(key) || key.is_identity_state());
2616 tombstones.clear();
2617 }
2618 }
2619 Ok(())
2620 }
2621
2622 fn retain_materialized_candidate_entries(
2623 &mut self,
2624 candidate: &CandidateSchemaRevision,
2625 root_slot: usize,
2626 ) -> Result<(), InternalError> {
2627 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
2628 self.accepted_bundle_cache.get_mut().take();
2629 let SchemaStoreBackend::Journaled {
2630 canonical,
2631 live,
2632 tombstones,
2633 } = &mut self.backend
2634 else {
2635 return Err(InternalError::store_invariant());
2636 };
2637 live.retain(|key, _| keep.contains(key) || key.is_identity_state());
2638 let canonical_keys = canonical
2639 .iter()
2640 .map(|entry| *entry.key())
2641 .collect::<Vec<_>>();
2642 for key in canonical_keys {
2643 if keep.contains(&key) || key.is_identity_state() {
2644 tombstones.remove(&key);
2645 } else {
2646 tombstones.insert(key);
2647 }
2648 }
2649 Ok(())
2650 }
2651
2652 fn retain_canonical_candidate_entries(
2653 &mut self,
2654 candidate: &CandidateSchemaRevision,
2655 root_slot: usize,
2656 ) -> Result<(), InternalError> {
2657 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
2658 self.accepted_bundle_cache.get_mut().take();
2659 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
2660 return Err(InternalError::store_invariant());
2661 };
2662 let stale = canonical
2663 .iter()
2664 .filter_map(|entry| {
2665 (!keep.contains(entry.key()) && !entry.key().is_identity_state())
2666 .then_some(*entry.key())
2667 })
2668 .collect::<Vec<_>>();
2669 for key in stale {
2670 canonical.remove(&key);
2671 }
2672 Ok(())
2673 }
2674
2675 #[must_use]
2677 #[cfg(test)]
2678 fn contains_raw_snapshot(&self, key: &RawSchemaKey) -> bool {
2679 match &self.backend {
2680 SchemaStoreBackend::Heap(map) => map.contains_key(key),
2681 SchemaStoreBackend::Journaled { .. } => {
2682 self.get_raw_snapshot_for_backend(key).is_some()
2683 }
2684 }
2685 }
2686
2687 #[must_use]
2689 #[cfg(test)]
2690 pub(in crate::db) fn len(&self) -> u64 {
2691 match &self.backend {
2692 SchemaStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
2693 SchemaStoreBackend::Journaled { .. } => {
2694 let mut count = 0_u64;
2695 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
2696 count = count.saturating_add(1);
2697 Ok(SchemaStoreVisit::Continue)
2698 });
2699 count
2700 }
2701 }
2702 }
2703
2704 #[must_use]
2706 #[cfg(test)]
2707 pub(in crate::db) fn is_empty(&self) -> bool {
2708 match &self.backend {
2709 SchemaStoreBackend::Heap(map) => map.is_empty(),
2710 SchemaStoreBackend::Journaled { .. } => {
2711 let mut empty = true;
2712 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
2713 empty = false;
2714 Ok(SchemaStoreVisit::Stop)
2715 });
2716 empty
2717 }
2718 }
2719 }
2720
2721 #[cfg(test)]
2723 pub(in crate::db) fn clear(&mut self) {
2724 self.accepted_bundle_cache.get_mut().take();
2725 match &mut self.backend {
2726 SchemaStoreBackend::Heap(map) => map.clear(),
2727 SchemaStoreBackend::Journaled {
2728 canonical,
2729 live,
2730 tombstones,
2731 } => {
2732 live.clear();
2733 tombstones.clear();
2734 let keys = canonical
2735 .iter()
2736 .map(|entry| *entry.key())
2737 .collect::<Vec<_>>();
2738 for key in keys {
2739 if key.is_entity_snapshot() {
2740 tombstones.insert(key);
2741 } else {
2742 canonical.remove(&key);
2743 }
2744 }
2745 }
2746 }
2747 }
2748
2749 fn current_accepted_schema_bundle_ref(
2750 &self,
2751 ) -> Result<Option<Ref<'_, AcceptedSchemaRevisionBundle>>, InternalError> {
2752 let Some(selection) = self.current_accepted_schema_root()? else {
2753 self.accepted_bundle_cache
2754 .try_borrow_mut()
2755 .map_err(|_| InternalError::store_invariant())?
2756 .take();
2757 return Ok(None);
2758 };
2759
2760 let cache_matches = self
2761 .accepted_bundle_cache
2762 .try_borrow()
2763 .map_err(|_| InternalError::store_invariant())?
2764 .as_ref()
2765 .is_some_and(|cached| cached.selection == selection);
2766 if !cache_matches {
2767 let key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
2768 let raw = self
2769 .get_raw_snapshot(&key)
2770 .ok_or_else(InternalError::store_corruption)?;
2771 let bundle =
2772 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes())?;
2773 self.validate_constraint_validation_job_closure(&bundle)?;
2774 #[cfg(test)]
2775 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES
2776 .with(|misses| misses.set(misses.get().saturating_add(1)));
2777 let value_catalog = AcceptedValueCatalogHandle::new(
2778 bundle.enum_catalog().clone(),
2779 bundle.composite_catalog().clone(),
2780 self.accepted_catalog_scope
2781 .get_or_init(AcceptedStoreCatalogScope::new)
2782 .clone(),
2783 bundle.revision(),
2784 selection.root().fingerprint(),
2785 );
2786 *self
2787 .accepted_bundle_cache
2788 .try_borrow_mut()
2789 .map_err(|_| InternalError::store_invariant())? = Some(AcceptedSchemaBundleCache {
2790 selection,
2791 bundle,
2792 value_catalog,
2793 entity_selections: RefCell::new(StdBTreeMap::new()),
2794 });
2795 }
2796
2797 let cache = self
2798 .accepted_bundle_cache
2799 .try_borrow()
2800 .map_err(|_| InternalError::store_invariant())?;
2801 let bundle = Ref::filter_map(cache, |cache| {
2802 cache
2803 .as_ref()
2804 .filter(|cached| cached.selection == selection)
2805 .map(|cached| &cached.bundle)
2806 })
2807 .map_err(|_| InternalError::store_invariant())?;
2808 self.validate_identity_state_closure(&bundle)?;
2809 Ok(Some(bundle))
2810 }
2811
2812 fn latest_raw_snapshots_by_entity(
2813 &self,
2814 ) -> StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)> {
2815 let mut latest_by_entity =
2816 StdBTreeMap::<EntityTag, (SchemaVersion, RawSchemaSnapshot)>::new();
2817
2818 let _: Result<(), std::convert::Infallible> = self.visit_raw_snapshots(|key, snapshot| {
2819 let version = SchemaVersion::new(key.version());
2820 match latest_by_entity.get_mut(&key.entity_tag()) {
2821 Some((latest_version, latest_snapshot)) if version > *latest_version => {
2822 *latest_version = version;
2823 *latest_snapshot = snapshot.clone();
2824 }
2825 None => {
2826 latest_by_entity.insert(key.entity_tag(), (version, snapshot.clone()));
2827 }
2828 Some(_) => {}
2829 }
2830 Ok(SchemaStoreVisit::Continue)
2831 });
2832
2833 latest_by_entity
2834 }
2835
2836 fn visit_raw_snapshots<E>(
2839 &self,
2840 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2841 ) -> Result<(), E> {
2842 let bounds = RawSchemaKey::all_entity_range_bounds();
2843 match &self.backend {
2844 SchemaStoreBackend::Heap(map) => {
2845 let mut visitor = visitor;
2846 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
2847 if visitor(key, snapshot)?.should_stop() {
2848 break;
2849 }
2850 }
2851 }
2852 SchemaStoreBackend::Journaled {
2853 canonical,
2854 live,
2855 tombstones,
2856 } => Self::visit_journaled_raw_snapshot_range(
2857 canonical,
2858 live,
2859 tombstones,
2860 bounds,
2861 Direction::Asc,
2862 visitor,
2863 )?,
2864 }
2865
2866 Ok(())
2867 }
2868
2869 fn visit_constraint_validation_jobs<E>(
2870 &self,
2871 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2872 ) -> Result<(), E> {
2873 let bounds = RawSchemaKey::all_constraint_validation_job_range_bounds();
2874 match &self.backend {
2875 SchemaStoreBackend::Heap(map) => {
2876 let mut visitor = visitor;
2877 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
2878 if visitor(key, snapshot)?.should_stop() {
2879 break;
2880 }
2881 }
2882 }
2883 SchemaStoreBackend::Journaled {
2884 canonical,
2885 live,
2886 tombstones,
2887 } => Self::visit_journaled_raw_snapshot_range(
2888 canonical,
2889 live,
2890 tombstones,
2891 bounds,
2892 Direction::Asc,
2893 visitor,
2894 )?,
2895 }
2896 Ok(())
2897 }
2898
2899 #[cfg(test)]
2900 #[must_use]
2901 pub(in crate::db) fn canonical_len_for_tests(&self) -> u64 {
2902 match &self.backend {
2903 SchemaStoreBackend::Journaled { canonical: map, .. } => map.len(),
2904 SchemaStoreBackend::Heap(_) => 0,
2905 }
2906 }
2907
2908 fn get_raw_snapshot_for_backend(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
2909 let SchemaStoreBackend::Journaled {
2910 canonical,
2911 live,
2912 tombstones,
2913 } = &self.backend
2914 else {
2915 return None;
2916 };
2917
2918 if tombstones.contains(key) {
2919 return None;
2920 }
2921 live.get(key).cloned().or_else(|| canonical.get(key))
2922 }
2923
2924 fn visit_journaled_raw_snapshot_range<E>(
2925 canonical: &StableBTreeMap<
2926 RawSchemaKey,
2927 RawSchemaSnapshot,
2928 VirtualMemory<DefaultMemoryImpl>,
2929 >,
2930 live: &StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
2931 tombstones: &BTreeSet<RawSchemaKey>,
2932 bounds: (RangeBound<RawSchemaKey>, RangeBound<RawSchemaKey>),
2933 direction: Direction,
2934 mut visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2935 ) -> Result<(), E> {
2936 match direction {
2937 Direction::Asc => visit_ordered_overlay(
2938 canonical.range((bounds.0, bounds.1)),
2939 live.range((bounds.0, bounds.1)),
2940 Direction::Asc,
2941 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
2942 |canonical_entry| !tombstones.contains(canonical_entry.key()),
2943 |live_entry| !tombstones.contains(live_entry.0),
2944 |entry| {
2945 let visit = match entry {
2946 OrderedOverlayEntry::Canonical(canonical_entry) => {
2947 visitor(canonical_entry.key(), &canonical_entry.value())?
2948 }
2949 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
2950 };
2951 Ok(if visit.should_stop() {
2952 OrderedOverlayVisit::Stop
2953 } else {
2954 OrderedOverlayVisit::Continue
2955 })
2956 },
2957 ),
2958 Direction::Desc => visit_ordered_overlay(
2959 canonical.range((bounds.0, bounds.1)).rev(),
2960 live.range((bounds.0, bounds.1)).rev(),
2961 Direction::Desc,
2962 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
2963 |canonical_entry| !tombstones.contains(canonical_entry.key()),
2964 |live_entry| !tombstones.contains(live_entry.0),
2965 |entry| {
2966 let visit = match entry {
2967 OrderedOverlayEntry::Canonical(canonical_entry) => {
2968 visitor(canonical_entry.key(), &canonical_entry.value())?
2969 }
2970 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
2971 };
2972 Ok(if visit.should_stop() {
2973 OrderedOverlayVisit::Stop
2974 } else {
2975 OrderedOverlayVisit::Continue
2976 })
2977 },
2978 ),
2979 }
2980 }
2981}
2982
2983fn map_schema_publication_error(error: AcceptedSchemaPublicationError) -> InternalError {
2984 match error {
2985 AcceptedSchemaPublicationError::StaleSchemaRevision { .. }
2986 | AcceptedSchemaPublicationError::RevisionExhausted => InternalError::store_unsupported(),
2987 AcceptedSchemaPublicationError::InvalidCandidate => InternalError::store_invariant(),
2988 AcceptedSchemaPublicationError::CorruptRootSlots => InternalError::store_corruption(),
2989 }
2990}
2991
2992fn derive_data_allocation_metadata(
2993 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
2994) -> Result<SchemaStoreCatalogMetadata, InternalError> {
2995 let mut max_version = SchemaVersion::initial();
2996 let mut hasher = new_hash_sha256();
2997 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN);
2998
2999 for (entity, (_, snapshot)) in latest_by_entity {
3000 let persisted = snapshot.decode_persisted_snapshot()?;
3001 if persisted.version() > max_version {
3002 max_version = persisted.version();
3003 }
3004
3005 let data_projection = PersistedSchemaSnapshot::new_with_primary_key_fields_and_indexes(
3006 persisted.version(),
3007 persisted.entity_path().to_string(),
3008 persisted.entity_name().to_string(),
3009 persisted.primary_key_field_ids().to_vec(),
3010 persisted.row_layout().clone(),
3011 persisted.fields().to_vec(),
3012 Vec::new(),
3013 );
3014 let constraint_catalog = crate::db::schema::AcceptedConstraintCatalog::initial(
3015 data_projection.fields(),
3016 data_projection.indexes(),
3017 data_projection.relations(),
3018 )
3019 .map_err(|_| InternalError::store_invariant())?;
3020 let data_projection = data_projection.with_constraint_catalog(constraint_catalog);
3021 let encoded = encode_persisted_schema_snapshot(&data_projection)?;
3022
3023 write_hash_u64(&mut hasher, entity.value());
3024 write_hash_u32(&mut hasher, persisted.version().get());
3025 write_hash_len_u32(&mut hasher, encoded.len());
3026 hasher.update(encoded);
3027 }
3028
3029 Ok(finalize_schema_metadata(
3030 max_version,
3031 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
3032 hasher,
3033 latest_by_entity.len(),
3034 ))
3035}
3036
3037fn derive_index_allocation_metadata(
3038 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
3039) -> Result<SchemaStoreCatalogMetadata, InternalError> {
3040 let mut max_version = SchemaVersion::initial();
3041 let mut hasher = new_hash_sha256();
3042 write_hash_tag_u8(
3043 &mut hasher,
3044 SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN,
3045 );
3046
3047 for (entity, (_, snapshot)) in latest_by_entity {
3048 let persisted = snapshot.decode_persisted_snapshot()?;
3049 if persisted.version() > max_version {
3050 max_version = persisted.version();
3051 }
3052
3053 write_hash_u64(&mut hasher, entity.value());
3054 write_hash_u32(&mut hasher, persisted.version().get());
3055 write_hash_len_u32(&mut hasher, persisted.indexes().len());
3056 for index in persisted.indexes() {
3057 write_hash_u32(&mut hasher, u32::from(index.ordinal()));
3058 write_hash_str_u32(&mut hasher, index.name());
3059 write_hash_str_u32(&mut hasher, index.store());
3060 write_hash_tag_u8(&mut hasher, u8::from(index.unique()));
3061 write_hash_str_u32(&mut hasher, persisted_index_origin_name(index.origin()));
3062 match index.predicate_sql() {
3063 Some(predicate_sql) => {
3064 write_hash_tag_u8(&mut hasher, 1);
3065 write_hash_str_u32(&mut hasher, predicate_sql);
3066 }
3067 None => write_hash_tag_u8(&mut hasher, 0),
3068 }
3069 hash_persisted_index_key(&mut hasher, index.key());
3070 }
3071 }
3072
3073 Ok(finalize_schema_metadata(
3074 max_version,
3075 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
3076 hasher,
3077 latest_by_entity.len(),
3078 ))
3079}
3080
3081fn derive_schema_catalog_metadata(
3082 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
3083) -> Result<SchemaStoreCatalogMetadata, InternalError> {
3084 let mut max_version = SchemaVersion::initial();
3085 let mut hasher = new_hash_sha256();
3086 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN);
3087
3088 for (entity, (version, snapshot)) in latest_by_entity {
3089 let persisted = snapshot.decode_persisted_snapshot()?;
3090 if persisted.version() > max_version {
3091 max_version = persisted.version();
3092 }
3093
3094 write_hash_u64(&mut hasher, entity.value());
3095 write_hash_u32(&mut hasher, version.get());
3096 write_hash_len_u32(&mut hasher, snapshot.as_bytes().len());
3097 hasher.update(snapshot.as_bytes());
3098 }
3099
3100 Ok(finalize_schema_metadata(
3101 max_version,
3102 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
3103 hasher,
3104 latest_by_entity.len(),
3105 ))
3106}
3107
3108fn finalize_schema_metadata(
3109 schema_version: SchemaVersion,
3110 schema_fingerprint_method_version: u8,
3111 hasher: sha2::Sha256,
3112 entity_count: usize,
3113) -> SchemaStoreCatalogMetadata {
3114 let digest = finalize_hash_sha256(hasher);
3115 let mut schema_fingerprint = [0u8; 16];
3116 schema_fingerprint.copy_from_slice(&digest[..16]);
3117
3118 SchemaStoreCatalogMetadata::new(
3119 schema_version,
3120 schema_fingerprint_method_version,
3121 schema_fingerprint,
3122 u64::try_from(entity_count).unwrap_or(u64::MAX),
3123 )
3124}
3125
3126fn hash_persisted_index_key(hasher: &mut sha2::Sha256, key: &PersistedIndexKeySnapshot) {
3127 match key {
3128 PersistedIndexKeySnapshot::FieldPath(paths) => {
3129 write_hash_tag_u8(hasher, 1);
3130 write_hash_len_u32(hasher, paths.len());
3131 for path in paths {
3132 hash_persisted_index_field_path(hasher, path);
3133 }
3134 }
3135 PersistedIndexKeySnapshot::Items(items) => {
3136 write_hash_tag_u8(hasher, 2);
3137 write_hash_len_u32(hasher, items.len());
3138 for item in items {
3139 match item {
3140 PersistedIndexKeyItemSnapshot::FieldPath(path) => {
3141 write_hash_tag_u8(hasher, 1);
3142 hash_persisted_index_field_path(hasher, path);
3143 }
3144 PersistedIndexKeyItemSnapshot::Expression(expression) => {
3145 write_hash_tag_u8(hasher, 2);
3146 write_hash_str_u32(hasher, persisted_expression_op_name(expression.op()));
3147 hash_persisted_index_field_path(hasher, expression.source());
3148 hash_accepted_field_kind(hasher, expression.input_kind());
3149 hash_accepted_field_kind(hasher, expression.output_kind());
3150 write_hash_str_u32(hasher, expression.canonical_text());
3151 }
3152 }
3153 }
3154 }
3155 }
3156}
3157
3158fn hash_persisted_index_field_path(
3159 hasher: &mut sha2::Sha256,
3160 path: &crate::db::schema::PersistedIndexFieldPathSnapshot,
3161) {
3162 write_hash_u32(hasher, path.field_id().get());
3163 write_hash_u32(hasher, u32::from(path.slot().get()));
3164 write_hash_len_u32(hasher, path.path().len());
3165 for segment in path.path() {
3166 write_hash_str_u32(hasher, segment);
3167 }
3168 hash_accepted_field_kind(hasher, path.kind());
3169 write_hash_tag_u8(hasher, u8::from(path.nullable()));
3170}
3171
3172fn hash_accepted_field_kind(hasher: &mut sha2::Sha256, kind: &AcceptedFieldKind) {
3173 match kind {
3174 AcceptedFieldKind::Account => write_hash_tag_u8(hasher, 1),
3175 AcceptedFieldKind::Blob { max_len } => {
3176 write_hash_tag_u8(hasher, 2);
3177 hash_optional_u32(hasher, *max_len);
3178 }
3179 AcceptedFieldKind::Bool => {
3180 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL);
3181 }
3182 AcceptedFieldKind::Date => write_hash_tag_u8(hasher, 4),
3183 AcceptedFieldKind::Decimal { scale } => {
3184 write_hash_tag_u8(hasher, 5);
3185 write_hash_u32(hasher, *scale);
3186 }
3187 AcceptedFieldKind::Duration => write_hash_tag_u8(hasher, 6),
3188 AcceptedFieldKind::Enum { type_id } => {
3189 write_hash_tag_u8(hasher, 7);
3190 write_hash_u32(hasher, type_id.get());
3191 }
3192 AcceptedFieldKind::Float32 => write_hash_tag_u8(hasher, 8),
3193 AcceptedFieldKind::Float64 => write_hash_tag_u8(hasher, 9),
3194 AcceptedFieldKind::Int8 => write_hash_tag_u8(hasher, 10),
3195 AcceptedFieldKind::Int16 => write_hash_tag_u8(hasher, 11),
3196 AcceptedFieldKind::Int32 => write_hash_tag_u8(hasher, 12),
3197 AcceptedFieldKind::Int64 => write_hash_tag_u8(hasher, 13),
3198 AcceptedFieldKind::Int128 => write_hash_tag_u8(hasher, 14),
3199 AcceptedFieldKind::IntBig { max_bytes } => {
3200 write_hash_tag_u8(hasher, 15);
3201 write_hash_u32(hasher, *max_bytes);
3202 }
3203 AcceptedFieldKind::Principal => write_hash_tag_u8(hasher, 16),
3204 AcceptedFieldKind::Subaccount => write_hash_tag_u8(hasher, 17),
3205 AcceptedFieldKind::Text { max_len } => {
3206 write_hash_tag_u8(hasher, 18);
3207 hash_optional_u32(hasher, *max_len);
3208 }
3209 AcceptedFieldKind::Timestamp => write_hash_tag_u8(hasher, 19),
3210 AcceptedFieldKind::Nat8 => write_hash_tag_u8(hasher, 20),
3211 AcceptedFieldKind::Nat16 => write_hash_tag_u8(hasher, 21),
3212 AcceptedFieldKind::Nat32 => write_hash_tag_u8(hasher, 22),
3213 AcceptedFieldKind::Nat64 => write_hash_tag_u8(hasher, 23),
3214 AcceptedFieldKind::Nat128 => write_hash_tag_u8(hasher, 24),
3215 AcceptedFieldKind::NatBig { max_bytes } => {
3216 write_hash_tag_u8(hasher, 25);
3217 write_hash_u32(hasher, *max_bytes);
3218 }
3219 AcceptedFieldKind::Ulid => write_hash_tag_u8(hasher, 26),
3220 AcceptedFieldKind::Unit => write_hash_tag_u8(hasher, 27),
3221 AcceptedFieldKind::Relation {
3222 target_path,
3223 target_entity_name,
3224 target_entity_tag,
3225 target_store_path,
3226 key_kind,
3227 } => {
3228 write_hash_tag_u8(hasher, 28);
3229 write_hash_str_u32(hasher, target_path);
3230 write_hash_str_u32(hasher, target_entity_name);
3231 write_hash_u64(hasher, target_entity_tag.value());
3232 write_hash_str_u32(hasher, target_store_path);
3233 hash_accepted_field_kind(hasher, key_kind);
3234 }
3235 AcceptedFieldKind::List(inner) => {
3236 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST);
3237 hash_accepted_field_kind(hasher, inner);
3238 }
3239 AcceptedFieldKind::Set(inner) => {
3240 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET);
3241 hash_accepted_field_kind(hasher, inner);
3242 }
3243 AcceptedFieldKind::Map { key, value } => {
3244 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP);
3245 hash_accepted_field_kind(hasher, key);
3246 hash_accepted_field_kind(hasher, value);
3247 }
3248 AcceptedFieldKind::Composite { type_id } => {
3249 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE);
3250 write_hash_u32(hasher, type_id.get());
3251 }
3252 }
3253}
3254
3255fn hash_optional_u32(hasher: &mut sha2::Sha256, value: Option<u32>) {
3256 match value {
3257 Some(value) => {
3258 write_hash_tag_u8(hasher, 1);
3259 write_hash_u32(hasher, value);
3260 }
3261 None => write_hash_tag_u8(hasher, 0),
3262 }
3263}
3264
3265const fn persisted_index_origin_name(
3266 origin: crate::db::schema::PersistedIndexOrigin,
3267) -> &'static str {
3268 match origin {
3269 crate::db::schema::PersistedIndexOrigin::Generated => "generated",
3270 crate::db::schema::PersistedIndexOrigin::SqlDdl => "sql_ddl",
3271 }
3272}
3273
3274const fn persisted_expression_op_name(
3275 op: crate::db::schema::PersistedIndexExpressionOp,
3276) -> &'static str {
3277 match op {
3278 crate::db::schema::PersistedIndexExpressionOp::Lower => "lower",
3279 crate::db::schema::PersistedIndexExpressionOp::Upper => "upper",
3280 crate::db::schema::PersistedIndexExpressionOp::Trim => "trim",
3281 crate::db::schema::PersistedIndexExpressionOp::LowerTrim => "lower_trim",
3282 crate::db::schema::PersistedIndexExpressionOp::Date => "date",
3283 crate::db::schema::PersistedIndexExpressionOp::Year => "year",
3284 crate::db::schema::PersistedIndexExpressionOp::Month => "month",
3285 crate::db::schema::PersistedIndexExpressionOp::Day => "day",
3286 }
3287}
3288
3289#[cfg(test)]
3294mod tests;