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