Cleanup coordinator recovery.

Makefile

@@ -6,4 +6,4 @@ bench:
 
 maelstrom:
 	cargo build --release --features maelstrom --bin maelstrom
-	maelstrom test -w lin-kv --bin target/release/maelstrom --latency 0 --rate 10 --time-limit 90 --concurrency 20 --nemesis partition --nemesis-interval 10
+	maelstrom test -w lin-kv --bin target/release/maelstrom --latency 0 --rate 10 --time-limit 90 --concurrency 20 --nemesis partition --nemesis-interval 20

src/tapir/replica.rs

@@ -469,7 +469,7 @@ impl<K: Key, V: Value> IrReplicaUpcalls for Replica<K, V> {
         membership: &IrMembership<T>,
         transport: &T,
     ) {
-        let threshold: u64 = transport.time() - 10 * 1000 * 1000;
+        let threshold: u64 = transport.time_offset(-500);
         for (transaction_id, (commit, transaction)) in &self.inner.prepared {
             if commit.time > threshold {
                 // Allow the client to finish on its own.

src/transport/channel.rs

@@ -93,12 +93,16 @@ impl<M: Message> Transport for Channel<M> {
     }
 
     fn time(&self) -> u64 {
+        self.time_offset(rand::thread_rng().gen_range(0..100))
+    }
+
+    fn time_offset(&self, offset: i64) -> u64 {
         use rand::Rng;
-        SystemTime::now()
+        (SystemTime::now()
             .duration_since(SystemTime::UNIX_EPOCH)
             .unwrap()
-            .as_nanos() as u64
-            + rand::thread_rng().gen_range(0..10 * 1000 * 1000)
+            .as_nanos() as u64)
+            .saturating_add_signed(offset.saturating_mul(1000 * 1000 / 10))
     }
 
     fn sleep(duration: Duration) -> Self::Sleep {

src/transport/mod.rs

@@ -27,6 +27,11 @@ pub trait Transport: Clone + Send + Sync + 'static {
             .as_nanos() as u64
     }
 
+    fn time_offset(&self, offset: i64) -> u64 {
+        self.time()
+            .saturating_add_signed(offset.saturating_mul(1000 * 1000))
+    }
+
     /// Sleep for duration.
     fn sleep(duration: Duration) -> Self::Sleep;