diff --git a/context/code/delivery.md b/context/code/delivery.md index 30eb8605..17db04d6 100644 --- a/context/code/delivery.md +++ b/context/code/delivery.md @@ -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`. `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 diff --git a/context/code/source.md b/context/code/source.md index 7ed4d1d1..056046c4 100644 --- a/context/code/source.md +++ b/context/code/source.md @@ -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. diff --git a/context/code/zset-ffi.md b/context/code/zset-ffi.md index b7e36581..64ed53cd 100644 --- a/context/code/zset-ffi.md +++ b/context/code/zset-ffi.md @@ -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 { 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 { 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 diff --git a/context/code/zset.md b/context/code/zset.md index 4c1380cb..ae025393 100644 --- a/context/code/zset.md +++ b/context/code/zset.md @@ -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. diff --git a/context/test-evidence/core.md b/context/test-evidence/core.md index 69226c41..32e16dc9 100644 --- a/context/test-evidence/core.md +++ b/context/test-evidence/core.md @@ -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). diff --git a/context/test-evidence/delivery.md b/context/test-evidence/delivery.md index 6ef10e55..fdb92e52 100644 --- a/context/test-evidence/delivery.md +++ b/context/test-evidence/delivery.md @@ -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. diff --git a/context/test-evidence/zset-ffi.md b/context/test-evidence/zset-ffi.md index b1a83910..b7c34f1c 100644 --- a/context/test-evidence/zset-ffi.md +++ b/context/test-evidence/zset-ffi.md @@ -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]`. diff --git a/context/test-evidence/zset.md b/context/test-evidence/zset.md index 0089a2f0..1605fa25 100644 --- a/context/test-evidence/zset.md +++ b/context/test-evidence/zset.md @@ -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]`. diff --git a/crates/improvise-command/src/ui/delivery_tests.rs b/crates/improvise-command/src/ui/delivery_tests.rs index 10e05a0b..dcad00d6 100644 --- a/crates/improvise-command/src/ui/delivery_tests.rs +++ b/crates/improvise-command/src/ui/delivery_tests.rs @@ -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, ¬ices); } + /// 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)?; } } diff --git a/crates/improvise-core/src/model/over_tests.rs b/crates/improvise-core/src/model/over_tests.rs index d8078a5d..305e8822 100644 --- a/crates/improvise-core/src/model/over_tests.rs +++ b/crates/improvise-core/src/model/over_tests.rs @@ -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; diff --git a/crates/improvise-core/src/model/source.rs b/crates/improvise-core/src/model/source.rs index 5c3fd8a0..944ae13f 100644 --- a/crates/improvise-core/src/model/source.rs +++ b/crates/improvise-core/src/model/source.rs @@ -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, + greatest: Option, + ) -> Self { + let held = |x: Option| 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, b: Option, pick: fn(f64, f64) -> f64) -> Option 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 { 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 + '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 = 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 = store.keys(&[]).collect(); + assert_eq!(all, rows.iter().map(|(k, _)| k.clone()).collect::>()); + let east = vec![pair("R", "E")]; + let under: Vec = store.keys(&east).collect(); + let want: Vec = store.cells(&east).map(|c| c.0).collect(); + assert_eq!(under, want); + assert!(!under.is_empty()); + } } diff --git a/crates/improvise-core/src/model/types.rs b/crates/improvise-core/src/model/types.rs index d9c161e8..0e5c0193 100644 --- a/crates/improvise-core/src/model/types.rs +++ b/crates/improvise-core/src/model/types.rs @@ -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) -> Result { @@ -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); } diff --git a/crates/improvise-core/src/view/layout.rs b/crates/improvise-core/src/view/layout.rs index 5c8ae905..4a4520d9 100644 --- a/crates/improvise-core/src/view/layout.rs +++ b/crates/improvise-core/src/view/layout.rs @@ -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(); diff --git a/crates/improvise-protocol/src/delivery/client.rs b/crates/improvise-protocol/src/delivery/client.rs index 49d899c6..87a20a77 100644 --- a/crates/improvise-protocol/src/delivery/client.rs +++ b/crates/improvise-protocol/src/delivery/client.rs @@ -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) { + 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) { + self.apply_with(c, true) + } + + fn apply_with(&mut self, c: &Correction, answer: bool) -> (Outcome, Vec) { 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| { 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); diff --git a/crates/zset-ffi/include/zset_ffi.h b/crates/zset-ffi/include/zset_ffi.h index ae38877f..93857477 100644 --- a/crates/zset-ffi/include/zset_ffi.h +++ b/crates/zset-ffi/include/zset_ffi.h @@ -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 diff --git a/crates/zset-ffi/src/engine.rs b/crates/zset-ffi/src/engine.rs index 68f90e50..7fb643ce 100644 --- a/crates/zset-ffi/src/engine.rs +++ b/crates/zset-ffi/src/engine.rs @@ -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>, 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 { 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) diff --git a/crates/zset-ffi/src/lib.rs b/crates/zset-ffi/src/lib.rs index 835e5e45..3d488319 100644 --- a/crates/zset-ffi/src/lib.rs +++ b/crates/zset-ffi/src/lib.rs @@ -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)] diff --git a/crates/zset-ffi/tests/abi.rs b/crates/zset-ffi/tests/abi.rs index 770efa4a..6f73d235 100644 --- a/crates/zset-ffi/tests/abi.rs +++ b/crates/zset-ffi/tests/abi.rs @@ -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); +} diff --git a/crates/zset-ffi/tests/rollback_abi.rs b/crates/zset-ffi/tests/rollback_abi.rs new file mode 100644 index 00000000..9c0e0f18 --- /dev/null +++ b/crates/zset-ffi/tests/rollback_abi.rs @@ -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}"); +} diff --git a/crates/zset-ffi/tests/smoke.lisp b/crates/zset-ffi/tests/smoke.lisp index 09af9937..30cf74e7 100644 --- a/crates/zset-ffi/tests/smoke.lisp +++ b/crates/zset-ffi/tests/smoke.lisp @@ -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~%") diff --git a/crates/zset/src/circuit.rs b/crates/zset/src/circuit.rs index 39ee55fd..8aee5bbe 100644 --- a/crates/zset/src/circuit.rs +++ b/crates/zset/src/circuit.rs @@ -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 { piece: T, - whole: T, + whole: Option, + /// Inside a transaction: the `(piece, whole)` the first publish replaced. + saved: Option<(T, Option)>, + open: bool, } impl Default for Slot { fn default() -> Self { Slot { piece: T::default(), - whole: T::default(), + whole: None, + saved: None, + open: false, } } } @@ -41,18 +48,51 @@ impl Piecewise for ZSet { pub struct Stream(Rc>>); impl Stream { - 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 Stream { + 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 Clone for Stream { impl Stream { /// 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(Rc>>); +/// 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(Rc>>, Rc<()>); impl Clone for Input { fn clone(&self) -> Self { - Input(Rc::clone(&self.0)) + Input(Rc::clone(&self.0), Rc::clone(&self.1)) } } @@ -84,7 +126,8 @@ impl Input { } } -/// 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(Rc>>); impl Output { @@ -93,78 +136,237 @@ impl Output { } } -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 { +/// One input, one output, and the state `f` may change. +struct Unary { src: Stream, out: Stream, + state: S, f: F, } -impl O> Node for Unary { +impl Transactional for Unary { + 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 O> Node + for Unary +{ 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 { +struct Binary { a: Stream, b: Stream, out: Stream, + state: S, f: F, } -impl O> Node for Binary { +impl Transactional for Binary { + 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 O> Node + for Binary +{ 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 { + held: ZSet, + open: bool, + saved: Option>, +} + +impl Held { + /// 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 Transactional for Held { + 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 { src: Stream>, out: Stream>, - held: ZSet, + state: Held, +} + +impl Transactional for Delay { + 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 Node for Delay { 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 { + acc: ZSet, + log: Log<(K, i64)>, +} + +impl Transactional for Acc { + 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 { src: Stream>, out: Stream>, - acc: ZSet, + state: Acc, +} + +impl Transactional for Integrate { + 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 Node for Integrate { 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 { pending: Rc>>, out: Stream>, + saved: Option>, +} + +impl Transactional for Source { + 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 Node for Source { @@ -174,12 +376,42 @@ impl Node for Source { } } +/// The sink's sum, and the log that takes a transaction's additions back. +struct Sink { + sum: Rc>>, + log: Log<(K, i64)>, +} + +impl Transactional for Sink { + 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>, 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, + /// 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 = ZSet; @@ -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) { + if self.open.is_some() { + node.begin(); + } + self.nodes.push(node); + } + + fn unary_with( + &mut self, + src: &Stream, + state: S, + f: impl FnMut(&mut S, &I) -> O + 'static, + ) -> Stream { + let out = Stream(Rc::default()); + self.add(Box::new(Unary { + src: src.clone(), + out: out.clone(), + state, + f, + })); + out + } + fn unary( &mut self, src: &Stream, - f: impl FnMut(&I) -> O + 'static, + mut f: impl FnMut(&I) -> O + 'static, + ) -> Stream { + 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, + b: &Stream, + state: S, + f: impl FnMut(&mut S, &A, &B) -> O + 'static, ) -> Stream { 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, b: &Stream, - f: impl FnMut(&A, &B) -> O + 'static, + mut f: impl FnMut(&A, &B) -> O + 'static, ) -> Stream { - 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(&mut self) -> (Input, Stream>) { let pending: Rc>> = 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(&mut self, s: &Stream>) -> Output { let sink = Output(Rc::default()); - let acc = Rc::clone(&sink.0); - self.unary(s, move |d: &Z| { - *acc.borrow_mut() += d; + let state = Sink { + sum: Rc::clone(&sink.0), + log: Log::default(), + }; + self.unary_with(s, state, |sink, d: &Z| { + 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(&mut self, s: &Stream>) -> Stream> { 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(&mut self, s: &Stream>) -> Stream> { 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(&mut self, s: &Stream>) -> Stream> { - let mut st = DistinctState::new(); - self.unary(s, move |d: &Z| st.step(d)) + self.unary_with(s, DistinctState::new(), |st, d: &Z| st.step(d)) } /// The bilinear delta join over in-memory traces. @@ -374,8 +691,8 @@ impl Circuit { TA: Trace + 'static, TB: Trace + '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 + '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 { input: Input, rest: std::collections::btree_map::IntoIter, chunk: usize, /// The circuit's tick count right after this job's first chunk, once it has run one. began_at: Option, + /// The circuit's `undone` count when this job first ran, even if that chunk panicked. + undone_at: Option, } impl Rebuild { @@ -437,27 +756,40 @@ impl Rebuild { 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 = 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(); diff --git a/crates/zset/src/lib.rs b/crates/zset/src/lib.rs index eef8e806..e18e5ebf 100644 --- a/crates/zset/src/lib.rs +++ b/crates/zset/src/lib.rs @@ -91,7 +91,7 @@ impl ZSet { 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 }; diff --git a/crates/zset/src/ops.rs b/crates/zset/src/ops.rs index 95a9701f..e1634285 100644 --- a/crates/zset/src/ops.rs +++ b/crates/zset/src/ops.rs @@ -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 { + on: bool, + entries: Vec, +} + +impl Default for Log { + fn default() -> Self { + Log { + on: false, + entries: Vec::new(), + } + } +} + +impl Log { + 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 { + self.on.then(entry) + } + + /// Logs an entry made by `entry`, once its change has been made. + pub(crate) fn push(&mut self, entry: Option) { + 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(z: &mut ZSet, d: &ZSet, 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(z: &mut ZSet, (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 { held: ZSet, + log: Log<(K, i64)>, } impl DistinctState { pub fn new() -> Self { - DistinctState { held: ZSet::new() } + DistinctState { + held: ZSet::new(), + log: Log::default(), + } } pub fn step(&mut self, d: &ZSet) -> ZSet { @@ -21,12 +101,84 @@ impl DistinctState { 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 Transactional for DistinctState { + 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 { + inner: T, + log: Log<(K, V, i64)>, +} + +impl Undoable { + pub(crate) fn new(inner: T) -> Self { + Undoable { + inner, + log: Log::default(), + } + } +} + +impl, K: Clone, V: Clone> Trace for Undoable { + 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, K, V> Transactional for Undoable { + 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 JoinState { } } +impl Transactional for JoinState { + 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 AggregateState { out } } + +impl Transactional for AggregateState { + fn begin(&mut self) { + self.groups.begin(); + } + + fn commit(&mut self) { + self.groups.commit(); + } + + fn rollback(&mut self) { + self.groups.rollback(); + } +} diff --git a/crates/zset/src/trace.rs b/crates/zset/src/trace.rs index c8c0a1a5..aa9bdc5c 100644 --- a/crates/zset/src/trace.rs +++ b/crates/zset/src/trace.rs @@ -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 { - /// 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(lo: Bound<&K>, hi: Bound<&K>) -> bool { impl Trace for MemTrace { 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 => {} } } diff --git a/crates/zset/tests/clone_panic.rs b/crates/zset/tests/clone_panic.rs new file mode 100644 index 00000000..fae17b24 --- /dev/null +++ b/crates/zset/tests/clone_panic.rs @@ -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 = ZSet; + +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> = 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: Input, + streams: Vec>>, + agg: Stream>, + sinks: Vec>, +} + +fn rig(which: u8) -> Rig { + let mut c = Circuit::new(); + let (a, sa) = c.input::(); + let (b, sb) = c.input::(); + 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::>(), + self.agg.current() + ) + } + fn drain(&self) -> String { + format!( + "{:?}", + self.sinks.iter().map(|s| s.take()).collect::>() + ) + } +} + +fn bz(v: &[(u8, u8, i64)]) -> Z { + v.iter() + .map(|(k, x, w)| ((Boom(*k), Boom(*x)), *w)) + .collect() +} + +fn clone_panic_sweep(which: u8) -> Vec { + 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 = 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" + ); +} diff --git a/crates/zset/tests/cost.rs b/crates/zset/tests/cost.rs new file mode 100644 index 00000000..7cce4d98 --- /dev/null +++ b/crates/zset/tests/cost.rs @@ -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 = const { Cell::new(0) }; + static LIVE: Cell = 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 { + (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::(); + 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::(); + 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::(); + let (b, sb) = c.input::(); + 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 { (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::(); + 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"); +} diff --git a/crates/zset/tests/rebuild.rs b/crates/zset/tests/rebuild.rs index cb549941..85958653 100644 --- a/crates/zset/tests/rebuild.rs +++ b/crates/zset/tests/rebuild.rs @@ -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::(); + 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::(); + 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 = [(1, 1), (2, 1), (3, 1)].into_iter().collect(); + let mut c = Circuit::new(); + let (input, s) = c.input::(); + 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); +} diff --git a/crates/zset/tests/rollback.proptest-regressions b/crates/zset/tests/rollback.proptest-regressions new file mode 100644 index 00000000..e24a730b --- /dev/null +++ b/crates/zset/tests/rollback.proptest-regressions @@ -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 diff --git a/crates/zset/tests/rollback.rs b/crates/zset/tests/rollback.rs new file mode 100644 index 00000000..46ffd72e --- /dev/null +++ b/crates/zset/tests/rollback.rs @@ -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 = ZSet; +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

