diff --git a/Cargo.lock b/Cargo.lock index ce3719f..4ccec60 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1386,6 +1386,7 @@ dependencies = [ "prettyplease 0.2.37", "proc-macro2", "quote", + "serde", "syn 2.0.119", "tempfile", "trace", @@ -1483,7 +1484,6 @@ name = "recorder" version = "0.1.0" dependencies = [ "libc-alloc", - "serde_json", "trace", "wit-bindgen 0.63.0", ] @@ -1876,6 +1876,7 @@ version = "0.1.0" dependencies = [ "serde", "serde_json", + "wit-bindgen 0.63.0", ] [[package]] @@ -2113,7 +2114,7 @@ dependencies = [ [[package]] name = "wasm-wave" version = "0.263.0" -source = "git+https://github.com/chenyan2002/wasm-tools.git?branch=extend-wave-2#7b107fc6879530c68e8c6a30f9aacc078de311b4" +source = "git+https://github.com/chenyan2002/wasm-tools.git?branch=extend-wave-2#1578e5afb7dd45cfeff6e42eed6dc49e058b28c5" dependencies = [ "logos 0.16.1", "thiserror 2.0.21", diff --git a/Cargo.toml b/Cargo.toml index 0ffc27e..4a391e8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,6 +16,7 @@ edition = "2024" [dependencies] trace = { path = "crates/trace" } +serde.workspace = true anyhow = "1.0.98" clap = { version = "4.5.40", features = ["derive"] } heck = "0.5.0" diff --git a/Makefile b/Makefile index 5f10073..3ec1269 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: all build-components build-cli test test-cargo test-record test-fixture test-fuzz test-dialog run-fuzz run-record run-dialog run-viceroy +.PHONY: all build-components build-cli test test-cargo test-record test-fixture test-record-p3 test-record-http-p3 test-fuzz-replay test-fuzz test-dialog run-fuzz run-fuzz-replay run-record run-dialog run-viceroy # `make test ONE_BYTE_PAGE=1` tests with 1-byte-page memories ifeq ($(ONE_BYTE_PAGE),1) @@ -7,6 +7,9 @@ WASMTIME_FLAGS = -W custom-page-sizes=y endif INSTRUMENT = target/release/proxy-component instrument $(INSTRUMENT_FLAGS) WASMTIME = wasmtime $(WASMTIME_FLAGS) +# Runs wasmtime, and exits with 2 if the main component trapped, rather than the proxy failed +# (see tests/run-wasm.sh) +RUN_WASM = tests/run-wasm.sh all: build-components build-cli build-cli: @@ -20,15 +23,22 @@ build-components: cp target/one-byte-page/wasm32-wasip2/release/debug.wasm assets/debug.one_byte_page.wasm cp target/one-byte-page/wasm32-wasip2/release/recorder.wasm assets/recorder.one_byte_page.wasm -test: test-cargo test-fuzz test-record test-fixture test-dialog +test: test-cargo test-fuzz test-record test-fixture test-record-p3 test-record-http-p3 test-fuzz-replay test-dialog # Snapshot tests of the generated sources test-cargo: cargo test +# The WASI p3 main component of tests/fixtures/p3, which uses error-context +P3_GUEST = tests/fixtures/p3/guest +P3_GUEST_OUT = target/fixture-p3 +P3_GUEST_WASM = $(P3_GUEST_OUT)/wasm32-wasip2/release/proxy_p3_guest.wasm test-fuzz: $(MAKE) run-fuzz WASM=tests/calculator.wasm $(MAKE) run-fuzz WASM=tests/wasi_http.wasm + $(MAKE) run-fuzz WASM=tests/wasi_http_p3.wasm + cargo build --manifest-path $(P3_GUEST)/Cargo.toml --target wasm32-wasip2 --release --target-dir $(P3_GUEST_OUT) + $(MAKE) run-fuzz WASM=$(P3_GUEST_WASM) WASMTIME_FLAGS="$(WASMTIME_FLAGS) -W component-model-error-context=y" # build-only test $(INSTRUMENT) -m fuzz tests/rust.wasm $(INSTRUMENT) -m fuzz tests/go.wasm @@ -47,9 +57,9 @@ test-record: $(INSTRUMENT) -m record tests/wasi_http.wasm $(INSTRUMENT) -m replay tests/wasi_http.wasm -# Records and replays the main component of tests/fixtures/proxy-run/guest, whose resources are +# Records and replays the main component of tests/fixtures/p2/guest, whose resources are # dropped by the guest and by the host, in a composition with a host and a driver (see compose.wac) -FIXTURE = tests/fixtures/proxy-run +FIXTURE = tests/fixtures/p2 FIXTURE_OUT = target/fixture FIXTURE_WASM = $(FIXTURE_OUT)/wasm32-wasip2/release test-fixture: @@ -60,21 +70,106 @@ test-fixture: wac compose $(FIXTURE)/compose.wac -d test:mockhost=$(FIXTURE_WASM)/proxy_host.wasm -d test:driver=$(FIXTURE_WASM)/proxy_driver.wasm -d test:main=composed.wasm -o $(FIXTURE_OUT)/record.wasm $(WASMTIME) --invoke 'run()' $(FIXTURE_OUT)/record.wasm > $(FIXTURE_OUT)/trace.out # the host's drops of the guest's sessions - grep -c '"ExportArgs":{"method":"drop"' $(FIXTURE_OUT)/trace.out | grep -qx 2 + grep -c '"by":"Host","at":null,"event":{"Drop"' $(FIXTURE_OUT)/trace.out | grep -qx 2 $(INSTRUMENT) -m replay $(FIXTURE_WASM)/proxy_guest.wasm $(WASMTIME) --invoke 'start()' composed.wasm < $(FIXTURE_OUT)/trace.out $(INSTRUMENT) -m replay --use-host-recorder $(FIXTURE_WASM)/proxy_guest.wasm target/release/proxy-component run composed.wasm --invoke 'start()' --trace $(FIXTURE_OUT)/trace.out +# Records the main component of tests/fixtures/p3-record/guest, whose async imports, streams +# and futures wasmtime implements itself, and replays each trace: `run` with each pair of delays, +# RETURNED for 0, or random ones, `race`, which cancels the call that loses, and `cat` with stdin +# in one chunk, in several, or none. +P3_RECORD = tests/fixtures/p3-record/guest +P3_RECORD_OUT = target/fixture-p3-record +P3_RECORD_WASM = $(P3_RECORD_OUT)/wasm32-wasip2/release/proxy_p3_record_guest.wasm +test-record-p3: + cargo build --manifest-path $(P3_RECORD)/Cargo.toml --target wasm32-wasip2 --release --target-dir $(P3_RECORD_OUT) + $(INSTRUMENT) -m record $(P3_RECORD_WASM) + cp composed.wasm $(P3_RECORD_OUT)/record.wasm + $(INSTRUMENT) -m replay $(P3_RECORD_WASM) + cp composed.wasm $(P3_RECORD_OUT)/replay.wasm + for call in "run 5 15" "run 15 5" "run 0 10" "run 10 0" "run 0 0" "run 5 5" "run random" "run random" "run random" "run random" "run random" \ + "race 5 30" "race 30 5" "race 10 0" "race 0 0" "race random" "race random" "race random"; do \ + set -- $$call; env=""; [ "$$2" = random ] || env="--env DELAY_A=$$2 --env DELAY_B=$$3"; \ + $(WASMTIME) $$env --invoke "$$1()" $(P3_RECORD_OUT)/record.wasm > $(P3_RECORD_OUT)/trace.out 2>&1 || { \ + cat $(P3_RECORD_OUT)/trace.out; echo "recording failed, with the output above"; exit 1; }; \ + $(WASMTIME) --invoke 'start()' $(P3_RECORD_OUT)/replay.wasm < $(P3_RECORD_OUT)/trace.out > $(P3_RECORD_OUT)/replay.out 2>&1 || { \ + cat $(P3_RECORD_OUT)/replay.out; echo "replay failed, with the output above, of the trace in $(P3_RECORD_OUT)/trace.out"; exit 1; }; \ + done + for input in "printf ''" "printf first; sleep 0.05; printf second" "for i in 1 2 3 4 5; do echo line \$$i; sleep 0.01; done" "head -c 10000 /dev/zero"; do \ + sh -c "$$input" | $(WASMTIME) --invoke 'cat()' $(P3_RECORD_OUT)/record.wasm > $(P3_RECORD_OUT)/trace.out 2> $(P3_RECORD_OUT)/stderr.out || { \ + cat $(P3_RECORD_OUT)/trace.out $(P3_RECORD_OUT)/stderr.out; echo "recording failed, with the output above"; exit 1; }; \ + $(WASMTIME) --invoke 'start()' $(P3_RECORD_OUT)/replay.wasm < $(P3_RECORD_OUT)/trace.out > $(P3_RECORD_OUT)/replay.out 2>&1 || { \ + cat $(P3_RECORD_OUT)/replay.out; echo "replay failed, with the output above, of the trace in $(P3_RECORD_OUT)/trace.out"; exit 1; }; \ + done + +# Records tests/wasi_http_p3.wasm under `wasmtime serve`, which reuses an instance for the requests, +# so its trace has an export call for each, and replays it. Not its echo (see +# docs/p3-record-replay.md, Labels), nor a fetch, which needs the network. +HTTP_OUT = target/fixture-http-p3 +HTTP_ADDR = 127.0.0.1:8789 +test-record-http-p3: + mkdir -p $(HTTP_OUT) + $(INSTRUMENT) -m record tests/wasi_http_p3.wasm + cp composed.wasm $(HTTP_OUT)/record.wasm + $(INSTRUMENT) -m replay tests/wasi_http_p3.wasm + cp composed.wasm $(HTTP_OUT)/replay.wasm + wasmtime serve $(WASMTIME_FLAGS) -S p3,http,cli --no-logging-prefix --addr $(HTTP_ADDR) $(HTTP_OUT)/record.wasm > $(HTTP_OUT)/trace.out 2> $(HTTP_OUT)/serve.err & echo $$! > $(HTTP_OUT)/serve.pid; \ + sleep 2; \ + for path in / "/wait?ms=20" /random "/wait?ms=5" /; do \ + curl -sf -m 10 "http://$(HTTP_ADDR)$$path" > /dev/null || { kill `cat $(HTTP_OUT)/serve.pid`; cat $(HTTP_OUT)/serve.err; echo "GET $$path failed"; exit 1; }; \ + done; \ + kill `cat $(HTTP_OUT)/serve.pid` + $(WASMTIME) --invoke 'start()' $(HTTP_OUT)/replay.wasm < $(HTTP_OUT)/trace.out > $(HTTP_OUT)/replay.out 2>&1 || { \ + cat $(HTTP_OUT)/replay.out; echo "replay failed, with the output above, of the trace in $(HTTP_OUT)/trace.out"; exit 1; } + +# The spike's guest, which joins two concurrent results in the order its wakeups receive them +RACE = tests/fixtures/spike-p3-race +RACE_OUT = target/fixture-race +RACE_WASM = $(RACE_OUT)/wasm32-wasip2/release/race_guest.wasm +# Fuzzes the main components of tests/fixtures/p2/guest, the p3 and p3-record fixtures, +# wasi_http_p3 and the spike's guest, and replays each trace it writes, which ends as the fuzzing run did: to the +# end, or in the main component's trap, e.g. a guest's panic on a fuzzed error. +test-fuzz-replay: + cargo build --manifest-path $(FIXTURE)/guest/Cargo.toml --target wasm32-wasip2 --release --target-dir $(FIXTURE_OUT) + $(MAKE) run-fuzz-replay WASM=$(FIXTURE_WASM)/proxy_guest.wasm OUT=$(FIXTURE_OUT) + cargo build --manifest-path $(P3_GUEST)/Cargo.toml --target wasm32-wasip2 --release --target-dir $(P3_GUEST_OUT) + $(MAKE) run-fuzz-replay WASM=$(P3_GUEST_WASM) OUT=$(P3_GUEST_OUT) WASMTIME_FLAGS="$(WASMTIME_FLAGS) -W component-model-error-context=y" + cargo build --manifest-path $(P3_RECORD)/Cargo.toml --target wasm32-wasip2 --release --target-dir $(P3_RECORD_OUT) + $(MAKE) run-fuzz-replay WASM=$(P3_RECORD_WASM) OUT=$(P3_RECORD_OUT) + $(MAKE) run-fuzz-replay WASM=tests/wasi_http_p3.wasm OUT=target/fixture-http-p3 + cargo build --manifest-path $(RACE)/Cargo.toml -p race-guest --target wasm32-wasip2 --release --target-dir $(RACE_OUT) + $(MAKE) run-fuzz-replay WASM=$(RACE_WASM) OUT=$(RACE_OUT) +run-fuzz-replay: + mkdir -p $(OUT) + $(INSTRUMENT) -m fuzz $(WASM) + cp composed.wasm $(OUT)/fuzz.wasm + $(INSTRUMENT) -m replay $(WASM) + cp composed.wasm $(OUT)/replay.wasm + for i in 1 2 3 4 5 6 7 8 9 10; do \ + $(RUN_WASM) $(OUT)/fuzz-trace.out $(WASMTIME_FLAGS) --invoke 'start()' $(OUT)/fuzz.wasm > /dev/null; fuzz=$$?; \ + if [ $$fuzz -eq 1 ]; then \ + cat $(OUT)/fuzz-trace.out; echo "fuzzing failed, with the output above, in $(OUT)/fuzz-trace.out"; exit 1; \ + fi; \ + $(RUN_WASM) $(OUT)/replay.out $(WASMTIME_FLAGS) --invoke 'start()' $(OUT)/replay.wasm < $(OUT)/fuzz-trace.out > /dev/null; \ + if [ $$? -ne $$fuzz ]; then \ + cat $(OUT)/replay.out; echo "replay failed, with the output above, in $(OUT)/replay.out, of the trace in $(OUT)/fuzz-trace.out"; exit 1; \ + fi; \ + done + test-dialog: rm tests/composed.wasm || true for wasm in tests/*.wasm; do \ - $(MAKE) run-dialog WASM=$$wasm; \ + $(MAKE) run-dialog WASM=$$wasm || exit 1; \ done + cargo build --manifest-path $(P3_GUEST)/Cargo.toml --target wasm32-wasip2 --release --target-dir $(P3_GUEST_OUT) + $(MAKE) run-dialog WASM=$(P3_GUEST_WASM) +# A trap of the main component is what fuzzing finds, not a failure. run-fuzz: $(INSTRUMENT) -m fuzz $(WASM) - $(WASMTIME) --invoke 'start()' composed.wasm + $(RUN_WASM) /tmp/composed.out $(WASMTIME_FLAGS) --invoke 'start()' composed.wasm; [ $$? -ne 1 ] run-record: $(INSTRUMENT) -m record $(WASM) diff --git a/README.md b/README.md index 58d953e..e06753a 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ To build the CLI tool, just run `make all`. To test record and fuzzing, run `make test`. `cargo test` checks the generated sources against the snapshots in `tests/snapshots`, generated -from the WIT in `tests/fixtures/proxy`. After an intended change, run `BLESS=1 cargo test` and +from the WIT in `tests/fixtures/p2/wit` and `tests/fixtures/p3/wit`. After an intended change, run `BLESS=1 cargo test` and review the snapshots' diff. ## Usage @@ -30,6 +30,12 @@ Run `composed.wasm` in the host runtime which the original wasm is supposed to r The host runtime can also choose to implement the [`record` interface](https://github.com/chenyan2002/proxy-component/blob/main/assets/recorder.wit#L3). Then we can use the `--use-host-recorder` flag to skip composing the guest-side record implementation. +Note that the recorded handle id is not an exact match with the handle id the main component sees. Each +component instance has its own handle table, and the main component can allocate handles in it +that the proxy never sees, e.g., its own resources and streams. Each of those +shifts the main component's handle ids, so the imports component's handle ids, from its own table, +can't match them exactly. + ### Replay Assuming the trace captured from the record phase is stored in `trace.out`. We can run the following to replay the trace. @@ -54,7 +60,7 @@ $ proxy-component instrument -m fuzz $ wasmtime --invoke 'start()' composed.wasm ``` -Fuzzing the import return and export input based on the WIT type. This mode requires a [Debug component](components/debug/) to get random numbers and logging. The `composed.wasm` can be run in +Fuzzing the import return and export input based on the WIT type. This mode requires a [Debug component](components/debug/) to get random numbers. An import's result is drawn from a hash of its arguments and a seed for the run, so the same call gets the same result within a run, and another in the next run. The run prints its trace to stdout, in the same format as record mode, so `wasmtime --invoke 'start()' composed.wasm > trace.out` stores it, for replay mode to replay. To read the trace instead, `wasmtime --env PROXY_TRACE=text --invoke 'start()' composed.wasm` prints it as text. The `composed.wasm` can be run in a standalone `wasmtime` without any special host functions. ### Dialog diff --git a/assets/executor.rs b/assets/executor.rs new file mode 100644 index 0000000..77c882b --- /dev/null +++ b/assets/executor.rs @@ -0,0 +1,304 @@ +// The executor of the imports component, which writes and reads the virtualized host's end of +// streams and futures, e.g. writing the body stream that a mock of the sync `consume-body` +// returns, or reading the body a mock of the sync `response.new` receives. +// +// Streams and futures themselves are not virtualized: `stream.new`, `stream.read` and the like +// are canonical built-ins, which the runtime implements, not imports. What the virtualized host +// plays is the other end. Only readable ends cross components; the writable end stays with the +// component that created the pair. So when a mock returns a stream, e.g. `consume-body`, it +// creates the pair, and must write to it after it has returned. Linking to the real host's +// streams would mean not virtualizing that interface, and in fuzz and replay modes there is no +// host. Playing the other end lets the host choose what it writes, and how it reads, e.g. +// stopping partway; replay the recorded chunks; observe what the guest writes; and put mocked +// resources in the values, e.g. the `fields` of a `future, error-code>>`. +// +// A sync function can't block or spawn a task, so it queues the work with `spawn` instead, and +// `run_executor` runs it in an async task, which lives for the whole run: the +// `proxy:conversion/executor.run` export, which the `start` function of the exports component +// calls. The exports component needs no executor: its work is only made by `start`, which is +// async, so its `spawn` spawns the work on the task of `start` (see `generate_start`). +// +// How it runs: `spawn` pushes a job and wakes `run_executor`, through wit-bindgen's inter-task +// wakeup, as a sync mock runs in another task. `run_executor` polls the running jobs, then the +// queued ones, until no more are queued, as a job may spawn another. As the jobs are polled in +// the task of `run`, a job blocked on a stream or future joins its waitable set, so the guest's +// side of the copy wakes `run`, which polls them all again: they share one waker. `start` calls +// `stop` once the exports it called are done, and `run` returns once the jobs are too. In dialog +// mode, `start` does so after each export call, and calls `run` again for the next one. +// +// When a value with a stream or future changes hands, e.g. an export returns a stream, the +// generated code doesn't await it. It gives it to `consume` (or `fuzz`, for one the host gives), +// which decides at once what the host does, e.g. in fuzz mode whether it reads the stream to the +// end, reads a few chunks, or drops it, and with what capacity; in dialog mode, the job asks the +// user at each step. A job then does it. When it runs is up to cooperative scheduling: first at +// the next point where its task suspends, e.g. the yields of `start` after each export call, an +// async call, or at the latest when `start` waits for the jobs: at the end in fuzz mode, and +// after each export call in dialog mode, so that the user is done with the streams and futures of +// a call before choosing the next one. Each read or write +// completes once the guest does its side. So the host's reads and writes overlap the following +// export calls, in an order the schedule decides, which is part of what fuzz mode explores. +// Awaiting the value right after the call instead would hide those orders, and deadlock a guest +// that only writes a stream in a later call, or keeps it for the rest of the run. +// +// The flip side: if the guest never finishes, or drops, a stream or future it gave the host, its +// job never ends, so neither do `run` nor `start`, and wasmtime traps with "deadlock detected". +// +// Steps: each run of the executor, i.e. each poll of `run_executor`, is a step, and so is each +// arrival of an export call, which the exports component counts with the `step` export. Host +// lines delivered outside a call carry the step, `at`, in which they happened (see +// `docs/p3-record-replay.md`, Steps). A job that waits for a later run, e.g. an async import +// the fuzzer made STARTED, makes the executor wake itself, as a host completion would. A run +// after it, as after a host completion, takes an event that came meanwhile (see +// `executor_callback`), so that a step means the same however the executor was woken. +type Job = std::pin::Pin>>; +#[derive(Default)] +struct Executor { + queue: Vec, + stopped: bool, + waker: Option, +} +thread_local! { + static EXECUTOR: std::cell::RefCell = std::cell::RefCell::new(Executor::default()); + // The last step, counted from 1. + static STEP: std::cell::Cell = const { std::cell::Cell::new(0) }; + // Set by a job waiting for a later run, so that the executor wakes itself. + static WAKE_AGAIN: std::cell::Cell = const { std::cell::Cell::new(false) }; + // The waits that ended in a run, whose tasks haven't resumed yet, which hold the executor (see + // `wait_for_run`), and whether it was woken meanwhile. + static HELD: std::cell::Cell = const { std::cell::Cell::new(0) }; + static WOKEN_WHILE_HELD: std::cell::Cell = const { std::cell::Cell::new(false) }; + // Set by `executor_callback` when the executor resumes after a yield, for its next poll to + // be a probe, and whether a probe was the last poll. + static PROBE: std::cell::Cell = const { std::cell::Cell::new(false) }; + static PROBED: std::cell::Cell = const { std::cell::Cell::new(false) }; +} +/// Counts a step, and returns it. +#[allow(dead_code)] +fn next_step() -> u32 { + STEP.set(STEP.get() + 1); + STEP.get() +} +/// The step a job runs in. +#[allow(dead_code)] +fn current_step() -> u32 { + STEP.get() +} +/// Waits, from any task, until the executor has run `runs` more times, and returns the step of +/// the last run. +#[allow(dead_code)] +async fn wait_for_runs(runs: u32) -> u32 { + let mut left = runs; + wait_for_run(move || { + if left == 0 { + return true; + } + left -= 1; + false + }) + .await +} +/// Waits, from any task, until a run of the executor in `step` or later, or returns at once if +/// the count is already there, e.g. replay's for a host line recorded in that step, which checks +/// that it came in that step. +#[allow(dead_code)] +async fn wait_until_step(step: u32) { + if current_step() >= step { + return; + } + wait_for_run(move || current_step() >= step).await; +} +/// Waits, from any task, as `wait_until_step` does, and takes the next step, of a host line +/// `start` makes, e.g. an export call's arrival, which it returns: the one after the run it +/// waited for, as the executor doesn't run in between (see `wait_for_run`). +#[allow(dead_code)] +async fn step_after(step: u32) -> u32 { + wait_until_step(step).await; + next_step() +} +/// Waits, in a job of the executor, until its run in `step`, waking the executor itself until +/// then, e.g. replay's job for a copy recorded in that step. +/// +/// Panics if the count passed `step`: replay diverged from the recording. +#[allow(dead_code)] +async fn reach_step(step: u32) { + std::future::poll_fn(|_| { + let now = current_step(); + assert!(now <= step, "replay reached step {now} before the run in step {step}"); + if now == step { + std::task::Poll::Ready(()) + } else { + WAKE_AGAIN.set(true); + std::task::Poll::Pending + } + }) + .await +} +/// Waits until a run of the executor in which `done` returns true, and returns its step. A job +/// asks in each run, waking the executor itself in between, as no host event would. +async fn wait_for_run(mut done: impl FnMut() -> bool + 'static) -> u32 { + let job = std::future::poll_fn(move |_| { + if done() { + return std::task::Poll::Ready(()); + } + WAKE_AGAIN.set(true); + std::task::Poll::Pending + }); + wait_for_job(job).await.1 +} +/// Runs `job` in the executor, from any task, and returns its value, and the step of the run it +/// completed in, e.g. record mode's host call that blocked, whose completion wakes the executor. +/// +/// The executor doesn't run again until the waiting task resumes, e.g. for a guest's event, so +/// that what the task does next, e.g. return a STARTED call, or take an export call's step, +/// follows the run it waited for, in record and replay alike: replay knows only the step. +/// +/// `job` may borrow from the waiting task, e.g. a host call on the resource the guest passed: it +/// is only polled while the waiting task waits, and dropped if the task is dropped first, e.g. +/// the guest cancelled its call, which cancels the host's call too. +#[allow(dead_code)] +async fn wait_for_job<'a, T: 'static>(job: impl std::future::Future + 'a) -> (T, u32) { + type Job<'a, T> = std::pin::Pin + 'a>>; + struct Done { + job: Option>, + value: Option<(T, u32)>, + waker: Option, + held: bool, + } + /// The waiting task's side, which drops the job if it is still pending, and releases the + /// executor once the task resumes, or is dropped first. + struct Waiter(std::rc::Rc>>); + impl Drop for Waiter { + fn drop(&mut self) { + let mut done = self.0.borrow_mut(); + drop(done.job.take()); + if std::mem::take(&mut done.held) { + HELD.set(HELD.get() - 1); + if HELD.get() == 0 && WOKEN_WHILE_HELD.replace(false) { + wake_executor(); + } + } + } + } + let job: Job<'a, T> = Box::pin(job); + // SAFETY: the job outlives `'a` only as an empty `Option`: the waiter, which `'a` outlives, + // drops it when it is dropped, and it is only polled before. + let job: Job<'static, T> = unsafe { std::mem::transmute(job) }; + let done = Done { job: Some(job), value: None, waker: None, held: false }; + let waiter = Waiter(std::rc::Rc::new(std::cell::RefCell::new(done))); + let job_shared = waiter.0.clone(); + spawn(std::future::poll_fn(move |cx| { + let mut shared = job_shared.borrow_mut(); + let Some(job) = shared.job.as_mut() else { + return std::task::Poll::Ready(()); + }; + let std::task::Poll::Ready(value) = job.as_mut().poll(cx) else { + return std::task::Poll::Pending; + }; + shared.job = None; + HELD.set(HELD.get() + 1); + shared.value = Some((value, current_step())); + shared.held = true; + if let Some(waker) = shared.waker.take() { + waker.wake(); + } + std::task::Poll::Ready(()) + })); + let value = std::future::poll_fn(|cx| { + let mut shared = waiter.0.borrow_mut(); + match shared.value.take() { + Some(value) => std::task::Poll::Ready(value), + None => { + shared.waker = Some(cx.waker().clone()); + std::task::Poll::Pending + } + } + }) + .await; + drop(waiter); + value +} +/// Queues `job`, from any function, sync or async. +#[allow(dead_code)] +fn spawn(job: impl std::future::Future + 'static) { + EXECUTOR.with_borrow_mut(|e| { + e.queue.push(Box::pin(job)); + }); + wake_executor(); +} +/// Wakes `run_executor` from another task, with wit-bindgen's `inter-task-wakeup`. +fn wake_executor() { + let waker = EXECUTOR.with_borrow_mut(|e| e.waker.take()); + if let Some(waker) = waker { + waker.wake(); + } +} +/// Polls the queued jobs until `stop_executor` is called and they are all done. +#[allow(dead_code)] +async fn run_executor() { + let mut running: Vec = Vec::new(); + std::future::poll_fn(|cx| { + // A probe isn't a run: it wakes the executor, for wit-bindgen to take an event that came + // while it yielded, before its run (see `executor_callback`). + if PROBE.replace(false) { + PROBED.set(true); + cx.waker().wake_by_ref(); + return std::task::Poll::Pending; + } + PROBED.set(false); + // Held, it isn't a run: the jobs are polled at the next, as every run polls them all. + if HELD.get() > 0 { + WOKEN_WHILE_HELD.set(true); + EXECUTOR.with_borrow_mut(|e| e.waker = Some(cx.waker().clone())); + return std::task::Poll::Pending; + } + next_step(); + WAKE_AGAIN.set(false); + running.retain_mut(|job| job.as_mut().poll(cx).is_pending()); + // A job may spawn another, so `EXECUTOR` isn't borrowed while jobs run. + loop { + let mut queued = EXECUTOR.with_borrow_mut(|e| std::mem::take(&mut e.queue)); + if queued.is_empty() { + break; + } + queued.retain_mut(|job| job.as_mut().poll(cx).is_pending()); + running.append(&mut queued); + } + EXECUTOR.with_borrow_mut(|e| { + if e.stopped && running.is_empty() { + std::task::Poll::Ready(()) + } else { + e.waker = Some(cx.waker().clone()); + if WAKE_AGAIN.get() { + // Yields, and comes back as the next run. + cx.waker().wake_by_ref(); + } + std::task::Poll::Pending + } + }) + }) + .await +} +/// The callback of the executor's task, which `run` exports, in place of wit-bindgen's (see +/// `patch_bindings` in the generator). Resumed after a yield, e.g. by a job waiting for a later +/// step, the executor first takes an event that came meanwhile, as a waiting executor woken by +/// that event does, e.g. in record mode: wit-bindgen would only take it after a poll, in the +/// next run. So its first poll is a probe, which only wakes it: wit-bindgen then takes a +/// pending event, if any, and polls it again. If there was none, wit-bindgen yields, and the +/// callback runs it at once instead. +#[allow(dead_code)] +unsafe fn executor_callback(event0: u32, event1: u32, event2: u32) -> u32 { + const EVENT_NONE: u32 = 0; + if event0 == EVENT_NONE { + PROBE.set(true); + } + let code = unsafe { wit_bindgen::rt::async_support::callback(event0, event1, event2) }; + if PROBED.replace(false) { + return unsafe { wit_bindgen::rt::async_support::callback(EVENT_NONE, 0, 0) }; + } + code +} +#[allow(dead_code)] +fn stop_executor() { + EXECUTOR.with_borrow_mut(|e| e.stopped = true); + wake_executor(); +} diff --git a/assets/executor.wit b/assets/executor.wit new file mode 100644 index 0000000..6c8657f --- /dev/null +++ b/assets/executor.wit @@ -0,0 +1,25 @@ +package proxy:conversion; + +// Runs the jobs of the imports component, which writes and reads the virtualized host's end of +// streams and futures, e.g. writing the body stream that a mock of the sync `consume-body` +// returns. Like the generated `conversion` interface of this package, the imports component +// exports it, for the exports component to call: its `start` function runs it. See +// `assets/executor.rs`. +interface executor { + run: async func(); + stop: func(); + // Counts a step that isn't a run of the executor, e.g. the host's drop of a guest's resource, + // which the exports component makes, and returns it (see `docs/p3-record-replay.md`, Steps). + step: func() -> u32; + // The last step. + current-step: func() -> u32; + // Waits until the executor's run in `step`, or returns at once if the count is already there. + wait-until-step: async func(step: u32); + // Waits as `wait-until-step` does, and counts the next step, as `step` does, without a run of + // the executor in between: `start` waits for the step before an export call's, which fuzz mode + // chooses, and replay's trace has, and the call arrives in the next. + step-after: async func(step: u32) -> u32; + // Whether the exports component is dropping mocks of the imports component itself, e.g. those + // it made for the borrowed arguments of an export call: the host's drops, not the guest's. + host-drops: func(dropping: bool); +} diff --git a/assets/recorder.one_byte_page.wasm b/assets/recorder.one_byte_page.wasm index 046ae20..60f9b2e 100644 Binary files a/assets/recorder.one_byte_page.wasm and b/assets/recorder.one_byte_page.wasm differ diff --git a/assets/recorder.wasm b/assets/recorder.wasm index 9e284ac..6fa975e 100644 Binary files a/assets/recorder.wasm and b/assets/recorder.wasm differ diff --git a/assets/recorder.wit b/assets/recorder.wit index ec13214..46f2170 100644 --- a/assets/recorder.wit +++ b/assets/recorder.wit @@ -1,18 +1,89 @@ package proxy:recorder@0.1.0; +// The lines of a trace, as `crates/trace` defines them (see `docs/p3-record-replay.md`, Trace +// format). Values are WAVE text, e.g. `ok(#response-3)`. +interface types { + enum party { host, guest } + + // The generated component that numbers a handle or a task. + enum table { imports, exports } + + // A call, numbered in order by the component that sees it. + record task-id { table: table, id: u32 } + + // A handle, by its name as WAVE writes it, e.g. `#stream-7`. + record label { table: table, name: string } + + record call { id: task-id, is-async: bool, %func: string, args: list } + record return-value { id: task-id, value: option } + record chunk { %stream: label, items: string } + record future-value { %future: label, value: string } + + variant event { + // `by` is the caller. + call(call), + // The callee blocked before returning, so the caller got a subtask. + started(task-id), + %return(return-value), + cancel(task-id), + cancelled(task-id), + // The host wrote `items`, which the guest read. + write(chunk), + // The host read `items`, which the guest wrote. + read(chunk), + // A future's value, and who wrote it (`by`). If the host wrote it, the line is recorded when + // the host writes, before the guest can read it; the guest may drop its end instead, which + // isn't recorded. If the guest wrote it, the line is recorded when the host reads it. + value(future-value), + // `by` is who dropped it. + drop(label), + } + + // `at` is the step of a host line the guest observes outside its calls. + record line { by: party, at: option, event: event } + + // What a line is about: a call, or a stream or future. + variant line-key { task(task-id), handle(label) } +} + interface %record { - record-args: func(method: option, args: list, is-export: bool); - record-ret: func(method: option, ret: option, is-export: bool); + use types.{line}; + // Records a line, in the order the proxy observes it. + %record: func(line: line); + // Reports a panic of a generated component, which can't print its own, and traps. + fail: func(message: string); } interface replay { - replay-export: func() -> option>>; - assert-export-ret: func(method: option, ret: option); - replay-import: func(method: option, args: option>) -> option; + use types.{line, line-key, label}; + // The next line `start` performs, in trace order: an export call, or the host's drop of a + // resource of the guest. `none` once there are none. + next-host-line: func() -> option; + // The next host line about `key`, e.g. the `return` of a STARTED call, which its mock waits + // for. + take-host-line: func(key: line-key) -> option; + // Pairs the trace's label of a stream or future replay makes with replay's, of the end it + // keeps; the trace's first appears in a host line, which `check` doesn't pair. + pair: func(trace: label, replay: label); + // Checks that replay reached every line, once it is done; traps on the first it didn't. + finish: func(); + // Reports a panic of a generated component, e.g. a divergence a job found, which it can't + // print itself, and traps. + fail: func(message: string); + // Checks a guest line against the trace, and returns the host lines that answer it at once, + // e.g. the `return` of an import call. Traps on a mismatch, with the expected and the actual + // line, except for a guest's line about a stream or future that a cut trace lacks. + check: func(line: line) -> list; + // Whether the trace is cut short: an export call has no answer, e.g. the run trapped during + // it. Its last lines that the host records once it sees a copy complete may be missing, e.g. + // the guest's drop of a stream, or the value of a future it wrote. + trace-cut: func() -> bool; } +// Async, so that it can await the main component's async exports, and run the jobs of +// `proxy:conversion/executor`. interface start-replay { - start: func(); + start: async func(); } world host { @@ -20,7 +91,12 @@ world host { import replay; } +// The types alone, which `crates/trace` generates its Rust types from. +world trace { + import types; +} + world guest { export %record; export replay; -} \ No newline at end of file +} diff --git a/components/recorder/Cargo.toml b/components/recorder/Cargo.toml index 2eb201e..fc0c50a 100644 --- a/components/recorder/Cargo.toml +++ b/components/recorder/Cargo.toml @@ -8,7 +8,6 @@ crate-type = ["cdylib"] [dependencies] trace = { path = "../../crates/trace" } -serde_json.workspace = true wit-bindgen.workspace = true libc-alloc = { path = "../../crates/libc-alloc", optional = true } diff --git a/components/recorder/src/lib.rs b/components/recorder/src/lib.rs index 7b57a36..a471482 100644 --- a/components/recorder/src/lib.rs +++ b/components/recorder/src/lib.rs @@ -7,68 +7,90 @@ mod bindings { wit_bindgen::generate!({ path: "../../assets/recorder.wit", world: "guest", + // The trace's types, which `crates/trace` generates from the same WIT. + with: { "proxy:recorder/types@0.1.0": trace }, }); } -use trace::Logger; +use std::cell::RefCell; +use trace::{Label, Labels, Line, LineKey, Replay, Trace}; + +thread_local! { + // The bindings of the labels recorded so far, which each line's are named apart by. + static LABELS: RefCell = RefCell::new(Labels::default()); + // Whether to print the lines as text, with `PROXY_TRACE=text`, e.g. to read a fuzz run, + // rather than in their stored form, which replay reads. + static AS_TEXT: bool = std::env::var("PROXY_TRACE").is_ok_and(|format| format == "text"); +} + struct Component; impl bindings::exports::proxy::recorder::record::Guest for Component { - fn record_args(method: Option, args: Vec, is_export: bool) { - let mut logger = Logger::new(); - let call = logger.record_args(method, args, is_export); - let json = serde_json::to_string(&call).unwrap(); - println!("{json}"); + // Prints the line, so that stdout is the trace, also of a run that traps. + fn record(line: Line) { + let line = LABELS.with_borrow_mut(|labels| labels.label(line)); + if AS_TEXT.with(|as_text| *as_text) { + println!("{line}"); + } else { + println!("{}", line.to_json()); + } } - fn record_ret(method: Option, ret: Option, is_export: bool) { - let mut logger = Logger::new(); - let call = logger.record_ret(method, ret, is_export); - let json = serde_json::to_string(&call).unwrap(); - println!("{json}"); + fn fail(message: String) { + panic!("{message}"); } } -use std::cell::RefCell; thread_local! { - static TRACE: RefCell> = RefCell::new(None); + // The trace being replayed, read from stdin on first use. + static REPLAY: RefCell> = const { RefCell::new(None) }; } -fn load_trace() { - let load = TRACE.with_borrow(|v| v.is_none()); - if load { - TRACE.with_borrow_mut(|v| { +fn with_replay(f: impl FnOnce(&mut Replay) -> R) -> R { + REPLAY.with_borrow_mut(|replay| { + let replay = replay.get_or_insert_with(|| { use std::io::Read; let mut input = String::new(); std::io::stdin().read_to_string(&mut input).unwrap(); - let mut logger = Logger::new(); - logger.load_trace(&input); - *v = Some(logger); + Replay::new(Trace::parse(&input)) }); - } + f(replay) + }) } impl bindings::exports::proxy::recorder::replay::Guest for Component { - fn replay_export() -> Option<(String, Vec)> { - load_trace(); - TRACE.with_borrow_mut(|v| v.as_mut().unwrap().replay_export()) + fn next_host_line() -> Option { + let line = with_replay(|replay| replay.next_host_line())?; + println!("{line}"); + Some(line) } - fn assert_export_ret(assert_method: Option, assert_ret: Option) { - TRACE.with_borrow_mut(|v| { - v.as_mut() - .unwrap() - .assert_export_ret(assert_method, assert_ret) - }); + fn take_host_line(key: LineKey) -> Option { + let line = with_replay(|replay| replay.take_host_line(&key))?; + println!("{line}"); + Some(line) + } + fn pair(trace: Label, replay: Label) { + with_replay(|replay_| replay_.pair(&trace, &replay)); + } + fn fail(message: String) { + panic!("{message}"); + } + fn finish() { + with_replay(|replay| replay.finish()); + } + fn trace_cut() -> bool { + with_replay(|replay| replay.is_cut()) } - fn replay_import( - assert_method: Option, - assert_args: Option>, - ) -> Option { - TRACE.with_borrow_mut(|v| { - let (_, ret) = v - .as_mut() - .unwrap() - .replay_import(assert_method, assert_args, true); - ret - }) + fn check(line: Line) -> Vec { + println!("{line}"); + // A replayed `wasi:cli/exit` ends the replay with the guest's status. + if let Some(status) = line.exit_status() { + with_replay(|replay| replay.check(&line)); + std::process::exit(status); + } + let answers = with_replay(|replay| replay.check(&line)); + for answer in &answers { + println!("{answer}"); + } + answers } } bindings::export!(Component with_types_in bindings); diff --git a/crates/trace/Cargo.toml b/crates/trace/Cargo.toml index fb4614b..62cb5b9 100644 --- a/crates/trace/Cargo.toml +++ b/crates/trace/Cargo.toml @@ -6,3 +6,5 @@ edition = "2024" [dependencies] serde.workspace = true serde_json.workspace = true +# Generates the types from `assets/recorder.wit`. +wit-bindgen.workspace = true diff --git a/crates/trace/src/lib.rs b/crates/trace/src/lib.rs index 96dcb57..7ecdfd4 100644 --- a/crates/trace/src/lib.rs +++ b/crates/trace/src/lib.rs @@ -1,173 +1,689 @@ -use std::collections::VecDeque; - -use serde::{Deserialize, Serialize}; - -#[derive(Serialize, Deserialize, Debug, Clone)] -pub enum FuncCall { - ExportArgs { - method: String, - args: Vec, - }, - ExportRet { - method: Option, - ret: Option, - }, - ImportArgs { - method: Option, - args: Vec, - }, - ImportRet { - method: Option, - ret: Option, - }, +//! The trace of a run, which record mode writes and replay mode reads: the events the proxy +//! observed, as `docs/p3-record-replay.md` (Trace format) defines them. A trace is stored as one +//! JSON line per `Line`; other lines in the text, e.g. the guest's own output, are ignored. +//! +//! The types are generated from `proxy:recorder/types` (`assets/recorder.wit`), with serde, so +//! that the WIT defines them once: the bundled recorder maps the interface to this crate, with +//! wit-bindgen's `with` option, and a host converts its own bindings' types with `convert`, as +//! they serialize alike. Values are WAVE text, e.g. `ok(#response-3)`, whose type comes from +//! where they appear. +use std::fmt; + +use serde::{Deserialize, Serialize, de::DeserializeOwned}; + +mod renaming; +pub use renaming::Labels; +use renaming::Renaming; + +mod bindings { + wit_bindgen::generate!({ + path: "../../assets/recorder.wit", + world: "trace", + additional_derives: [serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash], + }); } +pub use bindings::proxy::recorder::types::*; -/// The method of a resource drop, whose only argument is the handle, e.g. `#x-1`. -pub const DROP: &str = "drop"; +/// The lines the proxy observed, in the order it observed them. +#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq)] +#[serde(transparent)] +pub struct Trace(pub Vec); -pub struct Logger(pub VecDeque); +/// Converts between this crate's types and another binding's types of `proxy:recorder/types`, +/// e.g. those of wasmtime's `bindgen!`, which derive serde with `additional_derives`. +pub fn convert(value: T) -> U { + serde_json::from_value(serde_json::to_value(value).unwrap()).unwrap() +} -impl Logger { - pub fn new() -> Self { - Self(VecDeque::new()) +impl Event { + /// The stream, future or resource the event is about, for the events of a handle. + pub fn handle(&self) -> Option<&Label> { + match self { + Event::Write(chunk) | Event::Read(chunk) => Some(&chunk.stream), + Event::Value(value) => Some(&value.future), + Event::Drop(label) => Some(label), + _ => None, + } } - pub fn load_trace(&mut self, text: &str) { - self.0.clear(); - for line in text.lines() { - match serde_json::from_str::(line) { - Ok(item) => self.0.push_back(item), - // Ignore non-JSON lines. - Err(_) => continue, + /// The task the event is about, for the events of a call. + pub fn task(&self) -> Option { + match self { + Event::Call(Call { id, .. }) + | Event::Started(id) + | Event::Return(ReturnValue { id, .. }) + | Event::Cancel(id) + | Event::Cancelled(id) => Some(*id), + _ => None, + } + } +} + +impl Line { + /// Whether the line is the host's answer to a guest call, which the host gives while handling + /// the call: a `Started`, or the `Return` or `Cancelled` of a call that didn't start, of an + /// import. Replay performs it as part of answering the call, rather than on its own. + pub fn answers_guest(&self) -> bool { + self.by == Party::Host + && self.at.is_none() + && matches!( + self.event, + Event::Started(_) | Event::Return(_) | Event::Cancelled(_) + ) + && self + .event + .task() + .is_some_and(|id| id.table == Table::Imports) + } + /// Whether the line is one the host performs from the exports component, which replay's + /// `start` performs: an export call, or the host's drop of a resource of the guest. The + /// host's drops of its own resources, in the imports table, the guest doesn't see; replay has + /// nothing to do for them. + pub fn is_start_line(&self) -> bool { + self.by == Party::Host + && match &self.event { + Event::Call(_) => true, + // Not a stream's or future's end, which the job that keeps it drops. + Event::Drop(label) => { + label.table == Table::Exports + && !label.name.starts_with("#stream-") + && !label.name.starts_with("#future-") + } + _ => false, } + } + /// Whether the line is about `key`: a line of the task, or of the stream or future. + pub fn is_about(&self, key: &LineKey) -> bool { + match (key, &self.event) { + (LineKey::Task(task), event) => event.task() == Some(*task), + (LineKey::Handle(label), event) => event.handle() == Some(label), } } - pub fn dump_trace(&self) -> String { - self.0 - .iter() - .map(|call| serde_json::to_string(call).unwrap()) - .collect::>() - .join("\n") - } - pub fn record_args( - &mut self, - method: Option, - args: Vec, - is_export: bool, - ) -> FuncCall { - let call = if is_export { - FuncCall::ExportArgs { - method: method.unwrap(), + /// The exit status of a call to `wasi:cli/exit`, which ends the run. + pub fn exit_status(&self) -> Option { + match &self.event { + Event::Call(Call { func, args, .. }) if func.starts_with("wasi:cli/exit") => { + let is_err = args.first().is_some_and(|arg| arg.starts_with("err")); + Some(if is_err { 1 } else { 0 }) + } + _ => None, + } + } + /// The stored form of the line, one line of text. + pub fn to_json(&self) -> String { + serde_json::to_string(self).unwrap() + } +} + +impl Trace { + /// Reads a trace from its stored form, ignoring the lines that aren't trace lines, e.g. the + /// guest's output: those that aren't JSON objects with a `by` and an `event`. + /// + /// Panics on a trace line that doesn't parse, e.g. of another version of the format, and on + /// text without trace lines, e.g. a trace printed as text. + pub fn parse(text: &str) -> Self { + let lines = text.lines().filter_map(|text| { + let value: serde_json::Value = serde_json::from_str(text).ok()?; + let object = value.as_object()?; + if !object.contains_key("by") || !object.contains_key("event") { + return None; + } + match serde_json::from_value(value) { + Ok(line) => Some(line), + Err(err) => panic!("invalid trace line `{text}`: {err}"), + } + }); + let trace = Trace(lines.collect()); + assert!( + !trace.0.is_empty() || text.trim().is_empty(), + "no trace lines: was it printed with `PROXY_TRACE=text`?" + ); + trace + } +} + +impl fmt::Display for TaskId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let prefix = match self.table { + Table::Exports => 't', + Table::Imports => 'c', + }; + write!(f, "{prefix}{}", self.id) + } +} + +impl fmt::Display for Label { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self.table { + Table::Imports => write!(f, "{}", self.name), + Table::Exports => write!(f, "{} (exports)", self.name), + } + } +} + +/// An event as text, after the party that did it (see `Line`'s): a call's, by its task, e.g. +/// `t1.call run(#request-1)` or `c2.return ok`, and a copy's by the direction its data moves +/// for that party, `<--` into the stream or future, `-->` out of it, e.g. the host's write +/// `#stream-6 <-- [52]`, its read `#stream-3 --> [97]`, or a writer's `#future-4 <-- ok`. +impl fmt::Display for Event { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Event::Call(Call { + id, + is_async, + func, args, + }) => { + let is_async = if *is_async { "async " } else { "" }; + write!(f, "{id}.call {is_async}{func}({})", args.join(", ")) + } + Event::Started(id) => write!(f, "{id}.started"), + Event::Return(ReturnValue { id, value }) => { + write!(f, "{id}.return {}", value.as_deref().unwrap_or("()")) } - } else { - FuncCall::ImportArgs { method, args } + Event::Cancel(id) => write!(f, "{id}.cancel"), + Event::Cancelled(id) => write!(f, "{id}.cancelled"), + Event::Write(Chunk { stream, items }) => write!(f, "{stream} <-- {items}"), + Event::Read(Chunk { stream, items }) => write!(f, "{stream} --> {items}"), + Event::Value(FutureValue { future, value }) => write!(f, "{future} <-- {value}"), + Event::Drop(handle) => write!(f, "drop {handle}"), + } + } +} + +impl fmt::Display for Line { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let by = match self.by { + Party::Host => "host", + Party::Guest => "guest", }; - self.0.push_back(call.clone()); - call - } - pub fn record_ret( - &mut self, - method: Option, - ret: Option, - is_export: bool, - ) -> FuncCall { - let call = if is_export { - FuncCall::ExportRet { method, ret } - } else { - FuncCall::ImportRet { method, ret } + match self.at { + Some(at) => write!(f, "{by} @{at}: {}", self.event), + None => write!(f, "{by}: {}", self.event), + } + } +} + +/// A trace being replayed: which of its lines replay has performed or checked, and how its +/// labels pair with replay's (see `renaming`). The lines `Replay` returns have replay's labels, +/// where it has paired them, and those it takes, the trace's. +pub struct Replay { + lines: Vec, + done: Vec, + renaming: Renaming, + /// Whether the trace is cut short (see `is_cut`). + cut: bool, +} + +impl Replay { + pub fn new(trace: Trace) -> Self { + let done = vec![false; trace.0.len()]; + let renaming = Renaming::new(&trace.0); + let answered = |id: TaskId| { + trace.0.iter().any(|line| { + matches!(&line.event, Event::Return(ret) if ret.id == id) + || matches!(&line.event, Event::Cancelled(task) if *task == id) + }) }; - self.0.push_back(call.clone()); - call - } - pub fn replay_export(&mut self) -> Option<(String, Vec)> { - let call = self.0.pop_front()?; - println!("export call: {}", call.to_string()); - let FuncCall::ExportArgs { method, args } = call else { - panic!("the trace has {call:?}, which the guest didn't call") + let cut = trace.0.iter().any(|line| match &line.event { + Event::Call(call) => call.id.table == Table::Exports && !answered(call.id), + _ => false, + }); + Replay { + lines: trace.0, + done, + renaming, + cut, + } + } + /// Whether the trace is cut short: an export call has no answer, e.g. the run trapped during + /// it. The host records some copies once it sees them complete, after the guest went on, so + /// a cut trace may lack the last of those: the guest's drops of streams and futures, and the + /// values of the futures it wrote. `check` accepts a guest's line about one that it lacks. + pub fn is_cut(&self) -> bool { + self.cut + } + fn pending(&self) -> impl Iterator { + self.lines + .iter() + .enumerate() + .filter(|(idx, _)| !self.done[*idx]) + } + /// The next line replay's `start` performs, in trace order: an export call, or the host's + /// drop of a resource of the guest (see `Line::is_start_line`). `None` once there are none. + /// The lines before it that are still pending may happen while `start` waits for its step; + /// `finish` reports those that never do. + pub fn next_host_line(&mut self) -> Option { + self.take(|line| line.is_start_line()) + } + /// The next host line about `key`, in replay's labels, e.g. the `Return` of a STARTED call, + /// which the import's mock waits for, or a copy of a stream, which a job performs. + pub fn take_host_line(&mut self, key: &LineKey) -> Option { + let key = match key { + LineKey::Handle(label) => LineKey::Handle(self.renaming.to_trace(label)?), + key => key.clone(), }; - Some((method, args)) - } - pub fn assert_export_ret(&mut self, assert_method: Option, assert_ret: Option) { - if let Some(FuncCall::ExportRet { .. }) = self.0.front() { - let call = self.0.pop_front().unwrap(); - println!("export ret: {}", call.to_string()); - let FuncCall::ExportRet { method, ret } = call else { - panic!() + self.take(|line| line.by == Party::Host && line.is_about(&key)) + } + fn take(&mut self, matches: impl Fn(&Line) -> bool) -> Option { + let idx = self.pending().find(|(_, line)| matches(line))?.0; + Some(self.hand(idx)) + } + /// Hands line `idx` to replay, in its labels. + fn hand(&mut self, idx: usize) -> Line { + self.done[idx] = true; + self.renaming.to_replay(&self.lines[idx]) + } + /// Pairs the trace's label of a stream or future replay makes, from the value of the host + /// line that hands it on, with its own, of the end it keeps, e.g. the writer of a stream a + /// mock returns: the trace's label first appears in a host line, which `check` doesn't pair. + /// + /// Panics if either is paired with another, or no line hands it on. + pub fn pair(&mut self, trace: &Label, replay: &Label) { + assert!( + self.renaming.pair_next(trace, replay), + "replay's {replay} can't be the one the trace's {trace} hands on" + ); + } + /// Checks that replay reached every line, once it is done: all but the host's drops of its + /// own resources, which replay has nothing to do for. + /// + /// Panics on the first line it didn't reach. + pub fn finish(&self) { + let missed = self.pending().find(|(_, line)| { + line.by == Party::Guest || line.is_start_line() || !matches!(line.event, Event::Drop(_)) + }); + if let Some((_, line)) = missed { + panic!("the trace has `{line}`, which replay didn't reach"); + } + } + /// Checks a guest line against the trace, up to the pairing of labels, and returns the host + /// lines that answer it at once, e.g. the `Return` of an import call. A guest line is matched + /// by its task, for the lines of a call, and by its handle, for those of a stream, a future + /// or a resource. + /// + /// Panics on a mismatch, with the expected and the actual line. + pub fn check(&mut self, actual: &Line) -> Vec { + assert_eq!(actual.by, Party::Guest, "`{actual}` isn't a guest line"); + let (task, handle) = (actual.event.task(), actual.event.handle()); + // The trace's label of the handle, if replay's is paired, and otherwise a binding of + // replay's number, e.g. a mock's, which carries the trace's. A pair whose trace label + // ended, in a line replay has done, is of another handle, which had the number before. + let paired = handle.and_then(|label| { + let trace = self.renaming.to_trace(label)?; + let live = self.renaming.end_of(&trace).is_none_or(|end| !self.done[end]); + live.then_some(trace) + }); + let found = self.pending().find(|(_, line)| match (task, handle) { + (Some(task), _) => line.event.task() == Some(task), + (None, Some(replay)) => { + line.by == Party::Guest + && line.event.handle().is_some_and(|trace| match &paired { + Some(paired) => trace == paired, + None => Renaming::binds(trace, replay), + }) + } + (None, None) => line.by == Party::Guest, + }); + let Some((idx, expected)) = found else { + let late = |label: &Label| { + ["#stream-", "#future-"].iter().any(|kind| label.name.starts_with(kind)) }; - if let (Some(method), Some(assert_method)) = (method, assert_method) { - assert_eq!(method, assert_method); + if self.cut && handle.is_some_and(late) { + return Vec::new(); } - assert_eq!(ret, assert_ret); + panic!("the guest did `{actual}`, which the trace doesn't have"); + }; + let expected = expected.clone(); + if expected.by != Party::Guest || !self.renaming.matches(idx, &expected, actual) { + panic!("the trace has `{expected}`, but the guest did `{actual}`"); } - } - pub fn replay_import( - &mut self, - assert_method: Option, - assert_args: Option>, - from_guest: bool, - ) -> (bool, Option) { - let mut exit_called = false; - let called = assert_method.clone(); - let mut call = self.0.pop_front().unwrap(); - if let FuncCall::ImportArgs { method, args } = &call { - if let (Some(method), Some(assert_method)) = (method, assert_method) { - assert_eq!(method, &assert_method); + self.done[idx] = true; + let Some(task) = task else { + return Vec::new(); + }; + // The task's next lines, while they answer the call. + let mut answers = Vec::new(); + for idx in idx + 1..self.lines.len() { + let line = &self.lines[idx]; + if self.done[idx] || line.event.task() != Some(task) { + continue; } - if let Some(assert_args) = &assert_args { - assert_eq!(args, assert_args); + if !line.answers_guest() { + break; } - println!("import call: {}", call.to_string()); - if method - .as_ref() - .is_some_and(|m| m.starts_with("wasi:cli/exit")) - { - if from_guest { - let code = if assert_args - .is_some_and(|args| args.get(0).is_some_and(|arg| arg.starts_with("err"))) - { - 1 - } else { - 0 - }; - std::process::exit(code); - } else { - exit_called = true; - return (exit_called, Some("Something that can crash".to_string())); - } + let started = matches!(line.event, Event::Started(_)); + answers.push(self.hand(idx)); + if !started { + break; } - call = self.0.pop_front().unwrap(); } - println!("import ret: {}", call.to_string()); - let FuncCall::ImportRet { ret, .. } = call else { - panic!( - "the guest called {}, but the trace has {call:?}", - called.as_deref().unwrap_or("an import"), - ) - }; - (exit_called, ret) + answers } } -impl FuncCall { - pub fn to_string(&self) -> String { - match self { - // As component-ui writes drops, e.g. `drop #x-1` - FuncCall::ExportArgs { method, args } - | FuncCall::ImportArgs { - method: Some(method), - args, - } if method == DROP => { - format!("drop {}", args.join(", ")) - } - FuncCall::ExportArgs { method, args } => format!("{method}({})", args.join(", ")), - FuncCall::ImportArgs { method, args } => format!( - "{}({})", - method.as_deref().unwrap_or(""), - args.join(", ") - ), - FuncCall::ExportRet { ret, .. } | FuncCall::ImportRet { ret, .. } => { - ret.as_deref().unwrap_or("()").to_owned() - } +#[cfg(test)] +mod tests { + use super::*; + + fn task(table: Table, id: u32) -> TaskId { + TaskId { table, id } + } + fn line(by: Party, event: Event) -> Line { + Line { + by, + at: None, + event, } } + fn call(id: TaskId, func: &str, args: &[&str]) -> Event { + Event::Call(Call { + id, + is_async: false, + func: func.to_string(), + args: args.iter().map(|arg| arg.to_string()).collect(), + }) + } + fn ret(id: TaskId, value: Option<&str>) -> Event { + Event::Return(ReturnValue { + id, + value: value.map(str::to_string), + }) + } + /// The trace a recorder writes of `lines`, with their labels' bindings named apart. + fn recorded(lines: &[Line]) -> Trace { + let mut labels = Labels::default(); + Trace(lines.iter().map(|line| labels.label(line.clone())).collect()) + } + + /// An export call `t1`, during which the guest calls an import `c1`, then the host drops a + /// resource of the guest. + fn p2_trace() -> Trace { + let (t1, c1) = (task(Table::Exports, 1), task(Table::Imports, 1)); + Trace(vec![ + line(Party::Host, call(t1, "run", &["#request-1"])), + line(Party::Guest, call(c1, "get", &["1"])), + line(Party::Host, ret(c1, Some("\"a\""))), + line(Party::Guest, ret(t1, None)), + line( + Party::Host, + Event::Drop(Label { + table: Table::Exports, + name: "#session-1".to_string(), + }), + ), + ]) + } + + #[test] + fn stored_form() { + let trace = p2_trace(); + let text: Vec<_> = trace.0.iter().map(Line::to_json).collect(); + assert_eq!( + text[0], + r##"{"by":"Host","at":null,"event":{"Call":{"id":{"table":"Exports","id":1},"is_async":false,"func":"run","args":["#request-1"]}}}"## + ); + let text = format!("guest output\n{}\n", text.join("\n")); + assert_eq!(Trace::parse(&text), trace); + let step = Line { + at: Some(4), + ..trace.0[2].clone() + }; + assert_eq!(Trace::parse(&step.to_json()).0, vec![step]); + } + + #[test] + #[should_panic(expected = "invalid trace line")] + fn stored_form_mismatch() { + Trace::parse(r##"{"by":"Guest","at":null,"event":{"Drop":{"handle":"#x-1"}}}"##); + } + + #[test] + #[should_panic(expected = "no trace lines")] + fn stored_form_as_text() { + let text: Vec<_> = p2_trace().0.iter().map(Line::to_string).collect(); + Trace::parse(&text.join("\n")); + } + + #[test] + fn display() { + let lines: Vec<_> = p2_trace().0.iter().map(Line::to_string).collect(); + assert_eq!( + lines, + [ + "host: t1.call run(#request-1)", + "guest: c1.call get(1)", + "host: c1.return \"a\"", + "guest: t1.return ()", + "host: drop #session-1 (exports)", + ] + ); + let label = |name: &str| Label { + table: Table::Imports, + name: name.to_string(), + }; + let chunk = |name: &str, items: &str| Chunk { + stream: label(name), + items: items.to_string(), + }; + let copies = [ + line(Party::Host, Event::Started(task(Table::Imports, 2))), + line(Party::Host, Event::Write(chunk("#stream-6", "[52]"))), + line(Party::Host, Event::Read(chunk("#stream-3", "[97]"))), + line( + Party::Guest, + Event::Value(FutureValue { + future: label("#future-4"), + value: "ok".to_string(), + }), + ), + ]; + let lines: Vec<_> = copies.iter().map(Line::to_string).collect(); + assert_eq!( + lines, + [ + "host: c2.started", + "host: #stream-6 <-- [52]", + "host: #stream-3 --> [97]", + "guest: #future-4 <-- ok", + ] + ); + } + + #[test] + fn replay() { + let trace = p2_trace(); + let mut replay = Replay::new(trace.clone()); + assert_eq!(replay.next_host_line(), Some(trace.0[0].clone())); + assert_eq!(replay.check(&trace.0[1]), vec![trace.0[2].clone()]); + assert_eq!(replay.check(&trace.0[3]), vec![]); + assert_eq!(replay.next_host_line(), Some(trace.0[4].clone())); + assert_eq!(replay.next_host_line(), None); + } + + #[test] + #[should_panic( + expected = "the trace has `guest: c1.call get(1)`, but the guest did `guest: c1.call get(2)`" + )] + fn replay_mismatch() { + let mut replay = Replay::new(p2_trace()); + replay.next_host_line(); + replay.check(&line( + Party::Guest, + call(task(Table::Imports, 1), "get", &["2"]), + )); + } + + #[test] + #[should_panic(expected = "the trace has `guest: c1.call get(1)`, which replay didn't reach")] + fn replay_missing_guest_line() { + let mut replay = Replay::new(p2_trace()); + // The guest's lines may still come while `start` waits for the next line's step. + assert_eq!(replay.next_host_line(), Some(p2_trace().0[0].clone())); + assert_eq!(replay.next_host_line(), Some(p2_trace().0[4].clone())); + assert_eq!(replay.next_host_line(), None); + replay.finish(); + } + + #[test] + fn host_drops_of_mocks() { + let mock = Label { + table: Table::Imports, + name: "#counter-1".to_string(), + }; + let trace = Trace(vec![line(Party::Host, Event::Drop(mock))]); + let mut replay = Replay::new(trace); + // Neither `start`'s, nor needed. + assert_eq!(replay.next_host_line(), None); + replay.finish(); + } + + #[test] + fn replay_renames_labels() { + let session = |name: &str| Label { + table: Table::Exports, + name: name.to_string(), + }; + let (t1, t2) = (task(Table::Exports, 1), task(Table::Exports, 2)); + let trace = recorded(&[ + line(Party::Host, call(t1, "new", &[])), + line(Party::Guest, ret(t1, Some("#session-3"))), + line(Party::Host, Event::Drop(session("#session-3"))), + line(Party::Host, call(t2, "new", &[])), + line(Party::Guest, ret(t2, Some("#session-3"))), + ]); + assert_eq!(trace.0[4].to_string(), "guest: t2.return #session-3-2"); + let mut replay = Replay::new(trace); + replay.next_host_line(); + // Replay's table numbers the session differently. + replay.check(&line(Party::Guest, ret(t1, Some("#session-1")))); + let drop = line(Party::Host, Event::Drop(session("#session-1"))); + assert_eq!(replay.next_host_line(), Some(drop)); + // Once dropped, both labels may name another session. + replay.next_host_line(); + replay.check(&line(Party::Guest, ret(t2, Some("#session-2")))); + replay.finish(); + } + + #[test] + fn replay_ends_labels_handed_on() { + let (t1, t2, t3) = ( + task(Table::Exports, 1), + task(Table::Exports, 2), + task(Table::Exports, 3), + ); + let trace = recorded(&[ + line(Party::Host, call(t1, "new", &[])), + line(Party::Guest, ret(t1, Some("#session-3"))), + // The host hands the session back, which frees its number. + line(Party::Host, call(t2, "close", &["#session-3"])), + line(Party::Guest, ret(t2, None)), + line(Party::Host, call(t3, "new", &[])), + line(Party::Guest, ret(t3, Some("#session-3"))), + ]); + let mut replay = Replay::new(trace); + replay.next_host_line(); + replay.check(&line(Party::Guest, ret(t1, Some("#session-1")))); + let close = replay.next_host_line().unwrap(); + assert_eq!(close.event, call(t2, "close", &["#session-1"])); + replay.check(&line(Party::Guest, ret(t2, None))); + replay.next_host_line(); + // The recording reused the number, and replay didn't, or the other way round. + replay.check(&line(Party::Guest, ret(t3, Some("#session-2")))); + replay.finish(); + } + + #[test] + fn replay_finds_reused_labels() { + // The guest hands the counter on to the host, which returns another with its number, as + // record mode's trace has it. + let (c1, c2) = (task(Table::Imports, 1), task(Table::Imports, 2)); + let counter = Label { + table: Table::Imports, + name: "#counter-2".to_string(), + }; + let lines = [ + line(Party::Guest, call(c1, "new", &[])), + line(Party::Host, ret(c1, Some("#counter-2"))), + line(Party::Guest, call(c2, "merge", &["#counter-2"])), + line(Party::Host, ret(c2, Some("#counter-2"))), + line(Party::Guest, Event::Drop(counter.clone())), + ]; + let trace = recorded(&lines); + assert_eq!(trace.0[4].to_string(), "guest: drop #counter-2-3"); + // Replay's own labels, which it reuses too. + let mut replay = Replay::new(trace); + assert_eq!(replay.check(&lines[0]), vec![lines[1].clone()]); + assert_eq!(replay.check(&lines[2]), vec![lines[3].clone()]); + replay.check(&lines[4]); + replay.finish(); + } + + #[test] + #[should_panic(expected = "the trace has `guest: t2.return #session-3`")] + fn replay_checks_labels() { + let (t1, t2) = (task(Table::Exports, 1), task(Table::Exports, 2)); + let trace = Trace(vec![ + line(Party::Host, call(t1, "new", &[])), + line(Party::Guest, ret(t1, Some("#session-3"))), + line(Party::Host, call(t2, "get", &[])), + line(Party::Guest, ret(t2, Some("#session-3"))), + ]); + let mut replay = Replay::new(trace); + replay.next_host_line(); + replay.check(&line(Party::Guest, ret(t1, Some("#session-1")))); + replay.next_host_line(); + // The trace's `#session-3` is replay's `#session-1`, not another session. + replay.check(&line(Party::Guest, ret(t2, Some("#session-2")))); + } + + #[test] + fn cut_traces_lack_the_guests_last_copies() { + let t1 = task(Table::Exports, 1); + let stream = Label { + table: Table::Imports, + name: "#stream-3".to_string(), + }; + let drop = line(Party::Guest, Event::Drop(stream)); + // The run trapped during `t1`, before the host saw the guest's drop of its stream. + let mut replay = Replay::new(Trace(vec![line(Party::Host, call(t1, "run", &[]))])); + assert!(replay.is_cut()); + replay.next_host_line(); + assert_eq!(replay.check(&drop), vec![]); + // A whole trace has every line. + let whole = Trace(vec![ + line(Party::Host, call(t1, "run", &[])), + line(Party::Guest, ret(t1, None)), + ]); + let mut replay = Replay::new(whole); + assert!(!replay.is_cut()); + replay.next_host_line(); + let check = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| replay.check(&drop))); + assert!(check.is_err()); + } + + #[test] + fn started_then_return() { + let c1 = task(Table::Imports, 1); + let late = Line { + at: Some(3), + ..line(Party::Host, ret(c1, Some("5"))) + }; + let async_call = Event::Call(Call { + id: c1, + is_async: true, + func: "wait".to_string(), + args: vec![], + }); + let trace = Trace(vec![ + line(Party::Guest, async_call), + line(Party::Host, Event::Started(c1)), + late.clone(), + ]); + let mut replay = Replay::new(trace.clone()); + // The `Return` comes at its step, not in answer to the call, and the mock takes it. + assert_eq!(replay.check(&trace.0[0]), vec![trace.0[1].clone()]); + assert_eq!(replay.next_host_line(), None); + assert_eq!(replay.take_host_line(&LineKey::Task(c1)), Some(late)); + replay.finish(); + } } diff --git a/crates/trace/src/renaming.rs b/crates/trace/src/renaming.rs new file mode 100644 index 0000000..0a7a026 --- /dev/null +++ b/crates/trace/src/renaming.rs @@ -0,0 +1,569 @@ +//! The labels of a trace, and their pairing with replay's. A label is a handle's number in a +//! generated component's table, which names the handle until it leaves the table: at a `Drop`, a +//! future's `Value`, or when the host hands it on, owned, to the guest, in an export call's +//! arguments or an import's result. After that, the number may name another handle, as a +//! variable may be bound again. So the recorder names each binding apart, as it records the lines +//! (see `Labels`): the first keeps the number, e.g. `#stream-7`, and the next are `#stream-7-2`, +//! `#stream-7-3`, and so on. A trace's label then names one handle, from its first line to the +//! line that ends it. +//! +//! The tables of the recording and of replay differ: besides the values the trace labels, they +//! hold waitable sets and subtasks, which each mode makes its own. So replay compares lines up to +//! renaming: it pairs each label of the trace with the label replay sees for the same value, one +//! to one, per table, from the first line that has both, and requires every later line to agree. +//! Replay's labels are its own handle numbers, which it reuses too: a replay label can be paired +//! again once the trace's label it was paired with has ended. The rule is the same for every +//! label, whatever it names. +use std::collections::{HashMap, HashSet}; + +use crate::{Event, Label, Line, Party, Table}; + +/// A trace label's handle in a table, e.g. `(imports, "stream-7")`. +type Handle = (Table, String); + +/// Names the bindings of labels apart, as a recorder records the lines (see the module's doc). +/// +/// A line can come after the guest acted on a handle it brings in: the host records a copy once +/// its job sees it complete, e.g. a future's value with a mock in it, which the guest may have +/// taken and dropped by then. Lines about different handles may come in any order. So a `Drop` +/// or `Value` of a label that no line has mentioned yet ends the binding of the line that brings +/// it in, which comes later. A line that brings a handle in and hands it on, e.g. a mock an +/// import returns, ends it at its first mention. +#[derive(Default)] +pub struct Labels { + /// The number of bindings of each handle that have ended. + ended: HashMap, + /// The handles the lines so far mention. + seen: HashSet, + /// The handles a line ended before any line brought them in. + early: HashSet, +} + +impl Labels { + /// `line`, with each label renamed to its binding's, and the bindings it ends ended. + pub fn label(&mut self, line: Line) -> Line { + let renamed = map_labels(&line, |table, handle| { + let ended = *self.ended.get(&(table, handle.to_string()))?; + Some(format!("{handle}-{}", ended + 1)) + }); + let mentioned = handles_of(&line).0; + // The line that brings in a handle an earlier line ended, whose binding then ends. + let late: Vec<_> = mentioned + .iter() + .filter(|handle| self.early.remove(*handle)) + .cloned() + .collect(); + for handle in &late { + *self.ended.entry(handle.clone()).or_insert(0) += 1; + } + for handle in ends_of(&line) { + if late.contains(&handle) { + continue; + } + let brings_in = matches!(line.event, Event::Call(_) | Event::Return(_)); + if brings_in || self.seen.contains(&handle) { + *self.ended.entry(handle).or_insert(0) += 1; + } else { + self.early.insert(handle); + } + } + self.seen.extend(mentioned); + renamed + } +} + +/// The handle a binding's label names, and which binding of its number it is, as `Labels` names +/// them, e.g. `("stream-7", 2)` for `stream-7-2`, and `("stream-7", 1)` for `stream-7`. A handle +/// ends with its number, and a WIT name's words don't start with a digit. +fn binding_of(label: &str) -> (&str, u32) { + let is_number = |text: &str| !text.is_empty() && text.bytes().all(|b| b.is_ascii_digit()); + match label.rsplit_once('-') { + Some((handle, binding)) + if is_number(binding) && handle.rsplit_once('-').is_some_and(|(_, n)| is_number(n)) => + { + (handle, binding.parse().unwrap()) + } + _ => (label, 1), + } +} + +/// The binding of a handle's number after `label`'s, e.g. `stream-7-2` after `stream-7`. +fn next_binding(label: &str) -> String { + let (handle, binding) = binding_of(label); + format!("{handle}-{}", binding + 1) +} + +pub(crate) struct Renaming { + /// The line each label of the trace ends in. + end_of: HashMap, + to_replay: HashMap, + to_trace: HashMap, +} + +/// A WAVE value, cut into its labels, e.g. `#counter-1` or `#borrow-counter-1`, and the text +/// between them, which may have a `#` in a string or a char. +#[derive(Debug, PartialEq)] +enum Token<'a> { + Text(&'a str), + Label(&'a str), +} + +fn tokens(wave: &str) -> Vec> { + let bytes = wave.as_bytes(); + let mut tokens = Vec::new(); + let (mut start, mut idx) = (0, 0); + while idx < bytes.len() { + match bytes[idx] { + quote @ (b'"' | b'\'') => { + idx += 1; + while idx < bytes.len() && bytes[idx] != quote { + idx += if bytes[idx] == b'\\' { 2 } else { 1 }; + } + idx += 1; + } + b'#' => { + if start < idx { + tokens.push(Token::Text(&wave[start..idx])); + } + let end = idx + + 1 + + bytes[idx + 1..] + .iter() + .take_while(|byte| byte.is_ascii_alphanumeric() || **byte == b'-') + .count(); + tokens.push(Token::Label(&wave[idx..end])); + (start, idx) = (end, end); + } + _ => idx += 1, + } + } + if start < bytes.len() { + tokens.push(Token::Text(&wave[start..])); + } + tokens +} + +/// A label, e.g. `#borrow-counter-1`, as its borrow, and the handle it names, `counter-1`. +fn handle_of(label: &str) -> (bool, &str) { + let label = label.trim_start_matches('#'); + match label.strip_prefix("borrow-") { + Some(handle) => (true, handle), + None => (false, label), + } +} + +/// The table a line's labels are in: its task's, or its handle's. +fn table_of(event: &Event) -> Table { + match event { + Event::Write(chunk) | Event::Read(chunk) => chunk.stream.table, + Event::Value(value) => value.future.table, + Event::Drop(label) => label.table, + event => event.task().unwrap().table, + } +} + +/// The handles a line's labels name, and the owned ones among those in its values. +fn handles_of(line: &Line) -> (Vec, Vec) { + let table = table_of(&line.event); + let (mut all, mut owned) = (Vec::new(), Vec::new()); + let mut wave = |wave: &String| { + for token in tokens(wave) { + if let Token::Label(label) = token { + let (borrow, handle) = handle_of(label); + all.push((table, handle.to_string())); + if !borrow { + owned.push((table, handle.to_string())); + } + } + } + }; + let mut fields = Vec::new(); + match &line.event { + Event::Call(call) => call.args.iter().for_each(&mut wave), + Event::Return(ret) => ret.value.iter().for_each(&mut wave), + Event::Write(chunk) | Event::Read(chunk) => { + fields.push(&chunk.stream); + wave(&chunk.items); + } + Event::Value(value) => { + fields.push(&value.future); + wave(&value.value); + } + Event::Drop(label) => fields.push(label), + _ => {} + } + all.extend( + fields + .iter() + .map(|label| (label.table, handle_of(&label.name).1.to_string())), + ); + (all, owned) +} + +/// The handles that leave the table with a line: a dropped one, a future whose value was taken, +/// or those the host hands on, owned, to the guest. +/// +/// The party of a call or return says which way its handles move. A host line hands them from +/// a generated component to the guest: an export call's arguments, or an import's result, whose +/// labels end. A guest line hands them to a generated component: an import call's arguments, or +/// an export's result, which come in as new handles. The guest's own handles aren't labels. +fn ends_of(line: &Line) -> Vec { + let handle = |label: &Label| (label.table, handle_of(&label.name).1.to_string()); + match &line.event { + Event::Drop(label) => vec![handle(label)], + Event::Value(value) => vec![handle(&value.future)], + Event::Call(_) | Event::Return(_) if line.by == Party::Host => handles_of(line).1, + _ => Vec::new(), + } +} + +/// `wave` with its labels renamed by `rename`, of the handle a label names, if it has one. +fn rename(wave: &str, rename: impl Fn(&str) -> Option) -> String { + tokens(wave) + .into_iter() + .map(|token| match token { + Token::Text(text) => text.to_string(), + Token::Label(label) => match (handle_of(label), rename(handle_of(label).1)) { + ((true, _), Some(handle)) => format!("#borrow-{handle}"), + ((false, _), Some(handle)) => format!("#{handle}"), + (_, None) => label.to_string(), + }, + }) + .collect() +} + +/// `line` with its labels renamed by `rename`, of a handle in a table, if it has a new name. +fn map_labels(line: &Line, rename: impl Fn(Table, &str) -> Option) -> Line { + let table = table_of(&line.event); + let wave = |wave: &String| self::rename(wave, |handle| rename(table, handle)); + let label = |label: &Label| Label { + table: label.table, + name: self::rename(&label.name, |handle| rename(label.table, handle)), + }; + let event = match &line.event { + Event::Call(call) => Event::Call(crate::Call { + args: call.args.iter().map(wave).collect(), + ..call.clone() + }), + Event::Return(ret) => Event::Return(crate::ReturnValue { + value: ret.value.as_ref().map(wave), + ..ret.clone() + }), + Event::Write(chunk) => Event::Write(crate::Chunk { + stream: label(&chunk.stream), + items: wave(&chunk.items), + }), + Event::Read(chunk) => Event::Read(crate::Chunk { + stream: label(&chunk.stream), + items: wave(&chunk.items), + }), + Event::Value(value) => Event::Value(crate::FutureValue { + future: label(&value.future), + value: wave(&value.value), + }), + Event::Drop(handle) => Event::Drop(label(handle)), + event => event.clone(), + }; + Line { + event, + ..line.clone() + } +} + +impl Renaming { + /// Panics if a label of the trace names a handle after its end: a trace whose bindings + /// `Labels` didn't name apart. + pub(crate) fn new(lines: &[Line]) -> Self { + let mut labels = Labels::default(); + let mut end_of = HashMap::new(); + for (idx, line) in lines.iter().enumerate() { + assert!( + labels.label(line.clone()) == *line, + "the trace has `{line}`, with a label after the line that ended it" + ); + for handle in ends_of(line) { + end_of.insert(handle, idx); + } + } + Renaming { + end_of, + to_replay: HashMap::new(), + to_trace: HashMap::new(), + } + } + /// Whether the trace's label of `handle` has ended before line `idx`. + fn has_ended(&self, handle: &Handle, idx: usize) -> bool { + self.end_of.get(handle).is_some_and(|end| *end < idx) + } + /// Pairs the trace's label `trace` with replay's `replay`, at line `idx`, or returns false if + /// either is paired with another, which is still live. + fn pair(&mut self, idx: usize, table: Table, trace: &str, replay: &str) -> bool { + let ((trace_borrow, trace), (replay_borrow, replay)) = + (handle_of(trace), handle_of(replay)); + if trace_borrow != replay_borrow { + return false; + } + let trace = (table, trace.to_string()); + if let Some(paired) = self.to_replay.get(&trace) { + return paired == replay; + } + let replay = (table, replay.to_string()); + if let Some(other) = self.to_trace.get(&replay) { + let other = (table, other.clone()); + if !self.has_ended(&other, idx) { + return false; + } + // Replay reused the number of a handle that ended, as the trace says. + self.to_replay.remove(&other); + } + self.to_replay.insert(trace.clone(), replay.1.clone()); + self.to_trace.insert(replay, trace.1); + true + } + /// Pairs the trace's label of a stream or future replay makes, from the value of the line that + /// hands it on, with replay's, of the end it keeps: the label's next binding, e.g. the writer + /// of the stream a mock returns, whose number labeled the reader in that line. Returns false + /// if no line hands it on, or either is paired with another. + pub(crate) fn pair_next(&mut self, trace: &Label, replay: &Label) -> bool { + let handle = (trace.table, handle_of(&trace.name).1.to_string()); + let Some(end) = self.end_of.get(&handle).copied() else { + return false; + }; + trace.table == replay.table + && self.pair(end + 1, trace.table, &next_binding(&handle.1), &replay.name) + } + /// Whether the trace's WAVE value `trace`, in line `idx`, is replay's `replay`, up to the + /// pairing, which it extends. + fn unify(&mut self, idx: usize, table: Table, trace: &str, replay: &str) -> bool { + let (trace, replay) = (tokens(trace), tokens(replay)); + trace.len() == replay.len() + && trace.iter().zip(&replay).all(|tokens| match tokens { + (Token::Text(trace), Token::Text(replay)) => trace == replay, + (Token::Label(trace), Token::Label(replay)) => self.pair(idx, table, trace, replay), + _ => false, + }) + } + /// The trace's label that replay's label, e.g. a job's for the stream it keeps, is paired + /// with, if any. + pub(crate) fn to_trace(&self, label: &Label) -> Option