1 files changed, 37 insertions, 11 deletions

diff --git a/src/remake/local.rs b/src/remake/local.rs
index 7810690..c80a5a5 100644
--- a/src/remake/local.rs
+++ b/src/remake/local.rs
@@ -9,7 +9,6 @@ use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, Mutex, MutexGuard};
 use std::time::Duration;
 
-use automerge::{ActorId, AutoCommit};
 use serde::Serialize;
 use tokio::time::Instant;
 
@@ -17,6 +16,7 @@ use super::records::Retention;
 use super::snapshot::Snapshot;
 use crate::clock::UnixMillis;
 use crate::document::replica::Epoch;
+use crate::document::{ActorId, ChangeHash, Document};
 use crate::locks::MutexExt;
 
 pub(crate) struct RemakeState {
@@ -25,6 +25,8 @@ pub(crate) struct RemakeState {
     started_at: UnixMillis,
     // A std mutex: every access is short and never spans an await.
     inner: Mutex<Inner>,
+    /// A download of content ended (`end_content_fetch`).
+    content_fetch_ended: tokio::sync::Notify,
 }
 
 impl RemakeState {
@@ -33,6 +35,7 @@ impl RemakeState {
         Self {
             started_at,
             inner: Mutex::default(),
+            content_fetch_ended: tokio::sync::Notify::new(),
         }
     }
 }
@@ -60,6 +63,9 @@ struct Inner {
     coordinating: HashSet<String>,
     /// The remake this node is about to vote for, and when (`vote_due`).
     vote_due: Option<(String, UnixMillis)>,
+    /// Remakes this node could have voted for but saw expire first (`vote::cast`), each
+    /// logged once.
+    missed: HashSet<String>,
     /// The actors this node writes or wrote with in the current document: this run's,
     /// and those of earlier runs it knows of (`--document-file`). Its own writes, for a
     /// rejoin. A switch or rejoin starts the list over with the new document's actor, so
@@ -84,8 +90,13 @@ pub(crate) struct CatchUp {
 pub(crate) struct Rejoin {
     pub(crate) previous: Epoch,
     pub(crate) target: Epoch,
-    /// The document of `previous`, until it is replayed, quarantined or restored.
-    pub(crate) old: AutoCommit,
+    /// The document of `previous`, until it is replayed, quarantined or restored; of a
+    /// node that was ready, kept by the node instead, where it takes its writes
+    /// meanwhile (`Host::begin_rejoin`).
+    pub(crate) old: Option<Document>,
+    /// That document's heads as the rejoin began: the changes after them it took
+    /// during the rejoin.
+    pub(crate) heads: Vec<ChangeHash>,
     pub(crate) started: Instant,
     /// The node served requests before; it does again once the rejoin is over.
     pub(crate) was_ready: bool,
@@ -102,7 +113,7 @@ pub(crate) struct Quarantined {
     /// Store paths the changes touched (minimal, prefix-free).
     pub(crate) paths: Vec<Vec<String>>,
     /// The old document; `paths` are read from its merged view.
-    pub(crate) document: AutoCommit,
+    pub(crate) document: Document,
 }
 
 #[derive(Clone, Debug, Serialize)]
@@ -138,7 +149,7 @@ struct Archived {
     epoch: Epoch,
     // Kept for auditing and recovery by hand; not read by the protocol.
     #[allow(dead_code)]
-    doc: AutoCommit,
+    doc: Document,
     archived_at: UnixMillis,
     retention: Retention,
 }
@@ -215,6 +226,12 @@ impl RemakeState {
         }
     }
 
+    /// Notes that this node saw `remake` expire without having voted for it: true the
+    /// first time.
+    pub(crate) fn missed(&self, remake: &str) -> bool {
+        self.lock().missed.insert(remake.to_owned())
+    }
+
     fn lock(&self) -> MutexGuard<'_, Inner> {
         self.inner.lock_or_recover()
     }
@@ -249,7 +266,7 @@ impl RemakeState {
     pub(crate) fn archive(
         &self,
         epoch: Epoch,
-        doc: AutoCommit,
+        doc: Document,
         retention: Retention,
         current_generation: u64,
         now: UnixMillis,
@@ -325,10 +342,19 @@ impl RemakeState {
 
     pub(crate) fn end_content_fetch(&self, hash: &[u8; 32]) {
         self.lock().content_fetches.remove(hash);
+        self.content_fetch_ended.notify_waiters();
     }
 
-    pub(crate) fn content_fetch_running(&self, hash: &[u8; 32]) -> bool {
-        self.lock().content_fetches.contains(hash)
+    /// Waits until no download of `hash` runs.
+    pub(crate) async fn content_fetch_ended(&self, hash: &[u8; 32]) {
+        loop {
+            // Before looking, so that an end in between isn't missed.
+            let ended = self.content_fetch_ended.notified();
+            if !self.lock().content_fetches.contains(hash) {
+                return;
+            }
+            ended.await;
+        }
     }
 
     /// Drops content that no remake in `referenced` needs, once its grace has passed.
@@ -431,7 +457,7 @@ mod tests {
         // Kept by epoch count only.
         remake.archive(
             epoch(0),
-            AutoCommit::new(),
+            Document::new(),
             Retention {
                 epochs: Some(1),
                 age_ms: None,
@@ -446,7 +472,7 @@ mod tests {
         // Kept by time while the epoch count has passed.
         remake.archive(
             epoch(1),
-            AutoCommit::new(),
+            Document::new(),
             Retention {
                 epochs: Some(0),
                 age_ms: Some(3_600_000),
@@ -459,7 +485,7 @@ mod tests {
         // No limits at all: not kept.
         remake.archive(
             epoch(2),
-            AutoCommit::new(),
+            Document::new(),
             Retention {
                 epochs: None,
                 age_ms: None,