diff --git a/crates/improvise-command/src/ui/delivery.rs b/crates/improvise-command/src/ui/delivery.rs index 96c6b4cf..4f52ed0a 100644 --- a/crates/improvise-command/src/ui/delivery.rs +++ b/crates/improvise-command/src/ui/delivery.rs @@ -131,7 +131,10 @@ impl Structure { } } -fn axis_changes(before: &[(String, Vec)], after: &[(String, Vec)]) -> Vec { +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 @@ -145,11 +148,14 @@ fn axis_changes(before: &[(String, Vec)], after: &[(String, Vec) let added = AxisChange::CategoryAdded { category: category.clone(), }; - let inserted = items.iter().enumerate().map(|(at, item)| AxisChange::ItemInserted { - category: category.clone(), - item: item.clone(), - at, - }); + 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) @@ -343,12 +349,12 @@ impl Delivery { 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)), + None => Reply::Correction(Box::new(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)), + None => Reply::Correction(Box::new(self.window(request, None, self.stamp(), None))), }, Ask::Fetch { watermark, @@ -356,13 +362,15 @@ impl Delivery { through, window, } => { - let sane = through.same_epoch(&self.stamp()) - && through.publication <= self.publication; + 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)), + _ => { + Reply::Correction(Box::new(self.window(request, None, self.stamp(), None))) + } } } } @@ -426,15 +434,15 @@ impl Delivery { let carried = covering .iter() .cloned() - .reduce(|a, b| a.merge(b).unwrap_or_else(|(_, b)| b)); + .reduce(|a, b| a.merge(b).unwrap_or_else(|e| e.1)); 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)); + return Reply::Correction(Box::new(self.window(request, None, self.stamp(), None))); } if window || restructured { - return Reply::Correction(self.window(request, Some(from), through, carried)); + return Reply::Correction(Box::new(self.window(request, Some(from), through, carried))); } let keys: BTreeSet = match only { Some(keys) => keys.clone(), @@ -452,7 +460,7 @@ impl Delivery { .get(&request.shape) .map(|l| l.output.target(&request.window, Some(&keys))) .unwrap_or_default(); - Reply::Correction(Correction { + Reply::Correction(Box::new(Correction { shape: request.shape.clone(), at: self.stamp(), from: Some(from), @@ -460,7 +468,7 @@ impl Delivery { body: CorrectionBody::Moved { keys }, target: wire::encode(&target), carried, - }) + })) } } diff --git a/crates/improvise-command/src/ui/delivery_tests.rs b/crates/improvise-command/src/ui/delivery_tests.rs index d5fa8f3c..0841dc86 100644 --- a/crates/improvise-command/src/ui/delivery_tests.rs +++ b/crates/improvise-command/src/ui/delivery_tests.rs @@ -5,15 +5,15 @@ 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::view::layout::GridLayout; 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 improvise_protocol::{ClientFrame, ModelCommand, RecordCell, Rejection, ServerFrame}; use zset::ZSet; use super::app::App; @@ -115,9 +115,7 @@ fn send(host: &mut Host, id: SessionId, cmds: Vec) -> Vec bool { - answer - .iter() - .any(|m| matches!(m, ServerFrame::Rejected(_))) + answer.iter().any(|m| matches!(m, ServerFrame::Rejected(_))) } /// What the grid draws, derived without the delivery module's own reader. @@ -262,13 +260,20 @@ fn a_command_that_moves_nothing_still_publishes_and_a_view_command_does_not() { 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"); + 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"); + assert!( + host.take_notices().is_empty(), + "the cursor is the view store's" + ); } #[test] @@ -296,7 +301,10 @@ fn a_notice_names_the_moved_cells_including_a_formula_cell_it_did_not_write() { .filter(|(c, _)| c == "_Measure") .map(|(_, i)| i.as_str()) .collect(); - assert!(measures.contains(&"Revenue") && measures.contains(&"Double"), "{keys:?}"); + assert!( + measures.contains(&"Revenue") && measures.contains(&"Double"), + "{keys:?}" + ); } #[test] @@ -305,10 +313,14 @@ fn a_structural_command_reports_the_axis_change_and_a_formula_the_flag() { 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::RemoveItem { + category: "Region".into(), + item: "North".into(), + }], + ); send( &mut host, id, @@ -353,7 +365,10 @@ fn a_dropped_notice_is_detected_and_a_correction_repairs_it() { let (state, _) = client.sync(&mut edge, &shape(), 6); assert_eq!(state, Synced::Settled); assert!( - matches!(asks(&edge.reached)[0], Ask::Correction { watermark: Some(_) }), + matches!( + asks(&edge.reached)[0], + Ask::Correction { watermark: Some(_) } + ), "{:?}", edge.reached ); @@ -375,7 +390,9 @@ fn a_reconnect_after_a_host_restart_asks_for_a_correction_from_the_old_watermark assert_eq!(state, Synced::Settled); assert_eq!( asks(&edge.reached)[0], - &Ask::Correction { watermark: Some(old) } + &Ask::Correction { + watermark: Some(old) + } ); assert_ne!(client.watermark(&shape()).unwrap().host, old.host); held_equals_drawn(&client, &second); @@ -413,7 +430,10 @@ fn a_refused_command_consumes_no_publication_number() { let answer = send( &mut host, id, - vec![add_item("West"), ModelCommand::AddCategory("Overflow".into())], + vec![ + add_item("West"), + ModelCommand::AddCategory("Overflow".into()), + ], ); assert!( answer @@ -436,20 +456,21 @@ fn a_refused_command_consumes_no_publication_number() { 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( + let answer = send( &mut host, id, - vec![set("North", 77.0), ModelCommand::AddCategory("Overflow".into())], + vec![ + set("North", 77.0), + ModelCommand::LoadModel("missing.improv".into()), + ], + ); + assert!( + answer + .iter() + .any(|m| matches!(m, ServerFrame::Rejected(Rejection::MutationRefused))), + "the write must have run before the refusal: {answer:?}" ); let mut edge = Flaky::new(&mut host); client.connection_lost(); @@ -479,7 +500,10 @@ fn commands_before_a_refused_one_are_rolled_back_with_their_notices_under_adr_00 send( &mut host, id, - vec![add_item("West"), ModelCommand::AddCategory("Overflow".into())], + vec![ + add_item("West"), + ModelCommand::AddCategory("Overflow".into()), + ], ); assert!( host.take_notices().is_empty(), @@ -501,7 +525,10 @@ fn a_dropped_connection_or_a_failed_request_triggers_a_correction_and_the_sessio 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 { .. })), + matches!( + client.wants(&shape()), + Some(improvise_protocol::delivery::Want::Correction { .. }) + ), "{fault:?}: a failed request asks for a correction" ); } @@ -510,7 +537,9 @@ fn a_dropped_connection_or_a_failed_request_triggers_a_correction_and_the_sessio assert_eq!(state, Synced::Settled); assert_eq!( asks(&edge.reached)[0], - &Ask::Correction { watermark: Some(at) }, + &Ask::Correction { + watermark: Some(at) + }, "it resumes from the watermark it held" ); assert!(edge.reached.iter().all(|r| r.session == session())); @@ -518,12 +547,20 @@ fn a_dropped_connection_or_a_failed_request_triggers_a_correction_and_the_sessio 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"); + 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 { +fn reply_for( + host: &mut Host, + ask: Ask, + window: Window, +) -> improvise_protocol::delivery::Correction { let request = Request { session: session(), shape: shape(), @@ -531,7 +568,7 @@ fn reply_for(host: &mut Host, ask: Ask, window: Window) -> improvise_protocol::d ask, }; match host.deliver(&request, 0) { - Reply::Correction(c) => c, + Reply::Correction(c) => *c, other => panic!("expected a correction: {other:?}"), } } @@ -546,15 +583,26 @@ fn a_watermark_the_log_covers_gets_the_moved_keys_and_one_it_does_not_gets_the_w 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:?}"); + 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()); + let far = reply_for( + &mut host, + Ask::Correction { watermark: Some(w) }, + everything(), + ); assert!(matches!(far.body, CorrectionBody::Window), "{far:?}"); assert_eq!(far.from, None); } @@ -589,7 +637,12 @@ fn an_oversized_notice_says_correct_and_costs_every_reader_a_correction() { 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 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); @@ -599,10 +652,68 @@ fn a_correction_carries_only_the_clients_window() { let _ = w; let moved = reply_for( &mut host, - Ask::Correction { watermark: Some(Stamp { publication: w.publication - 1, ..w }) }, + Ask::Correction { + watermark: Some(Stamp { + publication: w.publication - 1, + ..w + }), + }, top_left, ); - assert!(moved.target.is_empty(), "South is outside the window: {moved:?}"); + assert!( + moved.target.is_empty(), + "South is outside the window: {moved:?}" + ); +} + +#[test] +fn an_axis_change_that_shifts_an_unmoved_cell_into_the_window_refetches_the_window() { + let mut host = host(); + let id = connect(&mut host); + let one = Window { + top: 0, + left: 0, + rows: 1, + cols: 1, + }; + let mut client = EdgeClient::new(session()); + client.track(shape(), one); + client.sync(&mut Flaky::new(&mut host), &shape(), 6); + host.take_notices(); + + let gone = ModelCommand::RemoveItem { + category: "Region".into(), + item: "North".into(), + }; + send(&mut host, id, vec![gone]); + let notices = host.take_notices(); + assert_eq!( + notices[0] + .moved + .get(&shape()) + .map(|m| matches!(m, Moved::Keys(k) if k.len() == 1)), + Some(true) + ); + 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::Fetch { window: true, .. }), + "{:?}", + edge.reached + ); + assert_eq!(client.held(&shape()), Some(&drawn(&host, &one))); + let south = WireKey(vec![ + ("Region".into(), "South".into()), + ("_Measure".into(), "Revenue".into()), + ]); + assert!( + client + .held(&shape()) + .unwrap() + .iter() + .any(|((k, _), _)| *k == south) + ); } #[test] @@ -625,7 +736,10 @@ fn an_evicted_shape_is_announced_and_its_reader_corrects() { )); 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"); + assert!( + host.delivery().is_live(&shape()), + "naming it again revives it" + ); } #[test] @@ -640,7 +754,7 @@ fn a_shape_naming_no_view_is_answered_with_nothing_and_never_goes_live() { ask: Ask::Correction { watermark: None }, }; match host.deliver(&request, 0) { - Reply::Correction(c) => c, + Reply::Correction(c) => *c, other => panic!("{other:?}"), } }; @@ -671,7 +785,10 @@ fn a_document_swap_is_a_new_epoch_behind_every_old_watermark() { 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(); + 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); @@ -710,8 +827,16 @@ fn a_request_can_be_retried_and_answers_the_same() { 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()); + 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); } @@ -722,7 +847,15 @@ 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]; + 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)]); @@ -767,7 +900,16 @@ mod random { ] } - const VALUES: [f64; 8] = [1.0, 2.5, -0.0, 0.0, f64::INFINITY, f64::NEG_INFINITY, f64::NAN, 0.30000000000000004]; + 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 { diff --git a/crates/improvise-protocol/src/delivery/client.rs b/crates/improvise-protocol/src/delivery/client.rs index 88259a36..67ce2800 100644 --- a/crates/improvise-protocol/src/delivery/client.rs +++ b/crates/improvise-protocol/src/delivery/client.rs @@ -266,9 +266,9 @@ impl EdgeClient { 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.is_some_and(|r| { + !r.same_epoch(&c.through) || r.publication <= c.through.publication + }) { t.restructured = None; } diff --git a/crates/improvise-protocol/src/delivery/exchange.rs b/crates/improvise-protocol/src/delivery/exchange.rs index eaef636a..b8162604 100644 --- a/crates/improvise-protocol/src/delivery/exchange.rs +++ b/crates/improvise-protocol/src/delivery/exchange.rs @@ -78,7 +78,7 @@ pub struct Correction { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub enum Reply { - Correction(Correction), + Correction(Box), Notices(Vec), } diff --git a/crates/improvise-protocol/src/delivery/mod.rs b/crates/improvise-protocol/src/delivery/mod.rs index 38ef9804..6d85bee0 100644 --- a/crates/improvise-protocol/src/delivery/mod.rs +++ b/crates/improvise-protocol/src/delivery/mod.rs @@ -1,6 +1,5 @@ -//! Delivery (ADR 009 clause 9): the wire form, the stamp and notice, the correction -//! and the client end of an edge. The host end is in `improvise-command`. -//! Why: `context/code/delivery.md`. +//! Delivery (ADR 009 clause 9): the wire form, notices, corrections and the client end. +//! The host end is in `improvise-command`; why: `context/code/delivery.md`. pub mod client; pub mod exchange; diff --git a/crates/improvise-protocol/src/delivery/notice.rs b/crates/improvise-protocol/src/delivery/notice.rs index b4491dcf..616bf04d 100644 --- a/crates/improvise-protocol/src/delivery/notice.rs +++ b/crates/improvise-protocol/src/delivery/notice.rs @@ -176,10 +176,10 @@ impl Notice { /// `self` followed by `next`, if `next` starts where `self` ends. Anything else /// has a gap or another epoch in it, and merging would hide a loss. - pub fn merge(mut self, next: Notice) -> Result { + pub fn merge(mut self, next: Notice) -> Result> { let adjacent = self.range.end() == next.range.start(); match adjacent { - false => Err((self, next)), + false => Err(Box::new((self, next))), true => { self.range.to = next.range.to; self.axis.extend(next.axis); @@ -243,8 +243,14 @@ mod tests { let now = stamp(5); assert!(stamp(4).is_behind(&now)); assert!(!stamp(5).is_behind(&now)); - let other_host = Stamp { host: 8, ..stamp(99) }; - let other_generation = Stamp { generation: 2, ..stamp(99) }; + let other_host = Stamp { + host: 8, + ..stamp(99) + }; + let other_generation = Stamp { + generation: 2, + ..stamp(99) + }; assert!(other_host.is_behind(&now), "a restart is behind"); assert!(other_generation.is_behind(&now), "a swap is behind"); } @@ -264,16 +270,21 @@ mod tests { #[test] fn a_gap_does_not_merge() { - let (a, b) = notice(3, &[]).merge(notice(5, &[])).unwrap_err(); + let (a, b) = *notice(3, &[]).merge(notice(5, &[])).unwrap_err(); assert_eq!((a.range.to, b.range.to), (3, 5)); - let foreign = Notice::empty(PubRange::one(Stamp { host: 9, ..stamp(4) })); + let foreign = Notice::empty(PubRange::one(Stamp { + host: 9, + ..stamp(4) + })); assert!(notice(3, &[]).merge(foreign).is_err()); } #[test] fn merging_keeps_axis_changes_in_order_and_correct_absorbs_keys() { let mut a = notice(1, &[("s", &[1])]); - a.axis.push(AxisChange::CategoryAdded { category: "X".into() }); + a.axis.push(AxisChange::CategoryAdded { + category: "X".into(), + }); let mut b = notice(2, &[]); b.moved.insert(ShapeKey("s".into()), Moved::Correct); b.axis.push(AxisChange::ItemRemoved { @@ -291,7 +302,10 @@ mod tests { #[test] fn a_notice_round_trips_and_a_wide_stamp_crosses_as_a_string() { - let wide_host = Stamp { host: u64::MAX, ..stamp(1 << 53) }; + let wide_host = Stamp { + host: u64::MAX, + ..stamp(1 << 53) + }; let n = Notice::empty(PubRange::one(wide_host)); let json = serde_json::to_string(&n).unwrap(); assert!(json.contains(&format!("\"{}\"", u64::MAX)), "{json}");