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