feat(module): add bounded native terminal factories and physical cleanup ownership

crates/tinybus-module/Cargo.toml

@@ -9,6 +9,9 @@ description = "Module-side runtime and export macro for trusted tinybus cdylibs.
 
 [features]
 static-link = []
+# Private, off-by-default instrumentation for compiled native lifecycle fixtures.
+# Not a production API or a source of native authority.
+__native-fixture-support = []
 
 [dependencies]
 async-trait.workspace = true

crates/tinybus-module/src/execution_context.rs

@@ -32,6 +32,18 @@ impl ExecutionOwner {
             live,
         }
     }
+    #[cfg(feature = "__native-fixture-support")]
+    pub(super) fn is_current_for_fixture(&self) -> bool {
+        OWNER.with(|slot| {
+            slot.borrow()
+                .as_ref()
+                .is_some_and(|current| Arc::ptr_eq(&self.live, &current.live))
+        })
+    }
+    #[cfg(feature = "__native-fixture-support")]
+    pub(super) fn is_stopped_for_fixture(&self) -> bool {
+        !self.live.load(Ordering::Acquire)
+    }
     pub(super) fn enter(&self) -> ExecutionGuard {
         let previous = OWNER.with(|slot| slot.replace(Some(self.clone())));
         let dispatch = tracing::dispatcher::set_default(&self.dispatch);

crates/tinybus-module/src/execution_context_tests.rs

@@ -111,3 +111,138 @@ fn unowned_and_reentrant_panic_contexts_never_select_a_host_callback() {
     assert_eq!(count.load(Ordering::SeqCst), 0);
     assert!(owner.panic_location.lock().unwrap().is_none());
 }
+
+static CONFIG_DISPATCH: StdMutex<Option<tracing::Dispatch>> = StdMutex::new(None);
+struct CapturedConfig;
+impl<'de> serde::Deserialize<'de> for CapturedConfig {
+    fn deserialize<D: serde::Deserializer<'de>>(
+        deserializer: D,
+    ) -> std::result::Result<Self, D::Error> {
+        let _ = serde_json::Value::deserialize(deserializer)?;
+        *CONFIG_DISPATCH.lock().unwrap() = Some(tracing::dispatcher::get_default(Clone::clone));
+        Ok(Self)
+    }
+}
+
+#[test]
+fn shutdown_suppresses_the_exact_factory_and_configuration_caller_dispatch() {
+    for factory in [true, false] {
+        let (fixture, count) = owner();
+        let mut host = fixture.host.0;
+        host.config = tinybus::module::abi::TbSlice {
+            ptr: b"{}".as_ptr(),
+            len: 2,
+        };
+        let captured = Arc::new(StdMutex::new(None));
+        let capture = captured.clone();
+        let mut module = TbModuleVtable::default();
+        let started = if factory {
+            unsafe {
+                start_reconfigurable_module_with_factory::<serde_json::Value, _, _>(
+                    &host,
+                    &mut module,
+                    1,
+                    false,
+                    move || {
+                        *capture.lock().unwrap() =
+                            Some(tracing::dispatcher::get_default(Clone::clone));
+                        |_, _| async { Ok(()) }
+                    },
+                )
+            }
+        } else {
+            unsafe {
+                start_module_with_config::<CapturedConfig, _, _>(
+                    &host,
+                    &mut module,
+                    1,
+                    false,
+                    |_, _| async { Ok(()) },
+                )
+            }
+        };
+        let dispatch = if factory {
+            captured.lock().unwrap().take()
+        } else {
+            CONFIG_DISPATCH.lock().unwrap().take()
+        };
+        let before = count.load(Ordering::SeqCst);
+        if let Some(dispatch) = &dispatch {
+            tracing::dispatcher::with_default(dispatch, || {
+                tracing::info!("captured caller before stop")
+            });
+        }
+        let live = count.load(Ordering::SeqCst);
+        let stopped = if started == TB_OK {
+            unsafe { (module.shutdown)(module.module_ctx, 1000) }
+        } else {
+            TB_CLOSED
+        };
+        let after_shutdown = count.load(Ordering::SeqCst);
+        let (replacement_fixture, replacement_count) = owner();
+        let mut replacement_host = replacement_fixture.host.0;
+        replacement_host.config = host.config;
+        let mut replacement = TbModuleVtable::default();
+        let replaced = unsafe {
+            start_module(&replacement_host, &mut replacement, 1, false, |_| async {
+                Ok(())
+            })
+        };
+        let replacement_before = replacement_count.load(Ordering::SeqCst);
+        if let Some(dispatch) = &dispatch {
+            tracing::dispatcher::with_default(dispatch, || {
+                tracing::info!("captured caller after actual shutdown")
+            });
+        }
+        let late = count.load(Ordering::SeqCst);
+        let replacement_after = replacement_count.load(Ordering::SeqCst);
+        let replacement_stopped = if replaced == TB_OK {
+            unsafe { (replacement.shutdown)(replacement.module_ctx, 1000) }
+        } else {
+            TB_CLOSED
+        };
+        eprintln!(
+            "factory={factory}: start={started}, stop={stopped}, before={before}, live={live}, after_shutdown={after_shutdown}, late={late}, replacement={replacement_before}/{replacement_after}, replacement_stop={replacement_stopped}"
+        );
+        assert_eq!(started, TB_OK);
+        assert!(dispatch.is_some());
+        assert!(
+            live > before,
+            "real caller dispatch forwards before shutdown"
+        );
+        assert_eq!(stopped, TB_OK);
+        assert_eq!(replaced, TB_OK);
+        assert_eq!(replacement_stopped, TB_OK);
+        assert_eq!(
+            late, after_shutdown,
+            "stopped caller dispatch must not forward"
+        );
+        assert_eq!(
+            replacement_after, replacement_before,
+            "old dispatch cannot select replacement"
+        );
+    }
+}
+
+unsafe extern "C" fn tracing_native_query(_: *mut c_void) -> u32 {
+    tracing::info!("terminal delivery query");
+    tinybus::module::abi::TB_NATIVE_TERMINAL_LIVE
+}
+#[test]
+fn terminal_deliver_enters_exact_instance_and_restores_nested_caller() {
+    let (caller, caller_count) = owner();
+    let (instance, instance_count) = owner();
+    let host = HostCalls(TbHostVtable {
+        native_terminal_state: Some(tracing_native_query),
+        ..instance.host.0
+    });
+    let state = super::super::terminal::TerminalState::admit(host, Some(instance)).unwrap();
+    let _outer = caller.enter();
+    let byte = 0;
+    let code = unsafe { super::super::terminal_deliver(state.context(), &byte, 1) };
+    tracing::info!("restored caller");
+    state.empty_factory_failure();
+    assert_eq!(code, TB_CLOSED); // No inbound queue was installed.
+    assert_eq!(instance_count.load(Ordering::SeqCst), 1);
+    assert_eq!(caller_count.load(Ordering::SeqCst), 1);
+}