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