From 9cbc0e15ff3be97cd541f5e6492b96c8deca6422 Mon Sep 17 00:00:00 2001 From: richardanyalai Date: Wed, 14 Oct 2026 11:23:40 +0200 Subject: [PATCH 1/4] feat(core): extracted the scenario runner decentralization-manager wrote its integration suite in a Given/When/Then runner that lived in its test directory. Other repos could not use the runner without copying it. This commit moves the generic part of that runner into its own crate. A small Fixture trait replaces the direct use of decman's fixture type. Each consumer keeps its own rules for classifying a failed request, because only the consumer knows its protocol. CI mirrors decman's checks: fmt, clippy, tests, docs, cargo-audit, cargo-deny and cargo-machete. Dependabot covers cargo and GitHub Actions. --- .github/dependabot.yml | 28 ++ .github/workflows/ci.yml | 68 ++++ .gitignore | 1 + Cargo.lock | 128 ++++++++ Cargo.toml | 20 ++ LICENSE | 201 ++++++++++++ NOTICE | 6 + README.md | 158 ++++++++- deny.toml | 10 + src/lib.rs | 7 + src/probe.rs | 216 +++++++++++++ src/scenario.rs | 674 +++++++++++++++++++++++++++++++++++++++ 12 files changed, 1515 insertions(+), 2 deletions(-) create mode 100644 .github/dependabot.yml create mode 100644 .github/workflows/ci.yml create mode 100644 .gitignore create mode 100644 Cargo.lock create mode 100644 Cargo.toml create mode 100644 LICENSE create mode 100644 NOTICE create mode 100644 deny.toml create mode 100644 src/lib.rs create mode 100644 src/probe.rs create mode 100644 src/scenario.rs diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..6aa4079 --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,28 @@ +version: 2 +updates: + - package-ecosystem: cargo + directory: / + schedule: + interval: monthly + open-pull-requests-limit: 10 + commit-message: + prefix: "chore(deps)" + groups: + cargo-minor-and-patch: + update-types: + - minor + - patch + + - package-ecosystem: github-actions + directory: / + schedule: + interval: monthly + open-pull-requests-limit: 5 + commit-message: + prefix: "chore(ci)" + groups: + actions-all: + update-types: + - major + - minor + - patch diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..9f834c3 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,68 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + workflow_dispatch: + +# Only PR runs are cancelled, so a push to main always finishes. +concurrency: + group: ci-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + +env: + CARGO_TERM_COLOR: always + RUSTFLAGS: -D warnings + +jobs: + cargo: + name: Format, Clippy & Tests + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - name: Install Rust + run: | + rustup toolchain install stable --profile minimal --component rustfmt,clippy + rustup default stable + - uses: Swatinem/rust-cache@v2 + - name: cargo fmt + run: cargo fmt -- --check + - name: cargo clippy + run: cargo clippy --locked --all-targets --all-features -- -D warnings + # Also runs the README example, which is the crate-level doctest. + - name: cargo test + run: cargo test --locked --all-features + # Fails on a broken intra-doc link, which rustc alone does not check. + - name: cargo doc + run: cargo doc --locked --no-deps + env: + RUSTDOCFLAGS: -D warnings + + cargo-audit: + name: Cargo Audit + runs-on: ubuntu-latest + permissions: + checks: write + contents: read + steps: + - uses: actions/checkout@v7 + - uses: rustsec/audit-check@v2.0.0 + with: + token: ${{ secrets.GITHUB_TOKEN }} + + cargo-deny: + name: Cargo Deny + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: EmbarkStudios/cargo-deny-action@v2 + with: + command: check advisories + + cargo-machete: + name: Cargo Machete + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: bnjbvr/cargo-machete@v0.9.2 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ea8c4bf --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/target diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000..b07afda --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,128 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "anyhow" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "penelope" +version = "0.1.0" +dependencies = [ + "anyhow", + "tokio", + "tracing", +] + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "syn" +version = "3.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d78c8dee4c7bf0e14673097256fed6142ce9d3b85a408189d07482442145823b" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tokio" +version = "1.53.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce3335fa71841cda333a58d7615b03901380ecf09d59b3296d21f8bbac0dde4e" +dependencies = [ + "pin-project-lite", + "tokio-macros", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + +[[package]] +name = "unicode-ident" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2c754d6c33795a1c324727428e5a7dedb5b06195f9890bdbcba760d3e246563" diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..5afd241 --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "penelope" +version = "0.1.0" +edition = "2024" +description = "A Given/When/Then scenario runner for eventually consistent systems" +license = "Apache-2.0" +repository = "https://github.com/DLC-link/penelope" +readme = "README.md" + +[dependencies] +anyhow = { version = "1.0.98" } +tokio = { version = "1.48.0", features = ["time"] } +tracing = { version = "0.1" } + +[dev-dependencies] +tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "time"] } + +[lints.rust] +missing_docs = "warn" +unsafe_code = "forbid" diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..78173fa --- /dev/null +++ b/LICENSE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + +TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + +1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + +2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + +3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + +4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + +5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + +6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + +7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + +8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + +9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + +END OF TERMS AND CONDITIONS + +APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + +Copyright [yyyy] [name of copyright owner] + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. diff --git a/NOTICE b/NOTICE new file mode 100644 index 0000000..d54e84e --- /dev/null +++ b/NOTICE @@ -0,0 +1,6 @@ +Penelope +Copyright 2026 BitSafe Finance + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this software except in compliance with the License. +A copy of the License is provided in the LICENSE file in this repository. diff --git a/README.md b/README.md index 85aed26..c7c779d 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,156 @@ -# penelope -A Given/When/Then scenario runner for eventually consistent systems +# Penelope + +Penelope runs Given/When/Then scenarios against systems that become +consistent over time, such as a cluster of replicated nodes. + +In the myth, Penelope waited twenty years for Odysseus. When he came home, +she tested him before she accepted him. This crate does the same thing. It +waits for the system to settle, and then it checks the result. + +## Why + +A distributed system rarely shows a change at once. A test that checks too +early fails. A fixed sleep wastes time, and the test can still fail on a slow +day. Penelope polls each Then step until the expected state appears. It also +records why each attempt failed, so a timeout says what went wrong. + +## Install + +Add the crate as a git dependency, and pin a commit: + +```toml +[dev-dependencies] +penelope = { git = "https://github.com/DLC-link/penelope", rev = "" } +``` + +The pin means your tests change only when you move it to a new commit. + +## Example + +The fixture below is an order store that applies each write after a delay. +The Then step polls the store until the order appears. + +```rust +use std::{ + sync::{Arc, Mutex}, + time::Duration, +}; + +use penelope::{Fixture, ProbeDiag, Scenario}; + +#[derive(Default)] +struct World { + diag: ProbeDiag, + orders: Arc>>, +} + +impl World { + async fn count_orders(&self) -> anyhow::Result { + Ok(self.orders.lock().unwrap().len()) + } +} + +impl Fixture for World { + fn probe_diag(&self) -> &ProbeDiag { + &self.diag + } +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let mut world = World::default(); + + Scenario::::new("a placed order becomes visible") + .given("the store is empty", |w, _| { + Box::pin(async move { + anyhow::ensure!(w.count_orders().await? == 0, "the store is not empty"); + Ok(()) + }) + }) + .when("a client places an order", |w, _| { + let orders = w.orders.clone(); + Box::pin(async move { + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(500)).await; + orders.lock().unwrap().push("order-1".to_string()); + }); + Ok(()) + }) + }) + .then("the store lists one order", Duration::from_secs(10), |w, _| { + let diag = w.probe_diag().clone(); + Box::pin(async move { + let count = diag.ok("count orders", w.count_orders().await)?; + (count == 1).then_some(Ok(())) + }) + }) + .run(&mut world) + .await +} +``` + +## Step kinds + +| Step | How often it runs | Use it for | +|-------|-----------------------------------------|------------------------------------------------| +| Given | Once | State that the test knows or can check first | +| When | Once | An action on the system under test | +| Then | Until it passes, fails, or times out | An observation, such as "node 2 sees the row" | + +A Then probe returns one of three values: + +- `Some(Ok(()))` passes the step. +- `None` means "not observable yet". The runner tries again after 2 seconds. +- `Some(Err(e))` fails the step at once. + +Steps can share state through a context value. Start the scenario with +`Scenario::with_ctx(name, ctx)`. Each step then receives `&mut ctx` as its +second argument. A Then probe can store what it observed there for a later +step. + +## Probe diagnostics + +A probe that returns `None` gives the runner no reason. Without a reason, a +typo in a probe's request and a slow server both end in a bare "timed out". + +The fixture's `ProbeDiag` holds that reason. A probe records each failure +against a key, such as `GET /orders`: + +- `record(key, Class::Transient, message)` notes a failure that can still + clear, such as a restart or a 503. +- `record(key, Class::Fatal, message)` notes a failure that retries cannot + fix, such as a route that does not exist. +- `record_ok(key)` clears the streak for that key. +- `ok(key, result)` records an `Err` as transient and returns `None`. + Otherwise it returns the value. + +The runner ends a Then step early when the same fatal failure repeats +`FATAL_STREAK` times in a row. That limit is 3. On a timeout, the error +message includes the last recorded failure. The probe chooses each failure's +class, because only the probe knows its protocol. + +## Naming the fixture type once + +A closure that reads a fixture field needs the fixture type. Declare a type +alias once in your test suite: + +```rust,ignore +pub type Scenario = penelope::Scenario; +``` + +After that, `Scenario::new("...")` works without a type annotation. + +## Logging + +The runner logs each step through `tracing` at the `info` level, under the +target `penelope`. If you set `RUST_LOG` to a list of targets, add +`penelope=info` to the list. + +## Origin + +Penelope began as the integration-test DSL of +[decentralization-manager](https://github.com/DLC-link/decentralization-manager). + +## License + +Apache-2.0. See the `LICENSE` and `NOTICE` files. diff --git a/deny.toml b/deny.toml new file mode 100644 index 0000000..16c71ec --- /dev/null +++ b/deny.toml @@ -0,0 +1,10 @@ +# cargo-deny is the only RustSec advisories gate in the Bitsafe baseline, so +# this is the only advisory-exception list the repo has. +# +# Every ignored RUSTSEC ID carries a `reason` that names the crate that brings +# it in, why no fix exists, and what would make the ignore unsafe. Do not add +# an entry for a finding behind a disabled feature: cargo-deny already skips +# it. Confirm with `cargo tree -i `. + +[advisories] +ignore = [] diff --git a/src/lib.rs b/src/lib.rs new file mode 100644 index 0000000..f314cc5 --- /dev/null +++ b/src/lib.rs @@ -0,0 +1,7 @@ +#![doc = include_str!("../README.md")] + +mod probe; +mod scenario; + +pub use probe::{Class, FATAL_STREAK, ProbeDiag}; +pub use scenario::{BoxFut, BoxProbe, Fixture, Scenario}; diff --git a/src/probe.rs b/src/probe.rs new file mode 100644 index 0000000..4a8e6f6 --- /dev/null +++ b/src/probe.rs @@ -0,0 +1,216 @@ +//! Failure diagnostics for polled Then probes. +//! +//! A probe returns `None` to mean "not observable yet, keep polling". That +//! makes every failure — a restart window, a 500, a typo in the probe's own +//! request — look identical to the [`Scenario`](crate::Scenario) runner, +//! which then reports nothing but "timed out after 60s". +//! +//! [`ProbeDiag`] is the side channel that closes the gap: probes record what +//! went wrong, the runner reads it back to annotate the timeout and to end the +//! step early when the request itself is the problem. +//! +//! The probe classifies each failure, because only the probe knows its +//! protocol. An HTTP probe might call a 503 transient and a 400 fatal. + +use std::sync::{Arc, Mutex, MutexGuard}; + +use tracing::{debug, warn}; + +/// Consecutive identical [`Class::Fatal`] failures that end a polled step early +/// instead of letting it run to its deadline. +pub const FATAL_STREAK: u32 = 3; + +/// Whether a probe failure can still resolve if the probe is retried. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Class { + /// The SUT may still get there: connection refused during a restart, a + /// 5xx, a token blip, or a handler answering "that resource does not exist + /// yet" for something a later step creates. + Transient, + /// The request will stay wrong however long it is retried: no such route, + /// a malformed query, a success body the probe cannot parse. + Fatal, +} + +#[derive(Debug, Default)] +struct Diag { + last_error: Option, + key: Option, + class: Option, + repeats: u32, +} + +/// Shared record of the most recent probe failure, cleared by the runner at +/// the start of every polled step. +/// +/// Clones share one record, so a probe can move a clone into its future. +#[derive(Debug, Clone, Default)] +pub struct ProbeDiag(Arc>); + +impl ProbeDiag { + fn guard(&self) -> MutexGuard<'_, Diag> { + self.0 + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// Record a failure against `key` (a request identity, e.g. `GET /orders`). + /// Repeats of the same key are counted; the first is logged at `warn` and + /// the rest at `debug` so a 60s poll cannot bury the run's real warnings. + pub fn record(&self, key: &str, class: Class, message: String) { + let mut d = self.guard(); + if d.key.as_deref() == Some(key) && d.class == Some(class) { + d.repeats += 1; + } else { + d.key = Some(key.to_string()); + d.class = Some(class); + d.repeats = 1; + } + if d.repeats == 1 { + warn!("probe failure ({class:?}): {message}"); + } else { + debug!("probe failure ({class:?}, x{}): {message}", d.repeats); + } + d.last_error = Some(message); + } + + /// Keep polling on `Err`, but record why first. For probe reads that + /// cannot be classified — a database count, a multi-call query — so the + /// failure is recorded as transient and never ends the step early. + pub fn ok(&self, key: &str, result: anyhow::Result) -> Option { + match result { + Ok(v) => { + self.record_ok(key); + Some(v) + } + Err(e) => { + self.record(key, Class::Transient, format!("{key}: {e:#}")); + None + } + } + } + + /// Note that `key` succeeded, ending any streak it had accumulated. + pub fn record_ok(&self, key: &str) { + let mut d = self.guard(); + if d.key.as_deref() == Some(key) { + *d = Diag::default(); + } + } + + /// The failure a polled step should stop on: the same fatal request + /// repeated [`FATAL_STREAK`] times. + pub fn fatal(&self) -> Option { + let d = self.guard(); + (d.class == Some(Class::Fatal) && d.repeats >= FATAL_STREAK) + .then(|| d.last_error.clone()) + .flatten() + } + + /// The message of the most recently recorded failure, if any. + pub fn last_error(&self) -> Option { + self.guard().last_error.clone() + } + + /// Forget every recorded failure. + pub fn reset(&self) { + *self.guard() = Diag::default(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fatal_reported_only_after_the_streak() { + let diag = ProbeDiag::default(); + for _ in 0..FATAL_STREAK - 1 { + diag.record("GET /typo", Class::Fatal, "404".to_string()); + assert_eq!(diag.fatal(), None); + } + diag.record("GET /typo", Class::Fatal, "404".to_string()); + assert_eq!(diag.fatal().as_deref(), Some("404")); + } + + #[test] + fn a_different_request_restarts_the_streak() { + let diag = ProbeDiag::default(); + for _ in 0..FATAL_STREAK { + diag.record("GET /a", Class::Fatal, "404 a".to_string()); + } + diag.record("GET /b", Class::Fatal, "404 b".to_string()); + assert_eq!(diag.fatal(), None); + assert_eq!(diag.last_error().as_deref(), Some("404 b")); + } + + #[test] + fn a_class_change_on_the_same_request_restarts_the_streak() { + let diag = ProbeDiag::default(); + for _ in 0..FATAL_STREAK - 1 { + diag.record("GET /a", Class::Fatal, "404".to_string()); + } + diag.record("GET /a", Class::Transient, "503".to_string()); + diag.record("GET /a", Class::Fatal, "404".to_string()); + assert_eq!(diag.fatal(), None); + } + + #[test] + fn transient_repeats_never_become_fatal() { + let diag = ProbeDiag::default(); + for _ in 0..FATAL_STREAK * 3 { + diag.record("GET /x", Class::Transient, "connection refused".to_string()); + } + assert_eq!(diag.fatal(), None); + assert_eq!(diag.last_error().as_deref(), Some("connection refused")); + } + + #[test] + fn success_clears_its_own_streak_only() { + let diag = ProbeDiag::default(); + for _ in 0..FATAL_STREAK { + diag.record("GET /a", Class::Fatal, "404 a".to_string()); + } + diag.record_ok("GET /b"); + assert_eq!(diag.fatal().as_deref(), Some("404 a")); + diag.record_ok("GET /a"); + assert_eq!(diag.fatal(), None); + assert_eq!(diag.last_error(), None); + } + + #[test] + fn ok_records_an_error_as_transient_and_passes_a_value_through() { + let diag = ProbeDiag::default(); + for _ in 0..FATAL_STREAK * 2 { + let read: anyhow::Result = Err(anyhow::anyhow!("db locked")); + assert_eq!(diag.ok("COUNT orders", read), None); + } + assert_eq!(diag.fatal(), None); + assert_eq!( + diag.last_error().as_deref(), + Some("COUNT orders: db locked") + ); + + assert_eq!(diag.ok("COUNT orders", Ok(7)), Some(7)); + assert_eq!(diag.last_error(), None); + } + + #[test] + fn clones_share_one_record() { + let diag = ProbeDiag::default(); + let clone = diag.clone(); + clone.record("GET /a", Class::Transient, "503".to_string()); + assert_eq!(diag.last_error().as_deref(), Some("503")); + } + + #[test] + fn reset_clears_everything() { + let diag = ProbeDiag::default(); + for _ in 0..FATAL_STREAK { + diag.record("GET /a", Class::Fatal, "404 a".to_string()); + } + diag.reset(); + assert_eq!(diag.fatal(), None); + assert_eq!(diag.last_error(), None); + } +} diff --git a/src/scenario.rs b/src/scenario.rs new file mode 100644 index 0000000..add5263 --- /dev/null +++ b/src/scenario.rs @@ -0,0 +1,674 @@ +//! Given/When/Then DSL for integration tests against an asynchronous system. +//! +//! See [`Scenario`] for the runner and the three step kinds (single-shot +//! Given/When, polled Then) and how they map onto an asynchronous +//! distributed SUT. + +use std::{ + future::Future, + pin::Pin, + time::{Duration, Instant}, +}; + +use anyhow::Result; +use tracing::{error, info}; + +use crate::probe::{FATAL_STREAK, ProbeDiag}; + +/// The future a Given or When step returns. +pub type BoxFut<'a> = Pin> + Send + 'a>>; + +/// The future a Then probe returns: `Some(Ok(()))` passes the step, `None` +/// polls again, and `Some(Err(_))` fails the step at once. +pub type BoxProbe<'a> = Pin>> + Send + 'a>>; + +/// The pause between two attempts of a Then probe that is not ready yet. +const POLL_INTERVAL: Duration = Duration::from_secs(2); + +/// The test state a [`Scenario`] runs against. +/// +/// The runner reads the fixture's [`ProbeDiag`]: it clears it before every +/// Then step, ends the step early on a repeated fatal failure, and adds the +/// last recorded failure to a timeout. +pub trait Fixture { + /// The diagnostics record that this fixture's probes write to. + fn probe_diag(&self) -> &ProbeDiag; +} + +#[derive(Debug, Clone, Copy)] +enum StepKind { + Given, + When, + Then, +} + +impl StepKind { + fn label(self) -> &'static str { + match self { + StepKind::Given => "GIVEN", + StepKind::When => "WHEN ", + StepKind::Then => "THEN ", + } + } +} + +type ImmediateBody = + Box FnMut(&'a mut F, &'a mut Ctx) -> BoxFut<'a> + Send + 'static>; + +type ProbeBody = + Box FnMut(&'a mut F, &'a mut Ctx) -> BoxProbe<'a> + Send + 'static>; + +enum Step { + Immediate { + kind: StepKind, + name: String, + body: ImmediateBody, + }, + Polled { + name: String, + deadline: Duration, + probe: ProbeBody, + }, +} + +/// A propose / confirm / observe scenario builder. +/// +/// The DSL distinguishes only three kinds of step, matching how the test +/// reads against an asynchronous distributed SUT. All step bodies are +/// `async` (they return a `BoxFut`); the kinds differ in *how the runner +/// executes them*, not in whether they can do I/O: +/// +/// - **Given** (single-shot fact / setup): the body is awaited exactly once. +/// Use for state the test directly knows or can directly verify before +/// acting. +/// - **When** (single-shot action): the body is awaited exactly once. +/// Submits work to the SUT. +/// - **Then** (polled observation): the probe is awaited in a loop until it +/// returns `Some(Ok(()))`, the `deadline` elapses, or it fails fast on +/// `Some(Err(_))`. Every "wait until X is observable on node N" is a +/// Then — cross-node visibility waits are observations, not setup. A Then +/// probe may write observed state into `ctx` for use by later steps; +/// capture is intrinsic to "wait until observable, remember what was +/// observed". +/// +/// `F` is the [`Fixture`] every step receives, and `Ctx` is state the steps +/// of one scenario share. +pub struct Scenario { + name: String, + ctx: Ctx, + steps: Vec>, +} + +impl Scenario { + /// Starts a scenario whose steps share no state. + pub fn new(name: impl Into) -> Self { + Self::with_ctx(name, ()) + } +} + +impl Scenario { + /// Starts a scenario whose steps share `ctx`. + pub fn with_ctx(name: impl Into, ctx: Ctx) -> Self { + Self { + name: name.into(), + ctx, + steps: Vec::new(), + } + } + + /// Adds a Given step: a fact or setup, awaited once. + pub fn given(mut self, name: impl Into, body: B) -> Self + where + B: for<'a> FnMut(&'a mut F, &'a mut Ctx) -> BoxFut<'a> + Send + 'static, + { + self.steps.push(Step::Immediate { + kind: StepKind::Given, + name: name.into(), + body: Box::new(body), + }); + self + } + + /// Adds a When step: an action on the SUT, awaited once. + pub fn when(mut self, name: impl Into, body: B) -> Self + where + B: for<'a> FnMut(&'a mut F, &'a mut Ctx) -> BoxFut<'a> + Send + 'static, + { + self.steps.push(Step::Immediate { + kind: StepKind::When, + name: name.into(), + body: Box::new(body), + }); + self + } + + /// Adds a Then step: `probe` is polled until it passes, fails, or + /// `deadline` elapses. + pub fn then

