Skip to main content

icydb_core/db/schema/
store.rs

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