From 0d70a61b504c0a5b832c4427165c244ade5bef56 Mon Sep 17 00:00:00 2001 From: Amp Date: Sat, 14 Nov 2026 17:14:17 +0000 Subject: [PATCH 1/2] feat(zero-rust): add LiveView, a typed live query Zero::live(&query, &ttl) takes a typed Query or a named query's QueryRequest (through the new typed::TypedQuery trait) and returns a LiveView, the Rust counterpart of the Swift client's LiveQuery: the query's value read as its Rust type and kept current, with its result type, an error (the server's query error, or rows that didn't read, which keep the last value), listeners, update_ttl, and the underlying MaterializedView. It follows the client when the client replaces itself, and dropping it removes its listeners and destroys the view. An update reads only the rows whose entries changed, as Swift's RowCache does (QueryValue::update_from_view, with a default that reads anew), and keeps a copy of the entries rather than sharing the view's, so the view changes its rows in place. In turns, an optimistic update with a view of 1,000, 10,000, and 100,000 rows read as structs went from 82.8, 759, and 13,130 us (reading every row in a listener) to 19.1, 61.6, and 521 us. --- .changeset/zero-rust-live-view.md | 5 + rust/BENCHMARKS.md | 21 +- rust/PARITY.md | 17 +- rust/benchmarks/src/main.rs | 53 ++- rust/zero-client/src/client/live_view.rs | 225 ++++++++++++ rust/zero-client/src/client/mod.rs | 2 + rust/zero-client/src/client/zero.rs | 15 + rust/zero-client/src/define.rs | 6 +- rust/zero-client/src/typed.rs | 171 ++++++++++ rust/zero-client/tests/live_view.rs | 416 +++++++++++++++++++++++ 10 files changed, 919 insertions(+), 12 deletions(-) create mode 100644 .changeset/zero-rust-live-view.md create mode 100644 rust/zero-client/src/client/live_view.rs create mode 100644 rust/zero-client/tests/live_view.rs diff --git a/.changeset/zero-rust-live-view.md b/.changeset/zero-rust-live-view.md new file mode 100644 index 0000000000..3689d1052e --- /dev/null +++ b/.changeset/zero-rust-live-view.md @@ -0,0 +1,5 @@ +--- +'zero-rust': minor +--- + +Add `LiveView`, a typed live query after the Swift client's `LiveQuery` (zero-react's `useQuery`). `Zero::live(&query, &ttl)` takes a typed `Query` or a named query's `QueryRequest` and returns a `LiveView` whose `value()` is the query's rows, `one()` row, or aggregate read as its Rust type and kept current, with `result_type()`, `error()` (a `LiveError`: the server's query error, or rows that didn't read, which keep the last value), `add_listener`, `update_ttl`, and `view()` for the `MaterializedView`. An update reads only the rows whose entries changed, so an optimistic update with a live view of 100,000 rows read as structs no longer reads every row again. A live view follows the client when it replaces itself, and dropping it removes its listeners and destroys its view. `QueryValue` gains `update_from_view`, which has a default, and the new `typed::TypedQuery` trait names what a query reads as. diff --git a/rust/BENCHMARKS.md b/rust/BENCHMARKS.md index 99429d2faf..6c88c14b1b 100644 --- a/rust/BENCHMARKS.md +++ b/rust/BENCHMARKS.md @@ -26,11 +26,12 @@ The JS sides: | `memory`, `hydration`, `mutations` | `pnpm --filter zero-conformance bench` (`src/bench/client.bench.ts`) | | `pokes`, `first sync`, the round trip | none: the JS client applies pokes on the next frame, so a JS timing measures that | -Two of Swift's aren't ported: `btree-set.bench.ts` (the memory sources' +One of Swift's isn't ported: `btree-set.bench.ts` (the memory sources' B+ tree is internal to the crate; the benchmarks time it through views and -mutations) and the optimistic update with a live query of structs (there is -no typed live query; an app reads a view's rows as structs, which -`hydration` times). +mutations). The optimistic update with a live query of structs times a +`LiveView`, and `mutations` also times the same update with the view's rows +read as structs by hand on every change, as an app did before `LiveView`; +`hydration` times making a `LiveView` next to materializing as structs. The benchmarks don't run in CI; numbers from shared runners vary too much to gate on. CI builds them with the workspace (`cargo clippy --all-targets`, @@ -167,6 +168,18 @@ timed from the poke's frames arriving to the view being flushed. Materializing and memory didn't move (20.4 and 20.5 ms for 100,000 rows; 82.8 and 81.5 MB). +- **A view read as structs read every row on every change.** An app + kept a view's rows as structs by reading them all in its listener, so + an optimistic update with a view of 100,000 rows cost 13.1 ms where the + view itself took 18 µs. A `LiveView` reads only the rows whose entries + changed, as Swift's `LiveQuery` does, and lets the view change its rows + in place rather than copy them. In turns (three runs, `--time 1`), the + update with a view of 1,000, 10,000, and 100,000 rows read as structs + went from 82.8, 759, and 13,130 µs to 19.1, 61.6, and 521 µs; it still + compares every entry, about 5 ns each. Making a live view costs what + materializing and reading the rows did (116 µs and 119 µs at 1,000 + rows, 1.46 and 1.45 ms at 10,000). + Where Rust is still behind: - **The smallest changes.** Putting a row in its table's layout compares diff --git a/rust/PARITY.md b/rust/PARITY.md index 26e0546bc6..9b79396a96 100644 --- a/rust/PARITY.md +++ b/rust/PARITY.md @@ -194,11 +194,18 @@ registered throws in JS; in Rust both outcomes are a `ZeroError` saying so, and ### React and SolidJS -No UI bindings: `useQuery` is a `MaterializedView` and its listener. -`apps/zbugs-rune` shows how a Rune app binds views to its UI, on the -shared zbugs model (`apps/zbugs-model`), whose queries and mutators are -`define_query` and `define_mutator` definitions; `apps/zbugs-gpui` is a -GPUI app on the same model. +No UI bindings. `useQuery`'s typed equivalent is `LiveView`, after the +Swift client's `LiveQuery`: `Zero::live` takes a typed `Query` or a named +query's `QueryRequest` and keeps its value read as the query's Rust type, +reading only the rows whose entries changed, with its result type and an +error that is the server's query error or a row that didn't read (which +keeps the last value). It follows the client when the client replaces +itself, and has listeners a UI wakes itself from; the untyped equivalent is +a `MaterializedView` and its listener. `apps/zbugs-rune` shows how a Rune +app binds live views to its UI, on the shared zbugs model +(`apps/zbugs-model`), whose queries and mutators are `define_query` and +`define_mutator` definitions; `apps/zbugs-gpui` is a GPUI app on the same +model. ## How some vectors are compared diff --git a/rust/benchmarks/src/main.rs b/rust/benchmarks/src/main.rs index f8b7e316db..939b0203be 100644 --- a/rust/benchmarks/src/main.rs +++ b/rust/benchmarks/src/main.rs @@ -16,13 +16,15 @@ mod harness; use fixtures::{Client, sum, update_user_args, user_json, users}; use harness::{Bench, check, resident_memory}; -use std::cell::Cell; +use std::cell::{Cell, RefCell}; use std::hint::black_box; +use std::rc::Rc; use zero_client::ast::SimpleOperator; use zero_client::hash::xxhash32; use zero_client::ivm::{ViewEntry, ViewValue}; use zero_client::json::Value; -use zero_client::typed::{QueryElement, QueryValue, ZeroValueError, column}; +use zero_client::ttl::Ttl; +use zero_client::typed::{Query, QueryElement, QueryValue, ZeroValueError, column}; fn main() { let mut filters = Vec::new(); @@ -144,6 +146,7 @@ fn hydration(bench: &mut Bench) { for n in [1_000, 10_000, 100_000] { if !bench.wants(&format!("materialize {n} rows")) && !bench.wants(&format!("materialize {n} rows as structs")) + && !bench.wants(&format!("live view of {n} rows as structs")) { continue; } @@ -161,6 +164,13 @@ fn hydration(bench: &mut Bench) { black_box(rows.iter().map(User::total).sum::()); view.destroy(); }); + let typed = Query::::from_any(all.clone()); + bench.run(&format!("live view of {n} rows as structs"), || { + let live = client.zero.live(&typed, &Ttl::default()).unwrap(); + let rows = live.value(); + check(rows.len() == n, "every row"); + black_box(rows.iter().map(User::total).sum::()); + }); } }); } @@ -284,6 +294,45 @@ fn mutations(bench: &mut Bench) { }); check_updated(&local_view, before, "updated"); + // The same with the rows read as structs on every update, as an + // app reads them: by hand, and with a live view, which reads only + // the rows that changed. + let all = Query::::from_any(users(local.zero.schema())); + let read = Client::new(n, false); + let read_view = read.materialize(all.any()); + let rows = Rc::new(RefCell::new(Vec::new())); + let into = rows.clone(); + let _remove = read_view.add_listener(move |data, _, _| { + *into.borrow_mut() = Vec::::from_view(data).unwrap(); + }); + let before = value.get(); + bench.run( + &format!("optimistic update, {n}-row view read as structs"), + || { + value.set(value.get() + 1); + let args = update_user_args(value.get() % n, value.get()); + let _ = read.zero.mutate("updateUser", Some(args)); + }, + ); + let updated = |rows: &[User]| rows[value.get() % n].b == value.get() as f64; + if value.get() > before { + check(updated(&rows.borrow()), "read"); + } + let structs = Client::new(n, false); + let live = structs.zero.live(&all, &Ttl::default()).unwrap(); + let before = value.get(); + bench.run( + &format!("optimistic update, {n}-row live query of structs"), + || { + value.set(value.get() + 1); + let args = update_user_args(value.get() % n, value.get()); + let _ = structs.zero.mutate("updateUser", Some(args)); + }, + ); + if value.get() > before { + check(updated(&live.value()), "read"); + } + // The client's side of a round trip: the optimistic update, the // push, and the poke that confirms it, which rebases the pending // mutations. diff --git a/rust/zero-client/src/client/live_view.rs b/rust/zero-client/src/client/live_view.rs new file mode 100644 index 0000000000..cab6740359 --- /dev/null +++ b/rust/zero-client/src/client/live_view.rs @@ -0,0 +1,225 @@ +//! A view read as its query's Rust type and kept current (`LiveQuery` in +//! the Swift client, `useQuery` in zero-react). + +use super::query_context::MaterializedView; +use super::{Error, QueryError}; +use crate::ivm::{ResultType, ViewEntry, ViewValue}; +use crate::protocol::ErroredQuery; +use crate::ttl::Ttl; +use crate::typed::{QueryValue, ZeroValueError}; +use std::cell::{Cell, RefCell}; +use std::fmt; +use std::rc::{Rc, Weak}; + +/// Why a [`LiveView`]'s value may not be what the server has. +#[derive(Clone, Debug, PartialEq)] +pub enum LiveError { + /// The server could not run the query. + Query(ErroredQuery), + /// The rows could not be read as the query's result type. The value is + /// the last one read. + Decode(ZeroValueError), +} + +impl fmt::Display for LiveError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + LiveError::Query(e) => QueryError::new(Some(e.clone())).fmt(f), + LiveError::Decode(e) => e.fmt(f), + } + } +} + +impl std::error::Error for LiveError {} + +type Listener = Rc; + +/// What a [`LiveView`] last read. +struct State { + value: Rc, + /// The entries `value` was read from. + read: ViewValue, + /// The view's rows when they were read, which it shares until they + /// change. Unlike holding them, which would make the view copy them to + /// change them, it lets them change in place, into another allocation. + seen: Weak>, + result_type: ResultType, + error: Option, +} + +fn rows_of(data: &ViewValue) -> Weak> { + match data { + ViewValue::Many(rows) => Rc::downgrade(rows), + _ => Weak::new(), + } +} + +struct Inner { + view: MaterializedView, + state: RefCell>, + listeners: RefCell)>>, + next_listener_id: Cell, +} + +/// A query's rows, the row of a `one()` query, or the value of an +/// aggregate, read as its Rust type and kept current, from +/// [`Zero::live`](super::Zero::live) (`LiveQuery` in the Swift client, +/// `useQuery` in zero-react). +/// +/// Each update of the view is read into the value, reading only the rows +/// whose entries changed. When the client replaces itself, the view moves +/// to the new client, and the live view with it. Dropping it removes its +/// listeners and destroys the view; the server keeps syncing the query for +/// its TTL. +pub struct LiveView { + inner: Rc>, + remove: Box, +} + +impl LiveView { + /// Reads `view`, and follows it. Fails, destroying it, when its rows + /// don't read as `T`. + pub(crate) fn new(view: MaterializedView) -> Result { + let read = view.data(); + let value = match T::from_view(&read) { + Ok(value) => value, + Err(e) => { + view.destroy(); + return Err(Error::Message(e.0)); + } + }; + let inner = Rc::new(Inner { + state: RefCell::new(State { + value: Rc::new(value), + seen: rows_of(&read), + read, + result_type: view.result_type(), + error: view.error().map(LiveError::Query), + }), + view, + listeners: RefCell::default(), + next_listener_id: Cell::new(0), + }); + let weak: Weak> = Rc::downgrade(&inner); + let remove = inner.view.add_listener(move |data, result_type, error| { + if let Some(inner) = weak.upgrade() { + inner.update(data, result_type, error); + } + }); + Ok(LiveView { inner, remove }) + } +} + +impl LiveView { + /// The value last read. Holding it past the next update makes that + /// update read every row again rather than only those that changed. + pub fn value(&self) -> Rc { + self.inner.state.borrow().value.clone() + } + + /// Whether the server has confirmed the rows on this connection + /// (`Complete`) or an earlier one (`Cached`), or not yet (`Unknown`), or + /// the query failed (`Error`). + pub fn result_type(&self) -> ResultType { + self.inner.state.borrow().result_type + } + + /// Why the server could not run the query, or why the rows could not be + /// read as `T`, which keeps the value last read. + pub fn error(&self) -> Option { + self.inner.state.borrow().error.clone() + } + + /// The view the value is read from. + pub fn view(&self) -> &MaterializedView { + &self.inner.view + } + + /// Raises the query's TTL; a lower one is ignored. + pub fn update_ttl(&self, ttl: &Ttl) { + self.inner.view.update_ttl(ttl); + } + + /// Calls `listener` with the value and result type now, and after every + /// update of the view, once the value is read. After a read that + /// failed, it gets the value kept, and [`error`](LiveView::error) says + /// why. Returns the function that removes it. + pub fn add_listener(&self, listener: impl Fn(&T, ResultType) + 'static) -> Box { + let id = self.inner.next_listener_id.get(); + self.inner.next_listener_id.set(id + 1); + let listener: Listener = Rc::new(listener); + self.inner + .listeners + .borrow_mut() + .push((id, listener.clone())); + let (value, result_type) = { + let s = self.inner.state.borrow(); + (s.value.clone(), s.result_type) + }; + listener(&value, result_type); + let weak = Rc::downgrade(&self.inner); + Box::new(move || { + if let Some(inner) = weak.upgrade() { + inner.listeners.borrow_mut().retain(|(i, _)| *i != id); + } + }) + } +} + +impl Inner { + /// Reads the view's update, and tells the listeners. + fn update(&self, data: &ViewValue, result_type: ResultType, error: Option<&ErroredQuery>) { + let value = { + let mut state = self.state.borrow_mut(); + let s = &mut *state; + s.result_type = result_type; + let unchanged = match data { + ViewValue::Many(rows) => std::ptr::eq(s.seen.as_ptr(), Rc::as_ptr(rows)), + data => s.read == *data, + }; + let read = if unchanged { + Ok(()) + } else { + let read = match Rc::get_mut(&mut s.value) { + Some(value) => value.update_from_view(&mut s.read, data), + // A value the app still holds is left as it is. + None => T::from_view(data).map(|value| { + s.value = Rc::new(value); + s.read = data.clone(); + }), + }; + if read.is_ok() { + s.seen = rows_of(data); + } + read + }; + s.error = match read { + Ok(()) => error.cloned().map(LiveError::Query), + Err(e) => Some(LiveError::Decode(e)), + }; + s.value.clone() + }; + // Each listener in turn, reading the list again after every call, + // as `MaterializedView` does. + let mut last: Option = None; + loop { + let next = self + .listeners + .borrow() + .iter() + .find(|(id, _)| last.is_none_or(|l| *id > l)) + .cloned(); + let Some((id, listener)) = next else { break }; + last = Some(id); + listener(&value, result_type); + } + } +} + +impl Drop for LiveView { + fn drop(&mut self) { + (self.remove)(); + self.inner.listeners.borrow_mut().clear(); + self.inner.view.destroy(); + } +} diff --git a/rust/zero-client/src/client/mod.rs b/rust/zero-client/src/client/mod.rs index 00e207a486..072d782c8e 100644 --- a/rust/zero-client/src/client/mod.rs +++ b/rust/zero-client/src/client/mod.rs @@ -12,6 +12,7 @@ mod error; mod event_loop; mod inspector; mod ivm_branch; +mod live_view; mod log; mod measure_push; mod mutation_tracker; @@ -38,6 +39,7 @@ pub use inspector::{ AnalyzeQueryOptions, AnalyzeQueryResult, Inspector, InspectorClient, InspectorClientGroup, InspectorQuery, Metrics, PlanDebugEvent, RowCountsBySource, RowsByQuery, RowsBySource, }; +pub use live_view::{LiveError, LiveView}; pub use log::{LogLevel, LogSink, StderrLogSink}; pub use query_context::{MaterializedView, Preload, RunType, ViewListener}; pub use scheduler::*; diff --git a/rust/zero-client/src/client/zero.rs b/rust/zero-client/src/client/zero.rs index 149d8ceff7..9866e3ae44 100644 --- a/rust/zero-client/src/client/zero.rs +++ b/rust/zero-client/src/client/zero.rs @@ -5,6 +5,7 @@ use super::connection_manager::ConnectionState; use super::database::{ReplicaDatabase, replicache_name}; use super::engine::{Engine, MutatorFn, MutatorResult, UpdateNeededReason, ZeroOptions}; use super::inspector::{Inspector, InspectorHost}; +use super::live_view::LiveView; use super::query_context::{MaterializedView, Preload, RunType, WeakView}; use super::storage::DroppedDatabases; use super::transaction::Transaction; @@ -14,6 +15,7 @@ use crate::json::Value; use crate::query::{AnyQuery, ZeroQuery}; use crate::schema::Schema; use crate::ttl::Ttl; +use crate::typed::TypedQuery; use std::cell::{Cell, RefCell}; use std::collections::HashMap; use std::rc::Rc; @@ -236,6 +238,19 @@ impl Zero { Ok(view) } + /// The query's result read as its Rust type and kept current: its rows, + /// the row of a `one()` query, or the value of an aggregate, from a + /// typed [`Query`](crate::typed::Query) or a named query's + /// `QueryRequest`. Fails when the query can't be materialized, or when + /// its rows don't read as the type (an [`Error::Message`] saying why). + pub fn live(&self, query: &Q, ttl: &Ttl) -> Result, Error> + where + Q: TypedQuery + ?Sized, + Q::Value: 'static, + { + LiveView::new(self.materialize(query, ttl)?) + } + /// The query's rows: what the replica holds now (`Unknown`), or once /// the server has confirmed them on this connection (`Complete`) or /// any (`Cached`). Await the result, or read it once settled. diff --git a/rust/zero-client/src/define.rs b/rust/zero-client/src/define.rs index 0c94bac87a..b10fdcb2e1 100644 --- a/rust/zero-client/src/define.rs +++ b/rust/zero-client/src/define.rs @@ -39,7 +39,7 @@ use crate::client::{ApplicationError, Error, MutateRequest, Mutator, Transaction use crate::ivm::ViewValue; use crate::json::{self, Value}; use crate::query::{AnyQuery, ZeroQuery}; -use crate::typed::{Query, QueryValue, ZeroValueError}; +use crate::typed::{Query, QueryValue, TypedQuery, ZeroValueError}; use serde::Serialize; use serde::de::DeserializeOwned; use std::borrow::Cow; @@ -160,6 +160,10 @@ impl ZeroQuery for QueryRequest { } } +impl TypedQuery for QueryRequest { + type Value = T; +} + /// A named query whose arguments come as JSON, as the server gets them, /// whatever its types: for a list of an app's queries, such as a test that /// checks them against the TS definitions. diff --git a/rust/zero-client/src/typed.rs b/rust/zero-client/src/typed.rs index e23a6054b4..99a5031bd6 100644 --- a/rust/zero-client/src/typed.rs +++ b/rust/zero-client/src/typed.rs @@ -36,6 +36,7 @@ use crate::client::Error; use crate::query::{ExpressionBuilder, JsonPathReference}; use std::fmt; use std::marker::PhantomData; +use std::rc::Rc; use std::sync::Arc; /// A value that doesn't fit the Rust type a row declares for it. @@ -396,12 +397,170 @@ impl ColumnUpdate { pub trait QueryValue: Sized { /// Reads the query's result from its view. fn from_view(view: &ViewValue) -> Result; + + /// Reads `view` into `self`, which was read from `previous`, as + /// [`from_view`](QueryValue::from_view) would, keeping what it can of + /// `self`, and makes `previous` what was read: a + /// [`LiveView`](crate::client::LiveView) reads each update so. On an + /// error both are left as they were. By default it reads `view` anew; + /// rows (`Vec`) read only the entries that changed. + fn update_from_view( + &mut self, + previous: &mut ViewValue, + view: &ViewValue, + ) -> Result<(), ZeroValueError> { + *self = Self::from_view(view)?; + *previous = view.clone(); + Ok(()) + } } impl QueryValue for Vec { fn from_view(view: &ViewValue) -> Result { view.entries().into_iter().map(E::from_entry).collect() } + + /// Keeps the rows whose entries are equal, and reads the others (the + /// Swift client's `RowCache`). Most updates change a few rows in place, + /// or insert or remove one, so the entries are matched in order, + /// skipping at most one. `previous` becomes a copy of the entries + /// changed in the same way, so the view, which then shares its rows with + /// nothing, changes them in place. + fn update_from_view( + &mut self, + previous: &mut ViewValue, + view: &ViewValue, + ) -> Result<(), ZeroValueError> { + let (ViewValue::Many(old), ViewValue::Many(new)) = (&mut *previous, view) else { + *self = Self::from_view(view)?; + *previous = view.clone(); + return Ok(()); + }; + if old.len() != self.len() { + *self = Self::from_view(view)?; + *previous = view.clone(); + return Ok(()); + } + // What the rows become, in order: runs of old rows kept, by index, + // and of rows read anew. Rows are read before anything changes, so a + // row that fails leaves both as they were. + let mut runs: Vec = Vec::new(); + let mut fresh = Vec::new(); + // Whether every row kept stays at its index, so the rows read anew + // replace the others where they are. + let mut in_place = old.len() == new.len(); + let mut j = 0; + for (i, entry) in new.iter().enumerate() { + let kept = if old.get(j).is_some_and(|o| same_entry(o, entry)) { + Some(j) + } else if old.get(j + 1).is_some_and(|o| same_entry(o, entry)) { + Some(j + 1) + } else { + None + }; + match (kept, runs.last_mut()) { + (Some(k), Some(Run::Keep { from, len })) if *from + *len == k => *len += 1, + (Some(k), _) => runs.push(Run::Keep { from: k, len: 1 }), + (None, Some(Run::Fresh(len))) => *len += 1, + (None, _) => runs.push(Run::Fresh(1)), + } + match kept { + Some(k) => { + in_place &= k == i; + j = k + 1; + } + None => fresh.push(E::from_entry(entry)?), + } + } + if in_place && fresh.is_empty() { + return Ok(()); + } + let mut fresh = fresh.into_iter(); + apply_runs(self, &runs, in_place, new.len(), |_| { + fresh.next().expect("a row read for each new entry") + }); + apply_runs(Rc::make_mut(old), &runs, in_place, new.len(), |i| { + new[i].clone() + }); + Ok(()) + } +} + +/// Whether two entries are equal, as `==` says, comparing the rows and the +/// lists they share by address first: an entry a view didn't change shares +/// them with the copy read before. +fn same_entry(a: &ViewEntry, b: &ViewEntry) -> bool { + (Rc::ptr_eq(&a.row, &b.row) || a.row == b.row) + && a.relationships.len() == b.relationships.len() + && a.relationships + .iter() + .zip(&b.relationships) + .all(|((m, a), (n, b))| m == n && same_value(a, b)) +} + +fn same_value(a: &ViewValue, b: &ViewValue) -> bool { + match (a, b) { + (ViewValue::Many(a), ViewValue::Many(b)) => { + Rc::ptr_eq(a, b) + || (a.len() == b.len() && a.iter().zip(b.iter()).all(|(a, b)| same_entry(a, b))) + } + (ViewValue::One(Some(a)), ViewValue::One(Some(b))) => same_entry(a, b), + (a, b) => a == b, + } +} + +/// A run of `Vec::update_from_view`'s rows. +enum Run { + /// `len` old rows from `from`. + Keep { from: usize, len: usize }, + /// `len` rows read anew. + Fresh(usize), +} + +/// Makes `items` the `len` items `runs` lists: the old ones kept, and +/// `fresh(i)` for the `i`th, read anew. `in_place`: each item kept stays +/// where it is. +fn apply_runs( + items: &mut Vec, + runs: &[Run], + in_place: bool, + len: usize, + mut fresh: impl FnMut(usize) -> X, +) { + if in_place { + let mut at = 0; + for run in runs { + match *run { + Run::Keep { len, .. } => at += len, + Run::Fresh(n) => { + for (i, item) in items[at..at + n].iter_mut().enumerate() { + *item = fresh(at + i); + } + at += n; + } + } + } + return; + } + let mut out = Vec::with_capacity(len); + let mut old = std::mem::take(items).into_iter(); + let mut next = 0; + for run in runs { + match *run { + Run::Keep { from, len } => { + if from > next { + old.nth(from - next - 1); + } + out.extend(old.by_ref().take(len)); + next = from + len; + } + Run::Fresh(n) => { + let at = out.len(); + out.extend((at..at + n).map(&mut fresh)); + } + } + } + *items = out; } /// What an `Option` query value holds: the row of `one()`, or the value of @@ -1031,6 +1190,18 @@ impl Query { } } +/// A query whose view reads as a [`QueryValue`]: a typed [`Query`], or a +/// named query's `QueryRequest`. What +/// [`Zero::live`](crate::client::Zero::live) takes. +pub trait TypedQuery: crate::query::ZeroQuery { + /// What the query's view reads as. + type Value: QueryValue; +} + +impl TypedQuery for Query { + type Value = T; +} + fn names(columns: Vec>) -> Vec { columns.into_iter().map(|c| c.name.to_owned()).collect() } diff --git a/rust/zero-client/tests/live_view.rs b/rust/zero-client/tests/live_view.rs new file mode 100644 index 0000000000..5c8aba1c27 --- /dev/null +++ b/rust/zero-client/tests/live_view.rs @@ -0,0 +1,416 @@ +//! `Zero::live`: a view read as row structs and kept current, against a +//! mock server. + +use std::cell::{Cell, RefCell}; +use std::rc::Rc; +use std::sync::Arc; +use zero_client::ast::System; +use zero_client::client::{LiveError, LiveView, Zero, ZeroOptions}; +use zero_client::hash::query::hash_of_name_and_args; +use zero_client::ivm::{ResultType, ViewEntry, ViewValue}; +use zero_client::json; +use zero_client::query::AnyQuery; +use zero_client::schema::Schema; +use zero_client::testing::{ManualScheduler, MockSocket, MockSocketFactory, RecordingLogSink}; +use zero_client::ttl::Ttl; +use zero_client::typed::{ + OptionalQueryElement, Query, QueryElement, ZeroValueError, column, optional_row, +}; + +#[derive(Debug, PartialEq)] +struct Issue { + id: String, + title: String, +} + +thread_local! { + /// How many issues were read from entries. + static READS: Cell = const { Cell::new(0) }; +} + +fn reads() -> usize { + READS.with(Cell::get) +} + +impl QueryElement for Issue { + fn from_entry(entry: &ViewEntry) -> Result { + READS.with(|r| r.set(r.get() + 1)); + let title: String = column(entry, "title", 1)?; + // A title a newer server might send, which this app can't read. + if title == "unreadable" { + return Err(ZeroValueError("an unreadable title".into())); + } + Ok(Issue { + id: column(entry, "id", 0)?, + title, + }) + } +} + +impl OptionalQueryElement for Issue { + fn optional_from_view(view: &ViewValue) -> Result, ZeroValueError> { + optional_row(view) + } +} + +fn issue(id: &str, title: &str) -> String { + format!(r#"{{"op":"put","tableName":"issue","value":{{"id":"{id}","title":"{title}"}}}}"#) +} + +fn del(id: &str) -> String { + format!(r#"{{"op":"del","tableName":"issue","id":{{"id":"{id}"}}}}"#) +} + +struct Harness { + zero: Zero, + scheduler: Rc, + factory: Rc, + schema: Arc, + cookie: Cell, +} + +impl Harness { + fn new() -> Self { + let schema = Arc::new( + Schema::from_json( + &json::parse( + r#"{"tables": {"issue": {"name": "issue", "primaryKey": ["id"], + "columns": {"id": {"type": "string"}, "title": {"type": "string"}}}}}"#, + ) + .unwrap(), + ) + .unwrap(), + ); + 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()), + log_sink: Some(Rc::new(RecordingLogSink::default())), + ..ZeroOptions::default() + }; + let zero = Zero::new( + schema.clone(), + (), + vec![], + options, + scheduler.clone(), + Some(factory.clone()), + ) + .unwrap(); + let h = Harness { + zero, + scheduler, + factory, + schema, + cookie: Cell::new(0), + }; + h.connect(); + h + } + + 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) { + self.settle(); + self.socket().receive_text(r#"["connected",{"wsid":"w1"}]"#); + self.settle(); + } + + /// Every issue, by `id`. + fn issues(&self) -> Query { + Query::from_any( + AnyQuery::new(self.schema.clone(), "issue", System::Client) + .order_by("id", zero_client::ast::Direction::Asc), + ) + } + + /// Every issue, as the named query `issues`. + fn named_issues(&self) -> Query { + Query::from_any(self.issues().into_any().name_and_args("issues", vec![])) + } + + fn live(&self, query: &Query) -> LiveView> { + self.zero.live(query, &Ttl::default()).unwrap() + } + + /// A poke from the server: `rows` (row patch ops), and the queries it + /// has now got. + fn poke(&self, rows: &[String], got: &[&str]) { + let base = match self.cookie.get() { + 0 => "null".to_owned(), + c => format!("\"{c:04}\""), + }; + let cookie = self.cookie.get() + 1; + self.cookie.set(cookie); + let got: Vec = got + .iter() + .map(|h| format!(r#"{{"op":"put","hash":"{h}"}}"#)) + .collect(); + let socket = self.socket(); + socket.receive_text(&format!( + r#"["pokeStart",{{"pokeID":"p{cookie}","baseCookie":{base}}}]"# + )); + socket.receive_text(&format!( + r#"["pokePart",{{"pokeID":"p{cookie}","rowsPatch":[{}],"gotQueriesPatch":[{}]}}]"#, + rows.join(","), + got.join(",") + )); + socket.receive_text(&format!( + r#"["pokeEnd",{{"pokeID":"p{cookie}","cookie":"{cookie:04}"}}]"# + )); + self.settle(); + } +} + +fn titles(live: &LiveView>) -> Vec { + live.value().iter().map(|i| i.title.clone()).collect() +} + +#[test] +fn the_value_follows_pokes() { + let h = Harness::new(); + let live = h.live(&h.issues()); + assert!(live.value().is_empty()); + assert_eq!(live.result_type(), ResultType::Unknown); + h.poke(&[issue("a", "Apple"), issue("b", "Banana")], &[]); + assert_eq!( + *live.value(), + vec![ + Issue { + id: "a".into(), + title: "Apple".into() + }, + Issue { + id: "b".into(), + title: "Banana".into() + }, + ] + ); + assert_eq!(live.error(), None); +} + +#[test] +fn a_listener_hears_the_value_now_and_on_every_update() { + let h = Harness::new(); + let live = h.live(&h.issues()); + let heard: Rc>>> = Rc::default(); + let h2 = heard.clone(); + let remove = live.add_listener(move |issues, _| { + h2.borrow_mut() + .push(issues.iter().map(|i| i.title.clone()).collect()); + }); + h.poke(&[issue("a", "Apple")], &[]); + h.poke(&[issue("a", "Apricot")], &[]); + assert_eq!( + *heard.borrow(), + vec![vec![], vec!["Apple".to_owned()], vec!["Apricot".to_owned()]] + ); + remove(); + h.poke(&[issue("b", "Banana")], &[]); + assert_eq!(heard.borrow().len(), 3, "removed"); + assert_eq!(titles(&live), ["Apricot", "Banana"]); +} + +#[test] +fn the_result_type_goes_complete_once_the_server_has_the_query() { + let h = Harness::new(); + let live = h.live(&h.named_issues()); + let types: Rc>> = Rc::default(); + let t = types.clone(); + let _remove = live.add_listener(move |_, ty| t.borrow_mut().push(ty)); + assert_eq!(live.result_type(), ResultType::Unknown); + h.poke( + &[issue("a", "Apple")], + &[&hash_of_name_and_args("issues", &[])], + ); + assert_eq!(live.result_type(), ResultType::Complete); + assert_eq!(titles(&live), ["Apple"]); + assert_eq!(types.borrow().last(), Some(&ResultType::Complete)); +} + +#[test] +fn a_query_the_server_could_not_run_is_an_error() { + let h = Harness::new(); + let live = h.live(&h.named_issues()); + let hash = hash_of_name_and_args("issues", &[]); + h.socket().receive_text(&format!( + r#"["transformError",[{{"error":"app","id":"{hash}","name":"issues","message":"boom"}}]]"# + )); + h.settle(); + assert_eq!(live.result_type(), ResultType::Error); + let error = live.error().expect("an error"); + assert!(matches!(error, LiveError::Query(_)), "{error:?}"); + assert_eq!(error.to_string(), "boom"); +} + +#[test] +fn rows_that_do_not_read_keep_the_last_value_and_set_the_error() { + let h = Harness::new(); + let live = h.live(&h.issues()); + let heard = Rc::new(Cell::new(0)); + let n = heard.clone(); + let _remove = live.add_listener(move |_, _| n.set(n.get() + 1)); + h.poke(&[issue("a", "Apple")], &[]); + h.poke(&[issue("b", "unreadable")], &[]); + assert_eq!(titles(&live), ["Apple"], "the last value read"); + assert_eq!( + live.error(), + Some(LiveError::Decode(ZeroValueError( + "an unreadable title".into() + ))) + ); + assert_eq!(heard.get(), 3, "the listener heard the failed read"); + // Rows that read again clear it. + h.poke(&[issue("b", "Banana")], &[]); + assert_eq!(titles(&live), ["Apple", "Banana"]); + assert_eq!(live.error(), None); +} + +#[test] +fn rows_that_do_not_read_at_first_fail_live() { + let h = Harness::new(); + h.poke(&[issue("a", "unreadable")], &[]); + let error = h.zero.live(&h.issues(), &Ttl::default()).err(); + assert_eq!( + error.map(|e| e.message()), + Some("an unreadable title".to_owned()) + ); +} + +#[test] +fn only_the_rows_that_changed_are_read_again() { + let h = Harness::new(); + h.poke( + &["a", "b", "c", "d", "e"].map(|id| issue(id, &id.to_uppercase())), + &[], + ); + let live = h.live(&h.issues()); + assert_eq!(titles(&live), ["A", "B", "C", "D", "E"]); + let mut before = reads(); + + let mut read = |live: &LiveView>, titles_now: &[&str]| { + assert_eq!(titles(live), titles_now); + let n = reads() - before; + before = reads(); + n + }; + h.poke(&[issue("c", "C2")], &[]); + assert_eq!(read(&live, &["A", "B", "C2", "D", "E"]), 1, "a change"); + h.poke(&[issue("bb", "BB")], &[]); + assert_eq!( + read(&live, &["A", "B", "BB", "C2", "D", "E"]), + 1, + "an insert" + ); + h.poke(&[del("a")], &[]); + assert_eq!(read(&live, &["B", "BB", "C2", "D", "E"]), 0, "a removal"); + h.poke(&[del("c"), issue("f", "F")], &[]); + assert_eq!(read(&live, &["B", "BB", "D", "E", "F"]), 1, "both"); + h.poke(&[issue("b", "B2"), issue("e", "E2")], &[]); + assert_eq!(read(&live, &["B2", "BB", "D", "E2", "F"]), 2, "two changes"); + + // A value the app still holds is left alone: the update reads the rows + // anew. + let held = live.value(); + h.poke(&[issue("d", "D2")], &[]); + assert_eq!(read(&live, &["B2", "BB", "D2", "E2", "F"]), 5, "held"); + assert_eq!(held[2].title, "D"); +} + +#[test] +fn the_view_moves_to_the_client_that_replaces_this_one() { + let h = Harness::new(); + let live = h.live(&h.issues()); + let heard: Rc>>> = Rc::default(); + let h2 = heard.clone(); + let _remove = live.add_listener(move |issues, _| { + h2.borrow_mut() + .push(issues.iter().map(|i| i.title.clone()).collect()); + }); + h.poke(&[issue("a", "Apple")], &[]); + let first = h.zero.client_id(); + let sockets = h.factory.sockets().len(); + + // The server lost the client's state: the client replaces itself. + h.socket().receive_text( + r#"["error",{"kind":"ClientNotFound","message":"gone","origin":"zeroCache"}]"#, + ); + h.settle(); + assert_ne!(h.zero.client_id(), first, "replaced"); + assert!(h.factory.sockets().len() > sockets, "a new connection"); + h.cookie.set(0); + h.connect(); + h.poke(&[issue("b", "Banana")], &[]); + assert_eq!(titles(&live), ["Banana"]); + // The new client's view, of its own client group's rows, and then its + // first poke. + assert_eq!( + *heard.borrow(), + [ + vec![], + vec!["Apple".to_owned()], + vec![], + vec!["Banana".to_owned()] + ] + ); + assert!(!live.view().is_destroyed()); +} + +#[test] +fn dropping_it_removes_its_listeners_and_destroys_the_view() { + let h = Harness::new(); + let live = h.live(&h.issues()); + let heard = Rc::new(Cell::new(0)); + let n = heard.clone(); + let _remove = live.add_listener(move |_, _| n.set(n.get() + 1)); + let view = live.view().clone(); + assert_eq!(Rc::strong_count(&heard), 2); + drop(live); + assert_eq!(Rc::strong_count(&heard), 1, "the listener is gone"); + assert!(view.is_destroyed()); + h.poke(&[issue("a", "Apple")], &[]); + assert_eq!(heard.get(), 1); +} + +#[test] +fn a_one_query_reads_its_row() { + let h = Harness::new(); + let first: Query> = Query::from_any(h.issues().into_any().one()); + let first = h.zero.live(&first, &Ttl::default()).unwrap(); + assert_eq!(*first.value(), None); + h.poke(&[issue("a", "Apple"), issue("b", "Banana")], &[]); + assert_eq!( + first.value().as_ref().as_ref().map(|i| i.id.as_str()), + Some("a") + ); +} + +#[cfg(feature = "serde")] +#[test] +fn a_named_query_request_reads_as_its_rows() { + use zero_client::define::{QueryDefinition, define_query}; + thread_local! { + static SCHEMA: RefCell>> = const { RefCell::new(None) }; + } + const ISSUES: QueryDefinition<(), (), Issue> = define_query("issues", |_, _| { + let schema = SCHEMA.with(|s| s.borrow().clone()).unwrap(); + Query::from_any(AnyQuery::new(schema, "issue", System::Client)) + }); + let h = Harness::new(); + SCHEMA.with(|s| *s.borrow_mut() = Some(h.schema.clone())); + let live = h.zero.live(&ISSUES.with(()), &Ttl::default()).unwrap(); + h.poke( + &[issue("a", "Apple")], + &[&hash_of_name_and_args("issues", &[])], + ); + assert_eq!(titles(&live), ["Apple"]); + assert_eq!(live.result_type(), ResultType::Complete); +} From 7145c8e75465e6d411fcaaa6b9165f51d96588d0 Mon Sep 17 00:00:00 2001 From: Amp Date: Sat, 14 Nov 2026 17:14:17 +0000 Subject: [PATCH 2/2] feat(zero-rust): move the zbugs GPUI and Rune apps onto LiveView zbugs-gpui's Live is now a GPUI adapter over a LiveView that only wakes the view to notify it; its Snapshot and its own reading of the rows are gone, and the app reads live.value(). zbugs-rune's watch makes a LiveView per client on the sync thread and ships its value to the UI, in place of materializing and reading the rows itself; a row that doesn't read is still printed and not applied. --- apps/zbugs-gpui/src/app.rs | 102 ++++++++++++++------------------- apps/zbugs-gpui/src/app/e2e.rs | 24 +++----- apps/zbugs-gpui/src/live.rs | 82 ++++++++++---------------- apps/zbugs-rune/src/sync.rs | 87 +++++++++++++++++----------- 4 files changed, 134 insertions(+), 161 deletions(-) diff --git a/apps/zbugs-gpui/src/app.rs b/apps/zbugs-gpui/src/app.rs index aeed8672b6..7d6511188c 100644 --- a/apps/zbugs-gpui/src/app.rs +++ b/apps/zbugs-gpui/src/app.rs @@ -729,10 +729,8 @@ impl ZbugsApp { } let exists = self .labels - .snapshot() - .value + .value() .iter() - .flatten() .any(|l| l.name.eq_ignore_ascii_case(&name)); if exists { self.error = Some(format!("There's already a label called {name}.")); @@ -784,7 +782,7 @@ impl ZbugsApp { let login = self .me .as_ref() - .and_then(|me| me.snapshot().value.clone().flatten()) + .and_then(|me| (*me.value()).clone()) .map(|u| u.login); let logged_in = self.session.auth.is_some(); let app = cx.entity().downgrade(); @@ -835,8 +833,7 @@ impl ZbugsApp { /// just edited) doesn't drag the list. At the top, the list stays at /// the top. fn keep_scroll_anchor(&mut self) { - let snapshot = self.list.snapshot(); - let rows = snapshot.value.as_deref().unwrap_or_default(); + let rows = self.list.value(); let handle = self.scroll.0.borrow().base_handle.clone(); let mut offset = handle.offset(); if offset.y < px(0.) { @@ -992,7 +989,7 @@ impl ZbugsApp { /// Near the end of what's loaded: ask for the next page. The rows /// loaded so far stay where they are, so the list doesn't move. fn grow(&mut self, cx: &mut Context) { - let loaded = self.list.snapshot().value.as_ref().map_or(0, Vec::len) as i64; + let loaded = self.list.value().len() as i64; if loaded < self.limit { return; } @@ -1042,12 +1039,7 @@ impl ZbugsApp { if !self.select_first { return; } - let first = self - .list - .snapshot() - .value - .as_ref() - .and_then(|rows| Some(rows.first()?.id.clone())); + let first = self.list.value().first().map(|row| row.id.clone()); let Some(id) = first else { return; }; @@ -1060,8 +1052,7 @@ impl ZbugsApp { /// Selects the issue `delta` rows from the selected one, or the first. fn move_selection(&mut self, delta: isize, cx: &mut Context) { let next = { - let snapshot = self.list.snapshot(); - let rows = snapshot.value.as_deref().unwrap_or_default(); + let rows = self.list.value(); if rows.is_empty() { return; } @@ -1265,11 +1256,12 @@ impl ZbugsApp { /// Checks or unchecks the selected issue, for space on the list. fn toggle_selected_checked(&mut self, cx: &mut Context) { - let issue = self.list.snapshot().value.as_ref().and_then(|rows| { - rows.iter() - .find(|r| Some(&r.id) == self.selected.as_ref()) - .cloned() - }); + let issue = self + .list + .value() + .iter() + .find(|r| Some(&r.id) == self.selected.as_ref()) + .cloned(); if let Some(issue) = issue { self.toggle_checked(&issue, cx); } @@ -1331,8 +1323,8 @@ impl ZbugsApp { let Some(detail) = self.detail.as_ref().filter(|d| d.is_synced()) else { return; }; - let snapshot = detail.snapshot(); - let Some(issue) = snapshot.value.as_ref().and_then(Option::as_ref) else { + let issue = detail.value(); + let Some(issue) = issue.as_ref() else { return; }; let comments = issue.comments.as_deref().unwrap_or_default(); @@ -1374,10 +1366,7 @@ impl ZbugsApp { /// "Zero Bugs → " and the shown issue's title, or the list's, as /// zbugs' issue and list pages title themselves. fn window_title(&self) -> String { - let issue = self - .detail - .as_ref() - .and_then(|d| d.snapshot().value.clone().flatten()); + let issue = self.detail.as_ref().and_then(|d| (*d.value()).clone()); let f = &self.filters; let shown = match issue { Some(issue) => issue.title, @@ -1407,8 +1396,8 @@ impl ZbugsApp { .as_ref() .and(self.detail.as_ref()) .and_then(|d| { - let snapshot = d.snapshot(); - let issue = snapshot.value.as_ref()?.as_ref()?; + let issue = d.value(); + let issue = issue.as_ref().as_ref()?; is_unread(issue).then(|| issue.id.clone()) }); let Some(id) = unread else { @@ -1442,7 +1431,7 @@ impl ZbugsApp { let modified = this .detail .as_ref() - .and_then(|d| d.snapshot().value.clone().flatten()) + .and_then(|d| (*d.value()).clone()) .filter(|issue| issue.id == id) .map(|issue| issue.modified); if let Some(modified) = modified { @@ -1497,7 +1486,7 @@ impl ZbugsApp { let pref = self .skin_tone .as_ref() - .and_then(|live| live.snapshot().value.clone().flatten()); + .and_then(|live| (*live.value()).clone()); pref.and_then(|p| p.value.parse().ok()) .filter(|tone| *tone < SKIN_TONES.len() as u32) .unwrap_or(0) @@ -1618,10 +1607,8 @@ impl ZbugsApp { fn project_id(&self) -> Option { let name = self.filters.project.to_lowercase(); self.projects - .snapshot() - .value + .value() .iter() - .flatten() .find(|p| p.lower_case_name == name) .map(|p| p.id.clone()) } @@ -1801,7 +1788,7 @@ impl ZbugsApp { fn render_sidebar(&self, lights: Pixels, cx: &mut Context) -> impl IntoElement { let f = &self.filters; let current = f.project.to_lowercase(); - let mut projects = self.projects.snapshot().value.clone().unwrap_or_default(); + let mut projects = (*self.projects.value()).clone(); projects.sort_by(|a, b| a.lower_case_name.cmp(&b.lower_case_name)); let project = self.picker( Button::new("project") @@ -1889,7 +1876,7 @@ impl ZbugsApp { let login = self .me .as_ref() - .and_then(|me| me.snapshot().value.clone().flatten()) + .and_then(|me| (*me.value()).clone()) .map(|u| u.login); let mine = |id: &'static str, on: bool, assigned: bool, cx: &mut Context| { let login = login.clone(); @@ -1937,10 +1924,8 @@ impl ZbugsApp { ); let mut names = self .labels - .snapshot() - .value + .value() .iter() - .flatten() .map(|l| l.name.clone()) .collect::>(); names.sort_by_key(|n| n.to_lowercase()); @@ -2035,11 +2020,13 @@ impl ZbugsApp { /// and log in or out. fn render_list_toolbar(&self, lights: Pixels, cx: &mut Context) -> impl IntoElement { let count = { - let s = self.count.snapshot(); - match (s.value, self.list.snapshot().value.as_ref()) { - (Some(n), _) if self.count.is_synced() || n > 0 => format!("{n}"), - (_, Some(rows)) if !rows.is_empty() => format!("{}+", rows.len()), - _ => String::new(), + let (n, rows) = (*self.count.value(), self.list.value()); + if self.count.is_synced() || n > 0 { + format!("{n}") + } else if !rows.is_empty() { + format!("{}+", rows.len()) + } else { + String::new() } }; let logged_in = self.session.auth.is_some(); @@ -2167,7 +2154,7 @@ impl ZbugsApp { value: None, checked: current.is_none(), }]; - let mut users = users.snapshot().value.clone().unwrap_or_default(); + let mut users = (*users.value()).clone(); users.sort_by_key(|u| u.login.to_lowercase()); choices.extend(users.into_iter().map(|u| Choice { avatar: Some(avatar_url(&u)), @@ -2250,7 +2237,7 @@ impl ZbugsApp { } fn render_list(&self, cx: &mut Context) -> impl IntoElement { - let rows = self.list.snapshot().value.clone().unwrap_or_default(); + let rows = (*self.list.value()).clone(); let len = rows.len(); let rows = Rc::new(rows); if len == 0 { @@ -2624,7 +2611,7 @@ impl ZbugsApp { let Some(detail) = &self.detail else { return empty("Select an issue", cx); }; - let issue = detail.snapshot().value.clone().flatten(); + let issue = (*detail.value()).clone(); let Some(issue) = issue else { return empty( if detail.is_synced() { @@ -2638,10 +2625,7 @@ impl ZbugsApp { // The newest comments first; shown oldest first. The page takes // over once it has at least what the issue brought. let mut comments: Vec = issue.comments.clone().unwrap_or_default(); - if let Some(page) = self - .comment_page - .as_ref() - .and_then(|p| p.snapshot().value.clone()) + if let Some(page) = self.comment_page.as_ref().map(|p| (*p.value()).clone()) && page.len() >= comments.len() { comments = page; @@ -2658,7 +2642,7 @@ impl ZbugsApp { value: None, checked: assignee_id.is_none(), }]; - let mut crew = self.crew.snapshot().value.clone().unwrap_or_default(); + let mut crew = (*self.crew.value()).clone(); crew.sort_by_key(|u| u.login.to_lowercase()); assignee_choices.extend(crew.into_iter().map(|u| Choice { avatar: Some(avatar_url(&u)), @@ -2681,7 +2665,7 @@ impl ZbugsApp { Rc::new(move |this, user_id, _, cx| this.set_assignee(id.clone(), user_id, cx)), cx, ); - let mut all_labels = self.labels.snapshot().value.clone().unwrap_or_default(); + let mut all_labels = (*self.labels.value()).clone(); all_labels.sort_by_key(|l| l.name.to_lowercase()); let mut label_choices: Vec = all_labels .into_iter() @@ -2991,10 +2975,11 @@ impl Render for ZbugsApp { window.push_notification(Notification::error(message), cx); } if self.reveal_selected { - let at = self.list.snapshot().value.as_ref().and_then(|rows| { - rows.iter() - .position(|r| Some(&r.id) == self.selected.as_ref()) - }); + let at = self + .list + .value() + .iter() + .position(|r| Some(&r.id) == self.selected.as_ref()); if let Some(ix) = at { self.scroll.scroll_to_item(ix, ScrollStrategy::Center); self.reveal_selected = false; @@ -3013,10 +2998,7 @@ impl Render for ZbugsApp { } else { px(TRAFFIC_LIGHTS) }; - let issue = self - .detail - .as_ref() - .and_then(|d| d.snapshot().value.clone().flatten()); + let issue = self.detail.as_ref().and_then(|d| (*d.value()).clone()); h_flex() .size_full() .bg(cx.theme().background) diff --git a/apps/zbugs-gpui/src/app/e2e.rs b/apps/zbugs-gpui/src/app/e2e.rs index eb2fac09ac..f62ad3e842 100644 --- a/apps/zbugs-gpui/src/app/e2e.rs +++ b/apps/zbugs-gpui/src/app/e2e.rs @@ -123,11 +123,11 @@ async fn settled(app: &Entity, cx: &mut AsyncApp) -> Step { } fn rows(app: &ZbugsApp) -> Vec { - app.list.snapshot().value.clone().unwrap_or_default() + (*app.list.value()).clone() } fn issue(app: &ZbugsApp) -> Option { - app.detail.as_ref()?.snapshot().value.clone().flatten() + (*app.detail.as_ref()?.value()).clone() } async fn online(window: AnyWindowHandle, app: &Entity, cx: &mut AsyncApp) -> Step { @@ -143,7 +143,7 @@ async fn online(window: AnyWindowHandle, app: &Entity, cx: &mut AsyncA app.selected.is_some() && app.selected.as_ref() == rows(app).first().map(|i| &i.id) }) .await?; - let open = app.read_with(cx, |app, _| app.count.snapshot().value.unwrap_or(0)); + let open = app.read_with(cx, |app, _| *app.count.value()); shot("1-list"); // Issues created and deleted again, one from the issue and two @@ -206,8 +206,7 @@ async fn online(window: AnyWindowHandle, app: &Entity, cx: &mut AsyncA }) .await?; until(app, cx, "the new issue at the top of the list", |app| { - rows(app).first().is_some_and(|i| i.id == id) - && app.count.snapshot().value == Some(open + 1) + rows(app).first().is_some_and(|i| i.id == id) && *app.count.value() == open + 1 }) .await?; @@ -312,21 +311,14 @@ async fn online(window: AnyWindowHandle, app: &Entity, cx: &mut AsyncA // The bug label, and the issue assigned to the user. until(app, cx, "the labels", |app| { - app.labels - .snapshot() - .value - .iter() - .flatten() - .any(|l| l.name == "bug") + app.labels.value().iter().any(|l| l.name == "bug") }) .await?; act(window, app, cx, |app, _, cx| { let bug = app .labels - .snapshot() - .value + .value() .iter() - .flatten() .find(|l| l.name == "bug") .map(|l| l.id.clone()); let issue = issue(app).expect("the issue is shown"); @@ -397,9 +389,7 @@ async fn online(window: AnyWindowHandle, app: &Entity, cx: &mut AsyncA // and All shows it. let login = app .read_with(cx, |app, _| { - app.me - .as_ref() - .and_then(|me| me.snapshot().value.clone().flatten()) + app.me.as_ref().and_then(|me| (*me.value()).clone()) }) .map(|u| u.login) .ok_or("the user's row didn't sync")?; diff --git a/apps/zbugs-gpui/src/live.rs b/apps/zbugs-gpui/src/live.rs index 46710b20ae..aebfa7ee7d 100644 --- a/apps/zbugs-gpui/src/live.rs +++ b/apps/zbugs-gpui/src/live.rs @@ -1,36 +1,28 @@ -//! A typed Zero query kept current in a GPUI view: the view's rows are read -//! when the client's view changes, and the GPUI view is notified then. +//! A typed Zero query kept current in a GPUI view: a `LiveView`, whose +//! updates notify the GPUI view. //! //! The client calls a view's listeners while it applies a change, which may -//! be inside a GPUI update of the same view, so a listener only stores the -//! rows and sends a wakeup; a foreground task notifies the view afterwards. +//! be inside a GPUI update of the same view, so a listener only sends a +//! wakeup; a foreground task notifies the view afterwards. use futures::StreamExt; use futures::channel::mpsc; use gpui_kit::{Context, Task}; -use std::cell::RefCell; use std::rc::Rc; use zbugs_model::Context as Auth; -use zero_client::client::{MaterializedView, Zero}; +use zero_client::client::{Error, LiveView, Zero}; use zero_client::define::QueryRequest; use zero_client::ivm::ResultType; use zero_client::ttl::Ttl; use zero_client::typed::QueryValue; pub struct Live { - value: Rc>>, - view: MaterializedView, - remove: Box, + /// None when the query's rows didn't read at first. + live: Option>, _notify: Task<()>, } -pub struct Snapshot { - pub value: Option, - pub result_type: ResultType, - pub error: Option, -} - -impl Live { +impl Live { /// Materializes `query` and notifies the view `cx` is for whenever its /// rows change, until dropped. pub fn new( @@ -39,28 +31,23 @@ impl Live { ttl: &Ttl, cx: &mut Context, ) -> Live { - let value = Rc::new(RefCell::new(Snapshot { - value: None, - result_type: ResultType::Unknown, - error: None, - })); let (wake, mut woken) = mpsc::unbounded::<()>(); - let view = zero - .materialize(&query, ttl) - .expect("a query of the app's schema materializes"); - let v = value.clone(); - let remove = view.add_listener(move |data, result_type, _| { - let mut s = v.borrow_mut(); - match query.read(data) { - Ok(t) => { - s.value = Some(t); - s.error = None; - } - Err(e) => s.error = Some(e.0), + let live = match zero.live(&query, ttl) { + Ok(live) => { + // Removed when the live view drops. + let _ = live.add_listener(move |_, _| { + let _ = wake.unbounded_send(()); + }); + Some(live) } - s.result_type = result_type; - let _ = wake.unbounded_send(()); - }); + // Rows the generated types can't read are a bug of the port: + // say so, and show none until they read. + Err(Error::Message(e)) => { + eprintln!("zbugs-gpui: {}: {e}", query.name()); + None + } + Err(_) => panic!("a query of the app's schema materializes"), + }; let notify = cx.spawn(async move |this, cx| { while woken.next().await.is_some() { // Several changes at once notify once. @@ -71,30 +58,21 @@ impl Live { } }); Live { - value, - view, - remove, + live, _notify: notify, } } - pub fn snapshot(&self) -> std::cell::Ref<'_, Snapshot> { - self.value.borrow() + /// The rows last read; none if they never could be. + pub fn value(&self) -> Rc { + self.live.as_ref().map_or_else(Rc::default, LiveView::value) } /// The server has confirmed the rows, on this connection or an earlier /// one. pub fn is_synced(&self) -> bool { - matches!( - self.value.borrow().result_type, - ResultType::Complete | ResultType::Cached - ) - } -} - -impl Drop for Live { - fn drop(&mut self) { - (self.remove)(); - self.view.destroy(); + self.live + .as_ref() + .is_some_and(|l| matches!(l.result_type(), ResultType::Complete | ResultType::Cached)) } } diff --git a/apps/zbugs-rune/src/sync.rs b/apps/zbugs-rune/src/sync.rs index e15a4d1e8b..ddeabde304 100644 --- a/apps/zbugs-rune/src/sync.rs +++ b/apps/zbugs-rune/src/sync.rs @@ -4,11 +4,13 @@ //! `LocalSet`, while Rune owns the main thread and runs no executor. So the //! client lives on a thread of its own, with a current-thread runtime. The //! UI sends it commands (closures that run with the client), and it sends -//! back each view's rows, read as the generated row types, with -//! `AppHandle::update`, which runs them on the UI thread before the next -//! frame. Reading a poke into IVM and the rows out of it stays off the UI -//! thread. +//! back each view's rows, read as the generated row types by a `LiveView`, +//! with `AppHandle::update`, which runs them on the UI thread before the +//! next frame. Reading a poke into IVM and the rows out of it stays off the +//! UI thread. +use std::any::Any; +use std::cell::OnceCell; use std::collections::HashMap; use std::rc::Rc; use std::sync::Arc; @@ -16,8 +18,8 @@ use std::sync::Arc; use rune::AppHandle; use tokio::sync::mpsc; use zero_client::client::{ - ConnectionState, MaterializedView, MutationOutcome, Mutator, MutatorResult, Preload, Zero, - ZeroOptions, ZeroStorage, + ConnectionState, Error, LiveError, LiveView, MutationOutcome, Mutator, MutatorResult, Preload, + Zero, ZeroOptions, ZeroStorage, }; use zero_client::define::QueryRequest; use zero_client::ivm::ResultType; @@ -113,7 +115,7 @@ impl SyncThread { /// Keeps `slot` showing `query`'s rows: `apply` gets them now, from the /// local store, and on every change. Watching a slot again replaces its /// query; the old rows stay in the app's state until the new ones come. - pub fn watch( + pub fn watch( &self, slot: &'static str, query: QueryRequest, @@ -212,23 +214,15 @@ struct Client { unsubscribe: Box, } -type Remake = Rc) -> Option<(MaterializedView, Box)>>; +/// Makes a slot's live view on a client; dropping it stops it. +type Remake = Rc) -> Option>>; -/// A slot's view, and how to make it again on a new client. +/// A slot's live view, and how to make it again on a new client. struct Watch { - view: Option<(MaterializedView, Box)>, + _live: Option>, remake: Remake, } -impl Drop for Watch { - fn drop(&mut self) { - if let Some((view, remove)) = self.view.take() { - remove(); - view.destroy(); - } - } -} - pub struct Worker { ui: Ui, setup: Setup, @@ -301,7 +295,7 @@ impl Worker { } } - fn watch( + fn watch( &mut self, slot: &'static str, query: QueryRequest, @@ -311,17 +305,34 @@ impl Worker { // Built for each client, with its context: logging in changes what // a query asks for. let remake: Remake = Rc::new(move |zero| { - let view = zero.materialize(&query, &Ttl::seconds(10.0)).ok()?; - let (ui, query) = (ui.clone(), query.clone()); - let remove = view.add_listener(move |data, result_type, _| { - // Rows the generated types can't read are a bug of the - // port; say so rather than show nothing. - match query.read(data) { - Ok(rows) => ui.update(move |s, tx| apply(s, rows, result_type, tx)), - Err(e) => eprintln!("zbugs-rune: {slot}: {e}"), + // Rows the generated types can't read are a bug of the port; + // say so rather than show nothing. + let live = match zero.live(&query, &Ttl::seconds(10.0)) { + Ok(live) => live, + Err(Error::Message(e)) => { + eprintln!("zbugs-rune: {slot}: {e}"); + return None; + } + Err(_) => return None, + }; + let cell: Rc>> = Rc::default(); + let weak = Rc::downgrade(&cell); + let ui = ui.clone(); + // Removed when the live view drops. + let _ = live.add_listener(move |rows, result_type| { + // Only set once this first call is over, which reads the + // rows `zero.live` already read. + let error = weak.upgrade().and_then(|c| c.get()?.error()); + match error { + Some(LiveError::Decode(e)) => eprintln!("zbugs-rune: {slot}: {e}"), + _ => { + let rows = rows.clone(); + ui.update(move |s, tx| apply(s, rows, result_type, tx)); + } } }); - Some((view, remove)) + let _ = cell.set(live); + Some(cell as Rc) }); self.watches.remove(slot); self.install(slot, remake); @@ -329,11 +340,23 @@ impl Worker { fn install(&mut self, slot: &'static str, remake: Remake) { let Some(client) = &self.client else { - self.watches.insert(slot, Watch { view: None, remake }); + self.watches.insert( + slot, + Watch { + _live: None, + remake, + }, + ); return; }; - let view = remake(&client.zero); - self.watches.insert(slot, Watch { view, remake }); + let live = remake(&client.zero); + self.watches.insert( + slot, + Watch { + _live: live, + remake, + }, + ); } fn close(&mut self) {