(mut self, name: impl Into, deadline: Duration, probe: P) -> Self + where + P: for<'a> FnMut(&'a mut F, &'a mut Ctx) -> BoxProbe<'a> + Send + 'static, + { + self.steps.push(Step::Polled { + name: name.into(), + deadline, + probe: Box::new(probe), + }); + self + } + + /// Runs the steps in order against `f`, stopping at the first failure. + /// + /// # Errors + /// + /// Returns the failing step's error, with the step and scenario names + /// added as context. A Then step also fails when its deadline elapses, or + /// when the fixture's [`ProbeDiag`] records the same fatal failure + /// [`FATAL_STREAK`] times. + pub async fn run(mut self, f: &mut F) -> Result<()> { + info!(scenario = %self.name, " Scenario \"{}\"", self.name); + let scenario_start = Instant::now(); + + for step in &mut self.steps { + match step { + Step::Immediate { kind, name, body } => { + info!(step_kind = ?kind, step_name = %name, " {} {}", kind.label(), name); + let fut = body(f, &mut self.ctx); + match fut.await { + Ok(()) => {} + Err(e) => { + error!( + scenario = %self.name, + "Scenario \"{}\" failed at {} \"{}\"", + self.name, + kind.label(), + name + ); + return Err(e + .context(format!("{} \"{}\"", kind.label(), name)) + .context(format!("scenario \"{}\" failed", self.name))); + } + } + } + Step::Polled { + name, + deadline, + probe, + } => { + let kind = StepKind::Then; + info!(step_kind = ?kind, step_name = %name, " {} {}", kind.label(), name); + let step_start = Instant::now(); + f.probe_diag().reset(); + let outcome: Result<()> = loop { + match probe(f, &mut self.ctx).await { + Some(Ok(())) => break Ok(()), + Some(Err(e)) => { + break Err(e.context(format!("THEN \"{name}\" failed"))); + } + None => { + if let Some(fatal) = f.probe_diag().fatal() { + break Err(anyhow::anyhow!( + "THEN \"{}\" failed fast: same terminal probe error \ + {FATAL_STREAK}x: {}", + name, + fatal + )); + } + let elapsed = step_start.elapsed(); + if elapsed >= *deadline { + break Err(match f.probe_diag().last_error() { + Some(last) => anyhow::anyhow!( + "THEN \"{}\" timed out after {:?}; \ + last probe error: {}", + name, + *deadline, + last + ), + None => anyhow::anyhow!( + "THEN \"{}\" timed out after {:?}", + name, + *deadline + ), + }); + } + let remaining = *deadline - elapsed; + tokio::time::sleep(std::cmp::min(POLL_INTERVAL, remaining)).await; + } + } + }; + let took = step_start.elapsed(); + match outcome { + Ok(()) => info!(" ✓ (took {:.1}s)", took.as_secs_f64()), + Err(e) => { + error!( + scenario = %self.name, + "Scenario \"{}\" failed at THEN \"{}\"", + self.name, + name + ); + return Err(e.context(format!("scenario \"{}\" failed", self.name))); + } + } + } + } + } + + info!( + scenario = %self.name, + " Scenario \"{}\" complete ({:.1}s)", + self.name, + scenario_start.elapsed().as_secs_f64() + ); + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use std::{ + sync::{ + Arc, Mutex, + atomic::{AtomicU32, Ordering}, + }, + time::{Duration, Instant}, + }; + + use super::*; + use crate::probe::Class; + + /// A stand-in fixture: one field a step can borrow, plus the diagnostics + /// record the runner needs. + struct World { + port: u16, + diag: ProbeDiag, + } + + impl World { + fn new() -> Self { + Self { + port: 8081, + diag: ProbeDiag::default(), + } + } + } + + impl Fixture for World { + fn probe_diag(&self) -> &ProbeDiag { + &self.diag + } + } + + // The alias a consumer declares once, so `Scenario::new` knows the + // fixture type before a closure reads one of its fields. + type Scenario = super::Scenario; + + #[tokio::test(flavor = "multi_thread")] + async fn steps_execute_in_order() { + let mut f = World::new(); + let order = Arc::new(Mutex::new(Vec::::new())); + let o1 = order.clone(); + let o2 = order.clone(); + let o3 = order.clone(); + + Scenario::new("ordering") + .given("g", move |_f, _c| { + let o = o1.clone(); + Box::pin(async move { + o.lock().unwrap().push(1); + Ok(()) + }) + }) + .when("w", move |_f, _c| { + let o = o2.clone(); + Box::pin(async move { + o.lock().unwrap().push(2); + Ok(()) + }) + }) + .then("t", Duration::from_secs(1), move |_f, _c| { + let o = o3.clone(); + Box::pin(async move { + o.lock().unwrap().push(3); + Some(Ok(())) + }) + }) + .run(&mut f) + .await + .unwrap(); + + assert_eq!(*order.lock().unwrap(), vec![1, 2, 3]); + } + + #[tokio::test(flavor = "multi_thread")] + async fn step_error_wraps_with_scenario_and_step_context() { + let mut f = World::new(); + let err = Scenario::new("my scn") + .when("the failing step", |_f, _c| { + Box::pin(async move { Err(anyhow::anyhow!("boom")) }) + }) + .run(&mut f) + .await + .unwrap_err(); + + let chain = format!("{err:#}"); + assert!(chain.contains("scenario \"my scn\" failed"), "got: {chain}"); + assert!( + chain.contains("WHEN") && chain.contains("the failing step"), + "got: {chain}" + ); + assert!(chain.contains("boom"), "got: {chain}"); + } + + #[tokio::test(flavor = "multi_thread")] + async fn a_failed_step_stops_the_scenario() { + let mut f = World::new(); + let later = Arc::new(AtomicU32::new(0)); + let l = later.clone(); + + let outcome = Scenario::new("stops") + .given("fails", |_f, _c| { + Box::pin(async move { Err(anyhow::anyhow!("no")) }) + }) + .when("never runs", move |_f, _c| { + let l = l.clone(); + Box::pin(async move { + l.fetch_add(1, Ordering::SeqCst); + Ok(()) + }) + }) + .run(&mut f) + .await; + + assert!(outcome.is_err()); + assert_eq!(later.load(Ordering::SeqCst), 0); + } + + #[tokio::test(flavor = "multi_thread")] + async fn then_retries_until_some_ok() { + let mut f = World::new(); + let counter = Arc::new(AtomicU32::new(0)); + let c = counter.clone(); + Scenario::new("retry") + .then( + "eventually ready", + Duration::from_secs(10), + move |_f, _c| { + let c = c.clone(); + Box::pin(async move { + let n = c.fetch_add(1, Ordering::SeqCst) + 1; + if n >= 3 { Some(Ok(())) } else { None } + }) + }, + ) + .run(&mut f) + .await + .unwrap(); + assert!(counter.load(Ordering::SeqCst) >= 3); + } + + #[tokio::test(flavor = "multi_thread")] + async fn then_fail_fasts_on_some_err() { + let mut f = World::new(); + let started = Instant::now(); + let err = Scenario::new("fast-fail") + .then("explodes", Duration::from_secs(30), |_f, _c| { + Box::pin(async move { Some(Err(anyhow::anyhow!("kaboom"))) }) + }) + .run(&mut f) + .await + .unwrap_err(); + + // Should NOT wait for the 30s deadline. + assert!( + started.elapsed() < Duration::from_secs(5), + "elapsed: {:?}", + started.elapsed() + ); + let chain = format!("{err:#}"); + assert!(chain.contains("kaboom"), "got: {chain}"); + assert!(chain.contains("scenario \"fast-fail\""), "got: {chain}"); + // Probe-error path must wrap with "failed", not "timed out". + assert!(chain.contains("THEN \"explodes\" failed"), "got: {chain}"); + assert!(!chain.contains("timed out"), "got: {chain}"); + } + + #[tokio::test(flavor = "multi_thread")] + async fn then_times_out_when_probe_only_returns_none() { + let mut f = World::new(); + let started = Instant::now(); + let err = Scenario::new("never-ready") + .then("eternal None", Duration::from_millis(100), |_f, _c| { + Box::pin(async move { None }) + }) + .run(&mut f) + .await + .unwrap_err(); + + let elapsed = started.elapsed(); + assert!( + elapsed >= Duration::from_millis(100), + "elapsed: {elapsed:?}" + ); + // Sleep must respect the remaining deadline, not always wait the full + // POLL_INTERVAL. Loose upper bound to avoid flakes on slow CI. + assert!( + elapsed < Duration::from_secs(1), + "deadline overshoot: {elapsed:?}" + ); + let chain = format!("{err:#}"); + assert!( + chain.contains("scenario \"never-ready\" failed"), + "got: {chain}" + ); + assert!(chain.contains("eternal None"), "got: {chain}"); + assert!(chain.contains("timed out"), "got: {chain}"); + assert!(!chain.contains("last probe error"), "got: {chain}"); + } + + /// A probe that can never succeed must not spend the whole deadline, and + /// the failure it kept swallowing must reach the report. + #[tokio::test(flavor = "multi_thread")] + async fn then_fails_fast_on_a_repeated_fatal_probe_error() { + let mut f = World::new(); + let started = Instant::now(); + let attempts = Arc::new(AtomicU32::new(0)); + let a = attempts.clone(); + + let outcome = Scenario::new("typo") + .then( + "never observable", + Duration::from_secs(600), + move |f, _c| { + let a = a.clone(); + let diag = f.probe_diag().clone(); + Box::pin(async move { + a.fetch_add(1, Ordering::SeqCst); + diag.record( + "GET :8081/parties/participants-status", + Class::Fatal, + "GET http://localhost:8081/parties/participants-status returned 404 \ + Not Found: 404 Not Found" + .to_string(), + ); + None + }) + }, + ) + .run(&mut f) + .await; + + let Err(err) = outcome else { + panic!("expected the scenario to fail"); + }; + assert!( + started.elapsed() < Duration::from_secs(30), + "elapsed: {:?}", + started.elapsed() + ); + assert_eq!(attempts.load(Ordering::SeqCst), FATAL_STREAK); + let chain = format!("{err:#}"); + assert!(chain.contains("failed fast"), "got: {chain}"); + assert!(chain.contains("participants-status"), "got: {chain}"); + assert!(chain.contains("404"), "got: {chain}"); + } + + #[tokio::test(flavor = "multi_thread")] + async fn timeout_message_carries_the_last_probe_error() { + let mut f = World::new(); + let outcome = Scenario::new("slow") + .then("never ready", Duration::from_millis(100), |f, _c| { + let diag = f.probe_diag().clone(); + Box::pin(async move { + diag.record( + "GET :8081/workflows", + Class::Transient, + "GET http://localhost:8081/workflows returned 503".to_string(), + ); + None + }) + }) + .run(&mut f) + .await; + + let Err(err) = outcome else { + panic!("expected the scenario to fail"); + }; + let chain = format!("{err:#}"); + assert!(chain.contains("timed out"), "got: {chain}"); + assert!(chain.contains("last probe error"), "got: {chain}"); + assert!(chain.contains("503"), "got: {chain}"); + } + + /// Each step starts from a clean slate, so a step never gets blamed for + /// the failure an earlier one recovered from. + #[tokio::test(flavor = "multi_thread")] + async fn a_previous_steps_probe_error_does_not_leak_into_the_next() { + let mut f = World::new(); + let outcome = Scenario::new("two steps") + .then("recovers", Duration::from_secs(10), |f, _c| { + let diag = f.probe_diag().clone(); + Box::pin(async move { + diag.record( + "GET :8081/a", + Class::Transient, + "stale error from step one".to_string(), + ); + Some(Ok(())) + }) + }) + .then("times out clean", Duration::from_millis(100), |_f, _c| { + Box::pin(async move { None }) + }) + .run(&mut f) + .await; + + let Err(err) = outcome else { + panic!("expected the scenario to fail"); + }; + let chain = format!("{err:#}"); + assert!(chain.contains("times out clean"), "got: {chain}"); + assert!(!chain.contains("stale error"), "got: {chain}"); + } + + #[tokio::test(flavor = "multi_thread")] + async fn ctx_mutations_are_visible_to_later_steps() { + let mut f = World::new(); + #[derive(Default)] + struct C { + n: u32, + } + + Scenario::with_ctx("ctx", C::default()) + .when("write", |_f, c| { + Box::pin(async move { + c.n = 42; + Ok(()) + }) + }) + .then("read", Duration::from_secs(1), |_f, c| { + Box::pin(async move { + assert_eq!(c.n, 42); + Some(Ok(())) + }) + }) + .run(&mut f) + .await + .unwrap(); + } + + #[tokio::test(flavor = "multi_thread")] + async fn polled_then_completes_before_subsequent_step() { + // A polled Then that takes multiple probe attempts must finish + // (probe returns Some(Ok)) before the next step runs. + let mut f = World::new(); + let probe_count = Arc::new(AtomicU32::new(0)); + let when_count = Arc::new(AtomicU32::new(0)); + let pc = probe_count.clone(); + let wc = when_count.clone(); + + Scenario::new("polled-blocks-next") + .then( + "ready after a few probes", + Duration::from_secs(10), + move |_f, _c| { + let pc = pc.clone(); + Box::pin(async move { + let n = pc.fetch_add(1, Ordering::SeqCst) + 1; + if n >= 2 { Some(Ok(())) } else { None } + }) + }, + ) + .when("after polled then", move |_f, _c| { + let wc = wc.clone(); + Box::pin(async move { + wc.fetch_add(1, Ordering::SeqCst); + Ok(()) + }) + }) + .run(&mut f) + .await + .unwrap(); + + assert!(probe_count.load(Ordering::SeqCst) >= 2); + assert_eq!(when_count.load(Ordering::SeqCst), 1); + } + + /// Smoke test for the HRTB closure pattern every consumer relies on: + /// a closure that captures a non-static `String` AND borrows the + /// `&mut` fixture parameter inside the future. If this fails to compile, + /// every consumer's steps fail too. + #[tokio::test(flavor = "multi_thread")] + async fn closure_can_capture_string_and_borrow_fixture() { + let mut f = World::new(); + let label = String::from("hello"); + + Scenario::new("hrtb-smoke") + .when("uses captured string and borrows f", { + let label = label.clone(); + move |f, _c| { + let label = label.clone(); + Box::pin(async move { + f.port += 1; + assert_eq!(label, "hello"); + Ok(()) + }) + } + }) + .then( + "captured string still readable in polled then", + Duration::from_secs(1), + { + let label = label.clone(); + move |f, _c| { + let label = label.clone(); + Box::pin(async move { + assert_eq!(f.port, 8082); + if label == "hello" { Some(Ok(())) } else { None } + }) + } + }, + ) + .run(&mut f) + .await + .unwrap(); + } +} From 34b0a6bbf7a1226d3269520d455e58b30b93a5d4 Mon Sep 17 00:00:00 2001 From: richardanyalai Date: Wed, 14 Oct 2026 11:33:18 +0200 Subject: [PATCH 2/4] fix(core): bounded each probe call by the deadline The runner checked a Then step's deadline only after a probe returned None. A probe that never returned kept the scenario pending forever, and a probe that answered after the deadline still passed the step. Each probe call now runs under the step's deadline. When the deadline passes, the runner cancels the call and reports the same timeout error, with the last recorded probe failure. --- README.md | 3 ++ src/scenario.rs | 101 +++++++++++++++++++++++++++++++++++++++--------- 2 files changed, 85 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index c7c779d..907bb0d 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,9 @@ A Then probe returns one of three values: - `None` means "not observable yet". The runner tries again after 2 seconds. - `Some(Err(e))` fails the step at once. +The deadline also limits each probe call. The runner cancels a probe that is +still running when the deadline passes, and the step fails. + Steps can share state through a context value. Start the scenario with `Scenario::with_ctx(name, ctx)`. Each step then receives `&mut ctx` as its second argument. A Then probe can store what it observed there for a later diff --git a/src/scenario.rs b/src/scenario.rs index add5263..838205c 100644 --- a/src/scenario.rs +++ b/src/scenario.rs @@ -143,7 +143,8 @@ impl Scenario { } /// Adds a Then step: `probe` is polled until it passes, fails, or - /// `deadline` elapses. + /// `deadline` elapses. A probe call still running at the deadline is + /// cancelled, and the step fails. pub fn then

