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/.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..907bb0d 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,159 @@ -# 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. + +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 +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..d33cbeb --- /dev/null +++ b/deny.toml @@ -0,0 +1,15 @@ +# CI checks RustSec advisories with two tools. cargo-deny reads this file, +# and cargo-audit reads .cargo/audit.toml. +# +# 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 = [] 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..4f56540 --- /dev/null +++ b/src/scenario.rs @@ -0,0 +1,734 @@ +//! 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. 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, + { + 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, " {:<5} {}", 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, " {:<5} {}", 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 { + // 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"))); + } + Ok(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(timed_out(name, *deadline, f.probe_diag())); + } + 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(()) + } +} + +/// 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::{ + 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 \"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}"); + } + + /// 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(); + 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(); + } +}