aboutsummaryrefslogtreecommitdiffstats
path: root/src/remake/switch.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/remake/switch.rs')
-rw-r--r--src/remake/switch.rs30
1 files changed, 20 insertions, 10 deletions
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
index 9b03789..7d6611d 100644
--- a/src/remake/switch.rs
+++ b/src/remake/switch.rs
@@ -193,6 +193,7 @@ async fn switch<H: Host>(
.translator()
.ok_or_else(|| format!("unknown remake kind {:?}", request.kind))?;
let late_translator = translator.clone();
+ let superseded = host.superseded();
let previous = request.previous;
let base = request.base_heads.clone();
let retention = record.proposal.policy.archive_retention.clone();
@@ -235,20 +236,20 @@ async fn switch<H: Host>(
// A fresh actor: this node's writes in the new epoch.
let mut doc = snapshot.doc.with_actor(actor);
- // The carried record and the replayed writes go into one change.
let heads = copy.get_heads();
- let paths = delta::changed_paths(&copy, &base, &heads);
- let mut transaction = doc.transaction();
// Keep this remake's record visible in the new epoch, for its history.
- if carry_record
- && let Err(error) = store::copy_path(&copy, &mut transaction, &record_path)
- {
- warn!(error = %error, "Could not carry the remake record into the new epoch");
+ if carry_record {
+ let mut transaction = doc.transaction();
+ if let Err(error) = store::copy_path(&copy, &mut transaction, &record_path) {
+ warn!(error = %error, "Could not carry the remake record into the new epoch");
+ }
+ transaction.commit();
}
- let stats = delta::replay_paths(&copy, &mut transaction, &paths, &translator);
- transaction.commit();
+ // Each write at the time it was made, so later replays judge by it.
+ let stats =
+ delta::replay_newest(&copy, &base, &heads, &mut doc, &translator, superseded);
Ok((doc, heads, stats))
},
)
@@ -283,6 +284,7 @@ async fn switch<H: Host>(
replayed = stats.replayed,
failed = stats.failed,
dropped = stats.dropped,
+ superseded = stats.superseded,
quarantine = ?quarantine,
"Switched to the new epoch"
);
@@ -299,6 +301,7 @@ pub(super) async fn swap_in<H: Host>(
copy_heads: Vec<ChangeHash>,
translator: Translator,
) -> Result<(Document, ReplayStats), String> {
+ let superseded = host.superseded();
// The closure runs on the blocking pool; its replay numbers come back through this.
let late_stats = std::sync::Arc::new(std::sync::Mutex::new(ReplayStats::default()));
let late_out = std::sync::Arc::clone(&late_stats);
@@ -310,7 +313,14 @@ pub(super) async fn swap_in<H: Host>(
}
let mut new_doc = new_doc;
let live_heads = live.get_heads();
- let late = delta::replay(live, &copy_heads, &live_heads, &mut new_doc, &translator);
+ let late = delta::replay_newest(
+ live,
+ &copy_heads,
+ &live_heads,
+ &mut new_doc,
+ &translator,
+ superseded,
+ );
if late.replayed + late.failed > 0 {
info!(
replayed = late.replayed,