(mut self, name: impl Into, deadline: Duration, probe: P) -> Self where P: for<'a> FnMut(&'a mut F, &'a mut Ctx) -> BoxProbe<'a> + Send + 'static, @@ -197,14 +198,20 @@ impl Scenario { let kind = StepKind::Then; info!(step_kind = ?kind, step_name = %name, " {} {}", kind.label(), name); let step_start = Instant::now(); + let step_deadline = tokio::time::Instant::from_std(step_start) + *deadline; f.probe_diag().reset(); let outcome: Result<()> = loop { - match probe(f, &mut self.ctx).await { - Some(Ok(())) => break Ok(()), - Some(Err(e)) => { + // The deadline bounds each probe call too, so a probe + // that hangs cannot hold the scenario past it. + let attempt = + tokio::time::timeout_at(step_deadline, probe(f, &mut self.ctx)).await; + match attempt { + Err(_) => break Err(timed_out(name, *deadline, f.probe_diag())), + Ok(Some(Ok(()))) => break Ok(()), + Ok(Some(Err(e))) => { break Err(e.context(format!("THEN \"{name}\" failed"))); } - None => { + Ok(None) => { if let Some(fatal) = f.probe_diag().fatal() { break Err(anyhow::anyhow!( "THEN \"{}\" failed fast: same terminal probe error \ @@ -215,20 +222,7 @@ impl Scenario { } let elapsed = step_start.elapsed(); if elapsed >= *deadline { - break Err(match f.probe_diag().last_error() { - Some(last) => anyhow::anyhow!( - "THEN \"{}\" timed out after {:?}; \ - last probe error: {}", - name, - *deadline, - last - ), - None => anyhow::anyhow!( - "THEN \"{}\" timed out after {:?}", - name, - *deadline - ), - }); + break Err(timed_out(name, *deadline, f.probe_diag())); } let remaining = *deadline - elapsed; tokio::time::sleep(std::cmp::min(POLL_INTERVAL, remaining)).await; @@ -262,6 +256,19 @@ impl Scenario { } } +/// The error for a Then step that reached its deadline, carrying the last +/// recorded probe failure when there is one. +fn timed_out(name: &str, deadline: Duration, diag: &ProbeDiag) -> anyhow::Error { + match diag.last_error() { + Some(last) => { + anyhow::anyhow!( + "THEN \"{name}\" timed out after {deadline:?}; last probe error: {last}" + ) + } + None => anyhow::anyhow!("THEN \"{name}\" timed out after {deadline:?}"), + } +} + #[cfg(test)] mod tests { use std::{ @@ -511,6 +518,62 @@ mod tests { assert!(chain.contains("404"), "got: {chain}"); } + /// The deadline must bound the probe call itself, or a probe that never + /// returns keeps the scenario pending forever. + #[tokio::test(flavor = "multi_thread")] + async fn then_times_out_when_a_probe_hangs() { + let mut f = World::new(); + let scenario = Scenario::new("hang") + .then("never returns", Duration::from_millis(100), |_f, _c| { + Box::pin(std::future::pending()) + }) + .run(&mut f); + + // The outer limit turns a regression into a failure, not a hung suite. + let Ok(outcome) = tokio::time::timeout(Duration::from_secs(5), scenario).await else { + panic!("the step deadline did not end a probe that never returns"); + }; + let err = outcome.unwrap_err(); + let chain = format!("{err:#}"); + assert!( + chain.contains("THEN \"never returns\" timed out"), + "got: {chain}" + ); + } + + /// A probe still running at the deadline is cancelled, so an answer that + /// would only arrive later cannot pass the step. The failure it recorded + /// before the deadline still reaches the report. + #[tokio::test(flavor = "multi_thread")] + async fn a_probe_that_would_pass_after_the_deadline_fails() { + let mut f = World::new(); + let started = Instant::now(); + let err = Scenario::new("late") + .then("answers too late", Duration::from_millis(100), |f, _c| { + let diag = f.probe_diag().clone(); + Box::pin(async move { + diag.record("GET /slow", Class::Transient, "still syncing".to_string()); + tokio::time::sleep(Duration::from_secs(5)).await; + Some(Ok(())) + }) + }) + .run(&mut f) + .await + .unwrap_err(); + + assert!( + started.elapsed() < Duration::from_secs(1), + "elapsed: {:?}", + started.elapsed() + ); + let chain = format!("{err:#}"); + assert!(chain.contains("timed out"), "got: {chain}"); + assert!( + chain.contains("last probe error: still syncing"), + "got: {chain}" + ); + } + #[tokio::test(flavor = "multi_thread")] async fn timeout_message_carries_the_last_probe_error() { let mut f = World::new(); From c057d93249e5586416032d0f6bdd8026228dd9d6 Mon Sep 17 00:00:00 2001 From: richardanyalai Date: Wed, 14 Oct 2026 11:33:18 +0200 Subject: [PATCH 3/4] chore(ci): added cargo-audit's ignore file CI runs both cargo-audit and cargo-deny, and each reads its own advisory ignore list. The deny.toml header claimed cargo-deny was the only gate, so an ignore added there alone would still fail Cargo Audit. The new .cargo/audit.toml matches decman's layout, and the deny.toml header now says which file each kind of ignore belongs in. --- .cargo/audit.toml | 9 +++++++++ deny.toml | 17 +++++++++++------ 2 files changed, 20 insertions(+), 6 deletions(-) create mode 100644 .cargo/audit.toml diff --git a/.cargo/audit.toml b/.cargo/audit.toml new file mode 100644 index 0000000..aae46f9 --- /dev/null +++ b/.cargo/audit.toml @@ -0,0 +1,9 @@ +# cargo-audit reads Cargo.lock and flags every advisory in it, even for a +# crate behind a disabled feature that never compiles. deny.toml explains +# which file an ignore belongs in. +# +# cargo-audit takes plain IDs, so write each ignore's reason in a comment +# above it. + +[advisories] +ignore = [] diff --git a/deny.toml b/deny.toml index 16c71ec..d33cbeb 100644 --- a/deny.toml +++ b/deny.toml @@ -1,10 +1,15 @@ -# cargo-deny is the only RustSec advisories gate in the Bitsafe baseline, so -# this is the only advisory-exception list the repo has. +# CI checks RustSec advisories with two tools. cargo-deny reads this file, +# and cargo-audit reads .cargo/audit.toml. # -# Every ignored RUSTSEC ID carries a `reason` that names the crate that brings -# it in, why no fix exists, and what would make the ignore unsafe. Do not add -# an entry for a finding behind a disabled feature: cargo-deny already skips -# it. Confirm with `cargo tree -i `. +# cargo-deny skips a crate behind a disabled feature. cargo-audit reads +# Cargo.lock, so it still flags that crate. An ignore therefore goes in: +# * both files, for a finding in code the build compiles; +# * only .cargo/audit.toml, for a finding behind a disabled feature. +# `cargo tree -i ` prints nothing for such a crate. +# +# Every ignore names the crate that brings the advisory in, why no fix +# exists, and what would make the ignore unsafe. In this file, write it as +# { id = "RUSTSEC-...", reason = "..." }. [advisories] ignore = [] From ac36c0e1e0574ac95d88c0445d0059ef569db5ba Mon Sep 17 00:00:00 2001 From: richardanyalai Date: Wed, 14 Oct 2026 11:38:14 +0200 Subject: [PATCH 4/4] fix(core): dropped the padding from step labels The When and Then labels carried a trailing space so the step log lines lined up. The failure log and the error context added their own separator after the label, so a failed When step read `WHEN "..."`. The labels are now plain words, and only the step log line pads them to five characters. The log output is unchanged. --- src/scenario.rs | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/src/scenario.rs b/src/scenario.rs index 838205c..4f56540 100644 --- a/src/scenario.rs +++ b/src/scenario.rs @@ -46,8 +46,8 @@ impl StepKind { fn label(self) -> &'static str { match self { StepKind::Given => "GIVEN", - StepKind::When => "WHEN ", - StepKind::Then => "THEN ", + StepKind::When => "WHEN", + StepKind::Then => "THEN", } } } @@ -172,7 +172,7 @@ impl Scenario { for step in &mut self.steps { match step { Step::Immediate { kind, name, body } => { - info!(step_kind = ?kind, step_name = %name, " {} {}", kind.label(), name); + info!(step_kind = ?kind, step_name = %name, " {:<5} {}", kind.label(), name); let fut = body(f, &mut self.ctx); match fut.await { Ok(()) => {} @@ -196,7 +196,7 @@ impl Scenario { probe, } => { let kind = StepKind::Then; - info!(step_kind = ?kind, step_name = %name, " {} {}", kind.label(), name); + info!(step_kind = ?kind, step_name = %name, " {:<5} {}", kind.label(), name); let step_start = Instant::now(); let step_deadline = tokio::time::Instant::from_std(step_start) + *deadline; f.probe_diag().reset(); @@ -358,10 +358,7 @@ mod tests { let chain = format!("{err:#}"); assert!(chain.contains("scenario \"my scn\" failed"), "got: {chain}"); - assert!( - chain.contains("WHEN") && chain.contains("the failing step"), - "got: {chain}" - ); + assert!(chain.contains("WHEN \"the failing step\""), "got: {chain}"); assert!(chain.contains("boom"), "got: {chain}"); }