Merge dbsp/model-over-source into dbsp/live-source

# Conflicts:
#	context/code/source.md
This commit is contained in:
Claude (dbsp-sources)
2026-10-10 18:58:28 -07:00
committed by Edward Langley
31 changed files with 3011 additions and 119 deletions
+18 -1
View File
@@ -119,6 +119,14 @@ dead host's replies are refused. A duplicate answer also prunes every queued ent
and moves `seen` onto the live epoch, so a late notice from a dead host costs one extra round and cannot
livelock `sync`.
**The host that answers a request is live, whatever the client believed.** `left` is a guess, because
epochs of different hosts are unordered: a late reply from a host the client never joined moves it onto
that host, and writes off the live one. Refusing the live host's answers as "left" then strands the client
(`sync` returned `Exhausted`, found by the property test). So `apply_answer`, which `sync` and `poll` use
for the reply to a request they just sent, takes an epoch off `left` before it applies. `apply` is for a
reply that arrives on its own, and keeps the guard. A stale reply can still move the client for one round;
the next request it makes is answered by the live host and brings it back.
`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.
@@ -182,7 +190,9 @@ 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).
`Worker` still calls `Host::new` (improvise-bw4p.7). Such a restart, with the same instance id and
generation 0, is not recognised as a restart: the client's `older` guard refuses its replies as coming from
an older generation of the host it is on (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
@@ -207,6 +217,13 @@ The token in a request is carried and echoed by the client and never read by the
## 8. Edge adapters
**Transport contract.** A reply must reach the call that made the request. `Request` and `Correction`
carry no id, so the client cannot tell a late reply from an answer on its own; `sync` and `poll` treat
whatever `Edge::call` returns as the answer (`apply_answer`), and only a reply handed to `apply` is
treated as late. A transport that reorders or duplicates replies must therefore correlate them itself
or drop the strays (improvise-bw4p.10 adds the id to the frames).
`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
+13 -4
View File
@@ -65,10 +65,13 @@ the owned `DataStore`. That is the whole evaluator hook (improvise-rnak, bw4p.14
there is still one evaluator, `Model`'s, and it computes every formula value, over a source whose
cells it never owns.
1. **A model over a source is read-only.** `set_cell` and `clear_cell` `debug_assert` that no source
is set, because they write the owned store, which is then not read. A release build would drop the
write silently; the assert is the guard, not the design. The design is that a caller who holds a
source writes to the source (ADR 003: the trait has no writes), and the model is built over it.
1. **A model over a source is read-only, in every build.** `set_cell`, `clear_cell`, `remove_category` and
`remove_item` `assert!` that no source is set (a `debug_assert` first shipped and was lost in release:
the write went to the owned store, which nothing reads). A caller who holds a source writes to the
source (ADR 003: the trait has no writes). `Model::over` drops the model's own cells, so no stale copy
is left for `get_cell`, persistence or the harness to read. `data` is `pub`, so a caller can still write
it directly; it is not read, and nothing prevents that. Structural edits (a category or item removed)
are refused too, because the source's cells would still carry them.
2. **A shared source changes behind its readers by interior mutability of its own.** The model holds
an `Arc`, so a database connection or a test double behind a lock is the same object the
backend feeds from.
@@ -96,3 +99,9 @@ now plus the changes drained later counts no change twice or not at all. Why, an
uses it: `context/code/improvise-incremental.md`, step 7. `DataStore` is not a `ChangeSource`: its log
is `&mut` and is drained by the owner in the step that made the edit. A `SharedFeed` is the `Arc` a
backend and a model share.
## Building a `Summary` from an engine's answer
`Summary::from_parts(Numbers::from_parts(count, exact_sum, least, greatest), Agreement)`. `Agreement`
is what the non-number values hold: `None`, `One(v)` or `Mixed`. `from_parts` drops extremes that
contradict the count or are NaN, so a source cannot hand the evaluator a state `add` could not reach.
+22 -6
View File
@@ -33,11 +33,13 @@ link, so they call the `extern "C"` functions as plain functions.
4. **Failure leaves the Z-set unchanged.** `add`, `sub`, `neg`, `scale` and `decode_add`
compute a new value and then assign it, so an overflow half way through cannot leave a
half-applied update (`zs_zset_insert` is one addition, so it has no half).
5. **A failed step poisons the circuit.** A panic in `step` leaves operator state half
updated, and the circuit has no rollback (ADR 009 clause 11 is the host's; `zset.md`). So the first
failure in a step, or in a rebuild step, marks the circuit poisoned, and every later call
that uses it returns `ZS_ERR_POISONED`. A caller drops it and builds another. Ids that do
not exist, and malformed arguments, are refused before any state changes and do not poison.
5. **A failed step poisons the circuit, and a rollback revives it.** A panic in `step` leaves
operator state half updated. So the first failure in a step, or in a rebuild step, marks
the circuit poisoned, and every later call that uses it returns `ZS_ERR_POISONED`. The one
way out is `zs_circuit_rollback` of a transaction that the failed step ran in
(decision 13); a poison from outside a transaction is permanent, and the caller drops the
circuit and builds another. Ids that do not exist, and malformed arguments, are refused
before any state changes and do not poison.
6. **Buffers.** The caller owns every buffer it passes and may reuse it once the call
returns; the library copies. The library hands out no memory except handles. Output of
variable size uses a two-call protocol: `zs_zset_encode(z, buf, cap, &needed)` always sets
@@ -83,6 +85,20 @@ link, so they call the `extern "C"` functions as plain functions.
12. **Versioning.** `zs_abi_version()` and `ZS_ABI_VERSION` are 1. A change that breaks a
signature or a code bumps it.
13. **Transactions** (improvise-bw4p.5; `zset.md`, "Transactions"). `zs_circuit_begin`,
`zs_circuit_commit` and `zs_circuit_rollback` wrap `Circuit::begin/commit/rollback`; a
handle-returning `snapshot` was not chosen, because the circuit logs what changes and there
is no snapshot object to free. The new code is `ZS_ERR_TRANSACTION` (10) for a `begin` on an
open transaction, a `commit` or `rollback` with none open, and a `zs_circuit_take` while
one is open (take after commit: a rollback subtracts the step's output from the sink).
`begin` on a poisoned circuit is `ZS_ERR_POISONED`, and so is `commit` after a failed step:
the host rolls back. `rollback` is the only call that clears the poison, and only a poison
that came from a step inside the transaction. This is additive, so `ZS_ABI_VERSION` stays 1.
What a rollback cannot undo is a callback's own side effects and a `ZsRebuild` stepped in the
transaction, finished or not (it answers `ZS_ERR_INTERRUPTED` after, and the host starts another job; `rollback_abi.rs`). A
circuit left mid-step by a non-local exit answers `ZS_ERR_POISONED` to a rollback like to
everything else: the frames are skipped and the engine is not touched.
A rebuild is one logical tick however many `zs_rebuild_step` calls it takes (`zset.md`,
decision 8): a `delay` fed by it shows the previous tick once, and `done` is 1 as soon as the
last chunk ran, whatever the deadline callback says.
@@ -108,7 +124,7 @@ resolve to whatever is current, and only the nightly date is fixed. It passed at
this commit (about 12s):
```
nix shell --impure --expr 'let p = import <nixpkgs> { overlays = [ (builtins.getFlake "github:oxalica/rust-overlay").overlays.default ]; }; in p.rust-bin.nightly."2026-09-24".default.override { extensions = [ "miri" "rust-src" ]; }' --command cargo miri test -p zset-ffi --lib --test abi
nix shell --impure --expr 'let p = import <nixpkgs> { overlays = [ (builtins.getFlake "github:oxalica/rust-overlay").overlays.default ]; }; in p.rust-bin.nightly."2026-09-24".default.override { extensions = [ "miri" "rust-src" ]; }' --command cargo miri test -p zset-ffi --lib --test abi --test rollback_abi
```
It is not a CI job. A proposed one: a `miri` job in `.github/workflows/ci.yml` that takes the
+84 -4
View File
@@ -124,6 +124,7 @@ Clause 1's deferred scope, built without ADR 008. The ADR is not edited; this is
| `aggregate`, `aggregate_in` | group-by, re-folds only touched keys (`AggregateState`) |
| `Trace`, `MemTrace` | operator state behind lookup, range, apply |
| `Rebuild`, `Progress` | a snapshot loaded as a resumable job |
| `begin`, `commit`, `rollback` | an undo-log transaction over the circuit's state (below) |
Decisions, where the ADR left a choice:
@@ -185,7 +186,86 @@ Decisions, where the ADR left a choice:
The tests are `crates/zset/tests/circuit.rs`; the rivals and where they failed are in
`context/test-evidence/zset.md`.
A panic inside `step` (a weight overflow in an operator, say) leaves the circuit unusable:
nodes ahead of the panic have updated their state and the rest have not, and there is no
rollback. Drop the circuit and build another. ADR 009 clause 11's per-command snapshot and
restore is the host's job, not this crate's; `zset-ffi` poisons the circuit for this reason.
## Transactions (improvise-bw4p.5)
A panic inside `step` (a weight overflow in an operator, say) leaves nodes ahead of the panic
updated and the rest not. ADR 009 clause 11 says each command snapshots the circuit for
rollback; `Circuit::begin`, `commit` and `rollback` are that, and the ADR is not amended (the
clause's wording stands; this is how the crate reads it).
1. **An undo log, not a snapshot.** The ADR budgets memory, and a clone of the circuit's state
is the size of the circuit. `begin` instead makes every stateful node start logging, and
the log holds what the steps touch: the cost of a transaction is the keys a step touches,
not the circuit's size (`a_transaction_costs_the_keys_a_step_touches_and_not_the_size_of_the_circuit`
measures it in key clones: 16 extra for `begin; step; commit`, 22 for `rollback`, the same at
10 and at 300 keys of state, with an `integrate` in the pipeline). Outside a transaction
nothing is logged and nothing extra is cloned. Two costs are the circuit's size, and are
limits, not features:
- **A split tick.** A `step_continue` that is the first publish of an edge in a transaction
clones that edge's whole value (2009 key clones against 1006 outside one, at N=1000;
`a_split_tick_in_a_transaction_costs_the_size_of_the_edge`).
- **Memory held until `commit`.** Every edge's previous value is moved into the slot's undo
record and kept, so an `integrate -> delay` pipeline holds 4022 live keys during a
transaction against 2014 without one (N=1000). An integrate edge is back to three copies
of its integral for the length of the transaction.
2. **What is logged, per node.** Each integrated Z-set (`integrate`, `distinct`, a sink's sum)
keeps `(key, weight)` for every addition it applied; each trace (the join's two, the
aggregate's groups) keeps `(key, value, weight)` for every `apply`, through `Undoable`, a
wrapper over the caller's `Trace`; a `Delay` keeps one clone of what it holds on its first
eval, which is the previous tick's input; a `Source` keeps a clone of what was pending at
`begin`; an edge (`Slot`) keeps the `(piece, whole)` its first publish replaced, moved and
not cloned, except for a continued piece, which clones the edge once. `ticks` is restored.
3. **Rollback replays newest first.** Each undo is the exact inverse of an addition that
succeeded, so it never leaves `i64`, but only in this order: a key at `MAX` lowered and
raised again would reach `MAX + 1` undone oldest first
(`rollback_undoes_newest_first_so_no_undo_leaves_i64`). `i64::MIN` has no negation; a
trace undoes it as `MAX` then `1` (the weight before was non-negative, so neither leaves
`i64`), and a Z-set undoes it by `adjust(.., subtract)`.
4. **An entry is built before its change and logged after it.** An overflow panics before a
key changes (`insert` is one checked addition), and the entry's clones are made first, so
a panicking `Clone` changes nothing either (`c1_a_clone_panic_anywhere_in_a_step_is_rolled_back`
sweeps a panic over every key clone in a step). A step that dies half way has therefore
logged exactly the keys it applied. A custom `Trace` must do the same: a panicking `apply`
leaves the trace as it was. `MemTrace` does, since it makes every clone before it touches
its map (`a_panicking_value_clone_leaves_a_memtrace_unchanged`).
5. **The edge's two copies.** An integrate stream's piece is also its output, so a second copy
(the whole-tick value) only exists while a tick is split. `Slot::whole` is `None` otherwise
and `current()` reads the piece. `piece()` lends a borrow rather than a clone
(improvise-bw4p.11). An integrate stream now holds its integral twice (the node's `acc`
and the stream's piece), not three times; sharing those two needs an `Rc` in the slot and
was not done.
6. **What a transaction does not undo.**
- A closure's captured state (a `map` function that counts, an FFI callback's side
effects). Operators that keep state keep it in the crate's own types for this reason.
- A `Rebuild`: it owns an iterator over its snapshot, so a rolled-back chunk leaves the job
ahead of the circuit. The clock cannot tell: `rollback` rewinds `ticks`, which can make
a dead job's recorded tick current again. So the circuit also counts the rollbacks that
undid a step or piece (`undone`, never rewound), and a job records the count when it first
runs, before its first chunk can panic. The chunk's `push` is marked as a step too, since it
can overflow the pending delta before any step runs (`p1_a_chunk_whose_push_panics_in_a_txn_is_counted`). A later count makes `interrupted` true for good,
finished or not (`rollback_rebuild.rs`: begun before the transaction and continued inside,
finished inside, a first chunk that panicked, and a dead job revived by a later step).
- An `Output::take` made inside the transaction: a rollback subtracts the step's output
from the sink, which a `take` has emptied. Take after `commit`; `zset-ffi` refuses to take
while a transaction is open.
7. **Pushes.** `Input::push` is not logged (a handle has no circuit). `rollback` puts the
pending delta back to what it was at `begin`, so a push made after `begin` is dropped, and
one made before it is pending again. Operators built after `begin` start logging at once
and go back to empty; they are not removed.
8. **One transaction at a time, and misuse panics** like an overflow does: `begin` while open,
`commit` or `rollback` with none open. `zset-ffi` answers with a code instead.
The tests are `tests/rollback.rs` (the property: a step that panics half way, then a rollback,
leaves every stream and sink as before and the later ticks equal to a circuit that never ran
the step) and `tests/cost.rs`. A step that panics outside a transaction still leaves the
circuit unusable: drop it and build another.
## Rebuilds and circuits (improvise-bw4p.12)
A `Rebuild` takes its circuit's identity from its `Input`: the circuit holds an `Rc<()>` and every
`Input` it makes holds a clone, so `Rebuild::belongs_to(&circuit)` is a pointer comparison and no
global counter is needed. `Rebuild::step` on another circuit panics, where it used to report
`Done` and land its chunk in that circuit's next tick. An interrupted job (decision 8 above) can
never finish: `interrupted` stays true however many steps follow, because a split tick cannot
be mended and a rollback that took a chunk is counted forever (decision 6 above). Drop it and start another job from the same snapshot; the host keeps the
snapshot.
+42
View File
@@ -2380,3 +2380,45 @@ branch reads the owned store; it fails here.
`set_cell` on a model over a source panics in a debug build, naming the reason. Without the assert the
write goes to the owned store and nothing reads it.
### a_summary_built_from_an_engines_parts_reduces_as_one_built_from_the_values
Review of #356 (Codex): an engine outside core could not build a `Summary` without replaying rows. The
parts (count, exact sum, extremes, agreement of the non-numbers) are computed in the test without
`Summary::of`, for the 585 sequences of the reference test. Rivals: the greatest taken from the least;
a NaN allowed as an extreme. Each fails it.
### parts_that_contradict_themselves_are_made_consistent
Extremes with no numbers, and a NaN as an extreme, are dropped by `Numbers::from_parts`.
### keys_of_a_store_are_its_cells_keys_without_the_values
Review of #356 (Codex): `DataStore::keys` went through `cells` and cloned every value. It reads keys
only now; the test pins the keys and their order against `cells`. No clone-count test: a value clone is
not observable from outside, so the equality is all that can be asserted.
### a_model_over_a_source_refuses_a_clear
Review of #357: the `clear_cell` guard had no test. The refusal is an `assert!`, so the test holds in a
release build too (`cargo test --release -p improvise-core --lib over_tests` was run: the earlier
`debug_assert!` failed all four refusal tests there, `a_model_over_a_source_refuses_a_write` among
them, and was the bug). Rival: the guard removed from `clear_cell` fails this test only.
### a_model_over_a_source_refuses_to_remove_a_category
`remove_category` swept the owned store with no guard. Rival: the guard removed.
### a_model_over_a_source_refuses_to_remove_an_item
The same for `remove_item`.
### putting_a_populated_model_over_a_source_drops_its_own_cells
`Model::over` empties the owned store, so `get_cell` on an over-model is `None` for a cell the
source holds, and the evaluator still reads the source's value.
### records_mode_reads_the_records_from_a_source
Review of #357: a records-mode `GridLayout` over a source. `group_records` reading the owned store
left all 563 tests green; with this test it fails (the grid has no rows).
+20
View File
@@ -172,3 +172,23 @@ host, and the old reply then arrives. Rival: the `left` check removed.
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.
## The left-epoch livelock
### a_late_reply_from_a_host_the_client_never_joined_cannot_strand_it
The property test failed with `Exhausted` on a shrunk history of two restarts and four writes. Cause: the
client was on host 3, never having joined host 2; a late host-2 reply was applied (an unknown epoch is
accepted), which wrote host 3 into `left`, and every later answer from the live host 3 was then refused as
a duplicate. The test replays that history by hand: three hosts, the host-2 reply lost in flight, the client
settling on host 3, the late reply applied, the connection lost, and a sync that must settle on host 3's
grid. Rival: `apply_answer` treated like `apply` (the `left.retain` removed) fails this test. **The random
property does not catch it**: it passed 11 of 11 runs under that fault, and again after the late reply was
made an explicit event (a stashed reply from a host the client never joined, replayed before a lost
connection), because the property rarely produces a reply from a host the client never joined. This
test is the pin. Rival: the left-epoch guard removed fails
`a_reply_held_across_a_host_restart_cannot_pull_the_client_back` and the property, as before.
The property's invariant changed with it. "The watermark never returns to an epoch it left" was false by
design once an unknown-epoch reply can move the client, so the property now checks the narrower, true
rule: a replayed reply from an epoch the client was in and has left must be a duplicate.
+55
View File
@@ -166,3 +166,58 @@ A job outlives its circuit; sixty-four new circuits (some at the old address) al
Retyping `ZsKeepFn`'s return in the header fails. The `unary_op!` signatures are now read
from the macro body rather than written into the test.
## Transactions (improvise-bw4p.5)
Rivals were injected one at a time in `engine.rs` and watched fail by name; Miri passes on `abi`.
### a_rollback_after_a_failed_step_revives_the_circuit_in_its_old_state
An integrate held at `i64::MAX`; inside a transaction a step overflows (`ZS_ERR_OVERFLOW`, then
`ZS_ERR_POISONED`), `commit` of it is refused as poisoned, and `rollback` revives the circuit with
the integral where it was: `-1` then gives `MAX - 1`, with neither the failed `+1` nor the key
added beside it. Rivals: a rollback that leaves the poison on, and one that does not roll the
circuit back, fail this test and the callback one; a `commit` that does not check the poison
fails this test.
### a_committed_transaction_keeps_its_steps_and_a_new_one_may_follow
`commit` keeps the step and its sink, and a later transaction can be opened and rolled back.
### a_failed_callback_step_is_rolled_back_too
A failure the callback only reports (the step returns, the fault is noted) is rolled back like a
panic: the sink holds nothing after the rollback.
### transaction_misuse_is_a_code_and_changes_nothing
`ZS_ERR_TRANSACTION` for `commit` or `rollback` with none open, a second `begin`, and a `take`
while one is open (nothing written to the out-pointer). A rollback after a poison from outside any
transaction is also refused and the circuit stays poisoned. Rivals: `take` allowed in a
transaction, and a second `begin` allowed, each fail this test.
## Review of #355 at 51438911
### f1_rebuild_begun_before_a_txn_continued_inside_and_rolled_back
Through the ABI, a job continued inside a rolled-back transaction returned 0 with `done` 1 and
a partial load. It is now `ZS_ERR_INTERRUPTED`.
### f4_a_failed_rebuild_step_in_a_txn_then_rollback_then_resume
A job whose first chunk failed inside the transaction resumed without that chunk; now
`ZS_ERR_INTERRUPTED`.
### f2_a_rollback_from_inside_a_callback_is_refused_and_the_txn_stays_open
`ZS_ERR_POISONED` from the callback's rollback, and the transaction still commits.
### f3_begin_on_a_circuit_poisoned_outside_a_txn_is_poisoned
`begin` is `ZS_ERR_POISONED`; `commit` and `rollback` are `ZS_ERR_TRANSACTION`.
### f5_a_chunk_whose_push_overflows_in_a_txn_then_rollback_is_interrupted
The same path through the ABI: `zs_rebuild_step` returns `ZS_ERR_OVERFLOW` from the chunk's
push, `zs_circuit_rollback` succeeds, and the next `zs_rebuild_step` was a silent success; it is
now `ZS_ERR_INTERRUPTED`. Rival: the mark before the push removed fails `[f5]`.
+167
View File
@@ -429,3 +429,170 @@ A second job's first chunk is a tick, so the first job's next call panics.
### a_finished_rebuild_and_an_interrupted_check_do_not_trip_on_later_ticks
A finished job is never interrupted, and steps afterwards do not make it panic.
## `crates/zset/tests/{cost,rollback,rebuild}.rs`: rollback, the borrowed piece, the bound rebuild (improvise-bw4p.5, .11, .12)
Every rival below was injected one at a time and watched fail by test name (and tag where one
exists) before being reverted; the whole suite was re-run clean afterwards. Costs are counted
in key clones and live keys of a `Counted` key type, never in time, so they hold on any machine.
### an_eval_borrows_its_input_instead_of_cloning_it
bw4p.11. `input -> map(|k| k.0) -> output` over 100 keys makes **0** key clones in a step; before
the change it made 200 (the source's whole-tick copy and `piece()`'s clone). Rival: `Unary::eval`
clones the piece again fails `[piece]`.
### an_integrate_stream_holds_its_integral_twice_when_the_tick_is_whole
bw4p.11. After one step an `input -> integrate` pipeline holds 3N live keys (the input's
piece, the integral, the integrate stream's piece), where it held 5N before: the whole-tick copy
that duplicated the piece is kept only while a tick is split (`Slot::whole` is an `Option`).
Rival: a publish that always stores `whole = Some(v.clone())` fails `[whole]` (and the clone test).
### a_transaction_costs_the_keys_a_step_touches_and_not_the_size_of_the_circuit
bw4p.5. A pipeline with `distinct`, `delay`, `join_indexed`, `aggregate` and three sinks is
loaded with 10, then 300 keys, then given a two-key delta in each of two inputs. The extra key
clones of `begin; step; commit` over the same step outside a transaction are **16** at both
sizes (22 for `rollback`; the plain step is 20). Rival: `DistinctState::begin` clones what it
holds (a snapshot of the world) fails `[cost]` with 28 against 318. A `delay` clones what it
holds, the previous tick's input, so the test steps a small delta after the load. Review of
#355: the pipeline now includes an `integrate`; a rival `Acc::begin` that clones the integral
fails `[cost]`. (An `integrate` whose output is sunk is left out: its output is the integral,
so the sink logs all of it, which is the delta and not an overhead.)
### a_rolled_back_step_leaves_the_circuit_as_it_was_and_the_ticks_after_unchanged
bw4p.5, the property. A rig with every stateful operator (integrate, delay, differentiate,
distinct, join, aggregate, plus and eight sinks) runs a random prefix, then a key preloaded at
`i64::MAX - 1`. The failing step adds `(0,0): 1` and then `BIG: 5`, so operators have applied
the first key when the second overflows (the test asserts the step panicked iff it was meant
to). After `rollback`, every stream's `current()` and every sink equal what they were (`[rollback]`),
and the ticks after equal a reference circuit that never saw the step (`[post]`). The same test
rolls back a step that succeeds. `carry` is pending at `begin` and must stay pending.
Rivals, each failing this test: rollback that skips every node (`[rollback]`, and
`an_operator_built_inside_a_transaction_goes_back_to_empty`); an integrate log not undone
(`[post]`); a stream slot not restored (`[rollback]`); a `delay`'s held value not restored
(`[post]`); a source's pending delta not restored (`[post]`); a sink not undone (`[rollback]`);
`DistinctState` not undone (`[post]`); a trace not undone (`[post]`); an undo that drops the sign
(`[post]`); a log that records nothing (`[rollback]`, `[built]`).
### a_committed_transaction_equals_the_same_steps_outside_one
bw4p.5. A circuit that commits some of its steps equals one that never opened a transaction,
state and output, tick by tick. Rival: `commit` rolls back fails `[commit]`.
### rollback_undoes_the_clock
bw4p.5. A rolled-back step does not count as a tick. Rival: `ticks` not restored fails `[ticks]`.
### an_operator_built_inside_a_transaction_goes_back_to_empty
bw4p.5. An `integrate` built after `begin`, stepped, then rolled back, forgets the step
(`[built]`). Without the "a node built in a transaction starts recording" rule it kept it.
### rollback_undoes_newest_first_so_no_undo_leaves_i64
bw4p.5. A key at `i64::MAX` lowered by one and raised by one in one transaction. Undone oldest
first the first undo would reach `MAX + 1` and panic; undone newest first nothing leaves
`i64`. Rival: the log replayed oldest first fails this test (and no other: the property above
cannot see it, since its keys never sit at the boundary).
### a_minimum_weight_applied_to_every_kind_of_state_is_taken_back
bw4p.5. `i64::MIN` has no negation, so a trace undoes it as `i64::MAX` then `1`. Rival: the
second application dropped fails `[min]`.
### a_second_begin_is_refused_loudly
bw4p.5, with `a_rollback_with_nothing_open_is_refused_loudly` and
`a_commit_with_nothing_open_is_refused_loudly`: misuse panics, as an overflow does.
### a_rebuild_stepped_on_a_foreign_circuit_is_refused_loudly
bw4p.12. A job made from circuit A's input, stepped on circuit B, used to report Done and land
its chunk in B's next tick. It now panics. Rival: the check removed fails this test.
### a_rebuild_knows_the_circuit_its_input_came_from
bw4p.12. `Rebuild::belongs_to`, the question a host asks before the panic. Rival: always true
fails `[owner]`.
### an_interrupted_rebuild_stays_interrupted_and_a_new_job_loads_the_snapshot
bw4p.12. Documents, and pins, that an interrupted job can never finish: later steps never
clear it, and a new job over the same snapshot loads it. It did not fail before: the behaviour
was right and unstated.
## Review of #355 at 51438911 (improvise-bw4p.5, .12)
An adversarial review found a rebuild across a rollback losing data, a panicking `Clone`
mutating unlogged state, and two cost claims that did not hold. Its reproductions are adopted
as tests; each rival below was injected and watched fail by name.
### r1_rebuild_begun_before_txn_continued_inside_then_rolled_back_reports_interrupted
A job begun outside a transaction and continued inside, then rolled back, was not interrupted;
finishing it gave `{1,3,4}` for the snapshot `{1,2,3,4}`. Rival: `rollback` does not count the
undone step fails this and r2 to r4.
### r2_rebuild_finished_inside_a_rolled_back_txn_is_reported_done_with_nothing_loaded
A job that finished inside a rolled-back transaction was Done with an empty integral and not
interrupted.
### r3_an_interrupted_rebuild_resurrects_after_rollback_and_a_later_step
`rollback` rewinds `ticks`, so a later `step` made a dead job's recorded tick current again and
it continued without its first chunk.
### r4_rebuild_whose_first_chunk_panics_in_a_txn_then_rolled_back
A first chunk that panicked left `began_at` unset, and the job resumed without it. The count is
now recorded before the chunk runs. Rival: recording it never (the `get_or_insert` removed)
fails all four; a rival recording it after the chunk was not run separately.
### b1_a_closure_that_reads_a_stream_during_eval_does_not_panic
The borrowed `piece()` does not conflict with a closure that reads streams during an eval,
including its own source, inside a transaction. It passed before and after.
### c1_a_clone_panic_anywhere_in_a_step_is_rolled_back
Sweeps a panic over every key clone of one transactional step through `integrate`, `distinct`,
`join`, `delay`, `aggregate` and sinks; 138 panic points left state that differed after the
rollback. The post-rollback tick now pushes the keys the failed step touched, because a leaked
key is only visible when it is touched again (a rival that mutated `DistinctState` before
cloning its entry survived the first version). Rivals, each failing it: `add_logged`,
`DistinctState::step` and `Undoable::apply` each mutating before cloning the entry.
### a_panicking_value_clone_leaves_a_memtrace_unchanged
`MemTrace::apply` inserted an empty group before cloning the value. Rival: that order restored
fails `[c2]`.
### reviewer_rollback_matches_a_circuit_that_never_ran_the_step
The reviewer's model-based property (random circuits over every public operator, a custom
panicking trace, random push/step/continue/begin/commit/rollback histories, operators built in
a transaction), at the default 256 cases here (30,000 passed in review). Rival: the continued
path saving the slot without its `whole` fails it.
### a_rollback_across_a_split_tick_restores_the_whole_value
The tick is split before the transaction opens, so every edge holds a `whole` when the first
continued piece saves it. Rival: that save dropping `whole` fails `[split]` (an earlier version
split the tick inside the transaction and never reached the save).
### a_split_tick_in_a_transaction_costs_the_size_of_the_edge
Pins the limit `zset.md` states: a first continued piece in a transaction clones the edge's
whole value, growing with the state.
### p1_a_chunk_whose_push_panics_in_a_txn_is_counted
Delta review of #355 at b2445fed. `Rebuild::step` takes the chunk from its iterator and pushes
it before stepping; a push that overflows the pending delta panicked with no step run, so
`undone` was not bumped and the resumed job silently dropped the chunk (key 1's `+1`). The step
is now marked before the push. Rival: that mark removed fails `[p1]`.
@@ -1086,6 +1086,45 @@ fn a_reply_held_across_a_host_restart_cannot_pull_the_client_back() {
assert_eq!(state, Synced::Settled);
}
#[test]
fn a_late_reply_from_a_host_the_client_never_joined_cannot_strand_it() {
let window = Window {
top: 0,
left: 0,
rows: 1,
cols: 1,
};
let mut first = host_instance(1);
let mut client = EdgeClient::new(session());
client.track(shape(), window);
client.sync(&mut Flaky::new(&mut first), &shape(), 6);
let mut second = host_instance(2);
let id = connect(&mut second);
send(&mut second, id, vec![set("South", 1.0)]);
let mut lost = Flaky::scripted(&mut second, vec![Some(Fault::DropReply)]);
client.connection_lost();
client.sync(&mut lost, &shape(), 6);
let late = lost.replies.remove(0);
let mut third = host_instance(3);
client.connection_lost();
client.sync(&mut Flaky::new(&mut third), &shape(), 6);
let Reply::Correction(late) = late else {
panic!("the lost reply was a correction")
};
client.apply(&late);
client.connection_lost();
let (state, _) = client.sync(&mut Flaky::new(&mut third), &shape(), 6);
assert_eq!(
state,
Synced::Settled,
"the live host's answers must be taken"
);
assert_eq!(client.held(&shape()), Some(&drawn(&third, &window)));
}
// ── the whole thing, at random ─────────────────────────────────────────────
mod random {
@@ -1265,6 +1304,39 @@ mod random {
deliver_all(client, &notices);
}
/// Epochs the client has applied a correction from, oldest first.
fn note_epoch(client: &EdgeClient, epochs: &mut Vec<(u64, u64)>) {
if let Some(w) = client.watermark(&shape()) {
let now = (w.host, w.generation);
if !epochs.contains(&now) {
epochs.push(now);
}
}
}
/// A late reply arrives out of band. One from an epoch the client was in and has left is refused;
/// one from an epoch it never joined may legitimately move it, and the next answer brings it back.
fn replay(
client: &mut EdgeClient,
c: &improvise_protocol::delivery::Correction,
epochs: &mut Vec<(u64, u64)>,
) -> Result<(), TestCaseError> {
let theirs = (c.through.host, c.through.generation);
let current = client.watermark(&shape()).map(|w| (w.host, w.generation));
let left = epochs.contains(&theirs) && current != Some(theirs);
let (outcome, _) = client.apply(c);
if left {
prop_assert_eq!(
outcome,
Outcome::Duplicate,
"a reply from the left epoch {:?}",
theirs
);
}
note_epoch(client, epochs);
Ok(())
}
proptest! {
#[test]
fn a_corrected_state_is_the_target_and_correcting_it_again_is_the_zero_zset(
@@ -1306,9 +1378,7 @@ mod random {
}
if let Some(i) = weather.replay.filter(|_| !old.is_empty()) {
match &old[i % old.len()] {
Reply::Correction(c) => {
client.apply(c);
}
Reply::Correction(c) => replay(&mut client, c, &mut epochs)?,
Reply::Notices(list) => deliver_all(&mut client, list),
}
}
@@ -1316,20 +1386,10 @@ mod random {
let mut edge = Flaky::scripted(&mut host, script);
client.sync(&mut edge, &shape(), 6);
old.append(&mut edge.replies);
note_epoch(&client, &mut epochs);
let moved_host = matches!(op, Op::Restart | Op::Swap);
if let (true, Some(Reply::Correction(c))) = (moved_host, old.get(step % old.len().max(1))) {
client.apply(c);
}
if let Some(w) = client.watermark(&shape()) {
let now = (w.host, w.generation);
let seen_at = epochs.iter().position(|e| *e == now);
prop_assert!(
seen_at.is_none_or(|i| i + 1 == epochs.len()),
"the client went back to a superseded epoch {now:?} after {epochs:?}"
);
if seen_at.is_none() {
epochs.push(now);
}
replay(&mut client, c, &mut epochs)?;
}
}
@@ -219,6 +219,48 @@ fn a_model_over_a_source_refuses_a_write() {
over.set_cell(key("East", "A", "Revenue"), CellValue::Number(1.0));
}
#[test]
#[should_panic(expected = "read-only")]
fn a_model_over_a_source_refuses_a_clear() {
let mut over = lent(&fixture());
over.clear_cell(&key("East", "A", "Revenue"));
}
#[test]
#[should_panic(expected = "read-only")]
fn a_model_over_a_source_refuses_to_remove_a_category() {
let mut over = lent(&fixture());
over.remove_category("Product");
}
#[test]
#[should_panic(expected = "read-only")]
fn a_model_over_a_source_refuses_to_remove_an_item() {
let mut over = lent(&fixture());
over.remove_item("Product", "A");
}
#[test]
fn putting_a_populated_model_over_a_source_drops_its_own_cells() {
let owned = fixture();
assert!(owned.data.iter_cells().count() > 0);
let rows = owned
.data
.iter_cells()
.map(|(k, v)| (k, v.clone()))
.collect();
let over = owned.clone().over(Arc::new(TableSource { rows }));
assert_eq!(over.data.iter_cells().count(), 0, "no stale second copy");
let cell = key("East", "A", "Revenue");
assert_eq!(
over.get_cell(&cell),
None,
"the owned store is empty, the source is not"
);
let read = over.evaluate_aggregated(&cell, Aggregation::Exact);
assert_eq!(read, owned.evaluate_aggregated(&cell, Aggregation::Exact));
}
#[test]
fn the_records_a_drill_freezes_are_read_from_the_source() {
use crate::view::layout::drill_snapshot;
+94 -12
View File
@@ -51,6 +51,23 @@ impl Numbers {
numbers
}
/// Numbers from what an engine computed: how many, their exact sum and the least and
/// greatest that are not NaN. Extremes that contradict the count or are NaN are dropped.
pub fn from_parts(
count: u64,
sum: ExactSum,
least: Option<f64>,
greatest: Option<f64>,
) -> Self {
let held = |x: Option<f64>| x.filter(|x| count > 0 && !x.is_nan());
Numbers {
count,
sum,
least: held(least),
greatest: held(greatest),
}
}
pub fn add(&mut self, x: f64) {
self.count += 1;
self.sum.add(x, 1);
@@ -104,9 +121,9 @@ fn merged(a: Option<f64>, b: Option<f64>, pick: fn(f64, f64) -> f64) -> Option<f
}
}
/// The values that are not numbers, as far as unanimity needs them.
/// What the values that are not numbers agree on: nothing held, one value, or disagreement.
#[derive(Debug, Clone, Default, PartialEq)]
enum Others {
pub enum Agreement {
#[default]
None,
One(CellValue),
@@ -118,7 +135,7 @@ enum Others {
#[derive(Debug, Clone, Default)]
pub struct Summary {
numbers: Numbers,
others: Others,
others: Agreement,
}
impl Summary {
@@ -128,6 +145,11 @@ impl Summary {
summary
}
/// A summary from what an engine computed: the numbers, and what the rest agree on.
pub fn from_parts(numbers: Numbers, others: Agreement) -> Self {
Summary { numbers, others }
}
pub fn add(&mut self, value: &CellValue) {
match value.as_f64() {
Some(x) => self.numbers.add(x),
@@ -137,18 +159,18 @@ impl Summary {
fn add_other(&mut self, value: &CellValue) {
self.others = match std::mem::take(&mut self.others) {
Others::None => Others::One(value.clone()),
Others::One(held) if held == *value => Others::One(held),
_ => Others::Mixed,
Agreement::None => Agreement::One(value.clone()),
Agreement::One(held) if held == *value => Agreement::One(held),
_ => Agreement::Mixed,
};
}
pub fn merge(&mut self, other: &Summary) {
self.numbers.merge(&other.numbers);
match &other.others {
Others::None => {}
Others::One(v) => self.add_other(v),
Others::Mixed => self.others = Others::Mixed,
Agreement::None => {}
Agreement::One(v) => self.add_other(v),
Agreement::Mixed => self.others = Agreement::Mixed,
}
}
@@ -159,8 +181,8 @@ impl Summary {
pub fn reduce(&self, func: &AggFunc) -> Option<CellValue> {
match (&self.others, self.numbers.count()) {
(Others::None, _) => self.numbers.reduce(func).map(CellValue::Number),
(Others::One(only), 0) => Some(only.clone()),
(Agreement::None, _) => self.numbers.reduce(func).map(CellValue::Number),
(Agreement::One(only), 0) => Some(only.clone()),
_ => Some(CellValue::Error("mixed".to_string())),
}
}
@@ -220,7 +242,10 @@ impl CellSource for DataStore {
&'a self,
partial: &'a [(String, String)],
) -> Box<dyn Iterator<Item = CellKey> + 'a> {
Box::new(self.cells(partial).map(|(key, _)| key))
match partial.is_empty() {
true => Box::new(self.iter_cells().map(|(key, _)| key)),
false => Box::new(self.matching_cells(partial).into_iter().map(|(key, _)| key)),
}
}
fn cells<'a>(
@@ -569,4 +594,61 @@ mod tests {
"no cell is keyed by R=E alone"
);
}
/// The parts an engine would compute for `values`, found without `Summary::of`.
fn parts_of(values: &[CellValue]) -> Summary {
let numbers: Vec<f64> = values.iter().filter_map(|v| v.as_f64()).collect();
let mut sum = ExactSum::new();
numbers.iter().for_each(|x| sum.add(*x, 1));
let finite = || numbers.iter().copied().filter(|x| !x.is_nan());
let least = finite().min_by(f64::total_cmp);
let greatest = finite().max_by(f64::total_cmp);
let rest: Vec<&CellValue> = values.iter().filter(|v| v.as_f64().is_none()).collect();
let agreement = match rest.split_first() {
None => Agreement::None,
Some((first, others)) if others.iter().all(|v| v == first) => {
Agreement::One((*first).clone())
}
Some(_) => Agreement::Mixed,
};
let parts = Numbers::from_parts(numbers.len() as u64, sum, least, greatest);
Summary::from_parts(parts, agreement)
}
#[test]
fn a_summary_built_from_an_engines_parts_reduces_as_one_built_from_the_values() {
for values in sequences(3) {
let refs: Vec<&CellValue> = values.iter().collect();
for func in &FUNCS {
let want = reference(&refs, func);
let got = parts_of(&values).reduce(func);
assert!(
same(&want, &got),
"{func:?} of {values:?}: {want:?} != {got:?}"
);
}
}
}
#[test]
fn parts_that_contradict_themselves_are_made_consistent() {
let none = Numbers::from_parts(0, ExactSum::new(), Some(1.0), Some(f64::NAN));
assert_eq!(none.reduce(&AggFunc::Min), None, "no numbers, no extremes");
let nan = Numbers::from_parts(2, ExactSum::new(), Some(f64::NAN), None);
let min = nan.reduce(&AggFunc::Min).unwrap();
assert!(min.is_nan(), "a NaN is never an extreme");
}
#[test]
fn keys_of_a_store_are_its_cells_keys_without_the_values() {
let rows = rows();
let store = store_of(&rows);
let all: Vec<CellKey> = store.keys(&[]).collect();
assert_eq!(all, rows.iter().map(|(k, _)| k.clone()).collect::<Vec<_>>());
let east = vec![pair("R", "E")];
let under: Vec<CellKey> = store.keys(&east).collect();
let want: Vec<CellKey> = store.cells(&east).map(|c| c.0).collect();
assert_eq!(under, want);
assert!(!under.is_empty());
}
}
+12 -4
View File
@@ -430,13 +430,19 @@ impl Model {
}
}
/// This model, reading its raw cells from `source` and not `data`: read-only.
/// Why: `context/code/source.md`.
/// This model, reading its raw cells from `source` and not `data`: read-only. Its own
/// cells are dropped, so no stale second copy is left to read. Why: `context/code/source.md`.
pub fn over(mut self, source: SharedSource) -> Self {
self.data = DataStore::new();
self.external = Some(source);
self
}
/// Refuses, in every build, a write the source would never see.
fn writable(&self) {
assert!(self.external.is_none(), "{READ_ONLY}");
}
/// Add a pivot category, enforcing `MAX_CATEGORIES` over regular ones.
/// The caller registers it on every view — normally `Workbook`.
pub fn add_category(&mut self, name: impl Into<String>) -> Result<CategoryId> {
@@ -480,6 +486,7 @@ impl Model {
/// Remove a category, its cells and the formulas targeting it. The caller
/// removes it from the views — normally `Workbook`.
pub fn remove_category(&mut self, name: &str) {
self.writable();
if !self.categories.contains_key(name) || CategoryKind::of_name(name).is_virtual() {
return;
}
@@ -502,6 +509,7 @@ impl Model {
/// An item, the formulas targeting it and the cells naming it. The sweep
/// is `remove_category`'s one level down — `context/code/core.md` §9.
pub fn remove_item(&mut self, cat_name: &str, item_name: &str) {
self.writable();
if let Some(cat) = self.categories.get_mut(cat_name) {
cat.remove_item(item_name);
}
@@ -552,12 +560,12 @@ impl Model {
}
pub fn set_cell(&mut self, key: CellKey, value: CellValue) {
debug_assert!(self.external.is_none(), "{READ_ONLY}");
self.writable();
self.data.set(key, value);
}
pub fn clear_cell(&mut self, key: &CellKey) {
debug_assert!(self.external.is_none(), "{READ_ONLY}");
self.writable();
self.data.remove(key);
}
+26
View File
@@ -1348,6 +1348,32 @@ mod tests {
assert_eq!(layout.col_label(0), "X");
}
/// The records a records-mode grid draws come from the source the model is over.
#[test]
fn records_mode_reads_the_records_from_a_source() {
use crate::model::source::TableSource;
let mut wb = records_workbook();
let v = wb.active_view_mut();
v.set_axis("_Index", Axis::Row);
v.set_axis("_Dim", Axis::Column);
let rows = wb.model.data.iter_cells().map(|(k, v)| (k, v.clone()));
let over = wb.model.clone().over(std::sync::Arc::new(TableSource {
rows: rows.collect(),
}));
let (owned, lent) = (
GridLayout::new(&wb.model, wb.active_view()),
GridLayout::new(&over, wb.active_view()),
);
assert!(lent.is_records_mode());
assert_eq!(lent.row_count(), 1, "the source holds the records");
for col in 0..owned.col_count() {
assert_eq!(lent.records_display(0, col), owned.records_display(0, col));
}
let revenue = (0..lent.col_count()).find(|&c| lent.col_label(c) == "Revenue");
let revenue = revenue.unwrap();
assert_eq!(lent.records_display(0, revenue).as_deref(), Some("100"));
}
#[test]
fn records_mode_activated_when_index_and_dim_on_axes() {
let mut wb = records_workbook();
@@ -243,11 +243,26 @@ impl EdgeClient {
}
}
/// A correction. It applies only at the watermark it starts from: `held + (target - held)`.
/// A correction that arrives on its own: it applies only at the watermark it starts from,
/// as `held + (target - held)`, and never from an epoch the client has left.
pub fn apply(&mut self, c: &Correction) -> (Outcome, Vec<Event>) {
self.apply_with(c, false)
}
/// The reply to a request this client just made. The host that answers is the live one, so
/// an epoch the client had written off is live again (`context/code/delivery.md` section 4).
pub fn apply_answer(&mut self, c: &Correction) -> (Outcome, Vec<Event>) {
self.apply_with(c, true)
}
fn apply_with(&mut self, c: &Correction, answer: bool) -> (Outcome, Vec<Event>) {
let Some(t) = self.shapes.get_mut(&c.shape) else {
return (Outcome::Untracked, Vec::new());
};
if answer {
t.left
.retain(|e| *e != (c.through.host, c.through.generation));
}
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)
@@ -347,7 +362,7 @@ impl EdgeClient {
return (Synced::Settled, events);
};
match edge.call(request) {
Ok(Reply::Correction(c)) => events.extend(self.apply(&c).1),
Ok(Reply::Correction(c)) => events.extend(self.apply_answer(&c).1),
Ok(Reply::Notices(list)) => {
list.iter()
.for_each(|n| events.extend(self.accept(n, Some(shape))));
@@ -387,7 +402,7 @@ impl EdgeClient {
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),
Ok(Reply::Correction(c)) => events.extend(self.apply_answer(&c).1),
Err(EdgeError::Failed) => {
self.request_failed(shape);
return (Synced::Failed, events);
+21 -1
View File
@@ -47,6 +47,7 @@ extern "C" {
#define ZS_ERR_POISONED 7
#define ZS_ERR_CALLBACK 8
#define ZS_ERR_INTERRUPTED 9
#define ZS_ERR_TRANSACTION 10
#define ZS_AGG_COUNT 0
#define ZS_AGG_SUM 1
@@ -131,8 +132,27 @@ int32_t zs_circuit_aggregate(ZsCircuit *c, uint32_t stream, int32_t kind, uint32
int32_t zs_circuit_push(ZsCircuit *c, uint32_t input, const ZsZSet *z);
/* One clock tick. After a failure in a step the circuit is poisoned: every later call that
* uses it returns ZS_ERR_POISONED, except the free. */
* uses it returns ZS_ERR_POISONED, except the free, and zs_circuit_rollback of a transaction
* the failed step ran in. */
int32_t zs_circuit_step(ZsCircuit *c);
/* Transactions: undo a failed command. zs_circuit_begin opens one; zs_circuit_commit keeps
* what its steps did; zs_circuit_rollback puts the circuit's state back to what it was at
* begin, and revives a circuit that a failed step in the transaction poisoned. The cost of a
* transaction is the keys its steps touch (plus a delay's held input), not the circuit's size.
* - At most one is open. begin on an open one, and commit or rollback on none, is
* ZS_ERR_TRANSACTION. begin on a poisoned circuit is ZS_ERR_POISONED; so is commit after a
* failed step (roll back instead). A poison from outside a transaction is permanent.
* - Pushes made after begin are dropped by a rollback; what was pending at begin is pending
* again. zs_circuit_take is ZS_ERR_TRANSACTION while one is open: take after commit.
* - Operators built inside a transaction stay, and their state goes back to empty.
* - What is outside the circuit is not undone: a callback's own side effects, and a
* ZsRebuild stepped in the transaction (it reports ZS_ERR_INTERRUPTED afterwards; start
* another job). */
int32_t zs_circuit_begin(ZsCircuit *c);
int32_t zs_circuit_commit(ZsCircuit *c);
int32_t zs_circuit_rollback(ZsCircuit *c);
int32_t zs_circuit_take(ZsCircuit *c, uint32_t sink, ZsZSet **out);
/* Resumable snapshot load: chunk keys per piece (0 reads as 1). The whole load is one
+50
View File
@@ -40,6 +40,7 @@ codes! {
Poisoned = 7 => "ZS_ERR_POISONED",
Callback = 8 => "ZS_ERR_CALLBACK",
Interrupted = 9 => "ZS_ERR_INTERRUPTED",
Transaction = 10 => "ZS_ERR_TRANSACTION",
}
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -209,6 +210,8 @@ pub struct Engine {
sinks: Vec<Output<Bytes>>,
fault: Fault,
poisoned: bool,
/// A transaction is open (`begin`, and not yet `commit` or `rollback`).
txn: bool,
}
impl Default for Engine {
@@ -226,6 +229,7 @@ impl Engine {
sinks: Vec::new(),
fault: Rc::default(),
poisoned: false,
txn: false,
}
}
@@ -474,8 +478,54 @@ impl Engine {
self.guarded(|c| c.step())
}
fn no_transaction() -> Fail {
Fail::new(Code::Transaction, "no transaction is open")
}
/// Opens a transaction: the circuit logs what its steps change.
pub fn begin(&mut self) -> Res<()> {
self.alive()?;
if self.txn {
return Err(Fail::new(
Code::Transaction,
"a transaction is already open",
));
}
self.circuit.begin();
self.txn = true;
Ok(())
}
/// Keeps the transaction's steps. A transaction whose step failed cannot be committed.
pub fn commit(&mut self) -> Res<()> {
if !self.txn {
return Err(Self::no_transaction());
}
self.alive()?;
self.circuit.commit();
self.txn = false;
Ok(())
}
/// Undoes the transaction's steps and clears the poison a failed step left in it.
pub fn rollback(&mut self) -> Res<()> {
if !self.txn {
return Err(Self::no_transaction());
}
self.circuit.rollback();
self.poisoned = false;
self.txn = false;
Ok(())
}
pub fn take(&mut self, sink: u32) -> Res<BZ> {
self.alive()?;
if self.txn {
return Err(Fail::new(
Code::Transaction,
"a transaction is open: commit or roll back before taking output",
));
}
self.sinks
.get((sink & !SINK_BIT) as usize)
.filter(|_| sink & SINK_BIT != 0)
+24
View File
@@ -499,6 +499,30 @@ pub unsafe extern "C" fn zs_circuit_step(c: *mut ZsCircuit) -> i32 {
})
}
/// # Safety
/// `c` is a live circuit or null.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn zs_circuit_begin(c: *mut ZsCircuit) -> i32 {
// SAFETY: the caller's contract.
guard(|| unsafe { circuit(c)? }.engine().begin())
}
/// # Safety
/// As `zs_circuit_begin`.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn zs_circuit_commit(c: *mut ZsCircuit) -> i32 {
// SAFETY: the caller's contract.
guard(|| unsafe { circuit(c)? }.engine().commit())
}
/// # Safety
/// As `zs_circuit_begin`.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn zs_circuit_rollback(c: *mut ZsCircuit) -> i32 {
// SAFETY: the caller's contract.
guard(|| unsafe { circuit(c)? }.engine().rollback())
}
/// # Safety
/// `c` is a live circuit or null; `out` is null or writable.
#[unsafe(no_mangle)]
+91
View File
@@ -965,3 +965,94 @@ fn a_host_step_between_rebuild_calls_is_interrupted_not_a_poison() {
unsafe { zs_rebuild_free(job) };
unsafe { zs_zset_free(snapshot) };
}
/// An integrate over `k` held at `i64::MAX` (taken), and its sink.
fn at_max() -> (C, u32, u32) {
let c = C::new();
let i = c.input();
let mut int = 0;
assert_eq!(unsafe { zs_circuit_integrate(c.0, i, &mut int) }, OK);
let s = c.sink(int);
c.push(i, &[(b"k", i64::MAX)]);
assert_eq!(c.run(s), one(&[(b"k", i64::MAX)]));
(c, i, s)
}
#[test]
fn a_rollback_after_a_failed_step_revives_the_circuit_in_its_old_state() {
let (c, i, s) = at_max();
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
c.push(i, &[(b"a", 1), (b"k", 1)]);
assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Overflow as i32);
assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
assert_eq!(
unsafe { zs_circuit_commit(c.0) },
Code::Poisoned as i32,
"a failed step cannot be committed"
);
assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
c.push(i, &[(b"k", -1)]);
assert_eq!(
c.run(s),
one(&[(b"k", i64::MAX - 1)]),
"[rollback] neither `a` nor the failed +1 is in the integral"
);
}
#[test]
fn a_committed_transaction_keeps_its_steps_and_a_new_one_may_follow() {
let (c, i, s) = at_max();
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
c.push(i, &[(b"k", -1)]);
assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
assert_eq!(c.take(s), one(&[(b"k", i64::MAX - 1)]));
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
c.push(i, &[(b"k", -1)]);
assert_eq!(c.run(s), one(&[(b"k", i64::MAX - 2)]));
}
#[test]
fn a_failed_callback_step_is_rolled_back_too() {
let c = C::new();
let i = c.input();
let (mut f, mut int) = (0, 0);
assert_eq!(unsafe { zs_circuit_integrate(c.0, i, &mut int) }, OK);
let s = c.sink(int);
assert_eq!(
unsafe { zs_circuit_filter(c.0, i, Some(fails), null_mut(), &mut f) },
OK
);
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
c.push(i, &[(b"k", 1)]);
assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Callback as i32);
assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK, "usable again");
assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
assert_eq!(
c.take(s),
vec![],
"[callback] the failed step left no output"
);
}
#[test]
fn transaction_misuse_is_a_code_and_changes_nothing() {
let (c, i, s) = at_max();
let t = Code::Transaction as i32;
assert_eq!(unsafe { zs_circuit_commit(c.0) }, t);
assert_eq!(unsafe { zs_circuit_rollback(c.0) }, t);
assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
assert_eq!(unsafe { zs_circuit_begin(c.0) }, t);
let mut z = null_mut();
assert_eq!(unsafe { zs_circuit_take(c.0, s, &mut z) }, t);
assert!(z.is_null(), "nothing written");
assert!(last_error().contains("roll back"), "{}", last_error());
assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
// A poison from outside any transaction is not undone by a rollback.
c.push(i, &[(b"k", 1)]);
assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Overflow as i32);
assert_eq!(unsafe { zs_circuit_rollback(c.0) }, t);
assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
}
+193
View File
@@ -0,0 +1,193 @@
//! Transactions through the ABI: a rebuild across a rollback, callbacks, and poison.
//! From the review of #355. Evidence: `context/test-evidence/zset-ffi.md`.
use std::ffi::c_void;
use std::ptr::null_mut;
use zset_ffi::engine::Code;
use zset_ffi::*;
const OK: i32 = 0;
fn zset(entries: &[(&[u8], i64)]) -> *mut ZsZSet {
let mut z = null_mut();
assert_eq!(unsafe { zs_zset_new(&mut z) }, OK);
for (k, w) in entries {
assert_eq!(unsafe { zs_zset_insert(z, k.as_ptr(), k.len(), *w) }, OK);
}
z
}
fn weight(z: *const ZsZSet, k: &[u8]) -> i64 {
let mut w = 0;
assert_eq!(
unsafe { zs_zset_weight(z, k.as_ptr(), k.len(), &mut w) },
OK
);
w
}
fn rig() -> (*mut ZsCircuit, u32, u32) {
let mut c = null_mut();
assert_eq!(unsafe { zs_circuit_new(&mut c) }, OK);
let mut i = 0;
assert_eq!(unsafe { zs_circuit_input(c, &mut i) }, OK);
let mut int = 0;
assert_eq!(unsafe { zs_circuit_integrate(c, i, &mut int) }, OK);
let mut s = 0;
assert_eq!(unsafe { zs_circuit_output(c, int, &mut s) }, OK);
(c, i, s)
}
extern "C" fn out_of_time(_: *mut c_void) -> i32 {
1
}
fn take(c: *mut ZsCircuit, s: u32) -> *mut ZsZSet {
let mut z = null_mut();
assert_eq!(unsafe { zs_circuit_take(c, s, &mut z) }, OK);
z
}
#[test]
fn f1_rebuild_begun_before_a_txn_continued_inside_and_rolled_back() {
let (c, i, s) = rig();
let snap = zset(&[(b"a", 1), (b"b", 1), (b"c", 1), (b"d", 1)]);
let mut job = null_mut();
assert_eq!(unsafe { zs_rebuild_new(c, i, snap, 1, &mut job) }, OK);
let mut done = 0;
assert_eq!(
unsafe { zs_rebuild_step(job, c, Some(out_of_time), null_mut(), &mut done) },
OK
);
assert_eq!(unsafe { zs_circuit_begin(c) }, OK);
assert_eq!(
unsafe { zs_rebuild_step(job, c, Some(out_of_time), null_mut(), &mut done) },
OK
);
assert_eq!(unsafe { zs_circuit_rollback(c) }, OK);
let rc = unsafe { zs_rebuild_step(job, c, None, null_mut(), &mut done) };
let z = take(c, s);
let got = (
rc,
done,
weight(z, b"a"),
weight(z, b"b"),
weight(z, b"c"),
weight(z, b"d"),
);
unsafe { zs_zset_free(z) };
unsafe { zs_rebuild_free(job) };
unsafe { zs_zset_free(snap) };
unsafe { zs_circuit_free(c) };
assert_eq!(
got.0,
Code::Interrupted as i32,
"[f1] header: a rebuild stepped in a rolled-back txn reports ZS_ERR_INTERRUPTED; got (rc, done, a, b, c, d) = {got:?}"
);
}
extern "C" fn rolls_back(ctx: *mut c_void, _: *const u8, _: usize) -> i32 {
let (c, seen) = unsafe { &mut *(ctx as *mut (*mut ZsCircuit, i32)) };
*seen = unsafe { zs_circuit_rollback(*c) };
1
}
#[test]
fn f2_a_rollback_from_inside_a_callback_is_refused_and_the_txn_stays_open() {
let (c, i, s) = rig();
let mut state: (*mut ZsCircuit, i32) = (c, 0);
let mut f = 0;
let ctx = (&mut state as *mut (*mut ZsCircuit, i32)).cast();
assert_eq!(
unsafe { zs_circuit_filter(c, i, Some(rolls_back), ctx, &mut f) },
OK
);
assert_eq!(unsafe { zs_circuit_begin(c) }, OK);
let z = zset(&[(b"k", 1)]);
assert_eq!(unsafe { zs_circuit_push(c, i, z) }, OK);
assert_eq!(unsafe { zs_circuit_step(c) }, OK);
assert_eq!(
state.1,
Code::Poisoned as i32,
"[f2] rollback inside a callback"
);
assert_eq!(
unsafe { zs_circuit_commit(c) },
OK,
"[f2] txn still open, commit works"
);
let out = take(c, s);
assert_eq!(weight(out, b"k"), 1);
unsafe { zs_zset_free(out) };
unsafe { zs_zset_free(z) };
unsafe { zs_circuit_free(c) };
}
#[test]
fn f3_begin_on_a_circuit_poisoned_outside_a_txn_is_poisoned() {
let (c, i, _s) = rig();
let z = zset(&[(b"k", i64::MAX)]);
assert_eq!(unsafe { zs_circuit_push(c, i, z) }, OK);
assert_eq!(unsafe { zs_circuit_step(c) }, OK);
assert_eq!(unsafe { zs_circuit_push(c, i, z) }, OK);
assert_eq!(unsafe { zs_circuit_step(c) }, Code::Overflow as i32);
assert_eq!(unsafe { zs_circuit_begin(c) }, Code::Poisoned as i32);
assert_eq!(unsafe { zs_circuit_rollback(c) }, Code::Transaction as i32);
assert_eq!(unsafe { zs_circuit_commit(c) }, Code::Transaction as i32);
unsafe { zs_zset_free(z) };
unsafe { zs_circuit_free(c) };
}
#[test]
fn f4_a_failed_rebuild_step_in_a_txn_then_rollback_then_resume() {
let (c, i, s) = rig();
let big = zset(&[(b"a", i64::MAX)]);
assert_eq!(unsafe { zs_circuit_push(c, i, big) }, OK);
assert_eq!(unsafe { zs_circuit_step(c) }, OK);
let o = take(c, s);
unsafe { zs_zset_free(o) };
let snap = zset(&[(b"a", 1), (b"b", 1), (b"c", 1)]);
let mut job = null_mut();
assert_eq!(unsafe { zs_rebuild_new(c, i, snap, 1, &mut job) }, OK);
assert_eq!(unsafe { zs_circuit_begin(c) }, OK);
let mut done = 0;
assert_eq!(
unsafe { zs_rebuild_step(job, c, None, null_mut(), &mut done) },
Code::Overflow as i32
);
assert_eq!(unsafe { zs_circuit_rollback(c) }, OK);
let rc = unsafe { zs_rebuild_step(job, c, None, null_mut(), &mut done) };
unsafe { zs_rebuild_free(job) };
unsafe { zs_zset_free(snap) };
unsafe { zs_zset_free(big) };
unsafe { zs_circuit_free(c) };
assert_eq!(
rc,
Code::Interrupted as i32,
"[f4] header: a rebuild stepped in a rolled-back txn reports ZS_ERR_INTERRUPTED; got rc={rc} done={done}"
);
}
#[test]
fn f5_a_chunk_whose_push_overflows_in_a_txn_then_rollback_is_interrupted() {
let (c, i, _s) = rig();
let pending = zset(&[(b"a", i64::MAX)]);
assert_eq!(unsafe { zs_circuit_push(c, i, pending) }, OK);
let snap = zset(&[(b"a", 1), (b"b", 1)]);
let mut job = null_mut();
assert_eq!(unsafe { zs_rebuild_new(c, i, snap, 1, &mut job) }, OK);
assert_eq!(unsafe { zs_circuit_begin(c) }, OK);
let mut done = 0;
assert_eq!(
unsafe { zs_rebuild_step(job, c, None, null_mut(), &mut done) },
Code::Overflow as i32,
"the chunk's push overflows the pending delta"
);
assert_eq!(unsafe { zs_circuit_rollback(c) }, OK);
let rc = unsafe { zs_rebuild_step(job, c, None, null_mut(), &mut done) };
unsafe { zs_rebuild_free(job) };
unsafe { zs_zset_free(snap) };
unsafe { zs_zset_free(pending) };
unsafe { zs_circuit_free(c) };
assert_eq!(rc, Code::Interrupted as i32, "[f5] got rc={rc} done={done}");
}
+6
View File
@@ -48,6 +48,8 @@
(defcfun "zs_circuit_filter" :int32 (c :pointer) (stream :uint32) (keep :pointer) (ctx :pointer) (out :pointer))
(defcfun "zs_circuit_push" :int32 (c :pointer) (input :uint32) (z :pointer))
(defcfun "zs_circuit_step" :int32 (c :pointer))
(defcfun "zs_circuit_begin" :int32 (c :pointer))
(defcfun "zs_circuit_rollback" :int32 (c :pointer))
(defcallback keep-or-fail :int32 ((ctx :pointer) (key :pointer) (len :size))
(declare (ignore ctx key len))
@@ -61,9 +63,13 @@
(check "filter" (zs-circuit-filter c (mem-ref id :uint32) (callback keep-or-fail) (null-pointer) id) 0)
(check "zset" (zs-zset-new zout) 0)
(key-call #'zs-zset-insert (mem-ref zout :pointer) "k" 1)
(check "begin a transaction" (zs-circuit-begin c) 0)
(check "push" (zs-circuit-push c 0 (mem-ref zout :pointer)) 0)
(check "step reports the callback's failure" (zs-circuit-step c) 8)
(check "the circuit is dead afterwards" (zs-circuit-step c) 7)
(check "rollback revives it" (zs-circuit-rollback c) 0)
(check "a transaction opens again" (zs-circuit-begin c) 0)
(check "and a second begin is refused" (zs-circuit-begin c) 10)
(zs-zset-free (mem-ref zout :pointer))
(zs-circuit-free c)))
(format t "zset smoke: all ok~%")
+394 -62
View File
@@ -1,23 +1,30 @@
//! The synchronous, single-threaded circuit. This module is the wiring (`Rc`, `RefCell`,
//! boxed nodes); the operators' logic is in `ops.rs` and `lib.rs`.
use crate::ops::{AggregateState, DistinctState, JoinState};
use crate::ops::{
AggregateState, DistinctState, JoinState, Log, Transactional, Undoable, add_logged, unadd,
};
use crate::{MemTrace, Trace, ZSet};
use std::cell::RefCell;
use std::cell::{Ref, RefCell};
use std::rc::Rc;
/// What an edge carries: the latest piece, and the whole value of the current tick.
/// They differ only while a tick arrives in pieces (`Circuit::step_continue`).
/// What an edge carries: the latest piece, and the whole tick's value, kept only while a
/// tick is split (`Circuit::step_continue`); `None` means the piece is the whole.
pub(crate) struct Slot<T> {
piece: T,
whole: T,
whole: Option<T>,
/// Inside a transaction: the `(piece, whole)` the first publish replaced.
saved: Option<(T, Option<T>)>,
open: bool,
}
impl<T: Default> Default for Slot<T> {
fn default() -> Self {
Slot {
piece: T::default(),
whole: T::default(),
whole: None,
saved: None,
open: false,
}
}
}
@@ -41,18 +48,51 @@ impl<K: Ord + Clone> Piecewise for ZSet<K> {
pub struct Stream<T>(Rc<RefCell<Slot<T>>>);
impl<T: Piecewise> Stream<T> {
fn piece(&self) -> T {
self.0.borrow().piece.clone()
fn piece(&self) -> Ref<'_, T> {
Ref::map(self.0.borrow(), |s| &s.piece)
}
fn publish(&self, v: T, cont: bool) {
let mut slot = self.0.borrow_mut();
let slot = &mut *slot;
if cont {
slot.whole.absorb(&v);
if slot.open && slot.saved.is_none() {
slot.saved = Some((slot.piece.clone(), slot.whole.clone()));
}
let mut whole = slot.whole.take().unwrap_or_else(|| slot.piece.clone());
whole.absorb(&v);
slot.whole = Some(whole);
slot.piece = v;
} else {
slot.whole = v.clone();
let piece = std::mem::replace(&mut slot.piece, v);
let whole = slot.whole.take();
if slot.open && slot.saved.is_none() {
slot.saved = Some((piece, whole));
}
}
slot.piece = v;
}
}
impl<T> Stream<T> {
fn begin(&self) {
let mut slot = self.0.borrow_mut();
slot.open = true;
slot.saved = None;
}
fn commit(&self) {
let mut slot = self.0.borrow_mut();
slot.open = false;
slot.saved = None;
}
fn rollback(&self) {
let mut slot = self.0.borrow_mut();
if let Some((piece, whole)) = slot.saved.take() {
slot.piece = piece;
slot.whole = whole;
}
slot.open = false;
}
}
@@ -65,16 +105,18 @@ impl<T> Clone for Stream<T> {
impl<T: Clone> Stream<T> {
/// The value of the last tick, all its pieces together (the zero before the first).
pub fn current(&self) -> T {
self.0.borrow().whole.clone()
let slot = self.0.borrow();
slot.whole.as_ref().unwrap_or(&slot.piece).clone()
}
}
/// Where a caller feeds deltas; they enter the circuit at the next `step`.
pub struct Input<K>(Rc<RefCell<ZSet<K>>>);
/// Where a caller feeds deltas; they enter the circuit at the next `step`. It remembers
/// which circuit made it, so a `Rebuild` can refuse a circuit that is not its own.
pub struct Input<K>(Rc<RefCell<ZSet<K>>>, Rc<()>);
impl<K> Clone for Input<K> {
fn clone(&self) -> Self {
Input(Rc::clone(&self.0))
Input(Rc::clone(&self.0), Rc::clone(&self.1))
}
}
@@ -84,7 +126,8 @@ impl<K: Ord> Input<K> {
}
}
/// A sink that sums every step's delta until it is taken.
/// A sink that sums every step's delta until it is taken. Take inside no transaction: a
/// rollback subtracts the step's output, which a `take` in between has already removed.
pub struct Output<K>(Rc<RefCell<ZSet<K>>>);
impl<K: Ord> Output<K> {
@@ -93,78 +136,237 @@ impl<K: Ord> Output<K> {
}
}
trait Node {
trait Node: Transactional {
/// `cont` is true for a later piece of a tick that has already begun.
fn eval(&mut self, cont: bool);
}
struct Unary<I, O, F> {
/// One input, one output, and the state `f` may change.
struct Unary<I, O, S, F> {
src: Stream<I>,
out: Stream<O>,
state: S,
f: F,
}
impl<I: Piecewise, O: Piecewise, F: FnMut(&I) -> O> Node for Unary<I, O, F> {
impl<I, O, S: Transactional, F> Transactional for Unary<I, O, S, F> {
fn begin(&mut self) {
self.state.begin();
self.out.begin();
}
fn commit(&mut self) {
self.state.commit();
self.out.commit();
}
fn rollback(&mut self) {
self.state.rollback();
self.out.rollback();
}
}
impl<I: Piecewise, O: Piecewise, S: Transactional, F: FnMut(&mut S, &I) -> O> Node
for Unary<I, O, S, F>
{
fn eval(&mut self, cont: bool) {
let v = (self.f)(&self.src.piece());
let v = (self.f)(&mut self.state, &self.src.piece());
self.out.publish(v, cont);
}
}
struct Binary<A, B, O, F> {
struct Binary<A, B, O, S, F> {
a: Stream<A>,
b: Stream<B>,
out: Stream<O>,
state: S,
f: F,
}
impl<A: Piecewise, B: Piecewise, O: Piecewise, F: FnMut(&A, &B) -> O> Node for Binary<A, B, O, F> {
impl<A, B, O, S: Transactional, F> Transactional for Binary<A, B, O, S, F> {
fn begin(&mut self) {
self.state.begin();
self.out.begin();
}
fn commit(&mut self) {
self.state.commit();
self.out.commit();
}
fn rollback(&mut self) {
self.state.rollback();
self.out.rollback();
}
}
impl<A: Piecewise, B: Piecewise, O: Piecewise, S: Transactional, F: FnMut(&mut S, &A, &B) -> O> Node
for Binary<A, B, O, S, F>
{
fn eval(&mut self, cont: bool) {
let v = (self.f)(&self.a.piece(), &self.b.piece());
let v = (self.f)(&mut self.state, &self.a.piece(), &self.b.piece());
self.out.publish(v, cont);
}
}
/// What a `Delay` holds, and what it held when a transaction began.
struct Held<K> {
held: ZSet<K>,
open: bool,
saved: Option<ZSet<K>>,
}
impl<K: Ord + Clone> Held<K> {
/// Notes the value to go back to, once per transaction. `held` is one tick's delta.
fn save(&mut self) {
if self.open && self.saved.is_none() {
self.saved = Some(self.held.clone());
}
}
}
impl<K> Transactional for Held<K> {
fn begin(&mut self) {
self.open = true;
self.saved = None;
}
fn commit(&mut self) {
self.open = false;
self.saved = None;
}
fn rollback(&mut self) {
if let Some(held) = self.saved.take() {
self.held = held;
}
self.open = false;
}
}
/// out(t) = in(t - 1). A continued piece adds to what is held and emits nothing.
struct Delay<K> {
src: Stream<ZSet<K>>,
out: Stream<ZSet<K>>,
held: ZSet<K>,
state: Held<K>,
}
impl<K> Transactional for Delay<K> {
fn begin(&mut self) {
self.state.begin();
self.out.begin();
}
fn commit(&mut self) {
self.state.commit();
self.out.commit();
}
fn rollback(&mut self) {
self.state.rollback();
self.out.rollback();
}
}
impl<K: Ord + Clone> Node for Delay<K> {
fn eval(&mut self, cont: bool) {
let d = self.src.piece();
self.state.save();
let v = if cont {
self.held += &d;
self.state.held += &*d;
ZSet::new()
} else {
std::mem::replace(&mut self.held, d)
std::mem::replace(&mut self.state.held, d.clone())
};
drop(d);
self.out.publish(v, cont);
}
}
/// The integral, and the log that takes a transaction's additions back.
struct Acc<K> {
acc: ZSet<K>,
log: Log<(K, i64)>,
}
impl<K: Ord> Transactional for Acc<K> {
fn begin(&mut self) {
self.log.begin();
}
fn commit(&mut self) {
self.log.commit();
}
fn rollback(&mut self) {
let acc = &mut self.acc;
self.log.rollback(|e| unadd(acc, e));
}
}
/// out(t) = in(0) + .. + in(t). A continued piece emits only its own input.
struct Integrate<K> {
src: Stream<ZSet<K>>,
out: Stream<ZSet<K>>,
acc: ZSet<K>,
state: Acc<K>,
}
impl<K: Ord> Transactional for Integrate<K> {
fn begin(&mut self) {
self.state.begin();
self.out.begin();
}
fn commit(&mut self) {
self.state.commit();
self.out.commit();
}
fn rollback(&mut self) {
self.state.rollback();
self.out.rollback();
}
}
impl<K: Ord + Clone> Node for Integrate<K> {
fn eval(&mut self, cont: bool) {
let d = self.src.piece();
self.acc += &d;
let v = if cont { d } else { self.acc.clone() };
add_logged(&mut self.state.acc, &d, &mut self.state.log);
let v = if cont {
d.clone()
} else {
self.state.acc.clone()
};
drop(d);
self.out.publish(v, cont);
}
}
/// Moves what was pushed into the circuit's stream.
/// Moves what was pushed into the circuit's stream. Inside a transaction it remembers what
/// was pending, so a rollback gives back the delta the failed step took.
struct Source<K> {
pending: Rc<RefCell<ZSet<K>>>,
out: Stream<ZSet<K>>,
saved: Option<ZSet<K>>,
}
impl<K: Ord + Clone> Transactional for Source<K> {
fn begin(&mut self) {
self.saved = Some(self.pending.borrow().clone());
self.out.begin();
}
fn commit(&mut self) {
self.saved = None;
self.out.commit();
}
fn rollback(&mut self) {
if let Some(saved) = self.saved.take() {
*self.pending.borrow_mut() = saved;
}
self.out.rollback();
}
}
impl<K: Ord + Clone> Node for Source<K> {
@@ -174,12 +376,42 @@ impl<K: Ord + Clone> Node for Source<K> {
}
}
/// The sink's sum, and the log that takes a transaction's additions back.
struct Sink<K> {
sum: Rc<RefCell<ZSet<K>>>,
log: Log<(K, i64)>,
}
impl<K: Ord> Transactional for Sink<K> {
fn begin(&mut self) {
self.log.begin();
}
fn commit(&mut self) {
self.log.commit();
}
fn rollback(&mut self) {
let sum = &mut *self.sum.borrow_mut();
self.log.rollback(|e| unadd(sum, e));
}
}
/// Nodes run in creation order, which is a topological order because a node can only
/// be built from streams that already exist. There are no cycles to schedule.
#[derive(Default)]
pub struct Circuit {
nodes: Vec<Box<dyn Node>>,
ticks: u64,
/// This circuit's identity: every `Input` it makes holds a clone.
token: Rc<()>,
/// `Some(ticks)` as of `begin`, while a transaction is open.
open: Option<u64>,
/// A step or piece ran since `begin`.
stepped: bool,
/// Rollbacks that undid a step or piece. It never goes down, unlike `ticks`, so a
/// `Rebuild` can tell that a rollback took some of its chunks.
undone: u64,
}
type Z<K> = ZSet<K>;
@@ -206,21 +438,103 @@ impl Circuit {
self.run(true);
}
/// Opens a transaction: state changes are logged until `commit` or `rollback`, at the
/// cost of the keys touched. Panics if one is open. See `context/code/zset.md`.
pub fn begin(&mut self) {
assert!(
self.open.is_none(),
"zset: a transaction is already open on this circuit"
);
self.open = Some(self.ticks);
self.stepped = false;
for n in &mut self.nodes {
n.begin();
}
}
/// Keeps what the transaction did. Panics if none is open.
pub fn commit(&mut self) {
assert!(
self.open.take().is_some(),
"zset: no transaction is open on this circuit"
);
for n in &mut self.nodes {
n.commit();
}
}
/// Puts the state back to what it was at `begin`, even after a step panicked half way;
/// pushes since are dropped. Panics if none is open.
pub fn rollback(&mut self) {
let ticks = self
.open
.take()
.expect("zset: no transaction is open on this circuit");
for n in &mut self.nodes {
n.rollback();
}
self.ticks = ticks;
self.undone += u64::from(self.stepped);
self.stepped = false;
}
fn run(&mut self, cont: bool) {
self.stepped |= self.open.is_some();
for n in &mut self.nodes {
n.eval(cont);
}
}
/// Adds a node; one built inside a transaction starts recording at once.
fn add(&mut self, mut node: Box<dyn Node>) {
if self.open.is_some() {
node.begin();
}
self.nodes.push(node);
}
fn unary_with<I: Piecewise + 'static, O: Piecewise + 'static, S: Transactional + 'static>(
&mut self,
src: &Stream<I>,
state: S,
f: impl FnMut(&mut S, &I) -> O + 'static,
) -> Stream<O> {
let out = Stream(Rc::default());
self.add(Box::new(Unary {
src: src.clone(),
out: out.clone(),
state,
f,
}));
out
}
fn unary<I: Piecewise + 'static, O: Piecewise + 'static>(
&mut self,
src: &Stream<I>,
f: impl FnMut(&I) -> O + 'static,
mut f: impl FnMut(&I) -> O + 'static,
) -> Stream<O> {
self.unary_with(src, (), move |_, d| f(d))
}
fn binary_with<
A: Piecewise + 'static,
B: Piecewise + 'static,
O: Piecewise + 'static,
S: Transactional + 'static,
>(
&mut self,
a: &Stream<A>,
b: &Stream<B>,
state: S,
f: impl FnMut(&mut S, &A, &B) -> O + 'static,
) -> Stream<O> {
let out = Stream(Rc::default());
self.nodes.push(Box::new(Unary {
src: src.clone(),
self.add(Box::new(Binary {
a: a.clone(),
b: b.clone(),
out: out.clone(),
state,
f,
}));
out
@@ -230,33 +544,30 @@ impl Circuit {
&mut self,
a: &Stream<A>,
b: &Stream<B>,
f: impl FnMut(&A, &B) -> O + 'static,
mut f: impl FnMut(&A, &B) -> O + 'static,
) -> Stream<O> {
let out = Stream(Rc::default());
self.nodes.push(Box::new(Binary {
a: a.clone(),
b: b.clone(),
out: out.clone(),
f,
}));
out
self.binary_with(a, b, (), move |_, x, y| f(x, y))
}
pub fn input<K: Ord + Clone + 'static>(&mut self) -> (Input<K>, Stream<Z<K>>) {
let pending: Rc<RefCell<Z<K>>> = Rc::default();
let out = Stream(Rc::default());
self.nodes.push(Box::new(Source {
self.add(Box::new(Source {
pending: Rc::clone(&pending),
out: out.clone(),
saved: None,
}));
(Input(pending), out)
(Input(pending, Rc::clone(&self.token)), out)
}
pub fn output<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Output<K> {
let sink = Output(Rc::default());
let acc = Rc::clone(&sink.0);
self.unary(s, move |d: &Z<K>| {
*acc.borrow_mut() += d;
let state = Sink {
sum: Rc::clone(&sink.0),
log: Log::default(),
};
self.unary_with(s, state, |sink, d: &Z<K>| {
add_logged(&mut sink.sum.borrow_mut(), d, &mut sink.log);
});
sink
}
@@ -264,10 +575,14 @@ impl Circuit {
/// out(t) = in(t - 1); out(0) is zero.
pub fn delay<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
let out = Stream(Rc::default());
self.nodes.push(Box::new(Delay {
self.add(Box::new(Delay {
src: s.clone(),
out: out.clone(),
held: ZSet::new(),
state: Held {
held: ZSet::new(),
open: false,
saved: None,
},
}));
out
}
@@ -275,10 +590,13 @@ impl Circuit {
/// out(t) = in(0) + .. + in(t).
pub fn integrate<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
let out = Stream(Rc::default());
self.nodes.push(Box::new(Integrate {
self.add(Box::new(Integrate {
src: s.clone(),
out: out.clone(),
acc: ZSet::new(),
state: Acc {
acc: ZSet::new(),
log: Log::default(),
},
}));
out
}
@@ -337,8 +655,7 @@ impl Circuit {
/// Incremental `distinct`.
pub fn distinct<K: Ord + Clone + 'static>(&mut self, s: &Stream<Z<K>>) -> Stream<Z<K>> {
let mut st = DistinctState::new();
self.unary(s, move |d: &Z<K>| st.step(d))
self.unary_with(s, DistinctState::new(), |st, d: &Z<K>| st.step(d))
}
/// The bilinear delta join over in-memory traces.
@@ -374,8 +691,8 @@ impl Circuit {
TA: Trace<K, A> + 'static,
TB: Trace<K, B> + 'static,
{
let mut st = JoinState::new(left, right);
self.binary(a, b, move |da: &Z<(K, A)>, db: &Z<(K, B)>| {
let st = JoinState::new(Undoable::new(left), Undoable::new(right));
self.binary_with(a, b, st, move |st, da: &Z<(K, A)>, db: &Z<(K, B)>| {
st.step(da, db, &mut f)
})
}
@@ -407,8 +724,8 @@ impl Circuit {
R: Ord + Clone + 'static,
T: Trace<K, V> + 'static,
{
let mut st = AggregateState::new(groups);
self.unary(s, move |d: &Z<(K, V)>| st.step(d, &mut f))
let st = AggregateState::new(Undoable::new(groups));
self.unary_with(s, st, move |st, d: &Z<(K, V)>| st.step(d, &mut f))
}
}
@@ -420,13 +737,15 @@ pub enum Progress {
}
/// A snapshot load as a sliced job: one logical tick (`context/code/zset.md`). A pending
/// job owns the clock: a `Circuit::step` before it is Done makes `step` panic.
/// job owns the clock: a `Circuit::step` before it is Done makes `step` panic, for good.
pub struct Rebuild<K> {
input: Input<K>,
rest: std::collections::btree_map::IntoIter<K, i64>,
chunk: usize,
/// The circuit's tick count right after this job's first chunk, once it has run one.
began_at: Option<u64>,
/// The circuit's `undone` count when this job first ran, even if that chunk panicked.
undone_at: Option<u64>,
}
impl<K: Ord> Rebuild<K> {
@@ -437,27 +756,40 @@ impl<K: Ord> Rebuild<K> {
rest: snapshot.into_iter(),
chunk: chunk.max(1),
began_at: None,
undone_at: None,
}
}
/// Whether a `Circuit::step` since this job began has split its tick. A finished job
/// cannot be interrupted.
/// Whether `circuit` is the one that made this job's input.
pub fn belongs_to(&self, circuit: &Circuit) -> bool {
Rc::ptr_eq(&self.input.1, &circuit.token)
}
/// Whether a `Circuit::step` split this job's tick, or a rollback undid a step after it
/// first ran (finished or not). An interrupted job never finishes: start another.
pub fn interrupted(&self, circuit: &Circuit) -> bool {
self.rest.len() > 0 && self.began_at.is_some_and(|t| t != circuit.ticks())
let split = self.rest.len() > 0 && self.began_at.is_some_and(|t| t != circuit.ticks());
split || self.undone_at.is_some_and(|u| u != circuit.undone)
}
/// Runs chunks until done or `expired()`, at least one per call. A predicate, not an
/// `Instant` (wasm32). An empty snapshot takes its one tick; a finished job none.
pub fn step(&mut self, circuit: &mut Circuit, mut expired: impl FnMut() -> bool) -> Progress {
assert!(
self.belongs_to(circuit),
"zset: the rebuild's input belongs to another circuit"
);
self.undone_at.get_or_insert(circuit.undone);
assert!(
!self.interrupted(circuit),
"zset: the rebuild was interrupted by a Circuit::step before it finished"
"zset: the rebuild was interrupted by a Circuit::step or a rollback before it finished"
);
loop {
let chunk: ZSet<K> = self.rest.by_ref().take(self.chunk).collect();
if chunk.is_zero() && self.began_at.is_some() {
return Progress::Done;
}
circuit.stepped |= circuit.open.is_some();
self.input.push(chunk);
if self.began_at.is_some() {
circuit.step_continue();
+1 -1
View File
@@ -91,7 +91,7 @@ impl<K: Ord> ZSet<K> {
self.adjust(key, weight, false);
}
fn adjust(&mut self, key: K, weight: i64, subtract: bool) {
pub(crate) fn adjust(&mut self, key: K, weight: i64, subtract: bool) {
match self.weights.entry(key) {
Entry::Vacant(e) => {
let from_zero = if subtract { negated(weight) } else { weight };
+183 -1
View File
@@ -4,15 +4,95 @@
use crate::{Trace, ZSet, plus, times};
use std::collections::BTreeMap;
/// State a transaction can undo (`Circuit::begin`): `begin` starts recording, `commit`
/// forgets the record, `rollback` replays it backwards. The cost is the delta, not the state.
pub(crate) trait Transactional {
fn begin(&mut self);
fn commit(&mut self);
fn rollback(&mut self);
}
impl Transactional for () {
fn begin(&mut self) {}
fn commit(&mut self) {}
fn rollback(&mut self) {}
}
/// An undo log that records only while a transaction is open.
#[derive(Clone, Debug)]
pub(crate) struct Log<E> {
on: bool,
entries: Vec<E>,
}
impl<E> Default for Log<E> {
fn default() -> Self {
Log {
on: false,
entries: Vec::new(),
}
}
}
impl<E> Log<E> {
pub(crate) fn begin(&mut self) {
self.on = true;
self.entries.clear();
}
pub(crate) fn commit(&mut self) {
self.on = false;
self.entries = Vec::new();
}
/// Builds the entry if a transaction is open. Build it before the change it describes,
/// so a panicking `Clone` cannot leave a change unlogged.
pub(crate) fn entry(&self, entry: impl FnOnce() -> E) -> Option<E> {
self.on.then(entry)
}
/// Logs an entry made by `entry`, once its change has been made.
pub(crate) fn push(&mut self, entry: Option<E>) {
self.entries.extend(entry);
}
/// Hands the entries to `undo`, newest first, and closes the log.
pub(crate) fn rollback(&mut self, mut undo: impl FnMut(E)) {
self.on = false;
for e in std::mem::take(&mut self.entries).into_iter().rev() {
undo(e);
}
}
}
/// `z += d`, one key at a time, logging each key that was applied. A key whose addition
/// overflows panics before it changes anything, so the log is exactly what was applied.
pub(crate) fn add_logged<K: Ord + Clone>(z: &mut ZSet<K>, d: &ZSet<K>, log: &mut Log<(K, i64)>) {
for (k, w) in d.iter() {
let entry = log.entry(|| (k.clone(), w));
z.insert(k.clone(), w);
log.push(entry);
}
}
/// Takes one logged addition back. The result is the weight before it, which was in range.
pub(crate) fn unadd<K: Ord>(z: &mut ZSet<K>, (k, w): (K, i64)) {
z.adjust(k, w, true);
}
/// Incremental `distinct`: out = distinct(held + d) - distinct(held), per touched key.
#[derive(Clone, Debug, Default)]
pub struct DistinctState<K> {
held: ZSet<K>,
log: Log<(K, i64)>,
}
impl<K: Ord + Clone> DistinctState<K> {
pub fn new() -> Self {
DistinctState { held: ZSet::new() }
DistinctState {
held: ZSet::new(),
log: Log::default(),
}
}
pub fn step(&mut self, d: &ZSet<K>) -> ZSet<K> {
@@ -21,12 +101,84 @@ impl<K: Ord + Clone> DistinctState<K> {
let before = self.held.weight(k);
let after = plus(before, w);
out.insert(k.clone(), i64::from(after > 0) - i64::from(before > 0));
let entry = self.log.entry(|| (k.clone(), w));
self.held.insert(k.clone(), w);
self.log.push(entry);
}
out
}
}
impl<K: Ord> Transactional for DistinctState<K> {
fn begin(&mut self) {
self.log.begin();
}
fn commit(&mut self) {
self.log.commit();
}
fn rollback(&mut self) {
let held = &mut self.held;
self.log.rollback(|e| unadd(held, e));
}
}
/// A trace that logs what it applies, so a transaction can take it back.
#[derive(Clone, Debug)]
pub(crate) struct Undoable<T, K, V> {
inner: T,
log: Log<(K, V, i64)>,
}
impl<T, K, V> Undoable<T, K, V> {
pub(crate) fn new(inner: T) -> Self {
Undoable {
inner,
log: Log::default(),
}
}
}
impl<T: Trace<K, V>, K: Clone, V: Clone> Trace<K, V> for Undoable<T, K, V> {
fn apply(&mut self, k: &K, v: &V, w: i64) {
let entry = self.log.entry(|| (k.clone(), v.clone(), w));
self.inner.apply(k, v, w);
self.log.push(entry);
}
fn lookup(&self, k: &K) -> Vec<(V, i64)> {
self.inner.lookup(k)
}
fn range(&self, lo: std::ops::Bound<&K>, hi: std::ops::Bound<&K>) -> Vec<(K, V, i64)> {
self.inner.range(lo, hi)
}
}
impl<T: Trace<K, V>, K, V> Transactional for Undoable<T, K, V> {
fn begin(&mut self) {
self.log.begin();
}
fn commit(&mut self) {
self.log.commit();
}
fn rollback(&mut self) {
let inner = &mut self.inner;
self.log.rollback(|(k, v, w)| match w.checked_neg() {
Some(n) => inner.apply(&k, &v, n),
// -i64::MIN is 2^63: add it as i64::MAX then 1. The weight before the apply
// was non-negative, so neither step leaves i64.
None => {
inner.apply(&k, &v, i64::MAX);
inner.apply(&k, &v, 1);
}
});
}
}
/// The bilinear delta join of two keyed streams:
/// out = da x b_prev + a_prev x db + da x db, never a rescan of either side.
#[derive(Clone, Debug, Default)]
@@ -75,6 +227,22 @@ impl<TA, TB> JoinState<TA, TB> {
}
}
impl<TA: Transactional, TB: Transactional> Transactional for JoinState<TA, TB> {
fn begin(&mut self) {
self.left.begin();
self.right.begin();
}
fn commit(&mut self) {
self.left.commit();
self.right.commit();
}
fn rollback(&mut self) {
self.left.rollback();
self.right.rollback();
}
}
/// Incremental group-by: `f` folds one key's integrated values to at most one result.
/// An empty group has no row, and `f` is not asked about it. Only touched keys are re-folded: out = -(k, f(old)) + (k, f(new)).
#[derive(Clone, Debug, Default)]
@@ -122,3 +290,17 @@ impl<T> AggregateState<T> {
out
}
}
impl<T: Transactional> Transactional for AggregateState<T> {
fn begin(&mut self) {
self.groups.begin();
}
fn commit(&mut self) {
self.groups.commit();
}
fn rollback(&mut self) {
self.groups.rollback();
}
}
+15 -5
View File
@@ -7,7 +7,8 @@ use std::ops::Bound;
/// An integrated `ZSet<(K, V)>`, indexed by `K`. Reads return owned values, so an
/// implementation may page or compress what it holds. See `context/code/zset.md`.
pub trait Trace<K, V> {
/// Adds `w` to the weight of `(k, v)`.
/// Adds `w` to the weight of `(k, v)`. A panic must leave the trace as it was: a
/// transaction (`Circuit::begin`) logs an apply only once it has returned.
fn apply(&mut self, k: &K, v: &V, w: i64);
/// The values at `k` with their non-zero weights, ascending by value.
fn lookup(&self, k: &K) -> Vec<(V, i64)>;
@@ -52,10 +53,19 @@ fn is_empty_range<K: Ord>(lo: Bound<&K>, hi: Bound<&K>) -> bool {
impl<K: Ord + Clone, V: Ord + Clone> Trace<K, V> for MemTrace<K, V> {
fn apply(&mut self, k: &K, v: &V, w: i64) {
let group = self.groups.entry(k.clone()).or_default();
group.insert(v.clone(), w);
if group.is_zero() {
self.groups.remove(k);
// Every clone is made before the map changes, so a panicking one changes nothing.
match self.groups.get_mut(k) {
Some(group) => {
group.insert(v.clone(), w);
if group.is_zero() {
self.groups.remove(k);
}
}
None if w != 0 => {
let group = ZSet::singleton(v.clone(), w);
self.groups.insert(k.clone(), group);
}
None => {}
}
}
+187
View File
@@ -0,0 +1,187 @@
//! A panicking `Clone` anywhere in a transactional step is rolled back, and `MemTrace::apply`
//! survives one. Evidence: `context/test-evidence/zset.md`.
use std::cell::Cell;
use std::panic::{AssertUnwindSafe, catch_unwind};
use zset::{Circuit, Input, MemTrace, Output, Stream, Trace, ZSet};
type Z<K> = ZSet<K>;
fn quiet() {
static O: std::sync::Once = std::sync::Once::new();
O.call_once(|| {
let prev = std::panic::take_hook();
std::panic::set_hook(Box::new(move |info| {
let s = format!("{info}");
if !(s.contains("tripped") || s.contains("overflow")) {
prev(info);
}
}));
});
}
thread_local! { static CLONE_TRIP: Cell<Option<u32>> = const { Cell::new(None) }; }
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
struct Boom(u8);
impl Clone for Boom {
fn clone(&self) -> Self {
CLONE_TRIP.with(|t| {
if let Some(n) = t.get() {
if n == 0 {
t.set(None);
panic!("clone tripped");
}
t.set(Some(n - 1));
}
});
Boom(self.0)
}
}
type B = (Boom, Boom);
struct Rig {
c: Circuit,
a: Input<B>,
b: Input<B>,
streams: Vec<Stream<Z<B>>>,
agg: Stream<Z<(Boom, u8)>>,
sinks: Vec<Output<B>>,
}
fn rig(which: u8) -> Rig {
let mut c = Circuit::new();
let (a, sa) = c.input::<B>();
let (b, sb) = c.input::<B>();
let mut streams = vec![sa.clone(), sb.clone()];
if which & 1 != 0 {
streams.push(c.integrate(&sa));
}
if which & 2 != 0 {
streams.push(c.distinct(&sa));
}
if which & 4 != 0 {
streams.push(c.join_indexed(&sa, &sb, |k, x, y| (Boom(k.0), Boom(x.0 ^ y.0))));
}
if which & 8 != 0 {
streams.push(c.delay(&sa));
}
let agg = c.aggregate(&sa, |_, vals: &[(Boom, i64)]| Some(vals.len() as u8));
let sinks = streams.iter().map(|s| c.output(s)).collect();
Rig {
c,
a,
b,
streams,
agg,
sinks,
}
}
impl Rig {
fn view(&self) -> String {
format!(
"{:?} {:?}",
self.streams.iter().map(|s| s.current()).collect::<Vec<_>>(),
self.agg.current()
)
}
fn drain(&self) -> String {
format!(
"{:?}",
self.sinks.iter().map(|s| s.take()).collect::<Vec<_>>()
)
}
}
fn bz(v: &[(u8, u8, i64)]) -> Z<B> {
v.iter()
.map(|(k, x, w)| ((Boom(*k), Boom(*x)), *w))
.collect()
}
fn clone_panic_sweep(which: u8) -> Vec<String> {
quiet();
let mut failures = vec![];
for n in 0..400u32 {
let (mut x, mut y) = (rig(which), rig(which));
for r in [&x, &y] {
r.a.push(bz(&[(1, 1, 1), (2, 2, 1)]));
r.b.push(bz(&[(1, 3, 1), (3, 3, 1)]));
}
x.c.step();
y.c.step();
x.drain();
let (v0, d0) = (x.view(), x.drain());
y.drain();
x.c.begin();
x.a.push(bz(&[(1, 5, 1), (4, 4, 1), (2, 2, -1)]));
x.b.push(bz(&[(4, 6, 1), (1, 3, -1)]));
CLONE_TRIP.with(|t| t.set(Some(n)));
let r = catch_unwind(AssertUnwindSafe(|| x.c.step()));
let fired = CLONE_TRIP.with(|t| t.get()).is_none();
CLONE_TRIP.with(|t| t.set(None));
if !fired {
assert!(r.is_ok());
break;
}
x.c.rollback();
let (v1, d1) = (x.view(), x.drain());
if v1 != v0 || d1 != d0 {
failures.push(format!(
"n={n}: rollback state differs\n view {v1}\n want {v0}\n drain {d1}"
));
continue;
}
for r in [&x, &y] {
// The keys the failed step touched come back, so state leaked by one shows.
r.a.push(bz(&[
(1, 7, 1),
(4, 1, 1),
(1, 5, 1),
(4, 4, 1),
(2, 2, -1),
]));
r.b.push(bz(&[(4, 2, 1), (1, 9, 1)]));
}
x.c.step();
y.c.step();
if x.view() != y.view() || x.drain() != y.drain() {
failures.push(format!("n={n}: post tick differs"));
}
}
failures
}
#[test]
fn c1_a_clone_panic_anywhere_in_a_step_is_rolled_back() {
let mut all = vec![];
for which in [1u8, 2, 4, 8, 15] {
for f in clone_panic_sweep(which) {
all.push(format!("which={which} {f}"));
}
}
assert!(
all.is_empty(),
"{} failures:\n{}",
all.len(),
all.join("\n")
);
}
#[test]
fn a_panicking_value_clone_leaves_a_memtrace_unchanged() {
quiet();
let mut t: MemTrace<u8, Boom> = MemTrace::new();
CLONE_TRIP.with(|c| c.set(Some(0)));
let r = catch_unwind(AssertUnwindSafe(|| t.apply(&1, &Boom(1), 1)));
CLONE_TRIP.with(|c| c.set(None));
assert!(r.is_err());
assert_eq!(
t.keys(),
0,
"[c2] MemTrace::apply panicked but left a group"
);
}
+173
View File
@@ -0,0 +1,173 @@
//! What the circuit copies and holds, counted in key clones and live keys, never in time.
//! Evidence: `context/test-evidence/zset.md`.
use std::cell::Cell;
use zset::{Circuit, ZSet};
thread_local! {
static CLONES: Cell<usize> = const { Cell::new(0) };
static LIVE: Cell<isize> = const { Cell::new(0) };
}
/// A key that counts its clones and how many of it are alive on this thread.
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
struct Counted(u32);
impl Counted {
fn new(n: u32) -> Self {
LIVE.with(|l| l.set(l.get() + 1));
Counted(n)
}
}
impl Clone for Counted {
fn clone(&self) -> Self {
CLONES.with(|c| c.set(c.get() + 1));
Counted::new(self.0)
}
}
impl Drop for Counted {
fn drop(&mut self) {
LIVE.with(|l| l.set(l.get() - 1));
}
}
fn clones() -> usize {
CLONES.with(Cell::get)
}
fn live() -> isize {
LIVE.with(Cell::get)
}
fn keys(n: u32) -> ZSet<Counted> {
(0..n).map(|i| (Counted::new(i), 1)).collect()
}
const N: u32 = 100;
#[test]
fn an_eval_borrows_its_input_instead_of_cloning_it() {
let mut c = Circuit::new();
let (input, s) = c.input::<Counted>();
let m = c.map(&s, |k| k.0);
let sink = c.output(&m);
input.push(keys(N));
let before = clones();
c.step();
assert_eq!(
clones() - before,
0,
"[piece] a step over {N} keys cloned keys"
);
assert_eq!(sink.take().len(), N as usize);
}
#[test]
fn an_integrate_stream_holds_its_integral_twice_when_the_tick_is_whole() {
let base = live();
let mut c = Circuit::new();
let (input, s) = c.input::<Counted>();
let i = c.integrate(&s);
input.push(keys(N));
c.step();
// The input's piece (N), the integral (N) and the integrate stream's piece (N).
assert_eq!(
live() - base,
3 * N as isize,
"[whole] live keys after one step"
);
drop((c, input, s, i));
assert_eq!(live(), base, "[leak] every key is dropped with the circuit");
}
type E = (Counted, u8);
#[derive(Clone, Copy)]
enum Mode {
Plain,
Commit,
Rollback,
}
/// Every stateful operator over `Counted` keys, loaded with `n` keys; then one step of a
/// two-key delta in `mode`. Returns the key clones that step (and `begin`/`end`) made.
fn clones_of_one_step(n: u32, mode: Mode) -> usize {
let mut c = Circuit::new();
let (a, sa) = c.input::<E>();
let (b, sb) = c.input::<E>();
let d = c.distinct(&sa);
let dl = c.delay(&sa);
let j = c.join_indexed(&sa, &sb, |k, x, y| (k.0, *x, *y));
let g = c.aggregate(&sa, |_, vals| Some(vals.len() as i64));
let _ig = c.integrate(&sa);
let sinks = (c.output(&d), c.output(&dl), c.output(&j), c.output(&g));
let rows =
|lo: u32, hi: u32| -> ZSet<E> { (lo..hi).map(|i| ((Counted::new(i), 0), 1)).collect() };
a.push(rows(0, n));
b.push(rows(0, n));
c.step();
for z in [&sinks.0, &sinks.1] {
z.take();
}
sinks.2.take();
sinks.3.take();
// A small step after the load: a `delay` holds the previous tick's input, so a
// transaction opened right after a big load would copy that load once.
a.push(rows(n, n + 2));
b.push(rows(n, n + 2));
c.step();
let before = clones();
if !matches!(mode, Mode::Plain) {
c.begin();
}
a.push(rows(n + 2, n + 4));
b.push(rows(n + 2, n + 4));
c.step();
match mode {
Mode::Plain => {}
Mode::Commit => c.commit(),
Mode::Rollback => c.rollback(),
}
clones() - before
}
#[test]
fn a_transaction_costs_the_keys_a_step_touches_and_not_the_size_of_the_circuit() {
for mode in [Mode::Commit, Mode::Rollback] {
let extra = |n| clones_of_one_step(n, mode) - clones_of_one_step(n, Mode::Plain);
let (small, large) = (extra(10), extra(300));
assert!(small > 0, "[cost] the log copies the keys it records");
assert_eq!(small, large, "[cost] the overhead grew with the state");
}
}
/// Clones made by one 1-key `step_continue` over an integral of `n` keys.
fn clones_of_one_continue(n: u32, txn: bool) -> usize {
let mut c = Circuit::new();
let (input, s) = c.input::<Counted>();
let _i = c.integrate(&s);
input.push((0..n).map(|k| (Counted::new(k), 1)).collect());
c.step();
let before = clones();
if txn {
c.begin();
}
input.push((n..n + 1).map(|k| (Counted::new(k), 1)).collect());
c.step_continue();
if txn {
c.commit();
}
clones() - before
}
/// A limit, pinned so the docs cannot drift from it: a continued piece clones the edge's
/// whole value once per transaction, which is the size of the state.
#[test]
fn a_split_tick_in_a_transaction_costs_the_size_of_the_edge() {
let extra = |n| clones_of_one_continue(n, true) - clones_of_one_continue(n, false);
let (small, large) = (extra(10), extra(300));
assert!(large >= 300, "[split] the edge's value is cloned: {large}");
assert!(large > small, "[split] and it grows with the state");
}
+39
View File
@@ -154,3 +154,42 @@ fn a_finished_rebuild_and_an_interrupted_check_do_not_trip_on_later_ticks() {
assert!(!job.interrupted(&c));
assert_eq!(job.step(&mut c, || true), Progress::Done);
}
#[test]
#[should_panic(expected = "another circuit")]
fn a_rebuild_stepped_on_a_foreign_circuit_is_refused_loudly() {
let (mut a, mut b) = (Circuit::new(), Circuit::new());
let (input, _s) = a.input::<u8>();
let mut job = Rebuild::new(input, [(1u8, 1)].into_iter().collect(), 1);
job.step(&mut b, || false);
}
#[test]
fn a_rebuild_knows_the_circuit_its_input_came_from() {
let (mut a, b) = (Circuit::new(), Circuit::new());
let (input, _s) = a.input::<u8>();
let job = Rebuild::new(input.clone(), ZSet::new(), 1);
assert!(
job.belongs_to(&a),
"[owner] the circuit that made the input"
);
assert!(!job.belongs_to(&b), "[owner] another circuit");
}
#[test]
fn an_interrupted_rebuild_stays_interrupted_and_a_new_job_loads_the_snapshot() {
let snap: Z<u8> = [(1, 1), (2, 1), (3, 1)].into_iter().collect();
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
let i = c.integrate(&s);
let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
assert_eq!(job.step(&mut c, || true), Progress::Pending);
c.step();
for _ in 0..3 {
assert!(job.interrupted(&c), "[forever] steps never un-interrupt it");
c.step();
}
let mut fresh = Rebuild::new(input, snap, 1);
assert_eq!(fresh.step(&mut c, || false), Progress::Done);
assert_eq!(i.current().len(), 3);
}
@@ -0,0 +1,8 @@
# Seeds for failure cases proptest has generated in the past. It is
# automatically read and these particular cases re-run before any
# novel cases are generated.
#
# It is recommended to check this file in to source control so that
# everyone who runs the test benefits from these saved cases.
cc b611223baeb182175151c758f4f6877b5897e5b067f9e5010d14e20b1c6c4418 # shrinks to pre = [], post = [(ZSet { weights: {} }, ZSet { weights: {} })], carry = ZSet { weights: {} }, bad = (ZSet { weights: {} }, ZSet { weights: {} }), fail = false
cc 71d785a9a4f55f12c26afef2c65babad2f29c1f2a30becf55936928d570254b4 # shrinks to pre = [], post = [(ZSet { weights: {} }, ZSet { weights: {} })], carry = ZSet { weights: {} }, bad = (ZSet { weights: {} }, ZSet { weights: {} }), fail = false
+309
View File
@@ -0,0 +1,309 @@
//! `Circuit::begin`/`rollback`/`commit` (ADR 009 clause 11): a failed step is undone, and
//! the circuit then behaves as if the step never ran. Evidence: `context/test-evidence/zset.md`.
use proptest::prelude::*;
use std::panic::{AssertUnwindSafe, catch_unwind};
use zset::{Circuit, Input, ZSet};
type Z<K> = ZSet<K>;
type P = (u8, u8);
/// A circuit using every stateful operator, and two closures: every stream's
/// `current()`, and every sink drained.
struct Rig {
c: Circuit,
a: Input<P>,
b: Input<P>,
view: Box<dyn Fn() -> String>,
drain: Box<dyn Fn() -> String>,
}
fn rig() -> Rig {
let mut c = Circuit::new();
let (a, sa) = c.input::<P>();
let (b, sb) = c.input::<P>();
let ia = c.integrate(&sa);
let da = c.delay(&sa);
let df = c.differentiate(&sa);
let ds = c.distinct(&sa);
let j = c.join_indexed(&sa, &sb, |k, x, y| (*k, *x, *y));
let ij = c.integrate(&j);
let ag = c.aggregate(&sa, |_, vals| {
Some(vals.iter().fold(0i64, |s, (v, w)| {
s.wrapping_add(i64::from(*v).wrapping_mul(*w))
}))
});
let pl = c.plus(&ia, &sb);
let sinks = (
c.output(&ia),
c.output(&da),
c.output(&df),
c.output(&ds),
c.output(&j),
c.output(&ij),
c.output(&ag),
c.output(&pl),
);
let (s1, s2) = (sa.clone(), sb.clone());
let view = move || {
format!(
"{:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?}",
s1.current(),
s2.current(),
ia.current(),
da.current(),
df.current(),
ds.current(),
j.current(),
ij.current(),
ag.current(),
pl.current(),
s1.current().len(),
)
};
let drain = move || {
format!(
"{:?} {:?} {:?} {:?} {:?} {:?} {:?} {:?}",
sinks.0.take(),
sinks.1.take(),
sinks.2.take(),
sinks.3.take(),
sinks.4.take(),
sinks.5.take(),
sinks.6.take(),
sinks.7.take(),
)
};
Rig {
c,
a,
b,
view: Box::new(view),
drain: Box::new(drain),
}
}
impl Rig {
fn tick(&mut self, a: &Z<P>, b: &Z<P>) {
self.a.push(a.clone());
self.b.push(b.clone());
self.c.step();
}
}
fn delta() -> impl Strategy<Value = Z<P>> {
prop::collection::vec(((0u8..4, 0u8..3), -3i64..=3), 0..6).prop_map(|p| p.into_iter().collect())
}
fn ticks(n: std::ops::Range<usize>) -> impl Strategy<Value = Vec<(Z<P>, Z<P>)>> {
prop::collection::vec((delta(), delta()), n)
}
/// The key that sits near `i64::MAX` once preloaded, so a later `+5` overflows it.
const BIG: P = (9, 9);
fn poison() -> Z<P> {
// `(0, 0)` sorts first, so operators have applied it before `BIG` overflows.
[((0, 0), 1), (BIG, 5)].into_iter().collect()
}
proptest! {
/// A step that fails half way, then a rollback, leaves the circuit as it was; the ticks
/// after run as if the failed step had never been attempted.
#[test]
fn a_rolled_back_step_leaves_the_circuit_as_it_was_and_the_ticks_after_unchanged(
pre in ticks(0..4),
post in ticks(1..4),
carry in delta(),
bad in (delta(), delta()),
fail in any::<bool>(),
) {
let (mut x, mut y) = (rig(), rig());
let preload: Z<P> = [(BIG, i64::MAX - 1)].into_iter().collect();
for (a, b) in pre.iter().chain([&(preload, Z::new())]) {
x.tick(a, b);
y.tick(a, b);
prop_assert_eq!((x.view)(), (y.view)(), "[pre] state");
prop_assert_eq!((x.drain)(), (y.drain)(), "[pre] output");
}
x.a.push(carry.clone());
y.a.push(carry);
let (view_before, drain_before) = ((x.view)(), (x.drain)());
x.c.begin();
let mut a = bad.0.clone();
if fail {
a += poison();
}
x.a.push(a);
x.b.push(bad.1.clone());
let outcome = catch_unwind(AssertUnwindSafe(|| x.c.step()));
prop_assert_eq!(outcome.is_err(), fail, "[panic] the poison step panics, the others do not");
x.c.rollback();
prop_assert_eq!((x.view)(), view_before, "[rollback] every stream's current()");
prop_assert_eq!((x.drain)(), drain_before, "[rollback] every sink");
for (i, (a, b)) in post.iter().enumerate() {
x.tick(a, b);
y.tick(a, b);
prop_assert_eq!((x.view)(), (y.view)(), "[post] state, tick {}", i);
prop_assert_eq!((x.drain)(), (y.drain)(), "[post] output, tick {}", i);
}
}
/// `commit` keeps what the steps did; a transaction that commits is not observable.
#[test]
fn a_committed_transaction_equals_the_same_steps_outside_one(
steps in ticks(1..6),
split in 0usize..6,
) {
let (mut x, mut y) = (rig(), rig());
for (i, (a, b)) in steps.iter().enumerate() {
let open = i >= split;
if open {
x.c.begin();
}
x.tick(a, b);
if open {
x.c.commit();
}
y.tick(a, b);
prop_assert_eq!((x.view)(), (y.view)(), "[commit] state, tick {}", i);
prop_assert_eq!((x.drain)(), (y.drain)(), "[commit] output, tick {}", i);
}
}
}
#[test]
fn an_operator_built_inside_a_transaction_goes_back_to_empty() {
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
c.begin();
let i = c.integrate(&s);
input.push([(1u8, 1)].into_iter().collect());
c.step();
c.rollback();
input.push([(2u8, 1)].into_iter().collect());
c.step();
let want: Z<u8> = [(2u8, 1)].into_iter().collect();
assert_eq!(
i.current(),
want,
"[built] the integral forgot the rolled-back tick"
);
}
/// One key at `MAX` that a transaction lowers by one and raises again. Undone oldest first,
/// the first undo would take it to `MAX + 1`; newest first, nothing leaves `i64`.
#[test]
fn rollback_undoes_newest_first_so_no_undo_leaves_i64() {
let mut c = Circuit::new();
let (input, s) = c.input::<P>();
let (i, d) = (c.integrate(&s), c.distinct(&s));
let g = c.aggregate(&s, |_, vals| Some(vals.len() as i64));
let one = |w: i64| -> Z<P> { [((1, 1), w)].into_iter().collect() };
input.push(one(i64::MAX));
c.step();
let before = (i.current(), d.current(), g.current());
c.begin();
for w in [-1, 1] {
input.push(one(w));
c.step();
}
c.rollback();
assert_eq!(
(i.current(), d.current(), g.current()),
before,
"[order] streams"
);
input.push(one(-1));
c.step();
assert_eq!(
i.current(),
one(i64::MAX - 1),
"[order] the integral is back at MAX first"
);
}
/// `i64::MIN` has no negation: a trace undoes it in two applications.
#[test]
fn a_minimum_weight_applied_to_every_kind_of_state_is_taken_back() {
let mut x = rig();
let mut y = rig();
let min: Z<P> = [((1, 1), i64::MIN)].into_iter().collect();
x.c.begin();
x.tick(&min, &Z::new());
x.c.rollback();
let right: Z<P> = [((1, 2), 1)].into_iter().collect();
x.tick(&Z::new(), &right);
y.tick(&Z::new(), &right);
assert_eq!((x.view)(), (y.view)(), "[min] state");
assert_eq!(
(x.drain)(),
(y.drain)(),
"[min] output: the join saw no left side"
);
}
#[test]
fn rollback_undoes_the_clock() {
let mut c = Circuit::new();
let (input, _s) = c.input::<u8>();
c.step();
c.begin();
input.push([(1u8, 1)].into_iter().collect());
c.step();
assert_eq!(c.ticks(), 2);
c.rollback();
assert_eq!(c.ticks(), 1, "[ticks] the rolled-back step is not counted");
}
#[test]
#[should_panic(expected = "already open")]
fn a_second_begin_is_refused_loudly() {
let mut c = Circuit::new();
c.begin();
c.begin();
}
#[test]
#[should_panic(expected = "no transaction is open")]
fn a_rollback_with_nothing_open_is_refused_loudly() {
Circuit::new().rollback();
}
#[test]
#[should_panic(expected = "no transaction is open")]
fn a_commit_with_nothing_open_is_refused_loudly() {
Circuit::new().commit();
}
/// A tick in two pieces, rolled back: the edge's whole value comes back, not the last piece.
#[test]
fn a_rollback_across_a_split_tick_restores_the_whole_value() {
let (mut x, mut y) = (rig(), rig());
let first: Z<P> = [((1, 1), 2), ((2, 1), 1)].into_iter().collect();
x.tick(&first, &Z::new());
y.tick(&first, &Z::new());
// The tick is already split when the transaction opens, so every edge holds a `whole`.
for r in [&mut x, &mut y] {
r.a.push([((3, 1), 1)].into_iter().collect());
r.c.step_continue();
}
x.c.begin();
for k in [4u8, 5] {
x.a.push([((k, 1), 1)].into_iter().collect());
x.c.step_continue();
}
x.c.rollback();
assert_eq!((x.view)(), (y.view)(), "[split] state after the rollback");
assert_eq!(
(x.drain)(),
(y.drain)(),
"[split] output after the rollback"
);
let next: Z<P> = [((5, 1), 1)].into_iter().collect();
x.tick(&next, &Z::new());
y.tick(&next, &Z::new());
assert_eq!((x.view)(), (y.view)(), "[split] the tick after");
}
+462
View File
@@ -0,0 +1,462 @@
//! A model-based property over every public operator: a rollback leaves the circuit equal to one
//! that never ran the step. From the review of #355. Evidence: `context/test-evidence/zset.md`.
use proptest::prelude::*;
use proptest::test_runner::{Config, TestRunner};
use std::cell::Cell;
use std::collections::BTreeMap;
use std::ops::Bound;
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
use zset::{Circuit, Input, Output, Stream, Trace, ZSet};
type P = (u8, u8);
type Z = ZSet<P>;
thread_local! { static TRIP: Cell<Option<u32>> = const { Cell::new(None) }; }
fn maybe_trip() {
TRIP.with(|t| {
if let Some(n) = t.get() {
if n == 0 {
t.set(None);
panic!("tripped");
}
t.set(Some(n - 1));
}
})
}
#[derive(Default)]
struct VecTrace<K, V>(BTreeMap<(K, V), i64>);
impl<K: Ord + Clone, V: Ord + Clone> Trace<K, V> for VecTrace<K, V> {
fn apply(&mut self, k: &K, v: &V, w: i64) {
maybe_trip();
let key = (k.clone(), v.clone());
let cur = self.0.get(&key).copied().unwrap_or(0);
let n = cur.checked_add(w).expect("overflow in VecTrace");
if n == 0 {
self.0.remove(&key);
} else {
self.0.insert(key, n);
}
}
fn lookup(&self, k: &K) -> Vec<(V, i64)> {
self.0
.iter()
.filter(|((kk, _), _)| kk == k)
.map(|((_, v), w)| (v.clone(), *w))
.collect()
}
fn range(&self, lo: Bound<&K>, hi: Bound<&K>) -> Vec<(K, V, i64)> {
self.0
.iter()
.filter(|((k, _), _)| std::ops::RangeBounds::contains(&(lo, hi), k))
.map(|((k, v), w)| (k.clone(), v.clone(), *w))
.collect()
}
}
#[derive(Clone, Debug)]
enum Op {
Integrate(usize),
Delay(usize),
Diff(usize),
Distinct(usize),
Plus(usize, usize),
Map(usize),
Filter(usize),
FlatMap(usize),
IndexBy(usize),
Join(usize, usize),
JoinCustom(usize, usize),
Agg(usize),
AggCustom(usize),
}
#[derive(Clone, Debug)]
enum Act {
Push(bool, Z),
Step(Option<u32>),
Cont(Option<u32>),
Begin,
Commit,
Rollback,
Build(Op),
}
struct World {
c: Circuit,
inputs: [Input<P>; 2],
streams: Vec<Stream<Z>>,
sinks: Vec<Output<P>>,
}
fn agg_f(_: &u8, vals: &[(u8, i64)]) -> Option<u8> {
maybe_trip();
Some(
vals.iter()
.fold(0u8, |s, (v, w)| s.wrapping_add(v.wrapping_mul(*w as u8))),
)
}
impl World {
fn new() -> Self {
let mut c = Circuit::new();
let (a, sa) = c.input::<P>();
let (b, sb) = c.input::<P>();
let sinks = vec![c.output(&sa), c.output(&sb)];
World {
c,
inputs: [a, b],
streams: vec![sa, sb],
sinks,
}
}
fn build(&mut self, op: &Op) {
let c = &mut self.c;
let st = &self.streams;
struct S<'a>(&'a [Stream<Z>]);
impl S<'_> {
fn s(&self, i: usize) -> Stream<Z> {
self.0[i % self.0.len()].clone()
}
}
let this = S(st);
let out: Stream<Z> = match *op {
Op::Integrate(i) => {
let s = this.s(i);
c.integrate(&s)
}
Op::Delay(i) => {
let s = this.s(i);
c.delay(&s)
}
Op::Diff(i) => {
let s = this.s(i);
c.differentiate(&s)
}
Op::Distinct(i) => {
let s = this.s(i);
c.distinct(&s)
}
Op::Plus(i, j) => {
let (x, y) = (this.s(i), this.s(j));
c.plus(&x, &y)
}
Op::Map(i) => {
let s = this.s(i);
c.map(&s, |p: &P| {
maybe_trip();
(p.1, p.0 % 4)
})
}
Op::Filter(i) => {
let s = this.s(i);
c.filter(&s, |p: &P| {
maybe_trip();
p.0.is_multiple_of(2)
})
}
Op::FlatMap(i) => {
let s = this.s(i);
c.flat_map(&s, |p: &P| {
maybe_trip();
vec![*p, (p.0, p.1.wrapping_add(1) % 4)]
})
}
Op::IndexBy(i) => {
let s = this.s(i);
let ix = c.index_by(&s, |p: &P| p.1);
c.map(&ix, |(g, p): &(u8, P)| (*g, p.0))
}
Op::Join(i, j) => {
let (x, y) = (this.s(i), this.s(j));
c.join_indexed(&x, &y, |k, a, b| (*k, (a ^ b) % 4))
}
Op::JoinCustom(i, j) => {
let (x, y) = (this.s(i), this.s(j));
c.join_indexed_in(
&x,
&y,
VecTrace::default(),
VecTrace::default(),
|k, a, b| (*k, a.wrapping_add(*b) % 4),
)
}
Op::Agg(i) => {
let s = this.s(i);
c.aggregate(&s, agg_f)
}
Op::AggCustom(i) => {
let s = this.s(i);
c.aggregate_in(&s, VecTrace::default(), agg_f)
}
};
let o = self.c.output(&out);
self.streams.push(out);
self.sinks.push(o);
}
fn view(&self) -> Vec<Z> {
self.streams.iter().map(|s| s.current()).collect()
}
fn drain(&self) -> Vec<Z> {
self.sinks.iter().map(|s| s.take()).collect()
}
}
static CASES: AtomicUsize = AtomicUsize::new(0);
static RB_AFTER_PANIC: AtomicUsize = AtomicUsize::new(0);
static RB_CLEAN: AtomicUsize = AtomicUsize::new(0);
static RB_AFTER_CONT: AtomicUsize = AtomicUsize::new(0);
static COMMITS: AtomicUsize = AtomicUsize::new(0);
static BUILT_IN_TXN: AtomicUsize = AtomicUsize::new(0);
static DOUBLE_BEGIN: AtomicUsize = AtomicUsize::new(0);
static OUTSIDE_PANIC_STOP: AtomicUsize = AtomicUsize::new(0);
static MULTI_STEP_TXN: AtomicUsize = AtomicUsize::new(0);
static TRIPS: AtomicUsize = AtomicUsize::new(0);
fn weight() -> impl Strategy<Value = i64> {
prop_oneof![
8 => -3i64..=3,
1 => Just(i64::MAX),
1 => Just(i64::MIN),
1 => Just(i64::MAX / 2 + 1),
1 => Just(-(i64::MAX / 2) - 1),
]
}
fn delta() -> impl Strategy<Value = Z> {
prop::collection::vec(((0u8..4, 0u8..4), weight()), 0..5).prop_map(|v| {
let mut z = Z::new();
for (k, w) in v {
// build without overflowing: replace if it would
if z.weight(&k).checked_add(w).is_some() {
z.insert(k, w);
}
}
z
})
}
fn op() -> impl Strategy<Value = Op> {
let i = 0usize..64;
prop_oneof![
i.clone().prop_map(Op::Integrate),
i.clone().prop_map(Op::Delay),
i.clone().prop_map(Op::Diff),
i.clone().prop_map(Op::Distinct),
(i.clone(), i.clone()).prop_map(|(a, b)| Op::Plus(a, b)),
i.clone().prop_map(Op::Map),
i.clone().prop_map(Op::Filter),
i.clone().prop_map(Op::FlatMap),
i.clone().prop_map(Op::IndexBy),
(i.clone(), i.clone()).prop_map(|(a, b)| Op::Join(a, b)),
(i.clone(), i.clone()).prop_map(|(a, b)| Op::JoinCustom(a, b)),
i.clone().prop_map(Op::Agg),
i.prop_map(Op::AggCustom),
]
}
fn act() -> impl Strategy<Value = Act> {
let trip = prop_oneof![2 => Just(None), 1 => (0u32..40).prop_map(Some)];
prop_oneof![
4 => (any::<bool>(), delta()).prop_map(|(b, z)| Act::Push(b, z)),
4 => trip.clone().prop_map(Act::Step),
2 => trip.prop_map(Act::Cont),
2 => Just(Act::Begin),
1 => Just(Act::Commit),
2 => Just(Act::Rollback),
1 => op().prop_map(Act::Build),
]
}
fn apply_plain(w: &mut World, a: &Act) {
match a {
Act::Push(b, z) => w.inputs[*b as usize].push(z.clone()),
Act::Step(_) => w.c.step(),
Act::Cont(_) => w.c.step_continue(),
Act::Build(op) => w.build(op),
_ => unreachable!(),
}
}
fn check(x: &World, y: &World, at: &str) -> Result<(), TestCaseError> {
prop_assert_eq!(x.c.ticks(), y.c.ticks(), "ticks {}", at);
prop_assert_eq!(x.view(), y.view(), "view {}", at);
prop_assert_eq!(x.drain(), y.drain(), "sinks {}", at);
Ok(())
}
fn run(ops: &[Op], acts: &[Act]) -> Result<(), TestCaseError> {
CASES.fetch_add(1, Relaxed);
let (mut x, mut y) = (World::new(), World::new());
for o in ops {
x.build(o);
y.build(o);
}
let mut open = false;
let mut failed = false;
let mut buf: Vec<Act> = vec![];
let mut steps_in_txn = 0;
let mut cont_in_txn = false;
for (n, a) in acts.iter().enumerate() {
let at = format!("after act {n} {a:?}");
match a {
Act::Begin => {
if open {
let r = catch_unwind(AssertUnwindSafe(|| x.c.begin()));
prop_assert!(r.is_err(), "double begin must panic");
DOUBLE_BEGIN.fetch_add(1, Relaxed);
} else {
x.c.begin();
open = true;
failed = false;
steps_in_txn = 0;
cont_in_txn = false;
}
}
Act::Commit if open && !failed => {
x.c.commit();
open = false;
for b in buf.drain(..) {
apply_plain(&mut y, &b);
}
COMMITS.fetch_add(1, Relaxed);
if steps_in_txn > 1 {
MULTI_STEP_TXN.fetch_add(1, Relaxed);
}
check(&x, &y, &at)?;
}
Act::Commit | Act::Rollback if open => {
x.c.rollback();
open = false;
if failed {
RB_AFTER_PANIC.fetch_add(1, Relaxed);
} else {
RB_CLEAN.fetch_add(1, Relaxed);
}
if cont_in_txn {
RB_AFTER_CONT.fetch_add(1, Relaxed);
}
if steps_in_txn > 1 {
MULTI_STEP_TXN.fetch_add(1, Relaxed);
}
for b in buf.drain(..) {
if let Act::Build(op) = b {
y.build(&op);
}
}
check(&x, &y, &at)?;
}
Act::Commit | Act::Rollback => {}
_ if open && failed => {} // only rollback after a failed step
Act::Step(t) | Act::Cont(t) if open => {
TRIP.with(|c| c.set(*t));
let r = catch_unwind(AssertUnwindSafe(|| apply_plain(&mut x, a)));
let tripped = TRIP.with(|c| c.get()).is_none() && t.is_some();
TRIP.with(|c| c.set(None));
steps_in_txn += 1;
if matches!(a, Act::Cont(_)) {
cont_in_txn = true;
}
if r.is_err() {
failed = true;
if tripped {
TRIPS.fetch_add(1, Relaxed);
}
} else {
buf.push(a.clone());
}
}
_ if open => {
if matches!(a, Act::Build(_)) {
BUILT_IN_TXN.fetch_add(1, Relaxed);
}
let r = catch_unwind(AssertUnwindSafe(|| apply_plain(&mut x, a)));
if r.is_err() {
failed = true;
} else {
buf.push(a.clone());
}
}
_ => {
let r = catch_unwind(AssertUnwindSafe(|| apply_plain(&mut x, a)));
if r.is_err() {
OUTSIDE_PANIC_STOP.fetch_add(1, Relaxed);
return Ok(());
}
apply_plain(&mut y, a);
check(&x, &y, &at)?;
}
}
}
if open {
x.c.rollback();
for b in buf.drain(..) {
if let Act::Build(op) = b {
y.build(&op);
}
}
if failed {
RB_AFTER_PANIC.fetch_add(1, Relaxed);
} else {
RB_CLEAN.fetch_add(1, Relaxed);
}
check(&x, &y, "final rollback")?;
}
// ticks after: both behave the same on a fresh tick
for i in 0..3u8 {
let z: Z = [((i, i), 1), ((i + 1, 0), -1)].into_iter().collect();
let r = catch_unwind(AssertUnwindSafe(|| {
x.inputs[0].push(z.clone());
x.inputs[1].push(z.clone());
x.c.step()
}));
if r.is_err() {
return Ok(());
}
y.inputs[0].push(z.clone());
y.inputs[1].push(z);
y.c.step();
check(&x, &y, &format!("post tick {i}"))?;
}
Ok(())
}
#[test]
fn reviewer_rollback_matches_a_circuit_that_never_ran_the_step() {
let cases: u32 = std::env::var("ATTACK_CASES")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(256);
std::panic::set_hook(Box::new(|_| {}));
let mut runner = TestRunner::new(Config {
cases,
..Config::default()
});
let strat = (
prop::collection::vec(op(), 1..12),
prop::collection::vec(act(), 1..40),
);
let r = runner.run(&strat, |(ops, acts)| run(&ops, &acts));
eprintln!(
"cases={} rb_after_panic={} rb_clean={} rb_after_cont={} commits={} built_in_txn={} double_begin={} outside_panic_stop={} multi_step_txn={} trips={}",
CASES.load(Relaxed),
RB_AFTER_PANIC.load(Relaxed),
RB_CLEAN.load(Relaxed),
RB_AFTER_CONT.load(Relaxed),
COMMITS.load(Relaxed),
BUILT_IN_TXN.load(Relaxed),
DOUBLE_BEGIN.load(Relaxed),
OUTSIDE_PANIC_STOP.load(Relaxed),
MULTI_STEP_TXN.load(Relaxed),
TRIPS.load(Relaxed)
);
let _ = std::panic::take_hook();
if let Err(e) = r {
eprintln!("FAILURE: {e:?}");
panic!("attack failed");
}
}
+167
View File
@@ -0,0 +1,167 @@
//! A `Rebuild` stepped across a rollback, and closures that read streams during an eval.
//! Evidence: `context/test-evidence/zset.md`.
use std::panic::{AssertUnwindSafe, catch_unwind};
use zset::{Circuit, Progress, Rebuild, ZSet};
type Z<K> = ZSet<K>;
fn quiet() {
static O: std::sync::Once = std::sync::Once::new();
O.call_once(|| {
let prev = std::panic::take_hook();
std::panic::set_hook(Box::new(move |info| {
let s = format!("{info}");
if !(s.contains("tripped") || s.contains("overflow")) {
prev(info);
}
}));
});
}
fn z(keys: &[(u8, i64)]) -> Z<u8> {
quiet();
keys.iter().copied().collect()
}
#[test]
fn r1_rebuild_begun_before_txn_continued_inside_then_rolled_back_reports_interrupted() {
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
let i = c.integrate(&s);
let snap = z(&[(1, 1), (2, 1), (3, 1), (4, 1)]);
let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
assert_eq!(job.step(&mut c, || true), Progress::Pending);
c.begin();
assert_eq!(job.step(&mut c, || true), Progress::Pending);
c.rollback();
let interrupted = job.interrupted(&c);
// Docs (zset.md Transactions 6, header): the job reports itself dead after a rollback.
let mut finished = None;
if !interrupted {
while job.step(&mut c, || false) == Progress::Pending {}
finished = Some(i.current());
}
assert!(
interrupted,
"[r1] job not interrupted after rollback; finishing it gives {:?}, want {:?}",
finished, snap
);
}
#[test]
fn r2_rebuild_finished_inside_a_rolled_back_txn_is_reported_done_with_nothing_loaded() {
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
let i = c.integrate(&s);
let snap = z(&[(1, 1), (2, 1), (3, 1)]);
let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
c.begin();
assert_eq!(job.step(&mut c, || false), Progress::Done);
c.rollback();
let interrupted = job.interrupted(&c);
assert!(
interrupted,
"[r2] after rollback: not interrupted, integral {:?}",
i.current()
);
}
#[test]
fn r3_an_interrupted_rebuild_resurrects_after_rollback_and_a_later_step() {
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
let i = c.integrate(&s);
c.step(); // ticks 1
let snap = z(&[(1, 1), (2, 1), (3, 1), (4, 1)]);
let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
c.begin();
assert_eq!(job.step(&mut c, || true), Progress::Pending); // began_at = 2
c.rollback(); // ticks back to 1
assert!(job.interrupted(&c), "[r3] dead right after rollback");
c.step(); // ticks 2 again
let dead_after_step = job.interrupted(&c);
let mut got = None;
if !dead_after_step {
while job.step(&mut c, || false) == Progress::Pending {}
got = Some(i.current());
}
assert!(
dead_after_step,
"[r3] interrupted job came back to life after a later step; finishing it gives {:?}, want {:?}",
got, snap
);
}
#[test]
fn r4_rebuild_whose_first_chunk_panics_in_a_txn_then_rolled_back() {
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
let i = c.integrate(&s);
input.push(z(&[(1, i64::MAX)]));
c.step();
let snap = z(&[(1, 1), (2, 1), (3, 1)]);
let mut job = Rebuild::new(input.clone(), snap.clone(), 1);
c.begin();
let r = catch_unwind(AssertUnwindSafe(|| job.step(&mut c, || false)));
assert!(r.is_err());
c.rollback();
let interrupted = job.interrupted(&c);
let mut got = None;
if !interrupted {
input.push(z(&[(1, -5)])); // make room so the retry does not overflow
while job.step(&mut c, || false) == Progress::Pending {}
got = Some(i.current());
}
assert!(
interrupted,
"[r4] after a panicked first chunk + rollback, the job is not interrupted; finishing gives {:?}",
got
);
}
#[test]
fn b1_a_closure_that_reads_a_stream_during_eval_does_not_panic() {
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
let s2 = s.clone();
let i = c.integrate(&s);
let i2 = i.clone();
// map reads its own source and another node's output during eval
let m = c.map(&s, move |k| {
k.wrapping_add(s2.current().len() as u8 + i2.current().len() as u8)
});
let m2 = m.clone();
// a later node reads an earlier node's output inside its own eval
let _f = c.filter(&i, move |_| m2.current().len() < 1000);
input.push(z(&[(1, 1), (2, 1)]));
c.step();
c.begin();
input.push(z(&[(3, 1)]));
c.step_continue();
c.rollback();
c.step();
}
#[test]
fn p1_a_chunk_whose_push_panics_in_a_txn_is_counted() {
let mut c = Circuit::new();
let (input, s) = c.input::<u8>();
let i = c.integrate(&s);
input.push(z(&[(1, i64::MAX)]));
let mut job = Rebuild::new(input.clone(), z(&[(1, 1), (2, 1)]), 1);
c.begin();
let r = catch_unwind(AssertUnwindSafe(|| job.step(&mut c, || false)));
assert!(r.is_err(), "the push overflows");
c.rollback();
let interrupted = job.interrupted(&c);
let mut got = None;
if !interrupted {
while job.step(&mut c, || false) == Progress::Pending {}
got = Some(i.current());
}
assert!(
interrupted,
"[p1] the chunk left the job unstepped; finishing it gives {got:?}"
);
}