Merge pull request #348 from fiddlerwoaroof/adr009/delivery
feat(delivery): ADR 009 step 8, notices, watermarks, corrections and the lossless wire (improvise-z2vr)
This commit is contained in:
Ed Langley
GitHub
Generated
+5
@@ -910,11 +910,13 @@ dependencies = [
|
||||
"improvise-incremental",
|
||||
"improvise-io",
|
||||
"improvise-protocol",
|
||||
"proptest",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tempfile",
|
||||
"unicode-width",
|
||||
"web-time",
|
||||
"zset",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -992,9 +994,12 @@ name = "improvise-protocol"
|
||||
version = "0.1.0-rc2"
|
||||
dependencies = [
|
||||
"improvise-core",
|
||||
"numtower",
|
||||
"proptest",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"unicode-width",
|
||||
"zset",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -3024,6 +3024,10 @@ rec {
|
||||
name = "web-time";
|
||||
packageId = "web-time";
|
||||
}
|
||||
{
|
||||
name = "zset";
|
||||
packageId = "zset";
|
||||
}
|
||||
];
|
||||
devDependencies = [
|
||||
{
|
||||
@@ -3036,6 +3040,10 @@ rec {
|
||||
packageId = "improvise-protocol";
|
||||
features = [ "test-support" ];
|
||||
}
|
||||
{
|
||||
name = "proptest";
|
||||
packageId = "proptest";
|
||||
}
|
||||
{
|
||||
name = "tempfile";
|
||||
packageId = "tempfile";
|
||||
@@ -3290,6 +3298,10 @@ rec {
|
||||
name = "improvise-core";
|
||||
packageId = "improvise-core";
|
||||
}
|
||||
{
|
||||
name = "numtower";
|
||||
packageId = "numtower";
|
||||
}
|
||||
{
|
||||
name = "serde";
|
||||
packageId = "serde";
|
||||
@@ -3305,8 +3317,16 @@ rec {
|
||||
name = "unicode-width";
|
||||
packageId = "unicode-width";
|
||||
}
|
||||
{
|
||||
name = "zset";
|
||||
packageId = "zset";
|
||||
}
|
||||
];
|
||||
devDependencies = [
|
||||
{
|
||||
name = "proptest";
|
||||
packageId = "proptest";
|
||||
}
|
||||
{
|
||||
name = "serde_json";
|
||||
packageId = "serde_json";
|
||||
|
||||
@@ -0,0 +1,227 @@
|
||||
# Delivery: the reasoning
|
||||
|
||||
ADR 009 step 8 (improvise-z2vr): notices per committed command, watermarks, corrections, the shared
|
||||
bounded log, the lossless wire and the edge adapters. The decisions are the ADR's clause 9
|
||||
(`context/adr/009-incremental-evaluation.md`). This file records how the code reads them, the places
|
||||
the ADR left a choice open, and the places it could not be satisfied yet. The evidence for each test is
|
||||
in `context/test-evidence/delivery.md`.
|
||||
|
||||
**What is built, and against what.** Neither the incremental backend nor ADR 008's render model
|
||||
exists. So the source of every target is the existing Eager path: the cells a `GridLayout` draws for a
|
||||
view. Everything below is correct against that source and does not depend on what replaces it. The
|
||||
circuit will supply "moved output keys" where `Delivery::commit` diffs two `Output`s today, and
|
||||
nothing else changes.
|
||||
|
||||
| piece | where |
|
||||
|---|---|
|
||||
| wire values, keys, Z-set elements | `improvise-protocol/src/delivery/wire.rs` |
|
||||
| stamp, range, notice, merge | `.../notice.rs` |
|
||||
| request, correction, `Edge` | `.../exchange.rs` |
|
||||
| the client end: watermark, held cells, when to ask | `.../client.rs` |
|
||||
| the publisher, the log, live shapes, answering | `improvise-command/src/ui/delivery.rs` |
|
||||
| `Host` integration, in-process and JSON edges | `ui/host.rs`, `ui/delivery.rs` |
|
||||
|
||||
## 1. The wire form
|
||||
|
||||
A number crosses as ADR 005's canonical text through `numtower::Num`: `#i0.1`, `#i-0.0`, `inf`,
|
||||
`-inf`, `nan`. `WireNum` accepts only text the tower itself writes (`parse` re-prints and compares),
|
||||
because a second spelling of one value would make textual equality wrong. Equality of wire values is
|
||||
textual, so `-0.0` and `+0.0` are different wire values. That is the opposite of the cell delta in
|
||||
clause 3, which makes them equal; the two do not meet, because the delta is a store-side signal and a
|
||||
correction is a value the client displays.
|
||||
|
||||
An integer is a JSON number below 2^53 and a decimal string at or above it (`wide`, `wide_weight`).
|
||||
That covers stamps, positions, window sizes and weights. Text is a JSON string, byte-exact.
|
||||
|
||||
`CellValue` is still an `f64`, so `WireValue::of` writes it through `Num::inexact` and
|
||||
`to_cell_value` narrows an exact number to `f64`. The round-trip properties therefore run over
|
||||
28-digit decimals and `i128`-wide fractions at the `WireNum` level, where they are lossless, and over
|
||||
`f64` at the `CellValue` level, where only the NaN payload is dropped.
|
||||
|
||||
`CellKey` is not `Serialize`, so a cell is named by `WireKey`, its `(category, item)` tuple. Under
|
||||
ADR 007 those become ids; nothing here reads the strings.
|
||||
|
||||
## 2. Stamps and notices
|
||||
|
||||
`Stamp` is `(host instance, document_generation, publication)`. A stamp from another host or
|
||||
generation is behind every current one (`Stamp::is_behind`). The publication number increments once
|
||||
per committed command, and it keeps counting across a document swap rather than restarting, so a
|
||||
number is never reused within one host instance.
|
||||
|
||||
**What "a committed command" is.** The host publishes one notice for each `Command::Model` that
|
||||
lands. A view command (cursor, mode, a view switch) publishes nothing: it is view-store state and
|
||||
the model store has no business announcing it. A write that changes nothing still publishes, with
|
||||
empty `moved`, because the contract is one notice per commit and a client counts them.
|
||||
|
||||
**A notice names moved keys per live shape**, plus axis changes, relabels, flags, evicted shapes. Axis
|
||||
changes are computed from the categories' item lists before and after (`axis_changes`): insert and
|
||||
remove by item, and a reorder when the surviving items changed order. `Relabel` exists and is never
|
||||
emitted: with names as identity a rename is a remove and an insert, and display writes arrive with
|
||||
ADR 007. Flags come from comparing the formula texts, `measure_agg`, the views and the default view.
|
||||
|
||||
`Notice::merge` joins two notices only when the second starts where the first ends (same epoch).
|
||||
Anything else returns both unchanged, so a merge cannot hide a loss. `Moved::Correct` absorbs key
|
||||
lists in a merge.
|
||||
|
||||
## 3. Requests and corrections
|
||||
|
||||
A request is a shape key, a window and one of three asks. `Ask::Correction` is "bring me up from my
|
||||
watermark"; `Ask::Fetch` is "the current values of these keys, covering the publications up to
|
||||
`through`"; `Ask::Notices` is "the notices after this stamp".
|
||||
|
||||
**A correction's elements are `target`, not `target - held`.** The host does not know what the client
|
||||
holds, so it sends the target restricted as the body says, and the client computes `target - held`
|
||||
with ordinary Z-set subtraction. A `Moved` body restricts both sides to the named keys, so a key the
|
||||
target no longer holds is removed. A `Window` body replaces everything held.
|
||||
|
||||
**`through` is not `at`.** `at` is the state the values were read at, always the latest committed one.
|
||||
`through` is how far the correction advances the client's watermark. They differ for a `Fetch`:
|
||||
the client named the keys of the notices it has seen, and a later commit may have moved other keys
|
||||
whose notice it has not seen yet. Advancing to `at` would make that notice look already applied, so
|
||||
the watermark advances only to `through` and the later notice refetches its keys. The values are
|
||||
at least as new as the watermark claims, never older.
|
||||
|
||||
**An axis change makes the correction a window refetch.** A moved key list says which values changed,
|
||||
not which keys sit in the client's window, and an insert or remove shifts cells into and out of it
|
||||
without moving them. The ADR has the client re-derive positions from the axis changes and fetch what
|
||||
it lacks. That needs ADR 008's render model, which this layer does not have. So the client records a
|
||||
notice that moved an axis or set `flags.views` (`restructured`), asks `Fetch { window: true }`, and the host answers with a
|
||||
`Window` body whose `carried` notice holds the axis changes, in order. This is a fetch, not a
|
||||
correction, so the required test "a frame `[AddItem, SetCell]` ... and no client asks for a correction"
|
||||
holds: it asserts no `Ask::Correction` was sent. When the render model exists the client can narrow
|
||||
the fetch to the cells that entered its window (improvise-bw4p.9).
|
||||
|
||||
A view command (`SetAxis`, `TransposeAxes`, a page selection, a hide) moves positions and no key, and
|
||||
reaches the client only as `flags.views`. So that flag counts as an axis change on both sides: the client
|
||||
fetches its window, and the host answers a correction over a range holding the flag with a window body.
|
||||
|
||||
## 4. The client
|
||||
|
||||
`EdgeClient` holds, per shape: the window, the held cells (a `ZSet`), the **watermark** (applied
|
||||
through), `seen` (the last notice received), the notices received and not yet fetched, `restructured`
|
||||
and `stale`.
|
||||
|
||||
`on_notice` ignores a duplicate (`to <= seen`), treats a gap, another host or generation, a shape in the
|
||||
evicted list, or `Moved::Correct` as "ask for a correction", and otherwise queues the moved keys. The
|
||||
watermark advances over a notice only when nothing is owed, so it never claims a key it has not fetched.
|
||||
|
||||
`apply` takes a correction. It is a duplicate if the watermark already covers `through`; a `Moved`
|
||||
body is a gap unless it starts at or below the watermark. Otherwise `held += target - held`, the
|
||||
watermark becomes `through`, and the queued notices it covers are dropped. A late reply older than
|
||||
the watermark therefore cannot regress the state, which is what makes a retry safe.
|
||||
|
||||
**A correction from a generation or host the client has left is a duplicate, whatever the
|
||||
watermark.** A shape records the epochs its watermark has moved out of (`left`, the newest 64), and a
|
||||
correction from one of them, or from an older generation of the host it is on, changes nothing. Epochs of
|
||||
different hosts are not ordered, so a client that has never applied anything from the newer host cannot
|
||||
tell a late reply from a dead one from a legitimate first answer, and applies it; once it has moved on, the
|
||||
dead host's replies are refused. A duplicate answer also prunes every queued entry from another epoch
|
||||
and moves `seen` onto the live epoch, so a late notice from a dead host costs one extra round and cannot
|
||||
livelock `sync`.
|
||||
|
||||
`poll` hands a pulled notice only to the shape it asked about. A notice carries only that shape's keys,
|
||||
and applying it to another tracked shape would advance that shape past keys it never saw.
|
||||
|
||||
`connection_lost`, `request_failed` and `unreadable` set `stale`, which `wants` turns into
|
||||
`Ask::Correction` from the watermark the client held. A duplicate whose `through` equals the watermark
|
||||
also clears `stale`: the host has said there is nothing newer.
|
||||
|
||||
**A lost final notice is invisible to a listener.** The next notice shows the gap, but if none comes the
|
||||
client never learns. `poll` is the answer: it asks `Ask::Notices { after: seen }`, which the host
|
||||
answers from the log. The property test ends every history with a `poll` for this reason.
|
||||
|
||||
`Event`s (`Axis`, `Relabel`, `Flags`, `Resync`) are what the shape's consumer applies to its
|
||||
structure. They are deduplicated by `events_through`, because an axis change is not idempotent.
|
||||
`Resync` means positions are unknown: re-derive them from what is held.
|
||||
|
||||
## 5. The Eager stand-in for a shape key
|
||||
|
||||
The real key is the circuit's: axes, None categories, filters, pages, `measure_agg`, in canonical id
|
||||
form. Here a shape key is `eager:view/<name>` (`shape_of_view`) and the output is that view's layout.
|
||||
That is finer than the real key, which is always safe: two views that would share state do not, and
|
||||
nothing else is lost. A view that no longer exists is announced as evicted.
|
||||
|
||||
`Output` is the shape's whole output, keyed by `WireKey` with the position attached. `target` restricts
|
||||
it to a window. `moved_since` compares two outputs by value and presence; a changed position is an axis
|
||||
change, not a key.
|
||||
|
||||
**This costs O(grid) per live shape per commit.** It stands in for the circuit's delta, and it is the
|
||||
reason the module exists beside, not instead of, the incremental work: nothing else in the host reads
|
||||
the whole grid per command.
|
||||
|
||||
## 6. The host
|
||||
|
||||
`Delivery` is one value per document, held by `Host`. `Host::handle` calls `Delivery::commit` after
|
||||
each `Command::Model` that lands, and `Host::deliver` answers a `Request`.
|
||||
|
||||
**Reads see only the last committed state.** `answer` reads `Live::output`, which `commit` refreshed,
|
||||
and never the workbook. A value a failed command wrote is not in it.
|
||||
|
||||
**Rollback.** `Host::mutating_frame` clones the `Delivery` beside the app snapshot and the session
|
||||
mirrors, and `undo_frame` puts it back. The clone is cheap: `Arc` for outputs, notices and structure,
|
||||
and both the log and the outbox hold at most `capacity` notices. So a refused frame consumes no publication number and publishes nothing.
|
||||
|
||||
**The ADR and ADR 001 disagree here, and ADR 001 wins until ADR 006 exists.** Clause 9 says a frame
|
||||
whose later command fails keeps its earlier commands and their notices, "each command commits on its own
|
||||
(ADR 006)". Today's host undoes the whole frame (ADR 001 clause 4, `Host::mutating_frame`), and ADR 006 is
|
||||
not in the tree. The required test "a refused command consumes no publication number, and the commands
|
||||
before it in its frame stay published" is split in two. The first half is tested. The second is
|
||||
tested as its opposite, `commands_before_a_refused_one_are_rolled_back_with_their_notices_under_adr_001`,
|
||||
which is the test to flip when per-command commit lands (improvise-bw4p.8): the publisher already supports it, since it
|
||||
publishes whatever the host tells it committed.
|
||||
|
||||
**Nothing is published while nobody can hear it.** With no live shape and no attached transport
|
||||
(`Host::attach_notices`), a commit only advances the publication number: no structure is read, nothing is
|
||||
diffed, the log is cleared and the queued structure is forgotten. A shape that goes live, or a transport
|
||||
that attaches, re-reads the structure, so its first diff is against a known state. Without this every
|
||||
host pays for delivery on every command while nothing consumes it. The outbox exists only while a
|
||||
transport is attached, and keeps the newest `capacity` notices; a transport that falls behind sees a gap
|
||||
and asks for a correction. The axis diff is linear in the items (`diff_items`, set lookups), and its test
|
||||
counts comparisons on a counting item type rather than reading a clock.
|
||||
|
||||
**The host instance** is a random id, never a time (`Host::new` takes one from `RandomState`).
|
||||
`RandomState` has no entropy on `wasm32-unknown-unknown`, so a worker must pass a real one through
|
||||
`Host::with_instance`. A restarted host with the same id would accept a stale watermark as current.
|
||||
`Worker` still calls `Host::new` (improvise-bw4p.7).
|
||||
|
||||
**Eviction** (`evict_idle`) takes a local count and an idle length in the same unit, never a wall clock
|
||||
compared across machines. An evicted key is announced in the next notice and forgotten; naming it again
|
||||
revives it with `since` set to the current publication, so a watermark older than that is answered with
|
||||
a window.
|
||||
|
||||
## 7. Answering
|
||||
|
||||
`covered(shape, watermark)` is true when the watermark is in the current epoch, is not ahead of the
|
||||
host, and either equals the current publication or is no older than both the log's start and the time
|
||||
the shape went live. Only then are the moved keys of `(watermark, now]` complete.
|
||||
|
||||
- covered, only moved keys: a `Moved` body with those keys' current values in the window;
|
||||
- covered, with axis changes or relabels in range: a `Window` body with `carried`;
|
||||
- a `Moved::Correct` in range, or not covered: a `Window` body from `None`, which is `target - empty`.
|
||||
|
||||
A notice lists no keys for a shape whose list passed `MAX_KEYS`; it says `Moved::Correct` instead, and
|
||||
costs every reader a correction. `LOG_CAPACITY` bounds the single shared log. Both are set per host by
|
||||
`limit_delivery` so a test can reach them.
|
||||
|
||||
The token in a request is carried and echoed by the client and never read by the host.
|
||||
|
||||
## 8. Edge adapters
|
||||
|
||||
`Edge` is one end of a stream: `call(Request) -> Result<Reply, EdgeError>`. `InProcess` is the
|
||||
function call; `WireEdge` passes request and reply through JSON text, the adapter at a wire boundary,
|
||||
so the lossless properties are exercised by every test that uses it. Neither adds a `ClientFrame` or
|
||||
`ServerFrame` variant: the page does not read notices yet, and adding the frames means changing the
|
||||
browser and the worker together (ADR 008 replaces `Projection`; improvise-bw4p.10). A transport
|
||||
pushes `Host::take_notices()`.
|
||||
|
||||
## Not done, and why
|
||||
|
||||
- **The notices are not sent by the socket or worker hosts.** They need frame variants and the page
|
||||
(above).
|
||||
- **Positions and column totals are not on the wire.** A correction carries `(key, value)` and no row or
|
||||
column, and `Output` holds body cells only. The ADR attaches the position when output is sent and lists
|
||||
totals among the moved keys. Both need the render model and the totals' own keys (ADR 008, clause 10).
|
||||
- **A twin at the serving end** (a server-side watermark or filter) is not built; the ADR allows it and
|
||||
does not require it.
|
||||
- **A pushed filter** is the client's `window`, applied on every request.
|
||||
- **Hidden-item overlays, pruning queries and column widths** are ADR 008 and clause 10.
|
||||
@@ -0,0 +1,174 @@
|
||||
# Test evidence: delivery
|
||||
|
||||
What each delivery test guards and which injected fault it was seen to fail on. Keyed by test name.
|
||||
The reasoning is in `context/code/delivery.md`. Each fault was put into the source one at a time,
|
||||
the suite run, the fault reverted with `git checkout -- <file>`, and the suite run again.
|
||||
|
||||
The host-side tests are in `crates/improvise-command/src/ui/delivery_tests.rs`, the client's in
|
||||
`crates/improvise-protocol/src/delivery/client_tests.rs`, the wire's in `delivery/wire.rs`.
|
||||
|
||||
## The injected faults, by what failed
|
||||
|
||||
- **The host reports no moved keys** (`let keys = BTreeSet::new()` in `Delivery::commit`): seven
|
||||
host-side tests fail: `a_dropped_connection_or_a_failed_request_triggers_a_correction_and_the_session_resumes`,
|
||||
`a_pull_only_client_follows_the_log_without_a_correction`,
|
||||
`a_notice_names_the_moved_cells_including_a_formula_cell_it_did_not_write`,
|
||||
`values_keep_every_bit_from_the_cell_to_the_client`,
|
||||
`an_oversized_notice_says_correct_and_costs_every_reader_a_correction`,
|
||||
`a_dropped_notice_is_detected_and_a_correction_repairs_it` and the random property.
|
||||
`a_frame_of_two_commands_publishes_one_notice_each_and_no_client_asks_for_a_correction` still
|
||||
passes, on purpose: its `AddItem` makes the client refetch the window, which also fetches the write.
|
||||
- **The client treats every notice as contiguous** (`false => true` in `on_notice`):
|
||||
`a_dropped_notice_is_detected`, `a_dropped_notice_is_detected_and_a_correction_repairs_it` and the
|
||||
random property.
|
||||
- **A failed frame does not put the `Delivery` back** (`undo_frame` drops `published`):
|
||||
`a_refused_command_consumes_no_publication_number`,
|
||||
`commands_before_a_refused_one_are_rolled_back_with_their_notices_under_adr_001` and
|
||||
`a_refused_frame_is_never_read_back`.
|
||||
- **A moved correction replaces instead of subtracting held** (`target` for `target - held_there`):
|
||||
`a_moved_correction_applies_only_at_the_watermark_it_starts_from` and
|
||||
`a_moved_key_the_target_no_longer_holds_is_removed`. The same for a window correction (`target` for
|
||||
`target - held`): `a_correction_equals_the_target_and_a_second_one_is_the_zero_zset`,
|
||||
`a_correction_replaces_what_is_held_not_adds_to_it` and
|
||||
`a_correction_already_covered_is_idempotent_and_never_regresses`.
|
||||
- **A wire number loses the sign of zero** (`Num::inexact(x + 0.0)` in `WireNum::of_f64`):
|
||||
`a_float_crosses_in_canonical_text_with_the_sign_of_zero` and
|
||||
`a_float_keeps_every_bit_but_the_nan_payload`.
|
||||
- **`covered` ignores the log's start and the shape's liveness**:
|
||||
`a_watermark_the_log_covers_gets_the_moved_keys_and_one_it_does_not_gets_the_window` and
|
||||
`an_evicted_shape_is_announced_and_its_reader_corrects`.
|
||||
- **The client does not remember an axis change** (`restructured` never set):
|
||||
`an_axis_change_that_shifts_an_unmoved_cell_into_the_window_refetches_the_window`. The host's own
|
||||
`restructured` test alone fails only the random property, since the client's `window` flag also forces
|
||||
a window.
|
||||
- **`Stamp::same_epoch` ignores the host instance**:
|
||||
`a_notice_from_another_host_or_generation_asks_for_a_correction`,
|
||||
`a_stamp_from_another_epoch_is_behind_every_current_one`,
|
||||
`a_reconnect_after_a_host_restart_asks_for_a_correction_from_the_old_watermark` and
|
||||
`a_notice_from_a_restarted_host_asks_for_a_correction_too`.
|
||||
|
||||
## Entries
|
||||
|
||||
### a_refused_frame_is_never_read_back
|
||||
|
||||
The first fixture filled the document to the category limit and then refused the frame at its basis
|
||||
check, so the write never ran and the test passed with the `Delivery` not restored. A fixture that fails
|
||||
in a wider shape than the fault: the frame never reached `mutating_frame`. It now writes North, then
|
||||
loads a missing document, and asserts `Rejection::MutationRefused` so the write is known to have run. The
|
||||
rival above fails it with `77.0` held.
|
||||
|
||||
### a_corrected_state_is_the_target_and_correcting_it_again_is_the_zero_zset
|
||||
|
||||
The random property: histories of up to twelve single-command frames (writes of `0.0`, `-0.0`, both
|
||||
infinities, NaN and `0.1 + 0.2`, item adds and removes, a formula), a random window, notices delivered
|
||||
or dropped, connections lost, requests dropped, replies dropped and garbled. It ends with a `poll`, then
|
||||
asserts the client holds exactly what the grid draws in its window, and that `target - held` is the zero
|
||||
Z-set. The oracle (`drawn`) reads `GridLayout` itself and shares nothing with `Output`.
|
||||
|
||||
Without the closing `poll` a lost last notice is not detectable by a listener, and the property
|
||||
reported it as a stale client (minimal case: one `Remove`, notice dropped, no later notice).
|
||||
|
||||
The first version of the property also failed on a window of one cell after a remove, and that was a
|
||||
real gap in the correction, not in the test: the remove moved South into the window without moving its
|
||||
value. It is the reason for the `restructured` rule (`delivery.md` §3).
|
||||
|
||||
### a_frame_of_two_commands_publishes_one_notice_each_and_no_client_asks_for_a_correction
|
||||
|
||||
Step 8's first required test. The frame's first command carries one basis stating both records
|
||||
(`Host`'s frame rule: the first mutation's basis stands for the frame), which the test helper builds.
|
||||
|
||||
### commands_before_a_refused_one_are_rolled_back_with_their_notices_under_adr_001
|
||||
|
||||
Step 8 requires that the commands before a refused one stay published. Today's host undoes the whole
|
||||
frame (ADR 001 clause 4), so this test asserts the opposite and names the reason. Flip it when ADR 006
|
||||
makes each command commit on its own.
|
||||
|
||||
### a_dropped_connection_or_a_failed_request_triggers_a_correction_and_the_session_resumes
|
||||
|
||||
The three transport faults (request dropped, reply dropped, reply unreadable) each leave the client
|
||||
asking for `Ask::Correction` from the watermark it held, and the next request carries the same session
|
||||
token and that watermark.
|
||||
|
||||
### an_axis_change_that_shifts_an_unmoved_cell_into_the_window_refetches_the_window
|
||||
|
||||
A one-cell window, a remove that moves South into it. The notice lists only North, so a client that
|
||||
fetches moved keys alone ends with an empty window.
|
||||
|
||||
## Review of #348, round one
|
||||
|
||||
Each test below was written against the behaviour the review reported and failed before its fix, or, for
|
||||
the cost tests, was run against a quadratic rival.
|
||||
|
||||
### diffing_a_category_compares_each_item_a_bounded_number_of_times
|
||||
|
||||
Items are a type whose `eq` is counted. Rival: `contains` on the slices in `diff_items`, which makes the
|
||||
count quadratic and fails every size. A wall-clock test would have passed on a fast machine.
|
||||
|
||||
### a_commit_with_nothing_live_and_no_transport_reads_nothing
|
||||
|
||||
Counts `Structure::of` calls. Rival: the idle skip removed.
|
||||
|
||||
### the_outbox_and_the_clone_a_frame_takes_are_bounded
|
||||
|
||||
Rival: the outbox truncation removed, so twenty commits leave twenty queued.
|
||||
|
||||
### a_late_window_reply_from_an_older_generation_does_not_regress
|
||||
|
||||
Rival: the older-generation guard disabled (fails this test in the protocol crate).
|
||||
|
||||
### a_late_notice_from_a_dead_host_settles_against_the_live_one
|
||||
|
||||
Rival: the duplicate arm no longer prunes other epochs' entries; the random property also fails.
|
||||
|
||||
### a_poll_for_one_shape_leaves_the_other_shapes_alone
|
||||
|
||||
Rival: pulled notices applied to every shape.
|
||||
|
||||
### a_transpose_empties_a_partial_window_of_the_cell_that_left_it
|
||||
|
||||
A transpose moves no key. Rival: `flags.views` not counted as a structural change in the client and the
|
||||
host. `a_dropped_views_notice_is_answered_with_the_window_not_an_empty_key_list` fails with the host
|
||||
alone changed.
|
||||
|
||||
### wide_integers_are_canonical_on_input
|
||||
|
||||
Both spellings are refused outside their range: a JSON number at or above 2^53, a string below it, a
|
||||
sign, leading zeros and trailing space.
|
||||
|
||||
### an_item_insert_position_crosses_as_a_wide_integer
|
||||
|
||||
`ItemInserted.at` is a `u128` through `wide`.
|
||||
|
||||
### a_fetch_covers_only_the_notices_it_named_when_a_later_commit_is_already_in_the_state
|
||||
|
||||
Rival: a fetch answered with `through: at`. The first version of the random property did not catch it (it
|
||||
delivered each notice before the next commit); the property now splits, delays, duplicates and replays
|
||||
notices and old replies, and fails the rival in most runs (17 of 20 at 256 cases, measured on the
|
||||
author's machine); this test fails it on every run, by name.
|
||||
|
||||
### equal_values_compare_equal_whatever_the_zero_counts_say
|
||||
|
||||
`ExactSum`'s equality compared the sign-of-zero counters even when the total was nonzero.
|
||||
|
||||
## Review of #348, round two
|
||||
|
||||
### a_late_window_reply_from_a_dead_host_does_not_regress
|
||||
|
||||
A client that has applied a correction from host 2 receives a late `Window` reply from host 1. Epochs of
|
||||
different hosts are unordered, so the first fix (older generation of the same host) did not cover it.
|
||||
Rival: the `left` check removed; this test and the two below fail.
|
||||
|
||||
### a_reply_from_a_generation_the_client_left_does_not_return
|
||||
|
||||
Generations 1, 2, 3 of one host; a late generation-2 reply. Rival: both guards removed.
|
||||
|
||||
### a_reply_held_across_a_host_restart_cannot_pull_the_client_back
|
||||
|
||||
The host-side form: the reply is computed and lost, the host restarts, the client settles against the new
|
||||
host, and the old reply then arrives. Rival: the `left` check removed.
|
||||
|
||||
**The random property, round two.** The property now checks after every step that the watermark's epoch never goes back to one it had left
|
||||
(so a mid-history regression cannot be repaired by the closing `poll`), and replays an earlier reply after
|
||||
every restart and swap. Rival: the `left` check removed fails it. The older-generation guard on its own is
|
||||
not separable from `left` in the histories the property reaches, so it is held by
|
||||
`a_late_window_reply_from_an_older_generation_does_not_regress` alone.
|
||||
@@ -466,3 +466,30 @@ factors (3, 77, 2^20, 3^40), so the lcm and the product differ.
|
||||
The rival combines over the product of the denominators. Its value is right, so only the
|
||||
lcm assertion can see it. That is why the property asserts the denominator before the
|
||||
value.
|
||||
|
||||
### a_sum_of_negative_zeros_is_negative_zero
|
||||
|
||||
`ExactSum` kept only limbs, so every zero sum read `+0.0` and a lone `-0.0` addend lost its sign
|
||||
(ADR 009 clause 8 asks for IEEE's). Injected fault: `negative_zero()` always false. This test, the
|
||||
property below and the retraction property all fail, each by name.
|
||||
|
||||
### a_zero_sum_has_the_sign_ieee_gives_the_same_addends
|
||||
|
||||
The oracle is an IEEE fold from `-0.0` over the same addends. The generator builds zero sums on
|
||||
purpose (zeros of either sign plus cancelling pairs, shuffled), because filtering random addends
|
||||
for a zero sum rejected nearly every case. Rival: the sign rule without the `others` count
|
||||
(`neg_zeros > 0` alone) fails `one_positive_zero_makes_the_zero_positive_and_so_does_a_cancellation`
|
||||
and both properties, since one `+0.0` or one cancelling pair must make the zero positive.
|
||||
|
||||
### a_retracted_addend_leaves_the_sign_of_the_rest
|
||||
|
||||
Adds `keep` and `gone`, then retracts `gone`; the read and the sum itself must equal a fresh sum
|
||||
of `keep`. It is why both counters are net weights and why `PartialEq` compares the sign.
|
||||
|
||||
### a_sum_that_is_not_zero_never_reads_as_negative_zero
|
||||
|
||||
The first version of the sign rule returned `-0.0` whenever the `-0.0` count was positive and the
|
||||
other count was nil, without checking that the finite total was zero. The weights here (+5.0 and -7.0
|
||||
net to one "other" addend, and the `-0.0` is held) make the counts say "negative zero" over a total of
|
||||
-2. Found by the existing `the_accumulator_reads_the_exact_sum_or_its_stated_roundings`, which read
|
||||
`#i-0.0` for a nonzero sum; the rival is that version.
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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(¬ice));
|
||||
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
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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 += δ
|
||||
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(¬ice(1, 5, &[]));
|
||||
c.on_notice(¬ice(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(¬ice(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(¬ice(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(¬ice(1, 5, &[3]));
|
||||
let before = c.wants(&shape());
|
||||
assert_eq!(c.on_notice(¬ice(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(¬ice(1, 5, &[1]));
|
||||
c.on_notice(¬ice(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(¬ice(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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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)),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
@@ -69,6 +69,7 @@ const SECTIONS = [
|
||||
["context/code/gp.md", "code/gp", "gp-tree and improvise-gp"],
|
||||
["context/code/numtower.md", "code/numtower", "numtower, the numeric tower"],
|
||||
["context/code/zset.md", "code/zset", "zset, the Z-set group"],
|
||||
["context/code/delivery.md", "code/delivery", "delivery, notices and corrections"],
|
||||
["context/code/improvise-incremental.md", "code/improvise-incremental", "improvise-incremental, the plug"],
|
||||
["context/code/kb-tool.md", "code/kb-tool", "The kb tool"],
|
||||
["context/code/tooling.md", "code/tooling", "The repository's own checkers"],
|
||||
@@ -90,6 +91,7 @@ const SECTIONS = [
|
||||
["context/test-evidence/gp.md", "test-evidence/gp", "gp-tree and improvise-gp"],
|
||||
["context/test-evidence/numtower.md", "test-evidence/numtower", "numtower"],
|
||||
["context/test-evidence/zset.md", "test-evidence/zset", "zset"],
|
||||
["context/test-evidence/delivery.md", "test-evidence/delivery", "delivery"],
|
||||
["context/test-evidence/improvise-incremental.md", "test-evidence/improvise-incremental", "improvise-incremental"],
|
||||
["context/test-evidence/tooling.md", "test-evidence/tooling", "The repository's own gates"],
|
||||
["context/test-evidence/layout-cost.md", "test-evidence/layout-cost", "Layout and keypress cost"],
|
||||
|
||||
Reference in New Issue
Block a user