Merge dbsp/step5-formulas into dbsp/step6-exactsum

The sign rule FloatSum wrote is in ExactSum on gh/main now (PR 348), so
step 6 sums through ExactSum and FloatSum and its test are dropped.

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-backend)
2026-10-10 15:24:58 -07:00
committed by Edward Langley
35 changed files with 4324 additions and 180 deletions
+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" }
# The cell delta behind the record tier and the dirty flag (ADR 009 step 3).
improvise-incremental = { path = "../improvise-incremental" }
anyhow = "1"
@@ -28,3 +29,4 @@ web-time = "1"
improvise-core = { path = "../improvise-core", features = ["test-support"] }
improvise-protocol = { path = "../improvise-protocol", features = ["test-support"] }
tempfile = "3"
proptest = "1"
+573
View File
@@ -0,0 +1,573 @@
//! 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, HashSet, 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,
}
#[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 {
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: at as u128,
});
return std::iter::once(added).chain(inserted).collect::<Vec<_>>();
};
item_changes(category, was, items)
});
gone.chain(rest).collect()
}
/// 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()
.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 = diff
.inserted
.into_iter()
.map(|(at, item)| AxisChange::ItemInserted {
category: category.into(),
item: item.clone(),
at: at as u128,
});
let reordered = diff.reordered.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: 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, generation: u64) -> Self {
Self {
host,
generation,
publication: 0,
log: VecDeque::new(),
capacity: LOG_CAPACITY,
max_keys: MAX_KEYS,
live: BTreeMap::new(),
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;
}
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;
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, self.structure.as_deref()) {
(false, Some(before)) => structure.since(before),
_ => (Vec::new(), all_flags()),
};
self.structure = Some(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();
}
if self.attached {
self.outbox.push(notice);
let excess = self.outbox.len().saturating_sub(self.capacity);
self.outbox.drain(..excess);
}
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 };
if self.structure.is_none() {
self.structure = Some(Arc::new(Structure::of(wb)));
}
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(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(Box::new(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(Box::new(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(|e| e.1));
let restructured = covering
.iter()
.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)));
}
if window || restructured {
return Reply::Correction(Box::new(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(Box::new(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)
}
File diff suppressed because it is too large Load Diff
+61 -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,60 @@ 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.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);
}
/// 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()
}
pub fn app(&self) -> &App {
&self.app
}
@@ -224,11 +270,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 +299,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 +316,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 +331,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;
+3 -3
View File
@@ -15,10 +15,10 @@ pub const MAX_CATEGORIES: usize = 24;
/// They must agree; why, and why 1e-10: context/code/core.md §4.
const FLOAT_EQ_EPSILON: f64 = 1e-10;
/// The sum correctly rounded once, whatever the order (ADR 009 clause 8; `-0.0` only if every
/// addend is, as `Sum for f64` gives).
/// The sum correctly rounded once, whatever the order (ADR 009 clause 8), signed as IEEE
/// signs a zero total.
fn exact_sum(values: &[f64]) -> f64 {
let mut sum = numtower::FloatSum::new();
let mut sum = numtower::ExactSum::new();
values.iter().for_each(|v| sum.add(*v, 1));
sum.value()
}
+4 -4
View File
@@ -4,7 +4,7 @@
use crate::DeltaValue;
use improvise_core::formula::ast::AggFunc;
use improvise_core::model::cell::CellValue;
use numtower::FloatSum;
use numtower::ExactSum;
use std::collections::BTreeMap;
/// A group's cells: how many, how many are numbers, a multiset of their values and their exact
@@ -14,7 +14,7 @@ pub struct Group {
cells: i64,
numbers: i64,
values: BTreeMap<DeltaValue, i64>,
sum: FloatSum,
sum: ExactSum,
}
impl Group {
@@ -127,8 +127,8 @@ fn reduce(groups: &[&Group], func: Reduce, cells: i64) -> f64 {
}
/// The groups' sums, absorbed into one: exact, so the same in any order and for any grouping.
fn total(groups: &[&Group]) -> FloatSum {
let mut sum = FloatSum::new();
fn total(groups: &[&Group]) -> ExactSum {
let mut sum = ExactSum::new();
groups.iter().for_each(|g| sum.absorb(&g.sum));
sum
}
@@ -224,6 +224,7 @@ impl Incremental {
/// cells, recomputes every formula cell as eager does.
fn evaluate(&mut self, stems: impl IntoIterator<Item = CellKey>) {
if self.stems.is_wide() {
self.work.recomputes += 1;
return self.model.recompute_formulas(&self.none_cats);
}
for stem in stems {
@@ -24,6 +24,9 @@ pub struct Meter {
pub formula_cells: u64,
/// Times the stems were rebuilt from the store, as a formula edit does.
pub rebuilds: u64,
/// Whole-cache recomputes by the fallback for documents whose stems read each other. One is
/// every stem's formulas for every pass, so it is counted as an event, not as cells.
pub recomputes: u64,
}
impl Meter {
@@ -36,6 +39,7 @@ impl Meter {
stems: self.stems + o.stems,
formula_cells: self.formula_cells + o.formula_cells,
rebuilds: self.rebuilds + o.rebuilds,
recomputes: self.recomputes + o.recomputes,
}
}
}
+2 -2
View File
@@ -5,7 +5,7 @@ use improvise_core::formula::ast::AggFunc;
use improvise_core::model::Model;
use improvise_core::model::cell::{CellKey, CellValue};
use improvise_incremental::{Derived, Eager, Edit, Incremental, Maintained};
use numtower::FloatSum;
use numtower::ExactSum;
fn key(i: usize) -> CellKey {
CellKey::new(vec![
@@ -34,7 +34,7 @@ fn model_of(values: &[f64], func: AggFunc) -> Model {
}
fn exact(values: &[f64]) -> f64 {
let mut sum = FloatSum::new();
let mut sum = ExactSum::new();
values.iter().for_each(|v| sum.add(*v, 1));
sum.value()
}
@@ -191,6 +191,33 @@ fn two_target_categories_read_each_others_stems_whatever_the_order_they_are_adde
}
}
#[test]
fn the_whole_cache_recompute_of_a_wide_document_is_metered() {
let wb = formula_workbook();
let mut backend = Incremental::new(wb.model.clone(), none_of(&wb));
let edit = bump("East", "A", "Revenue", 3.0);
backend.apply(&edit);
assert_eq!(
backend.meter().recomputes,
0,
"one target category: by stem"
);
backend.apply(&formula_edit(14));
let before = backend.meter();
backend.apply(&edit);
backend.apply(&bump("East", "A", "Revenue", 4.0));
let spent = backend.meter();
assert_eq!(
spent.recomputes - before.recomputes,
2,
"Region and _Measure targets: wide"
);
assert_eq!(
spent.stems, before.stems,
"and no stem is evaluated on its own"
);
}
#[test]
fn a_name_in_another_category_reads_another_stems_cell() {
let mut wb = formula_workbook();
@@ -67,6 +67,7 @@ fn since(before: Meter, after: Meter) -> Meter {
stems: after.stems - before.stems,
formula_cells: after.formula_cells - before.formula_cells,
rebuilds: after.rebuilds - before.rebuilds,
recomputes: after.recomputes - before.recomputes,
}
}
+4
View File
@@ -8,6 +8,9 @@ repository = "https://github.com/fiddlerwoaroof/improvise"
[dependencies]
improvise-core = { path = "../improvise-core" }
# ADR 009 clause 9: the wire form carries a correction as Z-set arithmetic, a value as ADR 005 canonical text.
zset = { path = "../zset" }
numtower = { path = "../numtower" }
serde = { version = "1", features = ["derive"] }
# For the one width rule both front ends measure a header label with
# (`AxisFrame::header_widths`). A terminal column is a display cell, not a
@@ -24,3 +27,4 @@ test-support = ["dep:serde_json"]
[dev-dependencies]
serde_json = "1"
proptest = "1"
@@ -0,0 +1,404 @@
//! The client end of an edge: watermarks, the held window, and when to ask (ADR 009 clause 9).
//! Why each choice: `context/code/delivery.md` §4.
use std::collections::{BTreeMap, BTreeSet};
use zset::ZSet;
use super::exchange::{Ask, Correction, CorrectionBody, Edge, EdgeError, Reply, Request};
use super::exchange::{SessionToken, Window};
use super::notice::{AxisChange, Flags, Moved, Notice, Relabel, ShapeKey, Stamp};
use super::wire::{Cell, WireKey, decode};
/// What the shape's consumer must do with the structure around the cells.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Event {
Axis(AxisChange),
Relabel(Relabel),
Flags(Flags),
/// Positions and labels are unknown: re-derive them from what is held.
Resync,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Want {
Correction {
watermark: Option<Stamp>,
},
Fetch {
watermark: Option<Stamp>,
keys: BTreeSet<WireKey>,
through: Stamp,
window: bool,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Outcome {
/// Applied; `delta` is what changed in the held cells.
Applied { delta: ZSet<Cell> },
/// Already covered by the watermark: nothing was done.
Duplicate,
/// Starts after the watermark, or in another epoch: ask again from the watermark.
Gap,
/// Names a shape this client does not read.
Untracked,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Synced {
Settled,
Failed,
/// The rounds ran out with work left: a defect in the edge, not a state to wait out.
Exhausted,
}
/// How many abandoned epochs a shape remembers.
const LEFT_EPOCHS: usize = 64;
#[derive(Debug, Clone)]
struct Tracked {
window: Window,
held: ZSet<Cell>,
watermark: Option<Stamp>,
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>,
/// Epochs the watermark has moved out of: nothing from them applies again.
left: Vec<(u64, u64)>,
stale: bool,
}
#[derive(Debug, Clone, Default)]
pub struct EdgeClient {
session: Option<SessionToken>,
shapes: BTreeMap<ShapeKey, Tracked>,
events_through: Option<Stamp>,
}
impl EdgeClient {
pub fn new(session: Option<SessionToken>) -> Self {
Self {
session,
..Self::default()
}
}
/// Start reading `shape`; the first thing it wants is a full refetch.
pub fn track(&mut self, shape: ShapeKey, window: Window) {
self.shapes.insert(
shape,
Tracked {
window,
held: ZSet::new(),
watermark: None,
seen: None,
unapplied: Vec::new(),
restructured: None,
left: Vec::new(),
stale: true,
},
);
}
pub fn held(&self, shape: &ShapeKey) -> Option<&ZSet<Cell>> {
self.shapes.get(shape).map(|t| &t.held)
}
pub fn watermark(&self, shape: &ShapeKey) -> Option<Stamp> {
self.shapes.get(shape).and_then(|t| t.watermark)
}
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 owed => Some(Want::Fetch {
watermark: t.watermark,
keys,
through,
window: t.restructured.is_some(),
}),
_ => None,
}
}
pub fn request(&self, shape: &ShapeKey, want: Want) -> Option<Request> {
let t = self.shapes.get(shape)?;
Some(Request {
session: self.session.clone(),
shape: shape.clone(),
window: t.window,
ask: match want {
Want::Correction { watermark } => Ask::Correction { watermark },
Want::Fetch {
watermark,
keys,
through,
window,
} => Ask::Fetch {
watermark,
keys,
through,
window,
},
},
})
}
/// A pushed notice. Duplicates are ignored, a gap or a new epoch makes every shape ask.
pub fn on_notice(&mut self, notice: &Notice) -> Vec<Event> {
self.accept(notice, None)
}
/// `only` limits the notice to one shape: a pulled notice carries no other shape's keys.
fn accept(&mut self, notice: &Notice, only: Option<&ShapeKey>) -> Vec<Event> {
let end = notice.stamp();
for (shape, t) in &mut self.shapes {
if only.is_some_and(|o| o != shape) {
continue;
}
let contiguous = match t.seen {
Some(seen) if seen.same_epoch(&end) => match notice.range.to <= seen.publication {
true => continue,
false => notice.range.from <= seen.publication,
},
_ => false,
};
t.seen = Some(end);
if !contiguous || notice.evicted.contains(shape) {
t.unapplied.push((end, Moved::Correct));
continue;
}
if !notice.axis.is_empty() || !notice.relabels.is_empty() || notice.flags.views {
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.restructured.is_none() && !t.stale && current {
t.watermark = Some(end);
}
}
self.events_of(notice)
}
/// The connection dropped: every shape asks for a correction from its watermark.
pub fn connection_lost(&mut self) {
self.shapes.values_mut().for_each(|t| t.stale = true);
}
pub fn request_failed(&mut self, shape: &ShapeKey) {
if let Some(t) = self.shapes.get_mut(shape) {
t.stale = true;
}
}
/// A notice that could not be read: what it said is lost, so everything asks.
pub fn unreadable(&mut self) {
self.connection_lost();
self.events_through = None;
}
fn events_of(&mut self, n: &Notice) -> Vec<Event> {
let end = n.stamp();
let resync = vec![Event::Resync];
let usable = match self.events_through {
Some(t) if t.same_epoch(&end) => {
if n.range.to <= t.publication {
return Vec::new();
}
n.range.from == t.publication
}
_ => false,
};
self.events_through = Some(end);
match usable {
false => resync,
true => n
.axis
.iter()
.cloned()
.map(Event::Axis)
.chain(n.relabels.iter().cloned().map(Event::Relabel))
.chain(n.flags.any().then_some(Event::Flags(n.flags)))
.collect(),
}
}
/// A correction. It applies only at the watermark it starts from: `held + (target - held)`.
pub fn apply(&mut self, c: &Correction) -> (Outcome, Vec<Event>) {
let Some(t) = self.shapes.get_mut(&c.shape) else {
return (Outcome::Untracked, Vec::new());
};
let target = decode(&c.target);
let older = |s: Option<Stamp>| {
s.is_some_and(|s| s.host == c.through.host && c.through.generation < s.generation)
};
if older(t.watermark)
|| older(t.seen)
|| t.left.contains(&(c.through.host, c.through.generation))
{
return (Outcome::Duplicate, Vec::new());
}
let covered = t
.watermark
.is_some_and(|w| w.same_epoch(&c.through) && w.publication >= c.through.publication);
let starts_here = match (&c.from, t.watermark) {
(Some(from), Some(w)) => w.same_epoch(from) && w.publication >= from.publication,
_ => false,
};
let delta = match (&c.body, covered) {
(_, true) => {
t.stale &= t.watermark != Some(c.through);
let live = c.through;
t.unapplied
.retain(|(at, _)| at.same_epoch(&live) && at.publication > live.publication);
if t.restructured.is_some_and(|r| !r.same_epoch(&live)) {
t.restructured = None;
}
if t.seen.is_some_and(|s| !s.same_epoch(&live)) {
t.seen = Some(live);
}
return (Outcome::Duplicate, Vec::new());
}
(CorrectionBody::Window, false) => &target - &t.held,
(CorrectionBody::Moved { .. }, false) if !starts_here => {
return (Outcome::Gap, Vec::new());
}
(CorrectionBody::Moved { keys }, false) => {
let held_there: ZSet<Cell> = t
.held
.iter()
.filter(|((k, _), _)| keys.contains(k))
.map(|(cell, w)| (cell.clone(), w))
.collect();
&target - &held_there
}
};
t.held += &delta;
if let Some(w) = t.watermark.filter(|w| !w.same_epoch(&c.through)) {
t.left.push((w.host, w.generation));
let excess = t.left.len().saturating_sub(LEFT_EPOCHS);
t.left.drain(..excess);
}
t.watermark = Some(c.through);
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
&& !(at.same_epoch(&c.through) && at.publication <= c.through.publication)
});
let seen_ahead = t
.seen
.is_some_and(|s| s.same_epoch(&c.through) && s.publication >= c.through.publication);
if !seen_ahead {
t.seen = Some(c.through);
}
let events = match (&c.carried, window) {
(Some(carried), _) => self.events_of(carried),
(None, true) => {
self.events_through = Some(c.through);
vec![Event::Resync]
}
(None, false) => Vec::new(),
};
(Outcome::Applied { delta }, events)
}
/// Ask until `shape` wants nothing, at most `rounds` times.
pub fn sync(
&mut self,
edge: &mut impl Edge,
shape: &ShapeKey,
rounds: usize,
) -> (Synced, Vec<Event>) {
let mut events = Vec::new();
for _ in 0..rounds {
let Some(want) = self.wants(shape) else {
return (Synced::Settled, events);
};
let Some(request) = self.request(shape, want) else {
return (Synced::Settled, events);
};
match edge.call(request) {
Ok(Reply::Correction(c)) => events.extend(self.apply(&c).1),
Ok(Reply::Notices(list)) => {
list.iter()
.for_each(|n| events.extend(self.accept(n, Some(shape))));
}
Err(EdgeError::Failed) => {
self.request_failed(shape);
return (Synced::Failed, events);
}
Err(EdgeError::Unreadable) => {
self.unreadable();
return (Synced::Failed, events);
}
}
}
match self.wants(shape) {
None => (Synced::Settled, events),
Some(_) => (Synced::Exhausted, events),
}
}
/// Pull instead of listening: ask for the notices after the horizon seen.
pub fn poll(&mut self, edge: &mut impl Edge, shape: &ShapeKey) -> (Synced, Vec<Event>) {
let (Some(t), Some(after)) = (
self.shapes.get(shape),
self.shapes.get(shape).and_then(|t| t.seen),
) else {
return self.sync(edge, shape, 4);
};
let request = Request {
session: self.session.clone(),
shape: shape.clone(),
window: t.window,
ask: Ask::Notices { after },
};
let mut events = Vec::new();
match edge.call(request) {
Ok(Reply::Notices(list)) => list
.iter()
.for_each(|n| events.extend(self.accept(n, Some(shape)))),
Ok(Reply::Correction(c)) => events.extend(self.apply(&c).1),
Err(EdgeError::Failed) => {
self.request_failed(shape);
return (Synced::Failed, events);
}
Err(EdgeError::Unreadable) => {
self.unreadable();
return (Synced::Failed, events);
}
}
let (synced, more) = self.sync(edge, shape, 4);
events.extend(more);
(synced, events)
}
}
@@ -0,0 +1,292 @@
//! The client end of an edge against hand-built notices and corrections.
use super::client::*;
use super::exchange::*;
use super::notice::*;
use super::wire::{Cell, WireElem, WireKey, WireValue, decode, encode};
fn at(host: u64, publication: u64) -> Stamp {
Stamp {
host,
generation: 1,
publication,
}
}
fn shape() -> ShapeKey {
ShapeKey("s".into())
}
fn window() -> Window {
Window {
top: 0,
left: 0,
rows: 10,
cols: 10,
}
}
fn key(n: u8) -> WireKey {
WireKey(vec![("C".into(), n.to_string())])
}
fn cell(k: u8, v: &str) -> Cell {
(key(k), WireValue::Text(v.into()))
}
fn target(cells: &[Cell]) -> Vec<WireElem> {
encode(&cells.iter().cloned().map(|c| (c, 1)).collect())
}
fn window_correction(through: Stamp, cells: &[Cell]) -> Correction {
Correction {
shape: shape(),
at: through,
from: None,
through,
body: CorrectionBody::Window,
target: target(cells),
carried: None,
}
}
fn notice(host: u64, publication: u64, moved: &[u8]) -> Notice {
let mut n = Notice::empty(PubRange::one(at(host, publication)));
if !moved.is_empty() {
n.moved.insert(
shape(),
Moved::Keys(moved.iter().map(|k| key(*k)).collect()),
);
}
n
}
fn synced_at(publication: u64, cells: &[Cell]) -> EdgeClient {
let mut c = EdgeClient::default();
c.track(shape(), window());
c.apply(&window_correction(at(1, publication), cells));
assert_eq!(c.wants(&shape()), None, "the fixture must start settled");
c
}
fn moved_correction(from: u64, through: u64, keys: &[u8], cells: &[Cell]) -> Correction {
Correction {
shape: shape(),
at: at(1, through),
from: Some(at(1, from)),
through: at(1, through),
body: CorrectionBody::Moved {
keys: keys.iter().map(|k| key(*k)).collect(),
},
target: target(cells),
carried: None,
}
}
#[test]
fn a_tracked_shape_first_wants_a_full_refetch() {
let mut c = EdgeClient::default();
c.track(shape(), window());
assert_eq!(
c.wants(&shape()),
Some(Want::Correction { watermark: None })
);
}
#[test]
fn consecutive_notices_ask_for_no_correction_and_advance_the_watermark() {
let mut c = synced_at(4, &[cell(1, "a")]);
c.on_notice(&notice(1, 5, &[]));
c.on_notice(&notice(1, 6, &[]));
assert_eq!(c.wants(&shape()), None);
assert_eq!(c.watermark(&shape()), Some(at(1, 6)));
}
#[test]
fn a_moved_key_wants_a_fetch_and_not_a_correction() {
let mut c = synced_at(4, &[cell(1, "a")]);
c.on_notice(&notice(1, 5, &[1, 2]));
assert_eq!(
c.wants(&shape()),
Some(Want::Fetch {
watermark: Some(at(1, 4)),
keys: [key(1), key(2)].into(),
through: at(1, 5),
window: false
})
);
}
#[test]
fn a_dropped_notice_is_detected() {
let mut c = synced_at(4, &[cell(1, "a")]);
c.on_notice(&notice(1, 6, &[]));
assert_eq!(
c.wants(&shape()),
Some(Want::Correction {
watermark: Some(at(1, 4))
})
);
}
#[test]
fn a_notice_from_another_host_or_generation_asks_for_a_correction() {
for foreign in [
notice(2, 5, &[]),
Notice::empty(PubRange::one(Stamp {
generation: 2,
..at(1, 5)
})),
] {
let mut c = synced_at(4, &[cell(1, "a")]);
c.on_notice(&foreign);
assert!(
matches!(c.wants(&shape()), Some(Want::Correction { .. })),
"{foreign:?}"
);
}
}
#[test]
fn an_eviction_and_an_oversized_notice_each_ask_for_a_correction() {
let mut evicted = notice(1, 5, &[]);
evicted.evicted.insert(shape());
let mut oversized = notice(1, 5, &[]);
oversized.moved.insert(shape(), Moved::Correct);
for n in [evicted, oversized] {
let mut c = synced_at(4, &[cell(1, "a")]);
c.on_notice(&n);
assert!(matches!(c.wants(&shape()), Some(Want::Correction { .. })));
}
}
#[test]
fn a_duplicate_notice_changes_nothing() {
let mut c = synced_at(4, &[]);
c.on_notice(&notice(1, 5, &[3]));
let before = c.wants(&shape());
assert_eq!(c.on_notice(&notice(1, 5, &[3])), Vec::new());
assert_eq!(c.wants(&shape()), before);
}
#[test]
fn a_correction_equals_the_target_and_a_second_one_is_the_zero_zset() {
let mut c = EdgeClient::default();
c.track(shape(), window());
let fix = window_correction(at(1, 4), &[cell(1, "a"), cell(2, "b")]);
let (first, _) = c.apply(&fix);
let Outcome::Applied { delta } = first else {
panic!("{first:?}")
};
assert_eq!(delta.len(), 2);
assert_eq!(c.held(&shape()), Some(&decode(&fix.target)));
let later = window_correction(at(1, 5), &[cell(1, "a"), cell(2, "b")]);
let (second, _) = c.apply(&later);
let Outcome::Applied { delta } = second else {
panic!("{second:?}")
};
assert!(delta.is_zero(), "correct state corrected: {delta:?}");
}
#[test]
fn a_correction_replaces_what_is_held_not_adds_to_it() {
let mut c = synced_at(4, &[cell(1, "a"), cell(2, "b")]);
c.apply(&window_correction(at(1, 5), &[cell(1, "z")]));
assert_eq!(c.held(&shape()), Some(&decode(&target(&[cell(1, "z")]))));
}
#[test]
fn a_correction_already_covered_is_idempotent_and_never_regresses() {
let mut c = synced_at(4, &[cell(1, "a")]);
c.apply(&window_correction(at(1, 7), &[cell(1, "new")]));
let late = window_correction(at(1, 5), &[cell(1, "old")]);
assert_eq!(c.apply(&late).0, Outcome::Duplicate);
assert_eq!(c.held(&shape()), Some(&decode(&target(&[cell(1, "new")]))));
assert_eq!(c.watermark(&shape()), Some(at(1, 7)));
}
#[test]
fn a_moved_correction_applies_only_at_the_watermark_it_starts_from() {
let mut c = synced_at(4, &[cell(1, "a"), cell(2, "b")]);
let ahead = moved_correction(6, 8, &[1], &[cell(1, "x")]);
assert_eq!(c.apply(&ahead).0, Outcome::Gap);
assert_eq!(c.watermark(&shape()), Some(at(1, 4)));
let fits = moved_correction(4, 6, &[1], &[cell(1, "x")]);
assert!(matches!(c.apply(&fits).0, Outcome::Applied { .. }));
assert_eq!(c.apply(&fits).0, Outcome::Duplicate);
let expect = decode(&target(&[cell(1, "x"), cell(2, "b")]));
assert_eq!(c.held(&shape()), Some(&expect));
assert_eq!(c.watermark(&shape()), Some(at(1, 6)));
}
#[test]
fn a_moved_key_the_target_no_longer_holds_is_removed() {
let mut c = synced_at(4, &[cell(1, "a"), cell(2, "b")]);
c.apply(&moved_correction(4, 5, &[2], &[]));
assert_eq!(c.held(&shape()), Some(&decode(&target(&[cell(1, "a")]))));
}
#[test]
fn a_dropped_connection_or_failed_request_asks_from_the_watermark() {
let mut c = synced_at(4, &[cell(1, "a")]);
c.connection_lost();
assert_eq!(
c.wants(&shape()),
Some(Want::Correction {
watermark: Some(at(1, 4))
})
);
c.apply(&moved_correction(4, 4, &[], &[]));
assert_eq!(c.wants(&shape()), None, "an empty answer settles it");
c.request_failed(&shape());
assert!(matches!(c.wants(&shape()), Some(Want::Correction { .. })));
}
#[test]
fn a_fetch_does_not_advance_past_the_notices_it_did_not_name() {
let mut c = synced_at(4, &[cell(1, "a")]);
c.on_notice(&notice(1, 5, &[1]));
c.on_notice(&notice(1, 6, &[2]));
c.apply(&moved_correction(4, 5, &[1], &[cell(1, "x")]));
assert_eq!(c.watermark(&shape()), Some(at(1, 5)));
assert_eq!(
c.wants(&shape()),
Some(Want::Fetch {
watermark: Some(at(1, 5)),
keys: [key(2)].into(),
through: at(1, 6),
window: false
})
);
}
#[test]
fn carried_axis_changes_are_applied_once_and_in_order() {
let mut c = synced_at(4, &[]);
let mut n = notice(1, 5, &[]);
n.axis = vec![
AxisChange::CategoryAdded {
category: "X".into(),
},
AxisChange::ItemInserted {
category: "X".into(),
item: "i".into(),
at: 0,
},
];
let events = c.on_notice(&n);
assert_eq!(events.len(), 2, "{events:?}");
let mut carried = moved_correction(4, 5, &[], &[]);
carried.carried = Some(n);
let (_, again) = c.apply(&carried);
assert_eq!(again, Vec::new(), "the same changes must not apply twice");
}
#[test]
fn a_gap_in_structure_asks_the_consumer_to_resync() {
let mut c = synced_at(4, &[]);
assert_eq!(c.on_notice(&notice(1, 7, &[])), vec![Event::Resync]);
}
@@ -0,0 +1,96 @@
//! Requests, corrections and the edge between a client and the host (ADR 009 clause 9).
//! Why each choice: `context/code/delivery.md` §3.
use std::collections::BTreeSet;
use serde::{Deserialize, Serialize};
use super::notice::{Notice, ShapeKey, Stamp};
use super::wire::{WireElem, WireKey, wide};
/// The cross-product rectangle a request costs O(visible) for.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct Window {
#[serde(with = "wide")]
pub top: u128,
#[serde(with = "wide")]
pub left: u128,
#[serde(with = "wide")]
pub rows: u128,
#[serde(with = "wide")]
pub cols: u128,
}
impl Window {
pub fn contains(&self, row: u128, col: u128) -> bool {
(self.top..self.top.saturating_add(self.rows)).contains(&row)
&& (self.left..self.left.saturating_add(self.cols)).contains(&col)
}
}
/// Opaque transport, never read by the host (clause 9, "an edge is a session").
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionToken(pub String);
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
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`. `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 },
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Request {
pub session: Option<SessionToken>,
pub shape: ShapeKey,
pub window: Window,
pub ask: Ask,
}
/// What a correction replaces on the client: the keys it names, or the whole window.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum CorrectionBody {
Moved { keys: BTreeSet<WireKey> },
Window,
}
/// `target`, restricted as `body` says, read at `at` and covering `(from, through]`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Correction {
pub shape: ShapeKey,
pub at: Stamp,
pub from: Option<Stamp>,
pub through: Stamp,
pub body: CorrectionBody,
pub target: Vec<WireElem>,
/// The axis changes, relabels and flags of `(from, through]`, in order, as a merged notice.
pub carried: Option<Notice>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum Reply {
Correction(Box<Correction>),
Notices(Vec<Notice>),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EdgeError {
/// The request did not complete: dropped, timed out or refused by the transport.
Failed,
/// What arrived could not be read.
Unreadable,
}
/// One end of an edge: a request goes out and a reply, or a failure, comes back.
pub trait Edge {
fn call(&mut self, request: Request) -> Result<Reply, EdgeError>;
}
@@ -0,0 +1,20 @@
//! 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;
pub mod notice;
pub mod wire;
pub use client::{EdgeClient, Event, Outcome, Synced, Want};
pub use exchange::{
Ask, Correction, CorrectionBody, Edge, EdgeError, Reply, Request, SessionToken, Window,
};
pub use notice::{AxisChange, Flags, Moved, Notice, PubRange, Relabel, ShapeKey, Stamp};
pub use wire::{Cell, WireElem, WireKey, WireNum, WireValue};
#[cfg(test)]
mod client_tests;
#[cfg(test)]
mod review_tests;
@@ -0,0 +1,315 @@
//! The stamp, the change notice and the merge of notices (ADR 009 clause 9).
//! Why each choice: `context/code/delivery.md` §2.
use std::collections::{BTreeMap, BTreeSet};
use serde::{Deserialize, Serialize};
use super::wire::{WireKey, wide};
/// A logical position in a host's stream: never a wall clock.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct Stamp {
#[serde(with = "wide")]
pub host: u64,
#[serde(with = "wide")]
pub generation: u64,
#[serde(with = "wide")]
pub publication: u64,
}
impl Stamp {
pub fn same_epoch(&self, other: &Stamp) -> bool {
self.host == other.host && self.generation == other.generation
}
/// A stamp from another host instance or generation is behind every current one.
pub fn is_behind(&self, current: &Stamp) -> bool {
!self.same_epoch(current) || self.publication < current.publication
}
}
/// The publications `(from, to]` one notice covers, in one epoch.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct PubRange {
#[serde(with = "wide")]
pub host: u64,
#[serde(with = "wide")]
pub generation: u64,
#[serde(with = "wide")]
pub from: u64,
#[serde(with = "wide")]
pub to: u64,
}
impl PubRange {
pub fn one(at: Stamp) -> Self {
Self {
host: at.host,
generation: at.generation,
from: at.publication - 1,
to: at.publication,
}
}
pub fn end(&self) -> Stamp {
Stamp {
host: self.host,
generation: self.generation,
publication: self.to,
}
}
pub fn start(&self) -> Stamp {
Stamp {
host: self.host,
generation: self.generation,
publication: self.from,
}
}
}
/// A named query: the key operator state is selected by (clause 7).
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct ShapeKey(pub String);
/// A change to the items an axis is made of. The client re-derives positions from these.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum AxisChange {
CategoryAdded {
category: String,
},
CategoryRemoved {
category: String,
},
ItemInserted {
category: String,
item: String,
#[serde(with = "wide")]
at: u128,
},
ItemRemoved {
category: String,
item: String,
},
ItemsReordered {
category: String,
items: Vec<String>,
},
}
/// A display label that moved: the id and what to show.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Relabel {
pub id: String,
pub display: String,
}
/// What moved that no cell key names.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Flags {
pub formulas: bool,
pub measure_agg: bool,
pub views: bool,
pub default_view: bool,
}
impl Flags {
pub fn any(&self) -> bool {
self.formulas || self.measure_agg || self.views || self.default_view
}
pub fn union(self, other: Flags) -> Flags {
Flags {
formulas: self.formulas || other.formulas,
measure_agg: self.measure_agg || other.measure_agg,
views: self.views || other.views,
default_view: self.default_view || other.default_view,
}
}
}
/// The output keys of one shape that moved, or "correct" when the list was too long to send.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum Moved {
Keys(BTreeSet<WireKey>),
Correct,
}
impl Moved {
fn union(self, other: Moved) -> Moved {
match (self, other) {
(Moved::Keys(mut a), Moved::Keys(b)) => {
a.extend(b);
Moved::Keys(a)
}
_ => Moved::Correct,
}
}
}
/// One committed command, or a run of them merged. Addressed to no client.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Notice {
pub range: PubRange,
pub axis: Vec<AxisChange>,
pub relabels: Vec<Relabel>,
pub flags: Flags,
pub evicted: BTreeSet<ShapeKey>,
pub moved: BTreeMap<ShapeKey, Moved>,
}
impl Notice {
pub fn empty(range: PubRange) -> Self {
Self {
range,
axis: Vec::new(),
relabels: Vec::new(),
flags: Flags::default(),
evicted: BTreeSet::new(),
moved: BTreeMap::new(),
}
}
pub fn stamp(&self) -> Stamp {
self.range.end()
}
/// `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, Box<(Notice, Notice)>> {
let adjacent = self.range.end() == next.range.start();
match adjacent {
false => Err(Box::new((self, next))),
true => {
self.range.to = next.range.to;
self.axis.extend(next.axis);
self.relabels.extend(next.relabels);
self.flags = self.flags.union(next.flags);
self.evicted.extend(next.evicted);
for (shape, moved) in next.moved {
let merged = match self.moved.remove(&shape) {
Some(before) => before.union(moved),
None => moved,
};
self.moved.insert(shape, merged);
}
Ok(self)
}
}
}
/// This notice as one shape reads it: other shapes' key lists are not its business.
pub fn for_shape(&self, shape: &ShapeKey) -> Notice {
Notice {
moved: self
.moved
.get(shape)
.map(|m| BTreeMap::from([(shape.clone(), m.clone())]))
.unwrap_or_default(),
..self.clone()
}
}
}
#[cfg(test)]
mod tests {
use super::*;
pub(crate) fn stamp(publication: u64) -> Stamp {
Stamp {
host: 7,
generation: 1,
publication,
}
}
pub(crate) fn key(n: u8) -> WireKey {
WireKey(vec![("C".into(), n.to_string())])
}
pub(crate) fn notice(publication: u64, moved: &[(&str, &[u8])]) -> Notice {
let mut n = Notice::empty(PubRange::one(stamp(publication)));
for (shape, keys) in moved {
n.moved.insert(
ShapeKey((*shape).into()),
Moved::Keys(keys.iter().map(|k| key(*k)).collect()),
);
}
n
}
#[test]
fn a_stamp_from_another_epoch_is_behind_every_current_one() {
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)
};
assert!(other_host.is_behind(&now), "a restart is behind");
assert!(other_generation.is_behind(&now), "a swap is behind");
}
#[test]
fn adjacent_notices_merge_into_the_range_they_cover() {
let merged = notice(3, &[("a", &[1])])
.merge(notice(4, &[("a", &[2]), ("b", &[9])]))
.unwrap();
assert_eq!((merged.range.from, merged.range.to), (2, 4));
assert_eq!(
merged.moved[&ShapeKey("a".into())],
Moved::Keys([key(1), key(2)].into())
);
assert!(merged.moved.contains_key(&ShapeKey("b".into())));
}
#[test]
fn a_gap_does_not_merge() {
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)
}));
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(),
});
let mut b = notice(2, &[]);
b.moved.insert(ShapeKey("s".into()), Moved::Correct);
b.axis.push(AxisChange::ItemRemoved {
category: "X".into(),
item: "i".into(),
});
b.flags.formulas = true;
let m = a.merge(b).unwrap();
assert_eq!(m.axis.len(), 2);
assert!(matches!(m.axis[0], AxisChange::CategoryAdded { .. }));
assert!(matches!(m.axis[1], AxisChange::ItemRemoved { .. }));
assert_eq!(m.moved[&ShapeKey("s".into())], Moved::Correct);
assert!(m.flags.formulas && !m.flags.views);
}
#[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 n = Notice::empty(PubRange::one(wide_host));
let json = serde_json::to_string(&n).unwrap();
assert!(json.contains(&format!("\"{}\"", u64::MAX)), "{json}");
assert_eq!(serde_json::from_str::<Notice>(&json).unwrap(), n);
}
}
@@ -0,0 +1,203 @@
//! Findings from the review of #348: each is a hand-built history that failed before its fix.
use serde::{Deserialize, Serialize};
use super::client::*;
use super::exchange::*;
use super::notice::*;
use super::wire::{self, Cell, WireElem, WireKey, WireValue, encode};
fn at(host: u64, generation: u64, publication: u64) -> Stamp {
Stamp {
host,
generation,
publication,
}
}
fn shape(name: &str) -> ShapeKey {
ShapeKey(name.into())
}
fn window() -> Window {
Window {
top: 0,
left: 0,
rows: 10,
cols: 10,
}
}
fn key(n: u8) -> WireKey {
WireKey(vec![("C".into(), n.to_string())])
}
fn cell(k: u8, v: &str) -> Cell {
(key(k), WireValue::Text(v.into()))
}
fn target(cells: &[Cell]) -> Vec<WireElem> {
encode(&cells.iter().cloned().map(|c| (c, 1)).collect())
}
fn window_correction(name: &str, through: Stamp, cells: &[Cell]) -> Correction {
Correction {
shape: shape(name),
at: through,
from: None,
through,
body: CorrectionBody::Window,
target: target(cells),
carried: None,
}
}
fn synced(name: &str, through: Stamp, cells: &[Cell]) -> EdgeClient {
let mut c = EdgeClient::default();
c.track(shape(name), window());
c.apply(&window_correction(name, through, cells));
c
}
#[test]
fn a_late_window_reply_from_an_older_generation_does_not_regress() {
let mut c = synced("s", at(1, 2, 5), &[cell(1, "new")]);
let late = window_correction("s", at(1, 1, 9), &[cell(1, "old")]);
assert_eq!(c.apply(&late).0, Outcome::Duplicate);
assert_eq!(c.watermark(&shape("s")), Some(at(1, 2, 5)));
let kept = wire::decode(&target(&[cell(1, "new")]));
assert_eq!(c.held(&shape("s")), Some(&kept));
}
#[test]
fn a_late_notice_from_a_dead_host_settles_against_the_live_one() {
let mut c = synced("s", at(2, 1, 4), &[cell(1, "a")]);
c.on_notice(&Notice::empty(PubRange::one(at(1, 1, 9))));
assert!(matches!(
c.wants(&shape("s")),
Some(Want::Correction { .. })
));
let live = window_correction("s", at(2, 1, 4), &[cell(1, "a")]);
assert_eq!(c.apply(&live).0, Outcome::Duplicate);
assert_eq!(
c.wants(&shape("s")),
None,
"the dead host's notice is dropped"
);
c.on_notice(&Notice::empty(PubRange::one(at(2, 1, 5))));
assert_eq!(
c.wants(&shape("s")),
None,
"and the live host's next one is contiguous"
);
}
struct Replies(Vec<Reply>);
impl Edge for Replies {
fn call(&mut self, _: Request) -> Result<Reply, EdgeError> {
match self.0.is_empty() {
true => Err(EdgeError::Failed),
false => Ok(self.0.remove(0)),
}
}
}
#[test]
fn a_poll_for_one_shape_leaves_the_other_shapes_alone() {
let (a, b) = (shape("a"), shape("b"));
let mut c = EdgeClient::default();
for s in [&a, &b] {
c.track(s.clone(), window());
c.apply(&window_correction(&s.0, at(1, 1, 4), &[]));
}
let mut pulled = Notice::empty(PubRange::one(at(1, 1, 5)));
pulled.moved.insert(a.clone(), Moved::Keys([key(1)].into()));
let mut edge = Replies(vec![Reply::Notices(vec![pulled])]);
c.poll(&mut edge, &a);
assert_eq!(c.watermark(&b), Some(at(1, 1, 4)), "b was not asked about");
let mut next = Notice::empty(PubRange::one(at(1, 1, 6)));
next.moved.insert(b.clone(), Moved::Keys([key(2)].into()));
c.on_notice(&next);
assert!(
matches!(c.wants(&b), Some(Want::Correction { .. })),
"b missed publication 5, and must see the gap: {:?}",
c.wants(&b)
);
}
#[test]
fn a_views_flag_makes_the_client_fetch_its_window() {
let mut c = synced("s", at(1, 1, 4), &[cell(1, "a")]);
let mut n = Notice::empty(PubRange::one(at(1, 1, 5)));
n.flags.views = true;
c.on_notice(&n);
assert!(matches!(
c.wants(&shape("s")),
Some(Want::Fetch { window: true, .. })
));
}
#[test]
fn wide_integers_are_canonical_on_input() {
#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct P(
#[serde(with = "wire::wide")] u64,
#[serde(with = "wire::wide_weight")] i64,
);
let bad = [
r#"[9007199254740992,1]"#,
r#"["5",1]"#,
r#"["+9007199254740992",1]"#,
r#"["09007199254740992",1]"#,
r#"[1,9007199254740992]"#,
r#"[1,"5"]"#,
r#"[1,"-5"]"#,
r#"[1,"+9007199254740992"]"#,
r#"[1,"-9007199254740992 "]"#,
];
for text in bad {
assert!(serde_json::from_str::<P>(text).is_err(), "{text}");
}
let good = r#"["9007199254740992","-9007199254740992"]"#;
assert_eq!(
serde_json::from_str::<P>(good).unwrap(),
P(1 << 53, -(1 << 53))
);
assert_eq!(
serde_json::from_str::<P>("[9007199254740991,-9007199254740991]").unwrap(),
P((1 << 53) - 1, -((1 << 53) - 1))
);
}
#[test]
fn an_item_insert_position_crosses_as_a_wide_integer() {
let change = AxisChange::ItemInserted {
category: "C".into(),
item: "i".into(),
at: 1 << 60,
};
let json = serde_json::to_string(&change).unwrap();
assert!(json.contains(r#""at":"1152921504606846976""#), "{json}");
assert_eq!(serde_json::from_str::<AxisChange>(&json).unwrap(), change);
}
#[test]
fn a_late_window_reply_from_a_dead_host_does_not_regress() {
let mut c = synced("s", at(1, 1, 9), &[cell(1, "old")]);
c.apply(&window_correction("s", at(2, 1, 3), &[cell(1, "new")]));
let late = window_correction("s", at(1, 1, 12), &[cell(1, "dead")]);
assert_eq!(c.apply(&late).0, Outcome::Duplicate);
assert_eq!(c.watermark(&shape("s")), Some(at(2, 1, 3)));
let kept = wire::decode(&target(&[cell(1, "new")]));
assert_eq!(c.held(&shape("s")), Some(&kept));
}
#[test]
fn a_reply_from_a_generation_the_client_left_does_not_return() {
let mut c = synced("s", at(1, 1, 9), &[cell(1, "old")]);
c.apply(&window_correction("s", at(1, 2, 1), &[cell(1, "new")]));
c.apply(&window_correction("s", at(1, 3, 1), &[cell(1, "newer")]));
let late = window_correction("s", at(1, 2, 7), &[cell(1, "gen2")]);
assert_eq!(c.apply(&late).0, Outcome::Duplicate);
}
@@ -0,0 +1,342 @@
//! The lossless wire form of what flows on a delivery edge (ADR 009 clause 9).
//! Why each choice: `context/code/delivery.md` §1.
use improvise_core::model::cell::{CellKey, CellValue};
use numtower::Num;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use zset::ZSet;
const EXACT_JSON: u128 = 1 << 53;
/// An unsigned integer: a JSON number below 2^53, a decimal string above it.
pub mod wide {
use super::*;
#[derive(Serialize, Deserialize)]
#[serde(untagged)]
enum Repr {
Small(u64),
Wide(String),
}
pub fn serialize<T: Copy + Into<u128>, S: Serializer>(n: &T, s: S) -> Result<S::Ok, S::Error> {
let n: u128 = (*n).into();
match n < EXACT_JSON {
true => Repr::Small(n as u64).serialize(s),
false => Repr::Wide(n.to_string()).serialize(s),
}
}
pub fn deserialize<'de, T: TryFrom<u128>, D: Deserializer<'de>>(d: D) -> Result<T, D::Error> {
let n: u128 = match Repr::deserialize(d)? {
Repr::Small(n) if u128::from(n) < EXACT_JSON => n.into(),
Repr::Small(_) => {
return Err(serde::de::Error::custom("a wide integer must be a string"));
}
Repr::Wide(text) => {
let n: u128 = text.parse().map_err(serde::de::Error::custom)?;
match n >= EXACT_JSON && n.to_string() == text {
true => n,
false => return Err(serde::de::Error::custom("not a canonical wide integer")),
}
}
};
T::try_from(n).map_err(|_| serde::de::Error::custom("integer out of range"))
}
}
/// A signed weight, with the same rule as [`wide`].
pub mod wide_weight {
use super::*;
#[derive(Serialize, Deserialize)]
#[serde(untagged)]
enum Repr {
Small(i64),
Wide(String),
}
pub fn serialize<S: Serializer>(n: &i64, s: S) -> Result<S::Ok, S::Error> {
match u128::from(n.unsigned_abs()) < EXACT_JSON {
true => Repr::Small(*n).serialize(s),
false => Repr::Wide(n.to_string()).serialize(s),
}
}
pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<i64, D::Error> {
match Repr::deserialize(d)? {
Repr::Small(n) if u128::from(n.unsigned_abs()) < EXACT_JSON => Ok(n),
Repr::Small(_) => Err(serde::de::Error::custom("a wide weight must be a string")),
Repr::Wide(text) => {
let n: i64 = text.parse().map_err(serde::de::Error::custom)?;
let wide = u128::from(n.unsigned_abs()) >= EXACT_JSON;
match wide && n.to_string() == text {
true => Ok(n),
false => Err(serde::de::Error::custom("not a canonical wide weight")),
}
}
}
}
}
/// A number in ADR 005's canonical text. Equality is textual, so the sign of a zero survives.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct WireNum(String);
impl WireNum {
pub fn of(n: &Num) -> Self {
Self(n.to_string())
}
pub fn of_f64(x: f64) -> Self {
Self::of(&Num::inexact(x))
}
/// `None` for text the tower does not write, so one value has one spelling.
pub fn parse(text: &str) -> Option<Self> {
let n = Num::from_canonical(text)?;
(n.to_string() == text).then(|| Self(text.to_string()))
}
pub fn text(&self) -> &str {
&self.0
}
pub fn to_num(&self) -> Num {
Num::from_canonical(&self.0).expect("a WireNum holds canonical text by construction")
}
}
impl Serialize for WireNum {
fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
s.serialize_str(&self.0)
}
}
impl<'de> Deserialize<'de> for WireNum {
fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
let text = String::deserialize(d)?;
Self::parse(&text)
.ok_or_else(|| serde::de::Error::custom(format!("{text:?} is not a canonical number")))
}
}
/// A cell's value on the wire. Text is byte-exact; a number never crosses as a double.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub enum WireValue {
Num(WireNum),
Text(String),
Error(String),
}
impl WireValue {
pub fn of(value: &CellValue) -> Self {
match value {
CellValue::Number(x) => Self::Num(WireNum::of_f64(*x)),
CellValue::Text(t) => Self::Text(t.clone()),
CellValue::Error(e) => Self::Error(e.clone()),
}
}
/// Exact numbers narrow to `f64` here, because `CellValue` is still an `f64` (ADR 005).
pub fn to_cell_value(&self) -> CellValue {
match self {
Self::Num(n) => CellValue::Number(n.to_num().to_f64()),
Self::Text(t) => CellValue::Text(t.clone()),
Self::Error(e) => CellValue::Error(e.clone()),
}
}
}
/// A cell named by its coordinate tuple (ids once ADR 007 lands).
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct WireKey(pub Vec<(String, String)>);
impl WireKey {
pub fn of(key: &CellKey) -> Self {
Self(key.0.clone())
}
}
pub type Cell = (WireKey, WireValue);
/// One element of a Z-set on the wire.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WireElem {
pub key: WireKey,
pub value: WireValue,
#[serde(with = "wide_weight")]
pub weight: i64,
}
pub fn encode(z: &ZSet<Cell>) -> Vec<WireElem> {
z.iter()
.map(|((key, value), weight)| WireElem {
key: key.clone(),
value: value.clone(),
weight,
})
.collect()
}
pub fn decode(elems: &[WireElem]) -> ZSet<Cell> {
elems
.iter()
.map(|e| ((e.key.clone(), e.value.clone()), e.weight))
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use proptest::prelude::*;
fn json_round<T: Serialize + for<'de> Deserialize<'de>>(v: &T) -> T {
serde_json::from_str(&serde_json::to_string(v).expect("serializes")).expect("reads back")
}
#[test]
fn a_float_crosses_in_canonical_text_with_the_sign_of_zero() {
for (x, text) in [
(0.1, "#i0.1"),
(-0.0, "#i-0.0"),
(0.0, "#i0.0"),
(f64::INFINITY, "inf"),
(f64::NEG_INFINITY, "-inf"),
(f64::NAN, "nan"),
] {
let sent = WireValue::of(&CellValue::Number(x));
let json = serde_json::to_string(&sent).unwrap();
assert_eq!(json, format!(r#"{{"Num":"{text}"}}"#), "{x:?}");
let CellValue::Number(y) = json_round(&sent).to_cell_value() else {
panic!("a number came back as something else")
};
match x.is_nan() {
true => assert!(y.is_nan()),
false => assert_eq!(x.to_bits(), y.to_bits(), "{x:?} came back {y:?}"),
}
}
}
#[test]
fn every_nan_is_the_one_token() {
let payload = f64::from_bits(0x7ff8_dead_beef_0001);
assert!(payload.is_nan());
assert_eq!(WireNum::of_f64(payload), WireNum::of_f64(f64::NAN));
}
#[test]
fn exact_numbers_round_trip_beyond_a_double() {
let wide_decimal = Num::from_canonical("1234567890123456789.0123456789").unwrap();
let fraction = Num::ratio(i128::MAX / 3, 7).unwrap();
for n in [wide_decimal, fraction] {
let back = json_round(&WireNum::of(&n)).to_num();
assert_eq!(back.to_string(), n.to_string());
assert_eq!(back.is_exact(), n.is_exact());
assert!(back.is_exact(), "{n} lost its exactness: {back}");
}
}
#[test]
fn text_that_the_tower_does_not_write_is_refused() {
for bad in ["1e3", " 1", "+1", "#i1", "1.50x", "0x10", ""] {
assert_eq!(WireNum::parse(bad), None, "{bad:?}");
let json = serde_json::to_string(bad).unwrap();
assert!(serde_json::from_str::<WireNum>(&json).is_err(), "{bad:?}");
}
}
#[test]
fn wide_integers_cross_as_strings_and_narrow_ones_as_numbers() {
#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct P(
#[serde(with = "wide")] u128,
#[serde(with = "wide_weight")] i64,
);
let below = (1u128 << 53) - 1;
assert_eq!(
serde_json::to_string(&P(below, -3)).unwrap(),
format!("[{below},-3]")
);
let p = P(u128::MAX, i64::MIN);
let json = serde_json::to_string(&p).unwrap();
assert_eq!(json, format!(r#"["{}","{}"]"#, u128::MAX, i64::MIN));
assert_eq!(json_round(&p), p);
assert_eq!(json_round(&P(1 << 53, 1 << 53)), P(1 << 53, 1 << 53));
let past = r#"["340282366920938463463374607431768211456",1]"#;
assert!(serde_json::from_str::<P>(past).is_err());
}
#[test]
fn text_is_byte_exact() {
for t in ["", " é ", "e\u{301}", "\u{0}\n\r\"\\", "\u{1F600}"] {
let v = WireValue::Text(t.to_string());
assert_eq!(json_round(&v), v, "{t:?}");
}
}
fn any_f64() -> impl Strategy<Value = f64> {
prop_oneof![
any::<f64>(),
Just(-0.0),
Just(f64::INFINITY),
Just(f64::NEG_INFINITY),
Just(f64::NAN),
any::<u64>().prop_map(f64::from_bits),
]
}
fn a_decimal(mantissa: u64, scale: usize) -> Option<Num> {
let digits = format!("{mantissa:0>w$}", w = scale + 1);
let (whole, frac) = digits.split_at(digits.len() - scale);
let text = match scale {
0 => whole.to_string(),
_ => format!("{whole}.{frac}"),
};
Num::from_canonical(&text)
}
fn any_value() -> impl Strategy<Value = WireValue> {
prop_oneof![
any_f64().prop_map(|x| WireValue::of(&CellValue::Number(x))),
(any::<i128>(), 1..i128::MAX)
.prop_filter_map("fits", |(n, d)| Num::ratio(n, d))
.prop_map(|n| WireValue::Num(WireNum::of(&n))),
(any::<u64>(), 0usize..=28)
.prop_filter_map("decimal", |(m, s)| a_decimal(m, s))
.prop_map(|n| WireValue::Num(WireNum::of(&n))),
any::<String>().prop_map(WireValue::Text),
any::<String>().prop_map(WireValue::Error),
]
}
proptest! {
#[test]
fn a_value_is_the_identity_through_the_wire(v in any_value()) {
prop_assert_eq!(json_round(&v), v);
}
#[test]
fn a_float_keeps_every_bit_but_the_nan_payload(x in any_f64()) {
let sent = WireValue::of(&CellValue::Number(x));
let CellValue::Number(y) = json_round(&sent).to_cell_value() else {
panic!("a number came back as something else")
};
match x.is_nan() {
true => prop_assert!(y.is_nan()),
false => prop_assert_eq!(x.to_bits(), y.to_bits()),
}
}
#[test]
fn a_zset_is_the_identity_through_the_wire(
elems in proptest::collection::vec((0u8..4, any_value(), -3i64..=3), 0..12)
) {
let z: ZSet<Cell> = elems
.into_iter()
.map(|(k, v, w)| ((WireKey(vec![("c".into(), k.to_string())]), v), w))
.collect();
prop_assert_eq!(decode(&json_round(&encode(&z))), z);
}
}
}
+1
View File
@@ -6,6 +6,7 @@ use std::collections::HashMap;
pub mod basis;
pub mod client;
pub mod delivery;
pub mod face;
pub mod frame;
pub mod help;
+1 -1
View File
@@ -14,7 +14,7 @@ pub use arith::{CEILING_BITS, DivisionByZero, Interim};
pub use lang::{Typing, is_number, parse_number, shortest_round_trip};
pub use num::{Num, NumRef};
pub use round::Rounding;
pub use sum::{ExactSum, FloatSum, K, NumSum};
pub use sum::{ExactSum, K, NumSum};
pub use num_rational::Ratio;
pub use rust_decimal::Decimal;
+23 -41
View File
@@ -27,6 +27,10 @@ pub struct ExactSum {
nan: i128,
pos_inf: i128,
neg_inf: i128,
/// Net weight of the `-0.0` addends, and of every other finite one: a zero sum is
/// `-0.0` only when the first is positive and the second is nil (IEEE).
neg_zeros: i128,
others: i128,
}
/// Equal sums, not equal representations: carries propagated, `pending` ignored.
@@ -36,7 +40,9 @@ impl PartialEq for ExactSum {
propagate(&mut mine);
propagate(&mut theirs);
let counts = |s: &ExactSum| (s.nan, s.pos_inf, s.neg_inf);
mine == theirs && counts(self) == counts(o)
let zero = mine.iter().all(|l| *l == 0);
let same_zero = !zero || self.negative_zero() == o.negative_zero();
mine == theirs && counts(self) == counts(o) && same_zero
}
}
@@ -50,6 +56,8 @@ impl Default for ExactSum {
nan: 0,
pos_inf: 0,
neg_inf: 0,
neg_zeros: 0,
others: 0,
}
}
}
@@ -66,6 +74,10 @@ impl ExactSum {
(false, true, true) => self.pos_inf += weight as i128,
(false, true, false) => self.neg_inf += weight as i128,
(false, false, _) => {
match x == 0.0 && x.is_sign_negative() {
true => self.neg_zeros += weight as i128,
false => self.others += weight as i128,
}
let magnitude = weight.unsigned_abs();
for j in 0..3 {
let digit = (magnitude >> (30 * j)) & (CHUNK - 1);
@@ -122,6 +134,12 @@ impl ExactSum {
self.nan += o.nan;
self.pos_inf += o.pos_inf;
self.neg_inf += o.neg_inf;
self.neg_zeros += o.neg_zeros;
self.others += o.others;
}
fn negative_zero(&self) -> bool {
self.neg_zeros > 0 && self.others == 0
}
/// The exact sum rounded to the nearest f64, ties to even.
@@ -130,7 +148,10 @@ impl ExactSum {
(true, _, _) | (false, true, true) => f64::NAN,
(false, true, false) => f64::INFINITY,
(false, false, true) => f64::NEG_INFINITY,
(false, false, false) => ratio_to_f64(&self.total(), &(BigInt::one() << 1074)),
(false, false, false) => match (self.negative_zero(), self.total()) {
(true, total) if total.is_zero() => -0.0,
(_, total) => ratio_to_f64(&total, &(BigInt::one() << 1074)),
},
}
}
@@ -144,45 +165,6 @@ impl ExactSum {
}
}
/// An `ExactSum` that also gives a zero total the sign IEEE gives the same addends (clause 8):
/// `-0.0` only if every addend is `-0.0`, else `+0.0`. An empty sum is `-0.0`, as `Sum for f64` is.
#[derive(Clone, Debug, Default)]
pub struct FloatSum {
exact: ExactSum,
addends: i128,
negative_zeros: i128,
}
impl FloatSum {
pub fn new() -> FloatSum {
FloatSum::default()
}
/// Adds `x` with a Z-set `weight`, negative to retract.
pub fn add(&mut self, x: f64, weight: i64) {
self.exact.add(x, weight);
self.addends += i128::from(weight);
if x == 0.0 && x.is_sign_negative() {
self.negative_zeros += i128::from(weight);
}
}
/// Adds another accumulator's contents.
pub fn absorb(&mut self, o: &FloatSum) {
self.exact.absorb(&o.exact);
self.addends += o.addends;
self.negative_zeros += o.negative_zeros;
}
/// The exact sum rounded to the nearest f64, ties to even.
pub fn value(&self) -> f64 {
match self.exact.value() {
v if v == 0.0 && self.negative_zeros == self.addends => -0.0,
v => v,
}
}
}
/// Carry each limb into the next, leaving every limb but the top in [0, 2^32).
fn propagate(limbs: &mut [i64; LIMBS]) {
for k in 0..LIMBS - 1 {
-78
View File
@@ -1,78 +0,0 @@
//! `FloatSum`: ADR 009 clause 8's sign of a zero total, against IEEE addition.
//! Evidence: `context/test-evidence/numtower.md`.
use numtower::FloatSum;
use proptest::prelude::*;
/// Both zeros and small integers: totals that cancel, and every sum below 2^53 is exact.
fn addend() -> impl Strategy<Value = f64> {
prop_oneof![
Just(0.0),
Just(-0.0),
(-3i32..=3).prop_map(f64::from),
prop::sample::select(vec![0.5, -0.5]),
]
}
fn sum_of(values: &[f64]) -> FloatSum {
let mut sum = FloatSum::new();
for v in values {
sum.add(*v, 1);
}
sum
}
/// What adding the same addends in `f64` gives, starting from `-0.0` as `Sum for f64` does.
fn ieee(values: &[f64]) -> f64 {
values.iter().sum()
}
fn same_bits(a: f64, b: f64) -> bool {
a.to_bits() == b.to_bits()
}
#[test]
fn the_sign_of_a_zero_total_follows_ieee_for_the_cases_that_decide_it() {
assert!(same_bits(sum_of(&[-0.0, -0.0]).value(), -0.0));
assert!(same_bits(sum_of(&[-0.0, 0.0]).value(), 0.0));
assert!(same_bits(sum_of(&[0.0, 0.0]).value(), 0.0));
assert!(same_bits(sum_of(&[1.0, -1.0]).value(), 0.0));
assert!(same_bits(sum_of(&[-0.0, 1.0, -1.0]).value(), 0.0));
assert!(same_bits(sum_of(&[]).value(), -0.0), "as Sum for f64 is");
}
proptest! {
#[test]
fn a_sum_has_the_bits_ieee_gives_the_same_addends(
values in prop::collection::vec(addend(), 0..12),
) {
let (got, want) = (sum_of(&values).value(), ieee(&values));
prop_assert!(same_bits(got, want), "{got:?} for {want:?} over {values:?}");
}
#[test]
fn retracting_addends_leaves_the_sum_of_the_rest(
values in prop::collection::vec(addend(), 1..12),
keep in 0usize..12,
) {
let keep = keep.min(values.len());
let mut sum = sum_of(&values);
for v in &values[keep..] {
sum.add(*v, -1);
}
let (got, want) = (sum.value(), ieee(&values[..keep]));
prop_assert!(same_bits(got, want), "{got:?} for {want:?} over {values:?} keeping {keep}");
}
#[test]
fn absorbing_two_sums_is_adding_all_the_addends(
values in prop::collection::vec(addend(), 0..12),
cut in 0usize..12,
) {
let cut = cut.min(values.len());
let mut left = sum_of(&values[..cut]);
left.absorb(&sum_of(&values[cut..]));
let (got, want) = (left.value(), sum_of(&values).value());
prop_assert!(same_bits(got, want), "{got:?} for {want:?} over {values:?} cut at {cut}");
}
}
+102
View File
@@ -0,0 +1,102 @@
//! ADR 009 step 8's last test: the sign of a zero sum is IEEE's. Evidence: `context/test-evidence/numtower.md`.
use numtower::ExactSum;
use proptest::prelude::*;
fn addend() -> impl Strategy<Value = f64> {
prop_oneof![
Just(-0.0),
Just(0.0),
Just(0.5),
Just(-0.5),
Just(3.0),
Just(-3.0),
]
}
/// IEEE addition from `-0.0`, the identity that keeps a `-0.0` addend's sign.
fn ieee(addends: &[f64]) -> f64 {
addends.iter().fold(-0.0, |sum, x| sum + x)
}
fn exact(addends: &[f64]) -> f64 {
let mut sum = ExactSum::new();
addends.iter().for_each(|x| sum.add(*x, 1));
sum.value()
}
#[test]
fn a_sum_of_negative_zeros_is_negative_zero() {
assert_eq!(exact(&[-0.0, -0.0]).to_bits(), (-0.0f64).to_bits());
assert_eq!(exact(&[-0.0]).to_bits(), (-0.0f64).to_bits());
}
#[test]
fn one_positive_zero_makes_the_zero_positive_and_so_does_a_cancellation() {
assert_eq!(exact(&[-0.0, 0.0]).to_bits(), 0.0f64.to_bits());
assert_eq!(exact(&[0.5, -0.5]).to_bits(), 0.0f64.to_bits());
assert_eq!(exact(&[-0.0, 0.5, -0.5]).to_bits(), 0.0f64.to_bits());
}
#[test]
fn an_empty_sum_is_positive_zero() {
assert_eq!(ExactSum::new().value().to_bits(), 0.0f64.to_bits());
}
/// Addends whose exact sum is zero: some zeros of either sign, and cancelling pairs.
fn zero_sum() -> impl Strategy<Value = Vec<f64>> {
let zeros = proptest::collection::vec(prop_oneof![Just(-0.0), Just(0.0)], 0..5);
let pairs = proptest::collection::vec(prop_oneof![Just(0.5), Just(-3.0)], 0..3);
(zeros, pairs)
.prop_map(|(mut all, pairs)| {
all.extend(pairs.iter().flat_map(|p| [*p, -*p]));
all
})
.prop_filter("non-empty", |all| !all.is_empty())
.prop_shuffle()
}
proptest! {
#[test]
fn a_zero_sum_has_the_sign_ieee_gives_the_same_addends(addends in zero_sum()) {
let want = ieee(&addends);
prop_assert_eq!(want, 0.0, "the generator must build zero sums");
prop_assert_eq!(exact(&addends).to_bits(), want.to_bits());
}
#[test]
fn a_retracted_addend_leaves_the_sign_of_the_rest(
keep in zero_sum(),
gone in proptest::collection::vec(addend(), 0..6),
) {
let mut sum = ExactSum::new();
keep.iter().chain(&gone).for_each(|x| sum.add(*x, 1));
gone.iter().for_each(|x| sum.add(*x, -1));
prop_assert_eq!(sum.value().to_bits(), ieee(&keep).to_bits());
let mut fresh = ExactSum::new();
keep.iter().for_each(|x| fresh.add(*x, 1));
prop_assert_eq!(sum, fresh);
}
}
#[test]
fn a_sum_that_is_not_zero_never_reads_as_negative_zero() {
let mut sum = ExactSum::new();
sum.add(-0.0, 1);
sum.add(5.0, 1);
sum.add(7.0, -1);
assert_eq!(sum.value(), -2.0);
}
#[test]
fn equal_values_compare_equal_whatever_the_zero_counts_say() {
let mut with = ExactSum::new();
with.add(-0.0, 1);
with.add(5.0, 1);
with.add(7.0, -1);
let mut without = ExactSum::new();
without.add(5.0, 1);
without.add(7.0, -1);
assert_eq!(with.value(), without.value());
assert_eq!(with, without);
}