diff --git a/.github/workflows/nightly-tsan.yml b/.github/workflows/nightly-tsan.yml new file mode 100644 index 00000000..044f568d --- /dev/null +++ b/.github/workflows/nightly-tsan.yml @@ -0,0 +1,55 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2026 Praxis Contributors + +name: Nightly ThreadSanitizer + +# The engine is shared across threads behind `Arc` and mutated while +# requests are in flight. PR CI runs the seeded stress test on a +# multi-threaded runtime; this job rebuilds it under ThreadSanitizer so +# a data race is a red build rather than a wrong plugin count. +# +# Nightly only: TSan needs a nightly compiler, and the rebuild is too +# slow for every pull request. `workflow_dispatch` is here so a race +# report can be reproduced without waiting for the cron. + +on: + schedule: + - cron: "0 6 * * *" + workflow_dispatch: + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +permissions: {} + +env: + CARGO_TERM_COLOR: always + RUST_BACKTRACE: 1 + CARGO_INCREMENTAL: 0 + TSAN_OPTIONS: halt_on_error=1 + +jobs: + tsan-engine-stress: + name: engine concurrency under TSan + runs-on: ubuntu-24.04 + timeout-minutes: 45 + permissions: + contents: read + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0 + with: + path: | + ~/.cargo/bin + ~/.cargo/registry/index + ~/.cargo/registry/cache + ~/.cargo/git/db + target + key: ${{ runner.os }}-cargo-tsan-${{ hashFiles('**/Cargo.lock', 'rust-toolchain.toml') }} + restore-keys: ${{ runner.os }}-cargo-tsan- + - run: rustup toolchain install nightly --component rust-src --profile minimal + - run: rustup target add x86_64-unknown-linux-gnu --toolchain nightly + - run: make test-tsan diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 19130c4a..023dfbcc 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -141,3 +141,33 @@ Enforcing one of the allowed groups is welcome as a focused change, one lint at time, separate from feature work. `docs/lints.md` is worth reading first: it records which lints clippy reports as machine-fixable but cannot actually fix, and where a lint's suggested rewrite is worse than the code it replaces. + +## Multi-threaded Tokio tests + +`#[tokio::test]` defaults to `current_thread`. Tasks yield at `.await` but never +run on two OS threads at the same time, so a load and a store that have no await +between them cannot overlap. + +Tests that exercise registration, unregister, hot reload (`load_config` / +`from_config`), or route-cache fill and invalidation use +`#[tokio::test(flavor = "multi_thread")]`, including when the body is a single +task. The flavor is set by the surface under test so a later spawn starts from +the runtime a host uses. A single-task test of those APIs does not itself +create overlap. + +Overlap is asserted in `crates/ppe-core/tests/engine_concurrency.rs`. + +```rust +#[tokio::test(flavor = "multi_thread")] +async fn register_while_other_tasks_invoke() { /* ... */ } +``` + +Sequential tests that do not touch those surfaces stay on `current_thread`. + +A seeded stress test lives in `crates/ppe-core/tests/engine_concurrency.rs`. +Replay a failure with `PPE_STRESS_SEED`. Nightly CI runs that test under +ThreadSanitizer (`make test-tsan`; Linux only, the target is hardcoded). The +`Release` / `Acquire` pairing `mutate_runtime` describes is documented by the +extracted loom model in `crates/ppe-core/tests/loom_generation_snapshot.rs`. +That model is not wired to `engine.rs`. Default `cargo test` does not compile +it; run `RUSTFLAGS='--cfg loom' cargo test -p praxis-policy-core --test loom_generation_snapshot`. diff --git a/Cargo.lock b/Cargo.lock index a9850cfd..d44388cf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1147,6 +1147,21 @@ dependencies = [ "slab", ] +[[package]] +name = "generator" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3b854b0e584ead1a33f18b2fcad7cf7be18b3875c78816b753639aa501513ae" +dependencies = [ + "cc", + "cfg-if", + "libc", + "log", + "rustversion", + "windows-link", + "windows-result", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -1836,12 +1851,34 @@ dependencies = [ "logos-codegen", ] +[[package]] +name = "loom" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "419e0dc8046cb947daa77eb95ae174acfbddb7673b4151f56d1eed8e93fbfaca" +dependencies = [ + "cfg-if", + "generator", + "scoped-tls", + "tracing", + "tracing-subscriber", +] + [[package]] name = "lru" version = "0.18.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5d2f2f9b4ba7e6b24d95e7e899329d35be83bcded72c8540cdd5368932d1d90a" +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "matchit" version = "0.8.4" @@ -1988,6 +2025,15 @@ dependencies = [ "serde", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num" version = "0.4.3" @@ -2443,6 +2489,7 @@ dependencies = [ "chrono", "hashbrown 0.17.1", "http", + "loom", "praxis-policy-core", "praxis-policy-orchestration", "serde", @@ -3066,6 +3113,12 @@ dependencies = [ "serde_json", ] +[[package]] +name = "scoped-tls" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1cf6437eb19a8f4a6cc0f7dca544973b0b78843adbfeb3683d1a94a0024a294" + [[package]] name = "scopeguard" version = "1.2.0" @@ -3235,6 +3288,15 @@ dependencies = [ "keccak", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "2.0.1" @@ -3519,6 +3581,15 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + [[package]] name = "time" version = "0.3.55" @@ -3738,6 +3809,36 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", ] [[package]] @@ -3890,6 +3991,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "version_check" version = "0.9.5" diff --git a/Cargo.toml b/Cargo.toml index 05d14e03..523611b5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -108,6 +108,9 @@ paste = "1" futures = "0.3" hashbrown = "0.17" arc-swap = "1.9" +# Exhaustive scheduler for a tiny model of the generation / snapshot +# pairing. Dev-only, and only compiled when a crate enables `--cfg loom`. +loom = "0.7" wildmatch = "2" rmp-serde = "1" serde_bytes = "0.11" @@ -189,7 +192,7 @@ redundant_lifetimes = "deny" # A stale feature name in a `cfg` predicate is otherwise only a warning, which # is how a rename can silently disable a gated export. Denying it turns that # into a build error at the moment the rename happens. -unexpected_cfgs = "deny" +unexpected_cfgs = { level = "deny", check-cfg = ['cfg(loom)'] } # Closed from the import backlog: measured at zero violations across # --workspace --all-targets --all-features, so enforcing costs no code change. unused_extern_crates = "deny" diff --git a/Makefile b/Makefile index f8ad2822..9dc1602a 100644 --- a/Makefile +++ b/Makefile @@ -40,6 +40,7 @@ help: @echo "" @echo "Test:" @echo " test Run all workspace tests" + @echo " test-tsan Engine concurrency stress under ThreadSanitizer (nightly)" @echo "" @echo "Supply chain & coverage:" @echo " audit cargo deny check (advisories, licenses, bans, sources)" @@ -159,6 +160,19 @@ test: @$(CARGO) test --workspace @$(CARGO) test --workspace --all-features +# ThreadSanitizer on the engine concurrency stress test. Needs nightly, a +# Linux target, and an instrumented libstd (`-Zbuild-std`). The sanitizer +# does not run on the pinned stable toolchain or on macOS. `--test-threads=1` +# keeps TSan's own reports from overlapping. +.PHONY: test-tsan +test-tsan: + @echo "ThreadSanitizer: praxis-policy-core engine concurrency ..." + @RUSTFLAGS="-Zsanitizer=thread" CARGO_INCREMENTAL=0 \ + $(CARGO) +$(NIGHTLY) test -Zbuild-std=std,panic_abort \ + -p praxis-policy-core --test engine_concurrency \ + --target x86_64-unknown-linux-gnu -- --test-threads=1 + @echo "test-tsan passed" + # ============================================================================= # Supply chain & coverage # ============================================================================= diff --git a/crates/ppe-core/Cargo.toml b/crates/ppe-core/Cargo.toml index 8b39f99c..1cc182fe 100644 --- a/crates/ppe-core/Cargo.toml +++ b/crates/ppe-core/Cargo.toml @@ -77,5 +77,11 @@ praxis-policy-core = { path = ".", features = ["test-util"] } tokio = { workspace = true, features = ["test-util"] } tokio-util = { workspace = true } +# loom is pulled only under `--cfg loom` so default `cargo test` does not +# compile `generator` / `cc`. Run the model with +# `RUSTFLAGS='--cfg loom' cargo test -p praxis-policy-core --test loom_generation_snapshot`. +[target.'cfg(loom)'.dev-dependencies] +loom = { workspace = true } + [lints] workspace = true diff --git a/crates/ppe-core/src/engine.rs b/crates/ppe-core/src/engine.rs index c6f5c71d..43957621 100644 --- a/crates/ppe-core/src/engine.rs +++ b/crates/ppe-core/src/engine.rs @@ -3699,7 +3699,7 @@ mod tests { assert_eq!(result.violation.as_ref().unwrap().code, "denied"); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_has_hooks_for() { let mgr = PolicyEngine::default(); assert!(!mgr.has_hooks_for("test_hook")); @@ -3891,7 +3891,7 @@ mod tests { ); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_unregister() { let mgr = PolicyEngine::default(); let config = make_config("removable", 10, PluginMode::Sequential); @@ -3911,7 +3911,7 @@ mod tests { /// that runtime registration is safe alongside invocations — the whole /// point of the `ArcSwap`-based snapshot redesign. Before this fix, /// `register_*` was `&mut self`, so this pattern wouldn't even compile. - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_manager_arc_shareable_with_concurrent_dispatch_and_registration() { use std::sync::atomic::{AtomicUsize, Ordering}; @@ -5833,7 +5833,7 @@ plugins: /// segment-boundary rows mirror the host router's own suite: a prefix that /// matches a path only where a `/` follows it, and a trailing slash on the /// declared prefix that changes nothing. - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_routing_works_for_all_entity_types() { register_fixture_hooks(); use std::sync::Arc as StdArc; @@ -6399,7 +6399,7 @@ routes: ); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_from_config_creates_manager() { register_fixture_hooks(); let yaml = r#" @@ -6426,7 +6426,7 @@ engine_settings: assert!(mgr.has_hooks_for("test_hook")); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_from_config_invokes_correctly() { register_fixture_hooks(); let yaml = r#" @@ -6460,7 +6460,7 @@ plugins: assert_eq!(result.violation.as_ref().unwrap().code, "denied"); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_from_config_unknown_kind_rejected() { register_fixture_hooks(); let yaml = r#" @@ -6481,7 +6481,7 @@ plugins: } } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_from_config_multiple_plugins() { register_fixture_hooks(); let yaml = r#" @@ -6525,7 +6525,7 @@ plugins: // -- Routing cache tests -- - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_routing_cache_populated_on_first_invoke() { register_fixture_hooks(); let yaml = r#" @@ -6727,7 +6727,7 @@ routes: ); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_routing_cache_different_entities_separate() { register_fixture_hooks(); let yaml = r#" @@ -6778,7 +6778,7 @@ routes: assert_eq!(mgr.routing_cache_size(), 2); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_routing_cache_cleared() { register_fixture_hooks(); let yaml = r#" @@ -6816,7 +6816,7 @@ routes: assert_eq!(mgr.routing_cache_size(), 0); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_unregister_invalidates_routing_cache() { register_fixture_hooks(); let yaml = r#" @@ -6885,7 +6885,7 @@ routes: assert_eq!(mgr.routing_cache_size(), 0); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_routing_cache_rejects_inserts_at_capacity() { register_fixture_hooks(); // Cap of 2 — verifies bound holds AND uncached requests still resolve correctly. @@ -6959,7 +6959,7 @@ routes: assert_eq!(mgr.routing_cache_size(), 1); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_register_handler_invalidates_routing_cache() { register_fixture_hooks(); let yaml = r#" @@ -7003,7 +7003,7 @@ routes: assert_eq!(mgr.routing_cache_size(), 0); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_routing_cache_scope_creates_separate_entries() { register_fixture_hooks(); let yaml = r#" @@ -7056,7 +7056,7 @@ routes: // -- Override instance tests -- - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_route_override_creates_new_instance() { register_fixture_hooks(); let yaml = r#" @@ -7293,7 +7293,7 @@ routes: /// open DB connections / file handles / network clients on init don't /// run with default state. Uses a tracking factory whose plugin /// increments a counter inside its `initialize()`. - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_route_override_initializes_new_instance() { register_fixture_hooks(); use std::sync::atomic::{AtomicUsize, Ordering}; @@ -7612,7 +7612,7 @@ routes: /// config) must not silently disable the plugin for every other route /// using the base config — config is part of the failure surface, and /// per-route blast radius is the point of having overrides. - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_route_override_circuit_breaker_isolated_from_base() { register_fixture_hooks(); struct ErrorOnInvokeFactory; @@ -7683,7 +7683,7 @@ routes: ); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_register_factory_then_load_config() { register_fixture_hooks(); let yaml = r#" @@ -7741,7 +7741,7 @@ engine_settings: } } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_routing_disabled_fires_all_plugins() { register_fixture_hooks(); // Same plugins under hook dispatch: all fire regardless of entity @@ -7782,7 +7782,7 @@ plugins: assert!(!result.continue_processing); // denier fires (all plugins active) } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_routing_no_meta_fires_all_plugins() { register_fixture_hooks(); // Routing enabled but no meta on extensions → fallback to all @@ -8047,7 +8047,7 @@ routes: /// Verifies that a handler that genuinely `.await`s gets driven /// to completion before its result is observed. - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_async_handler_registers_and_invokes() { let mgr = PolicyEngine::default(); let counter = Arc::new(std::sync::atomic::AtomicU64::new(0)); @@ -8083,7 +8083,7 @@ routes: /// genuinely awaits (`AsyncCounterPlugin`) co-register on the same /// hook via the same `register_handler` call. Both run in priority /// order. - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_mixed_sync_and_async_handlers_in_same_hook() { let mgr = PolicyEngine::default(); let counter = Arc::new(std::sync::atomic::AtomicU64::new(0)); @@ -8703,7 +8703,7 @@ routes: (entity_type, names.remove(0)) } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn many_http_paths_matching_one_route_share_one_cache_entry() { let (mgr, ledger) = recording_engine(HTTP_ROUTES_YAML).await; @@ -9242,7 +9242,7 @@ routes: /// A config replacement rebuilds the snapshot, so the answer follows the /// config it was derived from. A stale answer would warn about routes that /// are gone, or stay silent about ones that arrived. - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn a_reload_recomputes_which_routes_declare_authentication() { // A load merges its plugins into the registry, so each generation names // its own rather than colliding with the one before it. diff --git a/crates/ppe-core/tests/engine_concurrency.rs b/crates/ppe-core/tests/engine_concurrency.rs new file mode 100644 index 00000000..db20f074 --- /dev/null +++ b/crates/ppe-core/tests/engine_concurrency.rs @@ -0,0 +1,716 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Praxis Contributors + +//! Concurrent invoke against concurrent mutation of `PolicyEngine`. +//! +//! The engine is shared behind `Arc` the way a host shares it: request +//! threads `invoke_*` while other threads register, unregister, reload +//! config, and annotate routes. These tests run that shape on real OS +//! threads and check two things a single-threaded Tokio runtime cannot: +//! +//! 1. A successful registration is still visible afterwards (lost-update). +//! 2. An invoke that overlaps a snapshot swap sees one complete lineup +//! (paired plugins from a single `load_config` both fire, or neither +//! does), not a mix of two snapshots. +//! +//! Route-cache fill needs `dispatch: policy`. Plugin dispatch in this +//! crate needs `dispatch: hooks`. The two stress tests use one fixture +//! each; mixing them in one engine leaves the invoke arm firing nothing. +//! +//! The stress tests are seeded. Override with `PPE_STRESS_SEED`, +//! `PPE_STRESS_OPS`, `PPE_STRESS_INVOKERS`, and `PPE_STRESS_MUTATORS`. +//! A failure prints the seed so the same schedule can be replayed. + +#![allow( + missing_docs, + clippy::expect_used, + clippy::panic, + clippy::print_stderr, + clippy::unwrap_used, + reason = "test and example code" +)] + +use std::collections::{HashMap, HashSet}; +use std::env; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Barrier, Mutex}; +use std::thread; + +use async_trait::async_trait; +use praxis_policy_core::config::parse_config; +use praxis_policy_core::context::PluginContext; +use praxis_policy_core::engine::PolicyEngine; +use praxis_policy_core::error::PluginError; +use praxis_policy_core::executor::erase_result; +use praxis_policy_core::extensions::MetaExtension; +use praxis_policy_core::factory::{PluginFactory, PluginInstance}; +use praxis_policy_core::hooks::adapter::TypedHandlerAdapter; +use praxis_policy_core::hooks::metadata::{HookMetadata, register_hook_metadata}; +use praxis_policy_core::hooks::payload::{Extensions, PluginPayload}; +use praxis_policy_core::hooks::trait_def::{HookHandler, HookTypeDef, PluginResult}; +use praxis_policy_core::plugin::{OnError, Plugin, PluginConfig, PluginMode}; +use praxis_policy_core::registry::AnyHookHandler; + +const HOOK: &str = "stress_hook"; +const BASE_PLUGIN: &str = "base"; +const TOOL: &str = "stress_tool"; +const KIND: &str = "stress/allow"; + +const DEFAULT_SEED: u64 = 0xC0_FF_EE; +const DEFAULT_OPS: u32 = 128; +const DEFAULT_INVOKERS: usize = 4; +const DEFAULT_MUTATORS: usize = 4; + +type Ledger = Arc>>>; + +#[derive(Debug, Clone)] +struct StressPayload { + invoke_id: u64, +} +praxis_policy_core::impl_plugin_payload!(StressPayload); + +struct StressHook; +impl HookTypeDef for StressHook { + type Payload = StressPayload; + type Result = PluginResult; + const NAME: &'static str = HOOK; +} + +struct StressPlugin { + cfg: PluginConfig, + ledger: Ledger, +} + +impl StressPlugin { + fn new(cfg: PluginConfig, ledger: Ledger) -> Arc { + Arc::new(Self { cfg, ledger }) + } + + fn record(&self, invoke_id: u64) { + self.ledger + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .entry(invoke_id) + .or_default() + .push(self.cfg.name.clone()); + } +} + +#[async_trait] +impl Plugin for StressPlugin { + fn config(&self) -> &PluginConfig { + &self.cfg + } +} + +impl HookHandler for StressPlugin { + async fn handle( + &self, + payload: &StressPayload, + _extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> PluginResult { + self.record(payload.invoke_id); + PluginResult::allow() + } +} + +#[async_trait] +impl AnyHookHandler for StressPlugin { + async fn invoke( + &self, + payload: &dyn PluginPayload, + _extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> Result, Box> { + if let Some(p) = payload.as_any().downcast_ref::() { + self.record(p.invoke_id); + } + Ok(erase_result(PluginResult::::allow())) + } + + fn hook_type_name(&self) -> &'static str { + StressHook::NAME + } +} + +struct StressFactory { + ledger: Ledger, +} + +impl PluginFactory for StressFactory { + fn create(&self, config: &PluginConfig) -> Result> { + let plugin = StressPlugin::new(config.clone(), Arc::clone(&self.ledger)); + let handler: Arc = Arc::new(TypedHandlerAdapter::< + StressHook, + StressPlugin, + >::new(Arc::clone(&plugin))); + Ok(PluginInstance { + plugin, + handlers: vec![(StressHook::NAME, handler)], + }) + } +} + +fn plugin_config(name: &str) -> PluginConfig { + PluginConfig { + name: name.to_owned(), + kind: KIND.to_owned(), + description: None, + author: None, + version: None, + hooks: vec![HOOK.to_owned()], + mode: PluginMode::Sequential, + priority: 10, + on_error: OnError::Fail, + capabilities: Default::default(), + tags: Vec::new(), + conditions: Vec::new(), + config: None, + } +} + +fn register_stress_hook() { + register_hook_metadata(StressHook::NAME, HookMetadata::permissive()); +} + +fn tool_extensions() -> Extensions { + Extensions { + meta: Some(Arc::new(MetaExtension { + entity_type: Some("tool".into()), + entity_name: Some(TOOL.into()), + ..Default::default() + })), + ..Default::default() + } +} + +fn new_ledger() -> Ledger { + Arc::new(Mutex::new(HashMap::new())) +} + +fn new_engine(ledger: &Ledger) -> Arc { + register_stress_hook(); + let engine = Arc::new(PolicyEngine::default()); + engine.register_factory( + KIND, + Box::new(StressFactory { + ledger: Arc::clone(ledger), + }), + ); + engine +} + +fn bootstrap_hooks(ledger: &Ledger) -> Arc { + let engine = new_engine(ledger); + let yaml = format!( + " +engine_settings: + dispatch: hooks +plugins: + - name: {BASE_PLUGIN} + kind: {KIND} + hooks: [{HOOK}] + mode: sequential + priority: 10 +" + ); + let config = parse_config(&yaml).expect("hooks bootstrap config must parse"); + engine + .load_config(config) + .expect("hooks bootstrap load_config must succeed"); + engine +} + +fn bootstrap_policy(ledger: &Ledger) -> Arc { + let engine = new_engine(ledger); + let yaml = format!( + " +engine_settings: + dispatch: policy +plugins: + - name: {BASE_PLUGIN} + kind: {KIND} + hooks: [{HOOK}] + mode: sequential +routes: + - tool: {TOOL} +" + ); + let config = parse_config(&yaml).expect("policy bootstrap config must parse"); + engine + .load_config(config) + .expect("policy bootstrap load_config must succeed"); + engine +} + +fn env_u64(name: &str, default: u64) -> u64 { + env::var(name) + .ok() + .map(|raw| { + raw.parse::().unwrap_or_else(|_| { + panic!("{name}={raw:?} is not a u64"); + }) + }) + .unwrap_or(default) +} + +fn env_usize(name: &str, default: usize) -> usize { + let default = u64::try_from(default).expect("fits u64"); + usize::try_from(env_u64(name, default)).expect("fits usize") +} + +/// `SplitMix64`. One stream per mutator (`seed ^ mix(mutator_id)`) so the +/// schedule is a function of the seed alone. +struct SplitMix64(u64); + +impl SplitMix64 { + fn from_seed(seed: u64, stream: u64) -> Self { + Self(seed ^ stream.wrapping_mul(0x9E37_79B9_7F4A_7C15)) + } + + fn next_u64(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + + fn choose(&mut self, n: u32) -> u32 { + let bound = u64::from(n.max(1)); + u32::try_from(self.next_u64() % bound).expect("bound is a u32") + } +} + +fn register_named( + engine: &PolicyEngine, + ledger: &Ledger, + name: &str, +) -> Result<(), Box> { + let cfg = plugin_config(name); + engine + .register_handler::(StressPlugin::new(cfg.clone(), Arc::clone(ledger)), cfg) +} + +fn take_fired(ledger: &Ledger, invoke_id: u64) -> Vec { + ledger + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(&invoke_id) + .unwrap_or_default() +} + +/// `load_config` publishes `*-L` and `*-R` in one snapshot. One invoke +/// that sees only one of the pair mixed two snapshots. An annotation is +/// the entire lineup, so that path is exempt. +fn assert_coherent_lineup(fired: &[String], seed: u64, invoke_id: u64) { + assert!( + !fired.is_empty(), + "invoke fired no plugins; seed={seed} invoke={invoke_id}" + ); + if fired.iter().any(|name| name.starts_with("ann-")) { + assert_eq!( + fired.len(), + 1, + "an annotation is the whole lineup; seed={seed} invoke={invoke_id} \ + fired={fired:?}" + ); + return; + } + for name in fired { + if let Some(stem) = name.strip_suffix("-L") { + let right = format!("{stem}-R"); + assert!( + fired.iter().any(|n| n == &right), + "torn snapshot: {name} fired without {right}; seed={seed} \ + invoke={invoke_id} fired={fired:?}" + ); + } + if let Some(stem) = name.strip_suffix("-R") { + let left = format!("{stem}-L"); + assert!( + fired.iter().any(|n| n == &left), + "torn snapshot: {name} fired without {left}; seed={seed} \ + invoke={invoke_id} fired={fired:?}" + ); + } + } +} + +fn stress_knobs() -> (u64, u64, usize, usize) { + let seed = env_u64("PPE_STRESS_SEED", DEFAULT_SEED); + let ops = env_u64("PPE_STRESS_OPS", u64::from(DEFAULT_OPS)); + let invokers = env_usize("PPE_STRESS_INVOKERS", DEFAULT_INVOKERS).max(1); + let mutators = env_usize("PPE_STRESS_MUTATORS", DEFAULT_MUTATORS).max(1); + (seed, ops, invokers, mutators) +} + +/// Distinct names, one per thread, all `register_handler` calls overlapping +/// at a barrier. Last-writer-wins on the snapshot would drop at least one. +#[test] +fn concurrent_writers_do_not_drop_registrations() { + const N: usize = 8; + let ledger = new_ledger(); + let engine = bootstrap_hooks(&ledger); + let barrier = Arc::new(Barrier::new(N)); + let mut joins = Vec::with_capacity(N); + for i in 0..N { + let engine = Arc::clone(&engine); + let ledger = Arc::clone(&ledger); + let barrier = Arc::clone(&barrier); + joins.push(thread::spawn(move || { + let name = format!("barrier-{i}"); + barrier.wait(); + register_named(&engine, &ledger, &name).expect("register"); + name + })); + } + let names: Vec = joins + .into_iter() + .map(|j| j.join().expect("writer thread")) + .collect(); + + let missing: Vec<&str> = names + .iter() + .map(String::as_str) + .filter(|name| engine.get_plugin(name).is_none()) + .collect(); + assert!( + missing.is_empty(), + "lost update: register returned Ok but the snapshot is missing {missing:?}; \ + present={:?}", + engine.plugin_names() + ); + assert!( + engine.get_plugin(BASE_PLUGIN).is_some(), + "a concurrent register must not drop the bootstrap plugin" + ); +} + +/// N invoke tasks against M mutator OS threads under `dispatch: hooks`, +/// so registered plugins actually run. Invokers loop until mutators +/// finish; joining the OS threads happens on a blocking pool so a tokio +/// worker is not stuck in `JoinHandle::join` for the whole mutation. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn stress_invoke_against_concurrent_mutation() { + let (seed, ops, invokers, mutators) = stress_knobs(); + eprintln!( + "engine concurrency stress seed={seed} ops={ops} invokers={invokers} \ + mutators={mutators}" + ); + + let ledger = new_ledger(); + let engine = bootstrap_hooks(&ledger); + engine.initialize().await.expect("initialize"); + let generation_at_start = engine.config_generation(); + let done = Arc::new(AtomicBool::new(false)); + let next_invoke = Arc::new(AtomicU64::new(1)); + let invoke_ok = Arc::new(AtomicU64::new(0)); + + let mut invoke_joins = Vec::with_capacity(invokers); + for i in 0..invokers { + let engine = Arc::clone(&engine); + let ledger = Arc::clone(&ledger); + let done = Arc::clone(&done); + let next_invoke = Arc::clone(&next_invoke); + let invoke_ok = Arc::clone(&invoke_ok); + invoke_joins.push(tokio::spawn(async move { + while !done.load(Ordering::Acquire) { + let invoke_id = next_invoke.fetch_add(1, Ordering::Relaxed); + let payload: Box = Box::new(StressPayload { invoke_id }); + let (result, _) = engine + .invoke_by_name(HOOK, payload, tool_extensions(), None) + .await; + assert!( + result.continue_processing, + "invoke must see a coherent allow snapshot; \ + seed={seed} invoker={i} invoke={invoke_id} denied={:?}", + result.violation + ); + let fired = take_fired(&ledger, invoke_id); + assert_coherent_lineup(&fired, seed, invoke_id); + invoke_ok.fetch_add(1, Ordering::Relaxed); + tokio::task::yield_now().await; + } + })); + } + + let mut mutator_joins = Vec::with_capacity(mutators); + for mutator_id in 0..mutators { + let engine = Arc::clone(&engine); + let ledger = Arc::clone(&ledger); + mutator_joins.push(thread::spawn(move || { + mutator_loop( + &engine, + &ledger, + seed, + u64::try_from(mutator_id).expect("fits u64"), + ops, + true, + ) + })); + } + + let outcomes = tokio::task::spawn_blocking(move || { + mutator_joins + .into_iter() + .map(|j| j.join().expect("mutator thread")) + .collect::>() + }) + .await + .expect("join mutators"); + done.store(true, Ordering::Release); + + let mut expected: HashSet = HashSet::new(); + expected.insert(BASE_PLUGIN.to_owned()); + let mut published = 0_u64; + for outcome in outcomes { + expected.extend(outcome.live); + published += outcome.published; + } + for join in invoke_joins { + join.await.expect("invoker task"); + } + + assert!( + invoke_ok.load(Ordering::Relaxed) > 0, + "invokers must overlap the mutation stream; seed={seed}" + ); + + engine.remove_route_annotation("tool", TOOL, None, HOOK); + + let present: HashSet = engine.plugin_names().into_iter().collect(); + let missing: Vec<&str> = expected + .iter() + .map(String::as_str) + .filter(|name| !present.contains(*name)) + .collect(); + let unexpected: Vec<&str> = present + .iter() + .map(String::as_str) + .filter(|name| mutator_owned_name(name) && !expected.contains(*name)) + .collect(); + assert!( + missing.is_empty() && unexpected.is_empty(), + "lost update under concurrent mutation; seed={seed} missing={missing:?} \ + unexpected={unexpected:?} expected={expected:?} present={present:?}" + ); + + let generation = engine.config_generation(); + // `published` is a lower bound: every counted op published a snapshot. + // A no-op `unregister` is not counted (generation may still bump). A + // no-op `remove_route_annotation` is counted because `mutate_runtime` + // always stores and bumps, even when the key is absent. + assert!( + generation >= generation_at_start + published, + "generation must not go backwards and must count every published \ + snapshot; seed={seed} start={generation_at_start} end={generation} \ + published={published}" + ); +} + +/// Route cache fill is a `dispatch: policy` property. Mutate, quiesce, +/// then miss and refill from the live snapshot. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn stress_route_cache_refills_after_concurrent_mutation() { + let (seed, ops, _, mutators) = stress_knobs(); + eprintln!("engine cache stress seed={seed} ops={ops} mutators={mutators}"); + + let ledger = new_ledger(); + let engine = bootstrap_policy(&ledger); + engine.initialize().await.expect("initialize"); + let generation_at_start = engine.config_generation(); + + let mut mutator_joins = Vec::with_capacity(mutators); + for mutator_id in 0..mutators { + let engine = Arc::clone(&engine); + let ledger = Arc::clone(&ledger); + mutator_joins.push(thread::spawn(move || { + mutator_loop( + &engine, + &ledger, + seed, + u64::try_from(mutator_id).expect("fits u64"), + ops, + false, + ) + })); + } + + let outcomes = tokio::task::spawn_blocking(move || { + mutator_joins + .into_iter() + .map(|j| j.join().expect("mutator thread")) + .collect::>() + }) + .await + .expect("join mutators"); + + let mut expected: HashSet = HashSet::new(); + expected.insert(BASE_PLUGIN.to_owned()); + let mut published = 0_u64; + for outcome in outcomes { + expected.extend(outcome.live); + published += outcome.published; + } + + engine.remove_route_annotation("tool", TOOL, None, HOOK); + + let present: HashSet = engine.plugin_names().into_iter().collect(); + let missing: Vec<&str> = expected + .iter() + .map(String::as_str) + .filter(|name| !present.contains(*name)) + .collect(); + assert!( + missing.is_empty(), + "lost update under concurrent mutation; seed={seed} missing={missing:?} \ + present={present:?}" + ); + + let generation = engine.config_generation(); + assert!( + generation >= generation_at_start + published, + "generation must not go backwards; seed={seed} start={generation_at_start} \ + end={generation} published={published}" + ); + + engine.clear_routing_cache(); + assert_eq!(engine.routing_cache_size(), 0); + let payload: Box = Box::new(StressPayload { invoke_id: 0 }); + let (result, _) = engine + .invoke_by_name(HOOK, payload, tool_extensions(), None) + .await; + assert!( + result.continue_processing, + "quiesced invoke must still allow; seed={seed}" + ); + assert!( + engine.routing_cache_size() >= 1, + "routing is on, so a tool invoke must memoize the resolved lineup; \ + seed={seed} cache={}", + engine.routing_cache_size() + ); +} + +fn mutator_owned_name(name: &str) -> bool { + matches!(name.as_bytes().first(), Some(b'm' | b'r')) && name.contains('-') +} + +struct MutatorOutcome { + live: HashSet, + published: u64, +} + +fn mutator_loop( + engine: &PolicyEngine, + ledger: &Ledger, + seed: u64, + mutator_id: u64, + ops: u64, + hooks: bool, +) -> MutatorOutcome { + let mut rng = SplitMix64::from_seed(seed, mutator_id + 1); + let mut live = HashSet::new(); + let mut owned: Vec = Vec::new(); + let mut published = 0_u64; + let mut next_id = 0_u64; + + for _ in 0..ops { + match rng.choose(5) { + 0 => { + let name = format!("m{mutator_id}-{next_id}"); + next_id += 1; + if register_named(engine, ledger, &name).is_ok() { + live.insert(name.clone()); + owned.push(name); + published += 1; + } + }, + 1 => { + if let Some(name) = owned.pop() + && engine.unregister(&name).is_some() + { + live.remove(&name); + published += 1; + } + }, + 2 => { + let name = format!("ann-{mutator_id}-{next_id}"); + next_id += 1; + let cfg = plugin_config(&name); + engine.annotate_route( + "tool", + TOOL, + None, + HOOK, + StressPlugin::new(cfg.clone(), Arc::clone(ledger)), + cfg, + ); + published += 1; + }, + 3 => { + // Shared key: a second mutator may find nothing to remove. + // `mutate_runtime` still publishes (generation bumps) on that + // no-op, so this counts a snapshot, not a hit. + engine.remove_route_annotation("tool", TOOL, None, HOOK); + published += 1; + }, + _ => { + let id = next_id; + next_id += 1; + let yaml = if hooks { + let left = format!("r{mutator_id}-{id}-L"); + let right = format!("r{mutator_id}-{id}-R"); + format!( + " +engine_settings: + dispatch: hooks +plugins: + - name: {left} + kind: {KIND} + hooks: [{HOOK}] + mode: sequential + priority: 10 + - name: {right} + kind: {KIND} + hooks: [{HOOK}] + mode: sequential + priority: 20 +" + ) + } else { + let name = format!("r{mutator_id}-{id}"); + format!( + " +engine_settings: + dispatch: policy +plugins: + - name: {name} + kind: {KIND} + hooks: [{HOOK}] + mode: sequential +routes: + - tool: {TOOL} +" + ) + }; + if parse_config(&yaml) + .and_then(|cfg| engine.load_config(cfg)) + .is_ok() + { + if hooks { + live.insert(format!("r{mutator_id}-{id}-L")); + live.insert(format!("r{mutator_id}-{id}-R")); + } else { + live.insert(format!("r{mutator_id}-{id}")); + } + published += 1; + } + }, + } + } + + MutatorOutcome { live, published } +} diff --git a/crates/ppe-core/tests/loom_generation_snapshot.rs b/crates/ppe-core/tests/loom_generation_snapshot.rs new file mode 100644 index 00000000..5378ea11 --- /dev/null +++ b/crates/ppe-core/tests/loom_generation_snapshot.rs @@ -0,0 +1,70 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Praxis Contributors + +//! Extracted model of the `generation` / snapshot pairing `PolicyEngine` +//! documents on `mutate_runtime`. +//! +//! Production code publishes a snapshot and then bumps `generation` with +//! `Release`. Orchestrators load `generation` with `Acquire` and then load +//! the snapshot. A reader who observes a higher generation is intended to +//! see the snapshot stored before that bump. +//! +//! This file restates that pairing with loom atomics. It is documentation +//! of the intended order, not a harness over `engine.rs`: flipping the +//! real `Release` / `Acquire` sites to `Relaxed` does not fail this test. +//! Wiring it to the engine needs loom-aware snapshot storage (`arc_swap` +//! is not) and a `--cfg loom` build that CI does not run. +//! +//! Loom explores every allowed interleaving of those atomics. The model is +//! this pairing only: two threads, two atomics. A third writer thread is +//! not included — exhaustive search does not scale past this size, and +//! writers are already serialised by `runtime_write` in the engine. + +#![cfg(loom)] +#![allow( + missing_docs, + clippy::expect_used, + clippy::panic, + clippy::unwrap_used, + reason = "test and example code" +)] + +use loom::sync::Arc; +use loom::sync::atomic::{AtomicU64, Ordering}; +use loom::thread; + +/// One writer: `store` the snapshot, then `fetch_add(Release)` on +/// generation. A reader that `Acquire`-loads a non-zero generation must +/// see the stored snapshot. +#[test] +fn acquire_on_generation_sees_snapshot_stored_before_release_bump() { + loom::model(|| { + let snapshot = Arc::new(AtomicU64::new(0)); + let generation = Arc::new(AtomicU64::new(0)); + + let writer_snapshot = Arc::clone(&snapshot); + let writer_generation = Arc::clone(&generation); + let writer = thread::spawn(move || { + writer_snapshot.store(1, Ordering::Relaxed); + writer_generation.fetch_add(1, Ordering::Release); + }); + + let reader_snapshot = Arc::clone(&snapshot); + let reader_generation = Arc::clone(&generation); + let reader = thread::spawn(move || { + let observed = reader_generation.load(Ordering::Acquire); + let snap = reader_snapshot.load(Ordering::Relaxed); + if observed >= 1 { + assert_eq!( + snap, 1, + "Acquire on generation must observe the snapshot \ + stored before the Release bump; generation={observed} \ + snap={snap}" + ); + } + }); + + writer.join().unwrap(); + reader.join().unwrap(); + }); +}