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 journal::{JournalBatch, JournalRecord},
23 ordered_overlay::{OrderedOverlayEntry, OrderedOverlayVisit, visit_ordered_overlay},
24 positioned_overlay::{
25 JournalOverlayPosition, PositionedOverlayMetadata, PositionedOverlayRetirement,
26 },
27 runtime_entity_catalog::AcceptedRuntimeEntity,
28 schema::{
29 AcceptedFieldKind, AcceptedRowLayoutRuntimeContract, AcceptedSchemaSnapshot,
30 ConstraintActivationKind, ConstraintActivationState, ConstraintId, ConstraintOrigin,
31 ConstraintValidationJob, FieldId, PersistedIndexKeyItemSnapshot,
32 PersistedIndexKeySnapshot, PersistedSchemaSnapshot, SchemaVersion,
33 accepted_schema_cache_fingerprint,
34 accepted_schema_cache_fingerprint_for_persisted_snapshot,
35 accepted_schema_cache_fingerprint_method_version, decode_constraint_validation_job,
36 decode_persisted_schema_snapshot, encode_constraint_validation_job,
37 encode_persisted_schema_snapshot,
38 enum_catalog::{
39 AcceptedSchemaAuthority, AcceptedSchemaPublicationError, AcceptedSchemaRevision,
40 AcceptedSchemaRevisionBundle, AcceptedSchemaRootSelection,
41 AcceptedStoreCatalogScope, AcceptedValueCatalogHandle, CandidateSchemaRevision,
42 decode_verified_accepted_schema_revision_bundle,
43 prepare_accepted_schema_root_publication, select_current_accepted_schema_root,
44 },
45 schema_snapshot_integrity_detail,
46 },
47 },
48 error::InternalError,
49 types::EntityTag,
50};
51use ic_stable_structures::{
52 BTreeMap as StableBTreeMap, DefaultMemoryImpl, Storable, memory_manager::VirtualMemory,
53 storable::Bound as StorableBound,
54};
55use sha2::Digest;
56use std::borrow::Cow;
57#[cfg(test)]
58use std::cell::Cell;
59use std::cell::{OnceCell, Ref, RefCell};
60use std::collections::{BTreeMap as StdBTreeMap, BTreeSet};
61#[cfg(test)]
62use std::convert::Infallible;
63use std::ops::Bound as RangeBound;
64use std::rc::Rc;
65
66const SCHEMA_KEY_BYTES_USIZE: usize = 16;
67const SCHEMA_KEY_BYTES: u32 = 16;
68const SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT: u8 = 0;
69const SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE: u8 = 1;
70const SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT: u8 = 2;
71const SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB: u8 = 3;
72const SCHEMA_KEY_NAMESPACE_IDENTITY_STATE: u8 = 4;
73const SCHEMA_STORE_FINGERPRINT_METHOD_VERSION: u8 = 1;
76const SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN: u8 = 1;
77const SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN: u8 = 2;
78const SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN: u8 = 3;
79const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL: u8 = 3;
80const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST: u8 = 29;
81const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET: u8 = 30;
82const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP: u8 = 31;
83const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE: u8 = 32;
84const RAW_SCHEMA_SNAPSHOT_MAGIC: &[u8; 8] = b"ICYDBCAT";
85const RAW_SCHEMA_SNAPSHOT_VALUE_VERSION: u8 = 1;
86const RAW_SCHEMA_SNAPSHOT_HEADER_BYTES: usize = 25;
87
88pub(in crate::db) fn load_accepted_schema_snapshot(
93 schema_store: &SchemaStore,
94 entity_tag: EntityTag,
95 entity_path: &str,
96) -> Result<AcceptedSchemaSnapshot, InternalError> {
97 let bundle = schema_store
98 .current_accepted_schema_bundle()?
99 .ok_or_else(InternalError::store_corruption)?;
100 let snapshot = bundle
101 .entity_snapshots()
102 .get(&entity_tag)
103 .cloned()
104 .ok_or_else(InternalError::store_corruption)?;
105 if snapshot.entity_path() != entity_path {
106 return Err(InternalError::store_corruption());
107 }
108 let accepted = AcceptedSchemaSnapshot::try_new(snapshot)?;
109 let _runtime_contract = AcceptedRowLayoutRuntimeContract::from_accepted_schema(&accepted)?;
110
111 Ok(accepted)
112}
113
114#[cfg(test)]
115thread_local! {
116 static ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES: Cell<u64> = const { Cell::new(0) };
117}
118
119#[cfg(test)]
120fn reset_accepted_schema_bundle_cache_miss_count_for_tests() {
121 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(|misses| misses.set(0));
122}
123
124#[cfg(test)]
125fn accepted_schema_bundle_cache_miss_count_for_tests() -> u64 {
126 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(Cell::get)
127}
128
129#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
138struct RawSchemaKey([u8; SCHEMA_KEY_BYTES_USIZE]);
139
140impl RawSchemaKey {
141 #[must_use]
143 fn from_entity_version(entity: EntityTag, version: SchemaVersion) -> Self {
144 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
145 out[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
146 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
147 out[12..].copy_from_slice(&version.get().to_be_bytes());
148
149 Self(out)
150 }
151
152 fn from_accepted_bundle(bundle_key: super::enum_catalog::AcceptedSchemaBundleKey) -> Self {
153 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
154 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE;
155 out[4..12].copy_from_slice(&bundle_key.get().to_be_bytes());
156 Self(out)
157 }
158
159 fn from_accepted_root_slot(slot: usize) -> Result<Self, InternalError> {
160 let slot = u32::try_from(slot).map_err(|_| InternalError::store_invariant())?;
161 if slot > 1 {
162 return Err(InternalError::store_invariant());
163 }
164 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
165 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT;
166 out[12..].copy_from_slice(&slot.to_be_bytes());
167 Ok(Self(out))
168 }
169
170 fn from_constraint_validation_job(entity: EntityTag, constraint_id: ConstraintId) -> Self {
171 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
172 out[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
173 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
174 out[12..].copy_from_slice(&constraint_id.get().to_be_bytes());
175 Self(out)
176 }
177
178 fn from_identity_state(entity: EntityTag, field_id: FieldId) -> Self {
179 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
180 out[0] = SCHEMA_KEY_NAMESPACE_IDENTITY_STATE;
181 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
182 out[12..].copy_from_slice(&field_id.get().to_be_bytes());
183 Self(out)
184 }
185
186 #[must_use]
188 fn entity_tag(self) -> EntityTag {
189 let mut bytes = [0u8; size_of::<u64>()];
190 bytes.copy_from_slice(&self.0[4..12]);
191
192 EntityTag::new(u64::from_be_bytes(bytes))
193 }
194
195 #[must_use]
197 fn version(self) -> u32 {
198 let mut bytes = [0u8; size_of::<u32>()];
199 bytes.copy_from_slice(&self.0[12..]);
200
201 u32::from_be_bytes(bytes)
202 }
203
204 const fn all_entity_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
205 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
206 end[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
207 (
208 RangeBound::Included(Self([0; SCHEMA_KEY_BYTES_USIZE])),
209 RangeBound::Included(Self(end)),
210 )
211 }
212
213 #[cfg(test)]
214 fn entity_range_bounds(entity: EntityTag) -> (RangeBound<Self>, RangeBound<Self>) {
215 (
216 RangeBound::Included(Self::from_entity_version(entity, SchemaVersion::initial())),
217 RangeBound::Included(Self::from_entity_version(
218 entity,
219 SchemaVersion::new(u32::MAX),
220 )),
221 )
222 }
223
224 const fn all_constraint_validation_job_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
225 let mut start = [0u8; SCHEMA_KEY_BYTES_USIZE];
226 start[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
227 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
228 end[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
229 (
230 RangeBound::Included(Self(start)),
231 RangeBound::Included(Self(end)),
232 )
233 }
234
235 const fn all_identity_state_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
236 let mut start = [0u8; SCHEMA_KEY_BYTES_USIZE];
237 start[0] = SCHEMA_KEY_NAMESPACE_IDENTITY_STATE;
238 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
239 end[0] = SCHEMA_KEY_NAMESPACE_IDENTITY_STATE;
240 (
241 RangeBound::Included(Self(start)),
242 RangeBound::Included(Self(end)),
243 )
244 }
245
246 #[cfg(test)]
247 const fn is_entity_snapshot(self) -> bool {
248 self.0[0] == SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT
249 }
250
251 const fn is_accepted_root(self) -> bool {
252 self.0[0] == SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT
253 }
254
255 const fn is_constraint_validation_job(self) -> bool {
256 self.0[0] == SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB
257 }
258
259 const fn is_identity_state(self) -> bool {
260 self.0[0] == SCHEMA_KEY_NAMESPACE_IDENTITY_STATE
261 }
262
263 fn constraint_id(self) -> Option<ConstraintId> {
264 self.is_constraint_validation_job()
265 .then(|| ConstraintId::new(self.version()))
266 .flatten()
267 }
268}
269
270impl Storable for RawSchemaKey {
271 fn to_bytes(&self) -> Cow<'_, [u8]> {
272 Cow::Borrowed(&self.0)
273 }
274
275 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
276 debug_assert_eq!(
277 bytes.len(),
278 SCHEMA_KEY_BYTES_USIZE,
279 "RawSchemaKey::from_bytes received unexpected byte length",
280 );
281
282 if bytes.len() != SCHEMA_KEY_BYTES_USIZE {
283 return Self([0u8; SCHEMA_KEY_BYTES_USIZE]);
284 }
285
286 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
287 out.copy_from_slice(bytes.as_ref());
288 Self(out)
289 }
290
291 fn into_bytes(self) -> Vec<u8> {
292 self.0.to_vec()
293 }
294
295 const BOUND: StorableBound = StorableBound::Bounded {
296 max_size: SCHEMA_KEY_BYTES,
297 is_fixed_size: true,
298 };
299}
300
301#[derive(Clone, Debug, Eq, PartialEq)]
312struct RawSchemaSnapshot {
313 payload: Vec<u8>,
314 accepted_schema_fingerprint: Option<CommitSchemaFingerprint>,
315}
316
317impl RawSchemaSnapshot {
318 fn from_persisted_snapshot(snapshot: &PersistedSchemaSnapshot) -> Result<Self, InternalError> {
320 validate_typed_schema_snapshot_for_store(snapshot)?;
321
322 let accepted_schema_fingerprint =
323 accepted_schema_cache_fingerprint_for_persisted_snapshot(snapshot)?;
324 let payload = encode_persisted_schema_snapshot(snapshot)?;
325
326 Ok(Self {
327 payload,
328 accepted_schema_fingerprint: Some(accepted_schema_fingerprint),
329 })
330 }
331
332 #[must_use]
334 const fn from_encoded_control_record(payload: Vec<u8>) -> Self {
335 Self {
336 payload,
337 accepted_schema_fingerprint: None,
338 }
339 }
340
341 #[cfg(test)]
344 #[must_use]
345 const fn from_unchecked_persisted_snapshot_payload(payload: Vec<u8>) -> Self {
346 Self {
347 payload,
348 accepted_schema_fingerprint: Some([0; size_of::<CommitSchemaFingerprint>()]),
349 }
350 }
351
352 #[must_use]
354 const fn as_bytes(&self) -> &[u8] {
355 self.payload.as_slice()
356 }
357
358 #[must_use]
360 fn into_bytes(self) -> Vec<u8> {
361 self.payload
362 }
363
364 fn accepted_schema_fingerprint(&self) -> Result<CommitSchemaFingerprint, InternalError> {
367 self.accepted_schema_fingerprint
368 .ok_or_else(InternalError::store_corruption)
369 }
370
371 fn decode_persisted_snapshot(&self) -> Result<PersistedSchemaSnapshot, InternalError> {
373 let _fingerprint = self.accepted_schema_fingerprint()?;
376 decode_persisted_schema_snapshot(self.as_bytes())
377 }
378}
379
380#[cfg(test)]
381pub(in crate::db::schema) fn validate_raw_schema_snapshot_bytes_for_tests(
382 bytes: Vec<u8>,
383) -> Result<(), InternalError> {
384 let raw = <RawSchemaSnapshot as Storable>::from_bytes(Cow::Owned(bytes));
385 raw.decode_persisted_snapshot().map(drop)
386}
387
388#[derive(Clone, Debug, Eq, PartialEq)]
389pub(in crate::db) struct AcceptedCatalogIdentity {
390 entity_tag: EntityTag,
391 entity_path: Rc<str>,
392 store_path: &'static str,
393 accepted_schema_revision: AcceptedSchemaRevision,
394 accepted_schema_version: SchemaVersion,
395 fingerprint_method_version: u8,
396 accepted_schema_fingerprint: CommitSchemaFingerprint,
397}
398
399impl AcceptedCatalogIdentity {
400 #[must_use]
401 pub(in crate::db) fn new(
402 entity_tag: EntityTag,
403 entity_path: impl Into<Rc<str>>,
404 store_path: &'static str,
405 accepted_schema_revision: AcceptedSchemaRevision,
406 accepted_schema_version: SchemaVersion,
407 accepted_schema_fingerprint: CommitSchemaFingerprint,
408 ) -> Self {
409 Self {
410 entity_tag,
411 entity_path: entity_path.into(),
412 store_path,
413 accepted_schema_revision,
414 accepted_schema_version,
415 fingerprint_method_version: accepted_schema_cache_fingerprint_method_version(),
416 accepted_schema_fingerprint,
417 }
418 }
419
420 #[must_use]
421 pub(in crate::db) const fn entity_tag(&self) -> EntityTag {
422 self.entity_tag
423 }
424
425 #[must_use]
426 pub(in crate::db) fn entity_path(&self) -> &str {
427 self.entity_path.as_ref()
428 }
429
430 #[must_use]
431 pub(in crate::db) fn entity_path_handle(&self) -> Rc<str> {
432 self.entity_path.clone()
433 }
434
435 #[must_use]
436 pub(in crate::db) const fn store_path(&self) -> &'static str {
437 self.store_path
438 }
439
440 #[must_use]
441 pub(in crate::db) const fn accepted_schema_revision(&self) -> AcceptedSchemaRevision {
442 self.accepted_schema_revision
443 }
444
445 #[must_use]
446 pub(in crate::db) const fn accepted_schema_version(&self) -> SchemaVersion {
447 self.accepted_schema_version
448 }
449
450 #[must_use]
451 pub(in crate::db) const fn fingerprint_method_version(&self) -> u8 {
452 self.fingerprint_method_version
453 }
454
455 #[must_use]
456 pub(in crate::db) const fn accepted_schema_fingerprint(&self) -> CommitSchemaFingerprint {
457 self.accepted_schema_fingerprint
458 }
459}
460
461#[derive(Clone, Debug, Eq, PartialEq)]
462pub(in crate::db) struct AcceptedCatalogSnapshotSelection {
463 identity: AcceptedCatalogIdentity,
464 value_catalog: AcceptedValueCatalogHandle,
465 raw_snapshot: Rc<[u8]>,
466}
467
468impl AcceptedCatalogSnapshotSelection {
469 #[must_use]
470 const fn new(
471 identity: AcceptedCatalogIdentity,
472 value_catalog: AcceptedValueCatalogHandle,
473 raw_snapshot: Rc<[u8]>,
474 ) -> Self {
475 Self {
476 identity,
477 value_catalog,
478 raw_snapshot,
479 }
480 }
481
482 #[must_use]
483 pub(in crate::db) fn identity(&self) -> AcceptedCatalogIdentity {
484 self.identity.clone()
485 }
486
487 #[must_use]
488 pub(in crate::db) const fn value_catalog_handle(&self) -> &AcceptedValueCatalogHandle {
489 &self.value_catalog
490 }
491
492 pub(in crate::db) fn from_candidate(
495 candidate: &CandidateSchemaRevision,
496 entity_tag: EntityTag,
497 entity_path: &str,
498 store_path: &'static str,
499 ) -> Result<Option<Self>, InternalError> {
500 if candidate.store_path() != store_path {
501 return Err(InternalError::store_corruption());
502 }
503 let Some(snapshot) = candidate.bundle().entity_snapshots().get(&entity_tag) else {
504 return Ok(None);
505 };
506 if snapshot.entity_path() != entity_path {
507 return Err(InternalError::store_corruption());
508 }
509
510 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
511 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
512 let identity = AcceptedCatalogIdentity::new(
513 entity_tag,
514 entity_path,
515 store_path,
516 candidate.revision(),
517 snapshot.version(),
518 fingerprint,
519 );
520
521 Ok(Some(Self::new(
522 identity,
523 AcceptedValueCatalogHandle::new(
524 candidate.bundle().enum_catalog().clone(),
525 candidate.bundle().composite_catalog().clone(),
526 AcceptedStoreCatalogScope::new(),
527 candidate.revision(),
528 candidate.root().fingerprint(),
529 ),
530 Rc::from(raw_snapshot.into_bytes()),
531 )))
532 }
533
534 pub(in crate::db) fn decode_verified(&self) -> Result<AcceptedSchemaSnapshot, InternalError> {
535 let snapshot = decode_persisted_schema_snapshot(self.raw_snapshot.as_ref())?;
536 let accepted = AcceptedSchemaSnapshot::try_new(snapshot)?;
537 let identity = self.identity();
538
539 if accepted.persisted_snapshot().version() != identity.accepted_schema_version() {
540 return Err(InternalError::store_invariant());
541 }
542 if accepted.entity_path() != identity.entity_path() {
543 return Err(InternalError::store_invariant());
544 }
545
546 let decoded_fingerprint = accepted_schema_cache_fingerprint(&accepted)?;
547 if decoded_fingerprint != identity.accepted_schema_fingerprint() {
548 return Err(InternalError::store_invariant());
549 }
550
551 Ok(accepted)
552 }
553}
554
555impl Storable for RawSchemaSnapshot {
556 fn to_bytes(&self) -> Cow<'_, [u8]> {
557 let Some(fingerprint) = self.accepted_schema_fingerprint else {
558 return Cow::Borrowed(self.as_bytes());
559 };
560
561 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
562 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
563 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
564 bytes.extend_from_slice(&fingerprint);
565 bytes.extend_from_slice(self.as_bytes());
566
567 Cow::Owned(bytes)
568 }
569
570 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
571 let bytes = bytes.into_owned();
572 if bytes.len() >= RAW_SCHEMA_SNAPSHOT_HEADER_BYTES
573 && &bytes[..RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_MAGIC
574 && bytes[RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_VALUE_VERSION
575 {
576 let fingerprint_start = RAW_SCHEMA_SNAPSHOT_MAGIC.len() + size_of::<u8>();
577 let fingerprint_end = fingerprint_start + size_of::<CommitSchemaFingerprint>();
578 let mut fingerprint = [0_u8; size_of::<CommitSchemaFingerprint>()];
579 fingerprint.copy_from_slice(&bytes[fingerprint_start..fingerprint_end]);
580
581 return Self {
582 payload: bytes[fingerprint_end..].to_vec(),
583 accepted_schema_fingerprint: Some(fingerprint),
584 };
585 }
586
587 Self {
588 payload: bytes,
589 accepted_schema_fingerprint: None,
590 }
591 }
592
593 fn into_bytes(self) -> Vec<u8> {
594 let Some(fingerprint) = self.accepted_schema_fingerprint else {
595 return self.payload;
596 };
597
598 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
599 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
600 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
601 bytes.extend_from_slice(&fingerprint);
602 bytes.extend_from_slice(&self.payload);
603
604 bytes
605 }
606
607 const BOUND: StorableBound = StorableBound::Unbounded;
608}
609
610fn validate_typed_schema_snapshot_for_store(
614 snapshot: &PersistedSchemaSnapshot,
615) -> Result<(), InternalError> {
616 if schema_snapshot_integrity_detail(
617 "schema snapshot",
618 snapshot.version(),
619 snapshot.primary_key_field_ids(),
620 snapshot.row_layout(),
621 snapshot.fields(),
622 )
623 .is_some()
624 {
625 return Err(InternalError::store_invariant());
626 }
627
628 Ok(())
629}
630
631#[derive(Clone, Copy, Debug, Eq, PartialEq)]
639pub(in crate::db) struct SchemaStoreCatalogMetadata {
640 schema_version: SchemaVersion,
641 schema_fingerprint_method_version: u8,
642 schema_fingerprint: CommitSchemaFingerprint,
643 entity_count: u64,
644}
645
646impl SchemaStoreCatalogMetadata {
647 #[must_use]
649 const fn new(
650 schema_version: SchemaVersion,
651 schema_fingerprint_method_version: u8,
652 schema_fingerprint: CommitSchemaFingerprint,
653 entity_count: u64,
654 ) -> Self {
655 Self {
656 schema_version,
657 schema_fingerprint_method_version,
658 schema_fingerprint,
659 entity_count,
660 }
661 }
662
663 #[must_use]
665 pub(in crate::db) const fn schema_version(self) -> SchemaVersion {
666 self.schema_version
667 }
668
669 #[must_use]
671 pub(in crate::db) const fn schema_fingerprint_method_version(self) -> u8 {
672 self.schema_fingerprint_method_version
673 }
674
675 #[must_use]
678 pub(in crate::db) const fn schema_fingerprint(self) -> CommitSchemaFingerprint {
679 self.schema_fingerprint
680 }
681
682 #[must_use]
684 pub(in crate::db) const fn entity_count(self) -> u64 {
685 self.entity_count
686 }
687}
688
689#[derive(Clone, Copy, Debug, Eq, PartialEq)]
698pub(in crate::db) struct SchemaStoreAllocationMetadata {
699 data: SchemaStoreCatalogMetadata,
700 index: SchemaStoreCatalogMetadata,
701 schema: SchemaStoreCatalogMetadata,
702}
703
704impl SchemaStoreAllocationMetadata {
705 #[must_use]
708 const fn new(
709 data: SchemaStoreCatalogMetadata,
710 index: SchemaStoreCatalogMetadata,
711 schema: SchemaStoreCatalogMetadata,
712 ) -> Self {
713 Self {
714 data,
715 index,
716 schema,
717 }
718 }
719
720 #[must_use]
722 pub(in crate::db) const fn data(self) -> SchemaStoreCatalogMetadata {
723 self.data
724 }
725
726 #[must_use]
728 pub(in crate::db) const fn index(self) -> SchemaStoreCatalogMetadata {
729 self.index
730 }
731
732 #[must_use]
735 pub(in crate::db) const fn schema(self) -> SchemaStoreCatalogMetadata {
736 self.schema
737 }
738}
739
740pub(in crate::db) struct PendingRelationActivationDeleteBarrier {
748 accepted_schema_fingerprint: CommitSchemaFingerprint,
749 source_entity_tag: EntityTag,
750 constraint_id: ConstraintId,
751}
752
753impl PendingRelationActivationDeleteBarrier {
754 #[must_use]
755 pub(in crate::db) const fn accepted_schema_fingerprint(&self) -> CommitSchemaFingerprint {
756 self.accepted_schema_fingerprint
757 }
758
759 #[must_use]
760 pub(in crate::db) const fn source_entity_tag(&self) -> EntityTag {
761 self.source_entity_tag
762 }
763
764 #[must_use]
766 pub(in crate::db) const fn constraint_id(&self) -> ConstraintId {
767 self.constraint_id
768 }
769}
770
771pub struct SchemaStore {
780 backend: SchemaStoreBackend,
781 accepted_bundle_cache: RefCell<Option<AcceptedSchemaBundleCache>>,
782 accepted_catalog_scope: OnceCell<AcceptedStoreCatalogScope>,
783}
784
785struct AcceptedSchemaBundleCache {
786 selection: AcceptedSchemaRootSelection,
787 bundle: AcceptedSchemaRevisionBundle,
788 value_catalog: AcceptedValueCatalogHandle,
789 entity_selections: RefCell<StdBTreeMap<EntityTag, AcceptedCatalogSnapshotSelection>>,
790}
791
792enum SchemaStoreBackend {
793 Heap(StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>),
794 Journaled {
795 canonical:
796 StableBTreeMap<RawSchemaKey, RawSchemaSnapshot, VirtualMemory<DefaultMemoryImpl>>,
797 live: StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
798 tombstones: BTreeSet<RawSchemaKey>,
799 positions: PositionedOverlayMetadata<RawSchemaKey>,
800 },
801}
802
803#[derive(Clone, Copy, Debug, Eq, PartialEq)]
805enum SchemaStoreVisit {
806 Continue,
807 #[cfg(test)]
808 Stop,
809}
810
811impl SchemaStoreVisit {
812 const fn should_stop(self) -> bool {
813 match self {
814 Self::Continue => false,
815 #[cfg(test)]
816 Self::Stop => true,
817 }
818 }
819}
820
821#[derive(Clone, Copy)]
822enum IdentityStateStorageView {
823 Effective,
824 Canonical,
825}
826
827#[derive(Clone)]
829pub(in crate::db) struct PreparedSchemaPositionPublication {
830 keys: Vec<RawSchemaKey>,
831 position: JournalOverlayPosition,
832}
833
834pub(in crate::db) struct PreparedSchemaPositionRetirement {
836 entries: Vec<(RawSchemaKey, PositionedOverlayRetirement)>,
837}
838
839#[derive(Clone, Copy)]
840enum IdentityStateWriteTarget {
841 Durable,
842 Materialized,
843 Canonical,
844}
845
846impl SchemaStore {
847 #[must_use]
849 pub const fn init_heap() -> Self {
850 Self {
851 backend: SchemaStoreBackend::Heap(StdBTreeMap::new()),
852 accepted_bundle_cache: RefCell::new(None),
853 accepted_catalog_scope: OnceCell::new(),
854 }
855 }
856
857 #[must_use]
862 pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
863 Self {
864 backend: SchemaStoreBackend::Journaled {
865 canonical: StableBTreeMap::init(memory),
866 live: StdBTreeMap::new(),
867 tombstones: BTreeSet::new(),
868 positions: PositionedOverlayMetadata::new(),
869 },
870 accepted_bundle_cache: RefCell::new(None),
871 accepted_catalog_scope: OnceCell::new(),
872 }
873 }
874
875 pub(in crate::db) fn preflight_fold_recovered_journal(&self) -> Result<(), InternalError> {
877 match self.backend {
878 SchemaStoreBackend::Journaled { .. } => Ok(()),
879 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
880 }
881 }
882
883 fn prepare_identity_state_transition(
884 &self,
885 incarnation: DatabaseIncarnationId,
886 candidate: &CandidateSchemaRevision,
887 view: IdentityStateStorageView,
888 ) -> Result<IdentityStateTransition, InternalError> {
889 let current = match view {
890 IdentityStateStorageView::Effective => self
891 .current_accepted_schema_bundle_ref()?
892 .as_ref()
893 .map(|bundle| (*bundle).clone()),
894 IdentityStateStorageView::Canonical => {
895 self.current_canonical_accepted_schema_bundle()?
896 }
897 };
898 let inventory = self.identity_state_inventory(view)?;
899 prepare_identity_state_transition(
900 incarnation,
901 current.as_ref(),
902 candidate.bundle(),
903 inventory,
904 )
905 }
906
907 fn validate_identity_state_closure(
908 &self,
909 bundle: &AcceptedSchemaRevisionBundle,
910 ) -> Result<(), InternalError> {
911 let inventory = self.identity_state_inventory(IdentityStateStorageView::Effective)?;
912 validate_identity_state_closure(bundle, &inventory)
913 }
914
915 pub(in crate::db) fn identity_statement_cursor(
917 &self,
918 database_incarnation_id: DatabaseIncarnationId,
919 entity_tag: EntityTag,
920 field_id: FieldId,
921 accepted_kind: &AcceptedFieldKind,
922 ) -> Result<IdentityStatementCursor, InternalError> {
923 let key = RawSchemaKey::from_identity_state(entity_tag, field_id);
924 let raw = self
925 .get_raw_snapshot(&key)
926 .ok_or_else(InternalError::identity_state_corruption)?;
927 let state = decode_identity_state(raw.as_bytes())?;
928 let owner = state.owner();
929 if owner.database_incarnation_id() != database_incarnation_id
930 || owner.entity_tag() != entity_tag
931 || owner.field_id() != field_id
932 || state.accepted_kind() != accepted_kind
933 || state.lifecycle() != IdentityStateLifecycle::Active
934 {
935 return Err(InternalError::identity_state_corruption());
936 }
937 IdentityStatementCursor::from_active_state(&state)
938 }
939
940 pub(in crate::db) fn identity_high_water_for_integrity(
942 &self,
943 database_incarnation_id: DatabaseIncarnationId,
944 entity_tag: EntityTag,
945 field_id: FieldId,
946 accepted_kind: &AcceptedFieldKind,
947 ) -> Result<u128, InternalError> {
948 let key = RawSchemaKey::from_identity_state(entity_tag, field_id);
949 let raw = self
950 .get_raw_snapshot(&key)
951 .ok_or_else(InternalError::identity_state_corruption)?;
952 let state = decode_identity_state(raw.as_bytes())?;
953 let owner = state.owner();
954 if owner.database_incarnation_id() != database_incarnation_id
955 || owner.entity_tag() != entity_tag
956 || owner.field_id() != field_id
957 || state.accepted_kind() != accepted_kind
958 || state.lifecycle() != IdentityStateLifecycle::Active
959 {
960 return Err(InternalError::identity_state_corruption());
961 }
962 Ok(state.materialized_high_water())
963 }
964
965 pub(in crate::db) fn preflight_identity_range_advance(
967 &self,
968 range: IdentityRangeAdvance,
969 ) -> Result<(), InternalError> {
970 let state =
971 self.identity_state_for_owner(range.owner(), IdentityStateStorageView::Effective)?;
972 if state.lifecycle() != IdentityStateLifecycle::Active
973 || range.new_high_water()
974 > identity_kind_maximum(state.accepted_kind())
975 .ok_or_else(InternalError::identity_state_corruption)?
976 {
977 return Err(InternalError::identity_state_corruption());
978 }
979 if state.materialized_high_water() != range.expected_high_water() {
980 return Err(InternalError::identity_state_conflict());
981 }
982 Ok(())
983 }
984
985 pub(in crate::db) fn apply_identity_range_advance(
987 &mut self,
988 range: IdentityRangeAdvance,
989 advance_id: IdentityAdvanceId,
990 ) -> Result<(), InternalError> {
991 self.apply_identity_range_advance_to(
992 range,
993 advance_id,
994 IdentityStateStorageView::Effective,
995 IdentityStateWriteTarget::Materialized,
996 )
997 }
998
999 pub(in crate::db) fn fold_identity_range_advance(
1001 &mut self,
1002 range: IdentityRangeAdvance,
1003 advance_id: IdentityAdvanceId,
1004 ) -> Result<(), InternalError> {
1005 self.apply_identity_range_advance_to(
1006 range,
1007 advance_id,
1008 IdentityStateStorageView::Canonical,
1009 IdentityStateWriteTarget::Canonical,
1010 )
1011 }
1012
1013 pub(in crate::db) fn preflight_fold_identity_range_advance(
1015 &self,
1016 range: IdentityRangeAdvance,
1017 advance_id: IdentityAdvanceId,
1018 ) -> Result<(), InternalError> {
1019 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1020 return Err(InternalError::store_invariant());
1021 }
1022 let state =
1023 self.identity_state_for_owner(range.owner(), IdentityStateStorageView::Canonical)?;
1024 let advanced = state.apply_range_advance(range, advance_id)?;
1025 let _encoded = encode_identity_state(&advanced)?;
1026 Ok(())
1027 }
1028
1029 pub(in crate::db) fn verify_identity_range_advance(
1031 &self,
1032 range: IdentityRangeAdvance,
1033 advance_id: IdentityAdvanceId,
1034 ) -> Result<(), InternalError> {
1035 let state =
1036 self.identity_state_for_owner(range.owner(), IdentityStateStorageView::Effective)?;
1037 if state.materialized_high_water() != range.new_high_water()
1038 || state.last_applied_advance() != Some(advance_id)
1039 {
1040 return Err(InternalError::recovery_effect_verification_failed());
1041 }
1042 Ok(())
1043 }
1044
1045 pub(in crate::db) fn identity_range_commit_state(
1047 &self,
1048 range: IdentityRangeAdvance,
1049 advance_id: IdentityAdvanceId,
1050 canonical: bool,
1051 ) -> Result<IdentityRangeCommitState, InternalError> {
1052 let view = if canonical {
1053 IdentityStateStorageView::Canonical
1054 } else {
1055 IdentityStateStorageView::Effective
1056 };
1057 self.identity_state_for_owner(range.owner(), view)?
1058 .range_commit_state(range, advance_id)
1059 }
1060
1061 pub(in crate::db) fn identity_state_inventory_for_integrity(
1064 &self,
1065 incarnation: DatabaseIncarnationId,
1066 ) -> Result<Vec<IdentityState>, InternalError> {
1067 let has_accepted_bundle = self.current_accepted_schema_bundle_ref()?.is_some();
1068 let inventory = self.identity_state_inventory(IdentityStateStorageView::Effective)?;
1069 if !has_accepted_bundle && !inventory.is_empty() {
1070 return Err(InternalError::identity_state_corruption());
1071 }
1072 if inventory
1073 .values()
1074 .any(|state| state.owner().database_incarnation_id() != incarnation)
1075 {
1076 return Err(InternalError::identity_state_corruption());
1077 }
1078 Ok(inventory.into_values().collect())
1079 }
1080
1081 fn identity_state_for_owner(
1082 &self,
1083 owner: crate::db::schema::identity_state::IdentityStateOwner,
1084 view: IdentityStateStorageView,
1085 ) -> Result<IdentityState, InternalError> {
1086 let key = RawSchemaKey::from_identity_state(owner.entity_tag(), owner.field_id());
1087 let raw = match view {
1088 IdentityStateStorageView::Effective => self.get_raw_snapshot(&key),
1089 IdentityStateStorageView::Canonical => self.get_canonical_raw_value(&key)?,
1090 }
1091 .ok_or_else(InternalError::identity_state_corruption)?;
1092 let state = decode_identity_state(raw.as_bytes())?;
1093 if state.owner() != owner {
1094 return Err(InternalError::identity_state_corruption());
1095 }
1096 Ok(state)
1097 }
1098
1099 fn apply_identity_range_advance_to(
1100 &mut self,
1101 range: IdentityRangeAdvance,
1102 advance_id: IdentityAdvanceId,
1103 view: IdentityStateStorageView,
1104 target: IdentityStateWriteTarget,
1105 ) -> Result<(), InternalError> {
1106 let state = self.identity_state_for_owner(range.owner(), view)?;
1107 let advanced = state.apply_range_advance(range, advance_id)?;
1108 let key = RawSchemaKey::from_identity_state(
1109 advanced.owner().entity_tag(),
1110 advanced.owner().field_id(),
1111 );
1112 let bytes = encode_identity_state(&advanced)?;
1113 match target {
1114 IdentityStateWriteTarget::Materialized => {
1115 self.insert_raw_snapshot(
1116 key,
1117 RawSchemaSnapshot::from_encoded_control_record(bytes),
1118 );
1119 }
1120 IdentityStateWriteTarget::Canonical => {
1121 self.insert_canonical_raw_value(key, bytes)?;
1122 }
1123 IdentityStateWriteTarget::Durable => {
1124 return Err(InternalError::store_invariant());
1125 }
1126 }
1127 Ok(())
1128 }
1129
1130 fn identity_state_inventory(
1131 &self,
1132 view: IdentityStateStorageView,
1133 ) -> Result<IdentityStateInventory, InternalError> {
1134 let bounds = RawSchemaKey::all_identity_state_range_bounds();
1135 let mut inventory = StdBTreeMap::new();
1136 let mut collect = |key: &RawSchemaKey,
1137 raw: &RawSchemaSnapshot|
1138 -> Result<SchemaStoreVisit, InternalError> {
1139 if inventory.len() >= MAX_IDENTITY_STATE_RECORDS_PER_DATABASE {
1140 return Err(InternalError::identity_state_corruption());
1141 }
1142 let state = decode_identity_state(raw.as_bytes())?;
1143 let state_key = (key.entity_tag(), FieldId::new(key.version()));
1144 if !key.is_identity_state()
1145 || state.owner().entity_tag() != state_key.0
1146 || state.owner().field_id() != state_key.1
1147 || inventory.insert(state_key, state).is_some()
1148 {
1149 return Err(InternalError::identity_state_corruption());
1150 }
1151 Ok(SchemaStoreVisit::Continue)
1152 };
1153
1154 match (&self.backend, view) {
1155 (SchemaStoreBackend::Heap(map), IdentityStateStorageView::Effective) => {
1156 for (key, raw) in map.range((bounds.0, bounds.1)) {
1157 collect(key, raw)?;
1158 }
1159 }
1160 (
1161 SchemaStoreBackend::Journaled {
1162 canonical,
1163 live,
1164 tombstones,
1165 ..
1166 },
1167 IdentityStateStorageView::Effective,
1168 ) => Self::visit_journaled_raw_snapshot_range(
1169 canonical,
1170 live,
1171 tombstones,
1172 bounds,
1173 Direction::Asc,
1174 &mut collect,
1175 )?,
1176 (
1177 SchemaStoreBackend::Journaled { canonical, .. },
1178 IdentityStateStorageView::Canonical,
1179 ) => {
1180 for entry in canonical.range((bounds.0, bounds.1)) {
1181 collect(entry.key(), &entry.value())?;
1182 }
1183 }
1184 (SchemaStoreBackend::Heap(_), IdentityStateStorageView::Canonical) => {
1185 return Err(InternalError::store_invariant());
1186 }
1187 }
1188
1189 Ok(inventory)
1190 }
1191
1192 fn apply_identity_state_transition(
1193 &mut self,
1194 transition: IdentityStateTransition,
1195 target: IdentityStateWriteTarget,
1196 ) -> Result<(), InternalError> {
1197 for state in transition.into_updates() {
1198 let key = RawSchemaKey::from_identity_state(
1199 state.owner().entity_tag(),
1200 state.owner().field_id(),
1201 );
1202 let bytes = encode_identity_state(&state)?;
1203 match target {
1204 IdentityStateWriteTarget::Durable => {
1205 self.insert_durable_raw_value(key, bytes);
1206 }
1207 IdentityStateWriteTarget::Materialized => {
1208 self.insert_raw_snapshot(
1209 key,
1210 RawSchemaSnapshot::from_encoded_control_record(bytes),
1211 );
1212 }
1213 IdentityStateWriteTarget::Canonical => {
1214 self.insert_canonical_raw_value(key, bytes)?;
1215 }
1216 }
1217 }
1218 Ok(())
1219 }
1220
1221 pub(in crate::db) fn current_canonical_accepted_schema_bundle(
1222 &self,
1223 ) -> Result<Option<AcceptedSchemaRevisionBundle>, InternalError> {
1224 let first = self.canonical_root_slot_bytes(0)?;
1225 let second = self.canonical_root_slot_bytes(1)?;
1226 let Some(selection) =
1227 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1228 else {
1229 return Ok(None);
1230 };
1231 let bundle_key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
1232 let raw = self
1233 .get_canonical_raw_value(&bundle_key)?
1234 .ok_or_else(InternalError::store_corruption)?;
1235 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes()).map(Some)
1236 }
1237
1238 pub(in crate::db) fn insert_persisted_snapshot(
1240 &mut self,
1241 entity: EntityTag,
1242 snapshot: &PersistedSchemaSnapshot,
1243 ) -> Result<(), InternalError> {
1244 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
1245 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1246 let _ = self.insert_raw_snapshot(key, raw_snapshot);
1247
1248 Ok(())
1249 }
1250
1251 pub(in crate::db) fn constraint_validation_job(
1253 &self,
1254 entity: EntityTag,
1255 constraint_id: ConstraintId,
1256 ) -> Result<Option<ConstraintValidationJob>, InternalError> {
1257 let key = RawSchemaKey::from_constraint_validation_job(entity, constraint_id);
1258 self.get_raw_snapshot(&key)
1259 .map(|raw| decode_constraint_validation_job(raw.as_bytes()))
1260 .transpose()
1261 }
1262
1263 pub(in crate::db) fn apply_constraint_validation_job(
1265 &mut self,
1266 job: &ConstraintValidationJob,
1267 ) -> Result<(), InternalError> {
1268 let key =
1269 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id());
1270 let bytes = encode_constraint_validation_job(job)?;
1271 let _ =
1272 self.insert_raw_snapshot(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
1273 Ok(())
1274 }
1275
1276 #[expect(
1278 clippy::unnecessary_wraps,
1279 reason = "marker apply operations share one fallible callback contract"
1280 )]
1281 pub(in crate::db) fn apply_constraint_validation_job_removal(
1282 &mut self,
1283 entity: EntityTag,
1284 constraint_id: ConstraintId,
1285 ) -> Result<(), InternalError> {
1286 let key = RawSchemaKey::from_constraint_validation_job(entity, constraint_id);
1287 match &mut self.backend {
1288 SchemaStoreBackend::Heap(map) => {
1289 map.remove(&key);
1290 }
1291 SchemaStoreBackend::Journaled {
1292 live, tombstones, ..
1293 } => {
1294 live.remove(&key);
1295 tombstones.insert(key);
1296 }
1297 }
1298 Ok(())
1299 }
1300
1301 pub(in crate::db) fn fold_constraint_validation_job(
1303 &mut self,
1304 job: &ConstraintValidationJob,
1305 ) -> Result<(), InternalError> {
1306 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1307 return Err(InternalError::store_invariant());
1308 };
1309 let key =
1310 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id());
1311 let bytes = encode_constraint_validation_job(job)?;
1312 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
1313 Ok(())
1314 }
1315
1316 pub(in crate::db) fn preflight_fold_constraint_validation_job(
1318 &self,
1319 job: &ConstraintValidationJob,
1320 ) -> Result<(), InternalError> {
1321 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1322 return Err(InternalError::store_invariant());
1323 }
1324 let _encoded = encode_constraint_validation_job(job)?;
1325 Ok(())
1326 }
1327
1328 pub(in crate::db) fn fold_constraint_validation_job_removal(
1330 &mut self,
1331 entity: EntityTag,
1332 constraint_id: ConstraintId,
1333 ) -> Result<(), InternalError> {
1334 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1335 return Err(InternalError::store_invariant());
1336 };
1337 canonical.remove(&RawSchemaKey::from_constraint_validation_job(
1338 entity,
1339 constraint_id,
1340 ));
1341 Ok(())
1342 }
1343
1344 pub(in crate::db) fn preflight_fold_constraint_validation_job_removal(
1346 &self,
1347 ) -> Result<(), InternalError> {
1348 match self.backend {
1349 SchemaStoreBackend::Journaled { .. } => Ok(()),
1350 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
1351 }
1352 }
1353
1354 pub(in crate::db) fn reset_journaled_live_projection(&mut self) -> Result<(), InternalError> {
1357 let SchemaStoreBackend::Journaled {
1358 live,
1359 tombstones,
1360 positions,
1361 ..
1362 } = &mut self.backend
1363 else {
1364 return Err(InternalError::store_invariant());
1365 };
1366
1367 live.clear();
1368 tombstones.clear();
1369 positions.clear();
1370 self.accepted_bundle_cache.get_mut().take();
1371
1372 Ok(())
1373 }
1374
1375 pub(in crate::db) fn prepare_positioned_journal_batch_publication(
1377 &self,
1378 incarnation: DatabaseIncarnationId,
1379 batch: &JournalBatch,
1380 position: JournalOverlayPosition,
1381 ) -> Result<PreparedSchemaPositionPublication, InternalError> {
1382 let SchemaStoreBackend::Journaled { positions, .. } = &self.backend else {
1383 return Err(InternalError::store_invariant());
1384 };
1385 let keys = self.positioned_journal_batch_keys(
1386 incarnation,
1387 batch,
1388 IdentityStateStorageView::Effective,
1389 )?;
1390 for key in &keys {
1391 positions.preflight_publish(key, position)?;
1392 }
1393 Ok(PreparedSchemaPositionPublication {
1394 keys: keys.into_iter().collect(),
1395 position,
1396 })
1397 }
1398
1399 pub(in crate::db) fn prepare_positioned_journal_batch_retirement(
1401 &self,
1402 incarnation: DatabaseIncarnationId,
1403 batch: &JournalBatch,
1404 position: JournalOverlayPosition,
1405 ) -> Result<PreparedSchemaPositionRetirement, InternalError> {
1406 let SchemaStoreBackend::Journaled { positions, .. } = &self.backend else {
1407 return Err(InternalError::store_invariant());
1408 };
1409 let keys = self.positioned_journal_batch_keys(
1410 incarnation,
1411 batch,
1412 IdentityStateStorageView::Canonical,
1413 )?;
1414 let entries = keys
1415 .into_iter()
1416 .map(|key| {
1417 positions
1418 .preflight_retirement(&key, position)
1419 .map(|retirement| (key, retirement))
1420 })
1421 .collect::<Result<Vec<_>, _>>()?;
1422 Ok(PreparedSchemaPositionRetirement { entries })
1423 }
1424
1425 pub(in crate::db) fn publish_prepared_journal_batch_positions(
1427 &mut self,
1428 prepared: PreparedSchemaPositionPublication,
1429 ) {
1430 let SchemaStoreBackend::Journaled { positions, .. } = &mut self.backend else {
1431 debug_assert!(
1432 false,
1433 "preflighted schema positions require a journaled store"
1434 );
1435 return;
1436 };
1437 for key in prepared.keys {
1438 positions.publish_preflighted(key, prepared.position);
1439 }
1440 }
1441
1442 pub(in crate::db) fn apply_prepared_journal_batch_retirement(
1444 &mut self,
1445 prepared: PreparedSchemaPositionRetirement,
1446 ) {
1447 for (key, retirement) in prepared.entries {
1448 if retirement != PositionedOverlayRetirement::Exact {
1449 continue;
1450 }
1451 self.invalidate_accepted_bundle_cache_for_key(key);
1452 let SchemaStoreBackend::Journaled {
1453 live,
1454 tombstones,
1455 positions,
1456 ..
1457 } = &mut self.backend
1458 else {
1459 debug_assert!(
1460 false,
1461 "preflighted schema retirement requires a journaled store"
1462 );
1463 return;
1464 };
1465 live.remove(&key);
1466 tombstones.remove(&key);
1467 positions.retire_preflighted(&key, retirement);
1468 }
1469 }
1470
1471 #[cfg(test)]
1472 fn publish_positioned_journal_entry(
1473 &mut self,
1474 key: RawSchemaKey,
1475 snapshot: Option<RawSchemaSnapshot>,
1476 position: JournalOverlayPosition,
1477 ) -> Result<Option<RawSchemaSnapshot>, InternalError> {
1478 let SchemaStoreBackend::Journaled { positions, .. } = &self.backend else {
1479 return Err(InternalError::store_invariant());
1480 };
1481 positions.preflight_publish(&key, position)?;
1482 self.invalidate_accepted_bundle_cache_for_key(key);
1483 let SchemaStoreBackend::Journaled {
1484 canonical,
1485 live,
1486 tombstones,
1487 positions,
1488 } = &mut self.backend
1489 else {
1490 return Err(InternalError::store_invariant());
1491 };
1492 let previous = if tombstones.contains(&key) {
1493 None
1494 } else {
1495 live.get(&key).cloned().or_else(|| canonical.get(&key))
1496 };
1497 if let Some(snapshot) = snapshot {
1498 tombstones.remove(&key);
1499 live.insert(key, snapshot);
1500 } else {
1501 live.remove(&key);
1502 tombstones.insert(key);
1503 }
1504 positions.publish_preflighted(key, position);
1505 Ok(previous)
1506 }
1507
1508 #[cfg(test)]
1509 fn retire_positioned_journal_effect(
1510 &mut self,
1511 key: RawSchemaKey,
1512 position: JournalOverlayPosition,
1513 ) -> Result<PositionedOverlayRetirement, InternalError> {
1514 let SchemaStoreBackend::Journaled { positions, .. } = &self.backend else {
1515 return Err(InternalError::store_invariant());
1516 };
1517 let retirement = positions.preflight_retirement(&key, position)?;
1518 let prepared = PreparedSchemaPositionRetirement {
1519 entries: vec![(key, retirement)],
1520 };
1521 self.apply_prepared_journal_batch_retirement(prepared);
1522 Ok(retirement)
1523 }
1524
1525 pub(in crate::db) fn fold_persisted_snapshot(
1527 &mut self,
1528 entity: EntityTag,
1529 snapshot: &PersistedSchemaSnapshot,
1530 ) -> Result<(), InternalError> {
1531 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1532 return Err(InternalError::store_invariant());
1533 };
1534
1535 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
1536 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1537 canonical.insert(key, raw_snapshot);
1538
1539 Ok(())
1540 }
1541
1542 pub(in crate::db) fn preflight_fold_persisted_snapshot(
1544 &self,
1545 snapshot: &PersistedSchemaSnapshot,
1546 ) -> Result<(), InternalError> {
1547 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1548 return Err(InternalError::store_invariant());
1549 }
1550 let _encoded = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1551 Ok(())
1552 }
1553
1554 pub(in crate::db) fn current_accepted_schema_root(
1556 &self,
1557 ) -> Result<Option<AcceptedSchemaRootSelection>, InternalError> {
1558 let first = self.accepted_root_slot_bytes(0)?;
1559 let second = self.accepted_root_slot_bytes(1)?;
1560 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])
1561 }
1562
1563 pub(in crate::db) fn current_accepted_schema_bundle(
1565 &self,
1566 ) -> Result<Option<AcceptedSchemaRevisionBundle>, InternalError> {
1567 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1568 return Ok(None);
1569 };
1570 self.validate_constraint_validation_job_closure(&bundle)?;
1571 Ok(Some(bundle.clone()))
1572 }
1573
1574 pub(in crate::db) fn current_accepted_runtime_entities(
1576 &self,
1577 registered_store_path: &'static str,
1578 ) -> Result<Vec<AcceptedRuntimeEntity>, InternalError> {
1579 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1580 return Ok(Vec::new());
1581 };
1582 if bundle.store_path() != registered_store_path {
1583 return Err(InternalError::store_corruption());
1584 }
1585
1586 bundle
1587 .entity_snapshots()
1588 .iter()
1589 .map(|(entity_tag, snapshot)| {
1590 AcceptedRuntimeEntity::from_accepted_snapshot(
1591 &bundle,
1592 *entity_tag,
1593 snapshot,
1594 registered_store_path,
1595 )
1596 })
1597 .collect()
1598 }
1599
1600 pub(in crate::db) fn current_accepted_runtime_entity_for_tag(
1602 &self,
1603 registered_store_path: &'static str,
1604 entity_tag: EntityTag,
1605 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1606 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1607 return Ok(None);
1608 };
1609 if bundle.store_path() != registered_store_path {
1610 return Err(InternalError::store_corruption());
1611 }
1612 let Some(snapshot) = bundle.entity_snapshots().get(&entity_tag) else {
1613 return Ok(None);
1614 };
1615
1616 AcceptedRuntimeEntity::from_accepted_snapshot(
1617 &bundle,
1618 entity_tag,
1619 snapshot,
1620 registered_store_path,
1621 )
1622 .map(Some)
1623 }
1624
1625 pub(in crate::db) fn current_canonical_accepted_runtime_entity_for_tag(
1627 &self,
1628 registered_store_path: &'static str,
1629 entity_tag: EntityTag,
1630 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1631 let Some(bundle) = self.current_canonical_accepted_schema_bundle()? else {
1632 return Ok(None);
1633 };
1634 if bundle.store_path() != registered_store_path {
1635 return Err(InternalError::store_corruption());
1636 }
1637 let Some(snapshot) = bundle.entity_snapshots().get(&entity_tag) else {
1638 return Ok(None);
1639 };
1640
1641 AcceptedRuntimeEntity::from_accepted_snapshot(
1642 &bundle,
1643 entity_tag,
1644 snapshot,
1645 registered_store_path,
1646 )
1647 .map(Some)
1648 }
1649
1650 pub(in crate::db) fn current_accepted_runtime_entity_for_path(
1652 &self,
1653 registered_store_path: &'static str,
1654 entity_path: &str,
1655 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1656 self.current_accepted_runtime_entity_matching(registered_store_path, |snapshot_path, _| {
1657 snapshot_path == entity_path
1658 })
1659 }
1660
1661 pub(in crate::db) fn current_canonical_accepted_runtime_entity_for_path(
1663 &self,
1664 registered_store_path: &'static str,
1665 entity_path: &str,
1666 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1667 let Some(bundle) = self.current_canonical_accepted_schema_bundle()? else {
1668 return Ok(None);
1669 };
1670 if bundle.store_path() != registered_store_path {
1671 return Err(InternalError::store_corruption());
1672 }
1673
1674 let mut matched = None;
1675 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1676 if snapshot.entity_path() != entity_path {
1677 continue;
1678 }
1679 let entity = AcceptedRuntimeEntity::from_accepted_snapshot(
1680 &bundle,
1681 *entity_tag,
1682 snapshot,
1683 registered_store_path,
1684 )?;
1685 if matched.replace(entity).is_some() {
1686 return Err(InternalError::store_corruption());
1687 }
1688 }
1689
1690 Ok(matched)
1691 }
1692
1693 #[cfg(test)]
1695 pub(in crate::db) fn current_accepted_runtime_entity_for_name(
1696 &self,
1697 registered_store_path: &'static str,
1698 entity_name: &str,
1699 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1700 self.current_accepted_runtime_entity_matching(registered_store_path, |_, snapshot_name| {
1701 snapshot_name == entity_name
1702 })
1703 }
1704
1705 fn current_accepted_runtime_entity_matching(
1706 &self,
1707 registered_store_path: &'static str,
1708 mut predicate: impl FnMut(&str, &str) -> bool,
1709 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
1710 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1711 return Ok(None);
1712 };
1713 if bundle.store_path() != registered_store_path {
1714 return Err(InternalError::store_corruption());
1715 }
1716
1717 let mut matched = None;
1718 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1719 if !predicate(snapshot.entity_path(), snapshot.entity_name()) {
1720 continue;
1721 }
1722 let entity = AcceptedRuntimeEntity::from_accepted_snapshot(
1723 &bundle,
1724 *entity_tag,
1725 snapshot,
1726 registered_store_path,
1727 )?;
1728 if matched.replace(entity).is_some() {
1729 return Err(InternalError::store_corruption());
1730 }
1731 }
1732
1733 Ok(matched)
1734 }
1735
1736 pub(in crate::db) fn current_accepted_schema_revision(
1738 &self,
1739 ) -> Result<Option<AcceptedSchemaRevision>, InternalError> {
1740 Ok(self
1741 .current_accepted_schema_root()?
1742 .map(|selection| selection.root().revision()))
1743 }
1744
1745 pub(in crate::db) fn pending_relation_activation_for_target(
1751 &self,
1752 target_path: &str,
1753 ) -> Result<Option<PendingRelationActivationDeleteBarrier>, InternalError> {
1754 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1755 return Ok(None);
1756 };
1757 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1758 let Some(candidate) = snapshot
1759 .candidate_relations()
1760 .iter()
1761 .find(|candidate| candidate.target_path() == target_path)
1762 else {
1763 continue;
1764 };
1765 let activation = snapshot
1766 .constraint_activations()
1767 .iter()
1768 .find(|activation| {
1769 matches!(
1770 activation.kind(),
1771 ConstraintActivationKind::Relation { relation_id }
1772 if *relation_id == candidate.id()
1773 )
1774 })
1775 .ok_or_else(InternalError::store_corruption)?;
1776 return Ok(Some(PendingRelationActivationDeleteBarrier {
1777 accepted_schema_fingerprint:
1778 accepted_schema_cache_fingerprint_for_persisted_snapshot(snapshot)?,
1779 source_entity_tag: *entity_tag,
1780 constraint_id: activation.id(),
1781 }));
1782 }
1783
1784 Ok(None)
1785 }
1786
1787 pub(in crate::db) fn entity_has_relation_to_target(
1789 &self,
1790 source_entity: EntityTag,
1791 target_path: &str,
1792 ) -> Result<bool, InternalError> {
1793 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1794 return Ok(false);
1795 };
1796 let Some(snapshot) = bundle.entity_snapshots().get(&source_entity) else {
1797 return Ok(false);
1798 };
1799
1800 Ok(snapshot
1801 .relations()
1802 .iter()
1803 .any(|relation| relation.target_path() == target_path))
1804 }
1805
1806 pub(in crate::db) fn validate_live_activation_transition(
1808 &self,
1809 candidate: &AcceptedSchemaRevisionBundle,
1810 ) -> Result<(), InternalError> {
1811 let Some(current) = self.current_accepted_schema_bundle()? else {
1812 return Ok(());
1813 };
1814 Self::validate_activation_transition_from(¤t, candidate)
1815 }
1816
1817 pub(in crate::db) fn validate_canonical_activation_transition(
1819 &self,
1820 candidate: &AcceptedSchemaRevisionBundle,
1821 ) -> Result<(), InternalError> {
1822 let Some(current) = self.current_canonical_accepted_schema_bundle()? else {
1823 return Ok(());
1824 };
1825 Self::validate_activation_transition_from(¤t, candidate)
1826 }
1827
1828 fn validate_activation_transition_from(
1829 current: &AcceptedSchemaRevisionBundle,
1830 candidate: &AcceptedSchemaRevisionBundle,
1831 ) -> Result<(), InternalError> {
1832 for (entity_tag, before) in current.entity_snapshots() {
1833 if before.constraint_activations().is_empty() {
1834 continue;
1835 }
1836 let after = candidate
1837 .entity_snapshots()
1838 .get(entity_tag)
1839 .ok_or_else(InternalError::store_invariant)?;
1840 if before == after {
1841 continue;
1842 }
1843 let expected_shape = before
1844 .clone()
1845 .with_constraint_catalog(after.constraint_catalog().clone());
1846 let catalog_only_transition = expected_shape == *after
1847 && before
1848 .constraint_catalog()
1849 .permits_live_activation_transition_to(after.constraint_catalog());
1850 let sql_row_local_abort_with_version =
1851 before.constraint_activations().iter().any(|activation| {
1852 activation.origin() == ConstraintOrigin::SqlDdl
1853 && matches!(
1854 activation.kind(),
1855 ConstraintActivationKind::Check { .. }
1856 | ConstraintActivationKind::NotNull { .. }
1857 )
1858 && before.version().get().checked_add(1) == Some(after.version().get())
1859 && before
1860 .constraint_catalog()
1861 .clone()
1862 .with_aborted_activation(activation.id())
1863 .is_ok_and(|catalog| catalog == *after.constraint_catalog())
1864 && before
1865 .clone()
1866 .with_constraint_catalog(after.constraint_catalog().clone())
1867 .with_schema_version(after.version())
1868 == *after
1869 });
1870 let sql_unique_abort_with_version =
1871 before.constraint_activations().iter().any(|activation| {
1872 activation.origin() == ConstraintOrigin::SqlDdl
1873 && matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
1874 && before.version().get().checked_add(1) == Some(after.version().get())
1875 && before
1876 .with_aborted_unique_activation(activation.id(), after.version())
1877 .is_ok_and(|expected| expected == *after)
1878 });
1879 let not_null_promotion = before.constraint_activations().iter().any(|activation| {
1880 matches!(activation.kind(), ConstraintActivationKind::NotNull { .. })
1881 && before
1882 .with_promoted_not_null_activation(activation.id(), after.version())
1883 .is_ok_and(|expected| expected == *after)
1884 });
1885 let unique_promotion = before.constraint_activations().iter().any(|activation| {
1886 matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
1887 && before
1888 .with_promoted_unique_activation(activation.id(), after.version())
1889 .is_ok_and(|expected| expected == *after)
1890 });
1891 let relation_promotion = before.constraint_activations().iter().any(|activation| {
1892 matches!(activation.kind(), ConstraintActivationKind::Relation { .. })
1893 && before
1894 .with_promoted_relation_activation(activation.id(), after.version())
1895 .is_ok_and(|expected| expected == *after)
1896 });
1897 if !catalog_only_transition
1898 && !sql_row_local_abort_with_version
1899 && !sql_unique_abort_with_version
1900 && !not_null_promotion
1901 && !unique_promotion
1902 && !relation_promotion
1903 {
1904 return Err(InternalError::store_invariant());
1905 }
1906 }
1907 Ok(())
1908 }
1909
1910 pub(in crate::db) fn validate_constraint_validation_job_closure(
1912 &self,
1913 bundle: &AcceptedSchemaRevisionBundle,
1914 ) -> Result<(), InternalError> {
1915 self.validate_constraint_validation_job_closure_with_change(bundle, None, None)
1916 }
1917
1918 pub(in crate::db) fn validate_constraint_validation_job_closure_with_change(
1921 &self,
1922 bundle: &AcceptedSchemaRevisionBundle,
1923 replacement: Option<&ConstraintValidationJob>,
1924 removal: Option<(EntityTag, ConstraintId)>,
1925 ) -> Result<(), InternalError> {
1926 self.validate_constraint_validation_job_closure_with_change_in_view(
1927 bundle,
1928 replacement,
1929 removal,
1930 IdentityStateStorageView::Effective,
1931 )
1932 }
1933
1934 pub(in crate::db) fn validate_canonical_constraint_validation_job_closure_with_change(
1936 &self,
1937 bundle: &AcceptedSchemaRevisionBundle,
1938 replacement: Option<&ConstraintValidationJob>,
1939 removal: Option<(EntityTag, ConstraintId)>,
1940 ) -> Result<(), InternalError> {
1941 self.validate_constraint_validation_job_closure_with_change_in_view(
1942 bundle,
1943 replacement,
1944 removal,
1945 IdentityStateStorageView::Canonical,
1946 )
1947 }
1948
1949 fn validate_constraint_validation_job_closure_with_change_in_view(
1950 &self,
1951 bundle: &AcceptedSchemaRevisionBundle,
1952 replacement: Option<&ConstraintValidationJob>,
1953 removal: Option<(EntityTag, ConstraintId)>,
1954 view: IdentityStateStorageView,
1955 ) -> Result<(), InternalError> {
1956 if replacement.is_some() && removal.is_some() {
1957 return Err(InternalError::store_invariant());
1958 }
1959 let replacement_key = replacement.map(|job| {
1960 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id())
1961 });
1962 let removal_key = removal.map(|(entity_tag, constraint_id)| {
1963 RawSchemaKey::from_constraint_validation_job(entity_tag, constraint_id)
1964 });
1965 let mut expected = BTreeSet::new();
1966 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1967 for activation in snapshot.constraint_activations() {
1968 let key =
1969 RawSchemaKey::from_constraint_validation_job(*entity_tag, activation.id());
1970 match activation.state() {
1971 ConstraintActivationState::EnforcingNewWrites => {
1972 if self
1973 .constraint_validation_job_after_change(
1974 key,
1975 replacement,
1976 replacement_key,
1977 removal_key,
1978 view,
1979 )?
1980 .is_some()
1981 {
1982 return Err(InternalError::store_corruption());
1983 }
1984 }
1985 ConstraintActivationState::Validating => {
1986 let job = self
1987 .constraint_validation_job_after_change(
1988 key,
1989 replacement,
1990 replacement_key,
1991 removal_key,
1992 view,
1993 )?
1994 .ok_or_else(InternalError::store_corruption)?;
1995 if job.entity_tag() != *entity_tag
1996 || job.entity_path() != snapshot.entity_path()
1997 {
1998 return Err(InternalError::store_corruption());
1999 }
2000 job.validate(Some(activation))?;
2001 expected.insert(key);
2002 }
2003 }
2004 }
2005 }
2006
2007 self.visit_constraint_validation_jobs_in_view(view, |key, raw| {
2008 if removal_key == Some(*key) || replacement_key == Some(*key) {
2009 return Ok(SchemaStoreVisit::Continue);
2010 }
2011 if !expected.contains(key) {
2012 return Err(InternalError::store_corruption());
2013 }
2014 let job = decode_constraint_validation_job(raw.as_bytes())?;
2015 if job.entity_tag() != key.entity_tag()
2016 || key.constraint_id() != Some(job.constraint_id())
2017 {
2018 return Err(InternalError::store_corruption());
2019 }
2020 Ok(SchemaStoreVisit::Continue)
2021 })?;
2022
2023 if let Some(key) = replacement_key
2024 && !expected.contains(&key)
2025 {
2026 return Err(InternalError::store_corruption());
2027 }
2028 if let Some(key) = removal_key
2029 && expected.contains(&key)
2030 {
2031 return Err(InternalError::store_corruption());
2032 }
2033
2034 Ok(())
2035 }
2036
2037 fn constraint_validation_job_after_change(
2038 &self,
2039 key: RawSchemaKey,
2040 replacement: Option<&ConstraintValidationJob>,
2041 replacement_key: Option<RawSchemaKey>,
2042 removal_key: Option<RawSchemaKey>,
2043 view: IdentityStateStorageView,
2044 ) -> Result<Option<ConstraintValidationJob>, InternalError> {
2045 if removal_key == Some(key) {
2046 return Ok(None);
2047 }
2048 if replacement_key == Some(key) {
2049 return Ok(replacement.cloned());
2050 }
2051 let raw = match view {
2052 IdentityStateStorageView::Effective => self.get_raw_snapshot(&key),
2053 IdentityStateStorageView::Canonical => self.get_canonical_raw_value(&key)?,
2054 };
2055 raw.map(|raw| decode_constraint_validation_job(raw.as_bytes()))
2056 .transpose()
2057 }
2058
2059 pub(in crate::db) fn current_accepted_schema_authority_matches(
2062 &self,
2063 expected: &AcceptedSchemaAuthority,
2064 ) -> Result<bool, InternalError> {
2065 let Some(store_scope) = self.accepted_catalog_scope.get() else {
2066 return Ok(false);
2067 };
2068
2069 if let Some(cached) = self
2072 .accepted_bundle_cache
2073 .try_borrow()
2074 .map_err(|_| InternalError::store_invariant())?
2075 .as_ref()
2076 {
2077 let root = cached.selection.root();
2078 return Ok(expected.matches_store_root(
2079 store_scope,
2080 root.revision(),
2081 root.fingerprint(),
2082 ));
2083 }
2084
2085 let Some(selection) = self.current_accepted_schema_root()? else {
2086 return Ok(false);
2087 };
2088 let root = selection.root();
2089
2090 Ok(expected.matches_store_root(store_scope, root.revision(), root.fingerprint()))
2091 }
2092
2093 pub(in crate::db) fn publish_accepted_schema_candidate(
2099 &mut self,
2100 incarnation: DatabaseIncarnationId,
2101 expected_revision: AcceptedSchemaRevision,
2102 candidate: &CandidateSchemaRevision,
2103 ) -> Result<(), InternalError> {
2104 let identity_transition = self.prepare_identity_state_transition(
2105 incarnation,
2106 candidate,
2107 IdentityStateStorageView::Effective,
2108 )?;
2109 if self.current_root_matches_candidate(candidate)? {
2110 if !identity_transition.is_empty() {
2111 return Err(InternalError::identity_state_corruption());
2112 }
2113 let selection = self
2114 .current_accepted_schema_root()?
2115 .ok_or_else(InternalError::store_corruption)?;
2116 self.retain_durable_candidate_entries(candidate, selection.slot())?;
2117 return Ok(());
2118 }
2119 let first = self.accepted_root_slot_bytes(0)?;
2120 let second = self.accepted_root_slot_bytes(1)?;
2121 prepare_accepted_schema_root_publication(
2122 [first.as_deref(), second.as_deref()],
2123 expected_revision,
2124 candidate,
2125 )
2126 .map_err(map_schema_publication_error)?;
2127
2128 self.insert_durable_candidate_snapshots(candidate)?;
2129 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2130 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
2131 let persisted_bundle = self
2132 .get_raw_snapshot(&bundle_key)
2133 .ok_or_else(InternalError::store_corruption)?;
2134 let _verified = decode_verified_accepted_schema_revision_bundle(
2135 candidate.root(),
2136 persisted_bundle.as_bytes(),
2137 )?;
2138 self.apply_identity_state_transition(
2139 identity_transition,
2140 IdentityStateWriteTarget::Durable,
2141 )?;
2142
2143 let first = self.accepted_root_slot_bytes(0)?;
2146 let second = self.accepted_root_slot_bytes(1)?;
2147 let publication = prepare_accepted_schema_root_publication(
2148 [first.as_deref(), second.as_deref()],
2149 expected_revision,
2150 candidate,
2151 )
2152 .map_err(map_schema_publication_error)?;
2153 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
2154 self.insert_durable_raw_value(root_key, publication.encoded_root().to_vec());
2155
2156 let selected = self
2157 .current_accepted_schema_root()?
2158 .ok_or_else(InternalError::store_corruption)?;
2159 if selected.root() != candidate.root() {
2160 return Err(InternalError::store_corruption());
2161 }
2162 self.retain_durable_candidate_entries(candidate, selected.slot())?;
2163 Ok(())
2164 }
2165
2166 pub(in crate::db) fn restore_live_accepted_schema_checkpoint(
2169 &mut self,
2170 incarnation: DatabaseIncarnationId,
2171 candidate: &CandidateSchemaRevision,
2172 checkpoint_identity_states: &IdentityStateInventory,
2173 ) -> Result<(), InternalError> {
2174 if !matches!(self.backend, SchemaStoreBackend::Heap(_)) {
2175 return Err(InternalError::store_invariant());
2176 }
2177 let checkpoint_validation = prepare_identity_state_transition(
2178 incarnation,
2179 Some(candidate.bundle()),
2180 candidate.bundle(),
2181 checkpoint_identity_states.clone(),
2182 )?;
2183 if !checkpoint_validation.is_empty() {
2184 return Err(InternalError::identity_state_corruption());
2185 }
2186 if self.current_root_matches_candidate(candidate)? {
2187 for state in checkpoint_identity_states.values() {
2188 let key = RawSchemaKey::from_identity_state(
2189 state.owner().entity_tag(),
2190 state.owner().field_id(),
2191 );
2192 self.insert_durable_raw_value(key, encode_identity_state(state)?);
2193 }
2194 if self.identity_state_inventory(IdentityStateStorageView::Effective)?
2195 != *checkpoint_identity_states
2196 {
2197 return Err(InternalError::identity_state_corruption());
2198 }
2199 let selection = self
2200 .current_accepted_schema_root()?
2201 .ok_or_else(InternalError::store_corruption)?;
2202 self.retain_durable_candidate_entries(candidate, selection.slot())?;
2203 return Ok(());
2204 }
2205 if self.current_accepted_schema_root()?.is_some()
2206 || !self
2207 .identity_state_inventory(IdentityStateStorageView::Effective)?
2208 .is_empty()
2209 {
2210 return Err(InternalError::store_corruption());
2211 }
2212
2213 self.insert_durable_candidate_snapshots(candidate)?;
2214 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2215 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
2216 for state in checkpoint_identity_states.values() {
2217 let key = RawSchemaKey::from_identity_state(
2218 state.owner().entity_tag(),
2219 state.owner().field_id(),
2220 );
2221 self.insert_durable_raw_value(key, encode_identity_state(state)?);
2222 }
2223 let root_key = RawSchemaKey::from_accepted_root_slot(0)?;
2224 self.insert_durable_raw_value(root_key, candidate.encoded_root().to_vec());
2225
2226 let selected = self
2227 .current_accepted_schema_root()?
2228 .ok_or_else(InternalError::store_corruption)?;
2229 if selected.root() != candidate.root() {
2230 return Err(InternalError::store_corruption());
2231 }
2232 self.retain_durable_candidate_entries(candidate, selected.slot())?;
2233 Ok(())
2234 }
2235
2236 pub(in crate::db) fn preflight_accepted_schema_candidate(
2243 &self,
2244 incarnation: DatabaseIncarnationId,
2245 expected_revision: AcceptedSchemaRevision,
2246 candidate: &CandidateSchemaRevision,
2247 ) -> Result<bool, InternalError> {
2248 let identity_transition = self.prepare_identity_state_transition(
2249 incarnation,
2250 candidate,
2251 IdentityStateStorageView::Effective,
2252 )?;
2253 if self.current_root_matches_candidate(candidate)? {
2254 if !identity_transition.is_empty() {
2255 return Err(InternalError::identity_state_corruption());
2256 }
2257 return Ok(true);
2258 }
2259 let first = self.accepted_root_slot_bytes(0)?;
2260 let second = self.accepted_root_slot_bytes(1)?;
2261 prepare_accepted_schema_root_publication(
2262 [first.as_deref(), second.as_deref()],
2263 expected_revision,
2264 candidate,
2265 )
2266 .map_err(map_schema_publication_error)?;
2267
2268 Ok(false)
2269 }
2270
2271 pub(in crate::db) fn preflight_fold_journaled_accepted_schema_candidate(
2273 &self,
2274 incarnation: DatabaseIncarnationId,
2275 expected_revision: AcceptedSchemaRevision,
2276 candidate: &CandidateSchemaRevision,
2277 ) -> Result<(), InternalError> {
2278 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
2279 return Err(InternalError::store_invariant());
2280 }
2281 let identity_transition = self.prepare_identity_state_transition(
2282 incarnation,
2283 candidate,
2284 IdentityStateStorageView::Canonical,
2285 )?;
2286 let candidate_is_current = self.canonical_root_matches_candidate(candidate)?;
2287 if candidate_is_current && !identity_transition.is_empty() {
2288 return Err(InternalError::identity_state_corruption());
2289 }
2290 for state in identity_transition.into_updates() {
2291 let _encoded = encode_identity_state(&state)?;
2292 }
2293 for snapshot in candidate.bundle().entity_snapshots().values() {
2294 let _encoded = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2295 }
2296
2297 let first = self.canonical_root_slot_bytes(0)?;
2298 let second = self.canonical_root_slot_bytes(1)?;
2299 let root_slot = if candidate_is_current {
2300 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2301 .ok_or_else(InternalError::store_corruption)?
2302 .slot()
2303 } else {
2304 prepare_accepted_schema_root_publication(
2305 [first.as_deref(), second.as_deref()],
2306 expected_revision,
2307 candidate,
2308 )
2309 .map_err(map_schema_publication_error)?
2310 .target_slot()
2311 };
2312 let _retained = Self::candidate_entry_keys(candidate, root_slot)?;
2313 Ok(())
2314 }
2315
2316 pub(in crate::db) fn projected_identity_state_count(
2318 &self,
2319 incarnation: DatabaseIncarnationId,
2320 candidate: &CandidateSchemaRevision,
2321 ) -> Result<usize, InternalError> {
2322 Ok(self
2323 .prepare_identity_state_transition(
2324 incarnation,
2325 candidate,
2326 IdentityStateStorageView::Effective,
2327 )?
2328 .projected_inventory_len())
2329 }
2330
2331 pub(in crate::db) fn apply_journaled_accepted_schema_candidate(
2333 &mut self,
2334 incarnation: DatabaseIncarnationId,
2335 expected_revision: AcceptedSchemaRevision,
2336 candidate: &CandidateSchemaRevision,
2337 ) -> Result<(), InternalError> {
2338 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
2339 return Err(InternalError::store_invariant());
2340 }
2341 let identity_transition = self.prepare_identity_state_transition(
2342 incarnation,
2343 candidate,
2344 IdentityStateStorageView::Effective,
2345 )?;
2346 if self.current_root_matches_candidate(candidate)? {
2347 if !identity_transition.is_empty() {
2348 return Err(InternalError::identity_state_corruption());
2349 }
2350 let selection = self
2351 .current_accepted_schema_root()?
2352 .ok_or_else(InternalError::store_corruption)?;
2353 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
2354 return Ok(());
2355 }
2356
2357 let first = self.accepted_root_slot_bytes(0)?;
2358 let second = self.accepted_root_slot_bytes(1)?;
2359 prepare_accepted_schema_root_publication(
2360 [first.as_deref(), second.as_deref()],
2361 expected_revision,
2362 candidate,
2363 )
2364 .map_err(map_schema_publication_error)?;
2365
2366 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2367 self.insert_persisted_snapshot(*entity_tag, snapshot)?;
2368 }
2369 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2370 self.insert_raw_snapshot(
2371 bundle_key,
2372 RawSchemaSnapshot::from_encoded_control_record(candidate.encoded_bundle().to_vec()),
2373 );
2374 let persisted_bundle = self
2375 .get_raw_snapshot(&bundle_key)
2376 .ok_or_else(InternalError::store_corruption)?;
2377 let _verified = decode_verified_accepted_schema_revision_bundle(
2378 candidate.root(),
2379 persisted_bundle.as_bytes(),
2380 )?;
2381 self.apply_identity_state_transition(
2382 identity_transition,
2383 IdentityStateWriteTarget::Materialized,
2384 )?;
2385
2386 let first = self.accepted_root_slot_bytes(0)?;
2387 let second = self.accepted_root_slot_bytes(1)?;
2388 let publication = prepare_accepted_schema_root_publication(
2389 [first.as_deref(), second.as_deref()],
2390 expected_revision,
2391 candidate,
2392 )
2393 .map_err(map_schema_publication_error)?;
2394 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
2395 self.insert_raw_snapshot(
2396 root_key,
2397 RawSchemaSnapshot::from_encoded_control_record(publication.encoded_root().to_vec()),
2398 );
2399
2400 if !self.current_root_matches_candidate(candidate)? {
2401 return Err(InternalError::store_corruption());
2402 }
2403 let selection = self
2404 .current_accepted_schema_root()?
2405 .ok_or_else(InternalError::store_corruption)?;
2406 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
2407 Ok(())
2408 }
2409
2410 pub(in crate::db) fn fold_journaled_accepted_schema_candidate(
2412 &mut self,
2413 incarnation: DatabaseIncarnationId,
2414 expected_revision: AcceptedSchemaRevision,
2415 candidate: &CandidateSchemaRevision,
2416 ) -> Result<(), InternalError> {
2417 let identity_transition = self.prepare_identity_state_transition(
2418 incarnation,
2419 candidate,
2420 IdentityStateStorageView::Canonical,
2421 )?;
2422 if self.canonical_root_matches_candidate(candidate)? {
2423 if !identity_transition.is_empty() {
2424 return Err(InternalError::identity_state_corruption());
2425 }
2426 let first = self.canonical_root_slot_bytes(0)?;
2427 let second = self.canonical_root_slot_bytes(1)?;
2428 let selection =
2429 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2430 .ok_or_else(InternalError::store_corruption)?;
2431 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
2432 return Ok(());
2433 }
2434
2435 let first = self.canonical_root_slot_bytes(0)?;
2436 let second = self.canonical_root_slot_bytes(1)?;
2437 prepare_accepted_schema_root_publication(
2438 [first.as_deref(), second.as_deref()],
2439 expected_revision,
2440 candidate,
2441 )
2442 .map_err(map_schema_publication_error)?;
2443
2444 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2445 self.fold_persisted_snapshot(*entity_tag, snapshot)?;
2446 }
2447 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2448 self.insert_canonical_raw_value(bundle_key, candidate.encoded_bundle().to_vec())?;
2449 let persisted_bundle = self
2450 .get_canonical_raw_value(&bundle_key)?
2451 .ok_or_else(InternalError::store_corruption)?;
2452 let _verified = decode_verified_accepted_schema_revision_bundle(
2453 candidate.root(),
2454 persisted_bundle.as_bytes(),
2455 )?;
2456 self.apply_identity_state_transition(
2457 identity_transition,
2458 IdentityStateWriteTarget::Canonical,
2459 )?;
2460
2461 let first = self.canonical_root_slot_bytes(0)?;
2462 let second = self.canonical_root_slot_bytes(1)?;
2463 let publication = prepare_accepted_schema_root_publication(
2464 [first.as_deref(), second.as_deref()],
2465 expected_revision,
2466 candidate,
2467 )
2468 .map_err(map_schema_publication_error)?;
2469 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
2470 self.insert_canonical_raw_value(root_key, publication.encoded_root().to_vec())?;
2471
2472 if !self.canonical_root_matches_candidate(candidate)? {
2473 return Err(InternalError::store_corruption());
2474 }
2475 let first = self.canonical_root_slot_bytes(0)?;
2476 let second = self.canonical_root_slot_bytes(1)?;
2477 let selection = select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2478 .ok_or_else(InternalError::store_corruption)?;
2479 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
2480 Ok(())
2481 }
2482
2483 pub(in crate::db) fn get_persisted_snapshot(
2485 &self,
2486 entity: EntityTag,
2487 version: SchemaVersion,
2488 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2489 let key = RawSchemaKey::from_entity_version(entity, version);
2490 self.get_raw_snapshot(&key)
2491 .map(|snapshot| snapshot.decode_persisted_snapshot())
2492 .transpose()
2493 }
2494
2495 #[cfg(test)]
2496 fn latest_staged_persisted_snapshot(
2497 &self,
2498 entity: EntityTag,
2499 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2500 self.latest_raw_snapshots_by_entity()
2501 .remove(&entity)
2502 .map(|(_, snapshot)| snapshot.decode_persisted_snapshot())
2503 .transpose()
2504 }
2505
2506 pub(in crate::db) fn current_accepted_persisted_snapshot(
2509 &self,
2510 entity: EntityTag,
2511 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
2512 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2513 return Ok(None);
2514 };
2515
2516 Ok(bundle.entity_snapshots().get(&entity).cloned())
2517 }
2518
2519 pub(in crate::db) fn current_accepted_catalog_selection(
2521 &self,
2522 entity: EntityTag,
2523 entity_path: &str,
2524 store_path: &'static str,
2525 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
2526 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2527 return Ok(None);
2528 };
2529 if bundle.store_path() != store_path {
2530 return Err(InternalError::store_corruption());
2531 }
2532 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
2533 return Ok(None);
2534 };
2535 if snapshot.entity_path() != entity_path {
2536 return Err(InternalError::store_corruption());
2537 }
2538
2539 let cache = self
2540 .accepted_bundle_cache
2541 .try_borrow()
2542 .map_err(|_| InternalError::store_invariant())?;
2543 let cached = cache.as_ref().ok_or_else(InternalError::store_invariant)?;
2544 if let Some(selection) = cached
2545 .entity_selections
2546 .try_borrow()
2547 .map_err(|_| InternalError::store_invariant())?
2548 .get(&entity)
2549 .cloned()
2550 {
2551 return Ok(Some(selection));
2552 }
2553
2554 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2555 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
2556 let identity = AcceptedCatalogIdentity::new(
2557 entity,
2558 entity_path,
2559 store_path,
2560 bundle.revision(),
2561 snapshot.version(),
2562 fingerprint,
2563 );
2564
2565 let selected = AcceptedCatalogSnapshotSelection::new(
2566 identity,
2567 cached.value_catalog.clone(),
2568 Rc::from(raw_snapshot.into_bytes()),
2569 );
2570 cached
2571 .entity_selections
2572 .try_borrow_mut()
2573 .map_err(|_| InternalError::store_invariant())?
2574 .insert(entity, selected.clone());
2575
2576 Ok(Some(selected))
2577 }
2578
2579 pub(in crate::db) fn current_canonical_accepted_catalog_selection(
2583 &self,
2584 entity: EntityTag,
2585 entity_path: &str,
2586 store_path: &'static str,
2587 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
2588 let first = self.canonical_root_slot_bytes(0)?;
2589 let second = self.canonical_root_slot_bytes(1)?;
2590 let Some(selection) =
2591 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2592 else {
2593 return Ok(None);
2594 };
2595 let bundle_key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
2596 let raw_bundle = self
2597 .get_canonical_raw_value(&bundle_key)?
2598 .ok_or_else(InternalError::store_corruption)?;
2599 let bundle = decode_verified_accepted_schema_revision_bundle(
2600 selection.root(),
2601 raw_bundle.as_bytes(),
2602 )?;
2603 if bundle.store_path() != store_path {
2604 return Err(InternalError::store_corruption());
2605 }
2606 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
2607 return Ok(None);
2608 };
2609 if snapshot.entity_path() != entity_path {
2610 return Err(InternalError::store_corruption());
2611 }
2612
2613 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2614 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
2615 let identity = AcceptedCatalogIdentity::new(
2616 entity,
2617 entity_path,
2618 store_path,
2619 bundle.revision(),
2620 snapshot.version(),
2621 fingerprint,
2622 );
2623
2624 Ok(Some(AcceptedCatalogSnapshotSelection::new(
2625 identity,
2626 AcceptedValueCatalogHandle::new(
2627 bundle.enum_catalog().clone(),
2628 bundle.composite_catalog().clone(),
2629 self.accepted_catalog_scope
2630 .get_or_init(AcceptedStoreCatalogScope::new)
2631 .clone(),
2632 bundle.revision(),
2633 selection.root().fingerprint(),
2634 ),
2635 Rc::from(raw_snapshot.into_bytes()),
2636 )))
2637 }
2638
2639 #[cfg(test)]
2645 pub(in crate::db) fn catalog_metadata(
2646 &self,
2647 ) -> Result<Option<SchemaStoreCatalogMetadata>, InternalError> {
2648 Ok(self
2649 .allocation_metadata()?
2650 .map(SchemaStoreAllocationMetadata::schema))
2651 }
2652
2653 pub(in crate::db) fn allocation_metadata(
2660 &self,
2661 ) -> Result<Option<SchemaStoreAllocationMetadata>, InternalError> {
2662 let latest_by_entity = self.latest_raw_snapshots_by_entity();
2663 if latest_by_entity.is_empty() {
2664 return Ok(None);
2665 }
2666
2667 Ok(Some(SchemaStoreAllocationMetadata::new(
2668 derive_data_allocation_metadata(&latest_by_entity)?,
2669 derive_index_allocation_metadata(&latest_by_entity)?,
2670 derive_schema_catalog_metadata(&latest_by_entity)?,
2671 )))
2672 }
2673
2674 fn insert_raw_snapshot(
2676 &mut self,
2677 key: RawSchemaKey,
2678 snapshot: RawSchemaSnapshot,
2679 ) -> Option<RawSchemaSnapshot> {
2680 self.invalidate_accepted_bundle_cache_for_key(key);
2681 let previous_journaled = if matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
2682 self.get_raw_snapshot_for_backend(&key)
2683 } else {
2684 None
2685 };
2686 match &mut self.backend {
2687 SchemaStoreBackend::Heap(map) => map.insert(key, snapshot),
2688 SchemaStoreBackend::Journaled {
2689 live, tombstones, ..
2690 } => {
2691 tombstones.remove(&key);
2692 live.insert(key, snapshot);
2693 previous_journaled
2694 }
2695 }
2696 }
2697
2698 #[must_use]
2700 fn get_raw_snapshot(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
2701 match &self.backend {
2702 SchemaStoreBackend::Heap(map) => map.get(key).cloned(),
2703 SchemaStoreBackend::Journaled { .. } => self.get_raw_snapshot_for_backend(key),
2704 }
2705 }
2706
2707 fn accepted_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
2708 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
2709 Ok(self
2710 .get_raw_snapshot(&key)
2711 .map(RawSchemaSnapshot::into_bytes))
2712 }
2713
2714 fn canonical_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
2715 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
2716 Ok(self
2717 .get_canonical_raw_value(&key)?
2718 .map(RawSchemaSnapshot::into_bytes))
2719 }
2720
2721 fn current_root_matches_candidate(
2722 &self,
2723 candidate: &CandidateSchemaRevision,
2724 ) -> Result<bool, InternalError> {
2725 let Some(selection) = self.current_accepted_schema_root()? else {
2726 return Ok(false);
2727 };
2728 if selection.root() != candidate.root() {
2729 return Ok(false);
2730 }
2731 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2732 let bundle = self
2733 .get_raw_snapshot(&key)
2734 .ok_or_else(InternalError::store_corruption)?;
2735 let _verified =
2736 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
2737 Ok(true)
2738 }
2739
2740 fn canonical_root_matches_candidate(
2741 &self,
2742 candidate: &CandidateSchemaRevision,
2743 ) -> Result<bool, InternalError> {
2744 let first = self.canonical_root_slot_bytes(0)?;
2745 let second = self.canonical_root_slot_bytes(1)?;
2746 let Some(selection) =
2747 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2748 else {
2749 return Ok(false);
2750 };
2751 if selection.root() != candidate.root() {
2752 return Ok(false);
2753 }
2754 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2755 let bundle = self
2756 .get_canonical_raw_value(&key)?
2757 .ok_or_else(InternalError::store_corruption)?;
2758 let _verified =
2759 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
2760 Ok(true)
2761 }
2762
2763 fn get_canonical_raw_value(
2764 &self,
2765 key: &RawSchemaKey,
2766 ) -> Result<Option<RawSchemaSnapshot>, InternalError> {
2767 match &self.backend {
2768 SchemaStoreBackend::Journaled { canonical, .. } => Ok(canonical.get(key)),
2769 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
2770 }
2771 }
2772
2773 fn insert_canonical_raw_value(
2774 &mut self,
2775 key: RawSchemaKey,
2776 bytes: Vec<u8>,
2777 ) -> Result<(), InternalError> {
2778 self.invalidate_accepted_bundle_cache_for_key(key);
2779 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
2780 return Err(InternalError::store_invariant());
2781 };
2782 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
2783 Ok(())
2784 }
2785
2786 fn insert_durable_raw_value(&mut self, key: RawSchemaKey, bytes: Vec<u8>) {
2790 self.invalidate_accepted_bundle_cache_for_key(key);
2791 let value = RawSchemaSnapshot::from_encoded_control_record(bytes);
2792 match &mut self.backend {
2793 SchemaStoreBackend::Heap(map) => {
2794 map.insert(key, value);
2795 }
2796 SchemaStoreBackend::Journaled {
2797 canonical,
2798 live,
2799 tombstones,
2800 ..
2801 } => {
2802 live.remove(&key);
2803 tombstones.remove(&key);
2804 canonical.insert(key, value);
2805 }
2806 }
2807 }
2808
2809 fn invalidate_accepted_bundle_cache_for_key(&mut self, key: RawSchemaKey) {
2810 if key.is_accepted_root() {
2811 self.accepted_bundle_cache.get_mut().take();
2812 }
2813 }
2814
2815 fn insert_durable_candidate_snapshots(
2816 &mut self,
2817 candidate: &CandidateSchemaRevision,
2818 ) -> Result<(), InternalError> {
2819 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2820 let key = RawSchemaKey::from_entity_version(*entity_tag, snapshot.version());
2821 let value = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2822 match &mut self.backend {
2823 SchemaStoreBackend::Heap(map) => {
2824 map.insert(key, value);
2825 }
2826 SchemaStoreBackend::Journaled {
2827 canonical,
2828 live,
2829 tombstones,
2830 ..
2831 } => {
2832 live.remove(&key);
2833 tombstones.remove(&key);
2834 canonical.insert(key, value);
2835 }
2836 }
2837 }
2838 Ok(())
2839 }
2840
2841 fn candidate_entry_keys(
2842 candidate: &CandidateSchemaRevision,
2843 root_slot: usize,
2844 ) -> Result<BTreeSet<RawSchemaKey>, InternalError> {
2845 let mut keys = candidate
2846 .bundle()
2847 .entity_snapshots()
2848 .iter()
2849 .map(|(entity_tag, snapshot)| {
2850 RawSchemaKey::from_entity_version(*entity_tag, snapshot.version())
2851 })
2852 .collect::<BTreeSet<_>>();
2853 keys.insert(RawSchemaKey::from_accepted_bundle(
2854 candidate.root().bundle_key(),
2855 ));
2856 keys.insert(RawSchemaKey::from_accepted_root_slot(root_slot)?);
2857 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2858 for activation in snapshot
2859 .constraint_activations()
2860 .iter()
2861 .filter(|activation| activation.state() == ConstraintActivationState::Validating)
2862 {
2863 keys.insert(RawSchemaKey::from_constraint_validation_job(
2864 *entity_tag,
2865 activation.id(),
2866 ));
2867 }
2868 }
2869 Ok(keys)
2870 }
2871
2872 fn positioned_candidate_effect_keys(
2873 &self,
2874 incarnation: DatabaseIncarnationId,
2875 expected_revision: AcceptedSchemaRevision,
2876 candidate: &CandidateSchemaRevision,
2877 view: IdentityStateStorageView,
2878 ) -> Result<BTreeSet<RawSchemaKey>, InternalError> {
2879 let identity_transition =
2880 self.prepare_identity_state_transition(incarnation, candidate, view)?;
2881 let (first, second, candidate_is_current) = match view {
2882 IdentityStateStorageView::Effective => (
2883 self.accepted_root_slot_bytes(0)?,
2884 self.accepted_root_slot_bytes(1)?,
2885 self.current_root_matches_candidate(candidate)?,
2886 ),
2887 IdentityStateStorageView::Canonical => (
2888 self.canonical_root_slot_bytes(0)?,
2889 self.canonical_root_slot_bytes(1)?,
2890 self.canonical_root_matches_candidate(candidate)?,
2891 ),
2892 };
2893 let root_slot = if candidate_is_current {
2894 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2895 .ok_or_else(InternalError::store_corruption)?
2896 .slot()
2897 } else {
2898 prepare_accepted_schema_root_publication(
2899 [first.as_deref(), second.as_deref()],
2900 expected_revision,
2901 candidate,
2902 )
2903 .map_err(map_schema_publication_error)?
2904 .target_slot()
2905 };
2906 let mut keys = Self::candidate_entry_keys(candidate, root_slot)?;
2907 for state in identity_transition.into_updates() {
2908 keys.insert(RawSchemaKey::from_identity_state(
2909 state.owner().entity_tag(),
2910 state.owner().field_id(),
2911 ));
2912 }
2913
2914 let SchemaStoreBackend::Journaled {
2915 canonical, live, ..
2916 } = &self.backend
2917 else {
2918 return Err(InternalError::store_invariant());
2919 };
2920 for entry in canonical.iter() {
2921 if !keys.contains(entry.key()) && !entry.key().is_identity_state() {
2922 keys.insert(*entry.key());
2923 }
2924 }
2925 if matches!(view, IdentityStateStorageView::Effective) {
2926 for key in live.keys() {
2927 if !keys.contains(key) && !key.is_identity_state() {
2928 keys.insert(*key);
2929 }
2930 }
2931 }
2932 Ok(keys)
2933 }
2934
2935 fn positioned_journal_batch_keys(
2936 &self,
2937 incarnation: DatabaseIncarnationId,
2938 batch: &JournalBatch,
2939 view: IdentityStateStorageView,
2940 ) -> Result<BTreeSet<RawSchemaKey>, InternalError> {
2941 let mut keys = BTreeSet::new();
2942 for record in batch.records() {
2943 match record {
2944 JournalRecord::SchemaPut {
2945 schema_snapshot_bytes,
2946 ..
2947 } => {
2948 let snapshot = decode_persisted_schema_snapshot(schema_snapshot_bytes)?;
2949 let entity_tag = match view {
2950 IdentityStateStorageView::Effective => self
2951 .current_accepted_schema_bundle_ref()?
2952 .ok_or_else(InternalError::store_corruption)?
2953 .entity_snapshots()
2954 .iter()
2955 .find_map(|(entity_tag, accepted)| {
2956 (accepted.entity_path() == snapshot.entity_path())
2957 .then_some(*entity_tag)
2958 }),
2959 IdentityStateStorageView::Canonical => self
2960 .current_canonical_accepted_schema_bundle()?
2961 .ok_or_else(InternalError::store_corruption)?
2962 .entity_snapshots()
2963 .iter()
2964 .find_map(|(entity_tag, accepted)| {
2965 (accepted.entity_path() == snapshot.entity_path())
2966 .then_some(*entity_tag)
2967 }),
2968 }
2969 .ok_or_else(InternalError::store_corruption)?;
2970 keys.insert(RawSchemaKey::from_entity_version(
2971 entity_tag,
2972 snapshot.version(),
2973 ));
2974 }
2975 JournalRecord::AcceptedSchemaPublish {
2976 expected_revision,
2977 schema_bundle_bytes,
2978 schema_root_bytes,
2979 ..
2980 } => {
2981 let candidate = CandidateSchemaRevision::from_encoded(
2982 schema_bundle_bytes.clone(),
2983 schema_root_bytes.clone(),
2984 )?;
2985 keys.extend(self.positioned_candidate_effect_keys(
2986 incarnation,
2987 *expected_revision,
2988 &candidate,
2989 view,
2990 )?);
2991 }
2992 JournalRecord::ConstraintValidationJobPut {
2993 entity_tag,
2994 constraint_id,
2995 ..
2996 }
2997 | JournalRecord::ConstraintValidationJobDelete {
2998 entity_tag,
2999 constraint_id,
3000 ..
3001 } => {
3002 keys.insert(RawSchemaKey::from_constraint_validation_job(
3003 *entity_tag,
3004 *constraint_id,
3005 ));
3006 }
3007 JournalRecord::IdentityRangeAdvance { range } => {
3008 keys.insert(RawSchemaKey::from_identity_state(
3009 range.owner().entity_tag(),
3010 range.owner().field_id(),
3011 ));
3012 }
3013 JournalRecord::RowPut { .. }
3014 | JournalRecord::RowDelete { .. }
3015 | JournalRecord::AcceptedSchemaIndexDelete { .. }
3016 | JournalRecord::AcceptedSchemaIndexPut { .. }
3017 | JournalRecord::ConstraintValidationIndexPut { .. } => {}
3018 #[cfg(any(test, feature = "migration"))]
3019 JournalRecord::SchemaMigrationRowPut { .. }
3020 | JournalRecord::SchemaMigrationIndexPut { .. } => {}
3021 }
3022 }
3023 Ok(keys)
3024 }
3025
3026 fn retain_durable_candidate_entries(
3030 &mut self,
3031 candidate: &CandidateSchemaRevision,
3032 root_slot: usize,
3033 ) -> Result<(), InternalError> {
3034 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
3035 self.accepted_bundle_cache.get_mut().take();
3036 match &mut self.backend {
3037 SchemaStoreBackend::Heap(map) => {
3038 map.retain(|key, _| keep.contains(key) || key.is_identity_state());
3039 }
3040 SchemaStoreBackend::Journaled {
3041 canonical,
3042 live,
3043 tombstones,
3044 ..
3045 } => {
3046 let stale = canonical
3047 .iter()
3048 .filter_map(|entry| {
3049 (!keep.contains(entry.key()) && !entry.key().is_identity_state())
3050 .then_some(*entry.key())
3051 })
3052 .collect::<Vec<_>>();
3053 for key in stale {
3054 canonical.remove(&key);
3055 }
3056 live.retain(|key, _| keep.contains(key) || key.is_identity_state());
3057 tombstones.clear();
3058 }
3059 }
3060 Ok(())
3061 }
3062
3063 fn retain_materialized_candidate_entries(
3064 &mut self,
3065 candidate: &CandidateSchemaRevision,
3066 root_slot: usize,
3067 ) -> Result<(), InternalError> {
3068 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
3069 self.accepted_bundle_cache.get_mut().take();
3070 let SchemaStoreBackend::Journaled {
3071 canonical,
3072 live,
3073 tombstones,
3074 ..
3075 } = &mut self.backend
3076 else {
3077 return Err(InternalError::store_invariant());
3078 };
3079 live.retain(|key, _| keep.contains(key) || key.is_identity_state());
3080 let canonical_keys = canonical
3081 .iter()
3082 .map(|entry| *entry.key())
3083 .collect::<Vec<_>>();
3084 for key in canonical_keys {
3085 if keep.contains(&key) || key.is_identity_state() {
3086 tombstones.remove(&key);
3087 } else {
3088 tombstones.insert(key);
3089 }
3090 }
3091 Ok(())
3092 }
3093
3094 fn retain_canonical_candidate_entries(
3095 &mut self,
3096 candidate: &CandidateSchemaRevision,
3097 root_slot: usize,
3098 ) -> Result<(), InternalError> {
3099 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
3100 self.accepted_bundle_cache.get_mut().take();
3101 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
3102 return Err(InternalError::store_invariant());
3103 };
3104 let stale = canonical
3105 .iter()
3106 .filter_map(|entry| {
3107 (!keep.contains(entry.key()) && !entry.key().is_identity_state())
3108 .then_some(*entry.key())
3109 })
3110 .collect::<Vec<_>>();
3111 for key in stale {
3112 canonical.remove(&key);
3113 }
3114 Ok(())
3115 }
3116
3117 #[must_use]
3119 #[cfg(test)]
3120 fn contains_raw_snapshot(&self, key: &RawSchemaKey) -> bool {
3121 match &self.backend {
3122 SchemaStoreBackend::Heap(map) => map.contains_key(key),
3123 SchemaStoreBackend::Journaled { .. } => {
3124 self.get_raw_snapshot_for_backend(key).is_some()
3125 }
3126 }
3127 }
3128
3129 #[must_use]
3131 #[cfg(test)]
3132 pub(in crate::db) fn len(&self) -> u64 {
3133 match &self.backend {
3134 SchemaStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
3135 SchemaStoreBackend::Journaled { .. } => {
3136 let mut count = 0_u64;
3137 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
3138 count = count.saturating_add(1);
3139 Ok(SchemaStoreVisit::Continue)
3140 });
3141 count
3142 }
3143 }
3144 }
3145
3146 #[must_use]
3148 #[cfg(test)]
3149 pub(in crate::db) fn is_empty(&self) -> bool {
3150 match &self.backend {
3151 SchemaStoreBackend::Heap(map) => map.is_empty(),
3152 SchemaStoreBackend::Journaled { .. } => {
3153 let mut empty = true;
3154 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
3155 empty = false;
3156 Ok(SchemaStoreVisit::Stop)
3157 });
3158 empty
3159 }
3160 }
3161 }
3162
3163 #[cfg(test)]
3165 pub(in crate::db) fn clear(&mut self) {
3166 self.accepted_bundle_cache.get_mut().take();
3167 match &mut self.backend {
3168 SchemaStoreBackend::Heap(map) => map.clear(),
3169 SchemaStoreBackend::Journaled {
3170 canonical,
3171 live,
3172 tombstones,
3173 ..
3174 } => {
3175 live.clear();
3176 tombstones.clear();
3177 let keys = canonical
3178 .iter()
3179 .map(|entry| *entry.key())
3180 .collect::<Vec<_>>();
3181 for key in keys {
3182 if key.is_entity_snapshot() {
3183 tombstones.insert(key);
3184 } else {
3185 canonical.remove(&key);
3186 }
3187 }
3188 }
3189 }
3190 }
3191
3192 fn current_accepted_schema_bundle_ref(
3193 &self,
3194 ) -> Result<Option<Ref<'_, AcceptedSchemaRevisionBundle>>, InternalError> {
3195 let Some(selection) = self.current_accepted_schema_root()? else {
3196 self.accepted_bundle_cache
3197 .try_borrow_mut()
3198 .map_err(|_| InternalError::store_invariant())?
3199 .take();
3200 return Ok(None);
3201 };
3202
3203 let cache_matches = self
3204 .accepted_bundle_cache
3205 .try_borrow()
3206 .map_err(|_| InternalError::store_invariant())?
3207 .as_ref()
3208 .is_some_and(|cached| cached.selection == selection);
3209 if !cache_matches {
3210 let key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
3211 let raw = self
3212 .get_raw_snapshot(&key)
3213 .ok_or_else(InternalError::store_corruption)?;
3214 let bundle =
3215 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes())?;
3216 self.validate_constraint_validation_job_closure(&bundle)?;
3217 #[cfg(test)]
3218 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES
3219 .with(|misses| misses.set(misses.get().saturating_add(1)));
3220 let value_catalog = AcceptedValueCatalogHandle::new(
3221 bundle.enum_catalog().clone(),
3222 bundle.composite_catalog().clone(),
3223 self.accepted_catalog_scope
3224 .get_or_init(AcceptedStoreCatalogScope::new)
3225 .clone(),
3226 bundle.revision(),
3227 selection.root().fingerprint(),
3228 );
3229 *self
3230 .accepted_bundle_cache
3231 .try_borrow_mut()
3232 .map_err(|_| InternalError::store_invariant())? = Some(AcceptedSchemaBundleCache {
3233 selection,
3234 bundle,
3235 value_catalog,
3236 entity_selections: RefCell::new(StdBTreeMap::new()),
3237 });
3238 }
3239
3240 let cache = self
3241 .accepted_bundle_cache
3242 .try_borrow()
3243 .map_err(|_| InternalError::store_invariant())?;
3244 let bundle = Ref::filter_map(cache, |cache| {
3245 cache
3246 .as_ref()
3247 .filter(|cached| cached.selection == selection)
3248 .map(|cached| &cached.bundle)
3249 })
3250 .map_err(|_| InternalError::store_invariant())?;
3251 self.validate_identity_state_closure(&bundle)?;
3252 Ok(Some(bundle))
3253 }
3254
3255 fn latest_raw_snapshots_by_entity(
3256 &self,
3257 ) -> StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)> {
3258 let mut latest_by_entity =
3259 StdBTreeMap::<EntityTag, (SchemaVersion, RawSchemaSnapshot)>::new();
3260
3261 let _: Result<(), std::convert::Infallible> = self.visit_raw_snapshots(|key, snapshot| {
3262 let version = SchemaVersion::new(key.version());
3263 match latest_by_entity.get_mut(&key.entity_tag()) {
3264 Some((latest_version, latest_snapshot)) if version > *latest_version => {
3265 *latest_version = version;
3266 *latest_snapshot = snapshot.clone();
3267 }
3268 None => {
3269 latest_by_entity.insert(key.entity_tag(), (version, snapshot.clone()));
3270 }
3271 Some(_) => {}
3272 }
3273 Ok(SchemaStoreVisit::Continue)
3274 });
3275
3276 latest_by_entity
3277 }
3278
3279 fn visit_raw_snapshots<E>(
3282 &self,
3283 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
3284 ) -> Result<(), E> {
3285 let bounds = RawSchemaKey::all_entity_range_bounds();
3286 match &self.backend {
3287 SchemaStoreBackend::Heap(map) => {
3288 let mut visitor = visitor;
3289 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
3290 if visitor(key, snapshot)?.should_stop() {
3291 break;
3292 }
3293 }
3294 }
3295 SchemaStoreBackend::Journaled {
3296 canonical,
3297 live,
3298 tombstones,
3299 ..
3300 } => Self::visit_journaled_raw_snapshot_range(
3301 canonical,
3302 live,
3303 tombstones,
3304 bounds,
3305 Direction::Asc,
3306 visitor,
3307 )?,
3308 }
3309
3310 Ok(())
3311 }
3312
3313 fn visit_constraint_validation_jobs_in_view<E>(
3314 &self,
3315 view: IdentityStateStorageView,
3316 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
3317 ) -> Result<(), E> {
3318 let bounds = RawSchemaKey::all_constraint_validation_job_range_bounds();
3319 match (&self.backend, view) {
3320 (SchemaStoreBackend::Heap(map), _) => {
3321 let mut visitor = visitor;
3322 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
3323 if visitor(key, snapshot)?.should_stop() {
3324 break;
3325 }
3326 }
3327 }
3328 (
3329 SchemaStoreBackend::Journaled {
3330 canonical,
3331 live,
3332 tombstones,
3333 ..
3334 },
3335 IdentityStateStorageView::Effective,
3336 ) => Self::visit_journaled_raw_snapshot_range(
3337 canonical,
3338 live,
3339 tombstones,
3340 bounds,
3341 Direction::Asc,
3342 visitor,
3343 )?,
3344 (
3345 SchemaStoreBackend::Journaled { canonical, .. },
3346 IdentityStateStorageView::Canonical,
3347 ) => {
3348 let mut visitor = visitor;
3349 for entry in canonical.range((bounds.0, bounds.1)) {
3350 if visitor(entry.key(), &entry.value())?.should_stop() {
3351 break;
3352 }
3353 }
3354 }
3355 }
3356 Ok(())
3357 }
3358
3359 #[cfg(test)]
3360 #[must_use]
3361 pub(in crate::db) fn canonical_len_for_tests(&self) -> u64 {
3362 match &self.backend {
3363 SchemaStoreBackend::Journaled { canonical: map, .. } => map.len(),
3364 SchemaStoreBackend::Heap(_) => 0,
3365 }
3366 }
3367
3368 fn get_raw_snapshot_for_backend(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
3369 let SchemaStoreBackend::Journaled {
3370 canonical,
3371 live,
3372 tombstones,
3373 ..
3374 } = &self.backend
3375 else {
3376 return None;
3377 };
3378
3379 if tombstones.contains(key) {
3380 return None;
3381 }
3382 live.get(key).cloned().or_else(|| canonical.get(key))
3383 }
3384
3385 fn visit_journaled_raw_snapshot_range<E>(
3386 canonical: &StableBTreeMap<
3387 RawSchemaKey,
3388 RawSchemaSnapshot,
3389 VirtualMemory<DefaultMemoryImpl>,
3390 >,
3391 live: &StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
3392 tombstones: &BTreeSet<RawSchemaKey>,
3393 bounds: (RangeBound<RawSchemaKey>, RangeBound<RawSchemaKey>),
3394 direction: Direction,
3395 mut visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
3396 ) -> Result<(), E> {
3397 match direction {
3398 Direction::Asc => visit_ordered_overlay(
3399 canonical.range((bounds.0, bounds.1)),
3400 live.range((bounds.0, bounds.1)),
3401 Direction::Asc,
3402 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
3403 |canonical_entry| !tombstones.contains(canonical_entry.key()),
3404 |live_entry| !tombstones.contains(live_entry.0),
3405 |entry| {
3406 let visit = match entry {
3407 OrderedOverlayEntry::Canonical(canonical_entry) => {
3408 visitor(canonical_entry.key(), &canonical_entry.value())?
3409 }
3410 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
3411 };
3412 Ok(if visit.should_stop() {
3413 OrderedOverlayVisit::Stop
3414 } else {
3415 OrderedOverlayVisit::Continue
3416 })
3417 },
3418 ),
3419 Direction::Desc => visit_ordered_overlay(
3420 canonical.range((bounds.0, bounds.1)).rev(),
3421 live.range((bounds.0, bounds.1)).rev(),
3422 Direction::Desc,
3423 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
3424 |canonical_entry| !tombstones.contains(canonical_entry.key()),
3425 |live_entry| !tombstones.contains(live_entry.0),
3426 |entry| {
3427 let visit = match entry {
3428 OrderedOverlayEntry::Canonical(canonical_entry) => {
3429 visitor(canonical_entry.key(), &canonical_entry.value())?
3430 }
3431 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
3432 };
3433 Ok(if visit.should_stop() {
3434 OrderedOverlayVisit::Stop
3435 } else {
3436 OrderedOverlayVisit::Continue
3437 })
3438 },
3439 ),
3440 }
3441 }
3442}
3443
3444fn map_schema_publication_error(error: AcceptedSchemaPublicationError) -> InternalError {
3445 match error {
3446 AcceptedSchemaPublicationError::StaleSchemaRevision { .. }
3447 | AcceptedSchemaPublicationError::RevisionExhausted => InternalError::store_unsupported(),
3448 AcceptedSchemaPublicationError::InvalidCandidate => InternalError::store_invariant(),
3449 AcceptedSchemaPublicationError::CorruptRootSlots => InternalError::store_corruption(),
3450 }
3451}
3452
3453fn derive_data_allocation_metadata(
3454 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
3455) -> Result<SchemaStoreCatalogMetadata, InternalError> {
3456 let mut max_version = SchemaVersion::initial();
3457 let mut hasher = new_hash_sha256();
3458 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN);
3459
3460 for (entity, (_, snapshot)) in latest_by_entity {
3461 let persisted = snapshot.decode_persisted_snapshot()?;
3462 if persisted.version() > max_version {
3463 max_version = persisted.version();
3464 }
3465
3466 let data_projection = PersistedSchemaSnapshot::new_with_primary_key_fields_and_indexes(
3467 persisted.version(),
3468 persisted.entity_path().to_string(),
3469 persisted.entity_name().to_string(),
3470 persisted.primary_key_field_ids().to_vec(),
3471 persisted.row_layout().clone(),
3472 persisted.fields().to_vec(),
3473 Vec::new(),
3474 );
3475 let constraint_catalog = crate::db::schema::AcceptedConstraintCatalog::initial(
3476 data_projection.fields(),
3477 data_projection.indexes(),
3478 data_projection.relations(),
3479 )
3480 .map_err(|_| InternalError::store_invariant())?;
3481 let data_projection = data_projection.with_constraint_catalog(constraint_catalog);
3482 let encoded = encode_persisted_schema_snapshot(&data_projection)?;
3483
3484 write_hash_u64(&mut hasher, entity.value());
3485 write_hash_u32(&mut hasher, persisted.version().get());
3486 write_hash_len_u32(&mut hasher, encoded.len());
3487 hasher.update(encoded);
3488 }
3489
3490 Ok(finalize_schema_metadata(
3491 max_version,
3492 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
3493 hasher,
3494 latest_by_entity.len(),
3495 ))
3496}
3497
3498fn derive_index_allocation_metadata(
3499 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
3500) -> Result<SchemaStoreCatalogMetadata, InternalError> {
3501 let mut max_version = SchemaVersion::initial();
3502 let mut hasher = new_hash_sha256();
3503 write_hash_tag_u8(
3504 &mut hasher,
3505 SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN,
3506 );
3507
3508 for (entity, (_, snapshot)) in latest_by_entity {
3509 let persisted = snapshot.decode_persisted_snapshot()?;
3510 if persisted.version() > max_version {
3511 max_version = persisted.version();
3512 }
3513
3514 write_hash_u64(&mut hasher, entity.value());
3515 write_hash_u32(&mut hasher, persisted.version().get());
3516 write_hash_len_u32(&mut hasher, persisted.indexes().len());
3517 for index in persisted.indexes() {
3518 write_hash_u32(&mut hasher, u32::from(index.ordinal()));
3519 write_hash_str_u32(&mut hasher, index.name());
3520 write_hash_str_u32(&mut hasher, index.store());
3521 write_hash_tag_u8(&mut hasher, u8::from(index.unique()));
3522 write_hash_str_u32(&mut hasher, persisted_index_origin_name(index.origin()));
3523 match index.predicate_sql() {
3524 Some(predicate_sql) => {
3525 write_hash_tag_u8(&mut hasher, 1);
3526 write_hash_str_u32(&mut hasher, predicate_sql);
3527 }
3528 None => write_hash_tag_u8(&mut hasher, 0),
3529 }
3530 hash_persisted_index_key(&mut hasher, index.key());
3531 }
3532 }
3533
3534 Ok(finalize_schema_metadata(
3535 max_version,
3536 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
3537 hasher,
3538 latest_by_entity.len(),
3539 ))
3540}
3541
3542fn derive_schema_catalog_metadata(
3543 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
3544) -> Result<SchemaStoreCatalogMetadata, InternalError> {
3545 let mut max_version = SchemaVersion::initial();
3546 let mut hasher = new_hash_sha256();
3547 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN);
3548
3549 for (entity, (version, snapshot)) in latest_by_entity {
3550 let persisted = snapshot.decode_persisted_snapshot()?;
3551 if persisted.version() > max_version {
3552 max_version = persisted.version();
3553 }
3554
3555 write_hash_u64(&mut hasher, entity.value());
3556 write_hash_u32(&mut hasher, version.get());
3557 write_hash_len_u32(&mut hasher, snapshot.as_bytes().len());
3558 hasher.update(snapshot.as_bytes());
3559 }
3560
3561 Ok(finalize_schema_metadata(
3562 max_version,
3563 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
3564 hasher,
3565 latest_by_entity.len(),
3566 ))
3567}
3568
3569fn finalize_schema_metadata(
3570 schema_version: SchemaVersion,
3571 schema_fingerprint_method_version: u8,
3572 hasher: sha2::Sha256,
3573 entity_count: usize,
3574) -> SchemaStoreCatalogMetadata {
3575 let digest = finalize_hash_sha256(hasher);
3576 let mut schema_fingerprint = [0u8; 16];
3577 schema_fingerprint.copy_from_slice(&digest[..16]);
3578
3579 SchemaStoreCatalogMetadata::new(
3580 schema_version,
3581 schema_fingerprint_method_version,
3582 schema_fingerprint,
3583 u64::try_from(entity_count).unwrap_or(u64::MAX),
3584 )
3585}
3586
3587fn hash_persisted_index_key(hasher: &mut sha2::Sha256, key: &PersistedIndexKeySnapshot) {
3588 match key {
3589 PersistedIndexKeySnapshot::FieldPath(paths) => {
3590 write_hash_tag_u8(hasher, 1);
3591 write_hash_len_u32(hasher, paths.len());
3592 for path in paths {
3593 hash_persisted_index_field_path(hasher, path);
3594 }
3595 }
3596 PersistedIndexKeySnapshot::Items(items) => {
3597 write_hash_tag_u8(hasher, 2);
3598 write_hash_len_u32(hasher, items.len());
3599 for item in items {
3600 match item {
3601 PersistedIndexKeyItemSnapshot::FieldPath(path) => {
3602 write_hash_tag_u8(hasher, 1);
3603 hash_persisted_index_field_path(hasher, path);
3604 }
3605 PersistedIndexKeyItemSnapshot::Expression(expression) => {
3606 write_hash_tag_u8(hasher, 2);
3607 write_hash_str_u32(hasher, persisted_expression_op_name(expression.op()));
3608 hash_persisted_index_field_path(hasher, expression.source());
3609 hash_accepted_field_kind(hasher, expression.input_kind());
3610 hash_accepted_field_kind(hasher, expression.output_kind());
3611 write_hash_str_u32(hasher, expression.canonical_text());
3612 }
3613 }
3614 }
3615 }
3616 }
3617}
3618
3619fn hash_persisted_index_field_path(
3620 hasher: &mut sha2::Sha256,
3621 path: &crate::db::schema::PersistedIndexFieldPathSnapshot,
3622) {
3623 write_hash_u32(hasher, path.field_id().get());
3624 write_hash_u32(hasher, u32::from(path.slot().get()));
3625 write_hash_len_u32(hasher, path.path().len());
3626 for segment in path.path() {
3627 write_hash_str_u32(hasher, segment);
3628 }
3629 hash_accepted_field_kind(hasher, path.kind());
3630 write_hash_tag_u8(hasher, u8::from(path.nullable()));
3631}
3632
3633fn hash_accepted_field_kind(hasher: &mut sha2::Sha256, kind: &AcceptedFieldKind) {
3634 match kind {
3635 AcceptedFieldKind::Account => write_hash_tag_u8(hasher, 1),
3636 AcceptedFieldKind::Blob { max_len } => {
3637 write_hash_tag_u8(hasher, 2);
3638 hash_optional_u32(hasher, *max_len);
3639 }
3640 AcceptedFieldKind::Bool => {
3641 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL);
3642 }
3643 AcceptedFieldKind::Date => write_hash_tag_u8(hasher, 4),
3644 AcceptedFieldKind::Decimal { scale } => {
3645 write_hash_tag_u8(hasher, 5);
3646 write_hash_u32(hasher, *scale);
3647 }
3648 AcceptedFieldKind::Duration => write_hash_tag_u8(hasher, 6),
3649 AcceptedFieldKind::Enum { type_id } => {
3650 write_hash_tag_u8(hasher, 7);
3651 write_hash_u32(hasher, type_id.get());
3652 }
3653 AcceptedFieldKind::Float32 => write_hash_tag_u8(hasher, 8),
3654 AcceptedFieldKind::Float64 => write_hash_tag_u8(hasher, 9),
3655 AcceptedFieldKind::Int8 => write_hash_tag_u8(hasher, 10),
3656 AcceptedFieldKind::Int16 => write_hash_tag_u8(hasher, 11),
3657 AcceptedFieldKind::Int32 => write_hash_tag_u8(hasher, 12),
3658 AcceptedFieldKind::Int64 => write_hash_tag_u8(hasher, 13),
3659 AcceptedFieldKind::Int128 => write_hash_tag_u8(hasher, 14),
3660 AcceptedFieldKind::IntBig { max_bytes } => {
3661 write_hash_tag_u8(hasher, 15);
3662 write_hash_u32(hasher, *max_bytes);
3663 }
3664 AcceptedFieldKind::Principal => write_hash_tag_u8(hasher, 16),
3665 AcceptedFieldKind::Subaccount => write_hash_tag_u8(hasher, 17),
3666 AcceptedFieldKind::Text { max_len } => {
3667 write_hash_tag_u8(hasher, 18);
3668 hash_optional_u32(hasher, *max_len);
3669 }
3670 AcceptedFieldKind::Timestamp => write_hash_tag_u8(hasher, 19),
3671 AcceptedFieldKind::Nat8 => write_hash_tag_u8(hasher, 20),
3672 AcceptedFieldKind::Nat16 => write_hash_tag_u8(hasher, 21),
3673 AcceptedFieldKind::Nat32 => write_hash_tag_u8(hasher, 22),
3674 AcceptedFieldKind::Nat64 => write_hash_tag_u8(hasher, 23),
3675 AcceptedFieldKind::Nat128 => write_hash_tag_u8(hasher, 24),
3676 AcceptedFieldKind::NatBig { max_bytes } => {
3677 write_hash_tag_u8(hasher, 25);
3678 write_hash_u32(hasher, *max_bytes);
3679 }
3680 AcceptedFieldKind::Ulid => write_hash_tag_u8(hasher, 26),
3681 AcceptedFieldKind::Unit => write_hash_tag_u8(hasher, 27),
3682 AcceptedFieldKind::Relation {
3683 target_path,
3684 target_entity_name,
3685 target_entity_tag,
3686 target_store_path,
3687 key_kind,
3688 } => {
3689 write_hash_tag_u8(hasher, 28);
3690 write_hash_str_u32(hasher, target_path);
3691 write_hash_str_u32(hasher, target_entity_name);
3692 write_hash_u64(hasher, target_entity_tag.value());
3693 write_hash_str_u32(hasher, target_store_path);
3694 hash_accepted_field_kind(hasher, key_kind);
3695 }
3696 AcceptedFieldKind::List(inner) => {
3697 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST);
3698 hash_accepted_field_kind(hasher, inner);
3699 }
3700 AcceptedFieldKind::Set(inner) => {
3701 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET);
3702 hash_accepted_field_kind(hasher, inner);
3703 }
3704 AcceptedFieldKind::Map { key, value } => {
3705 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP);
3706 hash_accepted_field_kind(hasher, key);
3707 hash_accepted_field_kind(hasher, value);
3708 }
3709 AcceptedFieldKind::Composite { type_id } => {
3710 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE);
3711 write_hash_u32(hasher, type_id.get());
3712 }
3713 }
3714}
3715
3716fn hash_optional_u32(hasher: &mut sha2::Sha256, value: Option<u32>) {
3717 match value {
3718 Some(value) => {
3719 write_hash_tag_u8(hasher, 1);
3720 write_hash_u32(hasher, value);
3721 }
3722 None => write_hash_tag_u8(hasher, 0),
3723 }
3724}
3725
3726const fn persisted_index_origin_name(
3727 origin: crate::db::schema::PersistedIndexOrigin,
3728) -> &'static str {
3729 match origin {
3730 crate::db::schema::PersistedIndexOrigin::Generated => "generated",
3731 crate::db::schema::PersistedIndexOrigin::SqlDdl => "sql_ddl",
3732 }
3733}
3734
3735const fn persisted_expression_op_name(
3736 op: crate::db::schema::PersistedIndexExpressionOp,
3737) -> &'static str {
3738 match op {
3739 crate::db::schema::PersistedIndexExpressionOp::Lower => "lower",
3740 crate::db::schema::PersistedIndexExpressionOp::Upper => "upper",
3741 crate::db::schema::PersistedIndexExpressionOp::Trim => "trim",
3742 crate::db::schema::PersistedIndexExpressionOp::LowerTrim => "lower_trim",
3743 crate::db::schema::PersistedIndexExpressionOp::Date => "date",
3744 crate::db::schema::PersistedIndexExpressionOp::Year => "year",
3745 crate::db::schema::PersistedIndexExpressionOp::Month => "month",
3746 crate::db::schema::PersistedIndexExpressionOp::Day => "day",
3747 }
3748}
3749
3750#[cfg(test)]
3755mod tests;