feat(command): the host end of delivery, one notice per committed command, with the log and corrections (improvise-z2vr)

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 11:59:30 -07:00
committed by Edward Langley
parent a4c309642f
commit 19cea8eded
9 changed files with 1480 additions and 28 deletions
Generated
+2
View File
@@ -909,11 +909,13 @@ dependencies = [
"improvise-gp",
"improvise-io",
"improvise-protocol",
"proptest",
"serde",
"serde_json",
"tempfile",
"unicode-width",
"web-time",
"zset",
]
[[package]]
+2
View File
@@ -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"
+505
View File
@@ -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<WireKey, Placed>,
}
impl Output {
pub fn of(wb: &Workbook, view: &str) -> Option<Output> {
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<WireKey>>) -> ZSet<Cell> {
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<WireKey> {
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<String>)>,
formulas: Vec<String>,
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<AxisChange>, 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<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
.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::<Vec<_>>();
};
item_changes(category, was, items)
});
gone.chain(rest).collect()
}
fn item_changes(category: &str, was: &[String], now: &[String]) -> Vec<AxisChange> {
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<Output>,
}
/// 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<Arc<Notice>>,
capacity: usize,
max_keys: usize,
live: BTreeMap<ShapeKey, Live>,
structure: Arc<Structure>,
evicted: BTreeSet<ShapeKey>,
outbox: Vec<Arc<Notice>>,
}
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<Notice> {
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(&notice));
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<ShapeKey> {
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<Stamp> {
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<Stamp>) -> Vec<Notice> {
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<Stamp>,
through: Stamp,
carried: Option<Notice>,
) -> 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<WireKey>>,
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<WireKey> = 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::<Vec<_>>(),
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<Reply, EdgeError> {
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<E>(pub E);
impl<E: Edge> Edge for WireEdge<E> {
fn call(&mut self, request: Request) -> Result<Reply, EdgeError> {
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<Notice, EdgeError> {
let sent = serde_json::to_string(notice).map_err(|_| EdgeError::Unreadable)?;
serde_json::from_str(&sent).map_err(|_| EdgeError::Unreadable)
}
@@ -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<ModelCommand>) -> ClientFrame {
let mut wire: Vec<improvise_protocol::Command> = 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<ModelCommand>) -> Vec<ServerFrame> {
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<Cell> {
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<Option<Fault>>,
reached: Vec<Request>,
}
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<Option<Fault>>) -> Self {
Self {
script: script.into(),
..Self::new(host)
}
}
}
impl Edge for Flaky<'_> {
fn call(&mut self, request: Request) -> Result<Reply, EdgeError> {
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<SessionToken> {
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(&notice_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<u64> = 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, &notices);
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, &notices[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, &notices);
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, &notices);
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<Value = Op> {
prop_oneof![
4 => (any::<bool>(), 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<Option<u8>>,
}
fn weather() -> impl Strategy<Value = Weather> {
(
any::<bool>(),
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<Value = Window> {
(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, &notices);
}
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);
}
}
}
+59 -5
View File
@@ -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<ShapeKey> {
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<Notice> {
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<Addressed> {
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<Addressed> {
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<Addressed> {
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
+4
View File
@@ -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;
@@ -29,6 +29,7 @@ pub enum Want {
watermark: Option<Stamp>,
keys: BTreeSet<WireKey>,
through: Stamp,
window: bool,
},
}
@@ -60,6 +61,8 @@ struct Tracked {
seen: Option<Stamp>,
/// 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<Stamp>,
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<Want> {
let t = self.shapes.get(shape)?;
let needs_all = t.stale || t.unapplied.iter().any(|(_, m)| *m == Moved::Correct);
let keys: BTreeSet<WireKey> = t
.unapplied
.iter()
.flat_map(|(_, m)| match m {
Moved::Keys(keys) => keys.iter().cloned().collect::<Vec<_>>(),
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 += &delta;
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)
}
@@ -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
})
);
}
@@ -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<Stamp> },
/// 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<Stamp>,
keys: BTreeSet<WireKey>,
through: Stamp,
window: bool,
},
/// The notices after `after`, if the host's log still has them.
Notices { after: Stamp },