use crate::db::codec::{
finalize_hash_sha256, new_hash_sha256_prefixed, write_hash_str_u32, write_hash_tag_u8,
write_hash_u64,
};
use crate::error::{ConstraintDiagnostic, InternalError};
use candid::CandidType;
use icydb_schema::{
ExpectedAcceptedHead, SchemaProposalDigest, SchemaSubmissionKey, TargetDatabaseIdentity,
TargetStoreIdentity,
};
use serde::Deserialize;
use sha2::Digest;
const SCHEMA_CHANGE_JOB_ID_PROFILE: &[u8] = b"icydb.schema-application.job-id.v1";
const MAX_SCHEMA_CHANGE_ACTIVATIONS: usize = 512;
#[derive(CandidType, Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd)]
#[serde(transparent)]
pub struct SchemaChangeJobId([u8; 32]);
impl SchemaChangeJobId {
fn from_bytes(bytes: [u8; 32]) -> Result<Self, InternalError> {
if bytes == [0; 32] {
return Err(InternalError::store_corruption());
}
Ok(Self(bytes))
}
#[must_use]
pub const fn to_bytes(self) -> [u8; 32] {
self.0
}
fn validate(self) -> Result<(), InternalError> {
if self.0 == [0; 32] {
return Err(InternalError::store_corruption());
}
Ok(())
}
}
#[derive(CandidType, Clone, Copy, Debug, Deserialize, Eq, PartialEq)]
pub struct SchemaChangeJob {
id: SchemaChangeJobId,
}
impl SchemaChangeJob {
pub(in crate::db) const fn new(id: SchemaChangeJobId) -> Self {
Self { id }
}
#[must_use]
pub const fn id(self) -> SchemaChangeJobId {
self.id
}
}
#[derive(CandidType, Clone, Copy, Debug, Deserialize, Eq, PartialEq)]
pub enum SchemaChangeValidationPhase {
Forward,
Verify,
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub enum SchemaChangeProgressStatus {
Started,
Advanced {
phase: SchemaChangeValidationPhase,
rows_scanned: u64,
},
Findings {
phase: SchemaChangeValidationPhase,
rows_scanned: u64,
page_sequence: u64,
findings: Vec<ConstraintDiagnostic>,
},
Restarted {
rows_scanned: u64,
},
Applied,
Aborted,
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub struct SchemaChangeProgress {
receipt: SchemaChangeReceipt,
status: SchemaChangeProgressStatus,
}
impl SchemaChangeProgress {
pub(in crate::db) const fn new(
receipt: SchemaChangeReceipt,
status: SchemaChangeProgressStatus,
) -> Self {
Self { receipt, status }
}
#[must_use]
pub const fn receipt(&self) -> &SchemaChangeReceipt {
&self.receipt
}
#[must_use]
pub const fn status(&self) -> &SchemaChangeProgressStatus {
&self.status
}
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub enum SchemaChangeOutcome {
NoOp {
accepted_head: ExpectedAcceptedHead,
},
Applied {
accepted_head: ExpectedAcceptedHead,
},
Pending {
job: SchemaChangeJob,
candidate_head: ExpectedAcceptedHead,
},
Aborted {
accepted_head: ExpectedAcceptedHead,
},
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub struct SchemaChangeReceipt {
database_identity: TargetDatabaseIdentity,
submission_key: SchemaSubmissionKey,
proposal_digest: SchemaProposalDigest,
prior_head: ExpectedAcceptedHead,
outcome: SchemaChangeOutcome,
}
impl SchemaChangeReceipt {
pub(in crate::db) fn new(
database_identity: TargetDatabaseIdentity,
submission_key: SchemaSubmissionKey,
proposal_digest: SchemaProposalDigest,
prior_head: ExpectedAcceptedHead,
outcome: SchemaChangeOutcome,
) -> Result<Self, InternalError> {
let receipt = Self {
database_identity,
submission_key,
proposal_digest,
prior_head,
outcome,
};
receipt.validate()?;
Ok(receipt)
}
#[must_use]
pub const fn database_identity(&self) -> TargetDatabaseIdentity {
self.database_identity
}
#[must_use]
pub const fn submission_key(&self) -> &SchemaSubmissionKey {
&self.submission_key
}
#[must_use]
pub const fn proposal_digest(&self) -> SchemaProposalDigest {
self.proposal_digest
}
#[must_use]
pub const fn prior_head(&self) -> &ExpectedAcceptedHead {
&self.prior_head
}
#[must_use]
pub const fn outcome(&self) -> &SchemaChangeOutcome {
&self.outcome
}
pub(in crate::db) fn is_exact_submission(
&self,
database_identity: TargetDatabaseIdentity,
submission_key: &SchemaSubmissionKey,
proposal_digest: SchemaProposalDigest,
prior_head: &ExpectedAcceptedHead,
) -> bool {
self.database_identity == database_identity
&& &self.submission_key == submission_key
&& self.proposal_digest == proposal_digest
&& &self.prior_head == prior_head
}
pub(in crate::db) fn validate(&self) -> Result<(), InternalError> {
if self.database_identity.to_bytes() == [0; 32]
|| self.proposal_digest.to_bytes() == [0; 32]
{
return Err(InternalError::store_corruption());
}
validate_head(&self.prior_head, true)?;
match &self.outcome {
SchemaChangeOutcome::NoOp { accepted_head } => {
validate_head(accepted_head, true)?;
if accepted_head != &self.prior_head {
return Err(InternalError::store_corruption());
}
}
SchemaChangeOutcome::Applied { accepted_head }
| SchemaChangeOutcome::Aborted { accepted_head } => {
validate_head(accepted_head, false)?;
if accepted_head == &self.prior_head {
return Err(InternalError::store_corruption());
}
}
SchemaChangeOutcome::Pending {
job,
candidate_head,
} => {
job.id.validate()?;
validate_head(candidate_head, false)?;
if candidate_head == &self.prior_head
|| job.id
!= derive_schema_change_job_id(
self.database_identity,
&self.submission_key,
self.proposal_digest,
&self.prior_head,
)?
{
return Err(InternalError::store_corruption());
}
}
}
Ok(())
}
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub(in crate::db) struct SchemaChangeActivation {
store: TargetStoreIdentity,
entity_tag: u64,
constraint_id: u32,
}
impl SchemaChangeActivation {
pub(in crate::db) fn new(
store: TargetStoreIdentity,
entity_tag: u64,
constraint_id: u32,
) -> Result<Self, InternalError> {
if entity_tag == 0 || constraint_id == 0 {
return Err(InternalError::store_invariant());
}
Ok(Self {
store,
entity_tag,
constraint_id,
})
}
pub(in crate::db) const fn store(&self) -> TargetStoreIdentity {
self.store
}
pub(in crate::db) const fn entity_tag(&self) -> u64 {
self.entity_tag
}
pub(in crate::db) const fn constraint_id(&self) -> u32 {
self.constraint_id
}
fn validate(&self) -> Result<(), InternalError> {
if self.store.to_bytes() == [0; 32] || self.entity_tag == 0 || self.constraint_id == 0 {
return Err(InternalError::store_corruption());
}
Ok(())
}
}
#[derive(CandidType, Clone, Debug, Deserialize, Eq, PartialEq)]
pub(in crate::db) struct SchemaApplicationRecord {
receipt: SchemaChangeReceipt,
activations: Vec<SchemaChangeActivation>,
}
impl SchemaApplicationRecord {
pub(in crate::db) fn new(
receipt: SchemaChangeReceipt,
activations: Vec<SchemaChangeActivation>,
) -> Result<Self, InternalError> {
let record = Self {
receipt,
activations,
};
record.validate()?;
Ok(record)
}
pub(in crate::db) const fn receipt(&self) -> &SchemaChangeReceipt {
&self.receipt
}
pub(in crate::db) const fn activations(&self) -> &[SchemaChangeActivation] {
self.activations.as_slice()
}
pub(in crate::db) fn validate(&self) -> Result<(), InternalError> {
self.receipt.validate()?;
if self.activations.len() > MAX_SCHEMA_CHANGE_ACTIVATIONS
|| self.activations.windows(2).any(|pair| {
(pair[0].store, pair[0].entity_tag, pair[0].constraint_id)
>= (pair[1].store, pair[1].entity_tag, pair[1].constraint_id)
})
|| self
.activations
.iter()
.any(|activation| activation.validate().is_err())
{
return Err(InternalError::store_corruption());
}
let pending = matches!(self.receipt.outcome(), SchemaChangeOutcome::Pending { .. });
if pending == self.activations.is_empty() {
return Err(InternalError::store_corruption());
}
Ok(())
}
}
pub(in crate::db) fn derive_schema_change_job_id(
database_identity: TargetDatabaseIdentity,
submission_key: &SchemaSubmissionKey,
proposal_digest: SchemaProposalDigest,
prior_head: &ExpectedAcceptedHead,
) -> Result<SchemaChangeJobId, InternalError> {
validate_head(prior_head, true)?;
let mut hasher = new_hash_sha256_prefixed(SCHEMA_CHANGE_JOB_ID_PROFILE);
hasher.update(database_identity.to_bytes());
write_hash_str_u32(&mut hasher, submission_key.as_str());
hasher.update(proposal_digest.to_bytes());
write_head(&mut hasher, prior_head);
SchemaChangeJobId::from_bytes(finalize_hash_sha256(hasher))
}
fn validate_head(head: &ExpectedAcceptedHead, empty_allowed: bool) -> Result<(), InternalError> {
match head {
ExpectedAcceptedHead::Empty if empty_allowed => Ok(()),
ExpectedAcceptedHead::Exact { revision: 0, .. } | ExpectedAcceptedHead::Empty => {
Err(InternalError::store_corruption())
}
ExpectedAcceptedHead::Exact { fingerprint, .. } if fingerprint.to_bytes() == [0; 32] => {
Err(InternalError::store_corruption())
}
ExpectedAcceptedHead::Exact { .. } => Ok(()),
}
}
fn write_head(hasher: &mut sha2::Sha256, head: &ExpectedAcceptedHead) {
match head {
ExpectedAcceptedHead::Empty => write_hash_tag_u8(hasher, 0),
ExpectedAcceptedHead::Exact {
revision,
fingerprint,
} => {
write_hash_tag_u8(hasher, 1);
write_hash_u64(hasher, *revision);
hasher.update(fingerprint.to_bytes());
}
}
}
#[cfg(test)]
mod tests {
use super::{
SchemaApplicationRecord, SchemaChangeActivation, SchemaChangeJob, SchemaChangeOutcome,
SchemaChangeReceipt, derive_schema_change_job_id,
};
use icydb_schema::{
ExpectedAcceptedHead, ExpectedSchemaFingerprint, SchemaProposalDigest, SchemaSubmissionKey,
TargetDatabaseIdentity, TargetStoreIdentity,
};
#[test]
fn schema_change_job_identity_covers_the_complete_idempotency_tuple() {
let database = TargetDatabaseIdentity::from_bytes([0x11; 32]);
let submission =
SchemaSubmissionKey::try_new("job-id").expect("submission key should admit");
let digest = SchemaProposalDigest::from_bytes([0x22; 32]);
let empty = derive_schema_change_job_id(
database,
&submission,
digest,
&ExpectedAcceptedHead::Empty,
)
.expect("empty-head identity should derive");
let exact = derive_schema_change_job_id(
database,
&submission,
digest,
&ExpectedAcceptedHead::Exact {
revision: 1,
fingerprint: ExpectedSchemaFingerprint::from_bytes([0x33; 32]),
},
)
.expect("exact-head identity should derive");
assert_ne!(empty, exact);
}
#[test]
fn pending_and_terminal_record_state_is_exact() {
let database = TargetDatabaseIdentity::from_bytes([0x11; 32]);
let submission =
SchemaSubmissionKey::try_new("state").expect("submission key should admit");
let digest = SchemaProposalDigest::from_bytes([0x22; 32]);
let head = ExpectedAcceptedHead::Empty;
let job = SchemaChangeJob::new(
derive_schema_change_job_id(database, &submission, digest, &head)
.expect("job identity should derive"),
);
let pending = SchemaChangeReceipt::new(
database,
submission.clone(),
digest,
head.clone(),
SchemaChangeOutcome::Pending {
job,
candidate_head: ExpectedAcceptedHead::Exact {
revision: 1,
fingerprint: ExpectedSchemaFingerprint::from_bytes([0x33; 32]),
},
},
)
.expect("pending receipt should admit");
assert!(
SchemaApplicationRecord::new(pending.clone(), Vec::new()).is_err(),
"pending records require exact activation ownership",
);
let activation =
SchemaChangeActivation::new(TargetStoreIdentity::from_bytes([0x44; 32]), 1, 1)
.expect("activation should admit");
SchemaApplicationRecord::new(pending, vec![activation])
.expect("pending record with activation should admit");
let terminal = SchemaChangeReceipt::new(
database,
submission,
digest,
head,
SchemaChangeOutcome::Aborted {
accepted_head: ExpectedAcceptedHead::Exact {
revision: 1,
fingerprint: ExpectedSchemaFingerprint::from_bytes([0x55; 32]),
},
},
)
.expect("terminal receipt should admit");
assert!(
SchemaApplicationRecord::new(
terminal,
vec![
SchemaChangeActivation::new(TargetStoreIdentity::from_bytes([0x44; 32]), 1, 1,)
.expect("activation should admit"),
],
)
.is_err(),
"terminal records cannot retain activation state",
);
}
#[test]
fn schema_change_receipt_outcome_heads_have_exact_temporal_closure() {
let database = TargetDatabaseIdentity::from_bytes([0x11; 32]);
let digest = SchemaProposalDigest::from_bytes([0x22; 32]);
let prior = ExpectedAcceptedHead::Exact {
revision: 7,
fingerprint: ExpectedSchemaFingerprint::from_bytes([0x33; 32]),
};
let changed = ExpectedAcceptedHead::Exact {
revision: 8,
fingerprint: ExpectedSchemaFingerprint::from_bytes([0x44; 32]),
};
SchemaChangeReceipt::new(
database,
SchemaSubmissionKey::try_new("noop").expect("submission key should admit"),
digest,
prior.clone(),
SchemaChangeOutcome::NoOp {
accepted_head: prior.clone(),
},
)
.expect("no-op must retain the exact prior head");
assert!(
SchemaChangeReceipt::new(
database,
SchemaSubmissionKey::try_new("invalid-noop").expect("submission key should admit"),
digest,
prior.clone(),
SchemaChangeOutcome::NoOp {
accepted_head: changed.clone(),
},
)
.is_err(),
);
assert!(
SchemaChangeReceipt::new(
database,
SchemaSubmissionKey::try_new("invalid-applied")
.expect("submission key should admit"),
digest,
prior.clone(),
SchemaChangeOutcome::Applied {
accepted_head: prior.clone(),
},
)
.is_err(),
);
SchemaChangeReceipt::new(
database,
SchemaSubmissionKey::try_new("applied").expect("submission key should admit"),
digest,
prior.clone(),
SchemaChangeOutcome::Applied {
accepted_head: changed.clone(),
},
)
.expect("applied receipt must identify a different exact head");
assert!(
SchemaChangeReceipt::new(
database,
SchemaSubmissionKey::try_new("invalid-abort").expect("submission key should admit"),
digest,
prior.clone(),
SchemaChangeOutcome::Aborted {
accepted_head: prior.clone(),
},
)
.is_err(),
);
SchemaChangeReceipt::new(
database,
SchemaSubmissionKey::try_new("aborted").expect("submission key should admit"),
digest,
prior,
SchemaChangeOutcome::Aborted {
accepted_head: changed,
},
)
.expect("aborted receipt must retain the post-abort accepted head");
}
}