1mod deep;
7mod derived;
8mod job;
9mod progress_codec;
10mod progress_store;
11mod proof;
12mod row;
13
14use crate::{
15 db::{
16 commit::ensure_recovery_admitted,
17 registry::{
18 StoreAllocationIdentities, StoreHandle, StoreRuntimeStorageCapabilities,
19 StoreRuntimeStorageMode,
20 },
21 schema::{
22 AcceptedInspectionPlan, IdentityStateLifecycle, MAX_IDENTITY_STATE_RECORDS_PER_DATABASE,
23 },
24 },
25 error::{ConstraintValuePath, ErrorClass, ErrorOrigin, InternalError},
26 traits::CanisterKind,
27};
28
29use serde::Deserialize;
30use std::{
31 collections::BTreeMap,
32 sync::atomic::{AtomicU64, Ordering},
33};
34
35pub(in crate::db) use deep::{
36 abort_deep_integrity_job, continue_deep_integrity_job, run_next_integrity_retention_page,
37 start_deep_integrity_job,
38};
39pub(in crate::db) use derived::{
40 DerivedInspectionLimits, execute_index_integrity_page, execute_reverse_integrity_page,
41};
42pub use job::{
43 DeepIntegrityPage, DeepIntegrityPageStatus, IntegrityAbortReceipt, IntegrityAbortStatus,
44 IntegrityDeepError, IntegrityJobError, IntegrityJobId, IntegrityJobOwner, IntegrityJobReceipt,
45 IntegrityPendingTerminal, IntegritySubmissionKey, IntegrityTerminalOutcome,
46};
47pub(in crate::db) use job::{
48 IntegrityCheckpoint, IntegrityJob, IntegrityJobState, IntegrityReceiptEnvelope,
49 IntegrityReceiptReplayKey, MAX_INTEGRITY_IN_PROGRESS_PAGES,
50};
51#[cfg(any(feature = "sql", test))]
52pub(in crate::db) use progress_store::InsertMutationJobResult;
53#[cfg(feature = "sql")]
54pub(in crate::db) use progress_store::replace_mutation_progress_record_op;
55pub(in crate::db) use progress_store::{
56 MutationProgressRecordOp, apply_mutation_progress_record_op,
57 apply_preflighted_mutation_progress_record_op, preflight_mutation_progress_record_op,
58 verify_mutation_progress_record_op, with_mutation_progress_store,
59 with_resumable_progress_store,
60};
61pub use progress_store::{
62 ProgressJobFamily, ProgressJobInventory, ProgressJobInventoryRecord, ProgressJobLifecycle,
63};
64#[cfg(all(test, feature = "sql"))]
65pub(in crate::db) use progress_store::{
66 miskey_progress_record_for_test, replace_progress_record_for_test,
67};
68pub(in crate::db) use proof::{IntegrityProofVector, capture_integrity_proof_vector};
69pub(in crate::db) use row::{
70 PhysicalUnitCheckpoint, RowInspectionLimits, execute_row_integrity_page,
71};
72
73pub(in crate::db) const MAX_INTEGRITY_PATH_BYTES: usize = 4 * 1024;
74
75#[derive(Clone, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
83pub enum IntegrityCheckRequest {
84 Quick {
86 entity: IntegrityEntityIdentity,
88 },
89 DeepStart {
91 entity: IntegrityEntityIdentity,
93 submission_key: IntegritySubmissionKey,
95 },
96 DeepContinue {
98 job_id: IntegrityJobId,
100 acknowledged_sequence: u64,
102 },
103 DeepAbort {
105 job_id: IntegrityJobId,
107 },
108}
109
110impl IntegrityCheckRequest {
111 #[must_use]
113 pub const fn deep_continue(job_id: IntegrityJobId, acknowledged_sequence: u64) -> Self {
114 Self::DeepContinue {
115 job_id,
116 acknowledged_sequence,
117 }
118 }
119
120 #[must_use]
122 pub const fn deep_abort(job_id: IntegrityJobId) -> Self {
123 Self::DeepAbort { job_id }
124 }
125}
126
127#[derive(Clone, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
130pub enum IntegrityCheckResult {
131 Quick(QuickIntegrityResult),
133 Deep(IntegrityJobReceipt),
135}
136
137fn resolve_integrity_participating_stores<C: CanisterKind>(
139 db: &crate::db::Db<C>,
140 plan: &AcceptedInspectionPlan,
141) -> Result<BTreeMap<String, StoreHandle>, InternalError> {
142 let identity = plan.identity();
143 let source_store = db.store_handle(identity.store_path())?;
144 let mut stores = BTreeMap::from([(identity.store_path().to_string(), source_store)]);
145 for relation in plan.relation_inspection() {
146 stores
147 .entry(relation.target_store_path().to_string())
148 .or_insert_with(|| relation.target_store());
149 }
150 Ok(stores)
151}
152
153fn validate_quick_integrity_control<C: CanisterKind>(
154 db: &crate::db::Db<C>,
155 plan: &AcceptedInspectionPlan,
156 incarnation: DatabaseIncarnationId,
157) -> Result<Vec<IntegrityFinding>, InternalError> {
158 let identity = plan.identity();
159 let participating_stores = resolve_integrity_participating_stores(db, plan)?;
160
161 proof::validate_integrity_allocation_registry()?;
166 validate_quick_identity_control(db, incarnation)?;
167 let mut findings = Vec::new();
168 for (store_path, store) in &participating_stores {
169 if let Some(finding) = validate_quick_store_control(plan, store_path, *store)? {
170 findings.push(finding);
171 }
172 }
173 for ordinal in 0..plan.index_inspection().len() {
174 let _domain = plan
175 .index_inspection()
176 .domain(ordinal, identity.entity_tag())?;
177 }
178
179 Ok(findings)
180}
181
182fn validate_quick_identity_control<C: CanisterKind>(
183 db: &crate::db::Db<C>,
184 incarnation: DatabaseIncarnationId,
185) -> Result<(), InternalError> {
186 let mut stores = db.with_store_registry(|registry| registry.iter().collect::<Vec<_>>());
187 icydb_schema::compact_sort_unstable_by(&mut stores, |left, right| left.0.cmp(right.0));
188
189 let mut owners = BTreeMap::new();
190 let mut state_count = 0usize;
191 for (store_path, store) in stores {
192 let states = store.with_schema(|schema_store| {
193 schema_store.identity_state_inventory_for_integrity(incarnation)
194 })?;
195 state_count = state_count
196 .checked_add(states.len())
197 .ok_or_else(InternalError::identity_state_corruption)?;
198 if state_count > MAX_IDENTITY_STATE_RECORDS_PER_DATABASE {
199 return Err(InternalError::identity_state_corruption());
200 }
201
202 for state in states {
203 let owner = state.owner();
204 record_quick_identity_owner(&mut owners, store_path, &state)?;
205 if state.lifecycle() == IdentityStateLifecycle::Active {
206 let runtime_entity = db
207 .accepted_runtime_entity_for_tag(owner.entity_tag())
208 .map_err(|_| InternalError::identity_state_corruption())?;
209 if runtime_entity.store_path() != store_path {
210 return Err(InternalError::identity_state_corruption());
211 }
212 }
213 }
214 }
215
216 Ok(())
217}
218
219fn record_quick_identity_owner<'a>(
220 owners: &mut BTreeMap<(crate::types::EntityTag, crate::db::schema::FieldId), &'a str>,
221 store_path: &'a str,
222 state: &crate::db::schema::IdentityState,
223) -> Result<(), InternalError> {
224 let owner = state.owner();
225 let key = (owner.entity_tag(), owner.field_id());
226 if owners.insert(key, store_path).is_some() {
227 return Err(InternalError::identity_state_corruption());
228 }
229 Ok(())
230}
231
232fn validate_quick_store_control(
233 plan: &AcceptedInspectionPlan,
234 store_path: &str,
235 store: StoreHandle,
236) -> Result<Option<IntegrityFinding>, InternalError> {
237 let capabilities = store.storage_capabilities();
238 let allocations = store.allocation_identities();
239 match capabilities.storage_mode() {
240 StoreRuntimeStorageMode::Heap => {
241 if capabilities != StoreRuntimeStorageCapabilities::heap()
242 || allocations != StoreAllocationIdentities::absent()
243 || store.journal_tail_store().is_some()
244 {
245 return Err(InternalError::store_invariant());
246 }
247 }
248 StoreRuntimeStorageMode::Journaled => {
249 if capabilities != StoreRuntimeStorageCapabilities::journaled()
250 || !allocations.matches_storage_capabilities(capabilities)
251 {
252 return Err(InternalError::store_invariant());
253 }
254 let journal = store
255 .journal_tail_store()
256 .ok_or_else(InternalError::store_invariant)?
257 .with_borrow(crate::db::journal::JournalTailStore::proof_identity)?;
258 if !journal.is_well_formed() {
259 return Ok(Some(quick_journal_control_finding(plan, store_path)));
260 }
261 }
262 }
263 Ok(None)
264}
265
266fn quick_journal_control_finding(
267 plan: &AcceptedInspectionPlan,
268 store_path: &str,
269) -> IntegrityFinding {
270 let error = InternalError::store_corruption();
271 IntegrityFinding {
272 diagnostic_code: error.diagnostic_code().error_code().raw(),
273 class: IntegrityFindingClass::Corruption,
274 severity: IntegritySeverity::Error,
275 kind: IntegrityFindingKind::JournalControlMismatch,
276 entity: IntegrityEntityIdentity::from_plan(plan),
277 store_path: store_path.to_string(),
278 phase: IntegrityPhase::QuickMetadata,
279 verifier_family: IntegrityVerifierFamily::JournalEnvelope,
280 physical_key: Vec::new(),
281 primary_key: None,
282 field_paths: Vec::new(),
283 value_path: None,
284 constraint_id: None,
285 constraint_name: None,
286 schema_index_id: None,
287 relation_id: None,
288 expected: Some("well-formed-journal-control".to_string()),
289 observed: Some("inconsistent-journal-control".to_string()),
290 }
291}
292
293fn relation_field_paths(plan: &AcceptedInspectionPlan, relation_id: u32) -> Vec<String> {
294 let snapshot = plan.snapshot().persisted_snapshot();
295 let Some(relation) = snapshot
296 .relations()
297 .iter()
298 .find(|relation| relation.id().get() == relation_id)
299 else {
300 return Vec::new();
301 };
302
303 relation
304 .source()
305 .root_field_ids()
306 .iter()
307 .filter_map(|field_id| {
308 snapshot
309 .fields()
310 .iter()
311 .find(|field| field.id() == *field_id)
312 .map(|field| field.name().to_string())
313 })
314 .collect()
315}
316
317const MAX_QUICK_RETURNED_FINDINGS: usize = 64;
318#[cfg(target_arch = "wasm32")]
319const DATABASE_INCARNATION_DOMAIN: &[u8] = b"icydb.database-incarnation.v1";
320static DATABASE_INCARNATION_SEQUENCE: AtomicU64 = AtomicU64::new(0);
321
322#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, serde::Serialize)]
329pub struct DatabaseIncarnationId([u8; 16]);
330
331impl DatabaseIncarnationId {
332 pub(crate) fn try_from_bytes(bytes: [u8; 16]) -> Result<Self, InternalError> {
334 if bytes == [0; 16] {
335 return Err(InternalError::database_incarnation_invalid());
336 }
337
338 Ok(Self(bytes))
339 }
340
341 #[must_use]
343 pub const fn to_bytes(self) -> [u8; 16] {
344 self.0
345 }
346
347 fn generate() -> Result<Self, InternalError> {
348 #[allow(
349 deprecated,
350 reason = "Atomic try_update requires Rust 1.95; MSRV is 1.91."
351 )]
352 let sequence = DATABASE_INCARNATION_SEQUENCE
353 .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
354 current.checked_add(1)
355 })
356 .map_err(|_| InternalError::database_incarnation_generation_failed())?
357 .checked_add(1)
358 .ok_or_else(InternalError::database_incarnation_generation_failed)?;
359
360 #[cfg(not(target_arch = "wasm32"))]
361 let bytes = {
362 let mut bytes = [0_u8; 16];
363 getrandom::fill(&mut bytes)
364 .map_err(|_| InternalError::database_incarnation_generation_failed())?;
365 bytes
366 };
367
368 #[cfg(target_arch = "wasm32")]
369 let bytes = {
370 use sha2::{Digest, Sha256};
371
372 let mut hasher = Sha256::new();
373 hasher.update(DATABASE_INCARNATION_DOMAIN);
374 let mut canister = [0_u8; 29];
375 let canister = canister
376 .get_mut(..ic_platform::ic0::canister_self_size())
377 .ok_or_else(InternalError::database_incarnation_generation_failed)?;
378 ic_platform::ic0::canister_self_copy(canister, 0);
379 hasher.update(canister);
380 hasher.update(ic_platform::ic0::time().to_be_bytes());
381 hasher.update(sequence.to_be_bytes());
382 let digest = hasher.finalize();
383 let mut bytes = [0_u8; 16];
384 bytes.copy_from_slice(&digest[..16]);
385 bytes
386 };
387
388 let _ = sequence;
389 Self::try_from_bytes(bytes)
390 }
391
392 #[cfg(test)]
393 pub(crate) const fn for_tests(fill: u8) -> Self {
394 let mut bytes = [fill; 16];
395 if fill == 0 {
396 bytes[15] = 1;
397 }
398 Self(bytes)
399 }
400}
401
402#[derive(Clone, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
404pub struct IntegrityEntityIdentity {
405 entity_tag: u64,
406 entity_path: String,
407 store_path: String,
408}
409
410impl IntegrityEntityIdentity {
411 fn from_plan(plan: &AcceptedInspectionPlan) -> Self {
412 Self::from_accepted_identity(plan.identity_ref())
413 }
414
415 pub(in crate::db) fn from_accepted_identity(
416 identity: &crate::db::schema::AcceptedCatalogIdentity,
417 ) -> Self {
418 Self {
419 entity_tag: identity.entity_tag().value(),
420 entity_path: identity.entity_path().to_string(),
421 store_path: identity.store_path().to_string(),
422 }
423 }
424
425 pub(in crate::db) const fn validate(&self) -> Result<(), IntegrityJobError> {
426 if self.entity_tag == 0
427 || self.entity_path.is_empty()
428 || self.entity_path.len() > MAX_INTEGRITY_PATH_BYTES
429 || self.store_path.is_empty()
430 || self.store_path.len() > MAX_INTEGRITY_PATH_BYTES
431 {
432 return Err(IntegrityJobError::InvalidEntityIdentity);
433 }
434 Ok(())
435 }
436
437 #[must_use]
439 pub const fn entity_tag(&self) -> u64 {
440 self.entity_tag
441 }
442
443 #[must_use]
445 pub const fn entity_path(&self) -> &str {
446 self.entity_path.as_str()
447 }
448
449 #[must_use]
451 pub const fn store_path(&self) -> &str {
452 self.store_path.as_str()
453 }
454}
455
456#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
458pub enum IntegrityAuthorityClass {
459 Corruption,
461 IncompatiblePersistedFormat,
463 InvariantViolation,
465 Unsupported,
467 Internal,
469}
470
471#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
473pub enum IntegrityFindingClass {
474 Corruption,
476 IncompatiblePersistedFormat,
478 ResourceLimited,
480}
481
482#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
485pub enum IntegrityFindingKind {
486 MalformedDataKey,
488
489 MalformedRow,
491
492 OversizedRow,
494
495 InvalidFieldValue,
497
498 PrimaryKeyMismatch,
500
501 InvalidIdentityValue,
503
504 IdentityHighWaterExceeded,
506
507 ConstraintViolation,
509
510 MissingIndexEntry,
512
513 DivergentIndexEntry,
515
516 MalformedIndexEntry,
518
519 OrphanIndexEntry,
521
522 DuplicateUniqueIndexKey,
524
525 MissingRelationTarget,
527
528 MissingReverseRelationEntry,
530
531 DivergentReverseRelationEntry,
533
534 MalformedReverseRelationEntry,
536
537 OrphanReverseRelationEntry,
539
540 MalformedJournalBatch,
542
543 JournalSequenceGap,
545
546 DuplicateJournalBatchIdentity,
548
549 JournalControlMismatch,
551}
552
553#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
556pub enum IntegrityPhase {
557 QuickMetadata,
559
560 Rows,
562
563 IndexEntries,
565
566 ReverseRelations,
568
569 JournalTails,
571
572 FinalProofVectorCheck,
574}
575
576#[derive(
579 serde::Serialize, Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd,
580)]
581pub enum IntegrityVerifierFamily {
582 DataKey,
584
585 RowEnvelope,
587
588 FieldValue,
590
591 PrimaryKey,
593
594 IdentityState,
596
597 ValidatedConstraints,
599
600 ForwardIndex,
602
603 IndexEntry,
605
606 UniqueIndex,
608
609 Relation,
611
612 ReverseRelationEntry,
614
615 JournalEnvelope,
617
618 JournalBatchIdentity,
620}
621
622#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
624pub enum IntegritySeverity {
625 Error,
627 Advisory,
629}
630
631#[derive(Clone, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
635pub struct IntegrityFinding {
636 diagnostic_code: u16,
637 class: IntegrityFindingClass,
638 severity: IntegritySeverity,
639 kind: IntegrityFindingKind,
640 entity: IntegrityEntityIdentity,
641 store_path: String,
642 phase: IntegrityPhase,
643 verifier_family: IntegrityVerifierFamily,
644 physical_key: Vec<u8>,
645 primary_key: Option<Vec<u8>>,
646 field_paths: Vec<String>,
647 value_path: Option<Box<ConstraintValuePath>>,
648 constraint_id: Option<u32>,
649 constraint_name: Option<String>,
650 schema_index_id: Option<u32>,
651 relation_id: Option<u32>,
652 expected: Option<String>,
653 observed: Option<String>,
654}
655
656impl IntegrityFinding {
657 #[must_use]
659 pub const fn diagnostic_code(&self) -> u16 {
660 self.diagnostic_code
661 }
662
663 #[must_use]
665 pub const fn class(&self) -> IntegrityFindingClass {
666 self.class
667 }
668
669 #[must_use]
671 pub const fn severity(&self) -> IntegritySeverity {
672 self.severity
673 }
674
675 #[must_use]
677 pub const fn kind(&self) -> IntegrityFindingKind {
678 self.kind
679 }
680
681 #[must_use]
683 pub const fn entity(&self) -> &IntegrityEntityIdentity {
684 &self.entity
685 }
686
687 #[must_use]
689 pub const fn store_path(&self) -> &str {
690 self.store_path.as_str()
691 }
692
693 #[must_use]
695 pub const fn phase(&self) -> IntegrityPhase {
696 self.phase
697 }
698
699 #[must_use]
701 pub const fn verifier_family(&self) -> IntegrityVerifierFamily {
702 self.verifier_family
703 }
704
705 #[must_use]
707 pub const fn physical_key(&self) -> &[u8] {
708 self.physical_key.as_slice()
709 }
710
711 #[must_use]
713 pub fn primary_key(&self) -> Option<&[u8]> {
714 self.primary_key.as_deref()
715 }
716
717 #[must_use]
719 pub const fn field_paths(&self) -> &[String] {
720 self.field_paths.as_slice()
721 }
722
723 #[must_use]
725 pub fn value_path(&self) -> Option<&ConstraintValuePath> {
726 self.value_path.as_deref()
727 }
728
729 #[must_use]
731 pub const fn constraint_id(&self) -> Option<u32> {
732 self.constraint_id
733 }
734
735 #[must_use]
737 pub fn constraint_name(&self) -> Option<&str> {
738 self.constraint_name.as_deref()
739 }
740
741 #[must_use]
743 pub const fn schema_index_id(&self) -> Option<u32> {
744 self.schema_index_id
745 }
746
747 #[must_use]
749 pub const fn relation_id(&self) -> Option<u32> {
750 self.relation_id
751 }
752
753 #[must_use]
755 pub fn expected(&self) -> Option<&str> {
756 self.expected.as_deref()
757 }
758
759 #[must_use]
761 pub fn observed(&self) -> Option<&str> {
762 self.observed.as_deref()
763 }
764}
765
766#[derive(Clone, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
768pub struct IntegrityAuthorityDiagnostic {
769 diagnostic_code: u16,
770 class: IntegrityAuthorityClass,
771}
772
773impl IntegrityAuthorityDiagnostic {
774 pub(in crate::db) fn from_internal(error: &InternalError) -> Self {
775 let class = match error.class {
776 ErrorClass::Corruption => IntegrityAuthorityClass::Corruption,
777 ErrorClass::IncompatiblePersistedFormat => {
778 IntegrityAuthorityClass::IncompatiblePersistedFormat
779 }
780 ErrorClass::InvariantViolation => IntegrityAuthorityClass::InvariantViolation,
781 ErrorClass::Unsupported | ErrorClass::NotFound | ErrorClass::Conflict => {
782 IntegrityAuthorityClass::Unsupported
783 }
784 ErrorClass::Internal => IntegrityAuthorityClass::Internal,
785 };
786 Self {
787 diagnostic_code: error.diagnostic_code().error_code().raw(),
788 class,
789 }
790 }
791
792 #[must_use]
794 pub const fn diagnostic_code(&self) -> u16 {
795 self.diagnostic_code
796 }
797
798 #[must_use]
800 pub const fn class(&self) -> IntegrityAuthorityClass {
801 self.class
802 }
803}
804
805#[derive(Clone, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
807pub struct IntegrityResourceDiagnostic {
808 diagnostic_code: u16,
809}
810
811impl IntegrityResourceDiagnostic {
812 #[must_use]
814 pub const fn diagnostic_code(&self) -> u16 {
815 self.diagnostic_code
816 }
817}
818
819#[derive(Clone, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
821pub enum QuickIntegrityStatus {
822 CompleteClean,
824 CompleteWithFindings,
826 Uninspectable(IntegrityAuthorityDiagnostic),
828 ResourceLimited(IntegrityResourceDiagnostic),
830}
831
832#[derive(Clone, Debug, Deserialize, Eq, PartialEq, serde::Serialize)]
834pub struct QuickIntegrityResult {
835 entity: IntegrityEntityIdentity,
836 database_incarnation_id: DatabaseIncarnationId,
837 accepted_schema_version: u32,
838 accepted_schema_fingerprint: [u8; 16],
839 status: QuickIntegrityStatus,
840 total_findings: u64,
841 omitted_findings: u64,
842 findings: Vec<IntegrityFinding>,
843}
844
845impl QuickIntegrityResult {
846 #[must_use]
848 pub const fn entity(&self) -> &IntegrityEntityIdentity {
849 &self.entity
850 }
851
852 #[must_use]
854 pub const fn database_incarnation_id(&self) -> DatabaseIncarnationId {
855 self.database_incarnation_id
856 }
857
858 #[must_use]
860 pub const fn accepted_schema_version(&self) -> u32 {
861 self.accepted_schema_version
862 }
863
864 #[must_use]
866 pub const fn accepted_schema_fingerprint(&self) -> [u8; 16] {
867 self.accepted_schema_fingerprint
868 }
869
870 #[must_use]
872 pub const fn status(&self) -> &QuickIntegrityStatus {
873 &self.status
874 }
875
876 #[must_use]
878 pub const fn total_findings(&self) -> u64 {
879 self.total_findings
880 }
881
882 #[must_use]
884 pub const fn omitted_findings(&self) -> u64 {
885 self.omitted_findings
886 }
887
888 #[must_use]
890 pub const fn findings(&self) -> &[IntegrityFinding] {
891 self.findings.as_slice()
892 }
893}
894
895struct QuickIntegrityAccumulator {
896 total_findings: u64,
897 findings: Vec<IntegrityFinding>,
898}
899
900impl QuickIntegrityAccumulator {
901 const fn new() -> Self {
902 Self {
903 total_findings: 0,
904 findings: Vec::new(),
905 }
906 }
907
908 fn record(&mut self, finding: IntegrityFinding) -> Result<(), IntegrityResourceDiagnostic> {
909 self.total_findings =
910 self.total_findings
911 .checked_add(1)
912 .ok_or(IntegrityResourceDiagnostic {
913 diagnostic_code: icydb_diagnostic_code::ErrorCode::RUNTIME_INTERNAL.raw(),
914 })?;
915 if self.findings.len() < MAX_QUICK_RETURNED_FINDINGS {
916 self.findings.push(finding);
917 }
918 Ok(())
919 }
920
921 fn complete(
922 self,
923 plan: &AcceptedInspectionPlan,
924 incarnation: DatabaseIncarnationId,
925 ) -> Result<QuickIntegrityResult, InternalError> {
926 let status = if self.total_findings == 0 {
927 QuickIntegrityStatus::CompleteClean
928 } else {
929 QuickIntegrityStatus::CompleteWithFindings
930 };
931 let omitted_findings = self.omitted_findings()?;
932 let identity = plan.identity();
933
934 Ok(QuickIntegrityResult {
935 entity: IntegrityEntityIdentity::from_plan(plan),
936 database_incarnation_id: incarnation,
937 accepted_schema_version: identity.accepted_schema_version().get(),
938 accepted_schema_fingerprint: identity.accepted_schema_fingerprint(),
939 status,
940 total_findings: self.total_findings,
941 omitted_findings,
942 findings: self.findings,
943 })
944 }
945
946 fn resource_limited(
947 self,
948 plan: &AcceptedInspectionPlan,
949 incarnation: DatabaseIncarnationId,
950 diagnostic: IntegrityResourceDiagnostic,
951 ) -> Result<QuickIntegrityResult, InternalError> {
952 let omitted_findings = self.omitted_findings()?;
953 let identity = plan.identity();
954
955 Ok(QuickIntegrityResult {
956 entity: IntegrityEntityIdentity::from_plan(plan),
957 database_incarnation_id: incarnation,
958 accepted_schema_version: identity.accepted_schema_version().get(),
959 accepted_schema_fingerprint: identity.accepted_schema_fingerprint(),
960 status: QuickIntegrityStatus::ResourceLimited(diagnostic),
961 total_findings: self.total_findings,
962 omitted_findings,
963 findings: self.findings,
964 })
965 }
966
967 fn omitted_findings(&self) -> Result<u64, InternalError> {
968 let returned = u64::try_from(self.findings.len()).map_err(|_| {
969 InternalError::classified(ErrorClass::InvariantViolation, ErrorOrigin::Response)
970 })?;
971 self.total_findings.checked_sub(returned).ok_or_else(|| {
972 InternalError::classified(ErrorClass::InvariantViolation, ErrorOrigin::Response)
973 })
974 }
975}
976
977pub(in crate::db) fn uninspectable_quick_integrity(
978 identity: crate::db::schema::AcceptedCatalogIdentity,
979 incarnation: DatabaseIncarnationId,
980 diagnostic: IntegrityAuthorityDiagnostic,
981) -> QuickIntegrityResult {
982 QuickIntegrityResult {
983 entity: IntegrityEntityIdentity::from_accepted_identity(&identity),
984 database_incarnation_id: incarnation,
985 accepted_schema_version: identity.accepted_schema_version().get(),
986 accepted_schema_fingerprint: identity.accepted_schema_fingerprint(),
987 status: QuickIntegrityStatus::Uninspectable(diagnostic),
988 total_findings: 0,
989 omitted_findings: 0,
990 findings: Vec::new(),
991 }
992}
993
994pub(in crate::db) fn execute_quick_integrity<C: CanisterKind>(
995 db: &crate::db::Db<C>,
996 plan: &AcceptedInspectionPlan,
997 incarnation: DatabaseIncarnationId,
998) -> Result<QuickIntegrityResult, InternalError> {
999 ensure_recovery_admitted(db)?;
1000 let findings = match validate_quick_integrity_control(db, plan, incarnation) {
1001 Ok(findings) => findings,
1002 Err(error) => {
1003 let IntegrityTerminalOutcome::Uninspectable(diagnostic) =
1004 IntegrityTerminalOutcome::from_internal(&error)
1005 else {
1006 return Err(error);
1007 };
1008 return Ok(uninspectable_quick_integrity(
1009 plan.identity(),
1010 incarnation,
1011 diagnostic,
1012 ));
1013 }
1014 };
1015 let mut accumulator = QuickIntegrityAccumulator::new();
1016 for finding in findings {
1017 if let Err(diagnostic) = accumulator.record(finding) {
1018 return accumulator.resource_limited(plan, incarnation, diagnostic);
1019 }
1020 }
1021
1022 accumulator.complete(plan, incarnation)
1023}
1024
1025pub(crate) fn generate_database_incarnation_id() -> Result<DatabaseIncarnationId, InternalError> {
1026 DatabaseIncarnationId::generate()
1027}
1028
1029#[cfg(test)]
1030mod tests {
1031 use super::*;
1032 use crate::{
1033 db::schema::{FieldStorageDecode, LeafCodec, ScalarCodec},
1034 db::{
1035 commit::CommitSchemaFingerprint,
1036 schema::{
1037 AcceptedCatalogIdentity, AcceptedCompositeCatalog, AcceptedFieldKind,
1038 AcceptedSchemaRevision, AcceptedSchemaSnapshot, AcceptedValueCatalogHandle,
1039 FieldId, IdentityState, IdentityStateOwner, PersistedFieldSnapshot,
1040 PersistedSchemaSnapshot, SchemaFieldSlot, SchemaInsertDefault, SchemaRowLayout,
1041 SchemaVersion, empty_accepted_enum_catalog_for_tests,
1042 },
1043 },
1044 types::EntityTag,
1045 };
1046
1047 fn plan() -> AcceptedInspectionPlan {
1048 let revision = AcceptedSchemaRevision::INITIAL;
1049 let identity = AcceptedCatalogIdentity::new(
1050 EntityTag::new(23),
1051 "tests::QuickEntity",
1052 "tests::QuickStore",
1053 revision,
1054 SchemaVersion::initial(),
1055 CommitSchemaFingerprint::from([0x44; 16]),
1056 );
1057 let snapshot = AcceptedSchemaSnapshot::new(PersistedSchemaSnapshot::new(
1058 SchemaVersion::initial(),
1059 "tests::QuickEntity".to_string(),
1060 "QuickEntity".to_string(),
1061 FieldId::new(1),
1062 SchemaRowLayout::initial(vec![(FieldId::new(1), SchemaFieldSlot::new(0))]),
1063 vec![PersistedFieldSnapshot::new_initial(
1064 FieldId::new(1),
1065 "id".to_string(),
1066 SchemaFieldSlot::new(0),
1067 AcceptedFieldKind::Nat64,
1068 Vec::new(),
1069 false,
1070 SchemaInsertDefault::None,
1071 FieldStorageDecode::ByKind,
1072 LeafCodec::Scalar(ScalarCodec::Nat64),
1073 )],
1074 ));
1075 let value_catalog = AcceptedValueCatalogHandle::new_for_tests(
1076 empty_accepted_enum_catalog_for_tests(),
1077 AcceptedCompositeCatalog::empty(),
1078 revision,
1079 );
1080
1081 AcceptedInspectionPlan::compile_relation_free_for_tests(
1082 identity,
1083 snapshot.into(),
1084 value_catalog,
1085 )
1086 .expect("accepted Quick plan should compile")
1087 }
1088
1089 fn finding(plan: &AcceptedInspectionPlan) -> IntegrityFinding {
1090 IntegrityFinding {
1091 diagnostic_code: icydb_diagnostic_code::ErrorCode::STORE_CORRUPTION.raw(),
1092 class: IntegrityFindingClass::Corruption,
1093 severity: IntegritySeverity::Error,
1094 kind: IntegrityFindingKind::MalformedRow,
1095 entity: IntegrityEntityIdentity::from_plan(plan),
1096 store_path: plan.identity().store_path().to_string(),
1097 phase: IntegrityPhase::Rows,
1098 verifier_family: IntegrityVerifierFamily::RowEnvelope,
1099 physical_key: vec![1],
1100 primary_key: None,
1101 field_paths: Vec::new(),
1102 value_path: None,
1103 constraint_id: None,
1104 constraint_name: None,
1105 schema_index_id: None,
1106 relation_id: None,
1107 expected: None,
1108 observed: None,
1109 }
1110 }
1111
1112 #[test]
1113 fn database_incarnation_rejects_zero_and_round_trips_current_bytes() {
1114 assert!(DatabaseIncarnationId::try_from_bytes([0; 16]).is_err());
1115
1116 let identity = DatabaseIncarnationId::for_tests(7);
1117 assert_eq!(
1118 DatabaseIncarnationId::try_from_bytes(identity.to_bytes())
1119 .expect("nonzero incarnation should decode"),
1120 identity,
1121 );
1122 }
1123
1124 #[test]
1125 fn integrity_finding_wire_preserves_targeted_constraint_path() {
1126 let plan = plan();
1127 let mut finding = finding(&plan);
1128 let path = ConstraintValuePath::new(vec![
1129 crate::error::ConstraintValuePathComponent::RootField { field_id: 1 },
1130 crate::error::ConstraintValuePathComponent::ListElement { index: 2 },
1131 ]);
1132 finding.kind = IntegrityFindingKind::ConstraintViolation;
1133 finding.value_path = Some(Box::new(path.clone()));
1134 finding.constraint_id = Some(7);
1135 finding.constraint_name = Some("nested_limit".to_string());
1136
1137 let bytes =
1138 icydb_schema::wire::encode_one(&finding).expect("integrity finding should encode");
1139 let decoded: IntegrityFinding =
1140 icydb_schema::wire::decode_one(&bytes).expect("integrity finding should decode");
1141 assert_eq!(decoded.value_path(), Some(&path));
1142 assert_eq!(decoded.constraint_id(), Some(7));
1143 assert_eq!(decoded.constraint_name(), Some("nested_limit"));
1144 }
1145
1146 #[test]
1147 fn quick_clean_result_binds_incarnation_and_accepted_plan_identity() {
1148 let plan = plan();
1149 let incarnation = DatabaseIncarnationId::for_tests(8);
1150 let result = QuickIntegrityAccumulator::new()
1151 .complete(&plan, incarnation)
1152 .expect("clean Quick accounting should remain valid");
1153
1154 assert_eq!(result.status(), &QuickIntegrityStatus::CompleteClean);
1155 assert_eq!(result.database_incarnation_id(), incarnation);
1156 assert_eq!(result.accepted_schema_version(), 1);
1157 assert_eq!(result.accepted_schema_fingerprint(), [0x44; 16]);
1158 assert_eq!(result.total_findings(), 0);
1159 assert_eq!(result.omitted_findings(), 0);
1160 }
1161
1162 #[test]
1163 fn quick_findings_keep_a_bounded_prefix_and_exact_omitted_count() {
1164 let plan = plan();
1165 let mut accumulator = QuickIntegrityAccumulator::new();
1166 for _ in 0..=MAX_QUICK_RETURNED_FINDINGS {
1167 accumulator
1168 .record(finding(&plan))
1169 .expect("bounded test finding count should fit");
1170 }
1171 let result = accumulator
1172 .complete(&plan, DatabaseIncarnationId::for_tests(9))
1173 .expect("one-over-cap Quick accounting should remain valid");
1174
1175 assert_eq!(result.status(), &QuickIntegrityStatus::CompleteWithFindings,);
1176 assert_eq!(result.total_findings(), 65);
1177 assert_eq!(result.findings().len(), MAX_QUICK_RETURNED_FINDINGS);
1178 assert_eq!(result.omitted_findings(), 1);
1179 assert_eq!(
1180 result.total_findings(),
1181 result.findings().len() as u64 + result.omitted_findings(),
1182 );
1183 }
1184
1185 #[test]
1186 fn quick_findings_at_the_exact_returned_cap_have_no_omissions() {
1187 let plan = plan();
1188 let mut accumulator = QuickIntegrityAccumulator::new();
1189 for _ in 0..MAX_QUICK_RETURNED_FINDINGS {
1190 accumulator
1191 .record(finding(&plan))
1192 .expect("exact-cap finding count should fit");
1193 }
1194 let result = accumulator
1195 .complete(&plan, DatabaseIncarnationId::for_tests(10))
1196 .expect("exact-cap Quick accounting should remain valid");
1197
1198 assert_eq!(result.total_findings(), 64);
1199 assert_eq!(result.findings().len(), MAX_QUICK_RETURNED_FINDINGS);
1200 assert_eq!(result.omitted_findings(), 0);
1201 }
1202
1203 #[test]
1204 fn quick_selected_authority_failure_is_not_a_clean_completion() {
1205 let plan = plan();
1206 let error = InternalError::accepted_row_constraint_program_corrupt();
1207 let result = uninspectable_quick_integrity(
1208 plan.identity(),
1209 DatabaseIncarnationId::for_tests(11),
1210 IntegrityAuthorityDiagnostic::from_internal(&error),
1211 );
1212
1213 assert!(matches!(
1214 result.status(),
1215 QuickIntegrityStatus::Uninspectable(IntegrityAuthorityDiagnostic {
1216 class: IntegrityAuthorityClass::Corruption,
1217 ..
1218 }),
1219 ));
1220 assert_eq!(result.total_findings(), 0);
1221 assert_eq!(result.omitted_findings(), 0);
1222 }
1223
1224 #[test]
1225 fn quick_identity_inventory_rejects_active_retired_owner_collision_first() {
1226 let incarnation = DatabaseIncarnationId::for_tests(12);
1227 let owner = IdentityStateOwner::try_new(incarnation, EntityTag::new(31), FieldId::new(1))
1228 .expect("identity owner should admit");
1229 let active = IdentityState::new_active(owner, AcceptedFieldKind::Nat64)
1230 .expect("active identity state should admit");
1231 let retired = active.retire().expect("active state should retire");
1232 let mut owners = BTreeMap::new();
1233
1234 record_quick_identity_owner(&mut owners, "tests::first", &active)
1235 .expect("the first owner should admit");
1236 let error = record_quick_identity_owner(&mut owners, "tests::second", &retired)
1237 .expect_err("an active/retired owner collision must reject");
1238
1239 assert_eq!(error.class(), ErrorClass::Corruption);
1240 assert_eq!(error.origin(), ErrorOrigin::Identity);
1241 }
1242}