use crate::db::{
codec::hex::encode_hex_lower,
integrity::{
DatabaseIncarnationId, IntegrityAuthorityDiagnostic, IntegrityEntityIdentity,
IntegrityFinding, IntegrityPhase, IntegrityProofVector, IntegrityResourceDiagnostic,
IntegrityVerifierFamily, MAX_INTEGRITY_PATH_BYTES, PhysicalUnitCheckpoint,
},
journal::JournalInspectionCheckpoint,
};
use candid::CandidType;
use serde::Deserialize;
pub(in crate::db) const MAX_INTEGRITY_OWNER_BYTES: usize = 256;
pub(in crate::db) const MAX_INTEGRITY_SUBMISSION_KEY_BYTES: usize = 256;
const MAX_INTEGRITY_RECEIPT_FINDINGS: usize = 64;
pub(in crate::db) const MAX_INTEGRITY_IN_PROGRESS_PAGES: u64 = u64::MAX - 1;
#[derive(CandidType, Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct IntegrityJobId([u8; 32]);
impl IntegrityJobId {
pub fn try_from_bytes(bytes: [u8; 32]) -> Result<Self, IntegrityJobError> {
if bytes == [0; 32] {
return Err(IntegrityJobError::CorruptProgressRecord);
}
Ok(Self(bytes))
}
pub fn try_from_hex(value: &str) -> Result<Self, IntegrityJobError> {
if value.len() != 64 {
return Err(IntegrityJobError::InvalidJobId);
}
let mut bytes = [0_u8; 32];
for (index, pair) in value.as_bytes().chunks_exact(2).enumerate() {
let high = decode_hex_nibble(pair[0]).ok_or(IntegrityJobError::InvalidJobId)?;
let low = decode_hex_nibble(pair[1]).ok_or(IntegrityJobError::InvalidJobId)?;
bytes[index] = (high << 4) | low;
}
if bytes == [0; 32] {
return Err(IntegrityJobError::InvalidJobId);
}
Ok(Self(bytes))
}
#[must_use]
pub const fn to_bytes(self) -> [u8; 32] {
self.0
}
#[must_use]
pub fn to_hex(self) -> String {
encode_hex_lower(&self.0)
}
pub(in crate::db) fn validate(self) -> Result<(), IntegrityJobError> {
if self.0 == [0; 32] {
return Err(IntegrityJobError::InvalidJobId);
}
Ok(())
}
}
const fn decode_hex_nibble(value: u8) -> Option<u8> {
match value {
b'0'..=b'9' => Some(value - b'0'),
b'a'..=b'f' => Some(value - b'a' + 10),
b'A'..=b'F' => Some(value - b'A' + 10),
_ => None,
}
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub struct IntegrityJobOwner(String);
impl IntegrityJobOwner {
pub fn new(value: impl Into<String>) -> Result<Self, IntegrityJobError> {
let value = value.into();
if value.is_empty() || value.len() > MAX_INTEGRITY_OWNER_BYTES {
return Err(IntegrityJobError::InvalidOwner);
}
Ok(Self(value))
}
#[must_use]
pub const fn as_str(&self) -> &str {
self.0.as_str()
}
pub(in crate::db) const fn validate(&self) -> Result<(), IntegrityJobError> {
if self.0.is_empty() || self.0.len() > MAX_INTEGRITY_OWNER_BYTES {
return Err(IntegrityJobError::InvalidOwner);
}
Ok(())
}
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub struct IntegritySubmissionKey(String);
impl IntegritySubmissionKey {
pub fn new(value: impl Into<String>) -> Result<Self, IntegrityJobError> {
let value = value.into();
if value.is_empty() || value.len() > MAX_INTEGRITY_SUBMISSION_KEY_BYTES {
return Err(IntegrityJobError::InvalidSubmissionKey);
}
Ok(Self(value))
}
#[must_use]
pub const fn as_str(&self) -> &str {
self.0.as_str()
}
pub(in crate::db) const fn validate(&self) -> Result<(), IntegrityJobError> {
if self.0.is_empty() || self.0.len() > MAX_INTEGRITY_SUBMISSION_KEY_BYTES {
return Err(IntegrityJobError::InvalidSubmissionKey);
}
Ok(())
}
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub(in crate::db) enum IntegrityCheckpoint {
QuickMetadata,
Rows(PhysicalUnitCheckpoint),
Index {
ordinal: u32,
checkpoint: PhysicalUnitCheckpoint,
},
ReverseRelation {
ordinal: u32,
checkpoint: PhysicalUnitCheckpoint,
},
Journal {
store_ordinal: u32,
checkpoint: JournalInspectionCheckpoint,
},
FinalProof,
}
impl IntegrityCheckpoint {
#[must_use]
pub(in crate::db) const fn phase(&self) -> IntegrityPhase {
match self {
Self::QuickMetadata => IntegrityPhase::QuickMetadata,
Self::Rows(_) => IntegrityPhase::Rows,
Self::Index { .. } => IntegrityPhase::IndexEntries,
Self::ReverseRelation { .. } => IntegrityPhase::ReverseRelations,
Self::Journal { .. } => IntegrityPhase::JournalTails,
Self::FinalProof => IntegrityPhase::FinalProofVectorCheck,
}
}
}
#[derive(CandidType, Clone, Copy, Debug, Deserialize, Eq, PartialEq)]
pub enum IntegrityPendingTerminal {
Expired,
Aborted,
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub enum IntegrityTerminalOutcome {
DeepCompleteClean,
DeepCompleteWithFindings,
Invalidated,
Uninspectable(IntegrityAuthorityDiagnostic),
ResourceLimited(IntegrityResourceDiagnostic),
Expired,
Aborted,
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub(in crate::db) enum IntegrityJobState {
InProgress,
TerminalPending(IntegrityPendingTerminal),
Terminal {
outcome: IntegrityTerminalOutcome,
receipt_acknowledged: bool,
},
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub enum DeepIntegrityPageStatus {
InProgress,
Terminal(IntegrityTerminalOutcome),
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub struct DeepIntegrityPage {
pub(super) job_id: IntegrityJobId,
pub(super) page_sequence: u64,
pub(super) phase: IntegrityPhase,
pub(super) status: DeepIntegrityPageStatus,
pub(super) pages_completed: u64,
pub(super) findings_seen: u64,
pub(super) findings: Vec<IntegrityFinding>,
pub(super) blocked_verifier_families: Vec<IntegrityVerifierFamily>,
}
impl DeepIntegrityPage {
#[must_use]
pub const fn job_id(&self) -> IntegrityJobId {
self.job_id
}
#[must_use]
pub const fn page_sequence(&self) -> u64 {
self.page_sequence
}
#[must_use]
pub const fn phase(&self) -> IntegrityPhase {
self.phase
}
#[must_use]
pub const fn status(&self) -> &DeepIntegrityPageStatus {
&self.status
}
#[must_use]
pub const fn pages_completed(&self) -> u64 {
self.pages_completed
}
#[must_use]
pub const fn findings_seen(&self) -> u64 {
self.findings_seen
}
#[must_use]
pub const fn findings(&self) -> &[IntegrityFinding] {
self.findings.as_slice()
}
#[must_use]
pub const fn blocked_verifier_families(&self) -> &[IntegrityVerifierFamily] {
self.blocked_verifier_families.as_slice()
}
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub enum IntegrityAbortStatus {
TerminationPending(IntegrityPendingTerminal),
Terminal(IntegrityTerminalOutcome),
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub struct IntegrityAbortReceipt {
pub(super) job_id: IntegrityJobId,
pub(super) page_sequence: u64,
pub(super) status: IntegrityAbortStatus,
}
impl IntegrityAbortReceipt {
#[must_use]
pub const fn job_id(&self) -> IntegrityJobId {
self.job_id
}
#[must_use]
pub const fn page_sequence(&self) -> u64 {
self.page_sequence
}
#[must_use]
pub const fn status(&self) -> &IntegrityAbortStatus {
&self.status
}
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub enum IntegrityJobReceipt {
Page(DeepIntegrityPage),
Abort(IntegrityAbortReceipt),
}
impl IntegrityJobReceipt {
#[must_use]
pub const fn job_id(&self) -> IntegrityJobId {
match self {
Self::Page(page) => page.job_id,
Self::Abort(receipt) => receipt.job_id,
}
}
#[must_use]
pub const fn page_sequence(&self) -> u64 {
match self {
Self::Page(page) => page.page_sequence,
Self::Abort(receipt) => receipt.page_sequence,
}
}
}
#[derive(CandidType, Clone, Copy, Debug, Deserialize, Eq, PartialEq)]
pub(in crate::db) enum IntegrityReceiptReplayKey {
Start,
Continue { acknowledged_sequence: u64 },
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub(in crate::db) struct IntegrityReceiptEnvelope {
pub(super) replay_key: IntegrityReceiptReplayKey,
pub(super) receipt: IntegrityJobReceipt,
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub(in crate::db) struct IntegrityJob {
pub(super) id: IntegrityJobId,
pub(super) database_incarnation_id: DatabaseIncarnationId,
pub(super) owner: IntegrityJobOwner,
pub(super) submission_key: IntegritySubmissionKey,
pub(super) entity: IntegrityEntityIdentity,
pub(super) accepted_schema_version: u32,
pub(super) accepted_schema_fingerprint: [u8; 16],
pub(super) inspection_plan_fingerprint: [u8; 32],
pub(super) checkpoint: IntegrityCheckpoint,
pub(super) captured_proof_vector: IntegrityProofVector,
pub(super) state: IntegrityJobState,
pub(super) lease_deadline_nanos: u64,
pub(super) findings_seen: u64,
pub(super) pages_completed: u64,
pub(super) blocked_verifier_families: Vec<IntegrityVerifierFamily>,
pub(super) last_receipt: IntegrityReceiptEnvelope,
}
impl IntegrityJob {
pub(super) fn validate(&self) -> Result<(), IntegrityJobError> {
if self.id != self.last_receipt.receipt.job_id()
|| self.database_incarnation_id != self.captured_proof_vector.database_incarnation_id()
|| self.accepted_schema_version != self.captured_proof_vector.accepted_schema_version()
|| self.accepted_schema_fingerprint
!= self.captured_proof_vector.accepted_schema_fingerprint()
|| self.inspection_plan_fingerprint
!= self.captured_proof_vector.inspection_plan_fingerprint()
|| self.entity.entity_path().is_empty()
|| self.entity.entity_path().len() > MAX_INTEGRITY_PATH_BYTES
|| self.entity.store_path().is_empty()
|| self.entity.store_path().len() > MAX_INTEGRITY_PATH_BYTES
|| self.entity.entity_tag() == 0
|| self.owner.as_str().is_empty()
|| self.owner.as_str().len() > MAX_INTEGRITY_OWNER_BYTES
|| self.submission_key.as_str().is_empty()
|| self.submission_key.as_str().len() > MAX_INTEGRITY_SUBMISSION_KEY_BYTES
|| self.accepted_schema_version == 0
|| self.lease_deadline_nanos == 0
|| matches!(
self.state,
IntegrityJobState::InProgress | IntegrityJobState::TerminalPending(_)
) && self.pages_completed > MAX_INTEGRITY_IN_PROGRESS_PAGES
|| !strictly_sorted_unique(&self.blocked_verifier_families)
|| self.captured_proof_vector.validate().is_err()
|| !self.checkpoint_is_well_formed()
{
return Err(IntegrityJobError::CorruptProgressRecord);
}
let receipt_matches_state = match (&self.state, &self.last_receipt.receipt) {
(
IntegrityJobState::InProgress | IntegrityJobState::TerminalPending(_),
IntegrityJobReceipt::Page(page),
) => {
page.status == DeepIntegrityPageStatus::InProgress
&& page.phase == self.checkpoint.phase()
&& self.page_matches_counters(page)
}
(IntegrityJobState::Terminal { outcome, .. }, IntegrityJobReceipt::Page(page)) => {
page.status == DeepIntegrityPageStatus::Terminal(outcome.clone())
&& page.phase == self.checkpoint.phase()
&& !matches!(
outcome,
IntegrityTerminalOutcome::Expired | IntegrityTerminalOutcome::Aborted
)
&& self.page_matches_counters(page)
}
(IntegrityJobState::Terminal { outcome, .. }, IntegrityJobReceipt::Abort(receipt)) => {
receipt.status == IntegrityAbortStatus::Terminal(outcome.clone())
&& matches!(
outcome,
IntegrityTerminalOutcome::Expired | IntegrityTerminalOutcome::Aborted
)
}
_ => false,
};
if !receipt_matches_state
|| self.last_receipt.receipt.page_sequence() != self.pages_completed
|| !self.replay_key_matches_receipt()
|| !self.terminal_outcome_matches_counts()
{
return Err(IntegrityJobError::CorruptProgressRecord);
}
Ok(())
}
fn page_matches_counters(&self, page: &DeepIntegrityPage) -> bool {
page.pages_completed == self.pages_completed
&& page.findings_seen == self.findings_seen
&& page.findings.len() <= MAX_INTEGRITY_RECEIPT_FINDINGS
&& u64::try_from(page.findings.len()).is_ok_and(|count| count <= self.findings_seen)
&& page.blocked_verifier_families == self.blocked_verifier_families
}
fn replay_key_matches_receipt(&self) -> bool {
match self.last_receipt.replay_key {
IntegrityReceiptReplayKey::Start => {
self.pages_completed == 0
&& self.last_receipt.receipt.page_sequence() == 0
&& matches!(
&self.last_receipt.receipt,
IntegrityJobReceipt::Page(DeepIntegrityPage {
status: DeepIntegrityPageStatus::InProgress,
..
})
)
}
IntegrityReceiptReplayKey::Continue {
acknowledged_sequence,
} => acknowledged_sequence
.checked_add(1)
.is_some_and(|sequence| sequence == self.last_receipt.receipt.page_sequence()),
}
}
const fn terminal_outcome_matches_counts(&self) -> bool {
match &self.state {
IntegrityJobState::Terminal {
outcome: IntegrityTerminalOutcome::DeepCompleteClean,
..
} => self.findings_seen == 0 && self.blocked_verifier_families.is_empty(),
IntegrityJobState::Terminal {
outcome: IntegrityTerminalOutcome::DeepCompleteWithFindings,
..
} => self.findings_seen > 0 || !self.blocked_verifier_families.is_empty(),
_ => true,
}
}
fn checkpoint_is_well_formed(&self) -> bool {
match &self.checkpoint {
IntegrityCheckpoint::Rows(checkpoint) => row_checkpoint_is_well_formed(checkpoint),
IntegrityCheckpoint::Index {
ordinal,
checkpoint,
} => {
usize::try_from(*ordinal).is_ok_and(|ordinal| {
ordinal < self.captured_proof_vector.index_generation_count()
}) && index_checkpoint_is_well_formed(checkpoint)
}
IntegrityCheckpoint::ReverseRelation {
ordinal,
checkpoint,
} => {
usize::try_from(*ordinal).is_ok_and(|ordinal| {
ordinal < self.captured_proof_vector.relation_generation_count()
}) && reverse_checkpoint_is_well_formed(checkpoint)
}
IntegrityCheckpoint::Journal {
store_ordinal,
checkpoint,
} => usize::try_from(*store_ordinal)
.ok()
.and_then(|ordinal| self.captured_proof_vector.stores().get(ordinal))
.is_some_and(|proof| {
let (fold_sequence, next_append_sequence) = proof.journal_interval();
journal_checkpoint_is_well_formed(
checkpoint,
fold_sequence,
next_append_sequence,
)
}),
IntegrityCheckpoint::QuickMetadata | IntegrityCheckpoint::FinalProof => true,
}
}
}
fn row_checkpoint_is_well_formed(checkpoint: &PhysicalUnitCheckpoint) -> bool {
checkpoint.raw_data_key().is_ok()
&& !matches!(
checkpoint,
PhysicalUnitCheckpoint::Within {
verifier_family: IntegrityVerifierFamily::IndexEntry
| IntegrityVerifierFamily::UniqueIndex
| IntegrityVerifierFamily::ReverseRelationEntry
| IntegrityVerifierFamily::JournalEnvelope
| IntegrityVerifierFamily::JournalBatchIdentity,
..
}
)
}
fn index_checkpoint_is_well_formed(checkpoint: &PhysicalUnitCheckpoint) -> bool {
checkpoint.raw_index_key().is_ok()
&& !matches!(
checkpoint,
PhysicalUnitCheckpoint::Within {
verifier_family: IntegrityVerifierFamily::DataKey
| IntegrityVerifierFamily::RowEnvelope
| IntegrityVerifierFamily::FieldValue
| IntegrityVerifierFamily::PrimaryKey
| IntegrityVerifierFamily::ValidatedConstraints
| IntegrityVerifierFamily::ForwardIndex
| IntegrityVerifierFamily::Relation
| IntegrityVerifierFamily::ReverseRelationEntry
| IntegrityVerifierFamily::JournalEnvelope
| IntegrityVerifierFamily::JournalBatchIdentity,
..
}
)
}
fn reverse_checkpoint_is_well_formed(checkpoint: &PhysicalUnitCheckpoint) -> bool {
checkpoint.raw_index_key().is_ok()
&& !matches!(
checkpoint,
PhysicalUnitCheckpoint::Within {
verifier_family: IntegrityVerifierFamily::DataKey
| IntegrityVerifierFamily::RowEnvelope
| IntegrityVerifierFamily::FieldValue
| IntegrityVerifierFamily::PrimaryKey
| IntegrityVerifierFamily::ValidatedConstraints
| IntegrityVerifierFamily::ForwardIndex
| IntegrityVerifierFamily::Relation
| IntegrityVerifierFamily::IndexEntry
| IntegrityVerifierFamily::UniqueIndex
| IntegrityVerifierFamily::JournalEnvelope
| IntegrityVerifierFamily::JournalBatchIdentity,
..
}
)
}
const fn journal_checkpoint_is_well_formed(
checkpoint: &JournalInspectionCheckpoint,
fold_sequence: u64,
next_append_sequence: u64,
) -> bool {
match checkpoint {
JournalInspectionCheckpoint::BeforeFirst => true,
JournalInspectionCheckpoint::BeforeBatch { sequence } => {
*sequence > fold_sequence && *sequence < next_append_sequence
}
JournalInspectionCheckpoint::CheckingBatchIdentity {
sequence,
next_prior_sequence,
..
} => {
*sequence > fold_sequence
&& *sequence < next_append_sequence
&& *next_prior_sequence > fold_sequence
&& *next_prior_sequence < *sequence
}
JournalInspectionCheckpoint::AfterBatch { sequence } => {
*sequence >= fold_sequence && *sequence < next_append_sequence
}
}
}
fn strictly_sorted_unique(values: &[IntegrityVerifierFamily]) -> bool {
values.windows(2).all(|pair| pair[0] < pair[1])
}
#[derive(CandidType, Clone, Copy, Debug, Deserialize, Eq, PartialEq)]
pub enum IntegrityJobError {
CapacityExceeded,
CorruptProgressHeader,
CorruptProgressRecord,
CounterExhausted,
EntityIdentityMismatch,
IncompatibleProgressFormat,
Internal,
InvalidEntityIdentity,
InvalidJobId,
InvalidOwner,
InvalidSubmissionKey,
JobIncarnationMismatch,
JobNotFound,
JobOwnerMismatch,
StaleAcknowledgement,
StartInvalidated,
SubmissionAlreadyAdvanced,
SubmissionConflict,
}
#[derive(Debug)]
pub enum IntegrityDeepError {
Internal(crate::error::InternalError),
Job(IntegrityJobError),
Uninspectable(IntegrityAuthorityDiagnostic),
}
impl From<IntegrityJobError> for IntegrityDeepError {
fn from(error: IntegrityJobError) -> Self {
Self::Job(error)
}
}
impl From<crate::error::InternalError> for IntegrityDeepError {
fn from(error: crate::error::InternalError) -> Self {
Self::Internal(error)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn public_job_inputs_revalidate_after_wire_decode() {
let owner_bytes =
candid::encode_one(IntegrityJobOwner(String::new())).expect("owner should encode");
let owner: IntegrityJobOwner =
candid::decode_one(&owner_bytes).expect("owner should decode");
assert_eq!(owner.validate(), Err(IntegrityJobError::InvalidOwner));
let submission_bytes = candid::encode_one(IntegritySubmissionKey(String::new()))
.expect("submission key should encode");
let submission: IntegritySubmissionKey =
candid::decode_one(&submission_bytes).expect("submission key should decode");
assert_eq!(
submission.validate(),
Err(IntegrityJobError::InvalidSubmissionKey),
);
let job_id_bytes =
candid::encode_one(IntegrityJobId([0; 32])).expect("job id should encode");
let job_id: IntegrityJobId =
candid::decode_one(&job_id_bytes).expect("job id should decode");
assert_eq!(job_id.validate(), Err(IntegrityJobError::InvalidJobId),);
}
#[test]
fn public_job_id_hex_round_trip_is_exact_and_fail_closed() {
let mut bytes = [0_u8; 32];
bytes[0] = 0x01;
bytes[31] = 0xfe;
let job_id = IntegrityJobId::try_from_bytes(bytes).expect("job id should admit");
let encoded = job_id.to_hex();
assert_eq!(encoded.len(), 64);
assert_eq!(IntegrityJobId::try_from_hex(encoded.as_str()), Ok(job_id),);
assert_eq!(
IntegrityJobId::try_from_hex(encoded.to_uppercase().as_str()),
Ok(job_id),
);
for malformed in [
"",
"01",
"0000000000000000000000000000000000000000000000000000000000000000",
"g001000000000000000000000000000000000000000000000000000000000000",
] {
assert_eq!(
IntegrityJobId::try_from_hex(malformed),
Err(IntegrityJobError::InvalidJobId),
);
}
}
#[test]
fn persisted_checkpoint_families_stay_phase_owned() {
let journal_in_row = PhysicalUnitCheckpoint::Within {
physical_key: vec![1],
verifier_family: IntegrityVerifierFamily::JournalEnvelope,
ordinal: 0,
};
let row_in_index = PhysicalUnitCheckpoint::Within {
physical_key: vec![1],
verifier_family: IntegrityVerifierFamily::FieldValue,
ordinal: 0,
};
let reverse_in_reverse = PhysicalUnitCheckpoint::Within {
physical_key: vec![1],
verifier_family: IntegrityVerifierFamily::ReverseRelationEntry,
ordinal: 0,
};
assert!(!row_checkpoint_is_well_formed(&journal_in_row));
assert!(!index_checkpoint_is_well_formed(&row_in_index));
assert!(reverse_checkpoint_is_well_formed(&reverse_in_reverse));
}
#[test]
fn persisted_journal_checkpoint_cannot_skip_the_captured_tail_interval() {
assert!(journal_checkpoint_is_well_formed(
&JournalInspectionCheckpoint::BeforeFirst,
4,
8,
));
assert!(journal_checkpoint_is_well_formed(
&JournalInspectionCheckpoint::BeforeBatch { sequence: 7 },
4,
8,
));
assert!(journal_checkpoint_is_well_formed(
&JournalInspectionCheckpoint::AfterBatch { sequence: 4 },
4,
8,
));
assert!(!journal_checkpoint_is_well_formed(
&JournalInspectionCheckpoint::BeforeBatch { sequence: 4 },
4,
8,
));
assert!(!journal_checkpoint_is_well_formed(
&JournalInspectionCheckpoint::AfterBatch { sequence: 8 },
4,
8,
));
assert!(!journal_checkpoint_is_well_formed(
&JournalInspectionCheckpoint::CheckingBatchIdentity {
sequence: 7,
batch_id: [1; 16],
next_prior_sequence: 4,
},
4,
8,
));
}
}