WEBDIENSTE/NLnetLabs-krill
1
0
Fork 0
mirror of https://github.com/NLnetLabs/krill.git synced 2026-09-13 13:07:40 +02:00
Code Issues Packages Projects Releases Wiki Activity

Simplify event replay logic - do not rely on info file for reply without limit.

Browse Source
This commit is contained in:
Tim Bruijnzeels
2022-02-23 14:17:18 +01:00
parent 808905efa7
commit 1f46abf8f4
3 changed files with 27 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/commons/eventsourcing/store.rs
+25 -40
View File
@@ -894,51 +894,36 @@ where
fn update_aggregate(&self, id: &Handle, aggregate: &mut A, limit: Option<u64>) -> Result<(), AggregateStoreError> {
let start = aggregate.version();
let limit = if let Some(limit) = limit {
debug!("Will attempt to update '{}' using explicit limit", id);
limit
} else if let Ok(info) = self.get_info(id) {
debug!("Will attempt to update '{}' using limit from info", id);
info.last_event
if let Some(limit) = limit {
debug!("Will update '{}' from version: {} to: {}", id, start, limit + 1);
} else {
let nr_events = self.kv.keys(Some(id.to_string()), "delta-")?.len();
if nr_events < 1 {
return Err(AggregateStoreError::InfoMissing(id.clone()));
} else {
let limit = (nr_events - 1) as u64;
debug!("Will attempt to update '{}' from limit based on nr of events", id,);
limit
debug!("Will update '{}' to latest version", id);
}
// check and apply any applicable events until:
// - the limit is reached (if supplied)
// - there are no more events
// - any event cannot be applied (return an error)
loop {
let version = aggregate.version();
if let Some(limit) = limit {
if limit == version - 1 {
debug!("Updated '{}' to: {}", id, version);
break;
}
}
};
if limit == aggregate.version() - 1 {
// already at version, done
// note that an event has the version of the aggregate it *affects*. So delta 10 results in version 11.
debug!("Snapshot for '{}' is up to date", id);
return Ok(());
}
debug!(
"Will attempt to update '{}' from version: {} to: {}",
id,
start,
limit + 1
);
if start > limit {
return Err(AggregateStoreError::ReplayError(id.clone(), limit, start));
}
for version in start..limit + 1 {
if let Some(e) = self.get_event(id, version)? {
if aggregate.version() != version {
if let Some(e) = self.get_event::<A::Event>(id, version)? {
if version != e.version() {
error!("Trying to apply event to wrong version of aggregate in replay");
return Err(AggregateStoreError::ReplayError(id.clone(), limit, version));
return Err(AggregateStoreError::ReplayError(id.clone(), version, e.version()));
}
aggregate.apply(e);
debug!("Applied event nr {} to aggregate {}", version, id);
} else {
return Err(AggregateStoreError::ReplayError(id.clone(), limit, version));
debug!("No more events found. updated '{}' to: {}", id, version);
break;
}
}
@@ -1023,10 +1008,10 @@ impl fmt::Display for AggregateStoreError {
AggregateStoreError::InitError(handle) => {
write!(f, "Init event exists for '{}', but cannot be applied", handle)
}
AggregateStoreError::ReplayError(handle, target_version, fail_version) => write!(
AggregateStoreError::ReplayError(handle, version, fail_version) => write!(
f,
"Cannot reconstruct '{}' to version '{}', failed at version {}",
handle, target_version, fail_version
"Event for '{}' version '{}' had version '{}'",
handle, version, fail_version
),
AggregateStoreError::InfoMissing(handle) => write!(f, "Missing stored value info for '{}'", handle),
AggregateStoreError::InfoCorrupt(handle) => write!(f, "Corrupt stored value info for '{}'", handle),
src/upgrades/v0_9_0/ca_objects_migration.rs
+1 -20
View File
@@ -81,26 +81,7 @@ impl CaObjectsMigration {
info!("Will migrate data for {} CAs", cas.len());
for ca_handle in cas {
let ca = match store.get_latest(&ca_handle) {
Ok(ca) => ca,
Err(_) => {
// most likely an off by one error in the info.json of early releases
// try to move the info file out of the way and reload.
let kv = KeyValueStore::disk(&config.data_dir, CASERVER_DIR)?;
let info_key = KeyStoreKey::scoped(ca_handle.to_string(), "info.json".to_string());
let tmp_key = KeyStoreKey::scoped(ca_handle.to_string(), "tmp-info.json".to_string());
kv.move_key(&info_key, &tmp_key)?;
// Get latest (may still error)
let res = store.get_latest(&ca_handle);
// move the key back to leave things as they were
kv.move_key(&tmp_key, &info_key)?;
// Get the CA or error out if this failed.
res?
}
};
let ca = store.get_latest(&ca_handle)?;
let objects = ca.ca_objects(repo_manager.as_ref())?;
src/upgrades/v0_9_0/pubd_objects_migration.rs
+1 -20
View File
@@ -78,26 +78,7 @@ impl PubdObjectsMigration {
let repo_handle = Self::repository_handle();
let old_repo = match old_store.get_latest(&repo_handle) {
Ok(repo) => repo,
Err(_) => {
// most likely an off by one error in the info.json of early releases
// try to move the info file out of the way and reload.
let kv = KeyValueStore::disk(&config.data_dir, PUBSERVER_DIR)?;
let info_key = KeyStoreKey::scoped(repo_handle.to_string(), "info.json".to_string());
let tmp_key = KeyStoreKey::scoped(repo_handle.to_string(), "tmp-info.json".to_string());
kv.move_key(&info_key, &tmp_key)?;
// Get latest (may still error)
let res = old_store.get_latest(&repo_handle);
// move the key back to leave things as they were
kv.move_key(&tmp_key, &info_key)?;
// Get the repo or error out if this failed.
res?
}
};
let old_repo = old_store.get_latest(&repo_handle)?;
let publishers = old_repo
.publishers