feat(zero-rust): mutation-outcome listener, after Swift's

.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.

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

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<Value>,
+    /// `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<OutcomeListeners>) {
+        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,

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::{

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<Option<Value>>;
 
+type OutcomeListener = Rc<dyn Fn(&MutationServerOutcome)>;
+
+/// 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<Vec<(u64, OutcomeListener)>>,
+    next_id: Cell<u64>,
+}
+
+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<OutcomeListeners>,
+    mutation_id: u64,
+    name: String,
+    args: Option<Value>,
+}
+
+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<f64>,
     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<u64, (String, Option<Value>)>,
 }
 
 pub(crate) struct MutationTracker {
     client_id: String,
     state: RefCell<State>,
+    /// Hear each mutation's outcome from the server once.
+    outcome_listeners: RefCell<Option<Rc<OutcomeListeners>>>,
     ack_mutations: Box<dyn Fn(MutationId)>,
     on_fatal_error: Box<dyn Fn(ConnectionError)>,
     on_all_mutations_applied: Box<dyn Fn()>,
@@ -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<Value>) {
+        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<OutcomeListeners>) {
+        *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<Report> {
+        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(())
     }
 

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<Vec<Rc<PreloadRecord>>>,
     /// What the client's inspectors share, once there is one.
     inspector: RefCell<Option<Rc<InspectorHost>>>,
+    /// The mutation outcome listeners, which every engine reports to.
+    outcome_listeners: Rc<OutcomeListeners>,
 }
 
 /// A preload, which a replacement starts again on the new engine until
@@ -175,6 +180,7 @@ impl<C: 'static> Zero<C> {
             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<C: 'static> Zero<C> {
         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<dyn Fn()> {
+        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<Core>, engine: &Rc<Engine>) {
+        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() {