diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d8c46c..f2bb6a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -59,6 +59,8 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/). ### Fixed +- **Concurrent registration no longer loses plugins.** Every mutation of the engine's runtime snapshot was a load, clone, store with nothing serializing writers, so two threads that loaded the same snapshot each published their own copy and the last write discarded the other silently. A test putting sixteen threads through `register_handler` at once kept one plugin, and all sixteen calls returned `Ok`. Writers now serialize on a mutex held across the copy-on-write, covering `load_config`'s inline swap as well; the read path is untouched and still lock-free, and the generation counter still bumps exactly once per published mutation. Nothing in the workspace registers concurrently today, so no shipped configuration was affected, but registration takes `&self` on an `Arc`-shared engine and a host is free to call it from any thread. Neither that mutex nor the factory-registry `RwLock` is reentrant, so `load_config` now runs every `PluginFactory::create` holding neither: it resolves the factories, releases the registry lock, instantiates, and only then takes the writer lock for the registry clone and the swap. A factory that calls back into `register_handler`, `annotate_route`, `unregister_plugin`, or `register_factory` while being built would otherwise block on a lock its own caller was holding. Whatever such a factory registers on its way through is picked up by the clone rather than discarded by it. The two route-override paths already released the registry lock before `create` and are unchanged. `mutate_runtime` and `try_mutate_runtime` likewise release the writer lock before the snapshot they replaced goes out of scope: that snapshot usually holds the last reference to whatever the mutation discarded, and a plugin or annotation handler's `Drop` is host code too, so `remove_route_annotation` on a handler that called back would have blocked on the lock its own release was holding. A rejected `load_config` is the same story from the other end: the plugins its factories just built have nowhere else to live, and the registry drops the one whose name collided while the entries behind it are never registered at all. Registration now borrows the instances and hands the registry `Arc` clones, so the load keeps the last reference to every one of them and releases the lock before letting go. ([#23](https://github.com/praxis-proxy/policy/issues/23)) + - **Subject claims keep their JSON shape.** `SubjectExtension.claims` holds `serde_json::Value` and flattens into the attribute bag through `payload::walk`, so Keycloak's nested `realm_access.roles` is a `StringSet` a policy can test instead of one opaque string. Client claims always worked this way. **Breaking** for Rust callers reading `claims`; `SubjectExtension::claim_str` covers the scalar lookups. Scalar policies such as `claim.tenant == 'acme'` are unaffected, but a structured claim now sets only the flattened children beneath `claim.`, not the key itself, and a claim whose value is `{}` or `null` sets no key at all where it previously landed as stringified text. ([#9](https://github.com/praxis-proxy/policy/pull/9)) - **A scalar array reaches the bag as a `StringSet` instead of no key.** `payload::walk` emitted nothing for `[]` or for any array holding a number or bool, so a user with no realm roles had no `claim.realm_access.roles`, and a provider minting `"group_ids": [1, 2]` had none of those either — a missing key is a CEL error that fail-closed handling denies. Numbers and bools now render as strings, so `claim.group_ids contains "1"`, matching how a float claim is carried through Cedar. An array holding a nested array or object still sets no key. Applies to `args.*`, `result.*`, `data.*` and `client.claim.*` too. ([#9](https://github.com/praxis-proxy/policy/pull/9)) diff --git a/Cargo.lock b/Cargo.lock index a05fcf0..d7b89de 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -100,15 +100,15 @@ dependencies = [ [[package]] name = "astral-tokio-tar" -version = "0.5.6" +version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ec179a06c1769b1e42e1e2cbe74c7dcdb3d6383c838454d063eaac5bbb7ebbe5" +checksum = "b18457efd137254e016bbde5e1d88df61c4e1a5ae2223746e56123bac6af2463" dependencies = [ - "filetime", "futures-core", "libc", "portable-atomic", "rustc-hash", + "rustix", "tokio", "tokio-stream", "xattr", @@ -308,9 +308,9 @@ dependencies = [ [[package]] name = "bollard" -version = "0.19.4" +version = "0.20.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "87a52479c9237eb04047ddb94788c41ca0d26eaff8b697ecfbb4c32f7fdc3b1b" +checksum = "ee04c4c84f1f811b017f2fbb7dd8815c976e7ca98593de9c1e2afad0f636bff4" dependencies = [ "async-stream", "base64 0.22.1", @@ -318,7 +318,6 @@ dependencies = [ "bollard-buildkit-proto", "bollard-stubs", "bytes", - "chrono", "futures-core", "futures-util", "hex", @@ -336,14 +335,13 @@ dependencies = [ "rand 0.9.5", "rustls", "rustls-native-certs", - "rustls-pemfile", "rustls-pki-types", "serde", "serde_derive", "serde_json", - "serde_repr", "serde_urlencoded", "thiserror", + "time", "tokio", "tokio-stream", "tokio-util", @@ -368,19 +366,18 @@ dependencies = [ [[package]] name = "bollard-stubs" -version = "1.49.1-rc.28.4.0" +version = "1.52.1-rc.29.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5731fe885755e92beff1950774068e0cae67ea6ec7587381536fca84f1779623" +checksum = "0f0a8ca8799131c1837d1282c3f81f31e76ceb0ce426e04a7fe1ccee3287c066" dependencies = [ "base64 0.22.1", "bollard-buildkit-proto", "bytes", - "chrono", "prost", "serde", "serde_json", "serde_repr", - "serde_with", + "time", ] [[package]] @@ -964,25 +961,15 @@ checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" [[package]] name = "ferroid" -version = "0.8.9" +version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bb330bbd4cb7a5b9f559427f06f98a4f853a137c8298f3bd3f8ca57663e21986" +checksum = "ee93edf3c501f0035bbeffeccfed0b79e14c311f12195ec0e661e114a0f60da4" dependencies = [ "portable-atomic", - "rand 0.9.5", + "rand 0.10.2", "web-time", ] -[[package]] -name = "filetime" -version = "0.2.29" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759" -dependencies = [ - "cfg-if", - "libc", -] - [[package]] name = "find-msvc-tools" version = "0.1.11" @@ -2924,15 +2911,6 @@ dependencies = [ "security-framework", ] -[[package]] -name = "rustls-pemfile" -version = "2.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50" -dependencies = [ - "rustls-pki-types", -] - [[package]] name = "rustls-pki-types" version = "1.15.1" @@ -3402,9 +3380,9 @@ dependencies = [ [[package]] name = "testcontainers" -version = "0.26.3" +version = "0.27.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a81ec0158db5fbb9831e09d1813fe5ea9023a2b5e6e8e0a5fe67e2a820733629" +checksum = "bfd5785b5483672915ed5fe3cddf9f546802779fc1eceff0a6fb7321fac81c1e" dependencies = [ "astral-tokio-tar", "async-trait", @@ -3415,6 +3393,7 @@ dependencies = [ "etcetera", "ferroid", "futures", + "http", "itertools 0.14.0", "log", "memchr", @@ -3432,9 +3411,9 @@ dependencies = [ [[package]] name = "testcontainers-modules" -version = "0.14.0" +version = "0.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e75e78ff453128a2c7da9a5d5a3325ea34ea214d4bf51eab3417de23a4e5147" +checksum = "e5985fde5befe4ffa77a052e035e16c2da86e8bae301baa9f9904ad3c494d357" dependencies = [ "testcontainers", ] diff --git a/builtins/session/valkey/Cargo.toml b/builtins/session/valkey/Cargo.toml index 9e0311e..1be5732 100644 --- a/builtins/session/valkey/Cargo.toml +++ b/builtins/session/valkey/Cargo.toml @@ -62,8 +62,8 @@ deadpool-redis = { version = "0.23", default-features = false, features = [ tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread"] } # Spins a real `valkey/valkey` container for integration tests; these are # `#[ignore]`d by default and run via a dedicated job (see tests/). -testcontainers-modules = { version = "0.14", features = ["valkey"] } -testcontainers = "0.26" +testcontainers-modules = { version = "0.15", features = ["valkey"] } +testcontainers = "0.27" [lints] workspace = true diff --git a/crates/ppe-core/src/engine.rs b/crates/ppe-core/src/engine.rs index d09821a..ff25fc9 100644 --- a/crates/ppe-core/src/engine.rs +++ b/crates/ppe-core/src/engine.rs @@ -25,7 +25,7 @@ use std::hash::{Hash, Hasher}; use std::path::Path; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{Arc, RwLock}; +use std::sync::{Arc, Mutex, RwLock}; use hashbrown::HashMap; use tracing::{error, info, warn}; @@ -247,6 +247,22 @@ pub struct PolicyEngine { /// so callers can use 0 as a "never observed" sentinel. generation: AtomicU64, + /// Serialises writers on `runtime`. Every mutation is a load, clone, + /// store, so the lock has to span the whole sequence: a writer that clones + /// a snapshot another writer has already replaced publishes one missing + /// that change, and the losing call still returns `Ok`. + /// + /// Guards `()` because the data lives in the `ArcSwap`. Readers never take + /// it, so the invoke path is lock-free. + /// + /// Held across the copy-on-write in `mutate_runtime`, + /// `try_mutate_runtime`, and `load_config`, and released before the + /// routing cache is cleared. It is not reentrant, so no host-supplied + /// callback may run beneath it: `load_config` calls every + /// `PluginFactory::create` before taking it, and config visitors reach + /// `annotate_route` after `load_config` has returned, not during. + runtime_write: Mutex<()>, + /// Tracks in-flight fire-and-forget background tasks across all /// invocations so `shutdown()` can wait for them to drain before /// returning. Without this, audit/telemetry tasks spawned by recent @@ -309,35 +325,86 @@ fn warn_on_inactive_settings(cfg: &PolicyConfig) { } } -/// Instantiate every plugin in `plugin_configs` via the matching factory -/// and register the resulting handlers into `target_registry`. Shared by -/// `PolicyEngine::from_config` (fresh registry) and `load_config` (clone -/// of the existing registry) so the instantiation loop lives in one place. +/// Look up the factory for every entry in `plugin_configs`, in config +/// order. Returns on the first `kind` with no registered factory. /// -/// Returns on the first failure (factory missing, factory.create error, or -/// duplicate-name registration). On error, `target_registry` is in a -/// partial state — both callers discard it on failure (`load_config` builds -/// the new registry on a clone and only swaps on Ok; `from_config` bails -/// before publishing the snapshot). -fn instantiate_plugins_into( - target_registry: &mut PluginRegistry, +/// Split from `create_plugin_instances` so a caller can release the +/// engine's `factories` lock before any factory runs — see that function +/// for what happens if it doesn't. Resolving first also means an unknown +/// `kind` is reported before any factory has run, so a rejected config +/// leaves behind no half-built plugins. +fn resolve_factories( plugin_configs: &[crate::plugin::PluginConfig], factories: &PluginFactoryRegistry, -) -> Result<(), Box> { - for plugin_config in plugin_configs { - let factory = factories - .get(&plugin_config.kind) - .ok_or_else(|| PluginError::Config { - message: format!( - "no factory registered for plugin kind '{}' (plugin '{}')", - plugin_config.kind, plugin_config.name - ), - })?; +) -> Result>, Box> { + plugin_configs + .iter() + .map(|plugin_config| { + factories.get(&plugin_config.kind).ok_or_else(|| { + Box::new(PluginError::Config { + message: format!( + "no factory registered for plugin kind '{}' (plugin '{}')", + plugin_config.kind, plugin_config.name + ), + }) + }) + }) + .collect() +} - let instance = factory.create(plugin_config)?; +/// Create one plugin instance per entry in `plugin_configs`, using the +/// factories `resolve_factories` returned for those same configs. +/// +/// Must run with no engine lock held. `factory.create` is host code and is +/// free to re-enter the engine: `register_handler`, `annotate_route` and +/// `unregister_plugin` take `runtime_write`, `register_factory` takes the +/// `factories` write side, and neither lock is reentrant, so a caller +/// holding either across this function deadlocks against a factory that +/// calls back. `RwLock` gives no guarantee for a recursive read either — a +/// waiting writer can park a second `factories.read()` behind it. +/// +/// Returns on the first factory that rejects its config; instances already +/// created are dropped. +fn create_plugin_instances( + plugin_configs: &[crate::plugin::PluginConfig], + factories: &[Arc], +) -> Result, Box> { + plugin_configs + .iter() + .zip(factories) + .map(|(plugin_config, factory)| factory.create(plugin_config)) + .collect() +} +/// Register instances produced by `create_plugin_instances` into +/// `target_registry`. `instances` is zipped against `plugin_configs`, so it +/// has to be the slice that call returned for these same configs: a shorter +/// one silently drops the tail, a reordered one pairs each plugin with the +/// wrong config. +/// +/// Borrows rather than consumes, and hands the registry `Arc` clones, so +/// the caller keeps the last reference to every instance. That matters on +/// the error path: `register_multi_handler` takes ownership and drops what +/// it was given when it rejects a name, and the entries past the failure +/// are never reached at all. Owning them here would run those plugins' +/// `Drop` — host code — inside whatever lock the caller holds. +/// +/// Returns on the first duplicate-name registration. On error, +/// `target_registry` is in a partial state — both callers discard it on +/// failure (`load_config` builds the new registry on a clone and only swaps +/// on Ok; `from_config` bails before publishing the snapshot). +fn register_instances_into( + target_registry: &mut PluginRegistry, + plugin_configs: &[crate::plugin::PluginConfig], + instances: &[crate::factory::PluginInstance], +) -> Result<(), Box> { + for (plugin_config, instance) in plugin_configs.iter().zip(instances) { target_registry - .register_multi_handler(instance.plugin, plugin_config.clone(), instance.handlers) + .register_multi_handler( + Arc::clone(&instance.plugin), + plugin_config.clone(), + instance.handlers.clone(), + ) .map_err(|msg| Box::new(PluginError::Config { message: msg }))?; info!( @@ -386,6 +453,7 @@ impl PolicyEngine { route_cache_full_warned: AtomicBool::new(false), initialized: AtomicBool::new(false), generation: AtomicU64::new(0), + runtime_write: Mutex::new(()), task_tracker: tokio_util::task::TaskTracker::new(), visitors: RwLock::new(Vec::new()), http_transport: std::sync::OnceLock::new(), @@ -397,14 +465,26 @@ impl PolicyEngine { self.runtime.load_full() } + /// Take the runtime writer lock, ignoring poisoning. A panic inside a + /// mutation closure leaves the `ArcSwap` holding whatever was last + /// published, which is a complete snapshot either way, so there is no + /// half-applied state for the next writer to inherit. + fn lock_runtime_writer(&self) -> std::sync::MutexGuard<'_, ()> { + self.runtime_write + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + /// Apply a mutation to the runtime snapshot via copy-on-write. /// Clones the current snapshot, runs the closure on the clone, and /// atomically swaps it in. Concurrent readers continue using the old - /// snapshot; subsequent readers see the new one. + /// snapshot; subsequent readers see the new one. Writers serialise on + /// `runtime_write` for the length of the call. fn mutate_runtime(&self, f: F) -> R where F: FnOnce(&mut RuntimeSnapshot) -> R, { + let writer = self.lock_runtime_writer(); let current = self.runtime.load_full(); let mut next = (*current).clone(); let result = f(&mut next); @@ -413,24 +493,42 @@ impl PolicyEngine { // config_generation() — external cache consumers that observe a // higher generation are guaranteed to see the new snapshot. self.generation.fetch_add(1, Ordering::Release); + // Release before `current` falls out of scope. The store above + // usually leaves this the last reference to the old snapshot, and + // dropping it drops the registry, whose plugin `Drop` impls are host + // code: one that re-enters the engine would block on this lock. + drop(writer); + drop(current); result } /// Like `mutate_runtime` but the mutation can fail — the new snapshot /// is only published on `Ok`. On `Err`, the original snapshot is - /// untouched, so a partially-mutated clone is silently discarded. + /// untouched, so a partially-mutated clone is silently discarded. An + /// `Err` releases the writer lock without storing, so a rejected + /// mutation publishes nothing and holds up no other writer. fn try_mutate_runtime(&self, f: F) -> Result where F: FnOnce(&mut RuntimeSnapshot) -> Result, { + let writer = self.lock_runtime_writer(); let current = self.runtime.load_full(); let mut next = (*current).clone(); - let result = f(&mut next)?; - self.runtime.store(Arc::new(next)); - // Same Release-ordered bump as mutate_runtime — only on Ok, since - // Err leaves the snapshot untouched. - self.generation.fetch_add(1, Ordering::Release); - Ok(result) + let result = f(&mut next); + if result.is_ok() { + self.runtime.store(Arc::new(next)); + // Same Release-ordered bump as mutate_runtime — only on Ok, since + // Err leaves the snapshot untouched. + self.generation.fetch_add(1, Ordering::Release); + } + // Released before `next` and `current` fall out of scope, for the + // reason mutate_runtime releases early: both can hold the last + // reference to a plugin, and a `Drop` impl is host code. `next` + // survives to here on the `Err` path — no `?` above — so a rejected + // mutation drops whatever it built with the lock already gone. + drop(writer); + drop(current); + result } /// Monotonic counter that increments on every runtime snapshot swap @@ -537,30 +635,50 @@ impl PolicyEngine { pub fn load_config(&self, policy_config: PolicyConfig) -> Result<(), Box> { warn_on_inactive_settings(&policy_config); + // Resolve under the factories read lock, then drop it and + // instantiate holding nothing, per `create_plugin_instances`. A + // factory that registers something on its way through publishes it + // here, so the clone below picks it up rather than swapping a + // snapshot taken before it existed. + let factories = { + let registry = self + .factories + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + resolve_factories(&policy_config.plugins, ®istry)? + }; + let instances = create_plugin_instances(&policy_config.plugins, &factories)?; + // Build the new snapshot from the current one — copy-on-write so // concurrent invokes keep using the existing config until we swap. // We can't use mutate_runtime here because we need to atomically // ALSO build a new executor + new cache cap from the same config — - // the snapshot fields are coupled. - let factories = self - .factories - .read() - .unwrap_or_else(std::sync::PoisonError::into_inner); + // the snapshot fields are coupled. Takes the writer lock by hand for + // the reason mutate_runtime takes it. + let writer = self.lock_runtime_writer(); + let current = self.runtime.load_full(); let mut new_registry = current.registry.clone(); - instantiate_plugins_into(&mut new_registry, &policy_config.plugins, &factories)?; - - // Drop the factories read lock before taking other locks - // (route_cache write below) to avoid lock-ordering hazards. - drop(factories); - - self.runtime - .store(Arc::new(snapshot_from_config(new_registry, policy_config))); - // Same generation bump as mutate_runtime — load_config doesn't - // go through that helper because it has to swap registry + executor - // + cache-cap atomically as one snapshot. - self.generation.fetch_add(1, Ordering::Release); + let registered = + register_instances_into(&mut new_registry, &policy_config.plugins, &instances); + if registered.is_ok() { + self.runtime + .store(Arc::new(snapshot_from_config(new_registry, policy_config))); + // Same generation bump as mutate_runtime — load_config doesn't + // go through that helper because it has to swap registry + executor + // + cache-cap atomically as one snapshot. + self.generation.fetch_add(1, Ordering::Release); + } + // Released before `instances`, `new_registry` and `current` fall out + // of scope, the same discipline try_mutate_runtime follows. A + // rejected load holds the only references to plugins the factories + // just built, and dropping one runs host `Drop` code that is free to + // re-enter the engine. No `?` above, so nothing unwinds past here + // still holding the lock. + drop(writer); + drop(current); + registered?; // Clear routing cache — config changed. self.clear_routing_cache(); @@ -788,8 +906,10 @@ impl PolicyEngine { }); // Instantiate into a fresh registry, then publish atomically. + let resolved = resolve_factories(&policy_config.plugins, factories)?; + let instances = create_plugin_instances(&policy_config.plugins, &resolved)?; let mut new_registry = PluginRegistry::new(); - instantiate_plugins_into(&mut new_registry, &policy_config.plugins, factories)?; + register_instances_into(&mut new_registry, &policy_config.plugins, &instances)?; engine .runtime @@ -2663,6 +2783,319 @@ mod tests { assert_eq!(mgr.plugin_count(), 2); } + /// N OS threads register N distinct plugins at the same instant, and all N + /// have to survive. A registration lost to a concurrent one still returns + /// `Ok`, so the count is the only thing that catches it. + /// + /// The barrier is load-bearing: it lines every thread up on the same + /// snapshot, which is what makes the loss reliable rather than occasional. + /// So are the real threads. The sibling test above runs on + /// `current_thread`, where a load and a store cannot interleave, so + /// nothing here survives being rewritten as `tokio::spawn`. + #[test] + fn concurrent_registration_loses_no_plugins() { + let mgr = Arc::new(PolicyEngine::default()); + let n = 16_usize; + let barrier = std::sync::Barrier::new(n); + + std::thread::scope(|s| { + for i in 0..n { + let mgr = Arc::clone(&mgr); + let barrier = &barrier; + s.spawn(move || { + let cfg = make_config(&format!("p{i}"), 10, PluginMode::Sequential); + let plugin: Arc = Arc::new(AllowPlugin { cfg: cfg.clone() }); + barrier.wait(); + mgr.register_handler::(plugin, cfg) + .expect("registration must succeed"); + }); + } + }); + + assert_eq!( + mgr.plugin_count(), + n, + "every registration that returned Ok must be present in the snapshot", + ); + assert_eq!( + mgr.config_generation(), + n as u64, + "generation bumps exactly once per published mutation", + ); + } + + /// A factory that reaches back into the engine on its way through + /// `create`, once per lock a host factory can plausibly hit: + /// `register_handler` takes `runtime_write`, `register_factory` takes + /// the `factories` write side. The shape of a host plugin that installs + /// a companion handler and the factory for its own sub-kind while being + /// built. + struct ReentrantFactory { + engine: std::sync::Weak, + } + + impl crate::factory::PluginFactory for ReentrantFactory { + fn create( + &self, + config: &PluginConfig, + ) -> Result> { + let engine = self.engine.upgrade().expect("engine outlives the factory"); + let side_cfg = make_config( + &format!("{}_companion", config.name), + 10, + PluginMode::Sequential, + ); + let side: Arc = Arc::new(AllowPlugin { + cfg: side_cfg.clone(), + }); + engine + .register_handler::(side, side_cfg) + .expect("re-entrant registration must succeed"); + engine.register_factory("test/reentrant_spawned", Box::new(AllowPluginFactory)); + AllowPluginFactory.create(config) + } + } + + /// Neither `runtime_write` nor the `factories` `RwLock` is reentrant, so + /// a factory that calls back into the engine deadlocks against either + /// one `load_config` is still holding while it runs `create`. The load + /// runs on its own thread so the timeout reports that as a failure + /// instead of hanging the test binary. + /// + /// The count also pins the ordering that makes the fix correct — the + /// registry is cloned after the factories run, so what a factory + /// registered mid-create is in the published snapshot, not overwritten + /// by it. + #[test] + fn load_config_holds_no_lock_across_factory_create() { + let engine = Arc::new(PolicyEngine::default()); + engine.register_factory( + "test/reentrant", + Box::new(ReentrantFactory { + engine: Arc::downgrade(&engine), + }), + ); + + let yaml = r" +plugins: + - name: main_plugin + kind: test/reentrant + hooks: [test_hook] + mode: sequential + priority: 10 +"; + let policy_config = crate::config::parse_config(yaml).unwrap(); + + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let loader = Arc::clone(&engine); + std::thread::spawn(move || { + let _ = done_tx.send(loader.load_config(policy_config).is_ok()); + }); + + let loaded = done_rx + .recv_timeout(std::time::Duration::from_secs(10)) + .expect("load_config deadlocked against a re-entrant factory"); + assert!(loaded, "load_config must succeed"); + + assert_eq!( + engine.plugin_count(), + 2, + "both the configured plugin and the one its factory registered \ + must be in the published snapshot", + ); + } + + /// An annotation handler whose `Drop` reaches back into the engine. + /// Host annotation handlers own host resources, so their teardown is + /// host code like any other callback. + struct DropReentrantHandler { + cfg: PluginConfig, + engine: std::sync::Weak, + } + + #[async_trait] + impl Plugin for DropReentrantHandler { + fn config(&self) -> &PluginConfig { + &self.cfg + } + async fn initialize(&self) -> Result<(), Box> { + Ok(()) + } + async fn shutdown(&self) -> Result<(), Box> { + Ok(()) + } + } + + #[async_trait] + impl crate::registry::AnyHookHandler for DropReentrantHandler { + async fn invoke( + &self, + _payload: &dyn crate::prelude::PluginPayload, + _extensions: &Extensions, + _ctx: &mut PluginContext, + ) -> Result, Box> { + Err(Box::new(PluginError::Config { + message: "this handler exists to be dropped, not invoked".to_owned(), + })) + } + + fn hook_type_name(&self) -> &'static str { + "test_hook" + } + } + + impl Drop for DropReentrantHandler { + fn drop(&mut self) { + let Some(engine) = self.engine.upgrade() else { + return; + }; + // Any mutating call reaches `runtime_write`; this one is a + // no-op on a key that was never annotated. + engine.remove_route_annotation("tool", "never_annotated", None, "test_hook"); + } + } + + /// A discarded annotation drops the handler the old snapshot held, and + /// that teardown is host code. `runtime_write` has to be released + /// before the old snapshot falls out of scope, or a handler whose + /// `Drop` calls back blocks on the lock its own release is holding. + /// + /// `remove_route_annotation` is the reachable path: the closure drops + /// the entry from the clone while the snapshot being replaced still + /// mirrors it, so the last reference goes with that snapshot, inside + /// `mutate_runtime`. + #[test] + fn a_dropped_annotation_handler_can_re_enter_the_engine() { + let engine = Arc::new(PolicyEngine::default()); + let cfg = make_config("annotated", 10, PluginMode::Sequential); + let handler = Arc::new(DropReentrantHandler { + cfg: cfg.clone(), + engine: Arc::downgrade(&engine), + }); + + engine.annotate_route("tool", "t1", None, "test_hook", handler, cfg); + + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let remover = Arc::clone(&engine); + std::thread::spawn(move || { + // Drops the sole remaining reference — the annotation the line + // above published — and with it the handler. + remover.remove_route_annotation("tool", "t1", None, "test_hook"); + let _ = done_tx.send(()); + }); + + done_rx + .recv_timeout(std::time::Duration::from_secs(10)) + .expect("a re-entrant handler Drop deadlocked against runtime_write"); + } + + /// Builds plugins whose `Drop` re-enters the engine, so a load that is + /// rejected after instantiation has host teardown to run. + struct DropReentrantFactory { + engine: std::sync::Weak, + } + + impl crate::factory::PluginFactory for DropReentrantFactory { + fn create( + &self, + config: &PluginConfig, + ) -> Result> { + let plugin = Arc::new(DropReentrantHandler { + cfg: config.clone(), + engine: self.engine.clone(), + }); + Ok(crate::factory::PluginInstance { + plugin: Arc::clone(&plugin) as Arc, + handlers: vec![("test_hook", plugin)], + }) + } + } + + /// A name conflict rejects the load with the factories' plugins already + /// built, and `load_config` holds the only references to them: the + /// registry drops what it refused, the entries past the conflict are + /// never registered, and the partial clone drops unpublished. All of + /// that is host `Drop` code, so none of it may run under + /// `runtime_write`. + /// + /// The config collides with a plugin registered beforehand rather than + /// with itself — `parse_config` rejects a name repeated inside one + /// document, so an in-file duplicate never reaches the registry. The + /// conflict lands on the first entry, leaving one instance refused by + /// the registry and one never reached. + #[test] + fn a_rejected_load_drops_its_plugins_outside_the_writer_lock() { + let engine = Arc::new(PolicyEngine::default()); + engine.register_factory( + "test/drop_reentrant", + Box::new(DropReentrantFactory { + engine: Arc::downgrade(&engine), + }), + ); + + let taken = make_config("taken", 5, PluginMode::Sequential); + let sitting: Arc = Arc::new(AllowPlugin { cfg: taken.clone() }); + engine + .register_handler::(sitting, taken) + .expect("the conflicting name has to be registered first"); + + let yaml = r" +plugins: + - name: taken + kind: test/drop_reentrant + hooks: [test_hook] + mode: sequential + priority: 10 + - name: never_reached + kind: test/drop_reentrant + hooks: [test_hook] + mode: sequential + priority: 20 +"; + let policy_config = crate::config::parse_config(yaml).unwrap(); + + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let loader = Arc::clone(&engine); + std::thread::spawn(move || { + let _ = done_tx.send(loader.load_config(policy_config).is_err()); + }); + + let rejected = done_rx + .recv_timeout(std::time::Duration::from_secs(10)) + .expect("a rejected load deadlocked dropping its own plugins"); + assert!(rejected, "a conflicting plugin name must reject the load"); + assert_eq!( + engine.plugin_count(), + 1, + "a rejected load publishes nothing, leaving the earlier plugin", + ); + } + + /// A rejected mutation must publish nothing and leave the generation + /// alone, so a downstream cache keyed on the generation does not evict + /// and rebuild over a registration that never happened. + #[test] + fn failed_registration_publishes_nothing() { + let mgr = PolicyEngine::default(); + let cfg = make_config("dupe", 10, PluginMode::Sequential); + let plugin: Arc = Arc::new(AllowPlugin { cfg: cfg.clone() }); + mgr.register_handler::(plugin, cfg.clone()) + .expect("first registration must succeed"); + + let generation_after_first = mgr.config_generation(); + + let dupe: Arc = Arc::new(AllowPlugin { cfg: cfg.clone() }); + mgr.register_handler::(dupe, cfg) + .expect_err("a duplicate name must be rejected"); + + assert_eq!(mgr.plugin_count(), 1, "the rejected plugin must not appear"); + assert_eq!( + mgr.config_generation(), + generation_after_first, + "a failed mutation must not bump the generation", + ); + } + #[tokio::test] async fn test_audit_plugin_cannot_block() { let mgr = PolicyEngine::default();