fiddlerwoaroof/improvise
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(protocol): box the large delivery variants, fmt and the axis-change fetch (improvise-z2vr)

Browse Source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze
This commit is contained in:
Claude (dbsp-delivery)
2026-10-10 12:11:48 -07:00
committed by Edward Langley
parent e3f5cee29a
commit 0e6ad9e96a
6 changed files with 238 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/improvise-command/src/ui/delivery.rs
+24 -16
View File
@@ -131,7 +131,10 @@ impl Structure {
}
}
fn axis_changes(before: &[(String, Vec<String>)], after: &[(String, Vec<String>)]) -> Vec<AxisChange> {
fn axis_changes(
before: &[(String, Vec<String>)],
after: &[(String, Vec<String>)],
) -> Vec<AxisChange> {
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<String>)], after: &[(String, Vec<String>)
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::<Vec<_>>();
};
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<WireKey> = 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,
})
}))
}
}
crates/improvise-command/src/ui/delivery_tests.rs
+186 -44
View File
@@ -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<ModelCommand>) -> Vec<ServerFr
}
fn refused(answer: &[ServerFrame]) -> 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, &notices);
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 {
crates/improvise-protocol/src/delivery/client.rs
+3 -3
View File
@@ -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;
}
crates/improvise-protocol/src/delivery/exchange.rs
+1 -1
View File
@@ -78,7 +78,7 @@ pub struct Correction {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum Reply {
Correction(Correction),
Correction(Box<Correction>),
Notices(Vec<Notice>),
}
crates/improvise-protocol/src/delivery/mod.rs
+2 -3
View File
@@ -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;
crates/improvise-protocol/src/delivery/notice.rs
+22 -8
View File
@@ -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<Notice, (Notice, Notice)> {
pub fn merge(mut self, next: Notice) -> Result<Notice, Box<(Notice, Notice)>> {
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}");