, + b: Input

, + view: Box String>, + drain: Box String>, +} + +fn rig() -> Rig { + let mut c = Circuit::new(); + let (a, sa) = c.input::

(); + let (b, sb) = c.input::

(); + 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

, b: &Z

) { + self.a.push(a.clone()); + self.b.push(b.clone()); + self.c.step(); + } +} + +fn delta() -> impl Strategy> { + prop::collection::vec(((0u8..4, 0u8..3), -3i64..=3), 0..6).prop_map(|p| p.into_iter().collect()) +} + +fn ticks(n: std::ops::Range) -> impl Strategy, Z

)>> { + 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

{ + // `(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::(), + ) { + let (mut x, mut y) = (rig(), rig()); + let preload: Z

= [(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::(); + 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 = [(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::

(); + 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

{ [((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

= [((1, 1), i64::MIN)].into_iter().collect(); + x.c.begin(); + x.tick(&min, &Z::new()); + x.c.rollback(); + let right: Z

= [((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::(); + 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

= [((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

= [((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"); +} diff --git a/crates/zset/tests/rollback_model.rs b/crates/zset/tests/rollback_model.rs new file mode 100644 index 00000000..afb09121 --- /dev/null +++ b/crates/zset/tests/rollback_model.rs @@ -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

; + +thread_local! { static TRIP: Cell> = 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(BTreeMap<(K, V), i64>); + +impl Trace for VecTrace { + 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), + Cont(Option), + Begin, + Commit, + Rollback, + Build(Op), +} + +struct World { + c: Circuit, + inputs: [Input

; 2], + streams: Vec>, + sinks: Vec>, +} + +fn agg_f(_: &u8, vals: &[(u8, i64)]) -> Option { + 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::

(); + let (b, sb) = c.input::

(); + 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]); + impl S<'_> { + fn s(&self, i: usize) -> Stream { + self.0[i % self.0.len()].clone() + } + } + let this = S(st); + let out: Stream = 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 { + self.streams.iter().map(|s| s.current()).collect() + } + fn drain(&self) -> Vec { + 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 { + 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 { + 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 { + 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 { + let trip = prop_oneof![2 => Just(None), 1 => (0u32..40).prop_map(Some)]; + prop_oneof![ + 4 => (any::(), 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 = 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"); + } +} diff --git a/crates/zset/tests/rollback_rebuild.rs b/crates/zset/tests/rollback_rebuild.rs new file mode 100644 index 00000000..88326bc6 --- /dev/null +++ b/crates/zset/tests/rollback_rebuild.rs @@ -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 = ZSet; + +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 { + 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::(); + 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::(); + 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::(); + 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::(); + 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::(); + 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::(); + 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:?}" + ); +}