diff --git a/Cargo.lock b/Cargo.lock index ab1f783..f7c680d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2310,6 +2310,7 @@ dependencies = [ "crossterm", "dialoguer", "env_logger", + "futures-util", "hex", "hmac", "json-patch", @@ -2327,6 +2328,8 @@ dependencies = [ "tempfile", "thiserror 2.0.12", "tokio", + "tokio-tungstenite", + "tokio-util", "tracing", "tracing-subscriber", "url", @@ -3348,6 +3351,7 @@ dependencies = [ "futures-core", "futures-sink", "pin-project-lite", + "slab", "tokio", ] diff --git a/replicant-client/Cargo.toml b/replicant-client/Cargo.toml index 576ce53..0ffff14 100644 --- a/replicant-client/Cargo.toml +++ 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"] } diff --git a/replicant-client/src/driver/mod.rs b/replicant-client/src/driver/mod.rs new file mode 100644 index 0000000..a9f0faf --- /dev/null +++ b/replicant-client/src/driver/mod.rs @@ -0,0 +1,3 @@ +//! The owner task around the sans-IO core. + +pub mod timers; diff --git a/replicant-client/src/driver/timers.rs b/replicant-client/src/driver/timers.rs new file mode 100644 index 0000000..1a8bcbe --- /dev/null +++ b/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, + keys: HashMap, +} + +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); + } +} diff --git a/replicant-client/src/engine/doc.rs b/replicant-client/src/engine/doc.rs index 9c9207b..dcf5b5b 100644 --- a/replicant-client/src/engine/doc.rs +++ b/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 { +/// 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 { 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 { 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 .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 = 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 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 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); diff --git a/replicant-client/src/engine/fuzz_tests.rs b/replicant-client/src/engine/fuzz_tests.rs index c1ba091..c9e0c53 100644 --- a/replicant-client/src/engine/fuzz_tests.rs +++ b/replicant-client/src/engine/fuzz_tests.rs @@ -59,11 +59,21 @@ fn next_input(rng: &mut Jitter, outstanding: &mut Vec<(u64, Request)>, socket_ge } else { socket_gen.saturating_sub(1) }; + let mut refused = super::types::ServerError::new(if rng.next_unit() < 0.2 { + "update_required" + } else { + "rate_limited" + }); + refused.is_fatal = refused.code == "update_required"; pick( rng, &[ Input::SocketOpened { gen: socket_gen }, Input::SocketClosed { gen: closed_gen }, + Input::ConnectRefused { + gen: closed_gen, + error: refused, + }, Input::Reconnect, Input::OutboxChanged, Input::Timer(TimerId::Reconnect), @@ -106,11 +116,13 @@ impl Harness { fn feed(&mut self, input: Input) { let mut queue = std::collections::VecDeque::from([input]); while let Some(input) = queue.pop_front() { - if input - == (Input::SocketClosed { - gen: self.socket_gen, - }) - { + let closes_current = match &input { + Input::SocketClosed { gen } | Input::ConnectRefused { gen, .. } => { + *gen == self.socket_gen + } + _ => false, + }; + if closes_current { self.socket_open = false; } for e in self.core.step(input) { diff --git a/replicant-client/src/engine/machine.rs b/replicant-client/src/engine/machine.rs index ac365ce..8332fc5 100644 --- a/replicant-client/src/engine/machine.rs +++ b/replicant-client/src/engine/machine.rs @@ -157,6 +157,11 @@ pub enum Input { SocketClosed { gen: u64, }, + /// The server refused the websocket upgrade (HTTP 426 → `update_required`). + ConnectRefused { + gen: u64, + error: ServerError, + }, Reply { req: u64, result: Result, @@ -473,6 +478,18 @@ impl Core { Conn::Connected(_) => self.lose_connection(fx), _ => {} }, + Input::ConnectRefused { gen, .. } if gen != self.socket_gen => {} + Input::ConnectRefused { error, .. } => { + if let Conn::Connecting { .. } = self.conn { + match Self::halt_reason_for(&error) { + Some(reason) => { + fx.push(Effect::Cancel(TimerId::ConnectTimeout)); + self.halt(reason, fx); + } + None => self.fail_connect(error.retry_after_ms, fx), + } + } + } Input::Reply { req, result } => self.on_reply(req, result, fx), Input::Timer(timer) => self.on_timer(timer, fx), Input::Cursors(list) => self.on_cursors(list, fx), @@ -1406,6 +1423,76 @@ mod connection_tests { assert_eq!(sends(&fx).pop().map(|(_, r)| r), Some(Request::Join)); } + fn update_required() -> ServerError { + let mut error = ServerError::new("update_required"); + error.is_fatal = true; + error + } + + #[test] + fn connect_refused_update_required_halts_without_retry() { + let mut c = core(); + c.step(Input::Start { + has_credentials: true, + }); + let fx = c.step(Input::ConnectRefused { + gen: c.socket_gen, + error: update_required(), + }); + assert_eq!( + c.state().connection, + ConnectionView::Halted(HaltReason::UpdateRequired) + ); + assert!(fx.contains(&Effect::Cancel(TimerId::ConnectTimeout))); + assert!(!fx.iter().any(|e| matches!( + e, + Effect::Schedule { + timer: TimerId::HaltRetry | TimerId::Reconnect, + .. + } + ))); + assert!(emitted(&fx).iter().any(|l| matches!( + l, + Lifecycle::SyncError { fatal: true, code, .. } if code == "update_required" + ))); + } + + #[test] + fn connect_refused_transient_backs_off_with_retry_after() { + let mut c = core(); + c.step(Input::Start { + has_credentials: true, + }); + let mut error = ServerError::new("rate_limited"); + error.retry_after_ms = Some(4_000); + let fx = c.step(Input::ConnectRefused { + gen: c.socket_gen, + error, + }); + assert_eq!(c.state().connection, ConnectionView::Disconnected); + assert!(fx.contains(&Effect::Schedule { + timer: TimerId::Reconnect, + after: Duration::from_millis(4_000) + })); + } + + #[test] + fn stale_connect_refused_is_ignored() { + let mut c = core(); + c.step(Input::Start { + has_credentials: true, + }); + c.step(Input::Timer(TimerId::ConnectTimeout)); + c.step(Input::Timer(TimerId::Reconnect)); + assert!(c + .step(Input::ConnectRefused { + gen: 1, + error: update_required(), + }) + .is_empty()); + assert_eq!(c.state().connection, ConnectionView::Connecting); + } + #[test] fn start_without_credentials_halts_not_enrolled_with_retry() { let mut c = core(); diff --git a/replicant-client/src/engine/types.rs b/replicant-client/src/engine/types.rs index 1a098bd..e8ef293 100644 --- a/replicant-client/src/engine/types.rs +++ b/replicant-client/src/engine/types.rs @@ -89,7 +89,8 @@ pub struct Change { pub upload_id: Option, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] pub enum UploadKind { Create, Update, @@ -107,9 +108,10 @@ pub struct Upload { pub payload: Value, } -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Deserialize)] pub struct ServerError { pub code: String, + #[serde(default)] pub is_fatal: bool, pub retry_after_ms: Option, /// Set on `hash_mismatch`. diff --git a/replicant-client/src/lib.rs b/replicant-client/src/lib.rs index cbe72ca..9fac0f4 100644 --- a/replicant-client/src/lib.rs +++ b/replicant-client/src/lib.rs @@ -1,5 +1,6 @@ pub mod client; pub mod database; +pub mod driver; pub mod engine; pub mod enrollment; pub mod error_code; @@ -7,6 +8,7 @@ pub mod events; pub mod offline_queue; pub mod queries; pub mod secret_store; +pub mod transport; pub mod websocket; // C FFI module diff --git a/replicant-client/src/transport/codec.rs b/replicant-client/src/transport/codec.rs new file mode 100644 index 0000000..14cc3fa --- /dev/null +++ b/replicant-client/src/transport/codec.rs @@ -0,0 +1,338 @@ +//! Phoenix v2 JSON framing for one socket: `[join_ref, ref, topic, event, payload]`. + +use std::collections::HashMap; + +use serde::Serialize; +use serde_json::{json, Value}; +use uuid::Uuid; + +use super::wire::{ + ChangesPage, GetChangesSince, GetDocument, GetSnapshot, JoinAuth, JoinReply, ReplyEnvelope, + SnapshotPage, UploadRequest, CHANNEL_TOPIC, +}; +use crate::engine::machine::{Request, Response}; +use crate::engine::types::{Change, DocEnvelope, ServerError}; + +const PHOENIX_TOPIC: &str = "phoenix"; + +#[derive(Debug, Clone, PartialEq)] +pub enum Incoming { + Reply { + req: u64, + result: Result, + }, + /// The join succeeded; the driver checks `user_id` before telling the core. + Joined { + req: u64, + user_id: Uuid, + }, + Push(Change), + /// The server closed or crashed the channel; the socket is no longer usable. + ChannelClosed, +} + +#[derive(Debug, Clone, Copy, PartialEq)] +enum ReplyKind { + Join, + Changes, + Snapshot, + Upload, + Document, + Heartbeat, +} + +#[derive(Debug, Default)] +pub struct Codec { + join_ref: Option, + pending: HashMap, +} + +impl Codec { + pub fn new() -> Self { + Self::default() + } + + pub fn encode( + &mut self, + req: u64, + request: &Request, + auth: &JoinAuth, + now_unix: i64, + ) -> String { + let reference = req.to_string(); + let (kind, topic, event, payload) = match request { + Request::Join => { + self.join_ref = Some(reference.clone()); + ( + ReplyKind::Join, + CHANNEL_TOPIC, + "phx_join", + to_value(auth.sign(now_unix)), + ) + } + Request::Heartbeat => (ReplyKind::Heartbeat, PHOENIX_TOPIC, "heartbeat", json!({})), + Request::GetChangesSince { + scope, + cursor, + limit, + } => ( + ReplyKind::Changes, + CHANNEL_TOPIC, + "get_changes_since", + to_value(GetChangesSince { + scope, + cursor: *cursor, + limit: *limit, + }), + ), + Request::GetSnapshot { scope, page_token } => ( + ReplyKind::Snapshot, + CHANNEL_TOPIC, + "get_snapshot", + to_value(GetSnapshot { + scope, + page_token: page_token.as_deref(), + }), + ), + Request::Upload(upload) => ( + ReplyKind::Upload, + CHANNEL_TOPIC, + "upload", + to_value(UploadRequest { + upload_id: upload.upload_id, + doc_id: upload.doc_id, + kind: upload.kind, + base_hash: upload.base_hash.as_deref(), + payload: &upload.payload, + }), + ), + Request::GetDocument { doc_id } => ( + ReplyKind::Document, + CHANNEL_TOPIC, + "get_document", + to_value(GetDocument { doc_id: *doc_id }), + ), + }; + self.pending.insert(req, kind); + let join_ref = if topic == CHANNEL_TOPIC { + self.join_ref.clone() + } else { + None + }; + json!([join_ref, reference, topic, event, payload]).to_string() + } + + pub fn decode(&mut self, text: &str) -> Option { + let (_join_ref, reference, topic, event, payload): ( + Option, + Option, + String, + String, + Value, + ) = serde_json::from_str(text).ok()?; + match event.as_str() { + "phx_reply" => { + let req: u64 = reference?.parse().ok()?; + let kind = self.pending.remove(&req)?; + Some(decode_reply(req, kind, payload)) + } + "change" if topic == CHANNEL_TOPIC => { + serde_json::from_value(payload).ok().map(Incoming::Push) + } + "phx_error" | "phx_close" if topic == CHANNEL_TOPIC => Some(Incoming::ChannelClosed), + _ => None, + } + } +} + +fn to_value(payload: impl Serialize) -> Value { + serde_json::to_value(payload).expect("request payloads serialize") +} + +fn decode_reply(req: u64, kind: ReplyKind, payload: Value) -> Incoming { + let Ok(envelope) = serde_json::from_value::(payload) else { + return Incoming::Reply { + req, + result: Err(protocol_error()), + }; + }; + if envelope.status != "ok" { + let error = serde_json::from_value(envelope.response).unwrap_or_else(|_| protocol_error()); + return Incoming::Reply { + req, + result: Err(error), + }; + } + let response = envelope.response; + let decoded = match kind { + ReplyKind::Join => { + return match serde_json::from_value::(response) { + Ok(join) => Incoming::Joined { + req, + user_id: join.user_id, + }, + Err(_) => Incoming::Reply { + req, + result: Err(protocol_error()), + }, + } + } + ReplyKind::Heartbeat => Ok(Response::HeartbeatOk), + ReplyKind::Changes => { + serde_json::from_value::(response).map(|page| Response::Changes { + changes: page.changes, + next_cursor: page.next_cursor, + has_more: page.has_more, + }) + } + ReplyKind::Snapshot => { + serde_json::from_value::(response).map(|page| Response::SnapshotPage { + docs: page.docs, + snapshot_seq: page.snapshot_seq, + next_page_token: page.next_page_token, + }) + } + ReplyKind::Upload => { + serde_json::from_value::(response).map(Response::Uploaded) + } + ReplyKind::Document => { + serde_json::from_value::(response).map(Response::Document) + } + }; + Incoming::Reply { + req, + result: decoded.map_err(|_| protocol_error()), + } +} + +/// Transient by construction: a frame the client cannot read is retried, never halts. +fn protocol_error() -> ServerError { + ServerError::new("protocol_error") +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn auth() -> JoinAuth { + JoinAuth { + email: "a@b.c".into(), + api_key: "rpa_k".into(), + api_secret: "rps_s".into(), + } + } + + fn changes_request() -> Request { + Request::GetChangesSince { + scope: "own".into(), + cursor: 0, + limit: 500, + } + } + + fn parse(text: &str) -> Value { + serde_json::from_str(text).unwrap() + } + + #[test] + fn channel_frames_carry_the_join_ref_and_heartbeats_use_the_phoenix_topic() { + let mut codec = Codec::new(); + let join = parse(&codec.encode(4, &Request::Join, &auth(), 1)); + assert_eq!(join[0], json!("4")); + assert_eq!(join[1], json!("4")); + assert_eq!(join[2], json!("sync:v2")); + assert_eq!(join[3], json!("phx_join")); + + let changes = parse(&codec.encode(5, &changes_request(), &auth(), 1)); + assert_eq!(changes[0], json!("4")); + assert_eq!(changes[1], json!("5")); + + let heartbeat = parse(&codec.encode(6, &Request::Heartbeat, &auth(), 1)); + assert_eq!(heartbeat, json!([null, "6", "phoenix", "heartbeat", {}])); + } + + #[test] + fn heartbeat_reply_decodes_as_heartbeat_ok() { + let mut codec = Codec::new(); + codec.encode(6, &Request::Heartbeat, &auth(), 1); + let reply = json!([null, "6", "phoenix", "phx_reply", {"status": "ok", "response": {}}]); + assert_eq!( + codec.decode(&reply.to_string()), + Some(Incoming::Reply { + req: 6, + result: Ok(Response::HeartbeatOk) + }) + ); + } + + #[test] + fn a_reply_is_matched_once_and_unknown_refs_are_ignored() { + let mut codec = Codec::new(); + codec.encode(2, &changes_request(), &auth(), 1); + let reply = json!(["1", "2", "sync:v2", "phx_reply", + {"status": "ok", "response": {"changes": [], "next_cursor": 9, "has_more": false}}]) + .to_string(); + assert!(matches!( + codec.decode(&reply), + Some(Incoming::Reply { + req: 2, + result: Ok(Response::Changes { next_cursor: 9, .. }) + }) + )); + assert_eq!(codec.decode(&reply), None); + let unknown = json!(["1", "77", "sync:v2", "phx_reply", {"status": "ok", "response": {}}]); + assert_eq!(codec.decode(&unknown.to_string()), None); + } + + #[test] + fn unreadable_replies_are_transient_protocol_errors() { + let mut codec = Codec::new(); + codec.encode(2, &changes_request(), &auth(), 1); + let bad_ok = json!(["1", "2", "sync:v2", "phx_reply", + {"status": "ok", "response": {"bogus": 1}}]); + let Some(Incoming::Reply { + result: Err(error), .. + }) = codec.decode(&bad_ok.to_string()) + else { + panic!("expected an error reply"); + }; + assert_eq!(error.code, "protocol_error"); + assert!(!error.is_fatal); + + codec.encode(3, &changes_request(), &auth(), 1); + let reasoned = json!(["1", "3", "sync:v2", "phx_reply", + {"status": "error", "response": {"reason": "unmatched topic"}}]); + let Some(Incoming::Reply { + result: Err(error), .. + }) = codec.decode(&reasoned.to_string()) + else { + panic!("expected an error reply"); + }; + assert_eq!(error.code, "protocol_error"); + } + + #[test] + fn channel_error_or_close_ends_the_channel() { + let mut codec = Codec::new(); + for event in ["phx_error", "phx_close"] { + let frame = json!(["1", "1", "sync:v2", event, {}]); + assert_eq!( + codec.decode(&frame.to_string()), + Some(Incoming::ChannelClosed), + "{event}" + ); + } + } + + #[test] + fn unknown_events_and_malformed_text_are_ignored() { + let mut codec = Codec::new(); + let presence = json!(["1", null, "sync:v2", "presence_diff", {}]); + assert_eq!(codec.decode(&presence.to_string()), None); + assert_eq!(codec.decode("not json"), None); + assert_eq!(codec.decode(&json!({"topic": "sync:v2"}).to_string()), None); + let bad_change = json!(["1", null, "sync:v2", "change", {"seq": "x"}]); + assert_eq!(codec.decode(&bad_change.to_string()), None); + } +} diff --git a/replicant-client/src/transport/connection.rs b/replicant-client/src/transport/connection.rs new file mode 100644 index 0000000..1ea0b54 --- /dev/null +++ b/replicant-client/src/transport/connection.rs @@ -0,0 +1,349 @@ +//! The driver's side of the socket: runs socket effects and turns socket events into core inputs. + +use tokio::sync::mpsc; +use uuid::Uuid; + +use super::codec::{Codec, Incoming}; +use super::socket::{SocketEvent, SocketHandle}; +use super::wire::JoinAuth; +use crate::engine::machine::{Input, Request}; + +const EVENT_BUFFER: usize = 64; + +#[derive(Debug, PartialEq)] +#[allow(clippy::large_enum_variant)] // Input is the natural payload; boxing it would fight every match site. +pub enum Received { + Input(Input), + /// The join succeeded; the driver checks `user_id` before feeding `Reply { Ok(Joined) }`. + Joined { + req: u64, + user_id: Uuid, + }, +} + +pub struct Connection { + url: String, + auth: JoinAuth, + current: Option<(SocketHandle, Codec)>, + events_tx: mpsc::Sender, + events_rx: mpsc::Receiver, +} + +impl Connection { + pub fn new(url: String, auth: JoinAuth) -> Connection { + let (events_tx, events_rx) = mpsc::channel(EVENT_BUFFER); + Connection { + url, + auth, + current: None, + events_tx, + events_rx, + } + } + + /// `Effect::OpenSocket`. Replaces (and so aborts) any current socket. + pub fn open(&mut self, gen: u64) { + let socket = SocketHandle::open(self.url.clone(), gen, self.events_tx.clone()); + self.current = Some((socket, Codec::new())); + } + + /// `Effect::CloseSocket`. Later events from that socket are dropped by `receive`. + pub fn close(&mut self, gen: u64) { + if self.current_gen() == Some(gen) { + self.current = None; + } + } + + /// `Effect::Send`. Returns the input to feed back when the frame could not be queued. + pub fn send(&mut self, req: u64, request: &Request, now_unix: i64) -> Option { + let (socket, codec) = self.current.as_mut()?; + let text = codec.encode(req, request, &self.auth, now_unix); + if socket.send(text) { + return None; + } + let gen = socket.gen(); + self.current = None; + Some(Input::SocketClosed { gen }) + } + + /// Cancel-safe; use as a `select!` branch. + pub async fn next_event(&mut self) -> SocketEvent { + self.events_rx + .recv() + .await + .expect("Connection keeps a sender alive") + } + + pub fn receive(&mut self, event: SocketEvent) -> Option { + let current_gen = self.current_gen()?; + match event { + SocketEvent::Opened { gen } if gen == current_gen => { + Some(Received::Input(Input::SocketOpened { gen })) + } + SocketEvent::Refused { gen, error } if gen == current_gen => { + self.current = None; + Some(Received::Input(Input::ConnectRefused { gen, error })) + } + SocketEvent::Closed { gen } if gen == current_gen => { + self.current = None; + Some(Received::Input(Input::SocketClosed { gen })) + } + SocketEvent::Text { gen, text } if gen == current_gen => { + let (_, codec) = self.current.as_mut()?; + match codec.decode(&text)? { + Incoming::Reply { req, result } => { + Some(Received::Input(Input::Reply { req, result })) + } + Incoming::Joined { req, user_id } => Some(Received::Joined { req, user_id }), + Incoming::Push(change) => Some(Received::Input(Input::Push(change))), + Incoming::ChannelClosed => { + self.current = None; + Some(Received::Input(Input::SocketClosed { gen })) + } + } + } + _ => None, + } + } + + fn current_gen(&self) -> Option { + self.current.as_ref().map(|(socket, _)| socket.gen()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::engine::machine::Response; + use crate::engine::types::{Upload, UploadKind}; + use crate::transport::test_server::{refusing_server, FakeConnection, FakeServer}; + use crate::transport::wire::socket_url; + use serde_json::json; + use std::time::Duration; + use tokio::time::timeout; + + fn auth() -> JoinAuth { + JoinAuth { + email: "a@b.c".into(), + api_key: "rpa_k".into(), + api_secret: "rps_s".into(), + } + } + + /// Next event that survives generation filtering. + async fn received(connection: &mut Connection) -> Received { + timeout(Duration::from_secs(5), async { + loop { + let event = connection.next_event().await; + if let Some(received) = connection.receive(event) { + return received; + } + } + }) + .await + .expect("input within 5 s") + } + + async fn opened(server: &mut FakeServer) -> (Connection, FakeConnection) { + let mut connection = Connection::new(server.url.clone(), auth()); + connection.open(1); + assert_eq!( + received(&mut connection).await, + Received::Input(Input::SocketOpened { gen: 1 }) + ); + let conn = server.accept().await; + (connection, conn) + } + + #[tokio::test] + async fn joins_requests_pushes_and_heartbeats_over_one_socket() { + let mut server = FakeServer::start().await; + let client_id = Uuid::from_u128(9); + let mut connection = Connection::new(socket_url(&server.url, client_id).unwrap(), auth()); + connection.open(1); + assert_eq!( + received(&mut connection).await, + Received::Input(Input::SocketOpened { gen: 1 }) + ); + let mut conn = server.accept().await; + assert!(conn.query.contains("protocol_version=2")); + assert!(conn.query.contains(&format!("client_id={client_id}"))); + assert!(conn.query.contains("vsn=2.0.0")); + + assert_eq!(connection.send(1, &Request::Join, 1_767_225_600), None); + let join = conn.recv_json().await; + assert_eq!(join[3], json!("phx_join")); + assert_eq!(join[4]["email"], json!("a@b.c")); + conn.send_json(json!(["1", "1", "sync:v2", "phx_reply", + {"status": "ok", "response": {"user_id": Uuid::nil(), "protocol_version": 2}}])) + .await; + assert_eq!( + received(&mut connection).await, + Received::Joined { + req: 1, + user_id: Uuid::nil() + } + ); + + let changes = Request::GetChangesSince { + scope: "own".into(), + cursor: 0, + limit: 500, + }; + assert_eq!(connection.send(2, &changes, 0), None); + assert_eq!(conn.recv_json().await[3], json!("get_changes_since")); + conn.send_json(json!(["1", "2", "sync:v2", "phx_reply", + {"status": "ok", "response": {"changes": [], "next_cursor": 4, "has_more": false}}])) + .await; + assert_eq!( + received(&mut connection).await, + Received::Input(Input::Reply { + req: 2, + result: Ok(Response::Changes { + changes: vec![], + next_cursor: 4, + has_more: false + }) + }) + ); + + assert_eq!(connection.send(3, &Request::Heartbeat, 0), None); + assert_eq!( + conn.recv_json().await, + json!([null, "3", "phoenix", "heartbeat", {}]) + ); + conn.send_json( + json!([null, "3", "phoenix", "phx_reply", {"status": "ok", "response": {}}]), + ) + .await; + assert_eq!( + received(&mut connection).await, + Received::Input(Input::Reply { + req: 3, + result: Ok(Response::HeartbeatOk) + }) + ); + + conn.send_json(json!(["1", null, "sync:v2", "change", { + "scope": "own", "seq": 5, "prev_seq": 4, "doc_id": Uuid::nil(), "kind": "delete", + "doc": null, "client_id": null, "upload_id": null + }])) + .await; + let Received::Input(Input::Push(change)) = received(&mut connection).await else { + panic!("expected a push"); + }; + assert_eq!(change.seq, 5); + } + + #[tokio::test] + async fn events_from_other_generations_are_dropped() { + let mut connection = Connection::new("ws://127.0.0.1:9".into(), auth()); + connection.open(1); + connection.open(2); + assert_eq!(connection.receive(SocketEvent::Opened { gen: 1 }), None); + assert_eq!( + connection.receive(SocketEvent::Text { + gen: 1, + text: "[]".into() + }), + None + ); + assert_eq!(connection.receive(SocketEvent::Closed { gen: 1 }), None); + connection.close(1); + assert_eq!( + connection.receive(SocketEvent::Opened { gen: 2 }), + Some(Received::Input(Input::SocketOpened { gen: 2 })) + ); + connection.close(2); + assert_eq!(connection.receive(SocketEvent::Closed { gen: 2 }), None); + } + + #[tokio::test] + async fn http_426_becomes_connect_refused() { + let url = refusing_server("426 Upgrade Required").await; + let mut connection = Connection::new(url, auth()); + connection.open(4); + let Received::Input(Input::ConnectRefused { gen: 4, error }) = + received(&mut connection).await + else { + panic!("expected ConnectRefused for gen 4"); + }; + assert_eq!(error.code, "update_required"); + assert!(error.is_fatal); + } + + #[tokio::test] + async fn channel_error_closes_the_socket_and_later_frames_are_dropped() { + let mut server = FakeServer::start().await; + let (mut connection, mut conn) = opened(&mut server).await; + conn.send_json(json!(["1", "1", "sync:v2", "phx_error", {}])) + .await; + assert_eq!( + received(&mut connection).await, + Received::Input(Input::SocketClosed { gen: 1 }) + ); + let late = json!(["1", null, "sync:v2", "phx_close", {}]).to_string(); + assert_eq!( + connection.receive(SocketEvent::Text { gen: 1, text: late }), + None + ); + assert_eq!(connection.send(5, &Request::Heartbeat, 0), None); + } + + #[tokio::test] + async fn server_drop_mid_request_yields_only_socket_closed() { + let mut server = FakeServer::start().await; + let (mut connection, mut conn) = opened(&mut server).await; + let upload = Request::Upload(Upload { + upload_id: Uuid::from_u128(1), + doc_id: Uuid::from_u128(2), + kind: UploadKind::Create, + base_hash: None, + payload: json!({"title": "t"}), + }); + assert_eq!(connection.send(2, &upload, 0), None); + assert_eq!(conn.recv_json().await[3], json!("upload")); + conn.close().await; + assert_eq!( + received(&mut connection).await, + Received::Input(Input::SocketClosed { gen: 1 }) + ); + let nothing_more = timeout(Duration::from_millis(300), connection.next_event()).await; + assert!( + nothing_more.is_err(), + "no reply may be invented for the upload" + ); + } + + #[tokio::test] + async fn send_after_the_socket_task_ended_reports_socket_closed() { + let mut server = FakeServer::start().await; + let (mut connection, conn) = opened(&mut server).await; + conn.close().await; + // Take Closed off the channel without `receive`, as when the core sends first. + let event = timeout(Duration::from_secs(5), connection.next_event()) + .await + .unwrap(); + assert_eq!(event, SocketEvent::Closed { gen: 1 }); + let reported = timeout(Duration::from_secs(5), async { + loop { + if let Some(input) = connection.send(9, &Request::Heartbeat, 0) { + return input; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("send fails once the task has ended"); + assert_eq!(reported, Input::SocketClosed { gen: 1 }); + assert_eq!(connection.receive(event), None, "reported once only"); + } + + #[tokio::test] + async fn dropping_the_connection_closes_the_server_side() { + let mut server = FakeServer::start().await; + let (connection, mut conn) = opened(&mut server).await; + drop(connection); + conn.wait_closed().await; + } +} diff --git a/replicant-client/src/transport/mod.rs b/replicant-client/src/transport/mod.rs new file mode 100644 index 0000000..3add14f --- /dev/null +++ b/replicant-client/src/transport/mod.rs @@ -0,0 +1,8 @@ +//! Protocol v2 transport: wire payloads, Phoenix framing, and the websocket. + +pub mod codec; +pub mod connection; +pub mod socket; +#[cfg(test)] +pub(crate) mod test_server; +pub mod wire; diff --git a/replicant-client/src/transport/socket.rs b/replicant-client/src/transport/socket.rs new file mode 100644 index 0000000..16b01a3 --- /dev/null +++ b/replicant-client/src/transport/socket.rs @@ -0,0 +1,229 @@ +//! One websocket connection, run in its own task and tagged with the core's socket generation. + +use futures_util::{SinkExt, StreamExt}; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use tokio_tungstenite::tungstenite::{self, Message}; + +use crate::engine::types::ServerError; + +#[derive(Debug, Clone, PartialEq)] +pub enum SocketEvent { + Opened { + gen: u64, + }, + /// The upgrade was answered with an error the core must classify (HTTP 426). + Refused { + gen: u64, + error: ServerError, + }, + Text { + gen: u64, + text: String, + }, + /// Dial failed, the peer closed, or a read or write failed. Sent at most once. + Closed { + gen: u64, + }, +} + +pub struct SocketHandle { + gen: u64, + outgoing: mpsc::UnboundedSender, + task: JoinHandle<()>, +} + +impl SocketHandle { + pub fn open(url: String, gen: u64, events: mpsc::Sender) -> SocketHandle { + let (outgoing, mut outgoing_rx) = mpsc::unbounded_channel::(); + let task = tokio::spawn(async move { + let ws = match tokio_tungstenite::connect_async(url.as_str()).await { + Ok((ws, _response)) => ws, + Err(error) => { + let event = match refusal(&error) { + Some(error) => SocketEvent::Refused { gen, error }, + None => SocketEvent::Closed { gen }, + }; + let _ = events.send(event).await; + return; + } + }; + if events.send(SocketEvent::Opened { gen }).await.is_err() { + return; + } + let (mut sink, mut stream) = ws.split(); + loop { + tokio::select! { + text = outgoing_rx.recv() => match text { + Some(text) => { + if sink.send(Message::Text(text)).await.is_err() { + break; + } + } + None => break, + }, + message = stream.next() => match message { + Some(Ok(Message::Text(text))) => { + if events.send(SocketEvent::Text { gen, text }).await.is_err() { + return; + } + } + Some(Ok(Message::Close(_))) | Some(Err(_)) | None => break, + Some(Ok(_)) => {} + }, + } + } + let _ = events.send(SocketEvent::Closed { gen }).await; + }); + SocketHandle { + gen, + outgoing, + task, + } + } + + pub fn gen(&self) -> u64 { + self.gen + } + + /// Queues a text frame; false once the socket task has ended. + pub fn send(&self, text: String) -> bool { + self.outgoing.send(text).is_ok() + } + + pub fn is_finished(&self) -> bool { + self.task.is_finished() + } +} + +impl Drop for SocketHandle { + fn drop(&mut self) { + self.task.abort(); + } +} + +fn refusal(error: &tungstenite::Error) -> Option { + match error { + tungstenite::Error::Http(response) if response.status().as_u16() == 426 => { + let mut update_required = ServerError::new("update_required"); + update_required.is_fatal = true; + Some(update_required) + } + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::test_server::{refusing_server, silent_server, FakeServer}; + use serde_json::json; + use std::time::Duration; + use tokio::time::timeout; + + async fn next(events: &mut mpsc::Receiver) -> SocketEvent { + timeout(Duration::from_secs(5), events.recv()) + .await + .expect("event within 5 s") + .expect("socket task still holds the sender") + } + + async fn until_finished(socket: &SocketHandle) { + timeout(Duration::from_secs(5), async { + while !socket.is_finished() { + tokio::task::yield_now().await; + } + }) + .await + .expect("socket task ends within 5 s"); + } + + #[tokio::test] + async fn opens_and_exchanges_text_frames() { + let mut server = FakeServer::start().await; + let (events_tx, mut events) = mpsc::channel(16); + let socket = SocketHandle::open(server.url.clone(), 1, events_tx); + assert_eq!(next(&mut events).await, SocketEvent::Opened { gen: 1 }); + let mut conn = server.accept().await; + assert_eq!(conn.query, ""); + + assert!(socket.send(json!(["1", "1", "sync:v2", "phx_join", {}]).to_string())); + assert_eq!(conn.recv_json().await[3], json!("phx_join")); + + let push = json!([null, null, "sync:v2", "change", {}]); + conn.send_json(push.clone()).await; + assert_eq!( + next(&mut events).await, + SocketEvent::Text { + gen: 1, + text: push.to_string() + } + ); + } + + #[tokio::test] + async fn peer_close_reports_closed_and_later_sends_fail() { + let mut server = FakeServer::start().await; + let (events_tx, mut events) = mpsc::channel(16); + let socket = SocketHandle::open(server.url.clone(), 3, events_tx); + assert_eq!(next(&mut events).await, SocketEvent::Opened { gen: 3 }); + server.accept().await.close().await; + assert_eq!(next(&mut events).await, SocketEvent::Closed { gen: 3 }); + until_finished(&socket).await; + assert!(!socket.send("late".into())); + } + + #[tokio::test] + async fn http_426_is_refused_with_update_required() { + let url = refusing_server("426 Upgrade Required").await; + let (events_tx, mut events) = mpsc::channel(16); + let _socket = SocketHandle::open(url, 2, events_tx); + let SocketEvent::Refused { gen: 2, error } = next(&mut events).await else { + panic!("expected Refused for gen 2"); + }; + assert_eq!(error.code, "update_required"); + assert!(error.is_fatal); + } + + #[tokio::test] + async fn other_http_errors_and_dial_failures_are_plain_closes() { + let url = refusing_server("403 Forbidden").await; + let (events_tx, mut events) = mpsc::channel(16); + let _forbidden = SocketHandle::open(url, 1, events_tx.clone()); + assert_eq!(next(&mut events).await, SocketEvent::Closed { gen: 1 }); + + let unused = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let dead_url = format!("ws://{}", unused.local_addr().unwrap()); + drop(unused); + let _dial = SocketHandle::open(dead_url, 2, events_tx); + assert_eq!(next(&mut events).await, SocketEvent::Closed { gen: 2 }); + } + + #[tokio::test] + async fn dropping_the_handle_ends_a_stalled_dial_without_events() { + let url = silent_server().await; + let (events_tx, mut events) = mpsc::channel(16); + let socket = SocketHandle::open(url, 1, events_tx); + tokio::time::sleep(Duration::from_millis(50)).await; + drop(socket); + let after_drop = timeout(Duration::from_secs(5), events.recv()) + .await + .expect("aborted task drops its sender"); + assert_eq!(after_drop, None); + } + + #[tokio::test] + async fn dropping_an_open_handle_ends_the_task_without_events() { + let mut server = FakeServer::start().await; + let (events_tx, mut events) = mpsc::channel(16); + let socket = SocketHandle::open(server.url.clone(), 1, events_tx); + assert_eq!(next(&mut events).await, SocketEvent::Opened { gen: 1 }); + let mut conn = server.accept().await; + drop(socket); + let after_drop = timeout(Duration::from_secs(5), events.recv()) + .await + .expect("aborted task drops its sender"); + assert_eq!(after_drop, None); + conn.wait_closed().await; + } +} diff --git a/replicant-client/src/transport/test_server.rs b/replicant-client/src/transport/test_server.rs new file mode 100644 index 0000000..0737a9d --- /dev/null +++ b/replicant-client/src/transport/test_server.rs @@ -0,0 +1,128 @@ +//! Local websocket servers for transport tests. + +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use serde_json::Value; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::mpsc; +use tokio::time::timeout; +use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response}; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::WebSocketStream; + +const WAIT: Duration = Duration::from_secs(5); + +pub struct FakeServer { + pub url: String, + connections: mpsc::UnboundedReceiver, +} + +pub struct FakeConnection { + pub query: String, + ws: WebSocketStream, +} + +impl FakeServer { + pub async fn start() -> FakeServer { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + let (connection_tx, connections) = mpsc::unbounded_channel(); + tokio::spawn(async move { + while let Ok((tcp, _)) = listener.accept().await { + let mut query = String::new(); + let accepted = tokio_tungstenite::accept_hdr_async( + tcp, + |request: &Request, response: Response| -> Result { + query = request.uri().query().unwrap_or_default().to_string(); + Ok(response) + }, + ) + .await; + if let Ok(ws) = accepted { + let _ = connection_tx.send(FakeConnection { query, ws }); + } + } + }); + FakeServer { url, connections } + } + + pub async fn accept(&mut self) -> FakeConnection { + timeout(WAIT, self.connections.recv()) + .await + .expect("client connects within 5 s") + .expect("server task alive") + } +} + +impl FakeConnection { + pub async fn recv_json(&mut self) -> Value { + loop { + match timeout(WAIT, self.ws.next()) + .await + .expect("frame within 5 s") + { + Some(Ok(Message::Text(text))) => return serde_json::from_str(&text).unwrap(), + Some(Ok(_)) => continue, + other => panic!("connection ended: {other:?}"), + } + } + } + + pub async fn send_json(&mut self, frame: Value) { + self.ws + .send(Message::Text(frame.to_string())) + .await + .unwrap(); + } + + pub async fn close(mut self) { + let _ = self.ws.close(None).await; + } + + /// Resolves once the client side has gone away. + pub async fn wait_closed(&mut self) { + timeout(WAIT, async { + while let Some(Ok(message)) = self.ws.next().await { + if message.is_close() { + break; + } + } + }) + .await + .expect("client closes within 5 s"); + } +} + +/// Answers every upgrade with `status_line` (e.g. "426 Upgrade Required") and a JSON body. +pub async fn refusing_server(status_line: &'static str) -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { + while let Ok((mut tcp, _)) = listener.accept().await { + let mut request = [0u8; 4096]; + let _ = tcp.read(&mut request).await; + let body = r#"{"code":"update_required","is_fatal":true}"#; + let response = format!( + "HTTP/1.1 {status_line}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + let _ = tcp.write_all(response.as_bytes()).await; + } + }); + url +} + +/// Accepts TCP connections and never answers the websocket handshake. +pub async fn silent_server() -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((tcp, _)) = listener.accept().await { + held.push(tcp); + } + }); + url +} diff --git a/replicant-client/src/transport/wire.rs b/replicant-client/src/transport/wire.rs new file mode 100644 index 0000000..a545a26 --- /dev/null +++ b/replicant-client/src/transport/wire.rs @@ -0,0 +1,222 @@ +//! Protocol v2 payloads as fixed by replicant-server 0.5.0 (`lib/replicant_server/sync`). + +use hmac::{Hmac, Mac}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::Sha256; +use uuid::Uuid; + +use crate::engine::types::{Change, DocEnvelope, Seq, UploadKind}; + +pub const PROTOCOL_VERSION: &str = "2"; +pub const CHANNEL_TOPIC: &str = "sync:v2"; +/// Selects Phoenix's V2 JSON serializer (array frames). +const SERIALIZER_VSN: &str = "2.0.0"; + +/// Credentials for the `sync:v2` join. Each join is signed with a fresh timestamp. +/// Holds `api_secret`, so it derives neither `Debug` nor `PartialEq`. +pub struct JoinAuth { + pub email: String, + pub api_key: String, + pub api_secret: String, +} + +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct JoinPayload { + pub email: String, + pub api_key: String, + pub signature: String, + pub timestamp: i64, +} + +impl JoinAuth { + /// Hex HMAC-SHA256 over `"{timestamp}.{email}.{api_key}."` (the body is empty for a join). + pub fn sign(&self, timestamp: i64) -> JoinPayload { + let mut mac = Hmac::::new_from_slice(self.api_secret.as_bytes()) + .expect("HMAC accepts any key size"); + mac.update(format!("{timestamp}.{}.{}.", self.email, self.api_key).as_bytes()); + JoinPayload { + email: self.email.clone(), + api_key: self.api_key.clone(), + signature: hex::encode(mac.finalize().into_bytes()), + timestamp, + } + } +} + +#[derive(Debug, Clone, PartialEq, Deserialize)] +pub struct JoinReply { + pub user_id: Uuid, + pub protocol_version: u32, +} + +#[derive(Debug, Clone, PartialEq, Deserialize)] +pub struct ChangesPage { + pub changes: Vec, + pub next_cursor: Seq, + pub has_more: bool, +} + +#[derive(Debug, Clone, PartialEq, Deserialize)] +pub struct SnapshotPage { + pub docs: Vec, + pub snapshot_seq: Seq, + pub next_page_token: Option, +} + +#[derive(Debug, Serialize)] +pub struct GetChangesSince<'a> { + pub scope: &'a str, + pub cursor: Seq, + pub limit: u32, +} + +#[derive(Debug, Serialize)] +pub struct GetSnapshot<'a> { + pub scope: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + pub page_token: Option<&'a str>, +} + +#[derive(Debug, Serialize)] +pub struct UploadRequest<'a> { + pub upload_id: Uuid, + pub doc_id: Uuid, + pub kind: UploadKind, + #[serde(skip_serializing_if = "Option::is_none")] + pub base_hash: Option<&'a str>, + pub payload: &'a Value, +} + +#[derive(Debug, Serialize)] +pub struct GetDocument { + pub doc_id: Uuid, +} + +/// Payload of every `phx_reply` frame. +#[derive(Debug, Clone, PartialEq, Deserialize)] +pub struct ReplyEnvelope { + pub status: String, + pub response: Value, +} + +/// `server_url` may be http(s) or ws(s), with or without the `/socket/websocket` path. +pub fn socket_url(server_url: &str, client_id: Uuid) -> Result { + let ws_url = if let Some(rest) = server_url.strip_prefix("http://") { + format!("ws://{rest}") + } else if let Some(rest) = server_url.strip_prefix("https://") { + format!("wss://{rest}") + } else { + server_url.to_string() + }; + let mut url = + url::Url::parse(&ws_url).map_err(|e| format!("invalid server url {server_url}: {e}"))?; + if !matches!(url.scheme(), "ws" | "wss") { + return Err(format!("unsupported server url scheme: {server_url}")); + } + if !url.path().ends_with("/socket/websocket") { + let path = format!("{}/socket/websocket", url.path().trim_end_matches('/')); + url.set_path(&path); + } + url.query_pairs_mut() + .append_pair("protocol_version", PROTOCOL_VERSION) + .append_pair("client_id", &client_id.to_string()) + .append_pair("vsn", SERIALIZER_VSN); + Ok(url.into()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::engine::types::ServerError; + use serde_json::json; + + #[test] + fn join_signature_matches_the_server_hmac_scheme() { + let auth = JoinAuth { + email: "wire@example.com".into(), + api_key: "rpa_key".into(), + api_secret: "rps_test".into(), + }; + let payload = auth.sign(1_767_225_600); + let expected = "a3ffd0937b146736f3b60ffb28b943cf66c9d45d8c5e401fac9044d10db6f904"; + assert_eq!( + serde_json::to_value(&payload).unwrap(), + json!({ + "email": "wire@example.com", + "api_key": "rpa_key", + "signature": expected, + "timestamp": 1_767_225_600 + }) + ); + } + + #[test] + fn socket_url_adds_path_version_client_and_serializer() { + let id = Uuid::from_u128(1); + let query = format!("protocol_version=2&client_id={id}&vsn=2.0.0"); + assert_eq!( + socket_url("http://localhost:4000", id).unwrap(), + format!("ws://localhost:4000/socket/websocket?{query}") + ); + assert_eq!( + socket_url("https://sync.example.com/", id).unwrap(), + format!("wss://sync.example.com/socket/websocket?{query}") + ); + assert_eq!( + socket_url("wss://sync.example.com/socket/websocket", id).unwrap(), + format!("wss://sync.example.com/socket/websocket?{query}") + ); + assert!(socket_url("ftp://sync.example.com", id).is_err()); + assert!(socket_url("not a url", id).is_err()); + } + + #[test] + fn optional_request_fields_are_omitted_not_null() { + assert_eq!( + serde_json::to_value(GetSnapshot { + scope: "own", + page_token: None + }) + .unwrap(), + json!({"scope": "own"}) + ); + let payload = json!({"title": "t"}); + let create = UploadRequest { + upload_id: Uuid::nil(), + doc_id: Uuid::nil(), + kind: UploadKind::Create, + base_hash: None, + payload: &payload, + }; + assert_eq!( + serde_json::to_value(create).unwrap(), + json!({ + "upload_id": Uuid::nil(), + "doc_id": Uuid::nil(), + "kind": "create", + "payload": {"title": "t"} + }) + ); + } + + #[test] + fn server_error_reads_known_fields_and_ignores_the_rest() { + let error: ServerError = serde_json::from_value(json!({ + "code": "hash_mismatch", + "is_fatal": false, + "current_hash": "h", + "current_seq": 3, + "doc_id": Uuid::nil(), + "scope": "own" + })) + .unwrap(); + assert_eq!(error.code, "hash_mismatch"); + assert_eq!(error.current_hash.as_deref(), Some("h")); + assert_eq!(error.current_seq, Some(3)); + assert_eq!(error.retry_after_ms, None); + + let bare: ServerError = serde_json::from_value(json!({"code": "internal"})).unwrap(); + assert!(!bare.is_fatal); + } +} diff --git a/replicant-client/tests/v2_smoke.rs b/replicant-client/tests/v2_smoke.rs new file mode 100644 index 0000000..83f1aff --- /dev/null +++ b/replicant-client/tests/v2_smoke.rs @@ -0,0 +1,145 @@ +//! End-to-end smoke test of the v2 transport against a real replicant-server v2. +//! +//! Ignored by default. `test/run_phoenix_interop_local.sh` boots a server, seeds an account +//! and exports the variables read below. + +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use replicant_client::engine::machine::{Input, Request, Response}; +use replicant_client::transport::connection::{Connection, Received}; +use replicant_client::transport::wire::{socket_url, JoinAuth}; +use tokio::time::timeout; +use uuid::Uuid; + +const WAIT: Duration = Duration::from_secs(10); + +fn required_env(name: &str) -> String { + std::env::var(name).unwrap_or_else(|_| panic!("{name} is required for the v2 smoke test")) +} + +fn server_socket_url() -> String { + socket_url(&required_env("SYNC_SERVER_URL"), Uuid::new_v4()).unwrap() +} + +fn seeded_auth() -> JoinAuth { + JoinAuth { + email: std::env::var("REPLICANT_TEST_EMAIL") + .unwrap_or_else(|_| "integration-test@example.com".into()), + api_key: required_env("REPLICANT_API_KEY"), + api_secret: required_env("REPLICANT_API_SECRET"), + } +} + +fn now_unix() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() as i64 +} + +/// Next event that survives generation filtering. +async fn received(connection: &mut Connection) -> Received { + timeout(WAIT, async { + loop { + let event = connection.next_event().await; + if let Some(received) = connection.receive(event) { + return received; + } + } + }) + .await + .expect("input within 10 s") +} + +async fn opened(url: String, auth: JoinAuth) -> Connection { + let mut connection = Connection::new(url, auth); + connection.open(1); + assert_eq!( + received(&mut connection).await, + Received::Input(Input::SocketOpened { gen: 1 }) + ); + connection +} + +#[tokio::test] +#[ignore = "needs a v2 server: run through test/run_phoenix_interop_local.sh"] +async fn joins_requests_heartbeats_and_closes() { + let mut connection = opened(server_socket_url(), seeded_auth()).await; + + assert_eq!(connection.send(1, &Request::Join, now_unix()), None); + let joined = received(&mut connection).await; + let Received::Joined { req: 1, user_id } = joined else { + panic!("expected the join to succeed, got {joined:?}"); + }; + let seeded_user = Uuid::parse_str(&required_env("REPLICANT_TEST_USER_ID")).unwrap(); + assert_eq!(user_id, seeded_user); + + let changes = Request::GetChangesSince { + scope: "own".into(), + cursor: 0, + limit: 500, + }; + assert_eq!(connection.send(2, &changes, now_unix()), None); + let reply = received(&mut connection).await; + assert!( + matches!( + reply, + Received::Input(Input::Reply { + req: 2, + result: Ok(Response::Changes { .. }) + }) + ), + "{reply:?}" + ); + + assert_eq!(connection.send(3, &Request::Heartbeat, now_unix()), None); + assert_eq!( + received(&mut connection).await, + Received::Input(Input::Reply { + req: 3, + result: Ok(Response::HeartbeatOk) + }) + ); + + connection.close(1); + let after_close = timeout(Duration::from_millis(500), connection.next_event()).await; + assert!( + after_close.is_err(), + "a closed socket reports nothing: {after_close:?}" + ); +} + +#[tokio::test] +#[ignore = "needs a v2 server: run through test/run_phoenix_interop_local.sh"] +async fn unsupported_protocol_version_is_refused_with_update_required() { + let url = server_socket_url().replace("protocol_version=2", "protocol_version=1"); + let mut connection = Connection::new(url, seeded_auth()); + connection.open(1); + let refused = received(&mut connection).await; + let Received::Input(Input::ConnectRefused { gen: 1, error }) = refused else { + panic!("expected HTTP 426 as ConnectRefused, got {refused:?}"); + }; + assert_eq!(error.code, "update_required"); + assert!(error.is_fatal); +} + +#[tokio::test] +#[ignore = "needs a v2 server: run through test/run_phoenix_interop_local.sh"] +async fn wrong_secret_is_a_fatal_auth_invalid_join_error() { + let wrong_secret = JoinAuth { + api_secret: "rps_wrong".into(), + ..seeded_auth() + }; + let mut connection = opened(server_socket_url(), wrong_secret).await; + assert_eq!(connection.send(1, &Request::Join, now_unix()), None); + let rejected = received(&mut connection).await; + let Received::Input(Input::Reply { + req: 1, + result: Err(error), + }) = rejected + else { + panic!("expected a join error, got {rejected:?}"); + }; + assert_eq!(error.code, "auth_invalid"); + assert!(error.is_fatal); +} diff --git a/replicant-client/tests/wire_contract_tests.rs b/replicant-client/tests/wire_contract_tests.rs index cde6042..6b3233b 100644 --- a/replicant-client/tests/wire_contract_tests.rs +++ b/replicant-client/tests/wire_contract_tests.rs @@ -1,112 +1,19 @@ -//! Contract test: every recorded server v2 wire frame must deserialize into the -//! client's engine types. Proves wire compatibility before the client driver exists. +//! Contract test: every recorded server v2 wire frame round-trips through the client's real +//! wire types and codec. use std::collections::BTreeSet; -use replicant_client::engine::types::{Change, ChangeKind, DocEnvelope, Scope, Seq, UploadKind}; +use replicant_client::engine::machine::{Request, Response}; +use replicant_client::engine::types::{ChangeKind, DocEnvelope, ServerError, Upload, UploadKind}; +use replicant_client::transport::codec::{Codec, Incoming}; +use replicant_client::transport::wire::{socket_url, JoinAuth, ReplyEnvelope}; use replicant_core::patches::calculate_checksum; -use serde::Deserialize; -use serde_json::Value; +use serde_json::{json, Map, Value}; use uuid::Uuid; const FRAMES_JSON: &str = include_str!("fixtures/server_v2/v2_frames.json"); - -/// Mirrors `machine::Response::Changes`, which is not a serde type. -#[derive(Debug, Deserialize)] -struct WireChangesResponse { - changes: Vec, - next_cursor: Seq, - has_more: bool, -} - -/// Mirrors `machine::Response::SnapshotPage`, which is not a serde type. -#[derive(Debug, Deserialize)] -struct WireSnapshotResponse { - docs: Vec, - snapshot_seq: Seq, - next_page_token: Option, -} - -/// Mirrors `types::ServerError` plus the `scope`/`doc_id` keys some error replies carry, -/// which `ServerError` itself does not store. -#[derive(Debug, Deserialize)] -struct WireError { - code: String, - is_fatal: bool, - retry_after_ms: Option, - current_hash: Option, - current_seq: Option, - existing_owner: Option, - scope: Option, - doc_id: Option, -} - -#[derive(Debug, Deserialize)] -struct WireJoinReply { - user_id: Uuid, - protocol_version: u32, -} - -/// Mirrors `machine::Request::Join`: the `phx_join` payload. -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -struct Join { - email: String, - api_key: String, - signature: String, - timestamp: i64, -} - -/// Mirrors `machine::Request::GetChangesSince`, which is not a serde type. -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -struct GetChangesSince { - scope: Scope, - cursor: Seq, - limit: u32, -} - -/// Mirrors `machine::Request::GetSnapshot`, which is not a serde type. -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -struct GetSnapshot { - scope: Scope, - page_token: Option, -} - -/// Mirrors `machine::Request::Upload` (`types::Upload`), which is not a serde type. -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -struct Upload { - upload_id: Uuid, - doc_id: Uuid, - kind: String, - base_hash: Option, - payload: Value, -} - -/// Mirrors `machine::Request::GetDocument`, which is not a serde type. -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -struct GetDocument { - doc_id: Uuid, -} - -/// The websocket connect query params. -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -struct SocketConnectParams { - protocol_version: String, - client_id: Uuid, -} - -#[derive(Debug, Deserialize)] -struct ReplyEnvelope { - status: String, - response: Value, -} - const FATAL_CODES: &[&str] = &["update_required", "auth_invalid", "account_disabled"]; +const JOIN_TIMESTAMP: i64 = 1_767_225_600; /// Every frame name the server fixture is expected to contain. Keeping this list explicit /// means a frame added on the server fails this test until the client covers it too. @@ -147,340 +54,398 @@ const EXPECTED_FRAME_NAMES: &[&str] = &[ "validation_reply", ]; -fn load_frames() -> serde_json::Map { +fn load_frames() -> Map { let root: Value = serde_json::from_str(FRAMES_JSON).expect("v2_frames.json must parse"); root.as_object() .expect("v2_frames.json must be a JSON object of named frames") .clone() } -/// Parses a named entry as a `[join_ref, ref, topic, event, payload]` frame and returns its -/// payload, asserting the topic and event match. -fn frame_payload(frames: &serde_json::Map, name: &str, event: &str) -> Value { - let frame = frames[name] - .as_array() - .unwrap_or_else(|| panic!("{name} must be a 5-element phoenix frame array")); - assert_eq!(frame.len(), 5, "{name} must have 5 frame elements"); - assert_eq!( - frame[2], - Value::String("sync:v2".to_string()), - "{name} has unexpected topic" - ); +fn auth() -> JoinAuth { + JoinAuth { + email: "wire@example.com".into(), + api_key: "".into(), + api_secret: "rps_test".into(), + } +} + +/// A codec that has joined with ref "1", as every recorded channel frame assumes. +fn joined_codec() -> Codec { + let mut codec = Codec::new(); + codec.encode(1, &Request::Join, &auth(), JOIN_TIMESTAMP); + codec +} + +fn encode(codec: &mut Codec, req: u64, request: &Request) -> Value { + serde_json::from_str(&codec.encode(req, request, &auth(), JOIN_TIMESTAMP)) + .expect("codec writes JSON") +} + +/// Sends `request` with ref "2", as recorded, and decodes the named reply to it. +fn reply_to( + frames: &Map, + name: &str, + request: &Request, +) -> Result { + let mut codec = joined_codec(); + codec.encode(2, request, &auth(), JOIN_TIMESTAMP); + match codec.decode(&frames[name].to_string()) { + Some(Incoming::Reply { req: 2, result }) => result, + other => panic!("{name}: expected a reply to ref 2, got {other:?}"), + } +} + +fn expect_error(name: &str, result: Result) -> ServerError { + let error = result.expect_err(name); assert_eq!( - frame[3], - Value::String(event.to_string()), - "{name} has unexpected event" + error.is_fatal, + FATAL_CODES.contains(&error.code.as_str()), + "{name}: is_fatal mismatch for code {}", + error.code ); - frame[4].clone() + error } -/// Parses a named entry as a `phx_reply` frame and returns its `response`, asserting `status`. -fn reply_response(frames: &serde_json::Map, name: &str, status: &str) -> Value { - let payload = frame_payload(frames, name, "phx_reply"); - let envelope: ReplyEnvelope = serde_json::from_value(payload) +/// The raw `response` of a recorded reply, for keys `ServerError` does not keep. +fn raw_response(frames: &Map, name: &str) -> Value { + let envelope: ReplyEnvelope = serde_json::from_value(frames[name][4].clone()) .unwrap_or_else(|e| panic!("{name} payload must be a reply envelope: {e}")); - assert_eq!(envelope.status, status, "{name} has unexpected status"); envelope.response } -fn assert_error_shape(name: &str, response: Value) -> WireError { - let err: WireError = serde_json::from_value(response) - .unwrap_or_else(|e| panic!("{name} must parse as WireError: {e}")); - assert_eq!( - err.is_fatal, - FATAL_CODES.contains(&err.code.as_str()), - "{name}: is_fatal mismatch for code {}", - err.code - ); - err +fn changes_request() -> Request { + Request::GetChangesSince { + scope: "own".into(), + cursor: 0, + limit: 500, + } } -#[test] -fn every_expected_frame_is_present_and_no_others() { - let frames = load_frames(); - let actual: BTreeSet<&str> = frames.keys().map(String::as_str).collect(); - let expected: BTreeSet<&str> = EXPECTED_FRAME_NAMES.iter().copied().collect(); - assert_eq!(actual, expected, "frame set changed on the server side"); +fn snapshot_request() -> Request { + Request::GetSnapshot { + scope: "own".into(), + page_token: None, + } } -#[test] -fn change_push_upsert_deserializes_into_change() { - let frames = load_frames(); - let payload = frame_payload(&frames, "change_push", "change"); - let change: Change = serde_json::from_value(payload).expect("change_push must be a Change"); - assert_eq!(change.kind, ChangeKind::Upsert); - assert!(change.doc.is_some()); - assert!(change.prev_seq < change.seq); +fn document_request() -> Request { + Request::GetDocument { + doc_id: Uuid::nil(), + } +} + +fn upload(kind: UploadKind, base_hash: Option, payload: Value) -> Request { + Request::Upload(Upload { + upload_id: Uuid::nil(), + doc_id: Uuid::nil(), + kind, + base_hash, + payload, + }) +} + +fn create_upload() -> Request { + upload(UploadKind::Create, None, json!({"title": "Wire"})) } #[test] -fn change_push_delete_has_no_doc_and_advances_seq() { +fn every_expected_frame_is_present_and_no_others() { let frames = load_frames(); - let payload = frame_payload(&frames, "change_push_delete", "change"); - let change: Change = - serde_json::from_value(payload).expect("change_push_delete must be a Change"); - assert_eq!(change.kind, ChangeKind::Delete); - assert_eq!(change.doc, None); - assert!(change.prev_seq < change.seq); + let actual: BTreeSet<&str> = frames.keys().map(String::as_str).collect(); + let expected: BTreeSet<&str> = EXPECTED_FRAME_NAMES.iter().copied().collect(); + assert_eq!(actual, expected, "frame set changed on the server side"); } #[test] -fn changes_reply_empty_deserializes_into_changes_response() { +fn change_pushes_decode_into_changes() { let frames = load_frames(); - let response = reply_response(&frames, "changes_reply", "ok"); - let page: WireChangesResponse = - serde_json::from_value(response).expect("changes_reply must match Response::Changes"); - assert!(page.changes.is_empty()); - assert!(!page.has_more); - assert_eq!(page.next_cursor, 2); + let mut codec = joined_codec(); + + let Some(Incoming::Push(upsert)) = codec.decode(&frames["change_push"].to_string()) else { + panic!("change_push must decode as a push"); + }; + assert_eq!(upsert.kind, ChangeKind::Upsert); + assert!(upsert.doc.is_some()); + assert!(upsert.prev_seq < upsert.seq); + + let Some(Incoming::Push(delete)) = codec.decode(&frames["change_push_delete"].to_string()) + else { + panic!("change_push_delete must decode as a push"); + }; + assert_eq!(delete.kind, ChangeKind::Delete); + assert_eq!(delete.doc, None); + assert!(delete.prev_seq < delete.seq); } #[test] -fn changes_reply_populated_deserializes_every_changes_item() { +fn changes_replies_decode_into_response_changes() { let frames = load_frames(); - let response = reply_response(&frames, "changes_reply_populated", "ok"); - let page: WireChangesResponse = serde_json::from_value(response) - .expect("changes_reply_populated must match Response::Changes"); - assert_eq!(page.changes.len(), 1); - assert_eq!(page.changes[0].kind, ChangeKind::Delete); - assert_eq!(page.changes[0].doc, None); + assert_eq!( + reply_to(&frames, "changes_reply", &changes_request()), + Ok(Response::Changes { + changes: vec![], + next_cursor: 2, + has_more: false + }) + ); + let Ok(Response::Changes { + changes, + next_cursor: 4, + has_more: false, + }) = reply_to(&frames, "changes_reply_populated", &changes_request()) + else { + panic!("changes_reply_populated must decode as a changes page"); + }; + assert_eq!(changes.len(), 1); + assert_eq!(changes[0].kind, ChangeKind::Delete); + assert_eq!(changes[0].doc, None); } #[test] -fn snapshot_reply_deserializes_into_snapshot_response() { +fn snapshot_reply_decodes_into_snapshot_page() { let frames = load_frames(); - let response = reply_response(&frames, "snapshot_reply", "ok"); - let page: WireSnapshotResponse = - serde_json::from_value(response).expect("snapshot_reply must match Response::SnapshotPage"); - assert_eq!(page.docs.len(), 1); - assert_eq!(page.snapshot_seq, 3); - assert_eq!(page.next_page_token, None); + let Ok(Response::SnapshotPage { + docs, + snapshot_seq, + next_page_token, + }) = reply_to(&frames, "snapshot_reply", &snapshot_request()) + else { + panic!("snapshot_reply must decode as a snapshot page"); + }; + assert_eq!(docs.len(), 1); + assert_eq!(snapshot_seq, 3); + assert_eq!(next_page_token, None); } #[test] -fn join_reply_carries_protocol_version_two() { +fn join_reply_decodes_as_joined_with_the_user_id() { let frames = load_frames(); - let response = reply_response(&frames, "join_reply", "ok"); - let join: WireJoinReply = - serde_json::from_value(response).expect("join_reply must parse as WireJoinReply"); - assert_eq!(join.protocol_version, 2); - assert_eq!(join.user_id, Uuid::from_u128(0)); + let mut codec = joined_codec(); + assert_eq!( + codec.decode(&frames["join_reply"].to_string()), + Some(Incoming::Joined { + req: 1, + user_id: Uuid::nil() + }) + ); } #[test] -fn document_shaped_replies_deserialize_into_doc_envelope() { +fn document_shaped_replies_decode_into_doc_envelopes() { let frames = load_frames(); - for name in [ - "document_reply", - "publish_reply", - "publish_update_reply", - "unpublish_reply", - "upload_reply", - ] { - let response = reply_response(&frames, name, "ok"); - let _doc: DocEnvelope = - serde_json::from_value(response).unwrap_or_else(|e| panic!("{name}: {e}")); + assert!(matches!( + reply_to(&frames, "document_reply", &document_request()), + Ok(Response::Document(_)) + )); + assert!(matches!( + reply_to(&frames, "upload_reply", &create_upload()), + Ok(Response::Uploaded(_)) + )); + // The client sends no publish requests yet; their replies still share the envelope. + for name in ["publish_reply", "publish_update_reply", "unpublish_reply"] { + let _doc: DocEnvelope = serde_json::from_value(raw_response(&frames, name)) + .unwrap_or_else(|e| panic!("{name}: {e}")); } } #[test] -fn error_replies_and_socket_refusal_parse_extra_keys() { +fn error_replies_decode_into_server_errors() { let frames = load_frames(); - let cursor_too_old = assert_error_shape( + let cursor_too_old = expect_error( "cursor_too_old_reply", - reply_response(&frames, "cursor_too_old_reply", "error"), + reply_to(&frames, "cursor_too_old_reply", &changes_request()), ); - assert_eq!(cursor_too_old.scope.as_deref(), Some("own")); + assert_eq!(cursor_too_old.code, "cursor_too_old"); assert_eq!(cursor_too_old.retry_after_ms, None); + assert_eq!( + raw_response(&frames, "cursor_too_old_reply")["scope"], + "own" + ); - let deleted = assert_error_shape( + let deleted = expect_error( "deleted_reply", - reply_response(&frames, "deleted_reply", "error"), + reply_to(&frames, "deleted_reply", &document_request()), ); + assert_eq!(deleted.code, "deleted"); assert_eq!(deleted.current_seq, Some(4)); - assert!(deleted.doc_id.is_some()); + assert!(raw_response(&frames, "deleted_reply")["doc_id"].is_string()); - let exists = assert_error_shape( + let exists = expect_error( "exists_reply", - reply_response(&frames, "exists_reply", "error"), + reply_to(&frames, "exists_reply", &create_upload()), ); + assert_eq!(exists.code, "exists"); assert!(exists.existing_owner.is_some()); - assert!(exists.doc_id.is_some()); - let hash_mismatch = assert_error_shape( + let hash_mismatch = expect_error( "hash_mismatch_reply", - reply_response(&frames, "hash_mismatch_reply", "error"), + reply_to(&frames, "hash_mismatch_reply", &create_upload()), ); assert!(hash_mismatch.current_hash.is_some()); assert_eq!(hash_mismatch.current_seq, Some(3)); - let join_error = assert_error_shape( - "join_error_reply", - reply_response(&frames, "join_error_reply", "error"), - ); - assert_eq!(join_error.code, "auth_invalid"); - assert!(join_error.is_fatal); - - let subscription_forbidden = assert_error_shape( + let forbidden = expect_error( "subscription_forbidden_reply", - reply_response(&frames, "subscription_forbidden_reply", "error"), + reply_to(&frames, "subscription_forbidden_reply", &snapshot_request()), ); + assert_eq!(forbidden.code, "subscription_forbidden"); assert_eq!( - subscription_forbidden.scope.as_deref(), - Some("collection:nope") + raw_response(&frames, "subscription_forbidden_reply")["scope"], + "collection:nope" ); - let too_large = assert_error_shape( + let too_large = expect_error( "too_large_reply", - reply_response(&frames, "too_large_reply", "error"), + reply_to(&frames, "too_large_reply", &create_upload()), ); - assert!(too_large.doc_id.is_some()); + assert_eq!(too_large.code, "too_large"); + assert!(raw_response(&frames, "too_large_reply")["doc_id"].is_string()); - assert_error_shape( + let validation = expect_error( "validation_reply", - reply_response(&frames, "validation_reply", "error"), + reply_to(&frames, "validation_reply", &create_upload()), ); + assert_eq!(validation.code, "validation"); +} - let socket_refusal = &frames["socket_refusal"]; - let refusal = assert_error_shape("socket_refusal", socket_refusal.clone()); +#[test] +fn join_errors_and_socket_refusal_are_fatal_server_errors() { + let frames = load_frames(); + let mut codec = joined_codec(); + let Some(Incoming::Reply { req: 1, result }) = + codec.decode(&frames["join_error_reply"].to_string()) + else { + panic!("join_error_reply must decode as a reply to the join"); + }; + let join_error = expect_error("join_error_reply", result); + assert_eq!(join_error.code, "auth_invalid"); + assert!(join_error.is_fatal); + + let refusal: ServerError = serde_json::from_value(frames["socket_refusal"].clone()) + .expect("socket_refusal must parse as ServerError"); assert_eq!(refusal.code, "update_required"); assert!(refusal.is_fatal); } #[test] -fn client_push_requests_carry_their_expected_ids() { +fn socket_url_carries_the_recorded_connect_params() { let frames = load_frames(); - - let publish = frame_payload(&frames, "publish_request", "publish"); - assert!(publish - .get("source_doc_id") - .and_then(Value::as_str) - .is_some()); - - let publish_update = frame_payload(&frames, "publish_update_request", "publish_update"); - assert!(publish_update - .get("publication_id") - .and_then(Value::as_str) - .is_some()); - - let unpublish = frame_payload(&frames, "unpublish_request", "unpublish"); - assert!(unpublish - .get("publication_id") - .and_then(Value::as_str) - .is_some()); + let url = url::Url::parse(&socket_url("ws://localhost:4000", Uuid::nil()).unwrap()).unwrap(); + let params: Map = url + .query_pairs() + .filter(|(key, _)| key != "vsn") + .map(|(key, value)| (key.into_owned(), Value::String(value.into_owned()))) + .collect(); + assert_eq!(Value::Object(params), frames["socket_connect_params"]); } -fn upload_kind(kind: &str) -> UploadKind { - match kind { - "create" => UploadKind::Create, - "update" => UploadKind::Update, - "delete" => UploadKind::Delete, - other => panic!("unknown upload kind {other}"), +#[test] +fn join_request_matches_the_recorded_frame_with_a_real_signature() { + let frames = load_frames(); + let join = encode(&mut Codec::new(), 1, &Request::Join); + let recorded = &frames["join_request"]; + for i in 0..4 { + assert_eq!(join[i], recorded[i], "join frame element {i}"); } -} - -fn upload_request(frames: &serde_json::Map, name: &str) -> (Value, Upload) { - let payload = frame_payload(frames, name, "upload"); - let upload: Upload = serde_json::from_value(payload.clone()) - .unwrap_or_else(|e| panic!("{name} must match Request::Upload: {e}")); - assert_eq!(upload.upload_id, Uuid::nil(), "{name}"); - assert_eq!(upload.doc_id, Uuid::nil(), "{name}"); - (payload, upload) + let ours = join[4].as_object().unwrap(); + let theirs = recorded[4].as_object().unwrap(); + assert_eq!( + ours.keys().collect::>(), + theirs.keys().collect::>() + ); + for key in ["email", "api_key", "timestamp"] { + assert_eq!(ours[key], theirs[key], "{key}"); + } + assert_eq!( + ours["signature"], + "c53b53ad283f08d4cd2e9261f8f65d08dc0815b8ad0ff41aa4574f293bb1d9c2" + ); } #[test] -fn socket_connect_and_join_requests_carry_the_contract_fields() { +fn feed_and_document_requests_match_the_recorded_frames() { let frames = load_frames(); - - let connect: SocketConnectParams = - serde_json::from_value(frames["socket_connect_params"].clone()) - .expect("socket_connect_params must match the connect query params"); - assert_eq!(connect.protocol_version, "2"); - assert_eq!(connect.client_id, Uuid::nil()); - - let join = frame_payload(&frames, "join_request", "phx_join"); - let join: Join = serde_json::from_value(join).expect("join_request must match Request::Join"); - assert!(join.email.contains('@')); - assert!(!join.api_key.is_empty()); - assert_eq!(join.signature.len(), 64); - assert!(join.timestamp > 0); + let mut codec = joined_codec(); + assert_eq!( + encode(&mut codec, 2, &changes_request()), + frames["changes_request"] + ); + assert_eq!( + encode(&mut codec, 2, &snapshot_request()), + frames["snapshot_request"] + ); + let next_page = Request::GetSnapshot { + scope: "own".into(), + page_token: Some(format!("3:{}", Uuid::nil())), + }; + assert_eq!( + encode(&mut codec, 2, &next_page), + frames["snapshot_page_request"] + ); + assert_eq!( + encode(&mut codec, 2, &document_request()), + frames["document_request"] + ); } #[test] -fn feed_and_document_requests_match_their_request_variants() { +fn upload_requests_match_the_recorded_frames_per_kind() { let frames = load_frames(); - - let changes: GetChangesSince = serde_json::from_value(frame_payload( - &frames, - "changes_request", - "get_changes_since", - )) - .expect("changes_request must match Request::GetChangesSince"); - assert_eq!(changes.scope, "own"); - assert!(changes.cursor >= 0 && changes.limit > 0); - - let first_page = frame_payload(&frames, "snapshot_request", "get_snapshot"); - assert!(first_page.get("page_token").is_none()); - let first_page: GetSnapshot = serde_json::from_value(first_page) - .expect("snapshot_request must match Request::GetSnapshot"); - assert_eq!(first_page.scope, "own"); - assert_eq!(first_page.page_token, None); - - let next_page: GetSnapshot = serde_json::from_value(frame_payload( - &frames, - "snapshot_page_request", - "get_snapshot", - )) - .expect("snapshot_page_request must match Request::GetSnapshot"); - let token = next_page - .page_token - .expect("snapshot_page_request must carry a page_token"); - let (seq, id) = token - .split_once(':') - .expect("page_token is :"); - assert!(seq.parse::().is_ok()); - assert!(Uuid::parse_str(id).is_ok()); - - let document: GetDocument = - serde_json::from_value(frame_payload(&frames, "document_request", "get_document")) - .expect("document_request must match Request::GetDocument"); - assert_eq!(document.doc_id, Uuid::nil()); + let mut codec = joined_codec(); + assert_eq!( + encode(&mut codec, 2, &create_upload()), + frames["upload_create_request"] + ); + let recorded_update = &frames["upload_update_request"][4]; + let update = upload( + UploadKind::Update, + recorded_update["base_hash"].as_str().map(String::from), + recorded_update["payload"].clone(), + ); + assert_eq!( + encode(&mut codec, 2, &update), + frames["upload_update_request"] + ); + assert_eq!( + encode( + &mut codec, + 2, + &upload(UploadKind::Delete, None, Value::Null) + ), + frames["upload_delete_request"] + ); } #[test] -fn upload_requests_match_request_upload_per_kind() { +fn publish_requests_carry_their_ids() { let frames = load_frames(); - - let (raw, create) = upload_request(&frames, "upload_create_request"); - assert_eq!(upload_kind(&create.kind), UploadKind::Create); - assert!(raw.get("base_hash").is_none()); - assert!(create.payload.is_object()); - - let (_raw, update) = upload_request(&frames, "upload_update_request"); - assert_eq!(upload_kind(&update.kind), UploadKind::Update); - assert!(update.base_hash.is_some()); - assert!(update.payload.is_array()); - - let (raw, delete) = upload_request(&frames, "upload_delete_request"); - assert_eq!(upload_kind(&delete.kind), UploadKind::Delete); - assert!(raw.get("base_hash").is_none()); - assert!(delete.payload.is_null()); + for (name, event, key) in [ + ("publish_request", "publish", "source_doc_id"), + ("publish_update_request", "publish_update", "publication_id"), + ("unpublish_request", "unpublish", "publication_id"), + ] { + assert_eq!(frames[name][2], "sync:v2", "{name}"); + assert_eq!(frames[name][3], event, "{name}"); + assert!(frames[name][4][key].is_string(), "{name}"); + } } /// Never compare a client-computed hash to a server hash: the server hash is authoritative. #[test] -fn float_content_deserializes_but_client_hash_is_not_the_server_hash() { +fn float_content_decodes_but_client_hash_is_not_the_server_hash() { let frames = load_frames(); - - let upload_response = reply_response(&frames, "float_upload_reply", "ok"); - let upload_doc: DocEnvelope = - serde_json::from_value(upload_response).expect("float_upload_reply must be a DocEnvelope"); + let Ok(Response::Uploaded(upload_doc)) = + reply_to(&frames, "float_upload_reply", &create_upload()) + else { + panic!("float_upload_reply must decode as an upload reply"); + }; assert_eq!(upload_doc.seq, 5); - - let document_response = reply_response(&frames, "float_document_reply", "ok"); - let document_doc: DocEnvelope = serde_json::from_value(document_response) - .expect("float_document_reply must be a DocEnvelope"); + let Ok(Response::Document(document_doc)) = + reply_to(&frames, "float_document_reply", &document_request()) + else { + panic!("float_document_reply must decode as a document reply"); + }; assert_eq!( upload_doc.hash, document_doc.hash,