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::db::schema::{
14 cardinality_build::{
15 CardinalityAcceptedDomain, CardinalityReadyCandidate, EmptyCardinalityReadyCandidate,
16 },
17 cardinality_generation::{
18 CardinalityAcceptedRootIdentity, CardinalityBuildCursor, CardinalityCountDigest,
19 CardinalityCountRecord, CardinalityCountSlot, CardinalityGenerationHeader,
20 CardinalityGenerationId, CardinalityGenerationState, CardinalitySourceIdentity,
21 },
22};
23use crate::{
24 db::{
25 codec::{
26 finalize_hash_sha256, new_hash_sha256, write_hash_len_u32, write_hash_str_u32,
27 write_hash_tag_u8, write_hash_u32, write_hash_u64,
28 },
29 commit::CommitSchemaFingerprint,
30 direction::Direction,
31 integrity::DatabaseIncarnationId,
32 journal::{JournalBatch, JournalRecord},
33 ordered_overlay::{OrderedOverlayEntry, OrderedOverlayVisit, visit_ordered_overlay},
34 positioned_overlay::{
35 JournalOverlayPosition, PositionedOverlayMetadata, PositionedOverlayRetirement,
36 },
37 runtime_entity_catalog::AcceptedRuntimeEntity,
38 schema::{
39 AcceptedFieldKind, AcceptedRowLayoutRuntimeContract, AcceptedSchemaSnapshot,
40 ConstraintActivationKind, ConstraintActivationState, ConstraintId, ConstraintOrigin,
41 ConstraintValidationJob, FieldId, PersistedIndexKeyItemSnapshot,
42 PersistedIndexKeySnapshot, PersistedSchemaSnapshot, SchemaVersion,
43 accepted_schema_cache_fingerprint,
44 accepted_schema_cache_fingerprint_for_persisted_snapshot,
45 accepted_schema_cache_fingerprint_method_version, decode_constraint_validation_job,
46 decode_persisted_schema_snapshot, encode_constraint_validation_job,
47 encode_persisted_schema_snapshot,
48 enum_catalog::{
49 AcceptedSchemaAuthority, AcceptedSchemaPublicationError, AcceptedSchemaRevision,
50 AcceptedSchemaRevisionBundle, AcceptedSchemaRootSelection,
51 AcceptedStoreCatalogScope, AcceptedValueCatalogHandle, CandidateSchemaRevision,
52 decode_verified_accepted_schema_revision_bundle,
53 prepare_accepted_schema_root_publication, select_current_accepted_schema_root,
54 },
55 schema_snapshot_integrity_detail,
56 },
57 },
58 error::InternalError,
59 types::EntityTag,
60};
61use ic_stable_structures::{
62 BTreeMap as StableBTreeMap, DefaultMemoryImpl, Storable, memory_manager::VirtualMemory,
63 storable::Bound as StorableBound,
64};
65use sha2::Digest;
66use std::borrow::Cow;
67#[cfg(test)]
68use std::cell::Cell;
69use std::cell::{OnceCell, Ref, RefCell};
70use std::collections::{BTreeMap as StdBTreeMap, BTreeSet};
71#[cfg(test)]
72use std::convert::Infallible;
73use std::ops::Bound as RangeBound;
74use std::rc::Rc;
75
76const SCHEMA_KEY_BYTES_USIZE: usize = 16;
77const SCHEMA_KEY_BYTES: u32 = 16;
78const SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT: u8 = 0;
79const SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE: u8 = 1;
80const SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT: u8 = 2;
81const SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB: u8 = 3;
82const SCHEMA_KEY_NAMESPACE_IDENTITY_STATE: u8 = 4;
83const SCHEMA_STORE_FINGERPRINT_METHOD_VERSION: u8 = 1;
86const SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN: u8 = 1;
87const SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN: u8 = 2;
88const SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN: u8 = 3;
89const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL: u8 = 3;
90const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST: u8 = 29;
91const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET: u8 = 30;
92const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP: u8 = 31;
93const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE: u8 = 32;
94const RAW_SCHEMA_SNAPSHOT_MAGIC: &[u8; 8] = b"ICYDBCAT";
95const RAW_SCHEMA_SNAPSHOT_VALUE_VERSION: u8 = 1;
96const RAW_SCHEMA_SNAPSHOT_HEADER_BYTES: usize = 25;
97
98pub(in crate::db) fn load_accepted_schema_snapshot(
103 schema_store: &SchemaStore,
104 entity_tag: EntityTag,
105 entity_path: &str,
106) -> Result<AcceptedSchemaSnapshot, InternalError> {
107 let bundle = schema_store
108 .current_accepted_schema_bundle()?
109 .ok_or_else(InternalError::store_corruption)?;
110 let snapshot = bundle
111 .entity_snapshots()
112 .get(&entity_tag)
113 .cloned()
114 .ok_or_else(InternalError::store_corruption)?;
115 if snapshot.entity_path() != entity_path {
116 return Err(InternalError::store_corruption());
117 }
118 let accepted = AcceptedSchemaSnapshot::try_new(snapshot)?;
119 let _runtime_contract = AcceptedRowLayoutRuntimeContract::from_accepted_schema(&accepted)?;
120
121 Ok(accepted)
122}
123
124#[cfg(test)]
125thread_local! {
126 static ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES: Cell<u64> = const { Cell::new(0) };
127}
128
129#[cfg(test)]
130fn reset_accepted_schema_bundle_cache_miss_count_for_tests() {
131 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(|misses| misses.set(0));
132}
133
134#[cfg(test)]
135fn accepted_schema_bundle_cache_miss_count_for_tests() -> u64 {
136 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(Cell::get)
137}
138
139#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
148struct RawSchemaKey([u8; SCHEMA_KEY_BYTES_USIZE]);
149
150impl RawSchemaKey {
151 #[must_use]
153 fn from_entity_version(entity: EntityTag, version: SchemaVersion) -> Self {
154 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
155 out[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
156 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
157 out[12..].copy_from_slice(&version.get().to_be_bytes());
158
159 Self(out)
160 }
161
162 fn from_accepted_bundle(bundle_key: super::enum_catalog::AcceptedSchemaBundleKey) -> Self {
163 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
164 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE;
165 out[4..12].copy_from_slice(&bundle_key.get().to_be_bytes());
166 Self(out)
167 }
168
169 fn from_accepted_root_slot(slot: usize) -> Result<Self, InternalError> {
170 let slot = u32::try_from(slot).map_err(|_| InternalError::store_invariant())?;
171 if slot > 1 {
172 return Err(InternalError::store_invariant());
173 }
174 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
175 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT;
176 out[12..].copy_from_slice(&slot.to_be_bytes());
177 Ok(Self(out))
178 }
179
180 fn from_constraint_validation_job(entity: EntityTag, constraint_id: ConstraintId) -> Self {
181 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
182 out[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
183 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
184 out[12..].copy_from_slice(&constraint_id.get().to_be_bytes());
185 Self(out)
186 }
187
188 fn from_identity_state(entity: EntityTag, field_id: FieldId) -> Self {
189 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
190 out[0] = SCHEMA_KEY_NAMESPACE_IDENTITY_STATE;
191 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
192 out[12..].copy_from_slice(&field_id.get().to_be_bytes());
193 Self(out)
194 }
195
196 #[must_use]
198 fn entity_tag(self) -> EntityTag {
199 let mut bytes = [0u8; size_of::<u64>()];
200 bytes.copy_from_slice(&self.0[4..12]);
201
202 EntityTag::new(u64::from_be_bytes(bytes))
203 }
204
205 #[must_use]
207 fn version(self) -> u32 {
208 let mut bytes = [0u8; size_of::<u32>()];
209 bytes.copy_from_slice(&self.0[12..]);
210
211 u32::from_be_bytes(bytes)
212 }
213
214 const fn all_entity_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
215 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
216 end[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
217 (
218 RangeBound::Included(Self([0; SCHEMA_KEY_BYTES_USIZE])),
219 RangeBound::Included(Self(end)),
220 )
221 }
222
223 #[cfg(test)]
224 fn entity_range_bounds(entity: EntityTag) -> (RangeBound<Self>, RangeBound<Self>) {
225 (
226 RangeBound::Included(Self::from_entity_version(entity, SchemaVersion::initial())),
227 RangeBound::Included(Self::from_entity_version(
228 entity,
229 SchemaVersion::new(u32::MAX),
230 )),
231 )
232 }
233
234 const fn all_constraint_validation_job_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
235 let mut start = [0u8; SCHEMA_KEY_BYTES_USIZE];
236 start[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
237 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
238 end[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
239 (
240 RangeBound::Included(Self(start)),
241 RangeBound::Included(Self(end)),
242 )
243 }
244
245 const fn all_identity_state_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
246 let mut start = [0u8; SCHEMA_KEY_BYTES_USIZE];
247 start[0] = SCHEMA_KEY_NAMESPACE_IDENTITY_STATE;
248 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
249 end[0] = SCHEMA_KEY_NAMESPACE_IDENTITY_STATE;
250 (
251 RangeBound::Included(Self(start)),
252 RangeBound::Included(Self(end)),
253 )
254 }
255
256 #[cfg(test)]
257 const fn is_entity_snapshot(self) -> bool {
258 self.0[0] == SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT
259 }
260
261 const fn is_accepted_root(self) -> bool {
262 self.0[0] == SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT
263 }
264
265 const fn is_constraint_validation_job(self) -> bool {
266 self.0[0] == SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB
267 }
268
269 const fn is_identity_state(self) -> bool {
270 self.0[0] == SCHEMA_KEY_NAMESPACE_IDENTITY_STATE
271 }
272
273 fn constraint_id(self) -> Option<ConstraintId> {
274 self.is_constraint_validation_job()
275 .then(|| ConstraintId::new(self.version()))
276 .flatten()
277 }
278}
279
280impl RawSchemaKey {
281 const NAMESPACE_CARDINALITY_CONTROL: u8 = 5;
282 const NAMESPACE_CARDINALITY_COUNT_A: u8 = 6;
283 const NAMESPACE_CARDINALITY_COUNT_B: u8 = 7;
284 const CARDINALITY_HEADER_DISCRIMINATOR: u8 = 0;
285 const CARDINALITY_BUILD_CURSOR_DISCRIMINATOR: u8 = 1;
286
287 const fn from_cardinality_generation_header() -> Self {
288 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
289 out[0] = Self::NAMESPACE_CARDINALITY_CONTROL;
290 out[SCHEMA_KEY_BYTES_USIZE - 1] = Self::CARDINALITY_HEADER_DISCRIMINATOR;
291 Self(out)
292 }
293
294 const fn from_cardinality_build_cursor() -> Self {
295 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
296 out[0] = Self::NAMESPACE_CARDINALITY_CONTROL;
297 out[SCHEMA_KEY_BYTES_USIZE - 1] = Self::CARDINALITY_BUILD_CURSOR_DISCRIMINATOR;
298 Self(out)
299 }
300
301 fn from_cardinality_count(slot: CardinalityCountSlot, digest: CardinalityCountDigest) -> Self {
302 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
303 out[0] = match slot {
304 CardinalityCountSlot::A => Self::NAMESPACE_CARDINALITY_COUNT_A,
305 CardinalityCountSlot::B => Self::NAMESPACE_CARDINALITY_COUNT_B,
306 };
307 out[1..].copy_from_slice(&digest.as_bytes()[..SCHEMA_KEY_BYTES_USIZE - 1]);
308 Self(out)
309 }
310
311 const fn cardinality_count_range_bounds(
312 slot: CardinalityCountSlot,
313 ) -> (RangeBound<Self>, RangeBound<Self>) {
314 let namespace = match slot {
315 CardinalityCountSlot::A => Self::NAMESPACE_CARDINALITY_COUNT_A,
316 CardinalityCountSlot::B => Self::NAMESPACE_CARDINALITY_COUNT_B,
317 };
318 let mut start = [0_u8; SCHEMA_KEY_BYTES_USIZE];
319 start[0] = namespace;
320 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
321 end[0] = namespace;
322 (
323 RangeBound::Included(Self(start)),
324 RangeBound::Included(Self(end)),
325 )
326 }
327}
328
329impl Storable for RawSchemaKey {
330 fn to_bytes(&self) -> Cow<'_, [u8]> {
331 Cow::Borrowed(&self.0)
332 }
333
334 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
335 debug_assert_eq!(
336 bytes.len(),
337 SCHEMA_KEY_BYTES_USIZE,
338 "RawSchemaKey::from_bytes received unexpected byte length",
339 );
340
341 if bytes.len() != SCHEMA_KEY_BYTES_USIZE {
342 return Self([0u8; SCHEMA_KEY_BYTES_USIZE]);
343 }
344
345 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
346 out.copy_from_slice(bytes.as_ref());
347 Self(out)
348 }
349
350 fn into_bytes(self) -> Vec<u8> {
351 self.0.to_vec()
352 }
353
354 const BOUND: StorableBound = StorableBound::Bounded {
355 max_size: SCHEMA_KEY_BYTES,
356 is_fixed_size: true,
357 };
358}
359
360#[derive(Clone, Debug, Eq, PartialEq)]
371struct RawSchemaSnapshot {
372 payload: Vec<u8>,
373 accepted_schema_fingerprint: Option<CommitSchemaFingerprint>,
374}
375
376impl RawSchemaSnapshot {
377 fn from_persisted_snapshot(snapshot: &PersistedSchemaSnapshot) -> Result<Self, InternalError> {
379 validate_typed_schema_snapshot_for_store(snapshot)?;
380
381 let accepted_schema_fingerprint =
382 accepted_schema_cache_fingerprint_for_persisted_snapshot(snapshot)?;
383 let payload = encode_persisted_schema_snapshot(snapshot)?;
384
385 Ok(Self {
386 payload,
387 accepted_schema_fingerprint: Some(accepted_schema_fingerprint),
388 })
389 }
390
391 #[must_use]
393 const fn from_encoded_control_record(payload: Vec<u8>) -> Self {
394 Self {
395 payload,
396 accepted_schema_fingerprint: None,
397 }
398 }
399
400 #[cfg(test)]
403 #[must_use]
404 const fn from_unchecked_persisted_snapshot_payload(payload: Vec<u8>) -> Self {
405 Self {
406 payload,
407 accepted_schema_fingerprint: Some([0; size_of::<CommitSchemaFingerprint>()]),
408 }
409 }
410
411 #[must_use]
413 const fn as_bytes(&self) -> &[u8] {
414 self.payload.as_slice()
415 }
416
417 #[must_use]
419 fn into_bytes(self) -> Vec<u8> {
420 self.payload
421 }
422
423 fn accepted_schema_fingerprint(&self) -> Result<CommitSchemaFingerprint, InternalError> {
426 self.accepted_schema_fingerprint
427 .ok_or_else(InternalError::store_corruption)
428 }
429
430 fn decode_persisted_snapshot(&self) -> Result<PersistedSchemaSnapshot, InternalError> {
432 let _fingerprint = self.accepted_schema_fingerprint()?;
435 decode_persisted_schema_snapshot(self.as_bytes())
436 }
437}
438
439#[cfg(test)]
440pub(in crate::db::schema) fn validate_raw_schema_snapshot_bytes_for_tests(
441 bytes: Vec<u8>,
442) -> Result<(), InternalError> {
443 let raw = <RawSchemaSnapshot as Storable>::from_bytes(Cow::Owned(bytes));
444 raw.decode_persisted_snapshot().map(drop)
445}
446
447#[derive(Clone, Debug, Eq, PartialEq)]
448pub(in crate::db) struct AcceptedCatalogIdentity {
449 entity_tag: EntityTag,
450 entity_path: Rc<str>,
451 store_path: &'static str,
452 accepted_schema_revision: AcceptedSchemaRevision,
453 accepted_schema_version: SchemaVersion,
454 fingerprint_method_version: u8,
455 accepted_schema_fingerprint: CommitSchemaFingerprint,
456}
457
458impl AcceptedCatalogIdentity {
459 #[must_use]
460 pub(in crate::db) fn new(
461 entity_tag: EntityTag,
462 entity_path: impl Into<Rc<str>>,
463 store_path: &'static str,
464 accepted_schema_revision: AcceptedSchemaRevision,
465 accepted_schema_version: SchemaVersion,
466 accepted_schema_fingerprint: CommitSchemaFingerprint,
467 ) -> Self {
468 Self {
469 entity_tag,
470 entity_path: entity_path.into(),
471 store_path,
472 accepted_schema_revision,
473 accepted_schema_version,
474 fingerprint_method_version: accepted_schema_cache_fingerprint_method_version(),
475 accepted_schema_fingerprint,
476 }
477 }
478
479 #[must_use]
480 pub(in crate::db) const fn entity_tag(&self) -> EntityTag {
481 self.entity_tag
482 }
483
484 #[must_use]
485 pub(in crate::db) fn entity_path(&self) -> &str {
486 self.entity_path.as_ref()
487 }
488
489 #[must_use]
490 pub(in crate::db) fn entity_path_handle(&self) -> Rc<str> {
491 self.entity_path.clone()
492 }
493
494 #[must_use]
495 pub(in crate::db) const fn store_path(&self) -> &'static str {
496 self.store_path
497 }
498
499 #[must_use]
500 pub(in crate::db) const fn accepted_schema_revision(&self) -> AcceptedSchemaRevision {
501 self.accepted_schema_revision
502 }
503
504 #[must_use]
505 pub(in crate::db) const fn accepted_schema_version(&self) -> SchemaVersion {
506 self.accepted_schema_version
507 }
508
509 #[must_use]
510 pub(in crate::db) const fn fingerprint_method_version(&self) -> u8 {
511 self.fingerprint_method_version
512 }
513
514 #[must_use]
515 pub(in crate::db) const fn accepted_schema_fingerprint(&self) -> CommitSchemaFingerprint {
516 self.accepted_schema_fingerprint
517 }
518}
519
520#[derive(Clone, Debug, Eq, PartialEq)]
521pub(in crate::db) struct AcceptedCatalogSnapshotSelection {
522 identity: AcceptedCatalogIdentity,
523 value_catalog: AcceptedValueCatalogHandle,
524 raw_snapshot: Rc<[u8]>,
525}
526
527impl AcceptedCatalogSnapshotSelection {
528 #[must_use]
529 const fn new(
530 identity: AcceptedCatalogIdentity,
531 value_catalog: AcceptedValueCatalogHandle,
532 raw_snapshot: Rc<[u8]>,
533 ) -> Self {
534 Self {
535 identity,
536 value_catalog,
537 raw_snapshot,
538 }
539 }
540
541 #[must_use]
542 pub(in crate::db) fn identity(&self) -> AcceptedCatalogIdentity {
543 self.identity.clone()
544 }
545
546 #[must_use]
547 pub(in crate::db) const fn value_catalog_handle(&self) -> &AcceptedValueCatalogHandle {
548 &self.value_catalog
549 }
550
551 pub(in crate::db) fn from_candidate(
554 candidate: &CandidateSchemaRevision,
555 entity_tag: EntityTag,
556 entity_path: &str,
557 store_path: &'static str,
558 ) -> Result<Option<Self>, InternalError> {
559 if candidate.store_path() != store_path {
560 return Err(InternalError::store_corruption());
561 }
562 let Some(snapshot) = candidate.bundle().entity_snapshots().get(&entity_tag) else {
563 return Ok(None);
564 };
565 if snapshot.entity_path() != entity_path {
566 return Err(InternalError::store_corruption());
567 }
568
569 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
570 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
571 let identity = AcceptedCatalogIdentity::new(
572 entity_tag,
573 entity_path,
574 store_path,
575 candidate.revision(),
576 snapshot.version(),
577 fingerprint,
578 );
579
580 Ok(Some(Self::new(
581 identity,
582 AcceptedValueCatalogHandle::new(
583 candidate.bundle().enum_catalog().clone(),
584 candidate.bundle().composite_catalog().clone(),
585 AcceptedStoreCatalogScope::new(),
586 candidate.revision(),
587 candidate.root().fingerprint(),
588 ),
589 Rc::from(raw_snapshot.into_bytes()),
590 )))
591 }
592
593 pub(in crate::db) fn decode_verified(&self) -> Result<AcceptedSchemaSnapshot, InternalError> {
594 let snapshot = decode_persisted_schema_snapshot(self.raw_snapshot.as_ref())?;
595 let accepted = AcceptedSchemaSnapshot::try_new(snapshot)?;
596 let identity = self.identity();
597
598 if accepted.persisted_snapshot().version() != identity.accepted_schema_version() {
599 return Err(InternalError::store_invariant());
600 }
601 if accepted.entity_path() != identity.entity_path() {
602 return Err(InternalError::store_invariant());
603 }
604
605 let decoded_fingerprint = accepted_schema_cache_fingerprint(&accepted)?;
606 if decoded_fingerprint != identity.accepted_schema_fingerprint() {
607 return Err(InternalError::store_invariant());
608 }
609
610 Ok(accepted)
611 }
612}
613
614impl Storable for RawSchemaSnapshot {
615 fn to_bytes(&self) -> Cow<'_, [u8]> {
616 let Some(fingerprint) = self.accepted_schema_fingerprint else {
617 return Cow::Borrowed(self.as_bytes());
618 };
619
620 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
621 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
622 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
623 bytes.extend_from_slice(&fingerprint);
624 bytes.extend_from_slice(self.as_bytes());
625
626 Cow::Owned(bytes)
627 }
628
629 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
630 let bytes = bytes.into_owned();
631 if bytes.len() >= RAW_SCHEMA_SNAPSHOT_HEADER_BYTES
632 && &bytes[..RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_MAGIC
633 && bytes[RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_VALUE_VERSION
634 {
635 let fingerprint_start = RAW_SCHEMA_SNAPSHOT_MAGIC.len() + size_of::<u8>();
636 let fingerprint_end = fingerprint_start + size_of::<CommitSchemaFingerprint>();
637 let mut fingerprint = [0_u8; size_of::<CommitSchemaFingerprint>()];
638 fingerprint.copy_from_slice(&bytes[fingerprint_start..fingerprint_end]);
639
640 return Self {
641 payload: bytes[fingerprint_end..].to_vec(),
642 accepted_schema_fingerprint: Some(fingerprint),
643 };
644 }
645
646 Self {
647 payload: bytes,
648 accepted_schema_fingerprint: None,
649 }
650 }
651
652 fn into_bytes(self) -> Vec<u8> {
653 let Some(fingerprint) = self.accepted_schema_fingerprint else {
654 return self.payload;
655 };
656
657 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
658 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
659 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
660 bytes.extend_from_slice(&fingerprint);
661 bytes.extend_from_slice(&self.payload);
662
663 bytes
664 }
665
666 const BOUND: StorableBound = StorableBound::Unbounded;
667}
668
669fn validate_typed_schema_snapshot_for_store(
673 snapshot: &PersistedSchemaSnapshot,
674) -> Result<(), InternalError> {
675 if schema_snapshot_integrity_detail(
676 "schema snapshot",
677 snapshot.version(),
678 snapshot.primary_key_field_ids(),
679 snapshot.row_layout(),
680 snapshot.fields(),
681 )
682 .is_some()
683 {
684 return Err(InternalError::store_invariant());
685 }
686
687 Ok(())
688}
689
690#[derive(Clone, Copy, Debug, Eq, PartialEq)]
698pub(in crate::db) struct SchemaStoreCatalogMetadata {
699 schema_version: SchemaVersion,
700 schema_fingerprint_method_version: u8,
701 schema_fingerprint: CommitSchemaFingerprint,
702 entity_count: u64,
703}
704
705impl SchemaStoreCatalogMetadata {
706 #[must_use]
708 const fn new(
709 schema_version: SchemaVersion,
710 schema_fingerprint_method_version: u8,
711 schema_fingerprint: CommitSchemaFingerprint,
712 entity_count: u64,
713 ) -> Self {
714 Self {
715 schema_version,
716 schema_fingerprint_method_version,
717 schema_fingerprint,
718 entity_count,
719 }
720 }
721
722 #[must_use]
724 pub(in crate::db) const fn schema_version(self) -> SchemaVersion {
725 self.schema_version
726 }
727
728 #[must_use]
730 pub(in crate::db) const fn schema_fingerprint_method_version(self) -> u8 {
731 self.schema_fingerprint_method_version
732 }
733
734 #[must_use]
737 pub(in crate::db) const fn schema_fingerprint(self) -> CommitSchemaFingerprint {
738 self.schema_fingerprint
739 }
740
741 #[must_use]
743 pub(in crate::db) const fn entity_count(self) -> u64 {
744 self.entity_count
745 }
746}
747
748#[derive(Clone, Copy, Debug, Eq, PartialEq)]
757pub(in crate::db) struct SchemaStoreAllocationMetadata {
758 data: SchemaStoreCatalogMetadata,
759 index: SchemaStoreCatalogMetadata,
760 schema: SchemaStoreCatalogMetadata,
761}
762
763impl SchemaStoreAllocationMetadata {
764 #[must_use]
767 const fn new(
768 data: SchemaStoreCatalogMetadata,
769 index: SchemaStoreCatalogMetadata,
770 schema: SchemaStoreCatalogMetadata,
771 ) -> Self {
772 Self {
773 data,
774 index,
775 schema,
776 }
777 }
778
779 #[must_use]
781 pub(in crate::db) const fn data(self) -> SchemaStoreCatalogMetadata {
782 self.data
783 }
784
785 #[must_use]
787 pub(in crate::db) const fn index(self) -> SchemaStoreCatalogMetadata {
788 self.index
789 }
790
791 #[must_use]
794 pub(in crate::db) const fn schema(self) -> SchemaStoreCatalogMetadata {
795 self.schema
796 }
797}
798
799pub(in crate::db) struct PendingRelationActivationDeleteBarrier {
807 accepted_schema_fingerprint: CommitSchemaFingerprint,
808 source_entity_tag: EntityTag,
809 constraint_id: ConstraintId,
810}
811
812impl PendingRelationActivationDeleteBarrier {
813 #[must_use]
814 pub(in crate::db) const fn accepted_schema_fingerprint(&self) -> CommitSchemaFingerprint {
815 self.accepted_schema_fingerprint
816 }
817
818 #[must_use]
819 pub(in crate::db) const fn source_entity_tag(&self) -> EntityTag {
820 self.source_entity_tag
821 }
822
823 #[must_use]
825 pub(in crate::db) const fn constraint_id(&self) -> ConstraintId {
826 self.constraint_id
827 }
828}
829
830pub struct SchemaStore {
839 backend: SchemaStoreBackend,
840 accepted_bundle_cache: RefCell<Option<AcceptedSchemaBundleCache>>,
841 cardinality_header_cache: RefCell<Option<(Vec<u8>, CardinalityGenerationHeader)>>,
842 accepted_catalog_scope: OnceCell<AcceptedStoreCatalogScope>,
843}
844
845struct AcceptedSchemaBundleCache {
846 selection: AcceptedSchemaRootSelection,
847 bundle: AcceptedSchemaRevisionBundle,
848 cardinality_domain: Rc<CardinalityAcceptedDomain>,
849 value_catalog: AcceptedValueCatalogHandle,
850 entity_selections: RefCell<StdBTreeMap<EntityTag, AcceptedCatalogSnapshotSelection>>,
851}
852
853enum SchemaStoreBackend {
854 Heap(StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>),
855 Journaled {
856 canonical:
857 StableBTreeMap<RawSchemaKey, RawSchemaSnapshot, VirtualMemory<DefaultMemoryImpl>>,
858 live: StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
859 tombstones: BTreeSet<RawSchemaKey>,
860 positions: PositionedOverlayMetadata<RawSchemaKey>,
861 },
862}
863
864#[derive(Clone, Copy, Debug, Eq, PartialEq)]
866enum SchemaStoreVisit {
867 Continue,
868 #[cfg(test)]
869 Stop,
870}
871
872impl SchemaStoreVisit {
873 const fn should_stop(self) -> bool {
874 match self {
875 Self::Continue => false,
876 #[cfg(test)]
877 Self::Stop => true,
878 }
879 }
880}
881
882#[derive(Clone, Copy)]
883enum IdentityStateStorageView {
884 Effective,
885 Canonical,
886}
887
888#[derive(Clone)]
890pub(in crate::db) struct PreparedSchemaPositionPublication {
891 keys: Vec<RawSchemaKey>,
892 position: JournalOverlayPosition,
893}
894
895pub(in crate::db) struct PreparedSchemaPositionRetirement {
897 entries: Vec<(RawSchemaKey, PositionedOverlayRetirement)>,
898}
899
900pub(in crate::db) struct PreparedCardinalityCountWrites {
902 slot: CardinalityCountSlot,
903 generation: CardinalityGenerationId,
904 entries: Vec<(RawSchemaKey, RawSchemaSnapshot)>,
905 new_count_keys: u64,
906}
907
908impl PreparedCardinalityCountWrites {
909 #[must_use]
910 pub(in crate::db) const fn new_count_keys(&self) -> u64 {
911 self.new_count_keys
912 }
913}
914
915pub(in crate::db) struct PreparedCardinalityBuildPage {
917 count_entries: Vec<(RawSchemaKey, RawSchemaSnapshot)>,
918 cursor: (RawSchemaKey, RawSchemaSnapshot),
919}
920
921pub(in crate::db) struct PreparedCardinalityMaintenance {
923 count_entries: Vec<(RawSchemaKey, Option<RawSchemaSnapshot>)>,
924 header: (RawSchemaKey, RawSchemaSnapshot),
925}
926
927#[derive(Clone, Copy)]
928enum IdentityStateWriteTarget {
929 Durable,
930 Materialized,
931 Canonical,
932}
933
934impl SchemaStore {
935 #[must_use]
937 pub const fn init_heap() -> Self {
938 Self {
939 backend: SchemaStoreBackend::Heap(StdBTreeMap::new()),
940 accepted_bundle_cache: RefCell::new(None),
941 cardinality_header_cache: RefCell::new(None),
942 accepted_catalog_scope: OnceCell::new(),
943 }
944 }
945
946 #[must_use]
951 pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
952 Self {
953 backend: SchemaStoreBackend::Journaled {
954 canonical: StableBTreeMap::init(memory),
955 live: StdBTreeMap::new(),
956 tombstones: BTreeSet::new(),
957 positions: PositionedOverlayMetadata::new(),
958 },
959 accepted_bundle_cache: RefCell::new(None),
960 cardinality_header_cache: RefCell::new(None),
961 accepted_catalog_scope: OnceCell::new(),
962 }
963 }
964
965 pub(in crate::db) fn cardinality_generation_header(
967 &self,
968 ) -> Result<Option<CardinalityGenerationHeader>, InternalError> {
969 let key = RawSchemaKey::from_cardinality_generation_header();
970 let raw = self.get_canonical_raw_value(&key)?;
971 if let Some(raw) = raw.as_ref() {
972 self.decode_cardinality_header_cached(raw).map(Some)
973 } else {
974 self.cardinality_header_cache
975 .try_borrow_mut()
976 .map_err(|_| InternalError::store_invariant())?
977 .take();
978 Ok(None)
979 }
980 }
981
982 pub(in crate::db) fn cardinality_build_cursor(
984 &self,
985 ) -> Result<Option<CardinalityBuildCursor>, InternalError> {
986 let key = RawSchemaKey::from_cardinality_build_cursor();
987 self.get_canonical_raw_value(&key)?
988 .map(|raw| CardinalityBuildCursor::decode(raw.as_bytes()))
989 .transpose()
990 }
991
992 pub(in crate::db) fn cardinality_generation_control(
994 &self,
995 ) -> Result<
996 (
997 Option<CardinalityGenerationHeader>,
998 Option<CardinalityBuildCursor>,
999 ),
1000 InternalError,
1001 > {
1002 let SchemaStoreBackend::Journaled { canonical, .. } = &self.backend else {
1003 return Err(InternalError::store_invariant());
1004 };
1005 let header_key = RawSchemaKey::from_cardinality_generation_header();
1006 let cursor_key = RawSchemaKey::from_cardinality_build_cursor();
1007 let mut header = None;
1008 let mut cursor = None;
1009 for entry in canonical.range(header_key..=cursor_key) {
1010 if *entry.key() == header_key {
1011 header = Some(self.decode_cardinality_header_cached(&entry.value())?);
1012 } else if *entry.key() == cursor_key {
1013 cursor = Some(CardinalityBuildCursor::decode(entry.value().as_bytes())?);
1014 } else {
1015 return Err(InternalError::store_corruption());
1016 }
1017 }
1018 Ok((header, cursor))
1019 }
1020
1021 fn decode_cardinality_header_cached(
1022 &self,
1023 raw: &RawSchemaSnapshot,
1024 ) -> Result<CardinalityGenerationHeader, InternalError> {
1025 if let Some((_, header)) = self
1026 .cardinality_header_cache
1027 .try_borrow()
1028 .map_err(|_| InternalError::store_invariant())?
1029 .as_ref()
1030 .filter(|(bytes, _)| bytes.as_slice() == raw.as_bytes())
1031 {
1032 return Ok(*header);
1033 }
1034 let header = CardinalityGenerationHeader::decode(raw.as_bytes())?;
1035 *self
1036 .cardinality_header_cache
1037 .try_borrow_mut()
1038 .map_err(|_| InternalError::store_invariant())? =
1039 Some((raw.as_bytes().to_vec(), header));
1040 Ok(header)
1041 }
1042
1043 pub(in crate::db) fn cardinality_storage_is_pristine(&self) -> Result<bool, InternalError> {
1045 if self.cardinality_generation_header()?.is_some()
1046 || self.cardinality_build_cursor()?.is_some()
1047 {
1048 return Ok(false);
1049 }
1050 Ok(
1051 self.cardinality_count_slot_is_empty(CardinalityCountSlot::A)?
1052 && self.cardinality_count_slot_is_empty(CardinalityCountSlot::B)?,
1053 )
1054 }
1055
1056 pub(in crate::db) fn write_cardinality_generation_header(
1058 &mut self,
1059 header: CardinalityGenerationHeader,
1060 ) -> Result<(), InternalError> {
1061 self.insert_canonical_raw_value(
1062 RawSchemaKey::from_cardinality_generation_header(),
1063 header.encode(),
1064 )
1065 }
1066
1067 pub(in crate::db) fn restart_cardinality_generation(
1073 &mut self,
1074 current: CardinalityGenerationHeader,
1075 source: CardinalitySourceIdentity,
1076 ) -> Result<CardinalityGenerationHeader, InternalError> {
1077 if self.cardinality_generation_header()? != Some(current) {
1078 return Err(InternalError::store_corruption());
1079 }
1080 if current.validate_source(source).is_ok() {
1081 return Err(InternalError::store_invariant());
1082 }
1083 if let Some(cursor) = self.cardinality_build_cursor()? {
1084 cursor.validate_header(current)?;
1085 }
1086 let next = CardinalityGenerationHeader::new(
1087 current.generation().checked_next()?,
1088 CardinalityGenerationState::Building,
1089 current.slot().alternate(),
1090 source,
1091 );
1092 let encoded = RawSchemaSnapshot::from_encoded_control_record(next.encode());
1093 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1094 return Err(InternalError::store_invariant());
1095 };
1096 canonical.insert(RawSchemaKey::from_cardinality_generation_header(), encoded);
1097 canonical.remove(&RawSchemaKey::from_cardinality_build_cursor());
1098 Ok(next)
1099 }
1100
1101 pub(in crate::db) fn publish_empty_cardinality_generation(
1103 &mut self,
1104 candidate: &EmptyCardinalityReadyCandidate,
1105 ) -> Result<CardinalityGenerationHeader, InternalError> {
1106 if !self.cardinality_storage_is_pristine()? {
1107 return Err(InternalError::store_corruption());
1108 }
1109 let ready = CardinalityGenerationHeader::new(
1110 CardinalityGenerationId::INITIAL,
1111 CardinalityGenerationState::Ready,
1112 CardinalityCountSlot::A,
1113 candidate.source(),
1114 );
1115 let encoded = RawSchemaSnapshot::from_encoded_control_record(ready.encode());
1116 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1117 return Err(InternalError::store_invariant());
1118 };
1119 canonical.insert(RawSchemaKey::from_cardinality_generation_header(), encoded);
1120 Ok(ready)
1121 }
1122
1123 pub(in crate::db) fn publish_ready_cardinality_generation(
1125 &mut self,
1126 candidate: &CardinalityReadyCandidate,
1127 source: CardinalitySourceIdentity,
1128 ) -> Result<CardinalityGenerationHeader, InternalError> {
1129 let building = candidate.header();
1130 candidate.cursor().validate_header(building)?;
1131 if building.state() != CardinalityGenerationState::Building
1132 || building.validate_source(source).is_err()
1133 || self.cardinality_generation_header()? != Some(building)
1134 || self.cardinality_build_cursor()?.as_ref() != Some(candidate.cursor())
1135 {
1136 return Err(InternalError::store_corruption());
1137 }
1138 let ready = CardinalityGenerationHeader::new(
1139 building.generation(),
1140 CardinalityGenerationState::Ready,
1141 building.slot(),
1142 building.source(),
1143 );
1144 let encoded = RawSchemaSnapshot::from_encoded_control_record(ready.encode());
1145 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1146 return Err(InternalError::store_invariant());
1147 };
1148 canonical.insert(RawSchemaKey::from_cardinality_generation_header(), encoded);
1149 canonical.remove(&RawSchemaKey::from_cardinality_build_cursor());
1150 Ok(ready)
1151 }
1152
1153 pub(in crate::db) fn clear_cardinality_count_slot_page(
1159 &mut self,
1160 header: CardinalityGenerationHeader,
1161 initial_cursor: &CardinalityBuildCursor,
1162 limit: usize,
1163 ) -> Result<bool, InternalError> {
1164 if limit == 0 {
1165 return Err(InternalError::store_invariant());
1166 }
1167 initial_cursor.validate_header(header)?;
1168 let encoded_cursor = initial_cursor.encode()?;
1169 if self.cardinality_generation_header()? != Some(header)
1170 || self.cardinality_build_cursor()?.is_some()
1171 {
1172 return Err(InternalError::store_corruption());
1173 }
1174 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1175 return Err(InternalError::store_invariant());
1176 };
1177 let bounds = RawSchemaKey::cardinality_count_range_bounds(header.slot());
1178 let collect_limit = limit
1179 .checked_add(1)
1180 .ok_or_else(InternalError::store_unsupported)?;
1181 let mut keys = Vec::new();
1182 keys.try_reserve_exact(collect_limit)
1183 .map_err(|_| InternalError::store_unsupported())?;
1184 for entry in canonical.range(bounds).take(collect_limit) {
1185 keys.push(*entry.key());
1186 }
1187 let has_more = keys.len() > limit;
1188 for key in keys.into_iter().take(limit) {
1189 canonical.remove(&key);
1190 }
1191 if !has_more {
1192 canonical.insert(
1193 RawSchemaKey::from_cardinality_build_cursor(),
1194 RawSchemaSnapshot::from_encoded_control_record(encoded_cursor),
1195 );
1196 }
1197 Ok(has_more)
1198 }
1199
1200 pub(in crate::db) fn prepare_cardinality_count_increments(
1202 &self,
1203 slot: CardinalityCountSlot,
1204 generation: CardinalityGenerationId,
1205 increments: &[(CardinalityCountDigest, u64)],
1206 ) -> Result<PreparedCardinalityCountWrites, InternalError> {
1207 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1208 return Err(InternalError::store_invariant());
1209 }
1210 let mut entries = Vec::new();
1211 entries
1212 .try_reserve_exact(increments.len())
1213 .map_err(|_| InternalError::store_unsupported())?;
1214 let mut physical_keys = StdBTreeMap::new();
1215 let mut new_count_keys = 0_u64;
1216 for (digest, increment) in increments {
1217 if *increment == 0 {
1218 return Err(InternalError::store_invariant());
1219 }
1220 let key = RawSchemaKey::from_cardinality_count(slot, *digest);
1221 if let Some(previous_digest) = physical_keys.insert(key, *digest) {
1222 return Err(if previous_digest == *digest {
1223 InternalError::store_invariant()
1224 } else {
1225 InternalError::store_corruption()
1226 });
1227 }
1228 let current = self.get_canonical_raw_value(&key)?;
1229 let count = if let Some(raw) = current {
1230 let record = CardinalityCountRecord::decode(raw.as_bytes())?;
1231 record
1232 .validate_identity(generation, *digest)
1233 .map_err(|_| InternalError::store_corruption())?
1234 .checked_add(*increment)
1235 .ok_or_else(InternalError::store_unsupported)?
1236 } else {
1237 new_count_keys = new_count_keys
1238 .checked_add(1)
1239 .ok_or_else(InternalError::store_unsupported)?;
1240 *increment
1241 };
1242 let record = CardinalityCountRecord::new(generation, *digest, count)?;
1243 entries.push((
1244 key,
1245 RawSchemaSnapshot::from_encoded_control_record(record.encode().to_vec()),
1246 ));
1247 }
1248 Ok(PreparedCardinalityCountWrites {
1249 slot,
1250 generation,
1251 entries,
1252 new_count_keys,
1253 })
1254 }
1255
1256 pub(in crate::db) fn prepare_cardinality_build_page(
1258 &self,
1259 header: CardinalityGenerationHeader,
1260 current_cursor: &CardinalityBuildCursor,
1261 counts: PreparedCardinalityCountWrites,
1262 next_cursor: &CardinalityBuildCursor,
1263 ) -> Result<PreparedCardinalityBuildPage, InternalError> {
1264 current_cursor.validate_header(header)?;
1265 next_cursor.validate_header(header)?;
1266 if counts.slot != header.slot() || counts.generation != header.generation() {
1267 return Err(InternalError::store_invariant());
1268 }
1269 if self.cardinality_generation_header()? != Some(header)
1270 || self.cardinality_build_cursor()?.as_ref() != Some(current_cursor)
1271 {
1272 return Err(InternalError::store_corruption());
1273 }
1274 let encoded_cursor = next_cursor.encode()?;
1275 Ok(PreparedCardinalityBuildPage {
1276 count_entries: counts.entries,
1277 cursor: (
1278 RawSchemaKey::from_cardinality_build_cursor(),
1279 RawSchemaSnapshot::from_encoded_control_record(encoded_cursor),
1280 ),
1281 })
1282 }
1283
1284 pub(in crate::db) fn apply_prepared_cardinality_build_page(
1286 &mut self,
1287 prepared: PreparedCardinalityBuildPage,
1288 ) -> Result<(), InternalError> {
1289 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1290 return Err(InternalError::store_invariant());
1291 };
1292 for (key, value) in prepared.count_entries {
1293 canonical.insert(key, value);
1294 }
1295 canonical.insert(prepared.cursor.0, prepared.cursor.1);
1296 Ok(())
1297 }
1298
1299 pub(in crate::db) fn cardinality_count(
1301 &self,
1302 slot: CardinalityCountSlot,
1303 generation: CardinalityGenerationId,
1304 digest: CardinalityCountDigest,
1305 ) -> Result<Option<u64>, InternalError> {
1306 let key = RawSchemaKey::from_cardinality_count(slot, digest);
1307 self.get_canonical_raw_value(&key)?
1308 .map(|raw| {
1309 CardinalityCountRecord::decode(raw.as_bytes())?
1310 .validate_identity(generation, digest)
1311 .map_err(|_| InternalError::store_corruption())
1312 })
1313 .transpose()
1314 }
1315
1316 pub(in crate::db) fn prepare_cardinality_maintenance(
1318 &self,
1319 current: CardinalityGenerationHeader,
1320 current_source: CardinalitySourceIdentity,
1321 next_source: CardinalitySourceIdentity,
1322 changes: &[(CardinalityCountDigest, i64)],
1323 ) -> Result<PreparedCardinalityMaintenance, InternalError> {
1324 if current.state() != CardinalityGenerationState::Ready
1325 || current.validate_source(current_source).is_err()
1326 || self.cardinality_generation_header()? != Some(current)
1327 || self.cardinality_build_cursor()?.is_some()
1328 {
1329 return Err(InternalError::store_corruption());
1330 }
1331 let next = CardinalityGenerationHeader::new(
1332 current.generation(),
1333 CardinalityGenerationState::Ready,
1334 current.slot(),
1335 next_source,
1336 );
1337 let mut count_entries = Vec::new();
1338 count_entries
1339 .try_reserve_exact(changes.len())
1340 .map_err(|_| InternalError::store_unsupported())?;
1341 let mut physical_keys = StdBTreeMap::new();
1342 for (digest, delta) in changes {
1343 if *delta == 0 {
1344 return Err(InternalError::store_invariant());
1345 }
1346 let key = RawSchemaKey::from_cardinality_count(current.slot(), *digest);
1347 if let Some(previous_digest) = physical_keys.insert(key, *digest) {
1348 return Err(if previous_digest == *digest {
1349 InternalError::store_invariant()
1350 } else {
1351 InternalError::store_corruption()
1352 });
1353 }
1354 let base = self
1355 .cardinality_count(current.slot(), current.generation(), *digest)?
1356 .unwrap_or(0);
1357 let count = if *delta > 0 {
1358 base.checked_add(
1359 u64::try_from(*delta).map_err(|_| InternalError::store_invariant())?,
1360 )
1361 } else {
1362 base.checked_sub(delta.unsigned_abs())
1363 }
1364 .ok_or_else(InternalError::store_corruption)?;
1365 let value = if count == 0 {
1366 None
1367 } else {
1368 Some(RawSchemaSnapshot::from_encoded_control_record(
1369 CardinalityCountRecord::new(current.generation(), *digest, count)?
1370 .encode()
1371 .to_vec(),
1372 ))
1373 };
1374 count_entries.push((key, value));
1375 }
1376 Ok(PreparedCardinalityMaintenance {
1377 count_entries,
1378 header: (
1379 RawSchemaKey::from_cardinality_generation_header(),
1380 RawSchemaSnapshot::from_encoded_control_record(next.encode()),
1381 ),
1382 })
1383 }
1384
1385 pub(in crate::db) fn apply_prepared_cardinality_maintenance(
1387 &mut self,
1388 prepared: PreparedCardinalityMaintenance,
1389 ) -> Result<(), InternalError> {
1390 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1391 return Err(InternalError::store_invariant());
1392 };
1393 for (key, value) in prepared.count_entries {
1394 if let Some(value) = value {
1395 canonical.insert(key, value);
1396 } else {
1397 canonical.remove(&key);
1398 }
1399 }
1400 canonical.insert(prepared.header.0, prepared.header.1);
1401 Ok(())
1402 }
1403
1404 pub(in crate::db) fn cardinality_count_slot_is_empty(
1405 &self,
1406 slot: CardinalityCountSlot,
1407 ) -> Result<bool, InternalError> {
1408 let SchemaStoreBackend::Journaled { canonical, .. } = &self.backend else {
1409 return Err(InternalError::store_invariant());
1410 };
1411 Ok(canonical
1412 .range(RawSchemaKey::cardinality_count_range_bounds(slot))
1413 .next()
1414 .is_none())
1415 }
1416
1417 pub(in crate::db) fn preflight_fold_recovered_journal(&self) -> Result<(), InternalError> {
1419 match self.backend {
1420 SchemaStoreBackend::Journaled { .. } => Ok(()),
1421 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
1422 }
1423 }
1424
1425 fn prepare_identity_state_transition(
1426 &self,
1427 incarnation: DatabaseIncarnationId,
1428 candidate: &CandidateSchemaRevision,
1429 view: IdentityStateStorageView,
1430 ) -> Result<IdentityStateTransition, InternalError> {
1431 let current = match view {
1432 IdentityStateStorageView::Effective => self
1433 .current_accepted_schema_bundle_ref()?
1434 .as_ref()
1435 .map(|bundle| (*bundle).clone()),
1436 IdentityStateStorageView::Canonical => {
1437 self.current_canonical_accepted_schema_bundle()?
1438 }
1439 };
1440 let inventory = self.identity_state_inventory(view)?;
1441 prepare_identity_state_transition(
1442 incarnation,
1443 current.as_ref(),
1444 candidate.bundle(),
1445 inventory,
1446 )
1447 }
1448
1449 fn validate_identity_state_closure(
1450 &self,
1451 bundle: &AcceptedSchemaRevisionBundle,
1452 ) -> Result<(), InternalError> {
1453 let inventory = self.identity_state_inventory(IdentityStateStorageView::Effective)?;
1454 validate_identity_state_closure(bundle, &inventory)
1455 }
1456
1457 pub(in crate::db) fn identity_statement_cursor(
1459 &self,
1460 database_incarnation_id: DatabaseIncarnationId,
1461 entity_tag: EntityTag,
1462 field_id: FieldId,
1463 accepted_kind: &AcceptedFieldKind,
1464 ) -> Result<IdentityStatementCursor, InternalError> {
1465 let key = RawSchemaKey::from_identity_state(entity_tag, field_id);
1466 let raw = self
1467 .get_raw_snapshot(&key)
1468 .ok_or_else(InternalError::identity_state_corruption)?;
1469 let state = decode_identity_state(raw.as_bytes())?;
1470 let owner = state.owner();
1471 if owner.database_incarnation_id() != database_incarnation_id
1472 || owner.entity_tag() != entity_tag
1473 || owner.field_id() != field_id
1474 || state.accepted_kind() != accepted_kind
1475 || state.lifecycle() != IdentityStateLifecycle::Active
1476 {
1477 return Err(InternalError::identity_state_corruption());
1478 }
1479 IdentityStatementCursor::from_active_state(&state)
1480 }
1481
1482 pub(in crate::db) fn identity_high_water_for_integrity(
1484 &self,
1485 database_incarnation_id: DatabaseIncarnationId,
1486 entity_tag: EntityTag,
1487 field_id: FieldId,
1488 accepted_kind: &AcceptedFieldKind,
1489 ) -> Result<u128, InternalError> {
1490 let key = RawSchemaKey::from_identity_state(entity_tag, field_id);
1491 let raw = self
1492 .get_raw_snapshot(&key)
1493 .ok_or_else(InternalError::identity_state_corruption)?;
1494 let state = decode_identity_state(raw.as_bytes())?;
1495 let owner = state.owner();
1496 if owner.database_incarnation_id() != database_incarnation_id
1497 || owner.entity_tag() != entity_tag
1498 || owner.field_id() != field_id
1499 || state.accepted_kind() != accepted_kind
1500 || state.lifecycle() != IdentityStateLifecycle::Active
1501 {
1502 return Err(InternalError::identity_state_corruption());
1503 }
1504 Ok(state.materialized_high_water())
1505 }
1506
1507 pub(in crate::db) fn preflight_identity_range_advance(
1509 &self,
1510 range: IdentityRangeAdvance,
1511 ) -> Result<(), InternalError> {
1512 let state =
1513 self.identity_state_for_owner(range.owner(), IdentityStateStorageView::Effective)?;
1514 if state.lifecycle() != IdentityStateLifecycle::Active
1515 || range.new_high_water()
1516 > identity_kind_maximum(state.accepted_kind())
1517 .ok_or_else(InternalError::identity_state_corruption)?
1518 {
1519 return Err(InternalError::identity_state_corruption());
1520 }
1521 if state.materialized_high_water() != range.expected_high_water() {
1522 return Err(InternalError::identity_state_conflict());
1523 }
1524 Ok(())
1525 }
1526
1527 pub(in crate::db) fn apply_identity_range_advance(
1529 &mut self,
1530 range: IdentityRangeAdvance,
1531 advance_id: IdentityAdvanceId,
1532 ) -> Result<(), InternalError> {
1533 self.apply_identity_range_advance_to(
1534 range,
1535 advance_id,
1536 IdentityStateStorageView::Effective,
1537 IdentityStateWriteTarget::Materialized,
1538 )
1539 }
1540
1541 pub(in crate::db) fn fold_identity_range_advance(
1543 &mut self,
1544 range: IdentityRangeAdvance,
1545 advance_id: IdentityAdvanceId,
1546 ) -> Result<(), InternalError> {
1547 self.apply_identity_range_advance_to(
1548 range,
1549 advance_id,
1550 IdentityStateStorageView::Canonical,
1551 IdentityStateWriteTarget::Canonical,
1552 )
1553 }
1554
1555 pub(in crate::db) fn preflight_fold_identity_range_advance(
1557 &self,
1558 range: IdentityRangeAdvance,
1559 advance_id: IdentityAdvanceId,
1560 ) -> Result<(), InternalError> {
1561 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1562 return Err(InternalError::store_invariant());
1563 }
1564 let state =
1565 self.identity_state_for_owner(range.owner(), IdentityStateStorageView::Canonical)?;
1566 let advanced = state.apply_range_advance(range, advance_id)?;
1567 let _encoded = encode_identity_state(&advanced)?;
1568 Ok(())
1569 }
1570
1571 pub(in crate::db) fn verify_identity_range_advance(
1573 &self,
1574 range: IdentityRangeAdvance,
1575 advance_id: IdentityAdvanceId,
1576 ) -> Result<(), InternalError> {
1577 let state =
1578 self.identity_state_for_owner(range.owner(), IdentityStateStorageView::Effective)?;
1579 if state.materialized_high_water() != range.new_high_water()
1580 || state.last_applied_advance() != Some(advance_id)
1581 {
1582 return Err(InternalError::recovery_effect_verification_failed());
1583 }
1584 Ok(())
1585 }
1586
1587 pub(in crate::db) fn identity_range_commit_state(
1589 &self,
1590 range: IdentityRangeAdvance,
1591 advance_id: IdentityAdvanceId,
1592 canonical: bool,
1593 ) -> Result<IdentityRangeCommitState, InternalError> {
1594 let view = if canonical {
1595 IdentityStateStorageView::Canonical
1596 } else {
1597 IdentityStateStorageView::Effective
1598 };
1599 self.identity_state_for_owner(range.owner(), view)?
1600 .range_commit_state(range, advance_id)
1601 }
1602
1603 pub(in crate::db) fn identity_state_inventory_for_integrity(
1606 &self,
1607 incarnation: DatabaseIncarnationId,
1608 ) -> Result<Vec<IdentityState>, InternalError> {
1609 let has_accepted_bundle = self.current_accepted_schema_bundle_ref()?.is_some();
1610 let inventory = self.identity_state_inventory(IdentityStateStorageView::Effective)?;
1611 if !has_accepted_bundle && !inventory.is_empty() {
1612 return Err(InternalError::identity_state_corruption());
1613 }
1614 if inventory
1615 .values()
1616 .any(|state| state.owner().database_incarnation_id() != incarnation)
1617 {
1618 return Err(InternalError::identity_state_corruption());
1619 }
1620 Ok(inventory.into_values().collect())
1621 }
1622
1623 fn identity_state_for_owner(
1624 &self,
1625 owner: crate::db::schema::identity_state::IdentityStateOwner,
1626 view: IdentityStateStorageView,
1627 ) -> Result<IdentityState, InternalError> {
1628 let key = RawSchemaKey::from_identity_state(owner.entity_tag(), owner.field_id());
1629 let raw = match view {
1630 IdentityStateStorageView::Effective => self.get_raw_snapshot(&key),
1631 IdentityStateStorageView::Canonical => self.get_canonical_raw_value(&key)?,
1632 }
1633 .ok_or_else(InternalError::identity_state_corruption)?;
1634 let state = decode_identity_state(raw.as_bytes())?;
1635 if state.owner() != owner {
1636 return Err(InternalError::identity_state_corruption());
1637 }
1638 Ok(state)
1639 }
1640
1641 fn apply_identity_range_advance_to(
1642 &mut self,
1643 range: IdentityRangeAdvance,
1644 advance_id: IdentityAdvanceId,
1645 view: IdentityStateStorageView,
1646 target: IdentityStateWriteTarget,
1647 ) -> Result<(), InternalError> {
1648 let state = self.identity_state_for_owner(range.owner(), view)?;
1649 let advanced = state.apply_range_advance(range, advance_id)?;
1650 let key = RawSchemaKey::from_identity_state(
1651 advanced.owner().entity_tag(),
1652 advanced.owner().field_id(),
1653 );
1654 let bytes = encode_identity_state(&advanced)?;
1655 match target {
1656 IdentityStateWriteTarget::Materialized => {
1657 self.insert_raw_snapshot(
1658 key,
1659 RawSchemaSnapshot::from_encoded_control_record(bytes),
1660 );
1661 }
1662 IdentityStateWriteTarget::Canonical => {
1663 self.insert_canonical_raw_value(key, bytes)?;
1664 }
1665 IdentityStateWriteTarget::Durable => {
1666 return Err(InternalError::store_invariant());
1667 }
1668 }
1669 Ok(())
1670 }
1671
1672 fn identity_state_inventory(
1673 &self,
1674 view: IdentityStateStorageView,
1675 ) -> Result<IdentityStateInventory, InternalError> {
1676 let bounds = RawSchemaKey::all_identity_state_range_bounds();
1677 let mut inventory = StdBTreeMap::new();
1678 let mut collect = |key: &RawSchemaKey,
1679 raw: &RawSchemaSnapshot|
1680 -> Result<SchemaStoreVisit, InternalError> {
1681 if inventory.len() >= MAX_IDENTITY_STATE_RECORDS_PER_DATABASE {
1682 return Err(InternalError::identity_state_corruption());
1683 }
1684 let state = decode_identity_state(raw.as_bytes())?;
1685 let state_key = (key.entity_tag(), FieldId::new(key.version()));
1686 if !key.is_identity_state()
1687 || state.owner().entity_tag() != state_key.0
1688 || state.owner().field_id() != state_key.1
1689 || inventory.insert(state_key, state).is_some()
1690 {
1691 return Err(InternalError::identity_state_corruption());
1692 }
1693 Ok(SchemaStoreVisit::Continue)
1694 };
1695
1696 match (&self.backend, view) {
1697 (SchemaStoreBackend::Heap(map), IdentityStateStorageView::Effective) => {
1698 for (key, raw) in map.range((bounds.0, bounds.1)) {
1699 collect(key, raw)?;
1700 }
1701 }
1702 (
1703 SchemaStoreBackend::Journaled {
1704 canonical,
1705 live,
1706 tombstones,
1707 ..
1708 },
1709 IdentityStateStorageView::Effective,
1710 ) => Self::visit_journaled_raw_snapshot_range(
1711 canonical,
1712 live,
1713 tombstones,
1714 bounds,
1715 Direction::Asc,
1716 &mut collect,
1717 )?,
1718 (
1719 SchemaStoreBackend::Journaled { canonical, .. },
1720 IdentityStateStorageView::Canonical,
1721 ) => {
1722 for entry in canonical.range((bounds.0, bounds.1)) {
1723 collect(entry.key(), &entry.value())?;
1724 }
1725 }
1726 (SchemaStoreBackend::Heap(_), IdentityStateStorageView::Canonical) => {
1727 return Err(InternalError::store_invariant());
1728 }
1729 }
1730
1731 Ok(inventory)
1732 }
1733
1734 fn apply_identity_state_transition(
1735 &mut self,
1736 transition: IdentityStateTransition,
1737 target: IdentityStateWriteTarget,
1738 ) -> Result<(), InternalError> {
1739 for state in transition.into_updates() {
1740 let key = RawSchemaKey::from_identity_state(
1741 state.owner().entity_tag(),
1742 state.owner().field_id(),
1743 );
1744 let bytes = encode_identity_state(&state)?;
1745 match target {
1746 IdentityStateWriteTarget::Durable => {
1747 self.insert_durable_raw_value(key, bytes);
1748 }
1749 IdentityStateWriteTarget::Materialized => {
1750 self.insert_raw_snapshot(
1751 key,
1752 RawSchemaSnapshot::from_encoded_control_record(bytes),
1753 );
1754 }
1755 IdentityStateWriteTarget::Canonical => {
1756 self.insert_canonical_raw_value(key, bytes)?;
1757 }
1758 }
1759 }
1760 Ok(())
1761 }
1762
1763 pub(in crate::db) fn current_canonical_accepted_schema_bundle(
1764 &self,
1765 ) -> Result<Option<AcceptedSchemaRevisionBundle>, InternalError> {
1766 self.current_canonical_accepted_schema_authority()
1767 .map(|authority| authority.map(|(_, bundle)| bundle))
1768 }
1769
1770 pub(in crate::db) fn current_canonical_accepted_schema_authority(
1772 &self,
1773 ) -> Result<Option<(AcceptedSchemaRootSelection, AcceptedSchemaRevisionBundle)>, InternalError>
1774 {
1775 let Some(selection) = self.current_canonical_accepted_schema_root()? else {
1776 return Ok(None);
1777 };
1778 let bundle_key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
1779 let raw = self
1780 .get_canonical_raw_value(&bundle_key)?
1781 .ok_or_else(InternalError::store_corruption)?;
1782 let bundle =
1783 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes())?;
1784 Ok(Some((selection, bundle)))
1785 }
1786
1787 pub(in crate::db) fn current_canonical_accepted_schema_root(
1789 &self,
1790 ) -> Result<Option<AcceptedSchemaRootSelection>, InternalError> {
1791 let first = self.canonical_root_slot_bytes(0)?;
1792 let second = self.canonical_root_slot_bytes(1)?;
1793 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])
1794 }
1795
1796 pub(in crate::db) fn current_effective_and_canonical_accepted_schema_roots(
1798 &self,
1799 ) -> Result<
1800 (
1801 Option<AcceptedSchemaRootSelection>,
1802 Option<AcceptedSchemaRootSelection>,
1803 ),
1804 InternalError,
1805 > {
1806 let SchemaStoreBackend::Journaled {
1807 canonical,
1808 live,
1809 tombstones,
1810 ..
1811 } = &self.backend
1812 else {
1813 return Err(InternalError::store_invariant());
1814 };
1815 let first_key = RawSchemaKey::from_accepted_root_slot(0)?;
1816 let second_key = RawSchemaKey::from_accepted_root_slot(1)?;
1817 let mut canonical_first = None;
1818 let mut canonical_second = None;
1819 for entry in canonical.range(first_key..=second_key) {
1820 if *entry.key() == first_key {
1821 canonical_first = Some(entry.value().clone());
1822 } else if *entry.key() == second_key {
1823 canonical_second = Some(entry.value().clone());
1824 } else {
1825 return Err(InternalError::store_corruption());
1826 }
1827 }
1828 let effective_first = if tombstones.contains(&first_key) {
1829 None
1830 } else {
1831 live.get(&first_key)
1832 .cloned()
1833 .or_else(|| canonical_first.clone())
1834 };
1835 let effective_second = if tombstones.contains(&second_key) {
1836 None
1837 } else {
1838 live.get(&second_key)
1839 .cloned()
1840 .or_else(|| canonical_second.clone())
1841 };
1842 let effective_first = effective_first.map(RawSchemaSnapshot::into_bytes);
1843 let effective_second = effective_second.map(RawSchemaSnapshot::into_bytes);
1844 let canonical_first = canonical_first.map(RawSchemaSnapshot::into_bytes);
1845 let canonical_second = canonical_second.map(RawSchemaSnapshot::into_bytes);
1846 Ok((
1847 select_current_accepted_schema_root([
1848 effective_first.as_deref(),
1849 effective_second.as_deref(),
1850 ])?,
1851 select_current_accepted_schema_root([
1852 canonical_first.as_deref(),
1853 canonical_second.as_deref(),
1854 ])?,
1855 ))
1856 }
1857
1858 pub(in crate::db) fn insert_persisted_snapshot(
1860 &mut self,
1861 entity: EntityTag,
1862 snapshot: &PersistedSchemaSnapshot,
1863 ) -> Result<(), InternalError> {
1864 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
1865 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1866 let _ = self.insert_raw_snapshot(key, raw_snapshot);
1867
1868 Ok(())
1869 }
1870
1871 pub(in crate::db) fn constraint_validation_job(
1873 &self,
1874 entity: EntityTag,
1875 constraint_id: ConstraintId,
1876 ) -> Result<Option<ConstraintValidationJob>, InternalError> {
1877 let key = RawSchemaKey::from_constraint_validation_job(entity, constraint_id);
1878 self.get_raw_snapshot(&key)
1879 .map(|raw| decode_constraint_validation_job(raw.as_bytes()))
1880 .transpose()
1881 }
1882
1883 pub(in crate::db) fn apply_constraint_validation_job(
1885 &mut self,
1886 job: &ConstraintValidationJob,
1887 ) -> Result<(), InternalError> {
1888 let key =
1889 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id());
1890 let bytes = encode_constraint_validation_job(job)?;
1891 let _ =
1892 self.insert_raw_snapshot(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
1893 Ok(())
1894 }
1895
1896 #[expect(
1898 clippy::unnecessary_wraps,
1899 reason = "marker apply operations share one fallible callback contract"
1900 )]
1901 pub(in crate::db) fn apply_constraint_validation_job_removal(
1902 &mut self,
1903 entity: EntityTag,
1904 constraint_id: ConstraintId,
1905 ) -> Result<(), InternalError> {
1906 let key = RawSchemaKey::from_constraint_validation_job(entity, constraint_id);
1907 match &mut self.backend {
1908 SchemaStoreBackend::Heap(map) => {
1909 map.remove(&key);
1910 }
1911 SchemaStoreBackend::Journaled {
1912 live, tombstones, ..
1913 } => {
1914 live.remove(&key);
1915 tombstones.insert(key);
1916 }
1917 }
1918 Ok(())
1919 }
1920
1921 pub(in crate::db) fn fold_constraint_validation_job(
1923 &mut self,
1924 job: &ConstraintValidationJob,
1925 ) -> Result<(), InternalError> {
1926 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1927 return Err(InternalError::store_invariant());
1928 };
1929 let key =
1930 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id());
1931 let bytes = encode_constraint_validation_job(job)?;
1932 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
1933 Ok(())
1934 }
1935
1936 pub(in crate::db) fn preflight_fold_constraint_validation_job(
1938 &self,
1939 job: &ConstraintValidationJob,
1940 ) -> Result<(), InternalError> {
1941 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1942 return Err(InternalError::store_invariant());
1943 }
1944 let _encoded = encode_constraint_validation_job(job)?;
1945 Ok(())
1946 }
1947
1948 pub(in crate::db) fn fold_constraint_validation_job_removal(
1950 &mut self,
1951 entity: EntityTag,
1952 constraint_id: ConstraintId,
1953 ) -> Result<(), InternalError> {
1954 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1955 return Err(InternalError::store_invariant());
1956 };
1957 canonical.remove(&RawSchemaKey::from_constraint_validation_job(
1958 entity,
1959 constraint_id,
1960 ));
1961 Ok(())
1962 }
1963
1964 pub(in crate::db) fn preflight_fold_constraint_validation_job_removal(
1966 &self,
1967 ) -> Result<(), InternalError> {
1968 match self.backend {
1969 SchemaStoreBackend::Journaled { .. } => Ok(()),
1970 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
1971 }
1972 }
1973
1974 pub(in crate::db) fn reset_journaled_live_projection(&mut self) -> Result<(), InternalError> {
1977 let SchemaStoreBackend::Journaled {
1978 live,
1979 tombstones,
1980 positions,
1981 ..
1982 } = &mut self.backend
1983 else {
1984 return Err(InternalError::store_invariant());
1985 };
1986
1987 live.clear();
1988 tombstones.clear();
1989 positions.clear();
1990 self.accepted_bundle_cache.get_mut().take();
1991
1992 Ok(())
1993 }
1994
1995 pub(in crate::db) fn prepare_positioned_journal_batch_publication(
1997 &self,
1998 incarnation: DatabaseIncarnationId,
1999 batch: &JournalBatch,
2000 position: JournalOverlayPosition,
2001 ) -> Result<PreparedSchemaPositionPublication, InternalError> {
2002 let SchemaStoreBackend::Journaled { positions, .. } = &self.backend else {
2003 return Err(InternalError::store_invariant());
2004 };
2005 let keys = self.positioned_journal_batch_keys(
2006 incarnation,
2007 batch,
2008 IdentityStateStorageView::Effective,
2009 )?;
2010 for key in &keys {
2011 positions.preflight_publish(key, position)?;
2012 }
2013 Ok(PreparedSchemaPositionPublication {
2014 keys: keys.into_iter().collect(),
2015 position,
2016 })
2017 }
2018
2019 pub(in crate::db) fn prepare_positioned_journal_batch_retirement(
2021 &self,
2022 incarnation: DatabaseIncarnationId,
2023 batch: &JournalBatch,
2024 position: JournalOverlayPosition,
2025 ) -> Result<PreparedSchemaPositionRetirement, InternalError> {
2026 let SchemaStoreBackend::Journaled { positions, .. } = &self.backend else {
2027 return Err(InternalError::store_invariant());
2028 };
2029 let keys = self.positioned_journal_batch_keys(
2030 incarnation,
2031 batch,
2032 IdentityStateStorageView::Canonical,
2033 )?;
2034 let entries = keys
2035 .into_iter()
2036 .map(|key| {
2037 positions
2038 .preflight_retirement(&key, position)
2039 .map(|retirement| (key, retirement))
2040 })
2041 .collect::<Result<Vec<_>, _>>()?;
2042 Ok(PreparedSchemaPositionRetirement { entries })
2043 }
2044
2045 pub(in crate::db) fn publish_prepared_journal_batch_positions(
2047 &mut self,
2048 prepared: PreparedSchemaPositionPublication,
2049 ) {
2050 let SchemaStoreBackend::Journaled { positions, .. } = &mut self.backend else {
2051 debug_assert!(
2052 false,
2053 "preflighted schema positions require a journaled store"
2054 );
2055 return;
2056 };
2057 for key in prepared.keys {
2058 positions.publish_preflighted(key, prepared.position);
2059 }
2060 }
2061
2062 pub(in crate::db) fn apply_prepared_journal_batch_retirement(
2064 &mut self,
2065 prepared: PreparedSchemaPositionRetirement,
2066 ) {
2067 for (key, retirement) in prepared.entries {
2068 if retirement != PositionedOverlayRetirement::Exact {
2069 continue;
2070 }
2071 self.invalidate_accepted_bundle_cache_for_key(key);
2072 let SchemaStoreBackend::Journaled {
2073 live,
2074 tombstones,
2075 positions,
2076 ..
2077 } = &mut self.backend
2078 else {
2079 debug_assert!(
2080 false,
2081 "preflighted schema retirement requires a journaled store"
2082 );
2083 return;
2084 };
2085 live.remove(&key);
2086 tombstones.remove(&key);
2087 positions.retire_preflighted(&key, retirement);
2088 }
2089 }
2090
2091 #[cfg(test)]
2092 fn publish_positioned_journal_entry(
2093 &mut self,
2094 key: RawSchemaKey,
2095 snapshot: Option<RawSchemaSnapshot>,
2096 position: JournalOverlayPosition,
2097 ) -> Result<Option<RawSchemaSnapshot>, InternalError> {
2098 let SchemaStoreBackend::Journaled { positions, .. } = &self.backend else {
2099 return Err(InternalError::store_invariant());
2100 };
2101 positions.preflight_publish(&key, position)?;
2102 self.invalidate_accepted_bundle_cache_for_key(key);
2103 let SchemaStoreBackend::Journaled {
2104 canonical,
2105 live,
2106 tombstones,
2107 positions,
2108 } = &mut self.backend
2109 else {
2110 return Err(InternalError::store_invariant());
2111 };
2112 let previous = if tombstones.contains(&key) {
2113 None
2114 } else {
2115 live.get(&key).cloned().or_else(|| canonical.get(&key))
2116 };
2117 if let Some(snapshot) = snapshot {
2118 tombstones.remove(&key);
2119 live.insert(key, snapshot);
2120 } else {
2121 live.remove(&key);
2122 tombstones.insert(key);
2123 }
2124 positions.publish_preflighted(key, position);
2125 Ok(previous)
2126 }
2127
2128 #[cfg(test)]
2129 fn retire_positioned_journal_effect(
2130 &mut self,
2131 key: RawSchemaKey,
2132 position: JournalOverlayPosition,
2133 ) -> Result<PositionedOverlayRetirement, InternalError> {
2134 let SchemaStoreBackend::Journaled { positions, .. } = &self.backend else {
2135 return Err(InternalError::store_invariant());
2136 };
2137 let retirement = positions.preflight_retirement(&key, position)?;
2138 let prepared = PreparedSchemaPositionRetirement {
2139 entries: vec![(key, retirement)],
2140 };
2141 self.apply_prepared_journal_batch_retirement(prepared);
2142 Ok(retirement)
2143 }
2144
2145 pub(in crate::db) fn fold_persisted_snapshot(
2147 &mut self,
2148 entity: EntityTag,
2149 snapshot: &PersistedSchemaSnapshot,
2150 ) -> Result<(), InternalError> {
2151 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
2152 return Err(InternalError::store_invariant());
2153 };
2154
2155 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
2156 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2157 canonical.insert(key, raw_snapshot);
2158
2159 Ok(())
2160 }
2161
2162 pub(in crate::db) fn preflight_fold_persisted_snapshot(
2164 &self,
2165 snapshot: &PersistedSchemaSnapshot,
2166 ) -> Result<(), InternalError> {
2167 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
2168 return Err(InternalError::store_invariant());
2169 }
2170 let _encoded = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2171 Ok(())
2172 }
2173
2174 pub(in crate::db) fn current_accepted_schema_root(
2176 &self,
2177 ) -> Result<Option<AcceptedSchemaRootSelection>, InternalError> {
2178 let first = self.accepted_root_slot_bytes(0)?;
2179 let second = self.accepted_root_slot_bytes(1)?;
2180 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])
2181 }
2182
2183 pub(in crate::db) fn current_accepted_schema_bundle(
2185 &self,
2186 ) -> Result<Option<AcceptedSchemaRevisionBundle>, InternalError> {
2187 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2188 return Ok(None);
2189 };
2190 self.validate_constraint_validation_job_closure(&bundle)?;
2191 Ok(Some(bundle.clone()))
2192 }
2193
2194 pub(in crate::db) fn current_accepted_runtime_entities(
2196 &self,
2197 registered_store_path: &'static str,
2198 ) -> Result<Vec<AcceptedRuntimeEntity>, InternalError> {
2199 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2200 return Ok(Vec::new());
2201 };
2202 if bundle.store_path() != registered_store_path {
2203 return Err(InternalError::store_corruption());
2204 }
2205
2206 bundle
2207 .entity_snapshots()
2208 .iter()
2209 .map(|(entity_tag, snapshot)| {
2210 AcceptedRuntimeEntity::from_accepted_snapshot(
2211 &bundle,
2212 *entity_tag,
2213 snapshot,
2214 registered_store_path,
2215 )
2216 })
2217 .collect()
2218 }
2219
2220 pub(in crate::db) fn current_accepted_runtime_entity_for_tag(
2222 &self,
2223 registered_store_path: &'static str,
2224 entity_tag: EntityTag,
2225 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
2226 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2227 return Ok(None);
2228 };
2229 if bundle.store_path() != registered_store_path {
2230 return Err(InternalError::store_corruption());
2231 }
2232 let Some(snapshot) = bundle.entity_snapshots().get(&entity_tag) else {
2233 return Ok(None);
2234 };
2235
2236 AcceptedRuntimeEntity::from_accepted_snapshot(
2237 &bundle,
2238 entity_tag,
2239 snapshot,
2240 registered_store_path,
2241 )
2242 .map(Some)
2243 }
2244
2245 pub(in crate::db) fn current_canonical_accepted_runtime_entity_for_tag(
2247 &self,
2248 registered_store_path: &'static str,
2249 entity_tag: EntityTag,
2250 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
2251 let Some(bundle) = self.current_canonical_accepted_schema_bundle()? else {
2252 return Ok(None);
2253 };
2254 if bundle.store_path() != registered_store_path {
2255 return Err(InternalError::store_corruption());
2256 }
2257 let Some(snapshot) = bundle.entity_snapshots().get(&entity_tag) else {
2258 return Ok(None);
2259 };
2260
2261 AcceptedRuntimeEntity::from_accepted_snapshot(
2262 &bundle,
2263 entity_tag,
2264 snapshot,
2265 registered_store_path,
2266 )
2267 .map(Some)
2268 }
2269
2270 pub(in crate::db) fn current_accepted_runtime_entity_for_path(
2272 &self,
2273 registered_store_path: &'static str,
2274 entity_path: &str,
2275 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
2276 self.current_accepted_runtime_entity_matching(registered_store_path, |snapshot_path, _| {
2277 snapshot_path == entity_path
2278 })
2279 }
2280
2281 pub(in crate::db) fn current_canonical_accepted_runtime_entity_for_path(
2283 &self,
2284 registered_store_path: &'static str,
2285 entity_path: &str,
2286 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
2287 let Some(bundle) = self.current_canonical_accepted_schema_bundle()? else {
2288 return Ok(None);
2289 };
2290 if bundle.store_path() != registered_store_path {
2291 return Err(InternalError::store_corruption());
2292 }
2293
2294 let mut matched = None;
2295 for (entity_tag, snapshot) in bundle.entity_snapshots() {
2296 if snapshot.entity_path() != entity_path {
2297 continue;
2298 }
2299 let entity = AcceptedRuntimeEntity::from_accepted_snapshot(
2300 &bundle,
2301 *entity_tag,
2302 snapshot,
2303 registered_store_path,
2304 )?;
2305 if matched.replace(entity).is_some() {
2306 return Err(InternalError::store_corruption());
2307 }
2308 }
2309
2310 Ok(matched)
2311 }
2312
2313 #[cfg(test)]
2315 pub(in crate::db) fn current_accepted_runtime_entity_for_name(
2316 &self,
2317 registered_store_path: &'static str,
2318 entity_name: &str,
2319 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
2320 self.current_accepted_runtime_entity_matching(registered_store_path, |_, snapshot_name| {
2321 snapshot_name == entity_name
2322 })
2323 }
2324
2325 fn current_accepted_runtime_entity_matching(
2326 &self,
2327 registered_store_path: &'static str,
2328 mut predicate: impl FnMut(&str, &str) -> bool,
2329 ) -> Result<Option<AcceptedRuntimeEntity>, InternalError> {
2330 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2331 return Ok(None);
2332 };
2333 if bundle.store_path() != registered_store_path {
2334 return Err(InternalError::store_corruption());
2335 }
2336
2337 let mut matched = None;
2338 for (entity_tag, snapshot) in bundle.entity_snapshots() {
2339 if !predicate(snapshot.entity_path(), snapshot.entity_name()) {
2340 continue;
2341 }
2342 let entity = AcceptedRuntimeEntity::from_accepted_snapshot(
2343 &bundle,
2344 *entity_tag,
2345 snapshot,
2346 registered_store_path,
2347 )?;
2348 if matched.replace(entity).is_some() {
2349 return Err(InternalError::store_corruption());
2350 }
2351 }
2352
2353 Ok(matched)
2354 }
2355
2356 pub(in crate::db) fn current_accepted_schema_revision(
2358 &self,
2359 ) -> Result<Option<AcceptedSchemaRevision>, InternalError> {
2360 Ok(self
2361 .current_accepted_schema_root()?
2362 .map(|selection| selection.root().revision()))
2363 }
2364
2365 pub(in crate::db) fn pending_relation_activation_for_target(
2371 &self,
2372 target_path: &str,
2373 ) -> Result<Option<PendingRelationActivationDeleteBarrier>, InternalError> {
2374 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2375 return Ok(None);
2376 };
2377 for (entity_tag, snapshot) in bundle.entity_snapshots() {
2378 let Some(candidate) = snapshot
2379 .candidate_relations()
2380 .iter()
2381 .find(|candidate| candidate.target_path() == target_path)
2382 else {
2383 continue;
2384 };
2385 let activation = snapshot
2386 .constraint_activations()
2387 .iter()
2388 .find(|activation| {
2389 matches!(
2390 activation.kind(),
2391 ConstraintActivationKind::Relation { relation_id }
2392 if *relation_id == candidate.id()
2393 )
2394 })
2395 .ok_or_else(InternalError::store_corruption)?;
2396 return Ok(Some(PendingRelationActivationDeleteBarrier {
2397 accepted_schema_fingerprint:
2398 accepted_schema_cache_fingerprint_for_persisted_snapshot(snapshot)?,
2399 source_entity_tag: *entity_tag,
2400 constraint_id: activation.id(),
2401 }));
2402 }
2403
2404 Ok(None)
2405 }
2406
2407 pub(in crate::db) fn entity_has_relation_to_target(
2409 &self,
2410 source_entity: EntityTag,
2411 target_path: &str,
2412 ) -> Result<bool, InternalError> {
2413 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
2414 return Ok(false);
2415 };
2416 let Some(snapshot) = bundle.entity_snapshots().get(&source_entity) else {
2417 return Ok(false);
2418 };
2419
2420 Ok(snapshot
2421 .relations()
2422 .iter()
2423 .any(|relation| relation.target_path() == target_path))
2424 }
2425
2426 pub(in crate::db) fn validate_live_activation_transition(
2428 &self,
2429 candidate: &AcceptedSchemaRevisionBundle,
2430 ) -> Result<(), InternalError> {
2431 let Some(current) = self.current_accepted_schema_bundle()? else {
2432 return Ok(());
2433 };
2434 Self::validate_activation_transition_from(¤t, candidate)
2435 }
2436
2437 pub(in crate::db) fn validate_canonical_activation_transition(
2439 &self,
2440 candidate: &AcceptedSchemaRevisionBundle,
2441 ) -> Result<(), InternalError> {
2442 let Some(current) = self.current_canonical_accepted_schema_bundle()? else {
2443 return Ok(());
2444 };
2445 Self::validate_activation_transition_from(¤t, candidate)
2446 }
2447
2448 fn validate_activation_transition_from(
2449 current: &AcceptedSchemaRevisionBundle,
2450 candidate: &AcceptedSchemaRevisionBundle,
2451 ) -> Result<(), InternalError> {
2452 for (entity_tag, before) in current.entity_snapshots() {
2453 if before.constraint_activations().is_empty() {
2454 continue;
2455 }
2456 let after = candidate
2457 .entity_snapshots()
2458 .get(entity_tag)
2459 .ok_or_else(InternalError::store_invariant)?;
2460 if before == after {
2461 continue;
2462 }
2463 let expected_shape = before
2464 .clone()
2465 .with_constraint_catalog(after.constraint_catalog().clone());
2466 let catalog_only_transition = expected_shape == *after
2467 && before
2468 .constraint_catalog()
2469 .permits_live_activation_transition_to(after.constraint_catalog());
2470 let sql_row_local_abort_with_version =
2471 before.constraint_activations().iter().any(|activation| {
2472 activation.origin() == ConstraintOrigin::SqlDdl
2473 && matches!(
2474 activation.kind(),
2475 ConstraintActivationKind::Check { .. }
2476 | ConstraintActivationKind::NotNull { .. }
2477 )
2478 && before.version().get().checked_add(1) == Some(after.version().get())
2479 && before
2480 .constraint_catalog()
2481 .clone()
2482 .with_aborted_activation(activation.id())
2483 .is_ok_and(|catalog| catalog == *after.constraint_catalog())
2484 && before
2485 .clone()
2486 .with_constraint_catalog(after.constraint_catalog().clone())
2487 .with_schema_version(after.version())
2488 == *after
2489 });
2490 let sql_unique_abort_with_version =
2491 before.constraint_activations().iter().any(|activation| {
2492 activation.origin() == ConstraintOrigin::SqlDdl
2493 && matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
2494 && before.version().get().checked_add(1) == Some(after.version().get())
2495 && before
2496 .with_aborted_unique_activation(activation.id(), after.version())
2497 .is_ok_and(|expected| expected == *after)
2498 });
2499 let not_null_promotion = before.constraint_activations().iter().any(|activation| {
2500 matches!(activation.kind(), ConstraintActivationKind::NotNull { .. })
2501 && before
2502 .with_promoted_not_null_activation(activation.id(), after.version())
2503 .is_ok_and(|expected| expected == *after)
2504 });
2505 let unique_promotion = before.constraint_activations().iter().any(|activation| {
2506 matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
2507 && before
2508 .with_promoted_unique_activation(activation.id(), after.version())
2509 .is_ok_and(|expected| expected == *after)
2510 });
2511 let relation_promotion = before.constraint_activations().iter().any(|activation| {
2512 matches!(activation.kind(), ConstraintActivationKind::Relation { .. })
2513 && before
2514 .with_promoted_relation_activation(activation.id(), after.version())
2515 .is_ok_and(|expected| expected == *after)
2516 });
2517 if !catalog_only_transition
2518 && !sql_row_local_abort_with_version
2519 && !sql_unique_abort_with_version
2520 && !not_null_promotion
2521 && !unique_promotion
2522 && !relation_promotion
2523 {
2524 return Err(InternalError::store_invariant());
2525 }
2526 }
2527 Ok(())
2528 }
2529
2530 pub(in crate::db) fn validate_constraint_validation_job_closure(
2532 &self,
2533 bundle: &AcceptedSchemaRevisionBundle,
2534 ) -> Result<(), InternalError> {
2535 self.validate_constraint_validation_job_closure_with_change(bundle, None, None)
2536 }
2537
2538 pub(in crate::db) fn validate_constraint_validation_job_closure_with_change(
2541 &self,
2542 bundle: &AcceptedSchemaRevisionBundle,
2543 replacement: Option<&ConstraintValidationJob>,
2544 removal: Option<(EntityTag, ConstraintId)>,
2545 ) -> Result<(), InternalError> {
2546 self.validate_constraint_validation_job_closure_with_change_in_view(
2547 bundle,
2548 replacement,
2549 removal,
2550 IdentityStateStorageView::Effective,
2551 )
2552 }
2553
2554 pub(in crate::db) fn validate_canonical_constraint_validation_job_closure_with_change(
2556 &self,
2557 bundle: &AcceptedSchemaRevisionBundle,
2558 replacement: Option<&ConstraintValidationJob>,
2559 removal: Option<(EntityTag, ConstraintId)>,
2560 ) -> Result<(), InternalError> {
2561 self.validate_constraint_validation_job_closure_with_change_in_view(
2562 bundle,
2563 replacement,
2564 removal,
2565 IdentityStateStorageView::Canonical,
2566 )
2567 }
2568
2569 fn validate_constraint_validation_job_closure_with_change_in_view(
2570 &self,
2571 bundle: &AcceptedSchemaRevisionBundle,
2572 replacement: Option<&ConstraintValidationJob>,
2573 removal: Option<(EntityTag, ConstraintId)>,
2574 view: IdentityStateStorageView,
2575 ) -> Result<(), InternalError> {
2576 if replacement.is_some() && removal.is_some() {
2577 return Err(InternalError::store_invariant());
2578 }
2579 let replacement_key = replacement.map(|job| {
2580 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id())
2581 });
2582 let removal_key = removal.map(|(entity_tag, constraint_id)| {
2583 RawSchemaKey::from_constraint_validation_job(entity_tag, constraint_id)
2584 });
2585 let mut expected = BTreeSet::new();
2586 for (entity_tag, snapshot) in bundle.entity_snapshots() {
2587 for activation in snapshot.constraint_activations() {
2588 let key =
2589 RawSchemaKey::from_constraint_validation_job(*entity_tag, activation.id());
2590 match activation.state() {
2591 ConstraintActivationState::EnforcingNewWrites => {
2592 if self
2593 .constraint_validation_job_after_change(
2594 key,
2595 replacement,
2596 replacement_key,
2597 removal_key,
2598 view,
2599 )?
2600 .is_some()
2601 {
2602 return Err(InternalError::store_corruption());
2603 }
2604 }
2605 ConstraintActivationState::Validating => {
2606 let job = self
2607 .constraint_validation_job_after_change(
2608 key,
2609 replacement,
2610 replacement_key,
2611 removal_key,
2612 view,
2613 )?
2614 .ok_or_else(InternalError::store_corruption)?;
2615 if job.entity_tag() != *entity_tag
2616 || job.entity_path() != snapshot.entity_path()
2617 {
2618 return Err(InternalError::store_corruption());
2619 }
2620 job.validate(Some(activation))?;
2621 expected.insert(key);
2622 }
2623 }
2624 }
2625 }
2626
2627 self.visit_constraint_validation_jobs_in_view(view, |key, raw| {
2628 if removal_key == Some(*key) || replacement_key == Some(*key) {
2629 return Ok(SchemaStoreVisit::Continue);
2630 }
2631 if !expected.contains(key) {
2632 return Err(InternalError::store_corruption());
2633 }
2634 let job = decode_constraint_validation_job(raw.as_bytes())?;
2635 if job.entity_tag() != key.entity_tag()
2636 || key.constraint_id() != Some(job.constraint_id())
2637 {
2638 return Err(InternalError::store_corruption());
2639 }
2640 Ok(SchemaStoreVisit::Continue)
2641 })?;
2642
2643 if let Some(key) = replacement_key
2644 && !expected.contains(&key)
2645 {
2646 return Err(InternalError::store_corruption());
2647 }
2648 if let Some(key) = removal_key
2649 && expected.contains(&key)
2650 {
2651 return Err(InternalError::store_corruption());
2652 }
2653
2654 Ok(())
2655 }
2656
2657 fn constraint_validation_job_after_change(
2658 &self,
2659 key: RawSchemaKey,
2660 replacement: Option<&ConstraintValidationJob>,
2661 replacement_key: Option<RawSchemaKey>,
2662 removal_key: Option<RawSchemaKey>,
2663 view: IdentityStateStorageView,
2664 ) -> Result<Option<ConstraintValidationJob>, InternalError> {
2665 if removal_key == Some(key) {
2666 return Ok(None);
2667 }
2668 if replacement_key == Some(key) {
2669 return Ok(replacement.cloned());
2670 }
2671 let raw = match view {
2672 IdentityStateStorageView::Effective => self.get_raw_snapshot(&key),
2673 IdentityStateStorageView::Canonical => self.get_canonical_raw_value(&key)?,
2674 };
2675 raw.map(|raw| decode_constraint_validation_job(raw.as_bytes()))
2676 .transpose()
2677 }
2678
2679 pub(in crate::db) fn current_accepted_schema_authority_matches(
2682 &self,
2683 expected: &AcceptedSchemaAuthority,
2684 ) -> Result<bool, InternalError> {
2685 let Some(store_scope) = self.accepted_catalog_scope.get() else {
2686 return Ok(false);
2687 };
2688
2689 if let Some(cached) = self
2692 .accepted_bundle_cache
2693 .try_borrow()
2694 .map_err(|_| InternalError::store_invariant())?
2695 .as_ref()
2696 {
2697 let root = cached.selection.root();
2698 return Ok(expected.matches_store_root(
2699 store_scope,
2700 root.revision(),
2701 root.fingerprint(),
2702 ));
2703 }
2704
2705 let Some(selection) = self.current_accepted_schema_root()? else {
2706 return Ok(false);
2707 };
2708 let root = selection.root();
2709
2710 Ok(expected.matches_store_root(store_scope, root.revision(), root.fingerprint()))
2711 }
2712
2713 pub(in crate::db) fn publish_accepted_schema_candidate(
2719 &mut self,
2720 incarnation: DatabaseIncarnationId,
2721 expected_revision: AcceptedSchemaRevision,
2722 candidate: &CandidateSchemaRevision,
2723 ) -> Result<(), InternalError> {
2724 let identity_transition = self.prepare_identity_state_transition(
2725 incarnation,
2726 candidate,
2727 IdentityStateStorageView::Effective,
2728 )?;
2729 if self.current_root_matches_candidate(candidate)? {
2730 if !identity_transition.is_empty() {
2731 return Err(InternalError::identity_state_corruption());
2732 }
2733 let selection = self
2734 .current_accepted_schema_root()?
2735 .ok_or_else(InternalError::store_corruption)?;
2736 self.retain_durable_candidate_entries(candidate, selection.slot())?;
2737 return Ok(());
2738 }
2739 let first = self.accepted_root_slot_bytes(0)?;
2740 let second = self.accepted_root_slot_bytes(1)?;
2741 prepare_accepted_schema_root_publication(
2742 [first.as_deref(), second.as_deref()],
2743 expected_revision,
2744 candidate,
2745 )
2746 .map_err(map_schema_publication_error)?;
2747
2748 self.insert_durable_candidate_snapshots(candidate)?;
2749 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2750 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
2751 let persisted_bundle = self
2752 .get_raw_snapshot(&bundle_key)
2753 .ok_or_else(InternalError::store_corruption)?;
2754 let _verified = decode_verified_accepted_schema_revision_bundle(
2755 candidate.root(),
2756 persisted_bundle.as_bytes(),
2757 )?;
2758 self.apply_identity_state_transition(
2759 identity_transition,
2760 IdentityStateWriteTarget::Durable,
2761 )?;
2762
2763 let first = self.accepted_root_slot_bytes(0)?;
2766 let second = self.accepted_root_slot_bytes(1)?;
2767 let publication = prepare_accepted_schema_root_publication(
2768 [first.as_deref(), second.as_deref()],
2769 expected_revision,
2770 candidate,
2771 )
2772 .map_err(map_schema_publication_error)?;
2773 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
2774 self.insert_durable_raw_value(root_key, publication.encoded_root().to_vec());
2775
2776 let selected = self
2777 .current_accepted_schema_root()?
2778 .ok_or_else(InternalError::store_corruption)?;
2779 if selected.root() != candidate.root() {
2780 return Err(InternalError::store_corruption());
2781 }
2782 self.retain_durable_candidate_entries(candidate, selected.slot())?;
2783 Ok(())
2784 }
2785
2786 pub(in crate::db) fn restore_live_accepted_schema_checkpoint(
2789 &mut self,
2790 incarnation: DatabaseIncarnationId,
2791 candidate: &CandidateSchemaRevision,
2792 checkpoint_identity_states: &IdentityStateInventory,
2793 ) -> Result<(), InternalError> {
2794 if !matches!(self.backend, SchemaStoreBackend::Heap(_)) {
2795 return Err(InternalError::store_invariant());
2796 }
2797 let checkpoint_validation = prepare_identity_state_transition(
2798 incarnation,
2799 Some(candidate.bundle()),
2800 candidate.bundle(),
2801 checkpoint_identity_states.clone(),
2802 )?;
2803 if !checkpoint_validation.is_empty() {
2804 return Err(InternalError::identity_state_corruption());
2805 }
2806 if self.current_root_matches_candidate(candidate)? {
2807 for state in checkpoint_identity_states.values() {
2808 let key = RawSchemaKey::from_identity_state(
2809 state.owner().entity_tag(),
2810 state.owner().field_id(),
2811 );
2812 self.insert_durable_raw_value(key, encode_identity_state(state)?);
2813 }
2814 if self.identity_state_inventory(IdentityStateStorageView::Effective)?
2815 != *checkpoint_identity_states
2816 {
2817 return Err(InternalError::identity_state_corruption());
2818 }
2819 let selection = self
2820 .current_accepted_schema_root()?
2821 .ok_or_else(InternalError::store_corruption)?;
2822 self.retain_durable_candidate_entries(candidate, selection.slot())?;
2823 return Ok(());
2824 }
2825 if self.current_accepted_schema_root()?.is_some()
2826 || !self
2827 .identity_state_inventory(IdentityStateStorageView::Effective)?
2828 .is_empty()
2829 {
2830 return Err(InternalError::store_corruption());
2831 }
2832
2833 self.insert_durable_candidate_snapshots(candidate)?;
2834 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2835 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
2836 for state in checkpoint_identity_states.values() {
2837 let key = RawSchemaKey::from_identity_state(
2838 state.owner().entity_tag(),
2839 state.owner().field_id(),
2840 );
2841 self.insert_durable_raw_value(key, encode_identity_state(state)?);
2842 }
2843 let root_key = RawSchemaKey::from_accepted_root_slot(0)?;
2844 self.insert_durable_raw_value(root_key, candidate.encoded_root().to_vec());
2845
2846 let selected = self
2847 .current_accepted_schema_root()?
2848 .ok_or_else(InternalError::store_corruption)?;
2849 if selected.root() != candidate.root() {
2850 return Err(InternalError::store_corruption());
2851 }
2852 self.retain_durable_candidate_entries(candidate, selected.slot())?;
2853 Ok(())
2854 }
2855
2856 pub(in crate::db) fn preflight_accepted_schema_candidate(
2863 &self,
2864 incarnation: DatabaseIncarnationId,
2865 expected_revision: AcceptedSchemaRevision,
2866 candidate: &CandidateSchemaRevision,
2867 ) -> Result<bool, InternalError> {
2868 let identity_transition = self.prepare_identity_state_transition(
2869 incarnation,
2870 candidate,
2871 IdentityStateStorageView::Effective,
2872 )?;
2873 if self.current_root_matches_candidate(candidate)? {
2874 if !identity_transition.is_empty() {
2875 return Err(InternalError::identity_state_corruption());
2876 }
2877 return Ok(true);
2878 }
2879 let first = self.accepted_root_slot_bytes(0)?;
2880 let second = self.accepted_root_slot_bytes(1)?;
2881 prepare_accepted_schema_root_publication(
2882 [first.as_deref(), second.as_deref()],
2883 expected_revision,
2884 candidate,
2885 )
2886 .map_err(map_schema_publication_error)?;
2887
2888 Ok(false)
2889 }
2890
2891 pub(in crate::db) fn preflight_fold_journaled_accepted_schema_candidate(
2893 &self,
2894 incarnation: DatabaseIncarnationId,
2895 expected_revision: AcceptedSchemaRevision,
2896 candidate: &CandidateSchemaRevision,
2897 ) -> Result<(), InternalError> {
2898 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
2899 return Err(InternalError::store_invariant());
2900 }
2901 let identity_transition = self.prepare_identity_state_transition(
2902 incarnation,
2903 candidate,
2904 IdentityStateStorageView::Canonical,
2905 )?;
2906 let candidate_is_current = self.canonical_root_matches_candidate(candidate)?;
2907 if candidate_is_current && !identity_transition.is_empty() {
2908 return Err(InternalError::identity_state_corruption());
2909 }
2910 for state in identity_transition.into_updates() {
2911 let _encoded = encode_identity_state(&state)?;
2912 }
2913 for snapshot in candidate.bundle().entity_snapshots().values() {
2914 let _encoded = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
2915 }
2916
2917 let first = self.canonical_root_slot_bytes(0)?;
2918 let second = self.canonical_root_slot_bytes(1)?;
2919 let root_slot = if candidate_is_current {
2920 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
2921 .ok_or_else(InternalError::store_corruption)?
2922 .slot()
2923 } else {
2924 prepare_accepted_schema_root_publication(
2925 [first.as_deref(), second.as_deref()],
2926 expected_revision,
2927 candidate,
2928 )
2929 .map_err(map_schema_publication_error)?
2930 .target_slot()
2931 };
2932 let _retained = Self::candidate_entry_keys(candidate, root_slot)?;
2933 Ok(())
2934 }
2935
2936 pub(in crate::db) fn projected_identity_state_count(
2938 &self,
2939 incarnation: DatabaseIncarnationId,
2940 candidate: &CandidateSchemaRevision,
2941 ) -> Result<usize, InternalError> {
2942 Ok(self
2943 .prepare_identity_state_transition(
2944 incarnation,
2945 candidate,
2946 IdentityStateStorageView::Effective,
2947 )?
2948 .projected_inventory_len())
2949 }
2950
2951 pub(in crate::db) fn apply_journaled_accepted_schema_candidate(
2953 &mut self,
2954 incarnation: DatabaseIncarnationId,
2955 expected_revision: AcceptedSchemaRevision,
2956 candidate: &CandidateSchemaRevision,
2957 ) -> Result<(), InternalError> {
2958 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
2959 return Err(InternalError::store_invariant());
2960 }
2961 let identity_transition = self.prepare_identity_state_transition(
2962 incarnation,
2963 candidate,
2964 IdentityStateStorageView::Effective,
2965 )?;
2966 if self.current_root_matches_candidate(candidate)? {
2967 if !identity_transition.is_empty() {
2968 return Err(InternalError::identity_state_corruption());
2969 }
2970 let selection = self
2971 .current_accepted_schema_root()?
2972 .ok_or_else(InternalError::store_corruption)?;
2973 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
2974 return Ok(());
2975 }
2976
2977 let first = self.accepted_root_slot_bytes(0)?;
2978 let second = self.accepted_root_slot_bytes(1)?;
2979 prepare_accepted_schema_root_publication(
2980 [first.as_deref(), second.as_deref()],
2981 expected_revision,
2982 candidate,
2983 )
2984 .map_err(map_schema_publication_error)?;
2985
2986 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
2987 self.insert_persisted_snapshot(*entity_tag, snapshot)?;
2988 }
2989 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
2990 self.insert_raw_snapshot(
2991 bundle_key,
2992 RawSchemaSnapshot::from_encoded_control_record(candidate.encoded_bundle().to_vec()),
2993 );
2994 let persisted_bundle = self
2995 .get_raw_snapshot(&bundle_key)
2996 .ok_or_else(InternalError::store_corruption)?;
2997 let _verified = decode_verified_accepted_schema_revision_bundle(
2998 candidate.root(),
2999 persisted_bundle.as_bytes(),
3000 )?;
3001 self.apply_identity_state_transition(
3002 identity_transition,
3003 IdentityStateWriteTarget::Materialized,
3004 )?;
3005
3006 let first = self.accepted_root_slot_bytes(0)?;
3007 let second = self.accepted_root_slot_bytes(1)?;
3008 let publication = prepare_accepted_schema_root_publication(
3009 [first.as_deref(), second.as_deref()],
3010 expected_revision,
3011 candidate,
3012 )
3013 .map_err(map_schema_publication_error)?;
3014 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
3015 self.insert_raw_snapshot(
3016 root_key,
3017 RawSchemaSnapshot::from_encoded_control_record(publication.encoded_root().to_vec()),
3018 );
3019
3020 if !self.current_root_matches_candidate(candidate)? {
3021 return Err(InternalError::store_corruption());
3022 }
3023 let selection = self
3024 .current_accepted_schema_root()?
3025 .ok_or_else(InternalError::store_corruption)?;
3026 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
3027 Ok(())
3028 }
3029
3030 pub(in crate::db) fn fold_journaled_accepted_schema_candidate(
3032 &mut self,
3033 incarnation: DatabaseIncarnationId,
3034 expected_revision: AcceptedSchemaRevision,
3035 candidate: &CandidateSchemaRevision,
3036 ) -> Result<(), InternalError> {
3037 let identity_transition = self.prepare_identity_state_transition(
3038 incarnation,
3039 candidate,
3040 IdentityStateStorageView::Canonical,
3041 )?;
3042 if self.canonical_root_matches_candidate(candidate)? {
3043 if !identity_transition.is_empty() {
3044 return Err(InternalError::identity_state_corruption());
3045 }
3046 let first = self.canonical_root_slot_bytes(0)?;
3047 let second = self.canonical_root_slot_bytes(1)?;
3048 let selection =
3049 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
3050 .ok_or_else(InternalError::store_corruption)?;
3051 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
3052 return Ok(());
3053 }
3054
3055 let first = self.canonical_root_slot_bytes(0)?;
3056 let second = self.canonical_root_slot_bytes(1)?;
3057 prepare_accepted_schema_root_publication(
3058 [first.as_deref(), second.as_deref()],
3059 expected_revision,
3060 candidate,
3061 )
3062 .map_err(map_schema_publication_error)?;
3063
3064 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
3065 self.fold_persisted_snapshot(*entity_tag, snapshot)?;
3066 }
3067 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
3068 self.insert_canonical_raw_value(bundle_key, candidate.encoded_bundle().to_vec())?;
3069 let persisted_bundle = self
3070 .get_canonical_raw_value(&bundle_key)?
3071 .ok_or_else(InternalError::store_corruption)?;
3072 let _verified = decode_verified_accepted_schema_revision_bundle(
3073 candidate.root(),
3074 persisted_bundle.as_bytes(),
3075 )?;
3076 self.apply_identity_state_transition(
3077 identity_transition,
3078 IdentityStateWriteTarget::Canonical,
3079 )?;
3080
3081 let first = self.canonical_root_slot_bytes(0)?;
3082 let second = self.canonical_root_slot_bytes(1)?;
3083 let publication = prepare_accepted_schema_root_publication(
3084 [first.as_deref(), second.as_deref()],
3085 expected_revision,
3086 candidate,
3087 )
3088 .map_err(map_schema_publication_error)?;
3089 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
3090 self.insert_canonical_raw_value(root_key, publication.encoded_root().to_vec())?;
3091
3092 if !self.canonical_root_matches_candidate(candidate)? {
3093 return Err(InternalError::store_corruption());
3094 }
3095 let first = self.canonical_root_slot_bytes(0)?;
3096 let second = self.canonical_root_slot_bytes(1)?;
3097 let selection = select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
3098 .ok_or_else(InternalError::store_corruption)?;
3099 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
3100 Ok(())
3101 }
3102
3103 pub(in crate::db) fn get_persisted_snapshot(
3105 &self,
3106 entity: EntityTag,
3107 version: SchemaVersion,
3108 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
3109 let key = RawSchemaKey::from_entity_version(entity, version);
3110 self.get_raw_snapshot(&key)
3111 .map(|snapshot| snapshot.decode_persisted_snapshot())
3112 .transpose()
3113 }
3114
3115 #[cfg(test)]
3116 fn latest_staged_persisted_snapshot(
3117 &self,
3118 entity: EntityTag,
3119 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
3120 self.latest_raw_snapshots_by_entity()
3121 .remove(&entity)
3122 .map(|(_, snapshot)| snapshot.decode_persisted_snapshot())
3123 .transpose()
3124 }
3125
3126 pub(in crate::db) fn current_accepted_persisted_snapshot(
3129 &self,
3130 entity: EntityTag,
3131 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
3132 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
3133 return Ok(None);
3134 };
3135
3136 Ok(bundle.entity_snapshots().get(&entity).cloned())
3137 }
3138
3139 pub(in crate::db) fn current_accepted_catalog_selection(
3141 &self,
3142 entity: EntityTag,
3143 entity_path: &str,
3144 store_path: &'static str,
3145 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
3146 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
3147 return Ok(None);
3148 };
3149 if bundle.store_path() != store_path {
3150 return Err(InternalError::store_corruption());
3151 }
3152 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
3153 return Ok(None);
3154 };
3155 if snapshot.entity_path() != entity_path {
3156 return Err(InternalError::store_corruption());
3157 }
3158
3159 let cache = self
3160 .accepted_bundle_cache
3161 .try_borrow()
3162 .map_err(|_| InternalError::store_invariant())?;
3163 let cached = cache.as_ref().ok_or_else(InternalError::store_invariant)?;
3164 if let Some(selection) = cached
3165 .entity_selections
3166 .try_borrow()
3167 .map_err(|_| InternalError::store_invariant())?
3168 .get(&entity)
3169 .cloned()
3170 {
3171 return Ok(Some(selection));
3172 }
3173
3174 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
3175 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
3176 let identity = AcceptedCatalogIdentity::new(
3177 entity,
3178 entity_path,
3179 store_path,
3180 bundle.revision(),
3181 snapshot.version(),
3182 fingerprint,
3183 );
3184
3185 let selected = AcceptedCatalogSnapshotSelection::new(
3186 identity,
3187 cached.value_catalog.clone(),
3188 Rc::from(raw_snapshot.into_bytes()),
3189 );
3190 cached
3191 .entity_selections
3192 .try_borrow_mut()
3193 .map_err(|_| InternalError::store_invariant())?
3194 .insert(entity, selected.clone());
3195
3196 Ok(Some(selected))
3197 }
3198
3199 pub(in crate::db) fn current_canonical_accepted_catalog_selection(
3203 &self,
3204 entity: EntityTag,
3205 entity_path: &str,
3206 store_path: &'static str,
3207 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
3208 let first = self.canonical_root_slot_bytes(0)?;
3209 let second = self.canonical_root_slot_bytes(1)?;
3210 let Some(selection) =
3211 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
3212 else {
3213 return Ok(None);
3214 };
3215 let bundle_key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
3216 let raw_bundle = self
3217 .get_canonical_raw_value(&bundle_key)?
3218 .ok_or_else(InternalError::store_corruption)?;
3219 let bundle = decode_verified_accepted_schema_revision_bundle(
3220 selection.root(),
3221 raw_bundle.as_bytes(),
3222 )?;
3223 if bundle.store_path() != store_path {
3224 return Err(InternalError::store_corruption());
3225 }
3226 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
3227 return Ok(None);
3228 };
3229 if snapshot.entity_path() != entity_path {
3230 return Err(InternalError::store_corruption());
3231 }
3232
3233 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
3234 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
3235 let identity = AcceptedCatalogIdentity::new(
3236 entity,
3237 entity_path,
3238 store_path,
3239 bundle.revision(),
3240 snapshot.version(),
3241 fingerprint,
3242 );
3243
3244 Ok(Some(AcceptedCatalogSnapshotSelection::new(
3245 identity,
3246 AcceptedValueCatalogHandle::new(
3247 bundle.enum_catalog().clone(),
3248 bundle.composite_catalog().clone(),
3249 self.accepted_catalog_scope
3250 .get_or_init(AcceptedStoreCatalogScope::new)
3251 .clone(),
3252 bundle.revision(),
3253 selection.root().fingerprint(),
3254 ),
3255 Rc::from(raw_snapshot.into_bytes()),
3256 )))
3257 }
3258
3259 #[cfg(test)]
3265 pub(in crate::db) fn catalog_metadata(
3266 &self,
3267 ) -> Result<Option<SchemaStoreCatalogMetadata>, InternalError> {
3268 Ok(self
3269 .allocation_metadata()?
3270 .map(SchemaStoreAllocationMetadata::schema))
3271 }
3272
3273 pub(in crate::db) fn allocation_metadata(
3280 &self,
3281 ) -> Result<Option<SchemaStoreAllocationMetadata>, InternalError> {
3282 let latest_by_entity = self.latest_raw_snapshots_by_entity();
3283 if latest_by_entity.is_empty() {
3284 return Ok(None);
3285 }
3286
3287 Ok(Some(SchemaStoreAllocationMetadata::new(
3288 derive_data_allocation_metadata(&latest_by_entity)?,
3289 derive_index_allocation_metadata(&latest_by_entity)?,
3290 derive_schema_catalog_metadata(&latest_by_entity)?,
3291 )))
3292 }
3293
3294 fn insert_raw_snapshot(
3296 &mut self,
3297 key: RawSchemaKey,
3298 snapshot: RawSchemaSnapshot,
3299 ) -> Option<RawSchemaSnapshot> {
3300 self.invalidate_accepted_bundle_cache_for_key(key);
3301 let previous_journaled = if matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
3302 self.get_raw_snapshot_for_backend(&key)
3303 } else {
3304 None
3305 };
3306 match &mut self.backend {
3307 SchemaStoreBackend::Heap(map) => map.insert(key, snapshot),
3308 SchemaStoreBackend::Journaled {
3309 live, tombstones, ..
3310 } => {
3311 tombstones.remove(&key);
3312 live.insert(key, snapshot);
3313 previous_journaled
3314 }
3315 }
3316 }
3317
3318 #[must_use]
3320 fn get_raw_snapshot(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
3321 match &self.backend {
3322 SchemaStoreBackend::Heap(map) => map.get(key).cloned(),
3323 SchemaStoreBackend::Journaled { .. } => self.get_raw_snapshot_for_backend(key),
3324 }
3325 }
3326
3327 fn accepted_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
3328 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
3329 Ok(self
3330 .get_raw_snapshot(&key)
3331 .map(RawSchemaSnapshot::into_bytes))
3332 }
3333
3334 fn canonical_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
3335 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
3336 Ok(self
3337 .get_canonical_raw_value(&key)?
3338 .map(RawSchemaSnapshot::into_bytes))
3339 }
3340
3341 fn current_root_matches_candidate(
3342 &self,
3343 candidate: &CandidateSchemaRevision,
3344 ) -> Result<bool, InternalError> {
3345 let Some(selection) = self.current_accepted_schema_root()? else {
3346 return Ok(false);
3347 };
3348 if selection.root() != candidate.root() {
3349 return Ok(false);
3350 }
3351 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
3352 let bundle = self
3353 .get_raw_snapshot(&key)
3354 .ok_or_else(InternalError::store_corruption)?;
3355 let _verified =
3356 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
3357 Ok(true)
3358 }
3359
3360 fn canonical_root_matches_candidate(
3361 &self,
3362 candidate: &CandidateSchemaRevision,
3363 ) -> Result<bool, InternalError> {
3364 let first = self.canonical_root_slot_bytes(0)?;
3365 let second = self.canonical_root_slot_bytes(1)?;
3366 let Some(selection) =
3367 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
3368 else {
3369 return Ok(false);
3370 };
3371 if selection.root() != candidate.root() {
3372 return Ok(false);
3373 }
3374 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
3375 let bundle = self
3376 .get_canonical_raw_value(&key)?
3377 .ok_or_else(InternalError::store_corruption)?;
3378 let _verified =
3379 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
3380 Ok(true)
3381 }
3382
3383 fn get_canonical_raw_value(
3384 &self,
3385 key: &RawSchemaKey,
3386 ) -> Result<Option<RawSchemaSnapshot>, InternalError> {
3387 match &self.backend {
3388 SchemaStoreBackend::Journaled { canonical, .. } => Ok(canonical.get(key)),
3389 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
3390 }
3391 }
3392
3393 fn insert_canonical_raw_value(
3394 &mut self,
3395 key: RawSchemaKey,
3396 bytes: Vec<u8>,
3397 ) -> Result<(), InternalError> {
3398 self.invalidate_accepted_bundle_cache_for_key(key);
3399 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
3400 return Err(InternalError::store_invariant());
3401 };
3402 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
3403 Ok(())
3404 }
3405
3406 fn insert_durable_raw_value(&mut self, key: RawSchemaKey, bytes: Vec<u8>) {
3410 self.invalidate_accepted_bundle_cache_for_key(key);
3411 let value = RawSchemaSnapshot::from_encoded_control_record(bytes);
3412 match &mut self.backend {
3413 SchemaStoreBackend::Heap(map) => {
3414 map.insert(key, value);
3415 }
3416 SchemaStoreBackend::Journaled {
3417 canonical,
3418 live,
3419 tombstones,
3420 ..
3421 } => {
3422 live.remove(&key);
3423 tombstones.remove(&key);
3424 canonical.insert(key, value);
3425 }
3426 }
3427 }
3428
3429 fn invalidate_accepted_bundle_cache_for_key(&mut self, key: RawSchemaKey) {
3430 if key.is_accepted_root() {
3431 self.accepted_bundle_cache.get_mut().take();
3432 }
3433 }
3434
3435 fn insert_durable_candidate_snapshots(
3436 &mut self,
3437 candidate: &CandidateSchemaRevision,
3438 ) -> Result<(), InternalError> {
3439 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
3440 let key = RawSchemaKey::from_entity_version(*entity_tag, snapshot.version());
3441 let value = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
3442 match &mut self.backend {
3443 SchemaStoreBackend::Heap(map) => {
3444 map.insert(key, value);
3445 }
3446 SchemaStoreBackend::Journaled {
3447 canonical,
3448 live,
3449 tombstones,
3450 ..
3451 } => {
3452 live.remove(&key);
3453 tombstones.remove(&key);
3454 canonical.insert(key, value);
3455 }
3456 }
3457 }
3458 Ok(())
3459 }
3460
3461 fn candidate_entry_keys(
3462 candidate: &CandidateSchemaRevision,
3463 root_slot: usize,
3464 ) -> Result<BTreeSet<RawSchemaKey>, InternalError> {
3465 let mut keys = candidate
3466 .bundle()
3467 .entity_snapshots()
3468 .iter()
3469 .map(|(entity_tag, snapshot)| {
3470 RawSchemaKey::from_entity_version(*entity_tag, snapshot.version())
3471 })
3472 .collect::<BTreeSet<_>>();
3473 keys.insert(RawSchemaKey::from_accepted_bundle(
3474 candidate.root().bundle_key(),
3475 ));
3476 keys.insert(RawSchemaKey::from_accepted_root_slot(root_slot)?);
3477 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
3478 for activation in snapshot
3479 .constraint_activations()
3480 .iter()
3481 .filter(|activation| activation.state() == ConstraintActivationState::Validating)
3482 {
3483 keys.insert(RawSchemaKey::from_constraint_validation_job(
3484 *entity_tag,
3485 activation.id(),
3486 ));
3487 }
3488 }
3489 Ok(keys)
3490 }
3491
3492 fn positioned_candidate_effect_keys(
3493 &self,
3494 incarnation: DatabaseIncarnationId,
3495 expected_revision: AcceptedSchemaRevision,
3496 candidate: &CandidateSchemaRevision,
3497 view: IdentityStateStorageView,
3498 ) -> Result<BTreeSet<RawSchemaKey>, InternalError> {
3499 let identity_transition =
3500 self.prepare_identity_state_transition(incarnation, candidate, view)?;
3501 let (first, second, candidate_is_current) = match view {
3502 IdentityStateStorageView::Effective => (
3503 self.accepted_root_slot_bytes(0)?,
3504 self.accepted_root_slot_bytes(1)?,
3505 self.current_root_matches_candidate(candidate)?,
3506 ),
3507 IdentityStateStorageView::Canonical => (
3508 self.canonical_root_slot_bytes(0)?,
3509 self.canonical_root_slot_bytes(1)?,
3510 self.canonical_root_matches_candidate(candidate)?,
3511 ),
3512 };
3513 let root_slot = if candidate_is_current {
3514 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
3515 .ok_or_else(InternalError::store_corruption)?
3516 .slot()
3517 } else {
3518 prepare_accepted_schema_root_publication(
3519 [first.as_deref(), second.as_deref()],
3520 expected_revision,
3521 candidate,
3522 )
3523 .map_err(map_schema_publication_error)?
3524 .target_slot()
3525 };
3526 let mut keys = Self::candidate_entry_keys(candidate, root_slot)?;
3527 for state in identity_transition.into_updates() {
3528 keys.insert(RawSchemaKey::from_identity_state(
3529 state.owner().entity_tag(),
3530 state.owner().field_id(),
3531 ));
3532 }
3533
3534 let SchemaStoreBackend::Journaled {
3535 canonical, live, ..
3536 } = &self.backend
3537 else {
3538 return Err(InternalError::store_invariant());
3539 };
3540 for entry in canonical.iter() {
3541 if !keys.contains(entry.key()) && !entry.key().is_identity_state() {
3542 keys.insert(*entry.key());
3543 }
3544 }
3545 if matches!(view, IdentityStateStorageView::Effective) {
3546 for key in live.keys() {
3547 if !keys.contains(key) && !key.is_identity_state() {
3548 keys.insert(*key);
3549 }
3550 }
3551 }
3552 Ok(keys)
3553 }
3554
3555 fn positioned_journal_batch_keys(
3556 &self,
3557 incarnation: DatabaseIncarnationId,
3558 batch: &JournalBatch,
3559 view: IdentityStateStorageView,
3560 ) -> Result<BTreeSet<RawSchemaKey>, InternalError> {
3561 let mut keys = BTreeSet::new();
3562 for record in batch.records() {
3563 match record {
3564 JournalRecord::SchemaPut {
3565 schema_snapshot_bytes,
3566 ..
3567 } => {
3568 let snapshot = decode_persisted_schema_snapshot(schema_snapshot_bytes)?;
3569 let entity_tag = match view {
3570 IdentityStateStorageView::Effective => self
3571 .current_accepted_schema_bundle_ref()?
3572 .ok_or_else(InternalError::store_corruption)?
3573 .entity_snapshots()
3574 .iter()
3575 .find_map(|(entity_tag, accepted)| {
3576 (accepted.entity_path() == snapshot.entity_path())
3577 .then_some(*entity_tag)
3578 }),
3579 IdentityStateStorageView::Canonical => self
3580 .current_canonical_accepted_schema_bundle()?
3581 .ok_or_else(InternalError::store_corruption)?
3582 .entity_snapshots()
3583 .iter()
3584 .find_map(|(entity_tag, accepted)| {
3585 (accepted.entity_path() == snapshot.entity_path())
3586 .then_some(*entity_tag)
3587 }),
3588 }
3589 .ok_or_else(InternalError::store_corruption)?;
3590 keys.insert(RawSchemaKey::from_entity_version(
3591 entity_tag,
3592 snapshot.version(),
3593 ));
3594 }
3595 JournalRecord::AcceptedSchemaPublish {
3596 expected_revision,
3597 schema_bundle_bytes,
3598 schema_root_bytes,
3599 ..
3600 } => {
3601 let candidate = CandidateSchemaRevision::from_encoded(
3602 schema_bundle_bytes.clone(),
3603 schema_root_bytes.clone(),
3604 )?;
3605 keys.extend(self.positioned_candidate_effect_keys(
3606 incarnation,
3607 *expected_revision,
3608 &candidate,
3609 view,
3610 )?);
3611 }
3612 JournalRecord::ConstraintValidationJobPut {
3613 entity_tag,
3614 constraint_id,
3615 ..
3616 }
3617 | JournalRecord::ConstraintValidationJobDelete {
3618 entity_tag,
3619 constraint_id,
3620 ..
3621 } => {
3622 keys.insert(RawSchemaKey::from_constraint_validation_job(
3623 *entity_tag,
3624 *constraint_id,
3625 ));
3626 }
3627 JournalRecord::IdentityRangeAdvance { range } => {
3628 keys.insert(RawSchemaKey::from_identity_state(
3629 range.owner().entity_tag(),
3630 range.owner().field_id(),
3631 ));
3632 }
3633 JournalRecord::RowPut { .. }
3634 | JournalRecord::RowDelete { .. }
3635 | JournalRecord::AcceptedSchemaIndexDelete { .. }
3636 | JournalRecord::AcceptedSchemaIndexPut { .. }
3637 | JournalRecord::ConstraintValidationIndexPut { .. } => {}
3638 #[cfg(any(test, feature = "migration"))]
3639 JournalRecord::SchemaMigrationRowPut { .. }
3640 | JournalRecord::SchemaMigrationIndexPut { .. } => {}
3641 }
3642 }
3643 Ok(keys)
3644 }
3645
3646 fn retain_durable_candidate_entries(
3650 &mut self,
3651 candidate: &CandidateSchemaRevision,
3652 root_slot: usize,
3653 ) -> Result<(), InternalError> {
3654 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
3655 self.accepted_bundle_cache.get_mut().take();
3656 match &mut self.backend {
3657 SchemaStoreBackend::Heap(map) => {
3658 map.retain(|key, _| keep.contains(key) || key.is_identity_state());
3659 }
3660 SchemaStoreBackend::Journaled {
3661 canonical,
3662 live,
3663 tombstones,
3664 ..
3665 } => {
3666 let stale = canonical
3667 .iter()
3668 .filter_map(|entry| {
3669 (!keep.contains(entry.key()) && !entry.key().is_identity_state())
3670 .then_some(*entry.key())
3671 })
3672 .collect::<Vec<_>>();
3673 for key in stale {
3674 canonical.remove(&key);
3675 }
3676 live.retain(|key, _| keep.contains(key) || key.is_identity_state());
3677 tombstones.clear();
3678 }
3679 }
3680 Ok(())
3681 }
3682
3683 fn retain_materialized_candidate_entries(
3684 &mut self,
3685 candidate: &CandidateSchemaRevision,
3686 root_slot: usize,
3687 ) -> Result<(), InternalError> {
3688 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
3689 self.accepted_bundle_cache.get_mut().take();
3690 let SchemaStoreBackend::Journaled {
3691 canonical,
3692 live,
3693 tombstones,
3694 ..
3695 } = &mut self.backend
3696 else {
3697 return Err(InternalError::store_invariant());
3698 };
3699 live.retain(|key, _| keep.contains(key) || key.is_identity_state());
3700 let canonical_keys = canonical
3701 .iter()
3702 .map(|entry| *entry.key())
3703 .collect::<Vec<_>>();
3704 for key in canonical_keys {
3705 if keep.contains(&key) || key.is_identity_state() {
3706 tombstones.remove(&key);
3707 } else {
3708 tombstones.insert(key);
3709 }
3710 }
3711 Ok(())
3712 }
3713
3714 fn retain_canonical_candidate_entries(
3715 &mut self,
3716 candidate: &CandidateSchemaRevision,
3717 root_slot: usize,
3718 ) -> Result<(), InternalError> {
3719 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
3720 self.accepted_bundle_cache.get_mut().take();
3721 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
3722 return Err(InternalError::store_invariant());
3723 };
3724 let stale = canonical
3725 .iter()
3726 .filter_map(|entry| {
3727 (!keep.contains(entry.key()) && !entry.key().is_identity_state())
3728 .then_some(*entry.key())
3729 })
3730 .collect::<Vec<_>>();
3731 for key in stale {
3732 canonical.remove(&key);
3733 }
3734 Ok(())
3735 }
3736
3737 #[must_use]
3739 #[cfg(test)]
3740 fn contains_raw_snapshot(&self, key: &RawSchemaKey) -> bool {
3741 match &self.backend {
3742 SchemaStoreBackend::Heap(map) => map.contains_key(key),
3743 SchemaStoreBackend::Journaled { .. } => {
3744 self.get_raw_snapshot_for_backend(key).is_some()
3745 }
3746 }
3747 }
3748
3749 #[must_use]
3751 #[cfg(test)]
3752 pub(in crate::db) fn len(&self) -> u64 {
3753 match &self.backend {
3754 SchemaStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
3755 SchemaStoreBackend::Journaled { .. } => {
3756 let mut count = 0_u64;
3757 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
3758 count = count.saturating_add(1);
3759 Ok(SchemaStoreVisit::Continue)
3760 });
3761 count
3762 }
3763 }
3764 }
3765
3766 #[must_use]
3768 #[cfg(test)]
3769 pub(in crate::db) fn is_empty(&self) -> bool {
3770 match &self.backend {
3771 SchemaStoreBackend::Heap(map) => map.is_empty(),
3772 SchemaStoreBackend::Journaled { .. } => {
3773 let mut empty = true;
3774 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
3775 empty = false;
3776 Ok(SchemaStoreVisit::Stop)
3777 });
3778 empty
3779 }
3780 }
3781 }
3782
3783 #[cfg(test)]
3785 pub(in crate::db) fn clear(&mut self) {
3786 self.accepted_bundle_cache.get_mut().take();
3787 match &mut self.backend {
3788 SchemaStoreBackend::Heap(map) => map.clear(),
3789 SchemaStoreBackend::Journaled {
3790 canonical,
3791 live,
3792 tombstones,
3793 ..
3794 } => {
3795 live.clear();
3796 tombstones.clear();
3797 let keys = canonical
3798 .iter()
3799 .map(|entry| *entry.key())
3800 .collect::<Vec<_>>();
3801 for key in keys {
3802 if key.is_entity_snapshot() {
3803 tombstones.insert(key);
3804 } else {
3805 canonical.remove(&key);
3806 }
3807 }
3808 }
3809 }
3810 }
3811
3812 fn current_accepted_schema_bundle_ref(
3813 &self,
3814 ) -> Result<Option<Ref<'_, AcceptedSchemaRevisionBundle>>, InternalError> {
3815 self.current_accepted_schema_authority_ref()
3816 .map(|authority| authority.map(|(_selection, bundle)| bundle))
3817 }
3818
3819 pub(in crate::db) fn current_accepted_schema_authority_ref(
3821 &self,
3822 ) -> Result<
3823 Option<(
3824 AcceptedSchemaRootSelection,
3825 Ref<'_, AcceptedSchemaRevisionBundle>,
3826 )>,
3827 InternalError,
3828 > {
3829 let selection = self.current_accepted_schema_root()?;
3830 self.accepted_schema_authority_ref_for_selection(selection)
3831 }
3832
3833 fn accepted_schema_authority_ref_for_selection(
3834 &self,
3835 selection: Option<AcceptedSchemaRootSelection>,
3836 ) -> Result<
3837 Option<(
3838 AcceptedSchemaRootSelection,
3839 Ref<'_, AcceptedSchemaRevisionBundle>,
3840 )>,
3841 InternalError,
3842 > {
3843 let Some(selection) = selection else {
3844 self.accepted_bundle_cache
3845 .try_borrow_mut()
3846 .map_err(|_| InternalError::store_invariant())?
3847 .take();
3848 return Ok(None);
3849 };
3850
3851 let cache_matches = self
3852 .accepted_bundle_cache
3853 .try_borrow()
3854 .map_err(|_| InternalError::store_invariant())?
3855 .as_ref()
3856 .is_some_and(|cached| cached.selection == selection);
3857 if !cache_matches {
3858 let key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
3859 let raw = self
3860 .get_raw_snapshot(&key)
3861 .ok_or_else(InternalError::store_corruption)?;
3862 let bundle =
3863 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes())?;
3864 self.validate_constraint_validation_job_closure(&bundle)?;
3865 #[cfg(test)]
3866 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES
3867 .with(|misses| misses.set(misses.get().saturating_add(1)));
3868 let value_catalog = AcceptedValueCatalogHandle::new(
3869 bundle.enum_catalog().clone(),
3870 bundle.composite_catalog().clone(),
3871 self.accepted_catalog_scope
3872 .get_or_init(AcceptedStoreCatalogScope::new)
3873 .clone(),
3874 bundle.revision(),
3875 selection.root().fingerprint(),
3876 );
3877 let cardinality_domain = Rc::new(CardinalityAcceptedDomain::derive(&bundle)?);
3878 *self
3879 .accepted_bundle_cache
3880 .try_borrow_mut()
3881 .map_err(|_| InternalError::store_invariant())? = Some(AcceptedSchemaBundleCache {
3882 selection,
3883 bundle,
3884 cardinality_domain,
3885 value_catalog,
3886 entity_selections: RefCell::new(StdBTreeMap::new()),
3887 });
3888 }
3889
3890 let cache = self
3891 .accepted_bundle_cache
3892 .try_borrow()
3893 .map_err(|_| InternalError::store_invariant())?;
3894 let bundle = Ref::filter_map(cache, |cache| {
3895 cache
3896 .as_ref()
3897 .filter(|cached| cached.selection == selection)
3898 .map(|cached| &cached.bundle)
3899 })
3900 .map_err(|_| InternalError::store_invariant())?;
3901 self.validate_identity_state_closure(&bundle)?;
3902 Ok(Some((selection, bundle)))
3903 }
3904
3905 pub(in crate::db) fn accepted_cardinality_domain_for_selection(
3907 &self,
3908 selection: Option<AcceptedSchemaRootSelection>,
3909 ) -> Result<Option<(AcceptedSchemaRootSelection, Rc<CardinalityAcceptedDomain>)>, InternalError>
3910 {
3911 let Some(selection) = selection else {
3912 self.accepted_bundle_cache
3913 .try_borrow_mut()
3914 .map_err(|_| InternalError::store_invariant())?
3915 .take();
3916 return Ok(None);
3917 };
3918 let cache_matches = self
3919 .accepted_bundle_cache
3920 .try_borrow()
3921 .map_err(|_| InternalError::store_invariant())?
3922 .as_ref()
3923 .is_some_and(|cached| cached.selection == selection);
3924 if !cache_matches {
3925 let authority = self
3926 .accepted_schema_authority_ref_for_selection(Some(selection))?
3927 .ok_or_else(InternalError::store_invariant)?;
3928 drop(authority);
3929 }
3930 let cache = self
3931 .accepted_bundle_cache
3932 .try_borrow()
3933 .map_err(|_| InternalError::store_invariant())?;
3934 let domain = cache
3935 .as_ref()
3936 .filter(|cached| cached.selection == selection)
3937 .map(|cached| Rc::clone(&cached.cardinality_domain))
3938 .ok_or_else(InternalError::store_invariant)?;
3939 Ok(Some((selection, domain)))
3940 }
3941
3942 pub(in crate::db) fn cached_cardinality_domain_for_root(
3944 &self,
3945 root: CardinalityAcceptedRootIdentity,
3946 ) -> Result<Option<Rc<CardinalityAcceptedDomain>>, InternalError> {
3947 let cache = self
3948 .accepted_bundle_cache
3949 .try_borrow()
3950 .map_err(|_| InternalError::store_invariant())?;
3951 Ok(cache
3952 .as_ref()
3953 .filter(|cached| root.matches(cached.selection.root()))
3954 .map(|cached| Rc::clone(&cached.cardinality_domain)))
3955 }
3956
3957 fn latest_raw_snapshots_by_entity(
3958 &self,
3959 ) -> StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)> {
3960 let mut latest_by_entity =
3961 StdBTreeMap::<EntityTag, (SchemaVersion, RawSchemaSnapshot)>::new();
3962
3963 let _: Result<(), std::convert::Infallible> = self.visit_raw_snapshots(|key, snapshot| {
3964 let version = SchemaVersion::new(key.version());
3965 match latest_by_entity.get_mut(&key.entity_tag()) {
3966 Some((latest_version, latest_snapshot)) if version > *latest_version => {
3967 *latest_version = version;
3968 *latest_snapshot = snapshot.clone();
3969 }
3970 None => {
3971 latest_by_entity.insert(key.entity_tag(), (version, snapshot.clone()));
3972 }
3973 Some(_) => {}
3974 }
3975 Ok(SchemaStoreVisit::Continue)
3976 });
3977
3978 latest_by_entity
3979 }
3980
3981 fn visit_raw_snapshots<E>(
3984 &self,
3985 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
3986 ) -> Result<(), E> {
3987 let bounds = RawSchemaKey::all_entity_range_bounds();
3988 match &self.backend {
3989 SchemaStoreBackend::Heap(map) => {
3990 let mut visitor = visitor;
3991 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
3992 if visitor(key, snapshot)?.should_stop() {
3993 break;
3994 }
3995 }
3996 }
3997 SchemaStoreBackend::Journaled {
3998 canonical,
3999 live,
4000 tombstones,
4001 ..
4002 } => Self::visit_journaled_raw_snapshot_range(
4003 canonical,
4004 live,
4005 tombstones,
4006 bounds,
4007 Direction::Asc,
4008 visitor,
4009 )?,
4010 }
4011
4012 Ok(())
4013 }
4014
4015 fn visit_constraint_validation_jobs_in_view<E>(
4016 &self,
4017 view: IdentityStateStorageView,
4018 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
4019 ) -> Result<(), E> {
4020 let bounds = RawSchemaKey::all_constraint_validation_job_range_bounds();
4021 match (&self.backend, view) {
4022 (SchemaStoreBackend::Heap(map), _) => {
4023 let mut visitor = visitor;
4024 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
4025 if visitor(key, snapshot)?.should_stop() {
4026 break;
4027 }
4028 }
4029 }
4030 (
4031 SchemaStoreBackend::Journaled {
4032 canonical,
4033 live,
4034 tombstones,
4035 ..
4036 },
4037 IdentityStateStorageView::Effective,
4038 ) => Self::visit_journaled_raw_snapshot_range(
4039 canonical,
4040 live,
4041 tombstones,
4042 bounds,
4043 Direction::Asc,
4044 visitor,
4045 )?,
4046 (
4047 SchemaStoreBackend::Journaled { canonical, .. },
4048 IdentityStateStorageView::Canonical,
4049 ) => {
4050 let mut visitor = visitor;
4051 for entry in canonical.range((bounds.0, bounds.1)) {
4052 if visitor(entry.key(), &entry.value())?.should_stop() {
4053 break;
4054 }
4055 }
4056 }
4057 }
4058 Ok(())
4059 }
4060
4061 #[cfg(test)]
4062 #[must_use]
4063 pub(in crate::db) fn canonical_len_for_tests(&self) -> u64 {
4064 match &self.backend {
4065 SchemaStoreBackend::Journaled { canonical: map, .. } => map.len(),
4066 SchemaStoreBackend::Heap(_) => 0,
4067 }
4068 }
4069
4070 fn get_raw_snapshot_for_backend(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
4071 let SchemaStoreBackend::Journaled {
4072 canonical,
4073 live,
4074 tombstones,
4075 ..
4076 } = &self.backend
4077 else {
4078 return None;
4079 };
4080
4081 if tombstones.contains(key) {
4082 return None;
4083 }
4084 live.get(key).cloned().or_else(|| canonical.get(key))
4085 }
4086
4087 fn visit_journaled_raw_snapshot_range<E>(
4088 canonical: &StableBTreeMap<
4089 RawSchemaKey,
4090 RawSchemaSnapshot,
4091 VirtualMemory<DefaultMemoryImpl>,
4092 >,
4093 live: &StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
4094 tombstones: &BTreeSet<RawSchemaKey>,
4095 bounds: (RangeBound<RawSchemaKey>, RangeBound<RawSchemaKey>),
4096 direction: Direction,
4097 mut visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
4098 ) -> Result<(), E> {
4099 match direction {
4100 Direction::Asc => visit_ordered_overlay(
4101 canonical.range((bounds.0, bounds.1)),
4102 live.range((bounds.0, bounds.1)),
4103 Direction::Asc,
4104 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
4105 |canonical_entry| !tombstones.contains(canonical_entry.key()),
4106 |live_entry| !tombstones.contains(live_entry.0),
4107 |entry| {
4108 let visit = match entry {
4109 OrderedOverlayEntry::Canonical(canonical_entry) => {
4110 visitor(canonical_entry.key(), &canonical_entry.value())?
4111 }
4112 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
4113 };
4114 Ok(if visit.should_stop() {
4115 OrderedOverlayVisit::Stop
4116 } else {
4117 OrderedOverlayVisit::Continue
4118 })
4119 },
4120 ),
4121 Direction::Desc => visit_ordered_overlay(
4122 canonical.range((bounds.0, bounds.1)).rev(),
4123 live.range((bounds.0, bounds.1)).rev(),
4124 Direction::Desc,
4125 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
4126 |canonical_entry| !tombstones.contains(canonical_entry.key()),
4127 |live_entry| !tombstones.contains(live_entry.0),
4128 |entry| {
4129 let visit = match entry {
4130 OrderedOverlayEntry::Canonical(canonical_entry) => {
4131 visitor(canonical_entry.key(), &canonical_entry.value())?
4132 }
4133 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
4134 };
4135 Ok(if visit.should_stop() {
4136 OrderedOverlayVisit::Stop
4137 } else {
4138 OrderedOverlayVisit::Continue
4139 })
4140 },
4141 ),
4142 }
4143 }
4144}
4145
4146fn map_schema_publication_error(error: AcceptedSchemaPublicationError) -> InternalError {
4147 match error {
4148 AcceptedSchemaPublicationError::StaleSchemaRevision { .. }
4149 | AcceptedSchemaPublicationError::RevisionExhausted => InternalError::store_unsupported(),
4150 AcceptedSchemaPublicationError::InvalidCandidate => InternalError::store_invariant(),
4151 AcceptedSchemaPublicationError::CorruptRootSlots => InternalError::store_corruption(),
4152 }
4153}
4154
4155fn derive_data_allocation_metadata(
4156 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
4157) -> Result<SchemaStoreCatalogMetadata, InternalError> {
4158 let mut max_version = SchemaVersion::initial();
4159 let mut hasher = new_hash_sha256();
4160 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN);
4161
4162 for (entity, (_, snapshot)) in latest_by_entity {
4163 let persisted = snapshot.decode_persisted_snapshot()?;
4164 if persisted.version() > max_version {
4165 max_version = persisted.version();
4166 }
4167
4168 let data_projection = PersistedSchemaSnapshot::new_with_primary_key_fields_and_indexes(
4169 persisted.version(),
4170 persisted.entity_path().to_string(),
4171 persisted.entity_name().to_string(),
4172 persisted.primary_key_field_ids().to_vec(),
4173 persisted.row_layout().clone(),
4174 persisted.fields().to_vec(),
4175 Vec::new(),
4176 );
4177 let constraint_catalog = crate::db::schema::AcceptedConstraintCatalog::initial(
4178 data_projection.fields(),
4179 data_projection.indexes(),
4180 data_projection.relations(),
4181 )
4182 .map_err(|_| InternalError::store_invariant())?;
4183 let data_projection = data_projection.with_constraint_catalog(constraint_catalog);
4184 let encoded = encode_persisted_schema_snapshot(&data_projection)?;
4185
4186 write_hash_u64(&mut hasher, entity.value());
4187 write_hash_u32(&mut hasher, persisted.version().get());
4188 write_hash_len_u32(&mut hasher, encoded.len());
4189 hasher.update(encoded);
4190 }
4191
4192 Ok(finalize_schema_metadata(
4193 max_version,
4194 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
4195 hasher,
4196 latest_by_entity.len(),
4197 ))
4198}
4199
4200fn derive_index_allocation_metadata(
4201 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
4202) -> Result<SchemaStoreCatalogMetadata, InternalError> {
4203 let mut max_version = SchemaVersion::initial();
4204 let mut hasher = new_hash_sha256();
4205 write_hash_tag_u8(
4206 &mut hasher,
4207 SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN,
4208 );
4209
4210 for (entity, (_, snapshot)) in latest_by_entity {
4211 let persisted = snapshot.decode_persisted_snapshot()?;
4212 if persisted.version() > max_version {
4213 max_version = persisted.version();
4214 }
4215
4216 write_hash_u64(&mut hasher, entity.value());
4217 write_hash_u32(&mut hasher, persisted.version().get());
4218 write_hash_len_u32(&mut hasher, persisted.indexes().len());
4219 for index in persisted.indexes() {
4220 write_hash_u32(&mut hasher, u32::from(index.ordinal()));
4221 write_hash_str_u32(&mut hasher, index.name());
4222 write_hash_str_u32(&mut hasher, index.store());
4223 write_hash_tag_u8(&mut hasher, u8::from(index.unique()));
4224 write_hash_str_u32(&mut hasher, persisted_index_origin_name(index.origin()));
4225 match index.predicate_sql() {
4226 Some(predicate_sql) => {
4227 write_hash_tag_u8(&mut hasher, 1);
4228 write_hash_str_u32(&mut hasher, predicate_sql);
4229 }
4230 None => write_hash_tag_u8(&mut hasher, 0),
4231 }
4232 hash_persisted_index_key(&mut hasher, index.key());
4233 }
4234 }
4235
4236 Ok(finalize_schema_metadata(
4237 max_version,
4238 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
4239 hasher,
4240 latest_by_entity.len(),
4241 ))
4242}
4243
4244fn derive_schema_catalog_metadata(
4245 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
4246) -> Result<SchemaStoreCatalogMetadata, InternalError> {
4247 let mut max_version = SchemaVersion::initial();
4248 let mut hasher = new_hash_sha256();
4249 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN);
4250
4251 for (entity, (version, snapshot)) in latest_by_entity {
4252 let persisted = snapshot.decode_persisted_snapshot()?;
4253 if persisted.version() > max_version {
4254 max_version = persisted.version();
4255 }
4256
4257 write_hash_u64(&mut hasher, entity.value());
4258 write_hash_u32(&mut hasher, version.get());
4259 write_hash_len_u32(&mut hasher, snapshot.as_bytes().len());
4260 hasher.update(snapshot.as_bytes());
4261 }
4262
4263 Ok(finalize_schema_metadata(
4264 max_version,
4265 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
4266 hasher,
4267 latest_by_entity.len(),
4268 ))
4269}
4270
4271fn finalize_schema_metadata(
4272 schema_version: SchemaVersion,
4273 schema_fingerprint_method_version: u8,
4274 hasher: sha2::Sha256,
4275 entity_count: usize,
4276) -> SchemaStoreCatalogMetadata {
4277 let digest = finalize_hash_sha256(hasher);
4278 let mut schema_fingerprint = [0u8; 16];
4279 schema_fingerprint.copy_from_slice(&digest[..16]);
4280
4281 SchemaStoreCatalogMetadata::new(
4282 schema_version,
4283 schema_fingerprint_method_version,
4284 schema_fingerprint,
4285 u64::try_from(entity_count).unwrap_or(u64::MAX),
4286 )
4287}
4288
4289fn hash_persisted_index_key(hasher: &mut sha2::Sha256, key: &PersistedIndexKeySnapshot) {
4290 match key {
4291 PersistedIndexKeySnapshot::FieldPath(paths) => {
4292 write_hash_tag_u8(hasher, 1);
4293 write_hash_len_u32(hasher, paths.len());
4294 for path in paths {
4295 hash_persisted_index_field_path(hasher, path);
4296 }
4297 }
4298 PersistedIndexKeySnapshot::Items(items) => {
4299 write_hash_tag_u8(hasher, 2);
4300 write_hash_len_u32(hasher, items.len());
4301 for item in items {
4302 match item {
4303 PersistedIndexKeyItemSnapshot::FieldPath(path) => {
4304 write_hash_tag_u8(hasher, 1);
4305 hash_persisted_index_field_path(hasher, path);
4306 }
4307 PersistedIndexKeyItemSnapshot::Expression(expression) => {
4308 write_hash_tag_u8(hasher, 2);
4309 write_hash_str_u32(hasher, persisted_expression_op_name(expression.op()));
4310 hash_persisted_index_field_path(hasher, expression.source());
4311 hash_accepted_field_kind(hasher, expression.input_kind());
4312 hash_accepted_field_kind(hasher, expression.output_kind());
4313 write_hash_str_u32(hasher, expression.canonical_text());
4314 }
4315 }
4316 }
4317 }
4318 }
4319}
4320
4321fn hash_persisted_index_field_path(
4322 hasher: &mut sha2::Sha256,
4323 path: &crate::db::schema::PersistedIndexFieldPathSnapshot,
4324) {
4325 write_hash_u32(hasher, path.field_id().get());
4326 write_hash_u32(hasher, u32::from(path.slot().get()));
4327 write_hash_len_u32(hasher, path.path().len());
4328 for segment in path.path() {
4329 write_hash_str_u32(hasher, segment);
4330 }
4331 hash_accepted_field_kind(hasher, path.kind());
4332 write_hash_tag_u8(hasher, u8::from(path.nullable()));
4333}
4334
4335fn hash_accepted_field_kind(hasher: &mut sha2::Sha256, kind: &AcceptedFieldKind) {
4336 match kind {
4337 AcceptedFieldKind::Account => write_hash_tag_u8(hasher, 1),
4338 AcceptedFieldKind::Blob { max_len } => {
4339 write_hash_tag_u8(hasher, 2);
4340 hash_optional_u32(hasher, *max_len);
4341 }
4342 AcceptedFieldKind::Bool => {
4343 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL);
4344 }
4345 AcceptedFieldKind::Date => write_hash_tag_u8(hasher, 4),
4346 AcceptedFieldKind::Decimal { scale } => {
4347 write_hash_tag_u8(hasher, 5);
4348 write_hash_u32(hasher, *scale);
4349 }
4350 AcceptedFieldKind::Duration => write_hash_tag_u8(hasher, 6),
4351 AcceptedFieldKind::Enum { type_id } => {
4352 write_hash_tag_u8(hasher, 7);
4353 write_hash_u32(hasher, type_id.get());
4354 }
4355 AcceptedFieldKind::Float32 => write_hash_tag_u8(hasher, 8),
4356 AcceptedFieldKind::Float64 => write_hash_tag_u8(hasher, 9),
4357 AcceptedFieldKind::Int8 => write_hash_tag_u8(hasher, 10),
4358 AcceptedFieldKind::Int16 => write_hash_tag_u8(hasher, 11),
4359 AcceptedFieldKind::Int32 => write_hash_tag_u8(hasher, 12),
4360 AcceptedFieldKind::Int64 => write_hash_tag_u8(hasher, 13),
4361 AcceptedFieldKind::Int128 => write_hash_tag_u8(hasher, 14),
4362 AcceptedFieldKind::IntBig { max_bytes } => {
4363 write_hash_tag_u8(hasher, 15);
4364 write_hash_u32(hasher, *max_bytes);
4365 }
4366 AcceptedFieldKind::Principal => write_hash_tag_u8(hasher, 16),
4367 AcceptedFieldKind::Subaccount => write_hash_tag_u8(hasher, 17),
4368 AcceptedFieldKind::Text { max_len } => {
4369 write_hash_tag_u8(hasher, 18);
4370 hash_optional_u32(hasher, *max_len);
4371 }
4372 AcceptedFieldKind::Timestamp => write_hash_tag_u8(hasher, 19),
4373 AcceptedFieldKind::Nat8 => write_hash_tag_u8(hasher, 20),
4374 AcceptedFieldKind::Nat16 => write_hash_tag_u8(hasher, 21),
4375 AcceptedFieldKind::Nat32 => write_hash_tag_u8(hasher, 22),
4376 AcceptedFieldKind::Nat64 => write_hash_tag_u8(hasher, 23),
4377 AcceptedFieldKind::Nat128 => write_hash_tag_u8(hasher, 24),
4378 AcceptedFieldKind::NatBig { max_bytes } => {
4379 write_hash_tag_u8(hasher, 25);
4380 write_hash_u32(hasher, *max_bytes);
4381 }
4382 AcceptedFieldKind::Ulid => write_hash_tag_u8(hasher, 26),
4383 AcceptedFieldKind::Unit => write_hash_tag_u8(hasher, 27),
4384 AcceptedFieldKind::Relation {
4385 target_path,
4386 target_entity_name,
4387 target_entity_tag,
4388 target_store_path,
4389 key_kind,
4390 } => {
4391 write_hash_tag_u8(hasher, 28);
4392 write_hash_str_u32(hasher, target_path);
4393 write_hash_str_u32(hasher, target_entity_name);
4394 write_hash_u64(hasher, target_entity_tag.value());
4395 write_hash_str_u32(hasher, target_store_path);
4396 hash_accepted_field_kind(hasher, key_kind);
4397 }
4398 AcceptedFieldKind::List(inner) => {
4399 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST);
4400 hash_accepted_field_kind(hasher, inner);
4401 }
4402 AcceptedFieldKind::Set(inner) => {
4403 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET);
4404 hash_accepted_field_kind(hasher, inner);
4405 }
4406 AcceptedFieldKind::Map { key, value } => {
4407 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP);
4408 hash_accepted_field_kind(hasher, key);
4409 hash_accepted_field_kind(hasher, value);
4410 }
4411 AcceptedFieldKind::Composite { type_id } => {
4412 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE);
4413 write_hash_u32(hasher, type_id.get());
4414 }
4415 }
4416}
4417
4418fn hash_optional_u32(hasher: &mut sha2::Sha256, value: Option<u32>) {
4419 match value {
4420 Some(value) => {
4421 write_hash_tag_u8(hasher, 1);
4422 write_hash_u32(hasher, value);
4423 }
4424 None => write_hash_tag_u8(hasher, 0),
4425 }
4426}
4427
4428const fn persisted_index_origin_name(
4429 origin: crate::db::schema::PersistedIndexOrigin,
4430) -> &'static str {
4431 match origin {
4432 crate::db::schema::PersistedIndexOrigin::Generated => "generated",
4433 crate::db::schema::PersistedIndexOrigin::SqlDdl => "sql_ddl",
4434 }
4435}
4436
4437const fn persisted_expression_op_name(
4438 op: crate::db::schema::PersistedIndexExpressionOp,
4439) -> &'static str {
4440 match op {
4441 crate::db::schema::PersistedIndexExpressionOp::Lower => "lower",
4442 crate::db::schema::PersistedIndexExpressionOp::Upper => "upper",
4443 crate::db::schema::PersistedIndexExpressionOp::Trim => "trim",
4444 crate::db::schema::PersistedIndexExpressionOp::LowerTrim => "lower_trim",
4445 crate::db::schema::PersistedIndexExpressionOp::Date => "date",
4446 crate::db::schema::PersistedIndexExpressionOp::Year => "year",
4447 crate::db::schema::PersistedIndexExpressionOp::Month => "month",
4448 crate::db::schema::PersistedIndexExpressionOp::Day => "day",
4449 }
4450}
4451
4452#[cfg(test)]
4457mod tests;