commit 1ce1da677c6992ce87129a59cd6db2a9a720e401
parent 4f13a8cf263fd45dd2794270a40058b752f3a27c
Author: triesap <tyson@radroots.org>
Date: Mon, 3 Aug 2026 10:29:56 +0000
sdk: define explicit client lifecycle and shutdown
- share lifecycle state across every client clone
- require explicit asynchronous storage shutdown
- expose deterministic closing and closed errors
- verify cancellation and concurrent close behavior
Diffstat:
3 files changed, 221 insertions(+), 24 deletions(-)
diff --git a/crates/sdk/src/client.rs b/crates/sdk/src/client.rs
@@ -1,6 +1,9 @@
//! Client construction and lifecycle.
-use std::sync::Arc;
+use std::sync::{
+ Arc,
+ atomic::{AtomicU8, Ordering},
+};
use radroots_signing::Signer;
use radroots_storage::Storage;
@@ -34,8 +37,14 @@ struct ClientInner {
sink: Option<Arc<dyn EventSink>>,
#[cfg(feature = "sync")]
sync: Option<radroots_sync::Engine>,
+ lifecycle: AtomicU8,
}
+const OPEN: u8 = 0;
+const CLOSING: u8 = 1;
+const CLOSE_RETRY_REQUIRED: u8 = 2;
+const CLOSED: u8 = 3;
+
impl ClientBuilder {
/// Creates an empty builder with no hidden storage, network, signing, or
/// runtime side effects.
@@ -101,6 +110,7 @@ impl ClientBuilder {
sink: self.sink,
#[cfg(feature = "sync")]
sync: self.sync,
+ lifecycle: AtomicU8::new(OPEN),
}),
})
}
@@ -108,34 +118,111 @@ impl ClientBuilder {
impl Client {
/// Returns the injected canonical storage capability.
- #[must_use]
- pub fn storage(&self) -> &dyn Storage {
- self.inner.storage.as_ref()
+ pub fn storage(&self) -> Result<&dyn Storage> {
+ self.require_open()?;
+ Ok(self.inner.storage.as_ref())
}
/// Returns the injected signer, when outbound authoring is enabled.
- #[must_use]
- pub fn signer(&self) -> Option<&dyn Signer> {
- self.inner.signer.as_deref()
+ pub fn signer(&self) -> Result<Option<&dyn Signer>> {
+ self.require_open()?;
+ Ok(self.inner.signer.as_deref())
}
/// Returns the injected inbound source, when pull is enabled.
- #[must_use]
- pub fn source(&self) -> Option<&dyn EventSource> {
- self.inner.source.as_deref()
+ pub fn source(&self) -> Result<Option<&dyn EventSource>> {
+ self.require_open()?;
+ Ok(self.inner.source.as_deref())
}
/// Returns the injected outbound sink, when delivery is enabled.
- #[must_use]
- pub fn sink(&self) -> Option<&dyn EventSink> {
- self.inner.sink.as_deref()
+ pub fn sink(&self) -> Result<Option<&dyn EventSink>> {
+ self.require_open()?;
+ Ok(self.inner.sink.as_deref())
}
/// Returns the explicit synchronization engine, when configured.
#[cfg(feature = "sync")]
+ pub fn sync_engine(&self) -> Result<Option<&radroots_sync::Engine>> {
+ self.require_open()?;
+ Ok(self.inner.sync.as_ref())
+ }
+
+ /// Returns whether explicit close completed successfully or reached the
+ /// lower storage commit point.
#[must_use]
- pub fn sync_engine(&self) -> Option<&radroots_sync::Engine> {
- self.inner.sync.as_ref()
+ pub fn is_closed(&self) -> bool {
+ self.inner.lifecycle.load(Ordering::Acquire) == CLOSED
+ }
+
+ /// Explicitly closes active storage resources across every client clone.
+ ///
+ /// Dropping this future before its first poll has no effect. Cancellation
+ /// after close begins leaves the client unavailable and permits an
+ /// explicit retry; it never reports rollback. Repeated completed close is
+ /// idempotent. No worker, executor, or blocking `Drop` path is installed.
+ pub async fn close(&self) -> Result<()> {
+ loop {
+ match self.inner.lifecycle.load(Ordering::Acquire) {
+ CLOSED => return Ok(()),
+ CLOSING => return Err(Error::CloseInProgress),
+ state @ (OPEN | CLOSE_RETRY_REQUIRED) => {
+ if self
+ .inner
+ .lifecycle
+ .compare_exchange(state, CLOSING, Ordering::AcqRel, Ordering::Acquire)
+ .is_ok()
+ {
+ break;
+ }
+ }
+ _ => return Err(Error::ClientClosed),
+ }
+ }
+
+ let attempt = CloseAttempt::new(Arc::clone(&self.inner));
+ let close_result = radroots_storage::BackupSource::close(self.inner.storage.as_ref()).await;
+ attempt.complete();
+ close_result
+ .map(|_| ())
+ .map_err(|_| Error::StorageCloseFailed)
+ }
+
+ fn require_open(&self) -> Result<()> {
+ match self.inner.lifecycle.load(Ordering::Acquire) {
+ OPEN => Ok(()),
+ CLOSING | CLOSE_RETRY_REQUIRED => Err(Error::ClientClosing),
+ _ => Err(Error::ClientClosed),
+ }
+ }
+}
+
+struct CloseAttempt {
+ inner: Arc<ClientInner>,
+ completed: bool,
+}
+
+impl CloseAttempt {
+ fn new(inner: Arc<ClientInner>) -> Self {
+ Self {
+ inner,
+ completed: false,
+ }
+ }
+
+ fn complete(mut self) {
+ self.inner.lifecycle.store(CLOSED, Ordering::Release);
+ self.completed = true;
+ }
+}
+
+impl Drop for CloseAttempt {
+ fn drop(&mut self) {
+ if !self.completed {
+ self.inner
+ .lifecycle
+ .store(CLOSE_RETRY_REQUIRED, Ordering::Release);
+ }
}
}
@@ -146,6 +233,7 @@ impl std::fmt::Debug for Client {
.field("signer", &self.inner.signer.is_some())
.field("source", &self.inner.source.is_some())
.field("sink", &self.inner.sink.is_some())
+ .field("closed", &self.is_closed())
.finish_non_exhaustive()
}
}
@@ -173,6 +261,11 @@ mod tests {
DeliveryReceipt, DeliveryRequest, Error as TransportError, FetchPage, FetchRequest,
SinkStatus, SourceStatus, source::BoxFuture as TransportFuture,
};
+ use std::{
+ future::Future,
+ sync::Arc,
+ task::{Context, Poll, Wake, Waker},
+ };
struct TestSource;
struct TestSink;
@@ -238,23 +331,23 @@ mod tests {
#[test]
fn memory_local_source_only_and_sink_only_compositions_are_explicit() {
let local = ClientBuilder::memory(generation()).build().expect("local");
- assert!(local.source().is_none());
- assert!(local.sink().is_none());
- assert!(local.signer().is_none());
+ assert!(local.source().expect("source capability").is_none());
+ assert!(local.sink().expect("sink capability").is_none());
+ assert!(local.signer().expect("signer capability").is_none());
let source = ClientBuilder::memory(generation())
.source(Arc::new(TestSource))
.build()
.expect("source-only");
- assert!(source.source().is_some());
- assert!(source.sink().is_none());
+ assert!(source.source().expect("source capability").is_some());
+ assert!(source.sink().expect("sink capability").is_none());
let sink = ClientBuilder::memory(generation())
.sink(Arc::new(TestSink))
.build()
.expect("sink-only");
- assert!(sink.source().is_none());
- assert!(sink.sink().is_some());
+ assert!(sink.source().expect("source capability").is_none());
+ assert!(sink.sink().expect("sink capability").is_some());
}
#[test]
@@ -264,10 +357,81 @@ mod tests {
.signer(Arc::new(TestSigner))
.build()
.expect("outbound client");
- assert!(client.signer().is_some());
+ assert!(client.signer().expect("signer capability").is_some());
assert_eq!(
format!("{client:?}"),
- "Client { signer: true, source: false, sink: true, .. }"
+ "Client { signer: true, source: false, sink: true, closed: false, .. }"
);
}
+
+ #[test]
+ fn close_is_clone_shared_idempotent_and_rejects_later_capability_access() {
+ let client = ClientBuilder::memory(generation()).build().expect("client");
+ let clone = client.clone();
+ assert!(!client.is_closed());
+ block_on(client.close()).expect("first close");
+ assert!(clone.is_closed());
+ block_on(clone.close()).expect("repeated close");
+ assert!(matches!(clone.storage(), Err(Error::ClientClosed)));
+ assert!(matches!(client.source(), Err(Error::ClientClosed)));
+ }
+
+ #[test]
+ fn close_cancellation_boundaries_are_explicit_and_retryable() {
+ let client = ClientBuilder::memory(generation()).build().expect("client");
+ let unpolled = client.close();
+ drop(unpolled);
+ assert!(client.storage().is_ok());
+
+ client.inner.lifecycle.store(CLOSING, Ordering::Release);
+ let attempt = CloseAttempt::new(Arc::clone(&client.inner));
+ drop(attempt);
+ assert!(matches!(client.storage(), Err(Error::ClientClosing)));
+ block_on(client.close()).expect("retry close");
+ assert!(client.is_closed());
+ }
+
+ #[test]
+ fn concurrent_clones_converge_on_one_closed_state() {
+ let client = ClientBuilder::memory(generation()).build().expect("client");
+ let first = client.clone();
+ let second = client.clone();
+ let outcomes = std::thread::scope(|scope| {
+ let first_close = scope.spawn(move || block_on(first.close()));
+ let second_close = scope.spawn(move || block_on(second.close()));
+ [
+ first_close.join().expect("first thread"),
+ second_close.join().expect("second thread"),
+ ]
+ });
+ assert!(
+ outcomes.iter().all(|outcome| {
+ outcome.is_ok() || matches!(outcome, Err(Error::CloseInProgress))
+ })
+ );
+ if !client.is_closed() {
+ block_on(client.close()).expect("finish close");
+ }
+ assert!(client.is_closed());
+ }
+
+ struct ThreadWaker;
+
+ impl Wake for ThreadWaker {
+ fn wake(self: Arc<Self>) {
+ std::thread::current().unpark();
+ }
+ }
+
+ fn block_on<F: Future>(future: F) -> F::Output {
+ let waker = Waker::from(Arc::new(ThreadWaker));
+ let mut context = Context::from_waker(&waker);
+ let mut future = Box::pin(future);
+ loop {
+ match future.as_mut().poll(&mut context) {
+ Poll::Ready(output) => return output,
+ Poll::Pending => std::thread::park(),
+ }
+ }
+ }
}
diff --git a/crates/sdk/src/error.rs b/crates/sdk/src/error.rs
@@ -9,6 +9,14 @@ pub enum Error {
MissingStorage,
/// A signer cannot be selected without an outbound event sink.
SignerWithoutSink,
+ /// Another clone is actively closing shared resources.
+ CloseInProgress,
+ /// Close was cancelled after beginning and requires an explicit retry.
+ ClientClosing,
+ /// The client has completed explicit shutdown.
+ ClientClosed,
+ /// The storage close operation failed after shutdown began.
+ StorageCloseFailed,
}
impl fmt::Display for Error {
@@ -18,6 +26,12 @@ impl fmt::Display for Error {
Self::SignerWithoutSink => {
formatter.write_str("SDK signer requires an outbound event sink")
}
+ Self::CloseInProgress => formatter.write_str("SDK client close is in progress"),
+ Self::ClientClosing => {
+ formatter.write_str("SDK client close requires completion or retry")
+ }
+ Self::ClientClosed => formatter.write_str("SDK client is closed"),
+ Self::StorageCloseFailed => formatter.write_str("SDK storage close failed"),
}
}
}
diff --git a/crates/sdk/tests/package_boundary.rs b/crates/sdk/tests/package_boundary.rs
@@ -2,6 +2,7 @@ use std::collections::BTreeSet;
const MANIFEST: &str = include_str!("../Cargo.toml");
const ROOT: &str = include_str!("../src/lib.rs");
+const CLIENT: &str = include_str!("../src/client.rs");
#[test]
fn manifest_has_final_identity_and_dependency_boundary() {
@@ -89,6 +90,24 @@ fn nostr_adapter_implements_independent_source_and_sink_capabilities() {
assert_sink::<radroots_transport_nostr::NostrTransport>();
}
+#[test]
+fn lifecycle_source_owns_no_hidden_worker_runtime_or_blocking_drop() {
+ for forbidden in [
+ "tokio::spawn",
+ "std::thread::spawn",
+ "Runtime::new",
+ "block_on(self.close",
+ "impl Drop for Client",
+ ] {
+ assert!(
+ !CLIENT.contains(forbidden),
+ "forbidden lifecycle source `{forbidden}`"
+ );
+ }
+ assert!(CLIENT.contains("pub async fn close(&self)"));
+ assert!(CLIENT.contains("impl Drop for CloseAttempt"));
+}
+
fn dependency_names(manifest: &str) -> BTreeSet<&str> {
let dependencies = manifest
.split_once("[dependencies]")
