diff --git a/Cargo.lock b/Cargo.lock index 1936579d..be376f6a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -909,11 +909,13 @@ dependencies = [ "improvise-gp", "improvise-io", "improvise-protocol", + "proptest", "serde", "serde_json", "tempfile", "unicode-width", "web-time", + "zset", ] [[package]] diff --git a/crates/improvise-command/Cargo.toml b/crates/improvise-command/Cargo.toml index bc1ef6e9..7d8e6a4c 100644 --- a/crates/improvise-command/Cargo.toml +++ b/crates/improvise-command/Cargo.toml @@ -12,6 +12,7 @@ improvise-protocol = { path = "../improvise-protocol" } improvise-formula = { path = "../improvise-formula" } improvise-io = { path = "../improvise-io" } improvise-gp = { path = "../improvise-gp" } +zset = { path = "../zset" } anyhow = "1" serde = { version = "1", features = ["derive"] } # Parses JSON a user's numbers cross, so it must read back the f64 written: per crate, @@ -26,3 +27,4 @@ web-time = "1" improvise-core = { path = "../improvise-core", features = ["test-support"] } improvise-protocol = { path = "../improvise-protocol", features = ["test-support"] } tempfile = "3" +proptest = "1" diff --git a/crates/improvise-command/src/ui/delivery.rs b/crates/improvise-command/src/ui/delivery.rs new file mode 100644 index 00000000..96c6b4cf --- /dev/null +++ b/crates/improvise-command/src/ui/delivery.rs @@ -0,0 +1,505 @@ +//! The host end of delivery: one notice per committed command, the shared bounded +//! log, and corrections (ADR 009 clause 9). Why: `context/code/delivery.md` §5-§7. + +use std::collections::{BTreeMap, BTreeSet, VecDeque}; +use std::sync::Arc; + +use improvise_core::view::layout::GridLayout; +use improvise_core::workbook::Workbook; +use improvise_protocol::delivery::wire::{self, Cell, WireKey, WireValue}; +use improvise_protocol::delivery::{ + Ask, AxisChange, Correction, CorrectionBody, Edge, EdgeError, Flags, Moved, Notice, PubRange, + Reply, Request, ShapeKey, Stamp, Window, +}; +use zset::ZSet; + +use crate::ui::host::Host; + +/// The most notices the log keeps, and the most keys one shape lists in one notice. +pub const LOG_CAPACITY: usize = 1024; +pub const MAX_KEYS: usize = 4096; + +const EAGER: &str = "eager:view/"; + +/// The Eager stand-in for a shape key: the view it draws (`context/code/delivery.md` §5). +pub fn shape_of_view(view: &str) -> ShapeKey { + ShapeKey(format!("{EAGER}{view}")) +} + +fn view_of(shape: &ShapeKey) -> Option<&str> { + shape.0.strip_prefix(EAGER) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct Placed { + row: u128, + col: u128, + value: WireValue, +} + +/// A shape's whole output, keyed by tuple identity with the position attached. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct Output { + cells: BTreeMap, +} + +impl Output { + pub fn of(wb: &Workbook, view: &str) -> Option { + let model = &wb.model; + let layout = GridLayout::with_frozen_records(model, wb.views.get(view)?, None); + let eval = layout.evaluator(model); + let mut cells = BTreeMap::new(); + for row in 0..layout.row_count() { + for col in 0..layout.col_count() { + let Some(key) = layout.cell_key(row, col) else { + continue; + }; + let value = match layout.is_records_mode() { + true => layout.records_value(row, col), + false => eval.cell(&key), + }; + if let Some(value) = value { + cells.insert( + WireKey::of(&key), + Placed { + row: row as u128, + col: col as u128, + value: WireValue::of(&value), + }, + ); + } + } + } + Some(Output { cells }) + } + + /// The cells in `window` as a Z-set of weight one: the target a correction is read from. + pub fn target(&self, window: &Window, only: Option<&BTreeSet>) -> ZSet { + self.cells + .iter() + .filter(|(k, p)| window.contains(p.row, p.col) && only.is_none_or(|s| s.contains(k))) + .map(|(k, p)| ((k.clone(), p.value.clone()), 1)) + .collect() + } + + /// The keys whose value, or presence, differs. A moved position is an axis change, not a key. + fn moved_since(&self, before: &Output) -> BTreeSet { + let changed = self + .cells + .iter() + .filter(|(k, p)| before.cells.get(*k).is_none_or(|b| b.value != p.value)); + let gone = before.cells.keys().filter(|k| !self.cells.contains_key(*k)); + changed.map(|(k, _)| k).chain(gone).cloned().collect() + } +} + +/// What a notice reports besides cells: the items on each axis and the document-wide settings. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +struct Structure { + categories: Vec<(String, Vec)>, + formulas: Vec, + measure_agg: String, + views: String, + default_view: String, +} + +impl Structure { + fn of(wb: &Workbook) -> Self { + let model = &wb.model; + let aggs: BTreeMap<_, _> = model.measure_agg.iter().collect(); + Self { + categories: model + .categories + .iter() + .map(|(name, cat)| (name.clone(), cat.items.keys().cloned().collect())) + .collect(), + formulas: model.formulas().iter().map(|f| f.raw.clone()).collect(), + measure_agg: format!("{aggs:?}"), + views: format!("{:?}", wb.views), + default_view: wb.active_view.clone(), + } + } + + fn since(&self, before: &Structure) -> (Vec, Flags) { + let flags = Flags { + formulas: self.formulas != before.formulas, + measure_agg: self.measure_agg != before.measure_agg, + views: self.views != before.views, + default_view: self.default_view != before.default_view, + }; + (axis_changes(&before.categories, &self.categories), flags) + } +} + +fn axis_changes(before: &[(String, Vec)], after: &[(String, Vec)]) -> Vec { + let held: BTreeMap<_, _> = before.iter().map(|(c, i)| (c.as_str(), i)).collect(); + let now: BTreeSet<_> = after.iter().map(|(c, _)| c.as_str()).collect(); + let gone = before + .iter() + .filter(|(c, _)| !now.contains(c.as_str())) + .map(|(category, _)| AxisChange::CategoryRemoved { + category: category.clone(), + }); + let rest = after.iter().flat_map(|(category, items)| { + let Some(was) = held.get(category.as_str()) else { + let added = AxisChange::CategoryAdded { + category: category.clone(), + }; + let inserted = items.iter().enumerate().map(|(at, item)| AxisChange::ItemInserted { + category: category.clone(), + item: item.clone(), + at, + }); + return std::iter::once(added).chain(inserted).collect::>(); + }; + item_changes(category, was, items) + }); + gone.chain(rest).collect() +} + +fn item_changes(category: &str, was: &[String], now: &[String]) -> Vec { + let removed = was + .iter() + .filter(|i| !now.contains(i)) + .map(|item| AxisChange::ItemRemoved { + category: category.into(), + item: item.clone(), + }); + let inserted = now + .iter() + .enumerate() + .filter(|(_, i)| !was.contains(i)) + .map(|(at, item)| AxisChange::ItemInserted { + category: category.into(), + item: item.clone(), + at, + }); + let kept_before: Vec<_> = was.iter().filter(|i| now.contains(i)).collect(); + let kept_after: Vec<_> = now.iter().filter(|i| was.contains(i)).collect(); + let reordered = (kept_before != kept_after).then(|| AxisChange::ItemsReordered { + category: category.into(), + items: now.to_vec(), + }); + removed.chain(inserted).chain(reordered).collect() +} + +#[derive(Debug, Clone)] +struct Live { + view: String, + since: u64, + last_touch: u64, + output: Arc, +} + +/// The publisher, the shared log and the live shapes of one document. Cheap to clone, +/// so a frame that fails can put it back (`context/code/delivery.md` §6). +#[derive(Debug, Clone)] +pub struct Delivery { + host: u64, + generation: u64, + publication: u64, + log: VecDeque>, + capacity: usize, + max_keys: usize, + live: BTreeMap, + structure: Arc, + evicted: BTreeSet, + outbox: Vec>, +} + +impl Delivery { + /// `host` is a random instance id, never a time (clause 9). + pub fn new(host: u64, wb: &Workbook, generation: u64) -> Self { + Self { + host, + generation, + publication: 0, + log: VecDeque::new(), + capacity: LOG_CAPACITY, + max_keys: MAX_KEYS, + live: BTreeMap::new(), + structure: Arc::new(Structure::of(wb)), + evicted: BTreeSet::new(), + outbox: Vec::new(), + } + } + + pub fn set_limits(&mut self, capacity: usize, max_keys: usize) { + self.capacity = capacity; + self.max_keys = max_keys; + } + + pub fn stamp(&self) -> Stamp { + Stamp { + host: self.host, + generation: self.generation, + publication: self.publication, + } + } + + pub fn log_len(&self) -> usize { + self.log.len() + } + + pub fn is_live(&self, shape: &ShapeKey) -> bool { + self.live.contains_key(shape) + } + + /// Notices published since the last call, for the transport to push. + pub fn take_outbox(&mut self) -> Vec { + std::mem::take(&mut self.outbox) + .into_iter() + .map(|n| (*n).clone()) + .collect() + } + + /// One command committed: publish its notice. The document is read once, here, and every + /// read after it sees this state until the next commit. + pub fn commit(&mut self, wb: &Workbook, generation: u64) -> Stamp { + let new_epoch = generation != self.generation; + if new_epoch { + self.generation = generation; + self.log.clear(); + } + self.publication += 1; + let range = PubRange::one(self.stamp()); + let structure = Arc::new(Structure::of(wb)); + let (axis, flags) = match new_epoch { + true => (Vec::new(), all_flags()), + false => structure.since(&self.structure), + }; + self.structure = structure; + + let mut notice = Notice::empty(range); + notice.axis = axis; + notice.flags = flags; + notice.evicted = std::mem::take(&mut self.evicted); + let max_keys = self.max_keys; + self.live.retain(|shape, live| { + let Some(output) = Output::of(wb, &live.view) else { + notice.evicted.insert(shape.clone()); + return false; + }; + let keys = output.moved_since(&live.output); + let moved = match (new_epoch, keys.len() > max_keys) { + (false, false) => Moved::Keys(keys), + _ => Moved::Correct, + }; + if moved != Moved::Keys(BTreeSet::new()) { + notice.moved.insert(shape.clone(), moved); + } + live.output = Arc::new(output); + true + }); + let notice = Arc::new(notice); + self.log.push_back(Arc::clone(¬ice)); + while self.log.len() > self.capacity { + self.log.pop_front(); + } + self.outbox.push(notice); + self.stamp() + } + + /// A request names `shape`: it goes live if it was not, and is not idle. + pub fn touch(&mut self, shape: &ShapeKey, wb: &Workbook, now: u64) { + match self.live.get_mut(shape) { + Some(live) => live.last_touch = now, + None => { + let Some(view) = view_of(shape) else { return }; + let Some(output) = Output::of(wb, view) else { + return; + }; + self.live.insert( + shape.clone(), + Live { + view: view.to_string(), + since: self.publication, + last_touch: now, + output: Arc::new(output), + }, + ); + } + } + } + + /// Drop shapes no request has named for `idle`, a local duration in the caller's unit. + pub fn evict_idle(&mut self, now: u64, idle: u64) -> Vec { + let gone: Vec<_> = self + .live + .iter() + .filter(|(_, l)| now.saturating_sub(l.last_touch) >= idle) + .map(|(s, _)| s.clone()) + .collect(); + for shape in &gone { + self.live.remove(shape); + self.evicted.insert(shape.clone()); + } + gone + } + + /// Answer a request from the last committed state. + pub fn answer(&mut self, wb: &Workbook, request: &Request, now: u64) -> Reply { + self.touch(&request.shape, wb, now); + match &request.ask { + Ask::Notices { after } => match self.covered(&request.shape, Some(after)) { + Some(after) => Reply::Notices(self.notices_after(&request.shape, after, None)), + None => Reply::Correction(self.window(request, None, self.stamp(), None)), + }, + Ask::Correction { watermark } => match self.covered(&request.shape, watermark.as_ref()) + { + Some(from) => self.moved(request, from, self.stamp(), None, false), + None => Reply::Correction(self.window(request, None, self.stamp(), None)), + }, + Ask::Fetch { + watermark, + keys, + through, + window, + } => { + let sane = through.same_epoch(&self.stamp()) + && through.publication <= self.publication; + match (self.covered(&request.shape, watermark.as_ref()), sane) { + (Some(from), true) if from.publication <= through.publication => { + self.moved(request, from, *through, Some(keys), *window) + } + _ => Reply::Correction(self.window(request, None, self.stamp(), None)), + } + } + } + } + + /// The watermark, if the log and the shape's liveness reach back to it. + fn covered(&self, shape: &ShapeKey, watermark: Option<&Stamp>) -> Option { + let w = *watermark?; + let live = self.live.get(shape)?; + let log_start = self.log.front().map_or(self.publication, |n| n.range.from); + let reaches = w.publication <= self.publication + && (w.publication == self.publication + || (w.publication >= log_start && w.publication >= live.since)); + (w.same_epoch(&self.stamp()) && reaches).then_some(w) + } + + fn notices_after(&self, shape: &ShapeKey, after: Stamp, through: Option) -> Vec { + self.log + .iter() + .filter(|n| n.range.to > after.publication) + .filter(|n| through.is_none_or(|t| n.range.to <= t.publication)) + .map(|n| n.for_shape(shape)) + .collect() + } + + fn window( + &self, + request: &Request, + from: Option, + through: Stamp, + carried: Option, + ) -> Correction { + let target = self + .live + .get(&request.shape) + .map(|l| l.output.target(&request.window, None)) + .unwrap_or_default(); + Correction { + shape: request.shape.clone(), + at: self.stamp(), + from, + through, + body: CorrectionBody::Window, + target: wire::encode(&target), + carried, + } + } + + fn moved( + &self, + request: &Request, + from: Stamp, + through: Stamp, + only: Option<&BTreeSet>, + window: bool, + ) -> Reply { + let covering = self.notices_after(&request.shape, from, Some(through)); + let oversized = covering + .iter() + .any(|n| n.moved.get(&request.shape) == Some(&Moved::Correct)); + let carried = covering + .iter() + .cloned() + .reduce(|a, b| a.merge(b).unwrap_or_else(|(_, b)| b)); + let restructured = covering + .iter() + .any(|n| !n.axis.is_empty() || !n.relabels.is_empty()); + if oversized { + return Reply::Correction(self.window(request, None, self.stamp(), None)); + } + if window || restructured { + return Reply::Correction(self.window(request, Some(from), through, carried)); + } + let keys: BTreeSet = match only { + Some(keys) => keys.clone(), + None => covering + .iter() + .filter_map(|n| n.moved.get(&request.shape)) + .flat_map(|m| match m { + Moved::Keys(k) => k.iter().cloned().collect::>(), + Moved::Correct => Vec::new(), + }) + .collect(), + }; + let target = self + .live + .get(&request.shape) + .map(|l| l.output.target(&request.window, Some(&keys))) + .unwrap_or_default(); + Reply::Correction(Correction { + shape: request.shape.clone(), + at: self.stamp(), + from: Some(from), + through, + body: CorrectionBody::Moved { keys }, + target: wire::encode(&target), + carried, + }) + } +} + +fn all_flags() -> Flags { + Flags { + formulas: true, + measure_agg: true, + views: true, + default_view: true, + } +} + +/// An edge that is a function call: the host in this process. +pub struct InProcess<'a> { + pub host: &'a mut Host, + pub now: u64, +} + +impl Edge for InProcess<'_> { + fn call(&mut self, request: Request) -> Result { + Ok(self.host.deliver(&request, self.now)) + } +} + +/// An edge whose request and reply cross as JSON text: the adapter at a wire boundary. +pub struct WireEdge(pub E); + +impl Edge for WireEdge { + fn call(&mut self, request: Request) -> Result { + let sent = serde_json::to_string(&request).map_err(|_| EdgeError::Unreadable)?; + let arrived = serde_json::from_str(&sent).map_err(|_| EdgeError::Unreadable)?; + let reply = self.0.call(arrived)?; + let sent = serde_json::to_string(&reply).map_err(|_| EdgeError::Unreadable)?; + serde_json::from_str(&sent).map_err(|_| EdgeError::Unreadable) + } +} + +/// A notice as it crosses a wire. +pub fn notice_over_the_wire(notice: &Notice) -> Result { + let sent = serde_json::to_string(notice).map_err(|_| EdgeError::Unreadable)?; + serde_json::from_str(&sent).map_err(|_| EdgeError::Unreadable) +} diff --git a/crates/improvise-command/src/ui/delivery_tests.rs b/crates/improvise-command/src/ui/delivery_tests.rs new file mode 100644 index 00000000..d5fa8f3c --- /dev/null +++ b/crates/improvise-command/src/ui/delivery_tests.rs @@ -0,0 +1,866 @@ +//! Delivery against the Eager path: the host publishes, a client folds, and the client's +//! held cells must end at what the grid draws. Evidence: `context/test-evidence/delivery.md`. + +use std::collections::VecDeque; + +use improvise_core::model::cell::{CellKey, CellValue}; +use improvise_core::model::record::RecordHash; +use improvise_core::view::layout::GridLayout; +use improvise_core::view::Axis; +use improvise_core::workbook::Workbook; +use improvise_protocol::delivery::wire::{self, Cell, WireKey, WireValue}; +use improvise_protocol::delivery::{ + Ask, AxisChange, CorrectionBody, Edge, EdgeClient, EdgeError, Moved, Notice, Outcome, Reply, + Request, SessionToken, ShapeKey, Stamp, Synced, Window, +}; +use improvise_protocol::{ClientFrame, ModelCommand, Rejection, RecordCell, ServerFrame}; +use zset::ZSet; + +use super::app::App; +use super::delivery::{InProcess, WireEdge, notice_over_the_wire, shape_of_view}; +use super::host::Host; +use super::project::SessionId; + +fn workbook() -> Workbook { + let mut wb = Workbook::new("Test"); + wb.add_category("Region").unwrap(); + for item in ["North", "South"] { + wb.add_item("Region", item).unwrap(); + } + wb.add_item("_Measure", "Revenue").unwrap(); + for (region, value) in [("North", 10.0), ("South", 20.0)] { + let key = CellKey::new(vec![ + ("Region".to_string(), region.to_string()), + ("_Measure".to_string(), "Revenue".to_string()), + ]); + wb.model.set_cell(key, CellValue::Number(value)); + } + wb.active_view_mut().set_axis("Region", Axis::Row); + wb.active_view_mut().set_axis("_Measure", Axis::Column); + wb +} + +fn host_instance(instance: u64) -> Host { + let mut app = App::new(workbook(), None); + app.rebuild_layout(); + Host::with_instance(app, instance) +} + +fn host() -> Host { + host_instance(1) +} + +fn shape() -> ShapeKey { + shape_of_view("Default") +} + +fn everything() -> Window { + Window { + top: 0, + left: 0, + rows: 1000, + cols: 1000, + } +} + +fn record(region: &str) -> RecordCell { + let key = CellKey::new(vec![ + ("Region".to_string(), region.to_string()), + ("_Measure".to_string(), "Revenue".to_string()), + ]); + RecordCell { + record: RecordHash::of(&key), + measure: Some("Revenue".to_string()), + } +} + +fn set(region: &str, value: f64) -> ModelCommand { + ModelCommand::SetCell { + view: "Default".into(), + cell: record(region), + value: CellValue::Number(value), + } +} + +fn add_item(item: &str) -> ModelCommand { + ModelCommand::AddItem { + category: "Region".into(), + item: item.into(), + } +} + +/// A frame of caught-up commands: the first one's basis states every record the frame names. +fn frame(host: &Host, cmds: Vec) -> ClientFrame { + let mut wire: Vec = cmds + .into_iter() + .map(|c| crate::ui::reduce::caught_up(host.app(), c)) + .collect(); + let records: Vec<_> = wire + .iter() + .filter_map(|c| match c { + improvise_protocol::Command::Model { basis, .. } => Some(basis.records.clone()), + _ => None, + }) + .flatten() + .collect(); + if let Some(improvise_protocol::Command::Model { basis, .. }) = wire.first_mut() { + basis.records = records; + } + ClientFrame::Commands(wire) +} + +fn send(host: &mut Host, id: SessionId, cmds: Vec) -> Vec { + let f = frame(host, cmds); + host.handle(id, f).into_iter().map(|a| a.msg).collect() +} + +fn refused(answer: &[ServerFrame]) -> bool { + answer + .iter() + .any(|m| matches!(m, ServerFrame::Rejected(_))) +} + +/// What the grid draws, derived without the delivery module's own reader. +fn drawn(host: &Host, window: &Window) -> ZSet { + let wb = &host.app().model_state.workbook; + let layout = GridLayout::with_frozen_records(&wb.model, &wb.views["Default"], None); + let mut cells = ZSet::new(); + for row in 0..layout.row_count() { + for col in 0..layout.col_count() { + let Some(key) = layout.cell_key(row, col) else { + continue; + }; + if let Some(value) = layout.evaluate(&wb.model, &key) + && window.contains(row as u128, col as u128) + { + cells.insert((WireKey::of(&key), WireValue::of(&value)), 1); + } + } + } + cells +} + +#[derive(Clone, Copy, Debug, PartialEq)] +enum Fault { + DropRequest, + DropReply, + Garble, +} + +/// A transport that loses things on a script and remembers every request that reached it. +struct Flaky<'a> { + host: &'a mut Host, + script: VecDeque>, + reached: Vec, +} + +impl<'a> Flaky<'a> { + fn new(host: &'a mut Host) -> Self { + Self { + host, + script: VecDeque::new(), + reached: Vec::new(), + } + } + + fn scripted(host: &'a mut Host, script: Vec>) -> Self { + Self { + script: script.into(), + ..Self::new(host) + } + } +} + +impl Edge for Flaky<'_> { + fn call(&mut self, request: Request) -> Result { + let fault = self.script.pop_front().flatten(); + if fault == Some(Fault::DropRequest) { + return Err(EdgeError::Failed); + } + self.reached.push(request.clone()); + let reply = self.host.deliver(&request, 0); + match fault { + Some(Fault::DropReply) => Err(EdgeError::Failed), + Some(Fault::Garble) => Err(EdgeError::Unreadable), + _ => Ok(reply), + } + } +} + +fn session() -> Option { + Some(SessionToken("s-1".into())) +} + +/// A client that has fetched everything the host holds. +fn settled(host: &mut Host) -> EdgeClient { + let mut client = EdgeClient::new(session()); + client.track(shape(), everything()); + let (state, _) = client.sync(&mut WireEdge(InProcess { host, now: 0 }), &shape(), 6); + assert_eq!(state, Synced::Settled); + client +} + +fn connect(host: &mut Host) -> SessionId { + host.connect(10, 10).0 +} + +fn asks(reached: &[Request]) -> Vec<&Ask> { + reached.iter().map(|r| &r.ask).collect() +} + +fn deliver_all(client: &mut EdgeClient, notices: &[Notice]) { + for n in notices { + client.on_notice(¬ice_over_the_wire(n).expect("a notice crosses the wire")); + } +} + +fn held_equals_drawn(client: &EdgeClient, host: &Host) { + let want = drawn(host, &everything()); + assert_eq!(client.held(&shape()), Some(&want)); +} + +// ── one notice per committed command ─────────────────────────────────────── + +#[test] +fn a_frame_of_two_commands_publishes_one_notice_each_and_no_client_asks_for_a_correction() { + let mut host = host(); + let id = connect(&mut host); + let mut client = settled(&mut host); + let before = host.delivery().stamp(); + host.take_notices(); + + let answer = send(&mut host, id, vec![add_item("East"), set("North", 5.0)]); + assert!(!refused(&answer), "{answer:?}"); + let notices = host.take_notices(); + + let stamps: Vec = notices.iter().map(|n| n.stamp().publication).collect(); + assert_eq!( + stamps, + vec![before.publication + 1, before.publication + 2], + "one notice per committed command, numbered consecutively" + ); + deliver_all(&mut client, ¬ices); + let mut edge = Flaky::new(&mut host); + let (state, _) = client.sync(&mut edge, &shape(), 6); + assert_eq!(state, Synced::Settled); + assert!( + asks(&edge.reached) + .iter() + .all(|a| !matches!(a, Ask::Correction { .. })), + "no correction was asked for: {:?}", + edge.reached + ); + held_equals_drawn(&client, &host); +} + +#[test] +fn a_command_that_moves_nothing_still_publishes_and_a_view_command_does_not() { + let mut host = host(); + let id = connect(&mut host); + host.take_notices(); + send(&mut host, id, vec![set("North", 1.0)]); + send(&mut host, id, vec![set("North", 1.0)]); + let same = host.take_notices(); + assert_eq!(same.len(), 2); + assert_eq!(same[1].moved, Default::default(), "an equal write moves nothing"); + + host.handle( + id, + ClientFrame::command(improvise_protocol::Command::SetSelected { row: 1, col: 0 }), + ); + assert!(host.take_notices().is_empty(), "the cursor is the view store's"); +} + +#[test] +fn a_notice_names_the_moved_cells_including_a_formula_cell_it_did_not_write() { + let mut host = host(); + let id = connect(&mut host); + send( + &mut host, + id, + vec![ModelCommand::AddFormula { + raw: "Double = Revenue * 2".into(), + target_category: "_Measure".into(), + }], + ); + let _ = settled(&mut host); + host.take_notices(); + send(&mut host, id, vec![set("North", 4.0)]); + let notice = host.take_notices().remove(0); + let Some(Moved::Keys(keys)) = notice.moved.get(&shape()) else { + panic!("the live shape must be listed: {notice:?}") + }; + let measures: Vec<&str> = keys + .iter() + .flat_map(|k| k.0.iter()) + .filter(|(c, _)| c == "_Measure") + .map(|(_, i)| i.as_str()) + .collect(); + assert!(measures.contains(&"Revenue") && measures.contains(&"Double"), "{keys:?}"); +} + +#[test] +fn a_structural_command_reports_the_axis_change_and_a_formula_the_flag() { + let mut host = host(); + let id = connect(&mut host); + host.take_notices(); + send(&mut host, id, vec![add_item("East")]); + send(&mut host, id, vec![ModelCommand::RemoveItem { + category: "Region".into(), + item: "North".into(), + }]); + send( + &mut host, + id, + vec![ModelCommand::AddFormula { + raw: "Double = Revenue * 2".into(), + target_category: "_Measure".into(), + }], + ); + let notices = host.take_notices(); + assert_eq!( + notices[0].axis, + vec![AxisChange::ItemInserted { + category: "Region".into(), + item: "East".into(), + at: 2 + }] + ); + assert_eq!( + notices[1].axis, + vec![AxisChange::ItemRemoved { + category: "Region".into(), + item: "North".into() + }] + ); + assert!(notices[2].flags.formulas && !notices[1].flags.formulas); +} + +// ── loss, restart, refusal ───────────────────────────────────────────────── + +#[test] +fn a_dropped_notice_is_detected_and_a_correction_repairs_it() { + let mut host = host(); + let id = connect(&mut host); + let mut client = settled(&mut host); + host.take_notices(); + send(&mut host, id, vec![set("North", 1.0)]); + send(&mut host, id, vec![set("South", 2.0)]); + let notices = host.take_notices(); + deliver_all(&mut client, ¬ices[1..]); + + let mut edge = Flaky::new(&mut host); + let (state, _) = client.sync(&mut edge, &shape(), 6); + assert_eq!(state, Synced::Settled); + assert!( + matches!(asks(&edge.reached)[0], Ask::Correction { watermark: Some(_) }), + "{:?}", + edge.reached + ); + held_equals_drawn(&client, &host); +} + +#[test] +fn a_reconnect_after_a_host_restart_asks_for_a_correction_from_the_old_watermark() { + let mut first = host_instance(1); + let id = connect(&mut first); + send(&mut first, id, vec![set("North", 9.0)]); + let mut client = settled(&mut first); + let old = client.watermark(&shape()).expect("settled"); + + let mut second = host_instance(2); + client.connection_lost(); + let mut edge = Flaky::new(&mut second); + let (state, _) = client.sync(&mut edge, &shape(), 6); + assert_eq!(state, Synced::Settled); + assert_eq!( + asks(&edge.reached)[0], + &Ask::Correction { watermark: Some(old) } + ); + assert_ne!(client.watermark(&shape()).unwrap().host, old.host); + held_equals_drawn(&client, &second); +} + +#[test] +fn a_notice_from_a_restarted_host_asks_for_a_correction_too() { + let mut first = host_instance(1); + let mut client = settled(&mut first); + let mut second = host_instance(2); + let id = connect(&mut second); + send(&mut second, id, vec![set("North", 3.0)]); + deliver_all(&mut client, &second.take_notices()); + let mut edge = Flaky::new(&mut second); + client.sync(&mut edge, &shape(), 6); + assert!(matches!(asks(&edge.reached)[0], Ask::Correction { .. })); + held_equals_drawn(&client, &second); +} + +#[test] +fn a_refused_command_consumes_no_publication_number() { + let mut host = host(); + let id = connect(&mut host); + for n in 0..crate::model::MAX_CATEGORIES - 1 { + host.app_mut() + .model_state + .workbook + .add_category(format!("Filler{n}")) + .unwrap(); + } + host.app_mut().rebuild_layout(); + let before = host.delivery().stamp(); + host.take_notices(); + + let answer = send( + &mut host, + id, + vec![add_item("West"), ModelCommand::AddCategory("Overflow".into())], + ); + assert!( + answer + .iter() + .any(|m| matches!(m, ServerFrame::Rejected(Rejection::MutationRefused))), + "the fixture must reach the refusal: {answer:?}" + ); + assert_eq!(host.delivery().stamp(), before, "nothing was committed"); + assert!(host.take_notices().is_empty()); + + send(&mut host, id, vec![set("North", 1.0)]); + assert_eq!( + host.take_notices()[0].stamp().publication, + before.publication + 1, + "the next commit takes the next number" + ); +} + +#[test] +fn a_refused_frame_is_never_read_back() { + let mut host = host(); + let id = connect(&mut host); + for n in 0..crate::model::MAX_CATEGORIES - 1 { + host.app_mut() + .model_state + .workbook + .add_category(format!("Filler{n}")) + .unwrap(); + } + host.app_mut().rebuild_layout(); + let mut client = settled(&mut host); + let committed = drawn(&host, &everything()); + send( + &mut host, + id, + vec![set("North", 77.0), ModelCommand::AddCategory("Overflow".into())], + ); + let mut edge = Flaky::new(&mut host); + client.connection_lost(); + client.sync(&mut edge, &shape(), 6); + assert_eq!(client.held(&shape()), Some(&committed)); + let leaked = client + .held(&shape()) + .unwrap() + .iter() + .any(|((_, v), _)| *v == WireValue::of(&CellValue::Number(77.0))); + assert!(!leaked, "a value the failed frame wrote was read"); +} + +#[test] +fn commands_before_a_refused_one_are_rolled_back_with_their_notices_under_adr_001() { + let mut host = host(); + let id = connect(&mut host); + for n in 0..crate::model::MAX_CATEGORIES - 1 { + host.app_mut() + .model_state + .workbook + .add_category(format!("Filler{n}")) + .unwrap(); + } + host.app_mut().rebuild_layout(); + host.take_notices(); + send( + &mut host, + id, + vec![add_item("West"), ModelCommand::AddCategory("Overflow".into())], + ); + assert!( + host.take_notices().is_empty(), + "ADR 001 clause 4 undoes the whole frame; ADR 009 (via ADR 006) would keep `West`'s notice" + ); +} + +#[test] +fn a_dropped_connection_or_a_failed_request_triggers_a_correction_and_the_session_resumes() { + let mut host = host(); + let id = connect(&mut host); + let mut client = settled(&mut host); + let at = client.watermark(&shape()).unwrap(); + send(&mut host, id, vec![set("North", 8.0)]); + deliver_all(&mut client, &host.take_notices()); + + for fault in [Fault::DropRequest, Fault::DropReply, Fault::Garble] { + let mut edge = Flaky::scripted(&mut host, vec![Some(fault)]); + let (state, _) = client.sync(&mut edge, &shape(), 6); + assert_eq!(state, Synced::Failed, "{fault:?}"); + assert!( + matches!(client.wants(&shape()), Some(improvise_protocol::delivery::Want::Correction { .. })), + "{fault:?}: a failed request asks for a correction" + ); + } + let mut edge = Flaky::new(&mut host); + let (state, _) = client.sync(&mut edge, &shape(), 6); + assert_eq!(state, Synced::Settled); + assert_eq!( + asks(&edge.reached)[0], + &Ask::Correction { watermark: Some(at) }, + "it resumes from the watermark it held" + ); + assert!(edge.reached.iter().all(|r| r.session == session())); + held_equals_drawn(&client, &host); + + client.connection_lost(); + let (state, _) = client.sync(&mut Flaky::new(&mut host), &shape(), 6); + assert_eq!(state, Synced::Settled, "and a dropped connection settles too"); +} + +// ── the log, windows, eviction ───────────────────────────────────────────── + +fn reply_for(host: &mut Host, ask: Ask, window: Window) -> improvise_protocol::delivery::Correction { + let request = Request { + session: session(), + shape: shape(), + window, + ask, + }; + match host.deliver(&request, 0) { + Reply::Correction(c) => c, + other => panic!("expected a correction: {other:?}"), + } +} + +#[test] +fn a_watermark_the_log_covers_gets_the_moved_keys_and_one_it_does_not_gets_the_window() { + let mut host = host(); + host.limit_delivery(3, 4096); + let id = connect(&mut host); + let client = settled(&mut host); + let w = client.watermark(&shape()).unwrap(); + for n in 0..2 { + send(&mut host, id, vec![set("North", n as f64)]); + } + let near = reply_for(&mut host, Ask::Correction { watermark: Some(w) }, everything()); + assert!(matches!(near.body, CorrectionBody::Moved { .. }), "{near:?}"); + assert_eq!(near.from, Some(w)); + + for n in 2..6 { + send(&mut host, id, vec![set("North", n as f64)]); + } + assert_eq!(host.delivery().log_len(), 3, "the log is bounded"); + let far = reply_for(&mut host, Ask::Correction { watermark: Some(w) }, everything()); + assert!(matches!(far.body, CorrectionBody::Window), "{far:?}"); + assert_eq!(far.from, None); +} + +#[test] +fn an_oversized_notice_says_correct_and_costs_every_reader_a_correction() { + let mut host = host(); + host.limit_delivery(64, 1); + let id = connect(&mut host); + let mut client = settled(&mut host); + host.take_notices(); + send(&mut host, id, vec![set("North", 1.0)]); + let small = host.take_notices(); + assert!(matches!(small[0].moved.get(&shape()), Some(Moved::Keys(k)) if k.len() == 1)); + + let formula = ModelCommand::AddFormula { + raw: "Double = Revenue * 2".into(), + target_category: "_Measure".into(), + }; + send(&mut host, id, vec![formula]); + let big = host.take_notices(); + assert_eq!(big[0].moved.get(&shape()), Some(&Moved::Correct)); + deliver_all(&mut client, &small); + deliver_all(&mut client, &big); + let mut edge = Flaky::new(&mut host); + client.sync(&mut edge, &shape(), 6); + assert!(matches!(asks(&edge.reached)[0], Ask::Correction { .. })); + held_equals_drawn(&client, &host); +} + +#[test] +fn a_correction_carries_only_the_clients_window() { + let mut host = host(); + let id = connect(&mut host); + let top_left = Window { top: 0, left: 0, rows: 1, cols: 1 }; + let all = reply_for(&mut host, Ask::Correction { watermark: None }, everything()); + let one = reply_for(&mut host, Ask::Correction { watermark: None }, top_left); + assert_eq!(all.target.len(), 2); + assert_eq!(one.target.len(), 1, "{one:?}"); + send(&mut host, id, vec![set("South", 5.0)]); + let w = host.delivery().stamp(); + let _ = w; + let moved = reply_for( + &mut host, + Ask::Correction { watermark: Some(Stamp { publication: w.publication - 1, ..w }) }, + top_left, + ); + assert!(moved.target.is_empty(), "South is outside the window: {moved:?}"); +} + +#[test] +fn an_evicted_shape_is_announced_and_its_reader_corrects() { + let mut host = host(); + let id = connect(&mut host); + let mut client = settled(&mut host); + assert!(host.delivery().is_live(&shape())); + let gone = host.evict_idle(100, 50); + assert_eq!(gone, vec![shape()]); + assert!(!host.delivery().is_live(&shape())); + host.take_notices(); + send(&mut host, id, vec![set("North", 2.0)]); + let notices = host.take_notices(); + assert!(notices[0].evicted.contains(&shape())); + deliver_all(&mut client, ¬ices); + assert!(matches!( + client.wants(&shape()), + Some(improvise_protocol::delivery::Want::Correction { .. }) + )); + client.sync(&mut Flaky::new(&mut host), &shape(), 6); + held_equals_drawn(&client, &host); + assert!(host.delivery().is_live(&shape()), "naming it again revives it"); +} + +#[test] +fn a_shape_naming_no_view_is_answered_with_nothing_and_never_goes_live() { + let mut host = host(); + let ghost = shape_of_view("Nope"); + let c = { + let request = Request { + session: None, + shape: ghost.clone(), + window: everything(), + ask: Ask::Correction { watermark: None }, + }; + match host.deliver(&request, 0) { + Reply::Correction(c) => c, + other => panic!("{other:?}"), + } + }; + assert!(c.target.is_empty()); + assert!(!host.delivery().is_live(&ghost)); +} + +#[test] +fn a_document_swap_is_a_new_epoch_behind_every_old_watermark() { + let storage = crate::storage::MemoryFs::new(); + let mut other = Workbook::new("Other"); + other.add_category("Product").unwrap(); + other.add_item("Product", "Widget").unwrap(); + other.add_item("_Measure", "Units").unwrap(); + other.active_view_mut().set_axis("Product", Axis::Row); + other.active_view_mut().set_axis("_Measure", Axis::Column); + crate::persistence::save_to(&storage, &other, std::path::Path::new("other.improv")).unwrap(); + let mut app = App::new(workbook(), None); + app.storage = Box::new(storage); + app.rebuild_layout(); + let mut host = Host::with_instance(app, 1); + let id = connect(&mut host); + let mut client = settled(&mut host); + let old = client.watermark(&shape()).unwrap(); + host.take_notices(); + + let load = ModelCommand::LoadModel("other.improv".into()); + let answer = send(&mut host, id, vec![load]); + assert!(!refused(&answer), "{answer:?}"); + let notices = host.take_notices(); + let now = notices.last().expect("the swap is a committed command").stamp(); + assert!(old.is_behind(&now)); + assert_ne!(old.generation, now.generation); + assert!(notices[0].flags.formulas && notices[0].flags.views); + + deliver_all(&mut client, ¬ices); + let mut edge = Flaky::new(&mut host); + client.sync(&mut edge, &shape(), 6); + assert!(matches!(asks(&edge.reached)[0], Ask::Correction { .. })); + held_equals_drawn(&client, &host); +} + +#[test] +fn a_pull_only_client_follows_the_log_without_a_correction() { + let mut host = host(); + let id = connect(&mut host); + let mut client = settled(&mut host); + send(&mut host, id, vec![set("North", 6.0)]); + send(&mut host, id, vec![set("South", 7.0)]); + let mut edge = Flaky::new(&mut host); + // The client has seen no pushes, so its horizon is its watermark. + let (state, _) = client.poll(&mut edge, &shape()); + assert_eq!(state, Synced::Settled); + assert!(matches!(asks(&edge.reached)[0], Ask::Notices { .. })); + assert!( + asks(&edge.reached) + .iter() + .all(|a| !matches!(a, Ask::Correction { .. })) + ); + held_equals_drawn(&client, &host); +} + +#[test] +fn a_request_can_be_retried_and_answers_the_same() { + let mut host = host(); + let id = connect(&mut host); + let client = settled(&mut host); + send(&mut host, id, vec![set("North", 6.0)]); + let w = client.watermark(&shape()).unwrap(); + let a = reply_for(&mut host, Ask::Correction { watermark: Some(w) }, everything()); + let b = reply_for(&mut host, Ask::Correction { watermark: Some(w) }, everything()); + assert_eq!(a, b); +} + +// ── lossless through the wire ────────────────────────────────────────────── + +#[test] +fn values_keep_every_bit_from_the_cell_to_the_client() { + let mut host = host(); + let id = connect(&mut host); + let mut client = settled(&mut host); + let values = [0.1 + 0.2, -0.0, f64::INFINITY, f64::NEG_INFINITY, f64::NAN, 1e308, 5e-324]; + for (n, v) in values.iter().enumerate() { + let region = ["North", "South"][n % 2]; + send(&mut host, id, vec![set(region, *v)]); + deliver_all(&mut client, &host.take_notices()); + let mut edge = WireEdge(Flaky::new(&mut host)); + assert_eq!(client.sync(&mut edge, &shape(), 6).0, Synced::Settled); + let held = client.held(&shape()).unwrap(); + let got = held.iter().find_map(|((k, v), _)| { + (k.0.contains(&("Region".to_string(), region.to_string()))).then(|| v.to_cell_value()) + }); + let Some(CellValue::Number(got)) = got else { + panic!("{region} holds no number after {v:?}") + }; + match v.is_nan() { + true => assert!(got.is_nan()), + false => assert_eq!(got.to_bits(), v.to_bits(), "{v:?} became {got:?}"), + } + } +} + +// ── the whole thing, at random ───────────────────────────────────────────── + +mod random { + use proptest::prelude::*; + + use super::*; + + #[derive(Debug, Clone)] + enum Op { + Set(bool, usize), + Add(usize), + Remove(usize), + Formula, + } + + fn op() -> impl Strategy { + prop_oneof![ + 4 => (any::(), 0usize..8).prop_map(|(r, v)| Op::Set(r, v)), + 2 => (0usize..3).prop_map(Op::Add), + 1 => (0usize..3).prop_map(Op::Remove), + 1 => Just(Op::Formula), + ] + } + + const VALUES: [f64; 8] = [1.0, 2.5, -0.0, 0.0, f64::INFINITY, f64::NEG_INFINITY, f64::NAN, 0.30000000000000004]; + const EXTRA: [&str; 3] = ["East", "West", "North"]; + + fn command(op: &Op) -> ModelCommand { + match op { + Op::Set(north, v) => set(["South", "North"][*north as usize], VALUES[*v]), + Op::Add(i) => add_item(EXTRA[*i]), + Op::Remove(i) => ModelCommand::RemoveItem { + category: "Region".into(), + item: EXTRA[*i].into(), + }, + Op::Formula => ModelCommand::AddFormula { + raw: "Double = Revenue * 2".into(), + target_category: "_Measure".into(), + }, + } + } + + #[derive(Debug, Clone)] + struct Weather { + deliver_notice: bool, + lose_connection: bool, + sync: Vec>, + } + + fn weather() -> impl Strategy { + ( + any::(), + 0u8..8, + proptest::collection::vec(proptest::option::of(0u8..3), 0..3), + ) + .prop_map(|(deliver_notice, lose, sync)| Weather { + deliver_notice, + lose_connection: lose == 0, + sync, + }) + } + + fn fault(code: u8) -> Fault { + [Fault::DropRequest, Fault::DropReply, Fault::Garble][code as usize] + } + + fn windows() -> impl Strategy { + (0u128..2, 0u128..2, 1u128..6, 1u128..4).prop_map(|(top, left, rows, cols)| Window { + top, + left, + rows, + cols, + }) + } + + proptest! { + #[test] + fn a_corrected_state_is_the_target_and_correcting_it_again_is_the_zero_zset( + steps in proptest::collection::vec((op(), weather()), 1..12), + window in windows(), + ) { + let mut host = host(); + let id = connect(&mut host); + let mut client = EdgeClient::new(session()); + client.track(shape(), window); + client.sync(&mut Flaky::new(&mut host), &shape(), 6); + host.take_notices(); + + for (op, weather) in &steps { + send(&mut host, id, vec![command(op)]); + let notices = host.take_notices(); + if weather.deliver_notice { + deliver_all(&mut client, ¬ices); + } + if weather.lose_connection { + client.connection_lost(); + } + let script = weather.sync.iter().map(|f| f.map(fault)).collect(); + client.sync(&mut Flaky::scripted(&mut host, script), &shape(), 6); + } + + let (state, _) = client.poll(&mut WireEdge(Flaky::new(&mut host)), &shape()); + prop_assert_eq!(state, Synced::Settled); + let want = drawn(&host, &window); + prop_assert_eq!(client.held(&shape()), Some(&want)); + + let full = reply_for(&mut host, Ask::Correction { watermark: None }, window); + let correction = &wire::decode(&full.target) - client.held(&shape()).unwrap(); + prop_assert!(correction.is_zero(), "correcting a correct state: {correction:?}"); + + let before = client.held(&shape()).cloned(); + let (outcome, _) = client.apply(&full); + match outcome { + Outcome::Applied { delta } => prop_assert!(delta.is_zero()), + Outcome::Duplicate => {} + other => prop_assert!(false, "{other:?}"), + } + prop_assert_eq!(client.held(&shape()).cloned(), before); + } + } +} diff --git a/crates/improvise-command/src/ui/host.rs b/crates/improvise-command/src/ui/host.rs index b9434734..58553143 100644 --- a/crates/improvise-command/src/ui/host.rs +++ b/crates/improvise-command/src/ui/host.rs @@ -2,9 +2,11 @@ //! *document*, many clients per `App` — see `context/code/command.md`. use crate::ui::app::App; +use crate::ui::delivery::Delivery; use crate::ui::effect; use crate::ui::project::{Applied, Broadcast, Dirty, SessionId, Sessions}; use crate::ui::reduce::{self, Checked, CheckedBatch}; +use improvise_protocol::delivery::{Notice, Reply, Request, ShapeKey}; use improvise_protocol::{AppMode, Command, Rejection}; use improvise_protocol::{ClientFrame, ServerFrame}; @@ -22,16 +24,58 @@ pub struct Addressed { pub struct Host { app: App, sessions: Sessions, + delivery: Delivery, } impl Host { pub fn new(app: App) -> Self { + use std::hash::{BuildHasher, Hasher}; + let instance = std::collections::hash_map::RandomState::new() + .build_hasher() + .finish(); + Self::with_instance(app, instance) + } + + /// A host with a chosen instance id, which must differ from every earlier run's + /// (`context/code/delivery.md` §6). + pub fn with_instance(app: App, instance: u64) -> Self { + let delivery = Delivery::new( + instance, + &app.model_state.workbook, + app.document_generation(), + ); Self { app, sessions: Sessions::default(), + delivery, } } + pub fn delivery(&self) -> &Delivery { + &self.delivery + } + + /// Answer a delivery request from the last committed state. `now` is a local monotonic count. + pub fn deliver(&mut self, request: &Request, now: u64) -> Reply { + self.delivery + .answer(&self.app.model_state.workbook, request, now) + } + + /// Forget shapes no request has named for `idle` (a local duration in `now`'s unit). + pub fn evict_idle(&mut self, now: u64, idle: u64) -> Vec { + self.delivery.evict_idle(now, idle) + } + + /// How many notices the shared log keeps, and how many keys one shape lists in a notice. + pub fn limit_delivery(&mut self, capacity: usize, max_keys: usize) { + self.delivery.set_limits(capacity, max_keys); + } + + /// Notices published since the last call, for the transport to push. + pub fn take_notices(&mut self) -> Vec { + self.delivery.take_outbox() + } + pub fn app(&self) -> &App { &self.app } @@ -224,11 +268,12 @@ impl Host { fn mutating_frame(&mut self, id: SessionId, batch: CheckedBatch<'_>) -> Vec { let undo = self.app.snapshot(); let mirrors = self.sessions.clone(); + let published = self.delivery.clone(); let mut out = Vec::new(); let mut asker_reset = false; for cmd in batch { if let Some(reason) = reduce::broken_chain(&self.app, cmd.command()) { - return self.undo_frame(id, undo, mirrors, reason); + return self.undo_frame(id, (undo, mirrors, published), reason); } // Only a mutation runs `apply_effects`, which resets both flags — so // after anything else they are STALE. A refusal need not abort (cz8d). @@ -252,7 +297,7 @@ impl Host { out.extend(answered); let failed = self.app.last_batch_refused() || self.app.last_batch_aborted(); if mutation && failed { - return self.undo_frame(id, undo, mirrors, Rejection::MutationRefused); + return self.undo_frame(id, (undo, mirrors, published), Rejection::MutationRefused); } } // Last, after everything the frame owed: a client folds the answer to @@ -269,12 +314,12 @@ impl Host { fn undo_frame( &mut self, id: SessionId, - undo: crate::ui::app::AppSnapshot, - mirrors: Sessions, + (undo, mirrors, published): (crate::ui::app::AppSnapshot, Sessions, Delivery), reason: Rejection, ) -> Vec { self.app.restore(undo); self.sessions = mirrors; + self.delivery = published; vec![Addressed { to: id, msg: ServerFrame::Rejected(reason), @@ -284,7 +329,16 @@ impl Host { /// The one door (gotcha 36); the match that follows is the mechanism (gotcha 39): /// asking for the `Dirty` means naming the case where a `Snapshot` is owed. fn command(&mut self, id: SessionId, cmd: Checked<'_>) -> Vec { - match self.sessions.apply_checked(&mut self.app, id, cmd) { + let mutation = matches!(cmd.command(), Command::Model { .. }); + let applied = self.sessions.apply_checked(&mut self.app, id, cmd); + let landed = !matches!(applied, Applied::Rejected { .. }) + && !(self.app.last_batch_refused() || self.app.last_batch_aborted()); + if mutation && landed { + let generation = self.app.document_generation(); + self.delivery + .commit(&self.app.model_state.workbook, generation); + } + match applied { // The document was replaced: nothing to diff and nothing to broadcast, // each snapshot carrying a whole view slice and its seed. Applied::Rebased { snapshots } => snapshots diff --git a/crates/improvise-command/src/ui/mod.rs b/crates/improvise-command/src/ui/mod.rs index 9026994d..12d9d679 100644 --- a/crates/improvise-command/src/ui/mod.rs +++ b/crates/improvise-command/src/ui/mod.rs @@ -1,7 +1,11 @@ pub mod app; pub mod cat_tree; +pub mod delivery; pub mod effect; pub mod host; pub mod project; pub mod reduce; pub mod viewport; + +#[cfg(test)] +mod delivery_tests; diff --git a/crates/improvise-protocol/src/delivery/client.rs b/crates/improvise-protocol/src/delivery/client.rs index 25dcd395..88259a36 100644 --- a/crates/improvise-protocol/src/delivery/client.rs +++ b/crates/improvise-protocol/src/delivery/client.rs @@ -29,6 +29,7 @@ pub enum Want { watermark: Option, keys: BTreeSet, through: Stamp, + window: bool, }, } @@ -60,6 +61,8 @@ struct Tracked { seen: Option, /// Notices received and not yet fetched: `(publication, what moved)`. unapplied: Vec<(Stamp, Moved)>, + /// The last notice that moved an axis: until a window is read past it, positions are unknown. + restructured: Option, stale: bool, } @@ -88,6 +91,7 @@ impl EdgeClient { watermark: None, seen: None, unapplied: Vec::new(), + restructured: None, stale: true, }, ); @@ -104,21 +108,24 @@ impl EdgeClient { pub fn wants(&self, shape: &ShapeKey) -> Option { let t = self.shapes.get(shape)?; let needs_all = t.stale || t.unapplied.iter().any(|(_, m)| *m == Moved::Correct); + let keys: BTreeSet = t + .unapplied + .iter() + .flat_map(|(_, m)| match m { + Moved::Keys(keys) => keys.iter().cloned().collect::>(), + Moved::Correct => Vec::new(), + }) + .collect(); + let owed = !t.unapplied.is_empty() || t.restructured.is_some(); match (needs_all, t.seen) { (true, _) => Some(Want::Correction { watermark: t.watermark, }), - (false, Some(through)) if !t.unapplied.is_empty() => Some(Want::Fetch { + (false, Some(through)) if owed => Some(Want::Fetch { watermark: t.watermark, - keys: t - .unapplied - .iter() - .flat_map(|(_, m)| match m { - Moved::Keys(keys) => keys.iter().cloned(), - Moved::Correct => unreachable!("handled above"), - }) - .collect(), + keys, through, + window: t.restructured.is_some(), }), _ => None, } @@ -136,10 +143,12 @@ impl EdgeClient { watermark, keys, through, + window, } => Ask::Fetch { watermark, keys, through, + window, }, }, }) @@ -161,13 +170,16 @@ impl EdgeClient { t.unapplied.push((end, Moved::Correct)); continue; } + if !notice.axis.is_empty() || !notice.relabels.is_empty() { + t.restructured = Some(end); + } match notice.moved.get(shape) { Some(Moved::Keys(keys)) if keys.is_empty() => {} Some(moved) => t.unapplied.push((end, moved.clone())), None => {} } let current = t.watermark.is_some_and(|w| w.same_epoch(&end)); - if t.unapplied.is_empty() && !t.stale && current { + if t.unapplied.is_empty() && t.restructured.is_none() && !t.stale && current { t.watermark = Some(end); } } @@ -251,10 +263,15 @@ impl EdgeClient { }; t.held += δ t.watermark = Some(c.through); - t.stale = match &c.body { - CorrectionBody::Window => false, - CorrectionBody::Moved { .. } => t.stale, - }; + let window = matches!(c.body, CorrectionBody::Window); + t.stale &= !window; + if window + && t + .restructured + .is_some_and(|r| !r.same_epoch(&c.through) || r.publication <= c.through.publication) + { + t.restructured = None; + } let seen_epoch = t.seen.map(|s| (s.host, s.generation)); t.unapplied.retain(|(at, _)| { Some((at.host, at.generation)) == seen_epoch @@ -266,16 +283,14 @@ impl EdgeClient { if !seen_ahead { t.seen = Some(c.through); } - let mut events = match &c.body { - CorrectionBody::Window => { + let events = match (&c.carried, window) { + (Some(carried), _) => self.events_of(carried), + (None, true) => { self.events_through = Some(c.through); vec![Event::Resync] } - CorrectionBody::Moved { .. } => Vec::new(), + (None, false) => Vec::new(), }; - if let Some(carried) = &c.carried { - events.extend(self.events_of(carried)); - } (Outcome::Applied { delta }, events) } diff --git a/crates/improvise-protocol/src/delivery/client_tests.rs b/crates/improvise-protocol/src/delivery/client_tests.rs index 0121d5e9..aa770870 100644 --- a/crates/improvise-protocol/src/delivery/client_tests.rs +++ b/crates/improvise-protocol/src/delivery/client_tests.rs @@ -111,7 +111,8 @@ fn a_moved_key_wants_a_fetch_and_not_a_correction() { Some(Want::Fetch { watermark: Some(at(1, 4)), keys: [key(1), key(2)].into(), - through: at(1, 5) + through: at(1, 5), + window: false }) ); } @@ -256,7 +257,8 @@ fn a_fetch_does_not_advance_past_the_notices_it_did_not_name() { Some(Want::Fetch { watermark: Some(at(1, 5)), keys: [key(2)].into(), - through: at(1, 6) + through: at(1, 6), + window: false }) ); } diff --git a/crates/improvise-protocol/src/delivery/exchange.rs b/crates/improvise-protocol/src/delivery/exchange.rs index adccc3d4..eaef636a 100644 --- a/crates/improvise-protocol/src/delivery/exchange.rs +++ b/crates/improvise-protocol/src/delivery/exchange.rs @@ -36,11 +36,13 @@ pub struct SessionToken(pub String); pub enum Ask { /// Everything since `watermark`; `None` is a full refetch, `target - empty`. Correction { watermark: Option }, - /// The current values of `keys`, covering the publications up to `through`. + /// The current values of `keys`, covering the publications up to `through`. `window` + /// asks for the whole window too: an axis change moved positions, not values. Fetch { watermark: Option, keys: BTreeSet, through: Stamp, + window: bool, }, /// The notices after `after`, if the host's log still has them. Notices { after: Stamp },