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