From 1f46abf8f42a93db960ded8038daa130757706bb Mon Sep 17 00:00:00 2001 From: Tim Bruijnzeels Date: Tue, 22 Feb 2022 21:32:43 +0100 Subject: [PATCH] Simplify event replay logic - do not rely on info file for reply without limit. --- src/commons/eventsourcing/store.rs | 65 +++++++------------ src/upgrades/v0_9_0/ca_objects_migration.rs | 21 +----- src/upgrades/v0_9_0/pubd_objects_migration.rs | 21 +----- 3 files changed, 27 insertions(+), 80 deletions(-) diff --git a/src/commons/eventsourcing/store.rs b/src/commons/eventsourcing/store.rs index dd875fcc..91e6134b 100644 --- a/src/commons/eventsourcing/store.rs +++ b/src/commons/eventsourcing/store.rs @@ -894,51 +894,36 @@ where fn update_aggregate(&self, id: &Handle, aggregate: &mut A, limit: Option) -> 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::(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), diff --git a/src/upgrades/v0_9_0/ca_objects_migration.rs b/src/upgrades/v0_9_0/ca_objects_migration.rs index de128168..2006cba6 100644 --- a/src/upgrades/v0_9_0/ca_objects_migration.rs +++ b/src/upgrades/v0_9_0/ca_objects_migration.rs @@ -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())?; diff --git a/src/upgrades/v0_9_0/pubd_objects_migration.rs b/src/upgrades/v0_9_0/pubd_objects_migration.rs index 9fa82315..af6276cd 100644 --- a/src/upgrades/v0_9_0/pubd_objects_migration.rs +++ b/src/upgrades/v0_9_0/pubd_objects_migration.rs @@ -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