From eb80befe15bcff339a1c0f696546fd7c1be38d10 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 21:16:52 +0300 Subject: [PATCH 1/4] test(connection): cover callback connection not retaining served object Add a test that serves an object holding a clone of its own connection and verifies that dropping the service connection releases both the served object and its requested bus name, guarding against reference cycles between a connection and the objects it serves. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection_tests.rs | 70 ++++++++++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/crates/tinybus/src/connection_tests.rs b/crates/tinybus/src/connection_tests.rs index c09a8828..7f3f2c24 100644 --- a/crates/tinybus/src/connection_tests.rs +++ b/crates/tinybus/src/connection_tests.rs @@ -714,3 +714,73 @@ async fn unserve_stops_new_admission_but_preserves_an_already_admitted_call() { release.notify_one(); assert_eq!(active.await.unwrap().unwrap(), serde_json::json!([1])); } + +/// A served object needs its connection to issue callbacks, but must not own it. +struct CallbackConnection { + connection: Connection, + dropped: Arc, +} + +impl Drop for CallbackConnection { + fn drop(&mut self) { + self.dropped + .store(true, std::sync::atomic::Ordering::SeqCst); + } +} + +#[async_trait] +impl Interface for CallbackConnection { + fn name(&self) -> InterfaceName { + InterfaceName::new("ai.test.Callback").unwrap() + } + fn members(&self) -> Vec { + vec![MemberName::new("Owner").unwrap()] + } + async fn call(&self, _member: &MemberName, _args: Value) -> Result { + Ok(serde_json::json!( + self.connection.unique_name().map(ToString::to_string) + )) + } +} + +#[tokio::test] +async fn callback_connection_does_not_keep_its_served_object_and_bus_name_alive() -> Result<()> { + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let service = Connection::connect(bus.connect().await?).await?; + let client = Connection::connect(bus.connect().await?).await?; + let dropped = Arc::new(std::sync::atomic::AtomicBool::new(false)); + service + .serve_at( + ObjectPath::new("/test/Callback")?, + CallbackConnection { + connection: service.clone(), + dropped: dropped.clone(), + }, + ) + .await?; + service.request_name("ai.test.Callback").await?; + let owner: Option = client + .proxy("ai.test.Callback", "/test/Callback", "ai.test.Callback")? + .call("Owner", ()) + .await?; + assert_eq!(owner, service.unique_name().map(ToString::to_string)); + drop(service); + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if dropped.load(std::sync::atomic::Ordering::SeqCst) + && !client + .list_names() + .await? + .iter() + .any(|name| name.as_str() == "ai.test.Callback") + { + return Result::Ok(()); + } + tokio::task::yield_now().await; + } + }) + .await + .map_err(|_| Error::failed("callback connection retained its object/name"))??; + Ok(()) +} From 0fda923c80cd4a469cf2c42cfa43efe33b351c40 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 21:18:14 +0300 Subject: [PATCH 2/4] test(tinybus): replace ToString::to_string with closure in connection tests Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection_tests.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/tinybus/src/connection_tests.rs b/crates/tinybus/src/connection_tests.rs index 7f3f2c24..82262a9a 100644 --- a/crates/tinybus/src/connection_tests.rs +++ b/crates/tinybus/src/connection_tests.rs @@ -738,7 +738,7 @@ impl Interface for CallbackConnection { } async fn call(&self, _member: &MemberName, _args: Value) -> Result { Ok(serde_json::json!( - self.connection.unique_name().map(ToString::to_string) + self.connection.unique_name().map(|name| name.to_string()) )) } } @@ -764,7 +764,7 @@ async fn callback_connection_does_not_keep_its_served_object_and_bus_name_alive( .proxy("ai.test.Callback", "/test/Callback", "ai.test.Callback")? .call("Owner", ()) .await?; - assert_eq!(owner, service.unique_name().map(ToString::to_string)); + assert_eq!(owner, service.unique_name().map(|name| name.to_string())); drop(service); tokio::time::timeout(Duration::from_secs(1), async { loop { From 5d6a646b60e592996114b690a58ef756e19bbf13 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 21:19:19 +0300 Subject: [PATCH 3/4] feat(connection): add WeakConnection for non-owning callback access Served objects need their connection to issue callbacks, but holding a strong handle creates an ownership cycle that keeps the service, transport and well-known name alive. WeakConnection borrows the connection only for the duration of a callback and cannot resurrect it once the last strong owner drops, even while dispatch state still holds an internal reference. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection.rs | 51 ++++++++++++++++++++++++++ crates/tinybus/src/connection_tests.rs | 29 ++++++++++++++- crates/tinybus/src/lib.rs | 2 +- 3 files changed, 79 insertions(+), 3 deletions(-) diff --git a/crates/tinybus/src/connection.rs b/crates/tinybus/src/connection.rs index 76dc60a9..b19d66b6 100644 --- a/crates/tinybus/src/connection.rs +++ b/crates/tinybus/src/connection.rs @@ -232,7 +232,58 @@ pub struct Connection { _close: Arc, } +/// A non-owning connection reference for services issuing callbacks. +/// +/// A served object must not retain its own strong connection: the connection +/// owns that object tree, so doing so creates a cycle that keeps the service, +/// transport and well-known name alive. Upgrade only for the duration of an +/// actual callback. The reference never keeps the connection open or revives +/// it after its last strong owner goes away. +#[derive(Clone)] +pub struct WeakConnection { + inner: std::sync::Weak, + close: std::sync::Weak, +} + +impl std::fmt::Debug for WeakConnection { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("WeakConnection") + .field("has_strong_owner", &(self.close.strong_count() != 0)) + .finish_non_exhaustive() + } +} + +impl WeakConnection { + /// Borrow an owning handle while another strong connection still exists. + /// + /// Dispatch tasks can retain internal state after the last public handle + /// drops. Upgrading the lifetime guard first prevents that state from + /// resurrecting a service whose transport is already being closed. + /// A still-owned but explicitly closed connection can be upgraded; calls + /// on it return the same closed-transport errors as its owning handles. + #[must_use] + pub fn upgrade(&self) -> Option { + let close = self.close.upgrade()?; + Some(Connection { + inner: self.inner.upgrade()?, + _close: close, + }) + } +} + impl Connection { + /// Retain callback access without owning this connection's lifetime. + /// + /// Store this reference in objects served by the same connection to avoid + /// a connection/object ownership cycle. Upgrade it only while issuing work. + #[must_use] + pub fn downgrade(&self) -> WeakConnection { + WeakConnection { + inner: Arc::downgrade(&self.inner), + close: Arc::downgrade(&self._close), + } + } + /// Attach to the bus over `transport` and complete the `Hello` handshake. /// /// Returns once the broker has assigned a unique name, so diff --git a/crates/tinybus/src/connection_tests.rs b/crates/tinybus/src/connection_tests.rs index 82262a9a..1b9436f7 100644 --- a/crates/tinybus/src/connection_tests.rs +++ b/crates/tinybus/src/connection_tests.rs @@ -717,7 +717,7 @@ async fn unserve_stops_new_admission_but_preserves_an_already_admitted_call() { /// A served object needs its connection to issue callbacks, but must not own it. struct CallbackConnection { - connection: Connection, + connection: WeakConnection, dropped: Arc, } @@ -738,7 +738,11 @@ impl Interface for CallbackConnection { } async fn call(&self, _member: &MemberName, _args: Value) -> Result { Ok(serde_json::json!( - self.connection.unique_name().map(|name| name.to_string()) + self.connection + .upgrade() + .ok_or(Error::ConnectionClosed)? + .unique_name() + .map(|name| name.to_string()) )) } } @@ -784,3 +788,24 @@ async fn callback_connection_does_not_keep_its_served_object_and_bus_name_alive( .map_err(|_| Error::failed("callback connection retained its object/name"))??; Ok(()) } + +#[tokio::test] +async fn weak_references_cannot_resurrect_dispatch_state_after_the_last_owner_drops() -> Result<()> +{ + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let connection = Connection::connect(bus.connect().await?).await?; + let weak = connection.downgrade(); + let weak_clone = weak.clone(); + let upgraded = weak.upgrade().ok_or(Error::ConnectionClosed)?; + assert_eq!(upgraded.unique_name(), connection.unique_name()); + drop(connection); + assert!(weak_clone.upgrade().is_some()); + drop(upgraded); + // The dispatch task still has its own Arc until it sees closure. + // That is not permission to recreate the lifetime guard. + assert!(weak.upgrade().is_none()); + assert!(weak_clone.upgrade().is_none()); + assert!(format!("{weak:?}").contains("has_strong_owner: false")); + Ok(()) +} diff --git a/crates/tinybus/src/lib.rs b/crates/tinybus/src/lib.rs index fc8d19be..6068de80 100644 --- a/crates/tinybus/src/lib.rs +++ b/crates/tinybus/src/lib.rs @@ -92,7 +92,7 @@ pub mod version; pub mod __private; pub use crate::attest::Attestation; -pub use crate::connection::Connection; +pub use crate::connection::{Connection, WeakConnection}; pub use crate::error::{Error, Result}; pub use crate::events::{ Event, EventBus, EventBusConfig, EventHandler, EventReceiver, SubscriptionHandle, TryRecvError, From 10579d486f276c78c9c4a2b5eaa8bfee0e302292 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 11 Oct 2026 21:24:47 +0300 Subject: [PATCH 4/4] test(connection): use weak reference in callback connection test The callback connection test now stores a downgraded connection so the served object no longer keeps its own connection alive, matching the ownership model documented for callback objects. The connection module README gains a section explaining why served objects should hold weak references and how upgrade behaves across the connection lifetime. Auto-committed-on: dragonfly Co-authored-by: Medulla --- crates/tinybus/src/connection_tests.rs | 2 +- docs/modules/connection/README.md | 17 +++++++++++++++++ 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/crates/tinybus/src/connection_tests.rs b/crates/tinybus/src/connection_tests.rs index 1b9436f7..426da268 100644 --- a/crates/tinybus/src/connection_tests.rs +++ b/crates/tinybus/src/connection_tests.rs @@ -758,7 +758,7 @@ async fn callback_connection_does_not_keep_its_served_object_and_bus_name_alive( .serve_at( ObjectPath::new("/test/Callback")?, CallbackConnection { - connection: service.clone(), + connection: service.downgrade(), dropped: dropped.clone(), }, ) diff --git a/docs/modules/connection/README.md b/docs/modules/connection/README.md index 9541d70c..ccb9357d 100644 --- a/docs/modules/connection/README.md +++ b/docs/modules/connection/README.md @@ -36,6 +36,23 @@ Without it, an exiting service would keep its well-known name until its process died, and `NameOwnerChanged` — the whole reason the kernel can react to a death — would not fire. +## Callback ownership + +A served object that needs to call another service stores +a non-owning callback reference from `Connection::downgrade()`. Upgrade the +`WeakConnection` for the duration of each callback. Storing a strong clone in +an object served by that same connection creates a cycle: connection → object +→ connection. Its external owner can then drop without releasing the object +or well-known name. + +Weak references never keep the link open. `upgrade()` returns `None` after +the last strong owner drops, even while dispatch tasks still retain internal +state. A successful upgrade owns a normal connection for that work's duration; +it preserves the existing transport, routing, timeout and admission behavior. +It does not recreate a dropped lifetime guard. An explicitly closed connection +that still has an owner can be upgraded; calls return its normal closed-link +errors. + ## Argument normalisation `to_body` wraps a scalar into the positional array the protocol specifies, so