commit 234824a8aa2d4ccd7c7d158d48afefa5c6dab89b
parent a5d132a3213b40c44f1ee9ec62955f2b6199f77a
Author: triesap <tyson@radroots.org>
Date: Tue, 11 Aug 2026 01:41:30 +0000
service-host: add cancellation contract
- wrap a proven cancel-safe cooperative token
- preserve directional parent and child cancellation
- expose idempotent state and observable waiters
- prove races and drops without installing signals
Diffstat:
7 files changed, 137 insertions(+), 2 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -3884,6 +3884,7 @@ dependencies = [
"serde",
"serde_json",
"tokio",
+ "tokio-util",
]
[[package]]
@@ -5532,6 +5533,7 @@ dependencies = [
"bytes",
"futures-core",
"futures-sink",
+ "futures-util",
"pin-project-lite",
"tokio",
]
diff --git a/Cargo.toml b/Cargo.toml
@@ -271,6 +271,7 @@ tempfile = { version = "3" }
tar = { version = "0.4" }
thiserror = { version = "1" }
tokio = { version = "1" }
+tokio-util = { version = "0.7", features = ["rt"] }
tokio-tungstenite = { version = "0.26.2", default-features = false, features = [
"connect",
"rustls-tls-webpki-roots",
diff --git a/crates/service_host/Cargo.toml b/crates/service_host/Cargo.toml
@@ -17,6 +17,7 @@ radroots_runtime_paths = { workspace = true }
serde = { workspace = true, features = ["derive", "std"] }
serde_json = { workspace = true, features = ["std"] }
tokio = { workspace = true, features = ["sync"] }
+tokio-util = { workspace = true }
[dev-dependencies]
tokio = { workspace = true, features = ["macros", "rt", "sync"] }
diff --git a/crates/service_host/src/lib.rs b/crates/service_host/src/lib.rs
@@ -16,8 +16,8 @@ pub use build_info::{
pub use entropy::{EntropyError, EntropySource, SystemEntropy};
pub use error::{HostError, HostErrorCode, HostErrorKind, SafeHostError};
pub use lifecycle::{
- ShutdownPhase, TaskClassification, TaskCompletionExpectation, TaskMetadata, TaskMetadataError,
- TaskName,
+ CancellationToken, ShutdownPhase, TaskClassification, TaskCompletionExpectation, TaskMetadata,
+ TaskMetadataError, TaskName,
};
pub use status::{
CONFIGURATION_SCHEMA_VERSION, CachedServiceState, CachedServiceStatePublisher,
diff --git a/crates/service_host/src/lifecycle/cancel.rs b/crates/service_host/src/lifecycle/cancel.rs
@@ -0,0 +1,120 @@
+//! Composable cooperative cancellation without process-global signal handling.
+
+use core::fmt;
+
+/// Cloneable cooperative cancellation shared by supervisor-owned tasks.
+///
+/// Child cancellation propagates to descendants but never back to its parent
+/// or sideways to siblings. This type does not install or interpret operating
+/// system signals.
+#[derive(Clone, Default)]
+pub struct CancellationToken {
+ inner: tokio_util::sync::CancellationToken,
+}
+
+impl CancellationToken {
+ #[must_use]
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// Creates a child that is cancelled with this token while retaining its own authority.
+ #[must_use]
+ pub fn child_token(&self) -> Self {
+ Self {
+ inner: self.inner.child_token(),
+ }
+ }
+
+ /// Requests cancellation. Repeated requests have no additional effect.
+ pub fn cancel(&self) {
+ self.inner.cancel();
+ }
+
+ #[must_use]
+ pub fn is_cancelled(&self) -> bool {
+ self.inner.is_cancelled()
+ }
+
+ /// Completes after cancellation and remains immediately ready thereafter.
+ pub async fn cancelled(&self) {
+ self.inner.cancelled().await;
+ }
+}
+
+impl fmt::Debug for CancellationToken {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("CancellationToken")
+ .field("cancelled", &self.is_cancelled())
+ .finish()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[tokio::test]
+ async fn parent_child_propagation_is_directional() {
+ let parent = CancellationToken::new();
+ let child = parent.child_token();
+ let grandchild = child.child_token();
+ let sibling = parent.child_token();
+
+ child.cancel();
+ child.cancelled().await;
+ grandchild.cancelled().await;
+ assert!(!parent.is_cancelled());
+ assert!(!sibling.is_cancelled());
+
+ parent.cancel();
+ parent.cancelled().await;
+ sibling.cancelled().await;
+ assert!(parent.child_token().is_cancelled());
+ }
+
+ #[tokio::test]
+ async fn cancellation_is_idempotent_and_observable_after_the_fact() {
+ let token = CancellationToken::new();
+ token.cancel();
+ token.cancel();
+ assert!(token.is_cancelled());
+ token.cancelled().await;
+ assert_eq!(
+ format!("{token:?}"),
+ "CancellationToken { cancelled: true }"
+ );
+ }
+
+ #[tokio::test]
+ async fn dropping_a_waiter_does_not_consume_cancellation() {
+ let token = CancellationToken::new();
+ let abandoned = token.cancelled();
+ drop(abandoned);
+
+ token.cancel();
+ token.cancelled().await;
+ assert!(token.is_cancelled());
+ }
+
+ #[tokio::test]
+ async fn concurrent_waiter_registration_and_cancel_has_no_lost_wakeup() {
+ for round in 0..64 {
+ let token = CancellationToken::new();
+ let waiter_token = token.clone();
+ let waiter = tokio::spawn(async move {
+ if round % 2 == 0 {
+ tokio::task::yield_now().await;
+ }
+ waiter_token.cancelled().await;
+ waiter_token.is_cancelled()
+ });
+ if round % 2 == 1 {
+ tokio::task::yield_now().await;
+ }
+ token.cancel();
+ assert!(waiter.await.unwrap());
+ }
+ }
+}
diff --git a/crates/service_host/src/lifecycle/mod.rs b/crates/service_host/src/lifecycle/mod.rs
@@ -1,7 +1,9 @@
//! Explicit task ownership and service lifecycle mechanics.
+mod cancel;
mod task;
+pub use cancel::CancellationToken;
pub use task::{
ShutdownPhase, TaskClassification, TaskCompletionExpectation, TaskMetadata, TaskMetadataError,
TaskName,
diff --git a/crates/service_host/tests/package_boundary.rs b/crates/service_host/tests/package_boundary.rs
@@ -8,6 +8,11 @@ const STATUS_SOURCE: &str = concat!(
include_str!("../src/status/reason.rs"),
include_str!("../src/status/service.rs"),
);
+const LIFECYCLE_SOURCE: &str = concat!(
+ include_str!("../src/lifecycle/mod.rs"),
+ include_str!("../src/lifecycle/cancel.rs"),
+ include_str!("../src/lifecycle/task.rs"),
+);
#[test]
fn service_host_is_unpublished_lint_governed_and_dependency_bounded() {
@@ -31,6 +36,7 @@ fn service_host_is_unpublished_lint_governed_and_dependency_bounded() {
"serde",
"serde_json",
"tokio",
+ "tokio-util",
])
);
assert_eq!(
@@ -45,6 +51,9 @@ fn service_host_is_unpublished_lint_governed_and_dependency_bounded() {
])
);
assert!(!STATUS_SOURCE.contains("serde(untagged)"));
+ for forbidden in ["tokio::signal", "ctrl_c", "signal_hook"] {
+ assert!(!LIFECYCLE_SOURCE.contains(forbidden));
+ }
}
fn public_modules(root: &str) -> BTreeSet<&str> {
