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