diff --git a/crates/tinybus-module/src/lib.rs b/crates/tinybus-module/src/lib.rs index 468175b6..396be457 100644 --- a/crates/tinybus-module/src/lib.rs +++ b/crates/tinybus-module/src/lib.rs @@ -757,6 +757,83 @@ unsafe fn start_reconfigurable_module_mode( setup: F, linked: bool, ) -> i32 +where + C: serde::de::DeserializeOwned + Send + 'static, + F: Fn(Connection, C) -> Fut + Clone + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + unsafe { + start_reconfigurable_module_factory_mode( + host, + out, + worker_threads, + detach_on_panic, + || setup, + linked, + ) + } +} + +/// Initialize one configured instance with fresh captured setup state. +#[doc(hidden)] +pub unsafe fn start_reconfigurable_module_with_factory( + host: *const TbHostVtable, + out: *mut TbModuleVtable, + worker_threads: usize, + detach_on_panic: bool, + factory: impl FnOnce() -> F, +) -> i32 +where + C: serde::de::DeserializeOwned + Send + 'static, + F: Fn(Connection, C) -> Fut + Clone + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + unsafe { + start_reconfigurable_module_factory_mode( + host, + out, + worker_threads, + detach_on_panic, + factory, + false, + ) + } +} + +/// Initialize one linked configured instance with fresh captured setup state. +#[doc(hidden)] +pub unsafe fn start_linked_reconfigurable_module_with_factory( + host: *const TbHostVtable, + out: *mut TbModuleVtable, + worker_threads: usize, + detach_on_panic: bool, + factory: impl FnOnce() -> F, +) -> i32 +where + C: serde::de::DeserializeOwned + Send + 'static, + F: Fn(Connection, C) -> Fut + Clone + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + unsafe { + start_reconfigurable_module_factory_mode( + host, + out, + worker_threads, + detach_on_panic, + factory, + true, + ) + } +} + +unsafe fn start_reconfigurable_module_factory_mode( + host: *const TbHostVtable, + out: *mut TbModuleVtable, + worker_threads: usize, + detach_on_panic: bool, + factory: impl FnOnce() -> F, + linked: bool, +) -> i32 where C: serde::de::DeserializeOwned + Send + 'static, F: Fn(Connection, C) -> Fut + Clone + Send + Sync + 'static, @@ -768,6 +845,10 @@ where Ok(Err(code)) => return code, Err(_) => return TB_PANICKED, }; + let setup = match catch_unwind(AssertUnwindSafe(factory)) { + Ok(setup) => setup, + Err(_) => return TB_PANICKED, + }; let initial_setup = setup.clone(); let reinitialize_factory: ReinitializeFactory = Box::new(move |handle, connection_slot| { let reinitialize_lock = StdMutex::new(()); @@ -819,6 +900,15 @@ where /// ```ignore /// tinybus_module::module_export! { setup = setup, worker_threads = 1 } /// ``` +/// +/// Configured modules may use `setup_factory = factory` in place of `setup`. +/// After valid configuration is parsed, the safe no-argument factory runs once +/// per ABI initialization and returns a +/// cloneable `Fn(Connection, Config) -> Future>`. Its captured +/// state is reused for that instance's reinitializations, allowing admission to +/// remain bound without process-global state. Separate instances receive fresh +/// state. A factory panic returns `TB_PANICKED` without unwinding into the host. +/// Existing `setup = function` callbacks retain their behavior. #[macro_export] macro_rules! module_export { (@linked) => { @@ -956,6 +1046,19 @@ macro_rules! module_export { unsafe { $start(host, out, $threads, true, $setup) } } }; + ( + setup_factory = $factory:path, + config = $config:ty, + $($rest:tt)* + ) => { + $crate::module_export! { + @configured export = {#[unsafe(no_mangle)]}, + start = $crate::start_reconfigurable_module_with_factory, + setup = $factory, + config = $config, + $($rest)* + } + }; ( setup = $setup:path, config = $config:ty, @@ -981,8 +1084,19 @@ macro_rules! module_export { /// Generate Rust-addressable entry points for linking several modules into a /// single executable. The entry points have no shared C linker symbol names. +/// Both configured `setup` and per-instance `setup_factory` forms are supported. #[macro_export] macro_rules! module_export_static { + (setup_factory = $factory:path, config = $config:ty, $($rest:tt)*) => { + $crate::module_export! { + @configured export = {}, + start = $crate::start_linked_reconfigurable_module_with_factory, + setup = $factory, + config = $config, + $($rest)* + } + $crate::module_export! { @linked } + }; (setup = $setup:path, config = $config:ty, $($rest:tt)*) => { $crate::module_export! { @configured export = {}, @@ -1026,3 +1140,7 @@ mod tests; #[cfg(test)] #[path = "static_link_tests.rs"] mod static_link_tests; + +#[cfg(test)] +#[path = "setup_factory_tests.rs"] +mod setup_factory_test; diff --git a/crates/tinybus-module/src/setup_factory_tests.rs b/crates/tinybus-module/src/setup_factory_tests.rs new file mode 100644 index 00000000..f41a6adb --- /dev/null +++ b/crates/tinybus-module/src/setup_factory_tests.rs @@ -0,0 +1,208 @@ +use super::*; +use std::{ + future::Future, + pin::Pin, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, +}; +use tinybus::{broker::Broker, module::ModuleHost, transport::memory::MemoryBus}; + +type SetupFuture = Pin> + Send>>; +type SetupFn = fn(Connection, Config) -> SetupFuture; +static TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); +static FACTORIES: AtomicUsize = AtomicUsize::new(0); +#[derive(serde::Deserialize)] +struct Config { + restricted: bool, +} +struct Service { + restricted: bool, +} +#[async_trait::async_trait] +impl tinybus::service::Interface for Service { + fn name(&self) -> tinybus::InterfaceName { + "ai.tinyhumans.tinybus.Factory".try_into().unwrap() + } + fn members(&self) -> Vec { + vec!["Mode".try_into().unwrap()] + } + async fn call( + &self, + _member: &tinybus::MemberName, + _args: serde_json::Value, + ) -> tinybus::Result { + Ok(serde_json::json!(self.restricted)) + } +} +async fn legacy_setup(connection: Connection, config: Config) -> Result<()> { + connection + .serve_at( + "/factory".try_into()?, + Service { + restricted: config.restricted, + }, + ) + .await?; + connection + .request_name("ai.tinyhumans.tinybus.Factory") + .await?; + Ok(()) +} +fn setup_factory() +-> impl Fn(Connection, Config) -> Pin> + Send>> + Clone + Send + Sync +{ + FACTORIES.fetch_add(1, Ordering::SeqCst); + let mode = Arc::new(tokio::sync::Mutex::new(None)); + move |connection, config: Config| { + let mode = Arc::clone(&mode); + Box::pin(async move { + let mut admitted = mode.lock().await; + if admitted.is_some_and(|previous| previous != config.restricted) { + return Err(tinybus::Error::failed("immutable mode")); + } + *admitted = Some(config.restricted); + legacy_setup(connection, config).await + }) + } +} +module_export_static! { + setup_factory = setup_factory, + config = Config, + worker_threads = 1, + provides = ["ai.tinyhumans.tinybus.Factory"], + methods = ["Mode"], signals = [], requires = [], optional = [], lazy = false, +} +async fn instance(restricted: bool) -> (ModuleHost, Connection, tinybus::Proxy) { + let bus = MemoryBus::new(); + let broker = Broker::new(); + broker.clone().spawn(bus.clone()); + let host = ModuleHost::new(broker); + host.attach_linked_with_config( + linked_module().unwrap(), + serde_json::json!({"restricted":restricted}), + ) + .unwrap(); + let caller = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let proxy = caller + .proxy( + "ai.tinyhumans.tinybus.Factory", + "/factory", + "ai.tinyhumans.tinybus.Factory", + ) + .unwrap(); + assert_eq!(proxy.call::("Mode", ()).await.unwrap(), restricted); + (host, caller, proxy) +} +#[tokio::test(flavor = "current_thread")] +async fn compiled_factory_is_per_instance_and_reinitialization_keeps_its_admission() { + let _guard = TEST_LOCK.lock().await; + let before = FACTORIES.load(Ordering::SeqCst); + let ((first, _first_caller, first_proxy), (second, _second_caller, second_proxy)) = + tokio::join!(instance(true), instance(false)); + assert_eq!( + FACTORIES.load(Ordering::SeqCst) - before, + 2, + "factory must run once per ABI init" + ); + let (same, changed) = tokio::join!( + first.reinitialize("tinybus-module", serde_json::json!({"restricted":true})), + second.reinitialize("tinybus-module", serde_json::json!({"restricted":true})), + ); + assert!(same.is_ok()); + assert!(changed.is_err()); + assert!(first_proxy.call::("Mode", ()).await.unwrap()); + assert!(!second_proxy.call::("Mode", ()).await.unwrap()); + assert_eq!( + FACTORIES.load(Ordering::SeqCst) - before, + 2, + "reinitialization must reuse factory state" + ); +} +#[tokio::test(flavor = "current_thread")] +async fn compiled_instance_refuses_mode_change_without_replacing_live_service() { + let _guard = TEST_LOCK.lock().await; + let (host, _caller, proxy) = instance(true).await; + let result = host + .reinitialize("tinybus-module", serde_json::json!({"restricted":false})) + .await; + assert!( + result.is_err(), + "setup factory admission must survive reinitialization" + ); + assert!(proxy.call::("Mode", ()).await.unwrap()); +} +#[tokio::test] +async fn captured_admission_serializes_concurrent_setup_attempts() { + let _guard = TEST_LOCK.lock().await; + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let connection = Connection::connect(bus.connect().await.unwrap()) + .await + .unwrap(); + let setup = setup_factory(); + let (first, second) = tokio::join!( + setup(connection.clone(), Config { restricted: true }), + setup(connection.clone(), Config { restricted: false }) + ); + assert_ne!(first.is_ok(), second.is_ok()); +} + +mod panicking { + use super::*; + fn factory() -> SetupFn { + panic!("factory failed") + } + module_export_static! { + setup_factory = factory, config = Config, worker_threads = 1, + provides = ["ai.tinyhumans.tinybus.PanickingFactory"], + methods = [], signals = [], requires = [], optional = [], lazy = false, + } +} +#[test] +fn factory_panic_is_a_native_initialization_refusal() { + let broker = Broker::new(); + let host = ModuleHost::new(broker); + let error = host + .attach_linked_with_config( + panicking::linked_module().unwrap(), + serde_json::json!({"restricted":true}), + ) + .unwrap_err(); + assert!(matches!(error, tinybus::Error::ModuleRefused { .. })); +} +#[tokio::test] +async fn invalid_initial_configuration_does_not_construct_factory_state() { + let _guard = TEST_LOCK.lock().await; + let before = FACTORIES.load(Ordering::SeqCst); + let host = ModuleHost::new(Broker::new()); + assert!( + host.attach_linked_with_config( + linked_module().unwrap(), + serde_json::json!({"restricted":"invalid"}) + ) + .is_err() + ); + assert_eq!(FACTORIES.load(Ordering::SeqCst), before); +} + +#[test] +fn native_factory_starter_validates_host_before_constructing_state() { + let code = unsafe { + start_reconfigurable_module_with_factory::( + std::ptr::null(), + std::ptr::null_mut(), + 1, + true, + || { + panic!("invalid host must be rejected before factory"); + #[allow(unreachable_code)] + setup_factory() + }, + ) + }; + assert_eq!(code, TB_BAD_ARGUMENT); +} diff --git a/crates/tinybus-module/src/static_link_tests.rs b/crates/tinybus-module/src/static_link_tests.rs index e197efa3..36590150 100644 --- a/crates/tinybus-module/src/static_link_tests.rs +++ b/crates/tinybus-module/src/static_link_tests.rs @@ -56,12 +56,27 @@ mod configured { } mod optional { - async fn setup(_: tinybus::Connection) -> tinybus::Result<()> { + #[derive(serde::Deserialize)] + struct Config {} + + async fn setup(_: tinybus::Connection, _: Config) -> tinybus::Result<()> { Ok(()) } + fn setup_factory() -> impl Fn( + tinybus::Connection, + Config, + ) -> std::pin::Pin< + Box> + Send>, + > + Clone + + Send + + Sync { + |connection, config| Box::pin(setup(connection, config)) + } + crate::module_export_optional_static! { - setup = setup, + setup_factory = setup_factory, + config = Config, worker_threads = 1, provides = ["ai.tinyhumans.tinybus.OptionalStatic"], methods = [], diff --git a/docs/modules/module/README.md b/docs/modules/module/README.md index 16bf09f0..733248cb 100644 --- a/docs/modules/module/README.md +++ b/docs/modules/module/README.md @@ -213,3 +213,20 @@ corresponding `NameOwnerChanged` announcement. See [abi.md](abi.md) for the binary contract and [the protocol](../../protocol.md) for the bus control members. + +### Per-instance setup state + +Configured exports can use `setup_factory = factory` instead of `setup = setup`. +The safe, no-argument factory returns a cloneable setup callback with the same +`Fn(Connection, Config) -> Future>` signature. TinyBus calls +it once after parsing valid initial configuration, then retains its captured +state for that instance's initial setup and every reinitialization. Separate +ABI instances get fresh state; no global registry is needed. + +This permits a module to capture an immutable admission or mode in an Arc and +reject later configuration that would change it before replacing its served +object. The module owns that admission logic. Existing callback serialization, +configuration validation, deadlines and error codes remain authoritative. +Factory panics return the native initialization panic code. The factory form +is available in module_export, module_export_static and +module_export_optional_static; legacy setup callbacks remain supported.