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