use chrono::{DateTime, Utc};
use super::cache;
use super::event::Event;
use super::kel::{GitKel, KelError};
use super::state::KeyState;
use super::types::Said;
use crate::domain::EventHash;
pub const MAX_INCREMENTAL_EVENTS: usize = 10_000;
#[derive(Debug)]
pub enum IncrementalResult {
CacheHit(KeyState),
IncrementalSuccess {
state: KeyState,
events_validated: usize,
},
NeedsFullReplay(ReplayReason),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReplayReason {
NoCacheFile,
CacheCorrupt,
CachedCommitMissing,
CachedCommitNotInAncestry,
CacheSaidMismatch,
TooManyEvents,
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum IncrementalError {
#[error("KEL error: {0}")]
Kel(#[from] KelError),
#[error("Chain continuity error: expected previous SAID {expected}, got {actual}")]
ChainContinuity { expected: Said, actual: Said },
#[error("Sequence error: expected {expected}, got {actual}")]
SequenceError { expected: u128, actual: u128 },
#[error("Malformed sequence number: {raw:?}")]
MalformedSequence { raw: String },
#[error("Invalid event type in KEL: {0}")]
InvalidEventType(String),
#[error("KEL history is non-linear: commit {commit} has {parent_count} parents (expected 1)")]
NonLinearHistory { commit: String, parent_count: usize },
#[error("KEL history is corrupted: commit {commit} has no parent but is not inception")]
MissingParent { commit: String },
}
impl auths_core::error::AuthsErrorInfo for IncrementalError {
fn error_code(&self) -> &'static str {
match self {
Self::Kel(_) => "AUTHS-E4951",
Self::ChainContinuity { .. } => "AUTHS-E4952",
Self::SequenceError { .. } => "AUTHS-E4953",
Self::MalformedSequence { .. } => "AUTHS-E4954",
Self::InvalidEventType(_) => "AUTHS-E4955",
Self::NonLinearHistory { .. } => "AUTHS-E4956",
Self::MissingParent { .. } => "AUTHS-E4957",
}
}
fn suggestion(&self) -> Option<&'static str> {
match self {
Self::Kel(_) => None,
Self::ChainContinuity { .. } => {
Some("The KEL chain is broken; clear the cache and retry")
}
Self::SequenceError { .. } => {
Some("The KEL has sequence gaps; re-sync from a trusted source")
}
Self::MalformedSequence { .. } => None,
Self::InvalidEventType(_) => None,
Self::NonLinearHistory { .. } => {
Some("The KEL has merge commits, indicating tampering")
}
Self::MissingParent { .. } => Some("The KEL commit history is corrupted"),
}
}
}
pub fn try_incremental_validation<'a>(
kel: &GitKel<'a>,
did: &str,
now: DateTime<Utc>,
) -> Result<IncrementalResult, IncrementalError> {
let tip_hash = kel.tip_commit_hash()?;
let tip_event = kel.read_event_from_commit_hash(tip_hash)?;
let tip_said = tip_event.said();
let cached = match cache::try_load_cached_state_full(kel.workdir(), did) {
Some(c) => c,
None => {
log::debug!("KEL cache miss for {}: no cache file", did);
return Ok(IncrementalResult::NeedsFullReplay(
ReplayReason::NoCacheFile,
));
}
};
if cached.validated_against_tip_said == *tip_said {
log::debug!("KEL cache hit for {}", did);
return Ok(IncrementalResult::CacheHit(cached.state));
}
let cached_hash = match GitKel::parse_hash(cached.last_commit_oid.as_str()) {
Ok(h) => h,
Err(_) => {
log::debug!("KEL cache corrupt for {}: invalid commit hash", did);
return Ok(IncrementalResult::NeedsFullReplay(
ReplayReason::CacheCorrupt,
));
}
};
if !kel.commit_exists(cached_hash) {
log::debug!(
"KEL cache miss for {}: cached commit {} doesn't exist",
did,
cached_hash
);
return Ok(IncrementalResult::NeedsFullReplay(
ReplayReason::CachedCommitMissing,
));
}
let cached_event = kel.read_event_from_commit_hash(cached_hash)?;
if *cached_event.said() != cached.validated_against_tip_said {
log::warn!(
"KEL cache SAID mismatch for {}: cache says {} but commit has {}",
did,
cached.validated_against_tip_said,
cached_event.said()
);
return Ok(IncrementalResult::NeedsFullReplay(
ReplayReason::CacheSaidMismatch,
));
}
let delta = match build_delta_with_linearity_check(kel, tip_hash, cached_hash)? {
Some(d) => d,
None => {
log::debug!("KEL cache miss for {}: cached commit not in ancestry", did);
return Ok(IncrementalResult::NeedsFullReplay(
ReplayReason::CachedCommitNotInAncestry,
));
}
};
if delta.len() > MAX_INCREMENTAL_EVENTS {
log::info!(
"KEL incremental skip for {}: {} events exceeds threshold {}",
did,
delta.len(),
MAX_INCREMENTAL_EVENTS
);
return Ok(IncrementalResult::NeedsFullReplay(
ReplayReason::TooManyEvents,
));
}
log::debug!(
"KEL incremental validation for {}: {} new events",
did,
delta.len()
);
let mut state = cached.state;
let events_validated = delta.len();
for hash in delta {
let event = kel.read_event_from_commit_hash(hash)?;
apply_event_to_state(&mut state, &event)?;
}
if state.last_event_said != *tip_said {
return Err(IncrementalError::ChainContinuity {
expected: tip_said.clone(),
actual: state.last_event_said.clone(),
});
}
let _ = cache::write_kel_cache(
kel.workdir(),
did,
&state,
tip_said.as_str(),
&tip_hash.to_hex(),
now,
);
Ok(IncrementalResult::IncrementalSuccess {
state,
events_validated,
})
}
fn build_delta_with_linearity_check(
kel: &GitKel<'_>,
tip_hash: EventHash,
cached_hash: EventHash,
) -> Result<Option<Vec<EventHash>>, IncrementalError> {
if tip_hash == cached_hash {
return Ok(Some(Vec::new()));
}
let mut delta = Vec::new();
let mut current = tip_hash;
loop {
delta.push(current);
let parent_count = kel.parent_count(current)?;
if parent_count > 1 {
return Err(IncrementalError::NonLinearHistory {
commit: current.to_hex(),
parent_count,
});
}
if parent_count == 0 {
return Ok(None);
}
#[allow(clippy::expect_used)]
let parent = kel.parent_hash(current)?.expect("parent_count was 1");
if parent == cached_hash {
break;
}
current = parent;
}
delta.reverse();
Ok(Some(delta))
}
fn apply_event_to_state(state: &mut KeyState, event: &Event) -> Result<(), IncrementalError> {
let expected_sequence = state.sequence + 1;
let actual_sequence = event.sequence().value();
if actual_sequence != expected_sequence {
return Err(IncrementalError::SequenceError {
expected: expected_sequence,
actual: actual_sequence,
});
}
if let Some(prev_said) = event.previous() {
if *prev_said != state.last_event_said {
return Err(IncrementalError::ChainContinuity {
expected: state.last_event_said.clone(),
actual: prev_said.clone(),
});
}
} else {
return Err(IncrementalError::InvalidEventType(
"Unexpected inception event in incremental validation".to_string(),
));
}
match event {
Event::Rot(rot) => {
state.apply_rotation(
rot.k.clone(),
rot.n.clone(),
rot.kt.clone(),
rot.nt.clone(),
actual_sequence,
rot.d.clone(),
&rot.br,
&rot.ba,
rot.bt.clone(),
rot.c.clone(),
);
}
Event::Ixn(ixn) => {
state.apply_interaction(actual_sequence, ixn.d.clone());
}
Event::Icp(_) | Event::Dip(_) => {
return Err(IncrementalError::InvalidEventType(
"Inception event after KEL start".to_string(),
));
}
Event::Drt(_) => {
return Err(IncrementalError::InvalidEventType(
"Delegated rotation not yet supported".to_string(),
));
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::keri::Prefix;
use crate::keri::{CesrKey, Threshold};
#[test]
fn test_incremental_result_variants() {
let state = KeyState::from_inception(
Prefix::new_unchecked("ETest".to_string()),
vec![CesrKey::new_unchecked("DKey".to_string())],
vec![Said::new_unchecked("ENext".to_string())],
Threshold::Simple(1),
Threshold::Simple(1),
Said::new_unchecked("ESaid".to_string()),
vec![],
Threshold::Simple(0),
vec![],
);
let _hit = IncrementalResult::CacheHit(state.clone());
let _success = IncrementalResult::IncrementalSuccess {
state,
events_validated: 5,
};
let _miss = IncrementalResult::NeedsFullReplay(ReplayReason::NoCacheFile);
}
#[test]
fn test_replay_reasons() {
assert_ne!(ReplayReason::NoCacheFile, ReplayReason::CacheCorrupt);
assert_ne!(
ReplayReason::CachedCommitMissing,
ReplayReason::CacheSaidMismatch
);
assert_ne!(
ReplayReason::TooManyEvents,
ReplayReason::CachedCommitNotInAncestry
);
}
}