From 922b07698db22ae3cc130f15d5303834faa48034 Mon Sep 17 00:00:00 2001 From: Abdallah Samara Date: Tue, 1 Sep 2026 17:39:18 +0300 Subject: [PATCH] test: exercise engine snapshot updates under real parallelism Signed-off-by: Abdallah Samara --- .github/workflows/nightly-tsan.yml | 56 ++ CONTRIBUTING.md | 24 + Cargo.lock | 107 ++++ Cargo.toml | 4 + Makefile | 12 + crates/ppe-core/Cargo.toml | 3 + crates/ppe-core/src/engine.rs | 50 +- crates/ppe-core/tests/engine_concurrency.rs | 477 ++++++++++++++++++ .../tests/loom_generation_snapshot.rs | 63 +++ 9 files changed, 771 insertions(+), 25 deletions(-) create mode 100644 .github/workflows/nightly-tsan.yml create mode 100644 crates/ppe-core/tests/engine_concurrency.rs create mode 100644 crates/ppe-core/tests/loom_generation_snapshot.rs diff --git a/.github/workflows/nightly-tsan.yml b/.github/workflows/nightly-tsan.yml new file mode 100644 index 0000000..615c375 --- /dev/null +++ b/.github/workflows/nightly-tsan.yml @@ -0,0 +1,56 @@ +# 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 + RUSTFLAGS: -Zsanitizer=thread + 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@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0 + with: + persist-credentials: false + - uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.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 07a1b52..d63a647 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -134,3 +134,27 @@ 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. + +Use a multi-threaded runtime when the test shares a `PolicyEngine` (or any other +`Arc` handle) across tasks **and** mutates it while invokes are in flight: + +```rust +#[tokio::test(flavor = "multi_thread")] +async fn register_while_other_tasks_invoke() { /* ... */ } +``` + +That is the right flavor for registration, unregister, hot reload, and +route-cache fill or invalidation. Sequential logic — one engine, one task, no +shared mutation — stays 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`). The `Release` / `Acquire` pairing between +the snapshot and `config_generation` is checked by the loom model in +`crates/ppe-core/tests/loom_generation_snapshot.rs`. diff --git a/Cargo.lock b/Cargo.lock index c6d0f5d..91fb71e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1097,6 +1097,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" @@ -1777,12 +1792,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" @@ -1929,6 +1966,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" @@ -2378,6 +2424,7 @@ dependencies = [ "chrono", "hashbrown 0.17.1", "http", + "loom", "praxis-policy-orchestration", "serde", "serde_json", @@ -2998,6 +3045,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" @@ -3167,6 +3220,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" @@ -3451,6 +3513,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" @@ -3670,6 +3741,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]] @@ -3822,6 +3923,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 96578ca..1cf2743 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -108,6 +108,10 @@ 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: the model lives in ppe-core's loom test, not in +# production code. +loom = "0.7" wildmatch = "2" rmp-serde = "1" serde_bytes = "0.11" diff --git a/Makefile b/Makefile index 1350102..4295265 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)" @@ -157,6 +158,17 @@ test: @$(CARGO) test --workspace @$(CARGO) test --workspace --all-features +# ThreadSanitizer on the engine concurrency stress test. Needs nightly and a +# Linux target; the sanitizer does not run on the pinned stable toolchain. +# `--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 -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 acdc200..eb99dc4 100644 --- a/crates/ppe-core/Cargo.toml +++ b/crates/ppe-core/Cargo.toml @@ -63,5 +63,8 @@ zeroize = { version = "1.9", features = ["zeroize_derive"] } # (and praxis-policy-apl-core's `Effect::Parallel`). Leaf crate, no cycles back here. praxis-policy-orchestration = { workspace = true } +[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 d8582b7..7b1c5a2 100644 --- a/crates/ppe-core/src/engine.rs +++ b/crates/ppe-core/src/engine.rs @@ -2961,7 +2961,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")); @@ -3157,7 +3157,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); @@ -3177,7 +3177,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}; @@ -4940,7 +4940,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; @@ -5506,7 +5506,7 @@ routes: ); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_from_config_creates_manager() { register_fixture_hooks(); let yaml = r#" @@ -5533,7 +5533,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#" @@ -5567,7 +5567,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#" @@ -5588,7 +5588,7 @@ plugins: } } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_from_config_multiple_plugins() { register_fixture_hooks(); let yaml = r#" @@ -5632,7 +5632,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#" @@ -5834,7 +5834,7 @@ routes: ); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_routing_cache_different_entities_separate() { register_fixture_hooks(); let yaml = r#" @@ -5885,7 +5885,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#" @@ -5923,7 +5923,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#" @@ -5992,7 +5992,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. @@ -6066,7 +6066,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#" @@ -6110,7 +6110,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#" @@ -6163,7 +6163,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#" @@ -6400,7 +6400,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}; @@ -6719,7 +6719,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; @@ -6790,7 +6790,7 @@ routes: ); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] async fn test_register_factory_then_load_config() { register_fixture_hooks(); let yaml = r#" @@ -6848,7 +6848,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 @@ -6889,7 +6889,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 @@ -7160,7 +7160,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)); @@ -7196,7 +7196,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)); @@ -7765,7 +7765,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; @@ -8304,7 +8304,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 0000000..b264e6e --- /dev/null +++ b/crates/ppe-core/tests/engine_concurrency.rs @@ -0,0 +1,477 @@ +// 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 a complete snapshot, +//! not a mix of two configs, and after mutators stop the route cache +//! matches the live snapshot. +//! +//! The stress test is 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, + dead_code, + clippy::expect_used, + clippy::panic, + clippy::print_stderr, + clippy::unwrap_used, + reason = "test and example code" +)] + +use std::collections::HashSet; +use std::env; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Barrier}; +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; + +#[derive(Debug, Clone)] +struct StressPayload { + value: String, +} +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, +} + +impl StressPlugin { + fn new(cfg: PluginConfig) -> Arc { + Arc::new(Self { cfg }) + } +} + +#[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 { + PluginResult::allow() + } +} + +#[async_trait] +impl AnyHookHandler for StressPlugin { + async fn invoke( + &self, + _payload: &dyn PluginPayload, + _extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> Result, Box> { + Ok(erase_result(PluginResult::::allow())) + } + + fn hook_type_name(&self) -> &'static str { + StressHook::NAME + } +} + +struct StressFactory; + +impl PluginFactory for StressFactory { + fn create(&self, config: &PluginConfig) -> Result> { + let plugin = StressPlugin::new(config.clone()); + 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 bootstrap() -> Arc { + register_stress_hook(); + let engine = Arc::new(PolicyEngine::default()); + engine.register_factory(KIND, Box::new(StressFactory)); + 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("bootstrap config must parse"); + engine + .load_config(config) + .expect("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, name: &str) -> Result<(), Box> { + let cfg = plugin_config(name); + engine.register_handler::(StressPlugin::new(cfg.clone()), cfg) +} + +/// 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 engine = bootstrap(); + 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 barrier = Arc::clone(&barrier); + joins.push(thread::spawn(move || { + let name = format!("barrier-{i}"); + barrier.wait(); + register_named(&engine, &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. Each mutator owns a name +/// prefix, so the expected live set is the union of per-thread logs and +/// does not depend on a global total order. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn stress_invoke_against_concurrent_mutation() { + 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); + eprintln!( + "engine concurrency stress seed={seed} ops={ops} invokers={invokers} \ + mutators={mutators}" + ); + + let engine = bootstrap(); + engine.initialize().await.expect("initialize"); + let generation_at_start = engine.config_generation(); + + 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 invoke_ok = Arc::clone(&invoke_ok); + invoke_joins.push(tokio::spawn(async move { + for n in 0..ops { + let payload: Box = Box::new(StressPayload { + value: format!("invoker-{i}-{n}"), + }); + 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} op={n} denied={:?}", + result.violation + ); + invoke_ok.fetch_add(1, Ordering::Relaxed); + } + })); + } + + let mut mutator_joins = Vec::with_capacity(mutators); + for mutator_id in 0..mutators { + let engine = Arc::clone(&engine); + mutator_joins.push(thread::spawn(move || { + mutator_loop( + &engine, + seed, + u64::try_from(mutator_id).expect("fits u64"), + ops, + ) + })); + } + + let mut expected: HashSet = HashSet::new(); + expected.insert(BASE_PLUGIN.to_owned()); + let mut published = 0_u64; + for join in mutator_joins { + let outcome = join.join().expect("mutator thread"); + expected.extend(outcome.live); + published += outcome.published; + } + for join in invoke_joins { + join.await.expect("invoker task"); + } + + // Annotations installed by mutators short-circuit routing and skip + // the route cache. Strip them so the quiesced invoke is a cache miss + // against the live snapshot. + 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(); + 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}" + ); + + // Mutators have stopped. A cache filled under their feet must not + // outlive the snapshot: clear, miss, refill from the live config. + engine.clear_routing_cache(); + assert_eq!(engine.routing_cache_size(), 0); + let payload: Box = Box::new(StressPayload { + value: "after-quiesce".into(), + }); + 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() + ); + assert_eq!( + invoke_ok.load(Ordering::Relaxed), + ops * u64::try_from(invokers).expect("fits u64"), + "every invoke must have finished; seed={seed}" + ); +} + +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, seed: u64, mutator_id: u64, ops: u64) -> 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, &name).is_ok() { + live.insert(name.clone()); + owned.push(name); + published += 1; + } + }, + 1 => { + if let Some(name) = owned.pop() { + if 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()), + cfg, + ); + published += 1; + }, + 3 => { + engine.remove_route_annotation("tool", TOOL, None, HOOK); + published += 1; + }, + _ => { + let name = format!("r{mutator_id}-{next_id}"); + next_id += 1; + let yaml = format!( + " +engine_settings: + dispatch: policy +plugins: + - name: {name} + kind: {KIND} + hooks: [{HOOK}] + mode: sequential +routes: + - tool: {TOOL} +" + ); + if let Ok(()) = parse_config(&yaml).and_then(|cfg| engine.load_config(cfg)) { + live.insert(name); + 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 0000000..f6294ac --- /dev/null +++ b/crates/ppe-core/tests/loom_generation_snapshot.rs @@ -0,0 +1,63 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Praxis Contributors + +//! Extracted model of the `generation` / snapshot pairing in `PolicyEngine`. +//! +//! Production code publishes a snapshot and then bumps `generation` with +//! `Release`. Orchestrators load `generation` with `Acquire` and then load +//! the snapshot. The comment on `mutate_runtime` claims that a reader who +//! observes a higher generation is guaranteed to see the snapshot stored +//! before that bump. +//! +//! 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. + +#![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(); + }); +}