fix(command): linear item diffs, no delivery work while nobody listens, a bounded outbox, view changes refetch the window (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 12:36:09 -07:00
committed by Edward Langley
parent f1485f80d4
commit edf474d2a3
3 changed files with 248 additions and 28 deletions
+82 -22
View File
@@ -1,7 +1,7 @@
//! 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::collections::{BTreeMap, BTreeSet, HashSet, VecDeque};
use std::sync::Arc;
use improvise_core::view::layout::GridLayout;
@@ -103,8 +103,15 @@ struct Structure {
default_view: String,
}
#[cfg(test)]
thread_local! {
pub(crate) static STRUCTURE_READS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
impl Structure {
fn of(wb: &Workbook) -> Self {
#[cfg(test)]
STRUCTURE_READS.with(|n| n.set(n.get() + 1));
let model = &wb.model;
let aggs: BTreeMap<_, _> = model.measure_agg.iter().collect();
Self {
@@ -154,7 +161,7 @@ fn axis_changes(
.map(|(at, item)| AxisChange::ItemInserted {
category: category.clone(),
item: item.clone(),
at,
at: at as u128,
});
return std::iter::once(added).chain(inserted).collect::<Vec<_>>();
};
@@ -163,26 +170,55 @@ fn axis_changes(
gone.chain(rest).collect()
}
fn item_changes(category: &str, was: &[String], now: &[String]) -> Vec<AxisChange> {
let removed = was
/// What changed between two item lists, in one pass over each: set lookups, never `contains`.
pub(crate) struct ItemDiff<'a, T> {
pub removed: Vec<&'a T>,
pub inserted: Vec<(usize, &'a T)>,
pub reordered: bool,
}
pub(crate) fn diff_items<'a, T: Eq + std::hash::Hash>(
was: &'a [T],
now: &'a [T],
) -> ItemDiff<'a, T> {
let had: HashSet<&T> = was.iter().collect();
let has: HashSet<&T> = now.iter().collect();
let removed = was.iter().filter(|i| !has.contains(i)).collect();
let inserted = now
.iter()
.filter(|i| !now.contains(i))
.enumerate()
.filter(|(_, i)| !had.contains(i))
.collect();
let kept_before = was.iter().filter(|i| has.contains(i));
let kept_after = now.iter().filter(|i| had.contains(i));
ItemDiff {
removed,
inserted,
reordered: !kept_before.eq(kept_after),
}
}
fn item_changes(category: &str, was: &[String], now: &[String]) -> Vec<AxisChange> {
if was == now {
return Vec::new();
}
let diff = diff_items(was, now);
let removed = diff
.removed
.into_iter()
.map(|item| AxisChange::ItemRemoved {
category: category.into(),
item: item.clone(),
});
let inserted = now
.iter()
.enumerate()
.filter(|(_, i)| !was.contains(i))
let inserted = diff
.inserted
.into_iter()
.map(|(at, item)| AxisChange::ItemInserted {
category: category.into(),
item: item.clone(),
at,
at: at as u128,
});
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 {
let reordered = diff.reordered.then(|| AxisChange::ItemsReordered {
category: category.into(),
items: now.to_vec(),
});
@@ -208,14 +244,15 @@ pub struct Delivery {
capacity: usize,
max_keys: usize,
live: BTreeMap<ShapeKey, Live>,
structure: Arc<Structure>,
structure: Option<Arc<Structure>>,
attached: bool,
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 {
pub fn new(host: u64, generation: u64) -> Self {
Self {
host,
generation,
@@ -224,12 +261,22 @@ impl Delivery {
capacity: LOG_CAPACITY,
max_keys: MAX_KEYS,
live: BTreeMap::new(),
structure: Arc::new(Structure::of(wb)),
structure: None,
attached: false,
evicted: BTreeSet::new(),
outbox: Vec::new(),
}
}
/// A transport is (or is not) draining notices. Without one, and with no live shape, a
/// commit only counts: nothing is read, diffed or kept.
pub fn set_attached(&mut self, on: bool, wb: &Workbook) {
self.attached = on;
if on && self.structure.is_none() {
self.structure = Some(Arc::new(Structure::of(wb)));
}
}
pub fn set_limits(&mut self, capacity: usize, max_keys: usize) {
self.capacity = capacity;
self.max_keys = max_keys;
@@ -268,13 +315,19 @@ impl Delivery {
self.log.clear();
}
self.publication += 1;
if !self.attached && self.live.is_empty() {
self.log.clear();
self.structure = None;
self.evicted.clear();
return self.stamp();
}
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),
let (axis, flags) = match (new_epoch, self.structure.as_deref()) {
(false, Some(before)) => structure.since(before),
_ => (Vec::new(), all_flags()),
};
self.structure = structure;
self.structure = Some(structure);
let mut notice = Notice::empty(range);
notice.axis = axis;
@@ -302,7 +355,11 @@ impl Delivery {
while self.log.len() > self.capacity {
self.log.pop_front();
}
self.outbox.push(notice);
if self.attached {
self.outbox.push(notice);
let excess = self.outbox.len().saturating_sub(self.capacity);
self.outbox.drain(..excess);
}
self.stamp()
}
@@ -312,6 +369,9 @@ impl Delivery {
Some(live) => live.last_touch = now,
None => {
let Some(view) = view_of(shape) else { return };
if self.structure.is_none() {
self.structure = Some(Arc::new(Structure::of(wb)));
}
let Some(output) = Output::of(wb, view) else {
return;
};
@@ -437,7 +497,7 @@ impl Delivery {
.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());
.any(|n| !n.axis.is_empty() || !n.relabels.is_empty() || n.flags.views);
if oversized {
return Reply::Correction(Box::new(self.window(request, None, self.stamp(), None)));
}
@@ -43,7 +43,9 @@ fn workbook() -> Workbook {
fn host_instance(instance: u64) -> Host {
let mut app = App::new(workbook(), None);
app.rebuild_layout();
Host::with_instance(app, instance)
let mut host = Host::with_instance(app, instance);
host.attach_notices();
host
}
fn host() -> Host {
@@ -776,6 +778,7 @@ fn a_document_swap_is_a_new_epoch_behind_every_old_watermark() {
app.storage = Box::new(storage);
app.rebuild_layout();
let mut host = Host::with_instance(app, 1);
host.attach_notices();
let id = connect(&mut host);
let mut client = settled(&mut host);
let old = client.watermark(&shape()).unwrap();
@@ -876,6 +879,161 @@ fn values_keep_every_bit_from_the_cell_to_the_client() {
}
}
// ── cost: what a commit does when nobody is listening ─────────────────────
/// An item whose every comparison is counted, so the cost is a number and not a clock.
#[derive(Clone, Debug)]
struct Counted(usize);
thread_local! {
static COMPARISONS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
impl PartialEq for Counted {
fn eq(&self, o: &Self) -> bool {
COMPARISONS.with(|c| c.set(c.get() + 1));
self.0 == o.0
}
}
impl Eq for Counted {}
impl std::hash::Hash for Counted {
fn hash<H: std::hash::Hasher>(&self, h: &mut H) {
self.0.hash(h);
}
}
#[test]
fn diffing_a_category_compares_each_item_a_bounded_number_of_times() {
use super::delivery::diff_items;
for n in [1000usize, 2000, 4000] {
let was: Vec<Counted> = (0..n).map(Counted).collect();
let mut now = was.clone();
now.remove(n / 2);
now.push(Counted(n + 1));
COMPARISONS.with(|c| c.set(0));
let diff = diff_items(&was, &now);
let comparisons = COMPARISONS.with(|c| c.get());
assert_eq!(diff.removed.len() + diff.inserted.len(), 2);
assert!(
comparisons <= 8 * n,
"{n} items cost {comparisons} comparisons"
);
}
}
#[test]
fn a_commit_with_nothing_live_and_no_transport_reads_nothing() {
use super::delivery::STRUCTURE_READS;
let mut app = App::new(workbook(), None);
app.rebuild_layout();
let mut host = Host::with_instance(app, 1);
let id = connect(&mut host);
STRUCTURE_READS.with(|c| c.set(0));
for n in 0..3 {
send(&mut host, id, vec![set("North", n as f64)]);
}
assert_eq!(STRUCTURE_READS.with(|c| c.get()), 0);
assert_eq!(
host.delivery().stamp().publication,
3,
"the count still moves"
);
assert_eq!(host.delivery().log_len(), 0);
assert!(host.take_notices().is_empty());
}
#[test]
fn a_shape_that_goes_live_after_idle_commits_still_hears_the_axis_changes() {
let mut app = App::new(workbook(), None);
app.rebuild_layout();
let mut host = Host::with_instance(app, 1);
let id = connect(&mut host);
send(&mut host, id, vec![set("North", 1.0)]);
let mut client = settled(&mut host);
host.attach_notices();
send(&mut host, id, vec![add_item("East")]);
let notices = host.take_notices();
assert_eq!(notices[0].axis.len(), 1, "{:?}", notices[0]);
deliver_all(&mut client, &notices);
client.sync(&mut Flaky::new(&mut host), &shape(), 6);
held_equals_drawn(&client, &host);
}
#[test]
fn the_outbox_and_the_clone_a_frame_takes_are_bounded() {
let mut host = host();
host.limit_delivery(4, 4096);
let id = connect(&mut host);
let mut client = settled(&mut host);
for n in 0..20 {
send(&mut host, id, vec![set("North", n as f64)]);
}
let queued = host.take_notices();
assert_eq!(
queued.len(),
4,
"a transport that never drains keeps the newest four"
);
deliver_all(&mut client, &queued);
let mut edge = Flaky::new(&mut host);
client.sync(&mut edge, &shape(), 6);
assert!(
matches!(asks(&edge.reached)[0], Ask::Correction { .. }),
"the gap is seen"
);
held_equals_drawn(&client, &host);
}
// ── view geometry moves positions without moving a key ────────────────────
fn transpose() -> ModelCommand {
ModelCommand::TransposeAxes {
view: "Default".into(),
}
}
fn south_window() -> Window {
Window {
top: 1,
left: 0,
rows: 1,
cols: 1,
}
}
#[test]
fn a_transpose_empties_a_partial_window_of_the_cell_that_left_it() {
let mut host = host();
let id = connect(&mut host);
let mut client = EdgeClient::new(session());
client.track(shape(), south_window());
client.sync(&mut Flaky::new(&mut host), &shape(), 6);
assert_eq!(client.held(&shape()).map(|h| h.len()), Some(1));
host.take_notices();
send(&mut host, id, vec![transpose()]);
let notices = host.take_notices();
assert!(notices[0].flags.views);
deliver_all(&mut client, &notices);
client.sync(&mut Flaky::new(&mut host), &shape(), 6);
assert_eq!(client.held(&shape()), Some(&drawn(&host, &south_window())));
assert_eq!(client.held(&shape()).map(|h| h.len()), Some(0));
}
#[test]
fn a_dropped_views_notice_is_answered_with_the_window_not_an_empty_key_list() {
let mut host = host();
let id = connect(&mut host);
let mut client = EdgeClient::new(session());
client.track(shape(), south_window());
client.sync(&mut Flaky::new(&mut host), &shape(), 6);
send(&mut host, id, vec![transpose()]);
client.connection_lost();
client.sync(&mut Flaky::new(&mut host), &shape(), 6);
assert_eq!(client.held(&shape()), Some(&drawn(&host, &south_window())));
}
// ── the whole thing, at random ─────────────────────────────────────────────
mod random {
+7 -5
View File
@@ -39,11 +39,7 @@ impl Host {
/// 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(),
);
let delivery = Delivery::new(instance, app.document_generation());
Self {
app,
sessions: Sessions::default(),
@@ -71,6 +67,12 @@ impl Host {
self.delivery.set_limits(capacity, max_keys);
}
/// A transport will drain `take_notices`; until then nothing is queued for it.
pub fn attach_notices(&mut self) {
self.delivery
.set_attached(true, &self.app.model_state.workbook);
}
/// Notices published since the last call, for the transport to push.
pub fn take_notices(&mut self) -> Vec<Notice> {
self.delivery.take_outbox()