Transport: protocol v2 codec, socket, connection and timers (DEV-1150 part B)

replicant-client/Cargo.toml

@@ -6,6 +6,9 @@ edition = "2021"
 [dependencies]
 replicant-core = { path = "../replicant-core" }
 tokio = { workspace = true }
+tokio-tungstenite = { version = "0.24", features = ["native-tls"] }
+futures-util = { version = "0.3", features = ["sink"] }
+tokio-util = { version = "0.7", features = ["time"] }
 phoenix_channels_client = { git = "https://github.com/liveview-native/phoenix-channels-client", features = ["native-tls"] }
 sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite", "json", "uuid", "chrono", "tls-rustls"] }
 serde = { workspace = true }
@@ -25,6 +28,7 @@ rand = "0.8"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 
 [dev-dependencies]
+tokio = { workspace = true, features = ["test-util"] }
 tempfile = "3"
 wiremock = "0.6"
 clap = { version = "4.4", features = ["derive", "env"] }

replicant-client/src/driver/mod.rs

@@ -0,0 +1,3 @@
+//! The owner task around the sans-IO core.
+
+pub mod timers;

replicant-client/src/driver/timers.rs

@@ -0,0 +1,106 @@
+//! The core's timers. Scheduling an id that is already pending replaces it.
+
+use std::collections::HashMap;
+use std::future::poll_fn;
+use std::task::Poll;
+use std::time::Duration;
+
+use tokio_util::time::{delay_queue, DelayQueue};
+
+use crate::engine::machine::TimerId;
+
+#[derive(Default)]
+pub struct Timers {
+    queue: DelayQueue<TimerId>,
+    keys: HashMap<TimerId, delay_queue::Key>,
+}
+
+impl Timers {
+    pub fn schedule(&mut self, timer: TimerId, after: Duration) {
+        match self.keys.get(&timer) {
+            Some(key) => self.queue.reset(key, after),
+            None => {
+                let key = self.queue.insert(timer.clone(), after);
+                self.keys.insert(timer, key);
+            }
+        }
+    }
+
+    pub fn cancel(&mut self, timer: &TimerId) {
+        if let Some(key) = self.keys.remove(timer) {
+            self.queue.remove(&key);
+        }
+    }
+
+    /// Resolves with the next due timer. Pending while none are scheduled; cancel-safe.
+    pub async fn next(&mut self) -> TimerId {
+        poll_fn(|cx| match self.queue.poll_expired(cx) {
+            Poll::Ready(Some(expired)) => {
+                let timer = expired.into_inner();
+                self.keys.remove(&timer);
+                Poll::Ready(timer)
+            }
+            // An empty queue keeps the waker and wakes it on the next insert.
+            Poll::Ready(None) | Poll::Pending => Poll::Pending,
+        })
+        .await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::time::{timeout, Instant};
+
+    const LONG: Duration = Duration::from_secs(3600);
+
+    #[tokio::test(start_paused = true)]
+    async fn fires_in_due_order() {
+        let mut timers = Timers::default();
+        timers.schedule(TimerId::Pump, Duration::from_millis(200));
+        timers.schedule(TimerId::Heartbeat, Duration::from_millis(100));
+        assert_eq!(timers.next().await, TimerId::Heartbeat);
+        assert_eq!(timers.next().await, TimerId::Pump);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn rescheduling_replaces_the_pending_timer() {
+        let mut timers = Timers::default();
+        let start = Instant::now();
+        timers.schedule(TimerId::Pump, Duration::from_millis(100));
+        timers.schedule(TimerId::Pump, Duration::from_millis(500));
+        assert_eq!(timers.next().await, TimerId::Pump);
+        assert!(start.elapsed() >= Duration::from_millis(500));
+        assert!(
+            timeout(LONG, timers.next()).await.is_err(),
+            "a replaced timer fires once"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn cancelled_timer_never_fires() {
+        let mut timers = Timers::default();
+        timers.schedule(TimerId::Reconnect, Duration::from_millis(100));
+        timers.cancel(&TimerId::Reconnect);
+        timers.cancel(&TimerId::HaltRetry);
+        assert!(timeout(LONG, timers.next()).await.is_err());
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn ids_with_different_payloads_are_independent() {
+        let mut timers = Timers::default();
+        timers.schedule(TimerId::Request(1), Duration::from_millis(100));
+        timers.schedule(TimerId::Request(2), Duration::from_millis(200));
+        timers.cancel(&TimerId::Request(1));
+        assert_eq!(timers.next().await, TimerId::Request(2));
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn a_fired_timer_can_be_scheduled_again() {
+        let mut timers = Timers::default();
+        timers.schedule(TimerId::Heartbeat, Duration::from_secs(30));
+        assert_eq!(timers.next().await, TimerId::Heartbeat);
+        timers.schedule(TimerId::Heartbeat, Duration::from_secs(30));
+        assert_eq!(timers.next().await, TimerId::Heartbeat);
+    }
+}

replicant-client/src/engine/doc.rs

@@ -198,8 +198,14 @@ fn membership_seq(snap: &DocSnapshot, scope: &str) -> Seq {
         .map_or(0, |m| m.seq)
 }
 
-/// Non-echo upsert: content guard already passed.
-fn apply_upsert(snap: &DocSnapshot, doc: &DocEnvelope, seq: Seq) -> Vec<DocOp> {
+/// Non-echo upsert: content guard already passed. When `local_base_known` is false the shadow
+/// is not the base of the pending rows, so edits that would need a rebase settle as a conflict.
+fn apply_upsert(
+    snap: &DocSnapshot,
+    doc: &DocEnvelope,
+    seq: Seq,
+    local_base_known: bool,
+) -> Vec<DocOp> {
     let new_shadow = Shadow {
         content: doc.content.clone(),
         hash: doc.hash.clone(),
@@ -234,19 +240,21 @@ fn apply_upsert(snap: &DocSnapshot, doc: &DocEnvelope, seq: Seq) -> Vec<DocOp> {
         ops.push(DocOp::SetShadow(new_shadow));
         return ops;
     }
-    let content = match &snap.shadow {
+    let rebased = match &snap.shadow {
+        _ if !local_base_known => None,
         // Migration only: no known base, keep local content; pending = diff(new shadow, content).
-        None => snap.content.clone(),
+        None => Some(snap.content.clone()),
         Some(old) => match rebase(&old.content, &doc.content, &snap.content) {
-            Rebased::Clean(v) => v,
-            Rebased::Conflict => {
-                let with_new_shadow = snap.project(&[DocOp::SetShadow(new_shadow.clone())]);
-                ops.push(DocOp::SetShadow(new_shadow));
-                ops.extend(conflict_ops(&with_new_shadow));
-                return ops;
-            }
+            Rebased::Clean(v) => Some(v),
+            Rebased::Conflict => None,
         },
     };
+    let Some(content) = rebased else {
+        let with_new_shadow = snap.project(&[DocOp::SetShadow(new_shadow.clone())]);
+        ops.push(DocOp::SetShadow(new_shadow));
+        ops.extend(conflict_ops(&with_new_shadow));
+        return ops;
+    };
     ops.push(DocOp::SetShadow(new_shadow));
     ops.push(DocOp::SetContent(content));
     ops
@@ -289,12 +297,27 @@ pub fn apply_change(snap: &DocSnapshot, change: &Change, me: Uuid) -> Vec<DocOp>
         .filter(|u| snap.rows.iter().any(|r| r.mutation_id == *u));
 
     match (change.kind, echo_upload) {
-        // Echo of an upload from this data dir: never rebase, it would re-apply the delta.
+        // Echo of an upload from this data dir: never rebase, local content already holds the
+        // uploaded delta and re-applying it would duplicate or conflict.
         (ChangeKind::Upsert, Some(upload_id)) => {
             if !is_member(&ops) {
                 return ops;
             }
             if let Some(doc) = &change.doc {
+                let covered: Vec<Uuid> = snap
+                    .rows
+                    .iter()
+                    .filter(|r| r.mutation_id <= upload_id)
+                    .map(|r| r.mutation_id)
+                    .collect();
+                // A newer envelope than the upload (a page without the stored reply): the upload
+                // content is unknown, so rows it did not cover cannot be rebased onto it.
+                if doc.seq != change.seq {
+                    ops.push(DocOp::DeleteRows(covered));
+                    let acked = snap.project(&ops);
+                    ops.extend(apply_upsert(&acked, doc, doc.seq, false));
+                    return with_settle_invariant(snap, ops, me);
+                }
                 ops.push(DocOp::SetMeta {
                     owner_id: doc.owner_id,
                     read_only: doc.read_only,
@@ -304,12 +327,6 @@ pub fn apply_change(snap: &DocSnapshot, change: &Change, me: Uuid) -> Vec<DocOp>
                     hash: doc.hash.clone(),
                     seq: change.seq,
                 }));
-                let covered = snap
-                    .rows
-                    .iter()
-                    .filter(|r| r.mutation_id <= upload_id)
-                    .map(|r| r.mutation_id)
-                    .collect();
                 ops.push(DocOp::DeleteRows(covered));
                 return with_settle_invariant(snap, ops, me);
             }
@@ -318,7 +335,7 @@ pub fn apply_change(snap: &DocSnapshot, change: &Change, me: Uuid) -> Vec<DocOp>
             debug_assert!(change.doc.is_some(), "upsert without doc");
             if is_member(&ops) {
                 if let Some(doc) = &change.doc {
-                    ops.extend(apply_upsert(snap, doc, change.seq));
+                    ops.extend(apply_upsert(snap, doc, change.seq, true));
                 }
             }
         }
@@ -355,7 +372,7 @@ pub fn apply_server_copy(snap: &DocSnapshot, doc: &DocEnvelope, _me: Uuid) -> Ve
     if doc.seq <= snap.server_seq() {
         return Vec::new();
     }
-    apply_upsert(snap, doc, doc.seq)
+    apply_upsert(snap, doc, doc.seq, true)
 }
 
 /// `get_document` said the document does not exist on the server: not_found means the server
@@ -870,6 +887,125 @@ mod echo_and_delete_tests {
         );
     }
 
+    /// The page's change for upload m1 (seq 2) carrying the current envelope (seq 4).
+    fn page_echo(content: Value) -> Change {
+        let mut c = with_upload(upsert("own", json!({}), 2), m(1));
+        c.doc = Some(env(content, 4));
+        c
+    }
+
+    /// Settled on the envelope at its own seq with no rows left.
+    fn assert_adopted(after: &DocSnapshot, envelope: Value) {
+        assert!(after.rows.is_empty());
+        assert_eq!(after.content, envelope);
+        let sh = after.shadow.as_ref().unwrap();
+        assert_eq!(sh.content, envelope);
+        assert_eq!(sh.seq, 4);
+    }
+
+    fn assert_no_conflict_or_marker(ops: &[DocOp]) {
+        assert!(!ops.iter().any(|o| matches!(
+            o,
+            DocOp::Recover { .. } | DocOp::Emit(_) | DocOp::InsertMarker(_)
+        )));
+    }
+
+    fn assert_conflict_recovers(ops: &[DocOp], local: Value) {
+        assert!(ops.contains(&DocOp::Recover {
+            content: local,
+            reason: RecoverReason::Conflict
+        }));
+        assert!(ops.contains(&DocOp::Emit(DocEvent::ConflictDetected)));
+        assert!(!ops.iter().any(|o| matches!(o, DocOp::InsertMarker(_))));
+    }
+
+    #[test]
+    fn page_echo_with_newer_envelope_and_pending_rows_settles_as_conflict() {
+        // m1 uploaded "mine"; m2 ("later") is still pending; the envelope also has "theirs".
+        let mut s = synced(json!({"a": 1}), 1);
+        s.content = json!({"a": 1, "mine": true, "later": true});
+        s.rows = vec![row(1, RowKind::Update), row(2, RowKind::Update)];
+        let envelope = json!({"a": 1, "mine": true, "theirs": true});
+
+        let ops = apply_change(&s, &page_echo(envelope.clone()), ME);
+        let after = s.project(&ops);
+
+        assert_adopted(&after, envelope);
+        assert_conflict_recovers(&ops, s.content.clone());
+    }
+
+    #[test]
+    fn page_echo_array_append_covering_all_rows_adopts_envelope() {
+        // m1 appended x; another device then appended z.
+        let mut s = synced(json!({"items": []}), 1);
+        s.content = json!({"items": ["x"]});
+        s.rows = vec![row(1, RowKind::Update)];
+        let envelope = json!({"items": ["x", "z"]});
+
+        let ops = apply_change(&s, &page_echo(envelope.clone()), ME);
+
+        assert_adopted(&s.project(&ops), envelope);
+        assert_no_conflict_or_marker(&ops);
+    }
+
+    #[test]
+    fn page_echo_array_append_with_pending_row_settles_as_conflict() {
+        // m1 appended x (covered), m2 appended y (pending); another device appended z.
+        let mut s = synced(json!({"items": []}), 1);
+        s.content = json!({"items": ["x", "y"]});
+        s.rows = vec![row(1, RowKind::Update), row(2, RowKind::Update)];
+        let envelope = json!({"items": ["x", "z"]});
+
+        let ops = apply_change(&s, &page_echo(envelope.clone()), ME);
+
+        assert_adopted(&s.project(&ops), envelope);
+        assert_conflict_recovers(&ops, json!({"items": ["x", "y"]}));
+    }
+
+    #[test]
+    fn page_echo_key_removal_covering_all_rows_adopts_envelope() {
+        // m1 removed k; another device then added b.
+        let mut s = synced(json!({"a": 1, "k": 1}), 1);
+        s.content = json!({"a": 1});
+        s.rows = vec![row(1, RowKind::Update)];
+        let envelope = json!({"a": 1, "b": 2});
+
+        let ops = apply_change(&s, &page_echo(envelope.clone()), ME);
+
+        assert_adopted(&s.project(&ops), envelope);
+        assert_no_conflict_or_marker(&ops);
+    }
+
+    #[test]
+    fn page_echo_key_removal_with_pending_row_settles_as_conflict() {
+        // m1 removed k (covered), m2 added c (pending); another device added b.
+        let mut s = synced(json!({"a": 1, "k": 1}), 1);
+        s.content = json!({"a": 1, "c": 3});
+        s.rows = vec![row(1, RowKind::Update), row(2, RowKind::Update)];
+        let envelope = json!({"a": 1, "b": 2});
+
+        let ops = apply_change(&s, &page_echo(envelope.clone()), ME);
+
+        assert_adopted(&s.project(&ops), envelope);
+        assert_conflict_recovers(&ops, json!({"a": 1, "c": 3}));
+    }
+
+    #[test]
+    fn page_echo_covering_all_rows_takes_server_content() {
+        let mut s = synced(json!({"n": 1}), 1);
+        s.content = json!({"n": 2});
+        s.rows = vec![row(1, RowKind::Update)];
+        let mut page_echo = with_upload(upsert("own", json!({}), 2), m(1));
+        page_echo.doc = Some(env(json!({"n": 2, "other": 1}), 5));
+
+        let ops = apply_change(&s, &page_echo, ME);
+        let after = s.project(&ops);
+
+        assert!(after.rows.is_empty());
+        assert_eq!(after.content, json!({"n": 2, "other": 1}));
+        assert!(!ops.iter().any(|op| matches!(op, DocOp::InsertMarker(_))));
+    }
+
     #[test]
     fn echo_covering_all_rows_settles_without_marker() {
         let mut s = synced(json!({"n": 1}), 1);