Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 51 additions & 0 deletions crates/tinybus/src/connection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,58 @@ pub struct Connection {
_close: Arc<CloseOnDrop>,
}

/// 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<Inner>,
close: std::sync::Weak<CloseOnDrop>,
}

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<Connection> {
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
Expand Down
95 changes: 95 additions & 0 deletions crates/tinybus/src/connection_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -714,3 +714,98 @@ 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: WeakConnection,
dropped: Arc<std::sync::atomic::AtomicBool>,
}

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<MemberName> {
vec![MemberName::new("Owner").unwrap()]
}
async fn call(&self, _member: &MemberName, _args: Value) -> Result<Value> {
Ok(serde_json::json!(
self.connection
.upgrade()
.ok_or(Error::ConnectionClosed)?
.unique_name()
.map(|name| name.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.downgrade(),
dropped: dropped.clone(),
},
)
.await?;
service.request_name("ai.test.Callback").await?;
let owner: Option<String> = client
.proxy("ai.test.Callback", "/test/Callback", "ai.test.Callback")?
.call("Owner", ())
.await?;
assert_eq!(owner, service.unique_name().map(|name| name.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(())
}

#[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<Inner> 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(())
}
2 changes: 1 addition & 1 deletion crates/tinybus/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
17 changes: 17 additions & 0 deletions docs/modules/connection/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading