From 441d979c1a0c35894111469393b039197e339b89 Mon Sep 17 00:00:00 2001 From: Amp Date: Sat, 14 Nov 2026 16:57:05 +0000 Subject: [PATCH] feat(zero-rust): report each mutation's server outcome to listeners Port zero-swift's addMutationOutcomeListener as Zero::add_mutation_outcome_listener: the listener hears each mutation's server outcome once, as a MutationServerOutcome, including that of a mutation replayed after its server result settled with a ZeroError when the client gave up connecting. The listeners belong to the Zero, so they outlive a replaced client. --- .changeset/zero-rust-mutation-outcomes.md | 5 + rust/PARITY.md | 8 + rust/zero-client/src/client/engine.rs | 27 +- rust/zero-client/src/client/mod.rs | 4 +- .../src/client/mutation_tracker.rs | 128 ++++++- rust/zero-client/src/client/zero.rs | 44 ++- rust/zero-client/tests/mutation_outcomes.rs | 319 ++++++++++++++++++ 7 files changed, 528 insertions(+), 7 deletions(-) create mode 100644 .changeset/zero-rust-mutation-outcomes.md create mode 100644 rust/zero-client/tests/mutation_outcomes.rs diff --git a/.changeset/zero-rust-mutation-outcomes.md b/.changeset/zero-rust-mutation-outcomes.md new file mode 100644 index 0000000000..98a94b4503 --- /dev/null +++ b/.changeset/zero-rust-mutation-outcomes.md @@ -0,0 +1,5 @@ +--- +'zero-rust': minor +--- + +Add `Zero::add_mutation_outcome_listener`, the Swift client's `addMutationOutcomeListener`, which JS has no counterpart for. The listener hears each mutation's server outcome once, as a `client::MutationServerOutcome` with the mutation's ID, name, arguments, and `MutationOutcome`, and the call returns the function that removes it. Unlike a mutation's `server` result, which settles with a `ZeroError` when the client gives up connecting (`disconnect_timeout_ms`), it hears the real outcome of a mutation replayed after the client reconnects, so an app can show each rejection in one place. Listeners stay with the `Zero` when it replaces its client; a mutation still pending when the app quits is replayed on the next launch, but its outcome isn't reported. diff --git a/rust/PARITY.md b/rust/PARITY.md index 26e0546bc6..4a1408fb21 100644 --- a/rust/PARITY.md +++ b/rust/PARITY.md @@ -102,11 +102,19 @@ whose window can stay hidden for long can set JS's 300,000. | `zero.mutate.issue.insert(...)` (CRUD mutators) | not ported: write a mutator | | `zero.inspector` | `inspector()`, an `Inspector`; see below | | `zero.mutateBatch`, `zero.online`, `zero.server`, `zero.version` | not ported; `protocol::PROTOCOL_VERSION` is the sync protocol's version | +| none (Swift's `addMutationOutcomeListener`) | `add_mutation_outcome_listener(listener)`, which returns the function that removes it | Rust adds `reconnect_now` (the network may be back, so skip the backoff), `when_pending_settled`, and `flush_storage` (wait for the store's queued writes, such as before the app exits). +`add_mutation_outcome_listener` is Swift's API, with no JS counterpart: the +listener hears each mutation's server outcome once, as a +`MutationServerOutcome`, including that of a mutation replayed after its +`server` result settled with a `ZeroError` when the client gave up +connecting. Swift's `mutationOutcomes` stream isn't ported: the listener can +feed a channel. + When the server no longer has the client's state, JS asks the app to reload. Without `on_client_state_not_found`, Rust replaces the client inside the same `Zero`, as Swift does: a new client ID and client group on diff --git a/rust/zero-client/src/client/engine.rs b/rust/zero-client/src/client/engine.rs index 26c3ccb217..3766a970e4 100644 --- a/rust/zero-client/src/client/engine.rs +++ b/rust/zero-client/src/client/engine.rs @@ -9,7 +9,7 @@ use super::database::{AggregateWrites, ReplicaDatabase, StorageOp, StoredMutatio use super::event_loop::{EventLoop, LoopScheduler}; use super::ivm_branch::IvmBranch; use super::log::Log; -use super::mutation_tracker::MutationTracker; +use super::mutation_tracker::{MutationTracker, OutcomeListeners}; use super::poke_handler::{MergedPoke, PokeHandler}; use super::query_context::QueryContext; use super::query_manager::QueryManager; @@ -348,6 +348,22 @@ impl MutationOutcome { } } +/// The server's outcome of a mutation this client made, reported by +/// [`Zero::add_mutation_outcome_listener`](super::Zero::add_mutation_outcome_listener). +/// Swift's `MutationServerOutcome`: the JS client has no counterpart. +#[derive(Clone, Debug, PartialEq)] +pub struct MutationServerOutcome { + /// The mutation's ID, counted per client from 1. + pub mutation_id: u64, + /// The mutator's name, such as `issue.create`. + pub name: String, + /// The mutator's arguments: `None` when it was given none. + pub args: Option, + /// `Success` or `AppError`, with the server's data or message; or + /// `ZeroError` when the server reported a mutation it couldn't process. + pub outcome: MutationOutcome, +} + /// What `mutate` returns: the outcome of the optimistic run on the client, /// and the server's. #[derive(Clone)] @@ -1598,6 +1614,12 @@ impl Engine { } } + /// Reports the server's outcome of each mutation this engine makes to + /// `listeners`, which `Zero` keeps (`onMutationOutcome`). + pub(crate) fn set_mutation_outcome_listeners(&self, listeners: Rc) { + self.mutation_tracker.set_outcome_listeners(listeners); + } + fn settle_pending_waiters(&self) { if self.pending_waiters.borrow().is_empty() || !self.pending_settled() { return; @@ -2688,7 +2710,7 @@ impl Engine { } /// What a mutation's caller hears for an error (`wrapErrorFor`). - fn outcome(error: &Error) -> MutationOutcome { + pub(crate) fn outcome(error: &Error) -> MutationOutcome { if let Error::Connection(e) = error && e.is_zero_error() { @@ -2881,6 +2903,7 @@ impl Engine { timestamp: self.scheduler.monotonic_now(), writes: write.writes.clone(), }; + self.mutation_tracker.mutation_ran(mutation_id, name, args); let stored = StoredMutation { client_id: self.client_id.clone(), id: mutation_id, diff --git a/rust/zero-client/src/client/mod.rs b/rust/zero-client/src/client/mod.rs index 00e207a486..18136dee52 100644 --- a/rust/zero-client/src/client/mod.rs +++ b/rust/zero-client/src/client/mod.rs @@ -30,8 +30,8 @@ mod zero; pub use connection_manager::{ConnectionState, ConnectionStatus, NeedsAuthReason}; pub use engine::{ - BatchViewUpdates, MutationOutcome, MutatorResult, StorageFailureHandler, UpdateNeededHandler, - UpdateNeededReason, ZeroOptions, + BatchViewUpdates, MutationOutcome, MutationServerOutcome, MutatorResult, StorageFailureHandler, + UpdateNeededHandler, UpdateNeededReason, ZeroOptions, }; pub use error::*; pub use inspector::{ diff --git a/rust/zero-client/src/client/mutation_tracker.rs b/rust/zero-client/src/client/mutation_tracker.rs index 84e5a0d2f6..438f09bcc7 100644 --- a/rust/zero-client/src/client/mutation_tracker.rs +++ b/rust/zero-client/src/client/mutation_tracker.rs @@ -1,6 +1,7 @@ //! Tracks the mutations waiting for the server's verdict and settles their //! `server` results (`packages/zero-client/src/client/mutation-tracker.ts`). +use super::engine::{Engine, MutationOutcome, MutationServerOutcome}; use super::replica_store::{KvDiff, KvDiffOp}; use super::{ApplicationError, ClientErrorKind, ConnectionError, Deferred, Error}; use crate::decode::JsonCodable; @@ -9,11 +10,69 @@ use crate::protocol::{ ErrorBody, ErrorKind, ErrorOrigin, ErrorReason, FailureCause, MutationError, MutationId, MutationResult, PushError, PushResponseBody, }; -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; use std::collections::{BTreeMap, HashMap}; +use std::rc::Rc; pub(crate) type ServerResult = Deferred>; +type OutcomeListener = Rc; + +/// The listeners of `Zero::add_mutation_outcome_listener`. The client +/// keeps them, and gives them to each engine it makes, so they outlive a +/// replaced one. +#[derive(Default)] +pub(crate) struct OutcomeListeners { + listeners: RefCell>, + next_id: Cell, +} + +impl OutcomeListeners { + pub(crate) fn add(&self, listener: OutcomeListener) -> u64 { + let id = self.next_id.get(); + self.next_id.set(id + 1); + self.listeners.borrow_mut().push((id, listener)); + id + } + + pub(crate) fn remove(&self, id: u64) { + self.listeners.borrow_mut().retain(|(i, _)| *i != id); + } + + fn notify(&self, outcome: &MutationServerOutcome) { + // A listener may add or remove one. + let listeners: Vec<_> = self + .listeners + .borrow() + .iter() + .map(|(_, l)| l.clone()) + .collect(); + for l in listeners { + l(outcome); + } + } +} + +/// A mutation taken off the unreported ones, to report to `listeners` +/// once its `server` result has settled. +struct Report { + listeners: Rc, + mutation_id: u64, + name: String, + args: Option, +} + +impl Report { + fn send(self, outcome: MutationOutcome) { + self.listeners.notify(&MutationServerOutcome { + mutation_id: self.mutation_id, + name: self.name, + args: self.args, + outcome, + }); + } +} + struct Entry { mutation_id: Option, server: ServerResult, @@ -30,11 +89,18 @@ struct State { next_ephemeral_id: u64, largest_outstanding_mutation_id: f64, current_mutation_id: f64, + /// The mutations sent to the server whose outcome hasn't been + /// reported: their names and arguments, by mutation ID. Unlike + /// `outstanding`, they stay when the connection gives up on them: + /// they're sent again on reconnect, and their outcome is reported then. + unreported: BTreeMap)>, } pub(crate) struct MutationTracker { client_id: String, state: RefCell, + /// Hear each mutation's outcome from the server once. + outcome_listeners: RefCell>>, ack_mutations: Box, on_fatal_error: Box, on_all_mutations_applied: Box, @@ -54,6 +120,7 @@ impl MutationTracker { MutationTracker { client_id, state: RefCell::default(), + outcome_listeners: RefCell::new(None), ack_mutations, on_fatal_error, on_all_mutations_applied, @@ -94,6 +161,39 @@ impl MutationTracker { s.largest_outstanding_mutation_id = s.largest_outstanding_mutation_id.max(mutation_id); } + /// Mutation `mutation_id` ran on the client and is stored: its outcome + /// is reported once the server gives one. Swift records it when the ID + /// is assigned and drops it when the mutator throws (`rejectMutation`); + /// recording it once the mutator ran is the same, and moves `args` here + /// instead of copying them. + pub(crate) fn mutation_ran(&self, mutation_id: f64, name: &str, args: Option) { + self.state + .borrow_mut() + .unreported + .insert(mid_key(mutation_id), (name.to_owned(), args)); + } + + /// The listeners to report outcomes to, from now on. + pub(crate) fn set_outcome_listeners(&self, listeners: Rc) { + *self.outcome_listeners.borrow_mut() = Some(listeners); + } + + /// Takes mutation `mid` off the unreported ones: its report, when it + /// wasn't reported yet and someone listens. + fn take_report(&self, mid: u64) -> Option { + let (name, args) = self.state.borrow_mut().unreported.remove(&mid)?; + let listeners = self.outcome_listeners.borrow().clone()?; + if listeners.listeners.borrow().is_empty() { + return None; + } + Some(Report { + listeners, + mutation_id: mid, + name, + args, + }) + } + /// Rejects a mutation the client could not run. It was not stored. pub(crate) fn reject_mutation(&self, ephemeral_id: u64, error: &Error) { let app = ApplicationError::wrapping(error); @@ -207,6 +307,22 @@ impl MutationTracker { return Ok(()); } s.current_mutation_id = last_mutation_id; + drop(s); + // Applied without a result of their own: they succeeded. + let lmid = mid_key(last_mutation_id); + loop { + let next = { + let s = self.state.borrow(); + match s.unreported.first_key_value() { + Some((&mid, _)) if mid <= lmid => mid, + _ => break, + } + }; + if let Some(report) = self.take_report(next) { + report.send(MutationOutcome::Success { data: None }); + } + } + let mut s = self.state.borrow_mut(); let mut settled = Vec::new(); let mut head = 0; while head < s.order.len() { @@ -269,9 +385,15 @@ impl MutationTracker { }; Error::Connection(invalid_push(message)) }; + let report = self + .take_report(mid_key(mid)) + .map(|r| (r, Engine::outcome(&failure))); if let Some(ephemeral_id) = self.ephemeral_id(mid) { self.settle(ephemeral_id, Err(failure)); } + if let Some((report, outcome)) = report { + report.send(outcome); + } if kind == "oooMutation" { (self.on_fatal_error)(invalid_push( "Server reported an out-of-order mutation".to_owned(), @@ -292,9 +414,13 @@ impl MutationTracker { "received mutation for the wrong client", )); } + let report = self.take_report(mid_key(mid)).map(|r| (r, data.clone())); if let Some(ephemeral_id) = self.ephemeral_id(mid) { self.settle(ephemeral_id, Ok(data)); } + if let Some((report, data)) = report { + report.send(MutationOutcome::Success { data }); + } Ok(()) } diff --git a/rust/zero-client/src/client/zero.rs b/rust/zero-client/src/client/zero.rs index 149d8ceff7..13b6fee73f 100644 --- a/rust/zero-client/src/client/zero.rs +++ b/rust/zero-client/src/client/zero.rs @@ -3,8 +3,11 @@ use super::connection_manager::ConnectionState; use super::database::{ReplicaDatabase, replicache_name}; -use super::engine::{Engine, MutatorFn, MutatorResult, UpdateNeededReason, ZeroOptions}; +use super::engine::{ + Engine, MutationServerOutcome, MutatorFn, MutatorResult, UpdateNeededReason, ZeroOptions, +}; use super::inspector::{Inspector, InspectorHost}; +use super::mutation_tracker::OutcomeListeners; use super::query_context::{MaterializedView, Preload, RunType, WeakView}; use super::storage::DroppedDatabases; use super::transaction::Transaction; @@ -107,6 +110,8 @@ struct Core { preloads: RefCell>>, /// What the client's inspectors share, once there is one. inspector: RefCell>>, + /// The mutation outcome listeners, which every engine reports to. + outcome_listeners: Rc, } /// A preload, which a replacement starts again on the new engine until @@ -175,6 +180,7 @@ impl Zero { views: RefCell::default(), preloads: RefCell::default(), inspector: RefCell::default(), + outcome_listeners: Rc::default(), }); Core::attach(&core, &engine); Ok(Zero { core, context }) @@ -321,6 +327,38 @@ impl Zero { self.mutate(&request.name, request.args) } + /// Calls `listener` with the server's outcome of each mutation this + /// client makes, once per mutation, as the server reports it. Returns + /// the function that removes it. + /// + /// Unlike a mutation's `server` result, it hears the outcome of a + /// mutation sent again after the connection came back: `server` settles + /// with [`MutationOutcome::ZeroError`](super::MutationOutcome::ZeroError) + /// when the client gives up connecting + /// ([`disconnect_timeout_ms`](ZeroOptions::disconnect_timeout_ms)), and + /// the mutation is replayed when it reconnects. A mutation still pending + /// when the app quits is replayed when it next launches, but its outcome + /// isn't reported, nor is a pending mutation's when the client replaces + /// itself. The listeners stay with the client when it does. + /// + /// Use it for a single place that shows rejections, instead of each + /// `server` result, so a rejection isn't shown twice. To read the + /// outcomes elsewhere, such as from an async task, send them to a + /// channel from the listener. Swift's `addMutationOutcomeListener`: the + /// JS client has no counterpart. + pub fn add_mutation_outcome_listener( + &self, + listener: impl Fn(&MutationServerOutcome) + 'static, + ) -> Box { + let id = self.core.outcome_listeners.add(Rc::new(listener)); + let weak = Rc::downgrade(&self.core.outcome_listeners); + Box::new(move || { + if let Some(listeners) = weak.upgrade() { + listeners.remove(id); + } + }) + } + /// The app was shown, or hidden. Hidden for /// `ZeroOptions::hidden_disconnect_delay_ms`, the client disconnects /// until it is shown again. @@ -380,8 +418,10 @@ impl Core { } /// Hooks `engine` up to the client: the connection hears it needs an - /// update, and the client replaces it when it lost its state. + /// update, the mutation outcome listeners hear its mutations' outcomes, + /// and the client replaces it when it lost its state. fn attach(core: &Rc, engine: &Rc) { + engine.set_mutation_outcome_listeners(core.outcome_listeners.clone()); let connection = Rc::downgrade(&core.connection.0); *engine.on_update_needed.borrow_mut() = Some(Rc::new(move |reason| { if let Some(c) = connection.upgrade() { diff --git a/rust/zero-client/tests/mutation_outcomes.rs b/rust/zero-client/tests/mutation_outcomes.rs new file mode 100644 index 0000000000..05e9befdb9 --- /dev/null +++ b/rust/zero-client/tests/mutation_outcomes.rs @@ -0,0 +1,319 @@ +//! `Zero::add_mutation_outcome_listener`, which zero-swift has and JS +//! doesn't: each mutation's server outcome is heard once, including that of +//! a mutation replayed after its `server` result gave up on it. + +use std::cell::RefCell; +use std::rc::Rc; +use std::sync::Arc; +use zero_client::client::{ + MutationOutcome, MutationServerOutcome, Mutator, MutatorResult, Transaction, Zero, ZeroOptions, +}; +use zero_client::json::{self, Object, Value}; +use zero_client::schema::Schema; +use zero_client::testing::{ManualScheduler, MockSocket, MockSocketFactory, RecordingLogSink}; + +struct Harness { + zero: Zero, + scheduler: Rc, + factory: Rc, + heard: Rc>>, +} + +impl Harness { + fn new(disconnect_timeout_ms: f64) -> Self { + let schema = Schema::from_json( + &json::parse( + r#"{"tables": {"issue": {"name": "issue", "primaryKey": ["id"], + "columns": {"id": {"type": "string"}, "title": {"type": "string"}}}}}"#, + ) + .unwrap(), + ) + .unwrap(); + let create = Mutator::new("issue.create", |tx: &Transaction, args, _: &()| { + let args = args.unwrap(); + let mut row = Object::new(); + row.insert("id", args.get("id").unwrap().clone()); + row.insert("title", args.get("title").unwrap().clone()); + tx.table("issue").insert(&row)?; + Ok(None) + }); + let scheduler = ManualScheduler::new(1_700_000_000_000.0); + let factory = MockSocketFactory::new(); + let options = ZeroOptions { + cache_url: Some("http://zero.test".into()), + user_id: Some("u1".into()), + disconnect_timeout_ms, + log_sink: Some(Rc::new(RecordingLogSink::default())), + ..ZeroOptions::default() + }; + let zero = Zero::new( + Arc::new(schema), + (), + vec![create], + options, + scheduler.clone(), + Some(factory.clone()), + ) + .unwrap(); + let heard = Rc::new(RefCell::new(Vec::new())); + let h = Harness { + zero, + scheduler, + factory, + heard, + }; + h.settle(); + h + } + + /// Adds a listener that records what it hears in `heard`. + fn listen(&self, heard: &Rc>>) -> Box { + let heard = heard.clone(); + self.zero + .add_mutation_outcome_listener(move |o| heard.borrow_mut().push(o.clone())) + } + + fn settle(&self) { + self.zero.dispatch(|| {}); + self.scheduler.run_due_timers(); + } + + fn socket(&self) -> Rc { + self.factory.sockets().last().cloned().expect("a socket") + } + + fn connect(&self) -> Rc { + let socket = self.socket(); + socket.receive_text(r#"["connected",{"wsid":"w1"}]"#); + self.settle(); + socket + } + + fn create(&self, id: &str, title: &str) -> MutatorResult { + let r = self.zero.mutate("issue.create", Some(issue(id, title))); + self.settle(); + r + } + + /// A poke from `base` to `cookie` with `part`'s fields. + fn poke(&self, socket: &MockSocket, base: &str, cookie: &str, part: &str) { + socket.receive_text(&format!( + r#"["pokeStart",{{"pokeID":"{cookie}","baseCookie":{base}}}]"# + )); + socket.receive_text(&format!(r#"["pokePart",{{"pokeID":"{cookie}",{part}}}]"#)); + socket.receive_text(&format!( + r#"["pokeEnd",{{"pokeID":"{cookie}","cookie":"{cookie}"}}]"# + )); + self.settle(); + } + + /// The IDs of the mutations `socket` pushed, in order. + fn pushed(&self, socket: &MockSocket) -> Vec { + socket + .messages() + .iter() + .filter(|m| m.starts_with(r#"["push""#)) + .flat_map(|m| { + let v = json::parse(m).unwrap(); + let ms = v.at(1).unwrap().get("mutations").unwrap().clone(); + let Value::Array(ms) = ms else { panic!() }; + ms.iter() + .map(|m| m.get("id").unwrap().as_f64().unwrap()) + .collect::>() + }) + .collect() + } +} + +fn issue(id: &str, title: &str) -> Value { + json::parse(&format!(r#"{{"id":"{id}","title":"{title}"}}"#)).unwrap() +} + +fn heard( + mutation_id: u64, + id: &str, + title: &str, + outcome: MutationOutcome, +) -> MutationServerOutcome { + MutationServerOutcome { + mutation_id, + name: "issue.create".into(), + args: Some(issue(id, title)), + outcome, + } +} + +fn server(r: &MutatorResult) -> MutationOutcome { + r.server.result().expect("settled").unwrap() +} + +fn app_error(message: &str, details: Option<&str>) -> MutationOutcome { + MutationOutcome::AppError { + message: message.into(), + details: details.map(|d| json::parse(d).unwrap()), + } +} + +fn success(data: Option<&str>) -> MutationOutcome { + MutationOutcome::Success { + data: data.map(|d| json::parse(d).unwrap()), + } +} + +/// A push response reports one mutation's data and another's app error, +/// and the poke that confirms all three reports the third's success. The +/// first two aren't heard again when the poke confirms them too. +#[test] +fn hears_each_server_outcome_once() { + let h = Harness::new(60_000.0); + let _remove = h.listen(&h.heard); + let socket = h.connect(); + let cid = h.zero.client_id(); + let one = h.create("i1", "one"); + let two = h.create("i2", "two"); + let three = h.create("i3", "three"); + assert_eq!(h.pushed(&socket), [1.0, 2.0, 3.0]); + assert!(h.heard.borrow().is_empty()); + + socket.receive_text(&format!( + r#"["pushResponse",{{"mutations":[ + {{"id":{{"clientID":"{cid}","id":1}},"result":{{"data":{{"n":1}}}}}}, + {{"id":{{"clientID":"{cid}","id":2}},"result":{{"error":"app","message":"taken","details":{{"code":7}}}}}} + ]}}]"# + )); + h.settle(); + h.poke( + &socket, + "null", + "01", + &format!(r#""lastMutationIDChanges":{{"{cid}":3}}"#), + ); + + assert_eq!(server(&one), success(Some(r#"{"n":1}"#))); + assert_eq!(server(&two), app_error("taken", Some(r#"{"code":7}"#))); + assert_eq!(server(&three), success(None)); + assert_eq!( + *h.heard.borrow(), + [ + heard(1, "i1", "one", success(Some(r#"{"n":1}"#))), + heard(2, "i2", "two", app_error("taken", Some(r#"{"code":7}"#))), + heard(3, "i3", "three", success(None)), + ] + ); +} + +/// The client gives up on the server, so `server` settles with a +/// `ZeroError`; it reconnects, sends the two mutations again, and the +/// server rejects one and applies the other. The listener hears those, not +/// the `ZeroError`s. +#[test] +fn hears_the_outcome_of_a_mutation_replayed_after_the_disconnect_timeout() { + let h = Harness::new(5_000.0); + let _remove = h.listen(&h.heard); + let first = h.connect(); + let cid = h.zero.client_id(); + let one = h.create("i1", "one"); + let two = h.create("i2", "two"); + assert_eq!(h.pushed(&first), [1.0, 2.0]); + + first.server_close(1006, "", false); + h.settle(); + for _ in 0..10 { + h.scheduler.advance(1_000.0); + h.settle(); + } + assert!(matches!(server(&one), MutationOutcome::ZeroError { .. })); + assert!(matches!(server(&two), MutationOutcome::ZeroError { .. })); + assert!(h.heard.borrow().is_empty()); + + let sockets = h.factory.sockets().len(); + for _ in 0..600 { + if h.factory.sockets().len() > sockets { + break; + } + h.scheduler.advance(100.0); + h.settle(); + } + let second = h.connect(); + assert!(!Rc::ptr_eq(&first, &second), "the client reconnected"); + assert_eq!(h.pushed(&second), [1.0, 2.0], "the mutations are replayed"); + assert!(h.heard.borrow().is_empty()); + + h.poke( + &second, + "null", + "01", + &format!( + r#""lastMutationIDChanges":{{"{cid}":2}},"mutationsPatch":[{{"op":"put","mutation":{{"id":{{"clientID":"{cid}","id":1}},"result":{{"error":"app","message":"taken"}}}}}}]"# + ), + ); + assert_eq!( + *h.heard.borrow(), + [ + heard(1, "i1", "one", app_error("taken", None)), + heard(2, "i2", "two", success(None)), + ] + ); +} + +/// A removed listener hears nothing more; the others still do. +#[test] +fn a_removed_listener_hears_nothing() { + let h = Harness::new(60_000.0); + let removed = Rc::new(RefCell::new(Vec::new())); + let remove = h.listen(&removed); + let _keep = h.listen(&h.heard); + let socket = h.connect(); + let cid = h.zero.client_id(); + h.create("i1", "one"); + h.poke( + &socket, + "null", + "01", + &format!(r#""lastMutationIDChanges":{{"{cid}":1}}"#), + ); + assert_eq!(*removed.borrow(), [heard(1, "i1", "one", success(None))]); + + remove(); + remove(); + h.create("i2", "two"); + h.poke( + &socket, + r#""01""#, + "02", + &format!(r#""lastMutationIDChanges":{{"{cid}":2}}"#), + ); + assert_eq!(*removed.borrow(), [heard(1, "i1", "one", success(None))]); + assert_eq!( + *h.heard.borrow(), + [ + heard(1, "i1", "one", success(None)), + heard(2, "i2", "two", success(None)), + ] + ); +} + +/// The listeners stay with the client when it replaces itself, after the +/// server lost its state, and hear the new client's mutations. +#[test] +fn listeners_outlive_a_replaced_client() { + let h = Harness::new(60_000.0); + let _remove = h.listen(&h.heard); + let socket = h.connect(); + let old = h.zero.client_id(); + socket.receive_text(r#"["error",{"kind":"ClientNotFound","message":"gone"}]"#); + h.settle(); + let cid = h.zero.client_id(); + assert_ne!(cid, old, "the client replaced itself"); + + let socket = h.connect(); + h.create("i1", "one"); + assert_eq!(h.pushed(&socket), [1.0]); + h.poke( + &socket, + "null", + "01", + &format!(r#""lastMutationIDChanges":{{"{cid}":1}}"#), + ); + assert_eq!(*h.heard.borrow(), [heard(1, "i1", "one", success(None))]); +}