diff --git a/CLAUDE.md b/CLAUDE.md index 4df848dd5..d4c52d1eb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1,6 +1,6 @@ # CLAUDE.md -Last verified: 2026-05-19 +Last verified: 2026-06-09 This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. @@ -11,10 +11,12 @@ cargo build # Debug build cargo build --release # Release build (LTO enabled) cargo test # Run all tests cargo test -p buttplug_server # Run tests for specific crate -cargo fmt --all -- --check # Check formatting -cargo fmt # Auto-format (2-space indent, edition 2024) +cargo +nightly fmt --all -- --check # Check formatting (MUST use nightly) +cargo +nightly fmt # Auto-format (2-space indent, edition 2024) ``` +**Formatting gotcha**: rustfmt.toml uses nightly-only options (`imports_layout`, `empty_item_single_line`). Running `cargo fmt` on the STABLE toolchain silently ignores them and rewrites the entire workspace into the wrong style (~190 files of import-collapsing churn). Always use `cargo +nightly fmt`. CI checks formatting with nightly. + **Linux dependencies**: `libudev-dev`, `libusb-1.0-0-dev` (for serial/HID support) **WASM build**: @@ -54,7 +56,7 @@ Buttplug is a framework for interfacing with intimate hardware devices. It uses **Message-Based Protocol**: All client-server communication uses versioned JSON messages (v0-v4). Version negotiation happens during handshake. -**Async Architecture**: Heavy use of tokio channels (mpsc, broadcast, oneshot) for communication between components. Runtime abstraction supports tokio (production) and WASM. +**Async Architecture**: Heavy use of tokio channels (mpsc, broadcast, oneshot) for communication between components. Runtime abstraction supports tokio (production) and WASM. Spawned tasks are owned by `TaskScope`s rather than fire-and-forget `spawn!` (see Task Lifecycle below). **Device Lifecycle**: ``` @@ -93,6 +95,17 @@ Simulated devices allow testing the full device lifecycle without real hardware. - Validation rejects unknown archetypes and duplicate addresses at config build time - `SimulatedProtocol` is a no-op handler; `SimulatedHardwareConnector` creates in-memory endpoints +**Task Lifecycle** (`buttplug_core::util::task`): +Spawned async tasks are owned by a `TaskScope` ownership tree rather than fire-and-forget, giving cooperative cancellation and global introspection. Key contracts: +- `TaskScope::root(name)` makes a root scope (path gets a unique numeric suffix, e.g. `server-2`, so parallel instances don't collide); `.child(name)` derives a sub-scope whose token is a child of the parent's. +- `scope.spawn(name, |token| async ...)` spawns a task owned by the scope; long-running tasks MUST `select!` on the passed token. Cancelling or dropping a scope cancels its whole subtree. +- `scope.spawn_and_hold(name, ...)` consumes the scope into the task, so drop-cancel can't fire before the task runs (used for `FnOnce` callbacks like ping-timeout and protocol subscription handlers). +- `scope.shutdown().await` cancels the subtree and waits until every task under it has deregistered (wrap in a timeout if tasks may be uncooperative). +- `spawn_detached(name, fut)` is a rare escape hatch: registered under `detached/{name}` but uncancellable. Prefer scopes. +- `TaskRegistry` (global, via `registry()`) records every live task: `snapshot()`, `live_count_under(prefix)` (segment-aware prefix match), `event_stream()` (`TaskEvent::Started`/`Ended`), `wait_empty_under(prefix)`. +- Ownership in the server: `ButtplugServer` owns a `server` root scope; `ServerDeviceManager` owns a `device-manager` root scope with per-device child scopes (io/event-forwarding/bringup); `PingTimer` is scope-owned (the old `PingMessage::End` + Drop-spawn shutdown hack is gone). `ProtocolHandler::handle_input_subscribe_cmd` now takes a `TaskScope` param. +- `intiface_engine` exposes this to frontends when `emit_task_events` is set: registry events forward as `EngineMessage::TaskStarted`/`TaskEnded`, and `IntifaceMessage::RequestTaskList` returns `EngineMessage::TaskList` (`Vec`). + ## Agent skills ### Issue tracker diff --git a/CONTEXT.md b/CONTEXT.md index 5ee2983c4..5489a1808 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -92,6 +92,14 @@ _Avoid_: "Discovery" as a distinct stage from identification — they're the sam An opt-in broadcast of every output command sent to a **Device** — device index, feature index, output type, and value. Used by frontends (e.g. Intiface Central) to visually display real-time device activity, verify hardware behaviour matches commands, and let developers see what *would* happen with simulated devices. Disabled by default to avoid overhead. _Avoid_: Treating as internal-only debugging; it's a user-facing observability feature. +**Task Scope**: +The owner of spawned async tasks within a module. Every task is spawned through a Task Scope, which links it to a parent, derives its hierarchical name (e.g. `device-manager-1/devices/device-3/io`), registers it in the **Task Registry**, and hands it a cooperative cancellation token. Dropping a scope cancels its children. Tasks cannot be spawned without a parent scope. +_Avoid_: "Detached task" or bare spawning as the normal pattern; detachment is the rare, explicit exception. + +**Task Registry**: +The queryable record of every live task — id, hierarchical path, parent, state. Populated as a side effect of spawning through a **Task Scope**. Exposed in-process for tests and embedders, and to frontends via TaskStarted/TaskEnded **Events** plus a snapshot query (same opt-in pattern as **Output Observation**). +_Avoid_: Treating as internal-only debugging; like Output Observations, it's user-facing observability. + **Command**: A message from **Client** to **Server** requesting an action — controlling a device, starting a scan, requesting device lists. Always has a non-zero message ID; the server responds with an `Ok` or `Error` using the same ID. _Avoid_: Referring to server-initiated messages as commands. diff --git a/crates/buttplug_core/Cargo.toml b/crates/buttplug_core/Cargo.toml index f979fff80..bb7f5cef0 100644 --- a/crates/buttplug_core/Cargo.toml +++ b/crates/buttplug_core/Cargo.toml @@ -57,3 +57,5 @@ wasm-bindgen-futures = { version = "0.4.64", optional = true } wasmtimer = { version = "0.4.3", optional = true } smallvec = { version = "1.15.1", features = ["serde", "const_generics"] } enumflags2 = "0.7.12" +tokio-util = "0.7.18" +dashmap = "6.1.0" diff --git a/crates/buttplug_core/src/util/mod.rs b/crates/buttplug_core/src/util/mod.rs index 439950aeb..5734abdb4 100644 --- a/crates/buttplug_core/src/util/mod.rs +++ b/crates/buttplug_core/src/util/mod.rs @@ -14,6 +14,7 @@ pub mod range; pub mod serializers; pub mod small_vec_enum_map; pub mod stream; +pub mod task; #[cfg(all(not(feature = "wasm"), feature = "tokio-runtime"))] pub use tokio::time::sleep; diff --git a/crates/buttplug_core/src/util/task/mod.rs b/crates/buttplug_core/src/util/task/mod.rs new file mode 100644 index 000000000..847a917cc --- /dev/null +++ b/crates/buttplug_core/src/util/task/mod.rs @@ -0,0 +1,101 @@ +// Buttplug Rust Source Code File - See https://buttplug.io for more info. +// +// Copyright 2016-2026 Nonpolynomial Labs LLC. All rights reserved. +// +// Licensed under the BSD 3-Clause license. See LICENSE file in the project root +// for full license information. + +//! Task Scope and Task Registry: ownership and introspection for spawned tasks. +//! +//! A task spawned through a [TaskScope] is linked into an ownership tree, given +//! a hierarchical path, registered in the global [TaskRegistry], and handed a +//! cooperative [CancellationToken]; dropping a scope cancels its subtree. The +//! `buttplug_server` crate is fully migrated onto scope-owned tasks; the +//! hardware-manager and client crates still use the bare `spawn!` macro and +//! migrate onto scopes in a follow-up. + +mod registry; +mod scope; + +pub use registry::{ + TaskEvent, + TaskId, + TaskInfo, + TaskOutcome, + TaskRegistry, + TaskSpawnError, + registry, +}; +pub use scope::TaskScope; + +use crate::util::async_manager; +use std::future::Future; + +/// Spawn a task with no owning scope. Registered in the Task Registry under +/// "detached/{name}" so it still shows up in snapshots, but nothing can cancel +/// it. RARE — prefer [TaskScope::spawn]. Valid uses: one-shot notifications +/// where the spawner is being destroyed. +#[cfg(not(feature = "wasm"))] +pub fn spawn_detached(name: &str, fut: Fut) +where + Fut: Future + Send + 'static, +{ + let path = format!("detached/{name}"); + let id = registry().register(path.clone(), true); + let span = tracing::span!(tracing::Level::INFO, "buttplug_task", task.path = %path); + async_manager::spawn( + async move { + // Deregister via a drop guard so a panicking detached task still leaves + // the registry (outcome Panicked) instead of leaking its entry forever. + // Detached tasks have no cancellation token, so the outcome is Panicked on + // panic, else Completed. + let _guard = scope::DeregisterGuard::new(id, None); + fut.await; + }, + span, + ); +} + +/// Spawn a task with no owning scope (WASM, no Send required). See the +/// non-WASM variant for semantics. +#[cfg(feature = "wasm")] +pub fn spawn_detached(name: &str, fut: Fut) +where + Fut: Future + 'static, +{ + let path = format!("detached/{name}"); + let id = registry().register(path.clone(), true); + let span = tracing::span!(tracing::Level::INFO, "buttplug_task", task.path = %path); + async_manager::spawn( + async move { + // Deregister via a drop guard so a panicking detached task still leaves + // the registry (outcome Panicked) instead of leaking its entry forever. + // Detached tasks have no cancellation token, so the outcome is Panicked on + // panic, else Completed. + let _guard = scope::DeregisterGuard::new(id, None); + fut.await; + }, + span, + ); +} + +#[cfg(test)] +mod test { + use super::*; + use std::time::Duration; + + #[tokio::test] + async fn test_spawn_detached_registers() { + let (tx, rx) = tokio::sync::oneshot::channel(); + spawn_detached("test-notify", async move { + let _ = tx.send(()); + }); + rx.await.unwrap(); + tokio::time::timeout( + Duration::from_secs(1), + registry().wait_empty_under("detached/test-notify"), + ) + .await + .expect("detached task did not deregister"); + } +} diff --git a/crates/buttplug_core/src/util/task/registry.rs b/crates/buttplug_core/src/util/task/registry.rs new file mode 100644 index 000000000..361cce292 --- /dev/null +++ b/crates/buttplug_core/src/util/task/registry.rs @@ -0,0 +1,276 @@ +// Buttplug Rust Source Code File - See https://buttplug.io for more info. +// +// Copyright 2016-2026 Nonpolynomial Labs LLC. All rights reserved. +// +// Licensed under the BSD 3-Clause license. See LICENSE file in the project root +// for full license information. + +use dashmap::DashMap; +use std::sync::{ + Mutex, + OnceLock, + atomic::{AtomicU64, Ordering}, +}; +use tokio::sync::broadcast; +use tokio_util::sync::CancellationToken; + +/// Why a task could not be spawned into its scope. +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +pub enum TaskSpawnError { + /// The scope was cancelled before registration could win the race. + #[error("task scope is closed")] + ScopeClosed, +} + +/// Unique identifier for a registered task. Process-lifetime unique. +/// +/// IDs are drawn from a monotonically-incrementing `AtomicU64` counter. At +/// 64-bit width and even at 10 million tasks per second the counter would take +/// roughly 58,000 years to wrap, so id reuse within a single process lifetime +/// is not a practical concern. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct TaskId(u64); + +impl TaskId { + pub fn value(&self) -> u64 { + self.0 + } +} + +/// How a task ended: ran to completion on its own, exited after observing +/// cancellation, or unwound from a panic. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TaskOutcome { + Completed, + Cancelled, + Panicked, +} + +/// Registry entry for a live task. +#[derive(Debug, Clone)] +pub struct TaskInfo { + pub id: TaskId, + /// Hierarchical path, e.g. "device-manager-1/devices/device-3/io". + pub path: String, + /// True if spawned via spawn_detached (no owning scope). + pub detached: bool, +} + +/// Lifecycle events broadcast by the Task Registry. +#[derive(Debug, Clone)] +pub enum TaskEvent { + Started { + id: TaskId, + path: String, + }, + Ended { + id: TaskId, + path: String, + outcome: TaskOutcome, + }, +} + +/// The global record of every live task, populated as a side effect of +/// spawning through a Task Scope. +/// +/// **Memory note**: entries are removed on `deregister`, but `DashMap` does not +/// shrink shard capacity after removals. Peak concurrent-task count therefore +/// becomes a memory high-water mark that is held for the lifetime of the +/// registry (i.e. the process). In practice buttplug servers run a bounded +/// number of concurrent tasks, so this is not expected to be significant. +#[derive(Debug)] +pub struct TaskRegistry { + tasks: DashMap, + /// Short-held gate ordering registration against scope cancellation. + gate: Mutex<()>, + counter: AtomicU64, + root_counter: AtomicU64, + events: broadcast::Sender, +} + +/// The global Task Registry. +pub fn registry() -> &'static TaskRegistry { + static REGISTRY: OnceLock = OnceLock::new(); + REGISTRY.get_or_init(TaskRegistry::new) +} + +impl TaskRegistry { + pub(super) fn new() -> Self { + Self { + tasks: DashMap::new(), + gate: Mutex::new(()), + counter: AtomicU64::new(1), + root_counter: AtomicU64::new(1), + events: broadcast::channel(256).0, + } + } + + /// Unique suffix for root scope names so parallel instances don't collide. + pub(super) fn next_root_suffix(&self) -> u64 { + self.root_counter.fetch_add(1, Ordering::Relaxed) + } + + pub(super) fn register(&self, path: String, detached: bool) -> TaskId { + let _gate = self + .gate + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + self.register_locked(path, detached) + } + + pub(super) fn register_scoped( + &self, + path: String, + token: &CancellationToken, + ) -> Result<(TaskId, CancellationToken), TaskSpawnError> { + let _gate = self + .gate + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if token.is_cancelled() { + return Err(TaskSpawnError::ScopeClosed); + } + let task_token = token.child_token(); + let id = self.register_locked(path, false); + Ok((id, task_token)) + } + + pub(super) fn cancel(&self, token: &CancellationToken) { + let _gate = self + .gate + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + token.cancel(); + } + + #[cfg(test)] + pub(crate) fn test_hold_gate(&self) -> std::sync::MutexGuard<'_, ()> { + self + .gate + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn register_locked(&self, path: String, detached: bool) -> TaskId { + let id = TaskId(self.counter.fetch_add(1, Ordering::Relaxed)); + self.tasks.insert( + id.0, + TaskInfo { + id, + path: path.clone(), + detached, + }, + ); + let _ = self.events.send(TaskEvent::Started { id, path }); + id + } + + pub(super) fn deregister(&self, id: TaskId, outcome: TaskOutcome) { + if let Some((_, info)) = self.tasks.remove(&id.0) { + let _ = self.events.send(TaskEvent::Ended { + id, + path: info.path, + outcome, + }); + } + } + + /// Snapshot of all live tasks. + pub fn snapshot(&self) -> Vec { + self.tasks.iter().map(|e| e.value().clone()).collect() + } + + /// Count of live tasks at or under the given path prefix. Prefix matching is + /// segment-aware: "server-1" matches "server-1/loop" but not "server-10/loop". + pub fn live_count_under(&self, prefix: &str) -> usize { + self + .tasks + .iter() + .filter(|e| path_is_under(&e.value().path, prefix)) + .count() + } + + /// Subscribe to task lifecycle events. + pub fn event_stream(&self) -> broadcast::Receiver { + self.events.subscribe() + } + + /// Wait until no live tasks remain at or under the given path prefix. + /// Subscribes to events BEFORE counting to avoid missing an Ended event + /// between the count and the wait. Callers should wrap in a timeout if the + /// subtree may contain uncooperative tasks. + pub async fn wait_empty_under(&self, prefix: &str) { + let mut events = self.events.subscribe(); + while self.live_count_under(prefix) > 0 { + match events.recv().await { + Ok(_) | Err(broadcast::error::RecvError::Lagged(_)) => continue, + // Sender is owned by the registry itself; Closed only happens for + // non-global registries in tests being dropped. + Err(broadcast::error::RecvError::Closed) => return, + } + } + } +} + +fn path_is_under(path: &str, prefix: &str) -> bool { + path == prefix || (path.starts_with(prefix) && path.as_bytes().get(prefix.len()) == Some(&b'/')) +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn test_register_and_snapshot() { + let reg = TaskRegistry::new(); + let id = reg.register("root-1/loop".to_owned(), false); + let snapshot = reg.snapshot(); + assert_eq!(snapshot.len(), 1); + assert_eq!(snapshot[0].id, id); + assert_eq!(snapshot[0].path, "root-1/loop"); + reg.deregister(id, TaskOutcome::Completed); + assert!(reg.snapshot().is_empty()); + } + + #[test] + fn test_prefix_boundary() { + let reg = TaskRegistry::new(); + reg.register("server-1/loop".to_owned(), false); + reg.register("server-10/loop".to_owned(), false); + // "server-1" must not match "server-10/loop" + assert_eq!(reg.live_count_under("server-1"), 1); + assert_eq!(reg.live_count_under("server-10"), 1); + assert_eq!(reg.live_count_under("server"), 0); + } + + #[tokio::test] + async fn test_events_emitted() { + let reg = TaskRegistry::new(); + let mut events = reg.event_stream(); + let id = reg.register("root-1/task".to_owned(), false); + reg.deregister(id, TaskOutcome::Cancelled); + assert!(matches!( + events.recv().await.unwrap(), + TaskEvent::Started { .. } + )); + let TaskEvent::Ended { outcome, .. } = events.recv().await.unwrap() else { + panic!("expected Ended event"); + }; + assert_eq!(outcome, TaskOutcome::Cancelled); + } + + #[tokio::test] + async fn test_wait_empty_under() { + let reg = std::sync::Arc::new(TaskRegistry::new()); + let id = reg.register("root-1/task".to_owned(), false); + let reg_clone = reg.clone(); + let waiter = tokio::spawn(async move { reg_clone.wait_empty_under("root-1").await }); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + assert!(!waiter.is_finished()); + reg.deregister(id, TaskOutcome::Completed); + tokio::time::timeout(std::time::Duration::from_secs(1), waiter) + .await + .expect("wait_empty_under did not resolve") + .unwrap(); + } +} diff --git a/crates/buttplug_core/src/util/task/scope.rs b/crates/buttplug_core/src/util/task/scope.rs new file mode 100644 index 000000000..1027781b4 --- /dev/null +++ b/crates/buttplug_core/src/util/task/scope.rs @@ -0,0 +1,567 @@ +// Buttplug Rust Source Code File - See https://buttplug.io for more info. +// +// Copyright 2016-2026 Nonpolynomial Labs LLC. All rights reserved. +// +// Licensed under the BSD 3-Clause license. See LICENSE file in the project root +// for full license information. + +use super::registry::{TaskId, TaskOutcome, TaskSpawnError, registry}; +use crate::util::async_manager; +use std::future::Future; +use tokio_util::sync::CancellationToken; + +/// The owner of spawned async tasks within a module. +/// +/// Tasks can only be spawned through a scope, which derives their hierarchical +/// path, registers them in the global Task Registry, and hands them a +/// cooperative [CancellationToken]. Dropping a scope cancels its subtree. +/// +/// Scopes are intentionally NOT [Clone]: ownership is singular. To share +/// spawning capability, create a [child][TaskScope::child] and move it, or +/// wrap a scope in [std::sync::Arc] when a cloneable holder needs it (the +/// subtree then cancels when the last Arc drops). +#[derive(Debug)] +pub struct TaskScope { + path: String, + token: CancellationToken, +} + +impl TaskScope { + /// Create a root scope. The path gets a unique numeric suffix + /// (e.g. "server-2") so parallel instances in one process don't collide. + pub fn root(name: &str) -> Self { + Self { + path: format!("{}-{}", name, registry().next_root_suffix()), + token: CancellationToken::new(), + } + } + + /// Create a child scope. Cancelling this scope cancels the child. + pub fn child(&self, name: &str) -> Self { + Self { + path: format!("{}/{}", self.path, name), + token: self.token.child_token(), + } + } + + pub fn path(&self) -> &str { + &self.path + } + + /// The scope's own cancellation token, for select!-ing in code that runs + /// inside an already-spawned context. + pub fn token(&self) -> &CancellationToken { + &self.token + } + + /// Request cancellation of every task in this scope's subtree. + pub fn cancel(&self) { + registry().cancel(&self.token); + } + + /// Cancel the subtree and wait until every task under this scope has + /// deregistered. Wrap in a timeout if the subtree may contain + /// uncooperative tasks. + pub async fn shutdown(&self) { + registry().cancel(&self.token); + registry().wait_empty_under(&self.path).await; + } + + /// Spawn a task owned by this scope. The closure receives the task's own + /// child token; long-running tasks MUST select on it. + #[cfg(not(feature = "wasm"))] + pub fn spawn(&self, name: &str, f: F) -> Result<(), TaskSpawnError> + where + F: FnOnce(CancellationToken) -> Fut, + Fut: Future + Send + 'static, + { + let (id, task_token, span) = self.register_task(name)?; + let guard = DeregisterGuard::new(id, Some(task_token.clone())); + let fut = f(task_token.clone()); + async_manager::spawn(finish_task(fut, guard), span); + Ok(()) + } + + /// Spawn a task owned by this scope (WASM, no Send required). + #[cfg(feature = "wasm")] + pub fn spawn(&self, name: &str, f: F) -> Result<(), TaskSpawnError> + where + F: FnOnce(CancellationToken) -> Fut, + Fut: Future + 'static, + { + let (id, task_token, span) = self.register_task(name)?; + let guard = DeregisterGuard::new(id, Some(task_token.clone())); + let fut = f(task_token.clone()); + async_manager::spawn(finish_task(fut, guard), span); + Ok(()) + } + + /// Consume the scope and spawn a task that holds it alive for its own + /// duration. Use when the caller has nowhere to store the scope (e.g. + /// protocol subscription handlers): drop-cancel must not fire before the + /// task runs, but parent cancellation still propagates. + #[cfg(not(feature = "wasm"))] + pub fn spawn_and_hold(self, name: &str, f: F) -> Result<(), TaskSpawnError> + where + F: FnOnce(CancellationToken) -> Fut, + Fut: Future + Send + 'static, + { + let (id, task_token, span) = self.register_task(name)?; + let guard = DeregisterGuard::new(id, Some(task_token.clone())); + let fut = f(task_token.clone()); + async_manager::spawn( + async move { + let _hold = self; + finish_task(fut, guard).await; + }, + span, + ); + Ok(()) + } + + /// Consume the scope and spawn a task that holds it alive (WASM, no Send). + #[cfg(feature = "wasm")] + pub fn spawn_and_hold(self, name: &str, f: F) -> Result<(), TaskSpawnError> + where + F: FnOnce(CancellationToken) -> Fut, + Fut: Future + 'static, + { + let (id, task_token, span) = self.register_task(name)?; + let guard = DeregisterGuard::new(id, Some(task_token.clone())); + let fut = f(task_token.clone()); + async_manager::spawn( + async move { + let _hold = self; + finish_task(fut, guard).await; + }, + span, + ); + Ok(()) + } + + fn register_task( + &self, + name: &str, + ) -> Result<(TaskId, CancellationToken, tracing::Span), TaskSpawnError> { + let path = format!("{}/{}", self.path, name); + let (id, task_token) = registry().register_scoped(path.clone(), &self.token)?; + // Span names must be const in tracing; the dynamic path goes in a field. + let span = tracing::span!(tracing::Level::INFO, "buttplug_task", task.path = %path); + Ok((id, task_token, span)) + } +} + +/// Deregisters a task from the global registry on drop. Created BEFORE the +/// task future is awaited so deregistration happens even if the future panics: +/// a panicking task unwinds through this guard, so the registry entry is removed +/// rather than leaked (which would hang `wait_empty_under` on that subtree +/// forever). The outcome is derived at drop time from whether we are unwinding +/// from a panic and whether the task observed cancellation. +/// +/// `token` is `None` for detached tasks, which have no cancellation concept: +/// their outcome is `Panicked` on panic, else `Completed`. +pub(super) struct DeregisterGuard { + id: TaskId, + token: Option, +} + +impl DeregisterGuard { + pub(super) fn new(id: TaskId, token: Option) -> Self { + Self { id, token } + } +} + +impl Drop for DeregisterGuard { + fn drop(&mut self) { + let outcome = if std::thread::panicking() { + TaskOutcome::Panicked + } else if self.token.as_ref().is_some_and(|t| t.is_cancelled()) { + TaskOutcome::Cancelled + } else { + TaskOutcome::Completed + }; + registry().deregister(self.id, outcome); + } +} + +async fn finish_task(fut: impl Future, _guard: DeregisterGuard) { + fut.await; +} + +impl Drop for TaskScope { + fn drop(&mut self) { + registry().cancel(&self.token); + } +} + +#[cfg(test)] +mod test { + use super::*; + use crate::util::task::registry; + use crate::util::task::registry::TaskEvent; + use std::time::Duration; + + #[test] + fn test_root_path_unique() { + let a = TaskScope::root("testroot"); + let b = TaskScope::root("testroot"); + assert_ne!(a.path(), b.path()); + assert!(a.path().starts_with("testroot-")); + } + + #[test] + fn test_child_path() { + let root = TaskScope::root("testroot"); + let child = root.child("devices"); + assert_eq!(child.path(), format!("{}/devices", root.path())); + } + + #[test] + fn test_spawn_rejected_after_cancel_does_not_invoke_closure() { + let root = TaskScope::root("rejecttest"); + let path = root.path().to_owned(); + root.cancel(); + let invoked = std::cell::Cell::new(false); + let result = root.spawn("worker", |_token| { + invoked.set(true); + async {} + }); + assert_eq!(result, Err(TaskSpawnError::ScopeClosed)); + assert!(!invoked.get()); + assert_eq!(registry().live_count_under(&path), 0); + } + + #[test] + fn test_rejected_spawn_emits_no_started_event() { + let root = TaskScope::root("rejecteventtest"); + let task_path = format!("{}/worker", root.path()); + let mut events = registry().event_stream(); + root.cancel(); + assert_eq!( + root.spawn("worker", |_token| async {}), + Err(TaskSpawnError::ScopeClosed) + ); + while let Ok(event) = events.try_recv() { + if let TaskEvent::Started { path, .. } = event { + assert_ne!(path, task_path, "rejected spawn emitted TaskStarted"); + } + } + } + + #[test] + fn test_retained_child_rejects_after_parent_cancel() { + let root = TaskScope::root("retainedchildtest"); + let child = root.child("child"); + root.cancel(); + assert_eq!( + child.spawn("worker", |_token| async {}), + Err(TaskSpawnError::ScopeClosed) + ); + } + + #[tokio::test] + async fn test_retained_arc_rejects_after_shutdown() { + let root = std::sync::Arc::new(TaskScope::root("retainedarctest")); + let path = root.path().to_owned(); + let task_root = root.clone(); + let (ready, wait) = tokio::sync::oneshot::channel(); + task_root + .spawn("worker", move |token| async move { + let _ = ready.send(()); + token.cancelled().await; + }) + .expect("worker registration should succeed"); + wait.await.expect("worker did not start"); + root.shutdown().await; + assert_eq!( + root.spawn("late", |_token| async {}), + Err(TaskSpawnError::ScopeClosed) + ); + assert_eq!(registry().live_count_under(&path), 0); + } + + #[tokio::test] + async fn test_sync_panic_deregisters_spawn() { + let root = TaskScope::root("syncpanictest"); + let path = root.path().to_owned(); + let mut events = registry().event_stream(); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _ = root.spawn("panicker", |_token| -> std::future::Ready<()> { + panic!("intentional construction panic"); + }); + })); + assert!(result.is_err()); + assert_eq!(registry().live_count_under(&path), 0); + let task_path = format!("{}/panicker", root.path()); + let mut saw_started = false; + let mut saw_panicked = false; + while let Ok(event) = events.try_recv() { + match event { + TaskEvent::Started { path, .. } if path == task_path => saw_started = true, + TaskEvent::Ended { path, outcome, .. } if path == task_path => { + saw_panicked = outcome == TaskOutcome::Panicked; + } + _ => {} + } + } + assert!(saw_started); + assert!(saw_panicked); + } + + #[test] + fn test_sync_panic_deregisters_spawn_and_hold() { + let root = TaskScope::root("syncpanicholdtest"); + let path = root.path().to_owned(); + let task_path = format!("{}/panicker", root.path()); + let mut events = registry().event_stream(); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _ = root.spawn_and_hold("panicker", |_token| -> std::future::Ready<()> { + panic!("intentional construction panic"); + }); + })); + assert!(result.is_err()); + assert_eq!(registry().live_count_under(&path), 0); + let mut saw_started = false; + let mut saw_panicked = false; + while let Ok(event) = events.try_recv() { + match event { + TaskEvent::Started { path, .. } if path == task_path => saw_started = true, + TaskEvent::Ended { path, outcome, .. } if path == task_path => { + saw_panicked = outcome == TaskOutcome::Panicked; + } + _ => {} + } + } + assert!(saw_started); + assert!(saw_panicked); + } + + #[tokio::test] + async fn test_spawn_registers_and_completes() { + let root = TaskScope::root("spawntest"); + let path = root.path().to_owned(); + let (tx, rx) = tokio::sync::oneshot::channel(); + let _ = root.spawn("worker", |_token| async move { + let _ = tx.send(()); + }); + rx.await.unwrap(); + tokio::time::timeout(Duration::from_secs(1), registry().wait_empty_under(&path)) + .await + .expect("task did not deregister after completion"); + } + + #[tokio::test] + async fn test_cancel_propagates_to_children() { + let root = TaskScope::root("canceltest"); + let path = root.path().to_owned(); + let child = root.child("inner"); + let _ = child.spawn("worker", |token| async move { + token.cancelled().await; + }); + root.cancel(); + tokio::time::timeout(Duration::from_secs(1), registry().wait_empty_under(&path)) + .await + .expect("cancel did not propagate to child scope task"); + } + + #[tokio::test] + async fn test_drop_cancels() { + let root = TaskScope::root("droptest"); + let path = root.path().to_owned(); + let _ = root.spawn("worker", |token| async move { + token.cancelled().await; + }); + drop(root); + tokio::time::timeout(Duration::from_secs(1), registry().wait_empty_under(&path)) + .await + .expect("drop did not cancel task"); + } + + #[tokio::test] + async fn test_shutdown_awaits_subtree() { + let root = TaskScope::root("shutdowntest"); + let _ = root.spawn("worker", |token| async move { + token.cancelled().await; + // Simulate cleanup work after observing cancellation. + tokio::time::sleep(Duration::from_millis(20)).await; + }); + tokio::time::timeout(Duration::from_secs(1), root.shutdown()) + .await + .expect("shutdown did not resolve"); + } + + #[tokio::test] + async fn test_panicking_task_deregisters() { + // A scoped task that panics must still deregister (via the drop guard), + // otherwise wait_empty_under on its root would hang forever. tokio catches + // the panic at the task boundary, so this test itself does not fail from the + // spawned panic. Without the guard, this wait would time out. + let root = TaskScope::root("panictest"); + let path = root.path().to_owned(); + let _ = root.spawn("panicker", |_token| async move { + panic!("intentional panic for deregistration test"); + }); + tokio::time::timeout(Duration::from_secs(1), registry().wait_empty_under(&path)) + .await + .expect("panicking task did not deregister — registry entry leaked"); + } + + #[tokio::test] + async fn test_registration_winning_before_cancel_is_awaited_by_shutdown() { + let root = std::sync::Arc::new(TaskScope::root("gatewintest")); + let path = root.path().to_owned(); + let gate = registry().test_hold_gate(); + let spawn_root = root.clone(); + let handle = tokio::runtime::Handle::current(); + let spawn_thread = std::thread::spawn(move || { + let _enter = handle.enter(); + spawn_root + .spawn("worker", |token| async move { + token.cancelled().await; + }) + .expect("registration should succeed after gate release"); + }); + std::thread::yield_now(); + drop(gate); + spawn_thread.join().expect("spawn thread panicked"); + tokio::time::timeout(Duration::from_secs(1), root.shutdown()) + .await + .expect("shutdown did not await the successful registration"); + assert_eq!(registry().live_count_under(&path), 0); + } + + #[tokio::test] + async fn test_concurrent_spawn_shutdown_gate_has_only_valid_outcomes() { + use registry::TaskEvent; + + let root = std::sync::Arc::new(TaskScope::root("gateracetest")); + let path = root.path().to_owned(); + let task_path = format!("{path}/worker"); + let mut events = registry().event_stream(); + let gate = registry().test_hold_gate(); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(3)); + let spawn_root = root.clone(); + let spawn_barrier = barrier.clone(); + let spawn_task = tokio::task::spawn_blocking(move || { + spawn_barrier.wait(); + spawn_root.spawn("worker", |token| async move { + token.cancelled().await; + }) + }); + let cancel_root = root.clone(); + let cancel_barrier = barrier.clone(); + let cancel_task = tokio::task::spawn_blocking(move || { + cancel_barrier.wait(); + cancel_root.cancel(); + }); + barrier.wait(); + drop(gate); + let spawn_result = spawn_task.await.expect("spawn task panicked"); + cancel_task.await.expect("cancel task panicked"); + tokio::time::timeout(Duration::from_secs(1), root.shutdown()) + .await + .expect("shutdown did not drain the subtree"); + assert_eq!(registry().live_count_under(&path), 0); + + match spawn_result { + Ok(()) => { + let mut saw_started = false; + let mut saw_ended = false; + tokio::time::timeout(Duration::from_secs(1), async { + while !(saw_started && saw_ended) { + match events.recv().await { + Ok(TaskEvent::Started { path, .. }) if path == task_path => saw_started = true, + Ok(TaskEvent::Ended { path, .. }) if path == task_path => saw_ended = true, + Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + }) + .await + .expect("successful spawn lifecycle events did not arrive"); + assert!( + saw_started, + "successful spawn emitted no matching TaskStarted" + ); + assert!(saw_ended, "successful spawn emitted no matching TaskEnded"); + } + Err(TaskSpawnError::ScopeClosed) => { + let drain = tokio::time::timeout(Duration::from_millis(100), async { + loop { + match events.recv().await { + Ok(TaskEvent::Started { path, .. }) | Ok(TaskEvent::Ended { path, .. }) + if path == task_path => + { + panic!("rejected spawn emitted a matching lifecycle event"); + } + Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + }) + .await; + assert!( + drain.is_err(), + "event drain ended without its bounded timeout" + ); + } + } + } + + #[tokio::test] + async fn test_spawn_and_hold_keeps_scope_alive() { + use registry::TaskEvent; + + let root = TaskScope::root("holdtest"); + let path = root.path().to_owned(); + // Subscribe BEFORE spawning so we observe both the Started and Ended events + // for the held task and can assert how it actually finished. + let mut events = registry().event_stream(); + let sub_scope = root.child("subscription"); + let (tx, rx) = tokio::sync::oneshot::channel(); + // Consuming spawn: the scope moves INTO the task and must not cancel it. + let _ = sub_scope.spawn_and_hold("worker", |_token| async move { + tokio::time::sleep(Duration::from_millis(20)).await; + let _ = tx.send(()); + }); + // If drop-cancel fired early this would hang (task cancelled before send). + tokio::time::timeout(Duration::from_secs(1), rx) + .await + .expect("spawn_and_hold task was cancelled early") + .unwrap(); + tokio::time::timeout(Duration::from_secs(1), registry().wait_empty_under(&path)) + .await + .expect("task did not deregister"); + + // The held task ran to its natural end, so its reported outcome MUST be + // Completed — not Cancelled. (If spawn_and_hold had wired drop-cancel to the + // held task, it would have been cancelled mid-sleep and reported Cancelled.) + // Drain the event stream looking for this task's Ended event. The path is + // exact ("/subscription/worker") so we don't match unrelated tasks + // from other tests sharing the global registry. + let task_path = format!("{path}/subscription/worker"); + let outcome = tokio::time::timeout(Duration::from_secs(1), async { + loop { + match events.recv().await { + Ok(TaskEvent::Ended { path, outcome, .. }) if path == task_path => return outcome, + Ok(_) => continue, + // The registry's broadcast channel is process-global; heavy parallel + // test load can evict buffered events (Lagged). Keep draining — our + // own Ended event fires ~20ms after subscribing, so under normal load + // it arrives well before any eviction window matters. + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + panic!("event stream closed before held task's Ended event") + } + } + } + }) + .await + .expect("did not observe Ended event for held task"); + assert_eq!( + outcome, + TaskOutcome::Completed, + "normally-finishing held task should report Completed, got {outcome:?}" + ); + } +} diff --git a/crates/buttplug_server/CHANGELOG.md b/crates/buttplug_server/CHANGELOG.md index f4dd07672..381a0a814 100644 --- a/crates/buttplug_server/CHANGELOG.md +++ b/crates/buttplug_server/CHANGELOG.md @@ -1,3 +1,10 @@ +## Unreleased + +## Compatibility Notice + +- `ProtocolHandler::handle_input_subscribe_cmd` now receives a `TaskScope` argument. This is technically a public Rust API signature change, so custom external implementations must update their method signature. +- `ProtocolHandler` is currently intended for protocol implementations shipped inside this repository; it is not treated as a stable external extension interface. In line with ADR 0008, servers are not intended to be independently reimplemented. This policy does not make the public trait change non-breaking for existing external implementations. + # 10.0.4 (2026-06-01) ## Features diff --git a/crates/buttplug_server/src/device/device_handle.rs b/crates/buttplug_server/src/device/device_handle.rs index f666b2a18..50a9f86ce 100644 --- a/crates/buttplug_server/src/device/device_handle.rs +++ b/crates/buttplug_server/src/device/device_handle.rs @@ -26,7 +26,7 @@ use buttplug_core::{ OutputValue, StopCmdV4, }, - util::stream::convert_broadcast_receiver_to_stream, + util::{async_manager, stream::convert_broadcast_receiver_to_stream, task::TaskScope}, }; use buttplug_server_device_config::{ DeviceConfigurationManager, @@ -36,10 +36,13 @@ use buttplug_server_device_config::{ }; use dashmap::DashMap; use futures::future::{self, BoxFuture, FutureExt}; -use tokio::sync::{ - broadcast, - mpsc::{Sender, channel}, - oneshot, +use tokio::{ + select, + sync::{ + broadcast, + mpsc::{Sender, channel}, + oneshot, + }, }; use tokio_stream::StreamExt; use uuid::Uuid; @@ -58,7 +61,7 @@ use crate::{ use super::{ InternalDeviceEvent, OutputObservation, - device_task::{DeviceTaskConfig, spawn_device_task}, + device_task::{DeviceTaskConfig, DeviceTaskMessage, spawn_device_task}, hardware::{Hardware, HardwareCommand, HardwareConnector, HardwareEvent}, protocol::{ProtocolHandler, ProtocolKeepaliveStrategy, ProtocolSpecializer}, }; @@ -109,20 +112,27 @@ pub struct DeviceHandle { legacy_attributes: ServerDeviceAttributes, last_output_command: Arc>, stop_commands: Arc>, - internal_hw_msg_sender: Sender>, + internal_hw_msg_sender: Sender, output_observation_sender: Option>, + /// Scope owning this device's tasks. Rides in an Arc since DeviceHandle is + /// Clone; the subtree cancels when the last clone drops. + task_scope: Arc, } impl DeviceHandle { - /// Create a new DeviceHandle with direct ownership of device state + /// Create a new DeviceHandle with direct ownership of device state. + /// The arguments bundle all device-state concerns (hardware, protocol, definition, identity, + /// commands, I/O channel, observability, lifecycle scope) — suppressing the lint is correct here. + #[allow(clippy::too_many_arguments)] pub(crate) fn new( hardware: Arc, handler: Arc, definition: ServerDeviceDefinition, identifier: UserDeviceIdentifier, stop_commands: Vec, - internal_hw_msg_sender: Sender>, + internal_hw_msg_sender: Sender, output_observation_sender: Option>, + task_scope: Arc, ) -> Self { Self { hardware, @@ -134,6 +144,7 @@ impl DeviceHandle { stop_commands: Arc::new(stop_commands), internal_hw_msg_sender, output_observation_sender, + task_scope, } } @@ -264,11 +275,32 @@ impl DeviceHandle { // --- Private command handling methods --- fn handle_outputcmd_v4(&self, msg: &CheckedOutputCmdV4) -> ButtplugServerResultFuture { + match self.output_cmd_hardware_commands(msg) { + // Deduped (no-op) or empty: nothing to send, succeed immediately. + None => future::ready(Ok(message::OkV0::default().into())).boxed(), + Some(Err(err)) => future::ready(Err(err)).boxed(), + Some(Ok(commands)) => self.handle_hardware_commands(commands, false), + } + } + + /// Run the bookkeeping for an output command (dedupe map, output-observation + /// emit) and ask the protocol handler for the hardware commands it produces. + /// + /// Returns `None` when the command is a deduped no-op (nothing to send), + /// `Some(Err(..))` when the handler errored, and `Some(Ok(cmds))` otherwise. + /// Shared by `handle_outputcmd_v4` (normal output) and the stop path, so both + /// keep identical dedupe-map and observation behaviour. The stop path + /// accumulates the returned commands across all stop messages and flushes them + /// in a single write-acknowledged batch. + fn output_cmd_hardware_commands( + &self, + msg: &CheckedOutputCmdV4, + ) -> Option, ButtplugError>> { if let Some(last_msg) = self.last_output_command.get(&msg.feature_id()) && *last_msg == *msg { trace!("No commands generated for incoming device packet, skipping and returning success."); - return future::ready(Ok(message::OkV0::default().into())).boxed(); + return None; } self .last_output_command @@ -276,7 +308,6 @@ impl DeviceHandle { if let Some(sender) = &self.output_observation_sender { // OutputType derives Display via strum, producing clean names like "Vibrate", "Rotate". - // The design uses format!("{:?}") but to_string() is preferred for clean output. let _ = sender.send(OutputObservation { device_index: self.definition.index(), feature_index: msg.feature_index(), @@ -285,37 +316,94 @@ impl DeviceHandle { }); } - self.handle_generic_command_result(self.handler.handle_output_cmd(msg)) + Some(self.handler.handle_output_cmd(msg).map_err(|e| e.into())) } - fn handle_hardware_commands(&self, commands: Vec) -> ButtplugServerResultFuture { + /// Queue hardware commands for the device io task. + /// + /// When `wait_for_write` is false (all normal output paths) this is + /// fire-and-forget: the commands are dropped into the io channel and the + /// future resolves immediately, leaving any batching to the io task. + /// + /// When `wait_for_write` is true (the stop path) the commands carry a + /// write-acknowledgement channel: the io task flushes the batch to hardware + /// immediately and fires the ack, and this future only resolves once that + /// write has gone out (or the bounded wait elapses / the device goes away). + fn handle_hardware_commands( + &self, + commands: Vec, + wait_for_write: bool, + ) -> ButtplugServerResultFuture { let sender = self.internal_hw_msg_sender.clone(); async move { - let _ = sender.send(commands).await; + if wait_for_write { + let (ack_sender, ack_receiver) = oneshot::channel(); + if sender + .send(DeviceTaskMessage { + commands, + ack: Some(ack_sender), + }) + .await + .is_err() + { + debug!("Device io task gone, skipping write acknowledgement wait."); + return Ok(message::OkV0::default().into()); + } + // Bounded wait for the write ack. There is no tokio::time on WASM, so + // race the receiver against a runtime-agnostic sleep instead of + // tokio::time::timeout. A wedged or dead device must not hang shutdown + // or error a stop, so any non-success outcome resolves Ok (matching the + // existing swallow semantics; scope cancellation handles stragglers). + select! { + ack = ack_receiver => { + if ack.is_err() { + debug!("Device io task dropped ack; device likely disconnecting."); + } + } + _ = async_manager::sleep(Duration::from_secs(1)) => { + warn!("Timed out waiting for stop command write acknowledgement."); + } + } + } else { + let _ = sender + .send(DeviceTaskMessage { + commands, + ack: None, + }) + .await; + } Ok(message::OkV0::default().into()) } .boxed() } - fn handle_generic_command_result( - &self, - command_result: Result, ButtplugDeviceError>, - ) -> ButtplugServerResultFuture { - let hardware_commands = match command_result { - Ok(commands) => commands, - Err(err) => return future::ready(Err(err.into())).boxed(), - }; - - self.handle_hardware_commands(hardware_commands) - } - fn handle_stop_device_cmd(&self, msg: &StopCmdV4) -> ButtplugServerResultFuture { let mut fut_vec = vec![]; if msg.outputs() { - self - .stop_commands - .iter() - .for_each(|msg| fut_vec.push(self.parse_message(msg.clone()))); + // Accumulate the hardware commands from every stop output into a single + // write-acknowledged batch. A multi-feature device produces one stop + // OutputCmd per feature, but the protocol handler folds them into one + // accumulated write; sending each individually with its own ack would + // emit one write per feature instead. Collecting them and issuing a + // single acked flush keeps the on-wire output identical to normal output + // batching while still resolving only once the stop write has landed. + // + // Stop commands are always OutputCmds (see build_device_handle); any + // other shape falls back to the normal fire-and-forget parse path. + let mut stop_hardware_commands = vec![]; + for stop_msg in self.stop_commands.iter() { + match stop_msg { + ButtplugDeviceCommandMessageUnionV4::OutputCmd(output_msg) => { + match self.output_cmd_hardware_commands(output_msg) { + None => {} + Some(Ok(commands)) => stop_hardware_commands.extend(commands), + Some(Err(err)) => fut_vec.push(future::ready(Err(err)).boxed()), + } + } + other => fut_vec.push(self.parse_message(other.clone())), + } + } + fut_vec.push(self.handle_hardware_commands(stop_hardware_commands, true)); } if msg.inputs() { self.definition.features().iter().for_each(|(i, f)| { @@ -402,9 +490,17 @@ impl DeviceHandle { info!("Handling input subscribe command"); let device = self.hardware.clone(); let handler = self.handler.clone(); + let task_scope = self.task_scope.child("subscription"); async move { handler - .handle_input_subscribe_cmd(device_index, device, feature_index, feature_id, input_type) + .handle_input_subscribe_cmd( + device_index, + device, + feature_index, + feature_id, + input_type, + task_scope, + ) .await .map(|_| message::OkV0::new(1).into()) .map_err(|e| e.into()) @@ -455,7 +551,12 @@ pub(super) async fn build_device_handle( protocol_specializers: Vec, device_event_sender: tokio::sync::mpsc::Sender, output_observation_sender: Option>, + task_scope: TaskScope, ) -> Result { + // DeviceHandle is Clone, so its scope rides in an Arc; the device subtree + // cancels when the last clone (and thus the last Arc) drops. + let task_scope = Arc::new(task_scope); + // At this point, we know we've got hardware that is waiting to connect, and enough protocol // info to actually do something after we connect. So go ahead and connect. trace!("Connecting to {:?}", hardware_connector); @@ -515,7 +616,7 @@ pub(super) async fn build_device_handle( let strategy = handler.keepalive_strategy(); // Create the hardware command channel and spawn the device task - let (internal_hw_msg_sender, internal_hw_msg_recv) = channel::>(1024); + let (internal_hw_msg_sender, internal_hw_msg_recv) = channel::(1024); let device_wait_duration = if let Some(gap) = definition.message_gap_ms() { Some(Duration::from_millis(gap as u64)) @@ -524,6 +625,7 @@ pub(super) async fn build_device_handle( }; spawn_device_task( + &task_scope, hardware.clone(), handler.clone(), DeviceTaskConfig { @@ -532,7 +634,7 @@ pub(super) async fn build_device_handle( keepalive_strategy: handler.keepalive_strategy(), }, internal_hw_msg_recv, - ); + )?; // Generate stop commands for this device let mut stop_commands: Vec = vec![]; @@ -590,6 +692,7 @@ pub(super) async fn build_device_handle( stop_commands, internal_hw_msg_sender, output_observation_sender, + task_scope.clone(), ); // If we need a keepalive with a packet replay, set this up via stopping the device on connect. @@ -614,44 +717,57 @@ pub(super) async fn build_device_handle( // to the device manager event loop via the provided sender. let event_stream = device_handle.event_stream(); let identifier = device_handle.identifier().clone(); - buttplug_core::spawn!("DeviceEventForwarding", async move { - futures::pin_mut!(event_stream); - loop { - let event = futures::StreamExt::next(&mut event_stream).await; - match event { - Some(DeviceEvent::Disconnected(id)) => { - if device_event_sender - .send(InternalDeviceEvent::Disconnected(id)) - .await - .is_err() - { - info!( - "Device event sender closed for device {:?}, stopping event forwarding.", - identifier - ); + task_scope + .spawn("event-forwarding", move |token| async move { + futures::pin_mut!(event_stream); + loop { + tokio::select! { + _ = token.cancelled() => { + info!("Event forwarding cancelled for device {:?}", identifier); break; } - } - Some(DeviceEvent::Notification(_, msg)) => { - if device_event_sender - .send(InternalDeviceEvent::Notification(msg)) - .await - .is_err() - { - info!( - "Device event sender closed for device {:?}, stopping event forwarding.", - identifier - ); - break; + event = futures::StreamExt::next(&mut event_stream) => { + match event { + Some(DeviceEvent::Disconnected(id)) => { + if device_event_sender + .send(InternalDeviceEvent::Disconnected(id)) + .await + .is_err() + { + info!( + "Device event sender closed for device {:?}, stopping event forwarding.", + identifier + ); + break; + } + } + Some(DeviceEvent::Notification(_, msg)) => { + if device_event_sender + .send(InternalDeviceEvent::Notification(msg)) + .await + .is_err() + { + info!( + "Device event sender closed for device {:?}, stopping event forwarding.", + identifier + ); + break; + } + } + None => { + // Stream ended (device likely disconnected) + break; + } + } } } - None => { - // Stream ended (device likely disconnected) - break; - } } - } - }); + }) + .map_err(|e| { + ButtplugDeviceError::DeviceConnectionError(format!( + "Could not spawn device event forwarding task: {e}" + )) + })?; Ok(device_handle) } diff --git a/crates/buttplug_server/src/device/device_task.rs b/crates/buttplug_server/src/device/device_task.rs index a54f1fa43..014bf351a 100644 --- a/crates/buttplug_server/src/device/device_task.rs +++ b/crates/buttplug_server/src/device/device_task.rs @@ -14,9 +14,13 @@ use std::{collections::VecDeque, sync::Arc, time::Duration}; -use buttplug_core::util::async_manager; +use buttplug_core::{ + errors::ButtplugDeviceError, + util::{async_manager, task::TaskScope}, +}; use futures::future; use tokio::{select, sync::mpsc::Receiver, time::Instant}; +use tokio_util::sync::CancellationToken; use super::{ hardware::{Hardware, HardwareCommand, HardwareEvent, HardwareWriteCmd}, @@ -33,6 +37,15 @@ pub struct DeviceTaskConfig { pub keepalive_strategy: ProtocolKeepaliveStrategy, } +/// A batch of hardware commands for the device io task, optionally carrying a +/// write acknowledgement channel. When `ack` is present the batch is urgent: +/// the io task flushes it (and any pending batched commands) to hardware +/// immediately, then fires the ack. +pub struct DeviceTaskMessage { + pub commands: Vec, + pub ack: Option>, +} + /// Spawn the device communication task. /// /// This task handles: @@ -43,14 +56,19 @@ pub struct DeviceTaskConfig { /// /// Returns immediately after spawning the task. pub fn spawn_device_task( + task_scope: &TaskScope, hardware: Arc, _handler: Arc, config: DeviceTaskConfig, - mut command_receiver: Receiver>, -) { - buttplug_core::spawn!("DeviceTask", async move { - run_device_task(hardware, config, &mut command_receiver).await; - }); + mut command_receiver: Receiver, +) -> Result<(), ButtplugDeviceError> { + task_scope + .spawn("io", move |token| async move { + run_device_task(hardware, config, &mut command_receiver, token).await; + }) + .map_err(|e| { + ButtplugDeviceError::DeviceConnectionError(format!("Could not spawn device I/O task: {e}")) + }) } /// Run the device communication task (internal implementation). @@ -60,7 +78,8 @@ pub fn spawn_device_task( async fn run_device_task( hardware: Arc, config: DeviceTaskConfig, - command_receiver: &mut Receiver>, + command_receiver: &mut Receiver, + token: CancellationToken, ) { let mut hardware_events = hardware.event_stream(); let device_wait_duration = config.message_gap; @@ -90,6 +109,23 @@ async fn run_device_task( let mut pending_commands: VecDeque = VecDeque::new(); let mut batch_deadline: Option = None; + // Write every pending command to hardware in order, updating the keepalive + // packet as writes go out. Shared by the receive arm (urgent acked flush), + // the batch-deadline arm, and the flush-on-exit hardening below. + async fn flush_pending( + hardware: &Hardware, + pending_commands: &mut VecDeque, + track_keepalive: bool, + keepalive_packet: &mut Option, + ) { + while let Some(cmd) = pending_commands.pop_front() { + let _ = hardware.parse_message(&cmd).await; + if track_keepalive && let HardwareCommand::Write(ref write_cmd) = cmd { + *keepalive_packet = Some(write_cmd.clone()); + } + } + } + loop { // Calculate keepalive timeout let keepalive_fut = async { @@ -115,14 +151,57 @@ async fn run_device_task( select! { biased; + // Priority 0: Cooperative cancellation - wins over new work. + _ = token.cancelled() => { + info!("Device task cancelled, shutting down"); + // Best-effort flush of any batched writes (e.g. a stop command still + // sitting in the batch window) before exiting. The hardware is still + // present here, so the writes can land; write errors are swallowed. + flush_pending( + &hardware, + &mut pending_commands, + track_keepalive, + &mut keepalive_packet, + ) + .await; + return; + } + // Priority 1: Incoming commands msg = command_receiver.recv() => { - let Some(commands) = msg else { + let Some(DeviceTaskMessage { commands, ack }) = msg else { info!("No longer receiving messages from device parent, breaking"); + // The command channel closed (all DeviceHandles dropped). Hardware is + // still present, so best-effort flush any batched writes before exit. + flush_pending( + &hardware, + &mut pending_commands, + track_keepalive, + &mut keepalive_packet, + ) + .await; break; }; - if let Some(device_wait_duration) = device_wait_duration { + if let Some(ack) = ack { + // Urgent write-acknowledged batch (stop path). Merge with any pending + // batch using the existing dedupe, flush everything to hardware now + // regardless of the batch deadline, then fire the ack so the caller + // only resolves once the writes have actually gone out. + for command in commands { + pending_commands.retain(|existing| !command.overlaps(existing)); + pending_commands.push_back(command); + } + flush_pending( + &hardware, + &mut pending_commands, + track_keepalive, + &mut keepalive_packet, + ) + .await; + batch_deadline = None; + let _ = ack.send(()); + } else if let Some(device_wait_duration) = device_wait_duration { // Batching enabled if pending_commands.is_empty() { // First batch - add directly without deduplication (matches old behavior) @@ -138,28 +217,27 @@ async fn run_device_task( } else { // No batching - send immediately trace!("No wait duration, sending commands immediately: {:?}", commands); - for cmd in commands { - let _ = hardware.parse_message(&cmd).await; - if track_keepalive - && let HardwareCommand::Write(ref write_cmd) = cmd - { - keepalive_packet = Some(write_cmd.clone()); - } - } + pending_commands.extend(commands); + flush_pending( + &hardware, + &mut pending_commands, + track_keepalive, + &mut keepalive_packet, + ) + .await; } } // Priority 2: Batch deadline reached - flush pending commands _ = batch_fut => { trace!("Batch deadline reached, sending {} commands", pending_commands.len()); - while let Some(cmd) = pending_commands.pop_front() { - let _ = hardware.parse_message(&cmd).await; - if track_keepalive - && let HardwareCommand::Write(ref write_cmd) = cmd - { - keepalive_packet = Some(write_cmd.clone()); - } - } + flush_pending( + &hardware, + &mut pending_commands, + track_keepalive, + &mut keepalive_packet, + ) + .await; batch_deadline = None; } @@ -199,6 +277,8 @@ async fn run_device_task( hw_event = hardware_events.recv() => { if matches!(hw_event, Ok(HardwareEvent::Disconnected(_))) || hw_event.is_err() { info!("Hardware disconnected, shutting down task"); + // Do NOT flush pending_commands here: the hardware is gone, so writes + // would fail and any pending stop is moot. return; } } diff --git a/crates/buttplug_server/src/device/protocol.rs b/crates/buttplug_server/src/device/protocol.rs index e624b5706..83cc94b8b 100644 --- a/crates/buttplug_server/src/device/protocol.rs +++ b/crates/buttplug_server/src/device/protocol.rs @@ -10,6 +10,7 @@ use buttplug_core::{ errors::ButtplugDeviceError, message::{InputReadingV4, InputType, InputValue, OutputCommand}, + util::task::TaskScope, }; use buttplug_server_device_config::{ Endpoint, @@ -372,6 +373,7 @@ pub trait ProtocolHandler: Sync + Send { _feature_index: u32, _feature_id: Uuid, _sensor_type: InputType, + _task_scope: TaskScope, ) -> BoxFuture<'_, Result<(), ButtplugDeviceError>> { future::ready(Err(ButtplugDeviceError::UnhandledCommand( "Command not implemented for this protocol: InputCmd (Subscribe)".to_string(), diff --git a/crates/buttplug_server/src/device/protocol_impl/galaku.rs b/crates/buttplug_server/src/device/protocol_impl/galaku.rs index f951ba2fb..57fed8654 100644 --- a/crates/buttplug_server/src/device/protocol_impl/galaku.rs +++ b/crates/buttplug_server/src/device/protocol_impl/galaku.rs @@ -14,6 +14,7 @@ use futures_util::{FutureExt, future}; use buttplug_core::errors::ButtplugDeviceError; use buttplug_core::message::{InputReadingV4, InputType, InputTypeReading, InputValue}; +use buttplug_core::util::task::TaskScope; use buttplug_server_device_config::Endpoint; use buttplug_server_device_config::{ @@ -245,6 +246,7 @@ impl ProtocolHandler for Galaku { _feature_index: u32, feature_id: Uuid, sensor_type: InputType, + _task_scope: TaskScope, ) -> BoxFuture<'_, Result<(), ButtplugDeviceError>> { match sensor_type { InputType::Battery => { diff --git a/crates/buttplug_server/src/device/protocol_impl/kgoal_boost.rs b/crates/buttplug_server/src/device/protocol_impl/kgoal_boost.rs index 7e99fef18..b2fbb1a4f 100644 --- a/crates/buttplug_server/src/device/protocol_impl/kgoal_boost.rs +++ b/crates/buttplug_server/src/device/protocol_impl/kgoal_boost.rs @@ -15,7 +15,7 @@ use crate::{ use buttplug_core::{ errors::ButtplugDeviceError, message::{InputReadingV4, InputType, InputValue}, - util::stream::convert_broadcast_receiver_to_stream, + util::{stream::convert_broadcast_receiver_to_stream, task::TaskScope}, }; use buttplug_server_device_config::Endpoint; use futures::{ @@ -65,6 +65,7 @@ impl ProtocolHandler for KGoalBoost { feature_index: u32, feature_id: Uuid, _sensor_type: InputType, + task_scope: TaskScope, ) -> BoxFuture<'_, Result<(), ButtplugDeviceError>> { let sensors = self.subscribed_sensors.load(Ordering::Relaxed); if (sensors & (1 << feature_index as u8)) > 0 { @@ -90,69 +91,89 @@ impl ProtocolHandler for KGoalBoost { let stream_sensors = stream_sensors.clone(); info!("Starting Kgoal subscription"); // If we subscribe successfully, we need to set up our event handler. - buttplug_core::spawn!("Kgoal subscription event handler", async move { - let mut cached_values = vec![0u32, 0u32]; - while let Ok(info) = hardware_stream.recv().await { - let subscribed_sensors = stream_sensors.load(Ordering::Relaxed); - // If we have no receivers, quit. - if sender.receiver_count() == 0 || subscribed_sensors == 0 { - return; - } - if let HardwareEvent::Notification(_, endpoint, data) = info - && endpoint == Endpoint::RxPressure - { - if data.len() < 7 { - // Not even sure how this would happen, error and continue on. - error!("KGoal Boost data not expected length!"); - continue; - } - // Extract our two pressure values. - let normalized = (data[3] as u32) << 8 | data[4] as u32; - let unnormalized = (data[5] as u32) << 8 | data[6] as u32; - info!( - "Kgoal Reading {} {} {}", - subscribed_sensors, normalized, unnormalized - ); - if (subscribed_sensors & (1 << 0)) > 0 - && cached_values[0] != normalized - && sender - .send( - InputReadingV4::new( - device_index, - 0, - buttplug_core::message::InputTypeReading::Pressure(InputValue::new( - normalized, - )), - ) - .into(), - ) - .is_err() - { - debug!("Hardware device listener for KGoal Boost shut down, returning from task."); + task_scope + .spawn_and_hold("kgoal-events", move |token| async move { + let mut cached_values = vec![0u32, 0u32]; + loop { + let info = tokio::select! { + biased; + _ = token.cancelled() => return, + info = hardware_stream.recv() => { + let Ok(info) = info else { + return; + }; + info + } + }; + let subscribed_sensors = stream_sensors.load(Ordering::Relaxed); + // If we have no receivers, quit. + if sender.receiver_count() == 0 || subscribed_sensors == 0 { return; } - if (subscribed_sensors & (1 << 1)) > 0 - && cached_values[1] != unnormalized - && sender - .send( - InputReadingV4::new( - device_index, - 1, - buttplug_core::message::InputTypeReading::Pressure(InputValue::new( - unnormalized, - )), - ) - .into(), - ) - .is_err() + if let HardwareEvent::Notification(_, endpoint, data) = info + && endpoint == Endpoint::RxPressure { - debug!("Hardware device listener for KGoal Boost shut down, returning from task."); - return; + if data.len() < 7 { + // Not even sure how this would happen, error and continue on. + error!("KGoal Boost data not expected length!"); + continue; + } + // Extract our two pressure values. + let normalized = (data[3] as u32) << 8 | data[4] as u32; + let unnormalized = (data[5] as u32) << 8 | data[6] as u32; + info!( + "Kgoal Reading {} {} {}", + subscribed_sensors, normalized, unnormalized + ); + if (subscribed_sensors & (1 << 0)) > 0 + && cached_values[0] != normalized + && sender + .send( + InputReadingV4::new( + device_index, + 0, + buttplug_core::message::InputTypeReading::Pressure(InputValue::new( + normalized, + )), + ) + .into(), + ) + .is_err() + { + debug!( + "Hardware device listener for KGoal Boost shut down, returning from task." + ); + return; + } + if (subscribed_sensors & (1 << 1)) > 0 + && cached_values[1] != unnormalized + && sender + .send( + InputReadingV4::new( + device_index, + 1, + buttplug_core::message::InputTypeReading::Pressure(InputValue::new( + unnormalized, + )), + ) + .into(), + ) + .is_err() + { + debug!( + "Hardware device listener for KGoal Boost shut down, returning from task." + ); + return; + } + cached_values = vec![normalized, unnormalized]; } - cached_values = vec![normalized, unnormalized]; } - } - }); + }) + .map_err(|e| { + ButtplugDeviceError::DeviceConnectionError(format!( + "Could not start Kgoal event forwarding: {e}" + )) + })?; } stream_sensors.store( stream_sensors.load(Ordering::Relaxed) | (1 << feature_index), diff --git a/crates/buttplug_server/src/device/server_device_manager.rs b/crates/buttplug_server/src/device/server_device_manager.rs index 77ce10877..437f1cffd 100644 --- a/crates/buttplug_server/src/device/server_device_manager.rs +++ b/crates/buttplug_server/src/device/server_device_manager.rs @@ -36,7 +36,7 @@ use buttplug_core::{ DeviceListV4, StopCmdV4, }, - util::stream::convert_broadcast_receiver_to_stream, + util::{stream::convert_broadcast_receiver_to_stream, task::TaskScope}, }; use buttplug_server_device_config::{DeviceConfigurationManager, UserDeviceIdentifier}; use dashmap::DashMap; @@ -54,7 +54,6 @@ use std::{ }, }; use tokio::sync::{broadcast, mpsc}; -use tokio_util::sync::CancellationToken; #[derive(Debug)] pub(super) enum DeviceManagerCommand { @@ -175,7 +174,8 @@ impl ServerDeviceManagerBuilder { } let devices = Arc::new(DashMap::new()); - let loop_cancellation_token = CancellationToken::new(); + let task_scope = TaskScope::root("device-manager"); + let devices_scope = task_scope.child("devices"); let output_sender = broadcast::channel(255).0; let output_observation_sender = if self.emit_output_observations { @@ -184,24 +184,39 @@ impl ServerDeviceManagerBuilder { None }; - let mut event_loop = ServerDeviceManagerEventLoop::new( - comm_managers, - self.device_configuration_manager.clone(), - devices.clone(), - loop_cancellation_token.child_token(), - output_sender.clone(), - device_event_receiver, - device_command_receiver, - output_observation_sender.clone(), - ); - buttplug_core::spawn!("ServerDeviceManager event loop", async move { - event_loop.run().await; - }); + // Clone everything the event loop needs, since the originals are still + // required to construct the ServerDeviceManager below. + let device_configuration_manager = self.device_configuration_manager.clone(); + let devices_clone = devices.clone(); + let output_sender_clone = output_sender.clone(); + let output_observation_sender_clone = output_observation_sender.clone(); + task_scope + .spawn("event-loop", move |token| async move { + let mut event_loop = ServerDeviceManagerEventLoop::new( + comm_managers, + device_configuration_manager, + devices_clone, + token, + devices_scope, + output_sender_clone, + device_event_receiver, + device_command_receiver, + output_observation_sender_clone, + ); + event_loop.run().await; + }) + .map_err(|e| { + ButtplugServerError::DeviceConfigurationManagerError( + ButtplugDeviceError::DeviceConnectionError(format!( + "Could not spawn device manager event loop: {e}" + )), + ) + })?; Ok(ServerDeviceManager { device_configuration_manager: self.device_configuration_manager.clone(), devices, device_command_sender, - loop_cancellation_token, + task_scope, running: Arc::new(AtomicBool::new(true)), output_sender, output_observation_sender, @@ -216,7 +231,7 @@ pub struct ServerDeviceManager { #[getset(get = "pub(crate)")] devices: Arc>, device_command_sender: mpsc::Sender, - loop_cancellation_token: CancellationToken, + task_scope: TaskScope, running: Arc, output_sender: broadcast::Sender, output_observation_sender: Option>, @@ -236,6 +251,15 @@ impl ServerDeviceManager { .map(|sender| convert_broadcast_receiver_to_stream(sender.subscribe())) } + /// The hierarchical path of this manager's root Task Scope + /// (e.g. "device-manager-3"). Every task this manager spawns lives at or + /// under this prefix in the global Task Registry. Useful for scoping + /// registry queries to a single manager instance — the registry is + /// process-global, so parallel managers must filter by their own prefix. + pub fn scope_path(&self) -> &str { + self.task_scope.path() + } + fn start_scanning(&self) -> ButtplugServerResultFuture { let command_sender = self.device_command_sender.clone(); async move { @@ -372,7 +396,13 @@ impl ServerDeviceManager { self.running.store(false, Ordering::Relaxed); let stop_scanning = self.stop_scanning(); let stop_devices = self.stop_devices(&StopCmdV4::default()); - let token = self.loop_cancellation_token.clone(); + // TaskScope is not Clone, but its CancellationToken is. Capture the token + // and path synchronously, but DO NOT cancel yet: device io tasks select + // biased on their token, so cancelling before the stop/disconnect commands + // drain would drop those queued commands and leave devices running. Run + // cleanup first, then cancel, then await the subtree draining. + let token = self.task_scope.token().clone(); + let scope_path = self.task_scope.path().to_owned(); async move { // Force stop scanning, otherwise we can disconnect and instantly try to reconnect while // cleaning up if we're still scanning. @@ -381,7 +411,12 @@ impl ServerDeviceManager { for device in devices.iter() { device.value().disconnect().await?; } + // Cleanup commands have drained; now cancel the scope and wait for every + // task under it to deregister. token.cancel(); + buttplug_core::util::task::registry() + .wait_empty_under(&scope_path) + .await; Ok(message::OkV0::default().into()) } .boxed() @@ -391,6 +426,9 @@ impl ServerDeviceManager { impl Drop for ServerDeviceManager { fn drop(&mut self) { info!("Dropping device manager!"); - self.loop_cancellation_token.cancel(); + // Explicitly cancel the scope here for clarity and to trigger shutdown + // eagerly (before field drop order kicks in). Field drop would also cancel + // via TaskScope::drop, but the explicit call makes the intent obvious. + self.task_scope.cancel(); } } diff --git a/crates/buttplug_server/src/device/server_device_manager_event_loop.rs b/crates/buttplug_server/src/device/server_device_manager_event_loop.rs index 6300586df..0b909139f 100644 --- a/crates/buttplug_server/src/device/server_device_manager_event_loop.rs +++ b/crates/buttplug_server/src/device/server_device_manager_event_loop.rs @@ -23,6 +23,7 @@ use crate::device::{ hardware::communication::{HardwareCommunicationManager, HardwareCommunicationManagerEvent}, protocol::ProtocolManager, }; +use buttplug_core::util::task::TaskScope; use dashmap::{DashMap, DashSet}; use futures::{FutureExt, future}; use std::sync::Arc; @@ -68,6 +69,9 @@ pub(super) struct ServerDeviceManagerEventLoop { connecting_devices: Arc>, /// Cancellation token for the event loop loop_cancellation_token: CancellationToken, + /// Scope owning all per-device task subtrees. Each device's tasks are spawned + /// through a `device-{...}` child of this scope. + devices_scope: TaskScope, /// Protocol map, for mapping user definitions to protocols protocol_manager: ProtocolManager, /// Optional sender for output observations, None when disabled @@ -81,6 +85,7 @@ impl ServerDeviceManagerEventLoop { device_config_manager: Arc, device_map: Arc>, loop_cancellation_token: CancellationToken, + devices_scope: TaskScope, server_sender: broadcast::Sender, device_comm_receiver: mpsc::Receiver, device_command_receiver: mpsc::Receiver, @@ -99,6 +104,7 @@ impl ServerDeviceManagerEventLoop { scanning_state: ScanningState::Idle, connecting_devices: Arc::new(DashSet::new()), loop_cancellation_token, + devices_scope, protocol_manager: ProtocolManager::default(), output_observation_sender, } @@ -292,45 +298,87 @@ impl ServerDeviceManagerEventLoop { let device_config_manager = self.device_config_manager.clone(); let connecting_devices = self.connecting_devices.clone(); let output_observation_sender = self.output_observation_sender.clone(); - let span = info_span!( - "device creation", - name = tracing::field::display(name), - address = tracing::field::display(address.clone()) - ); // Clone sender again for the forwarding task that build_device_handle will spawn let device_event_sender_for_forwarding = self.device_event_sender.clone(); - buttplug_core::util::async_manager::spawn( - async move { - match build_device_handle( - device_config_manager, - creator, - protocol_specializers, - device_event_sender_for_forwarding, - output_observation_sender, - ) - .await - { - Ok(device_handle) => { - if device_event_sender_clone - .send(InternalDeviceEvent::Connected(device_handle)) - .await - .is_err() - { - error!( - "Device manager disappeared before connection established, device will be dropped." - ); - } + // Create the per-device scope before spawning the bringup task. The + // device index is not known until identification completes inside + // build_device_handle, so both scopes are keyed by the device's stable + // address. + let device_scope = self.devices_scope.child(&format!("device-{address}")); + + // Spawn bringup through devices_scope so it is registered in the Task + // Registry for the duration of the connection attempt. This closes the + // shutdown race: wait_empty_under on the devices scope can only return + // zero after the bringup task deregisters, which happens *after* + // build_device_handle returns OR after cancellation is observed. + // build_device_handle registers the io and event-forwarding tasks (via + // task_scope.spawn) synchronously before returning, so on the success + // path the registry count never momentarily hits zero while work + // remains: + // 1. bringup registers (here) + // 2. io task registers (inside build_device_handle → spawn_device_task) + // 3. event-forwarding task registers (inside build_device_handle) + // 4. build_device_handle returns + // 5. bringup deregisters + // Step 2–3 happen before step 5, guaranteeing correct ordering. + // + // build_device_handle can stall indefinitely (e.g. a BLE connect that + // never completes), so the bringup MUST honor its cancellation token — + // otherwise ServerDeviceManager::shutdown's wait_empty_under would hang + // forever waiting for this task to deregister. We select on the token; + // on cancellation we drop the build_device_handle future, which drops + // device_scope and thereby cancels anything it has already spawned. + let bringup_address = address.clone(); + let bringup_result = self.devices_scope.spawn( + &format!("bringup-{address}"), + move |token| async move { + let address = bringup_address; + tokio::select! { + biased; + _ = token.cancelled() => { + info!( + "Device bringup for {address} cancelled before connection completed." + ); } - Err(e) => { - error!("Device errored while trying to connect: {:?}", e); + result = build_device_handle( + device_config_manager, + creator, + protocol_specializers, + device_event_sender_for_forwarding, + output_observation_sender, + device_scope, + ) => { + match result { + Ok(device_handle) => { + if device_event_sender_clone + .send(InternalDeviceEvent::Connected(device_handle)) + .await + .is_err() + { + error!( + "Device manager disappeared before connection established, device will be dropped." + ); + } + } + Err(e) => { + error!("Device errored while trying to connect: {:?}", e); + } + } } } + // Runs on every path (success, error, AND cancellation): the address + // must always leave the connecting set so a future scan can retry. connecting_devices.remove(&address); }, - span, ); + if let Err(e) = bringup_result { + debug!("Device bringup for {address} was not started because its scope is closed: {e}"); + // Registration was rejected synchronously, so the closure did not run + // and cannot perform its normal cleanup. + self.connecting_devices.remove(&address); + } } } } diff --git a/crates/buttplug_server/src/lib.rs b/crates/buttplug_server/src/lib.rs index 37e4cb9a5..fa3d76521 100644 --- a/crates/buttplug_server/src/lib.rs +++ b/crates/buttplug_server/src/lib.rs @@ -82,6 +82,9 @@ pub type ButtplugServerResultFuture = BoxFuture<'static, ButtplugServerResult>; /// Error enum for Buttplug Server configuration errors. #[derive(Error, Debug)] pub enum ButtplugServerError { + /// A server task could not be spawned during construction. + #[error(transparent)] + TaskSpawnError(#[from] buttplug_core::util::task::TaskSpawnError), /// DeviceConfigurationManager could not be built. #[error("The DeviceConfigurationManager could not be built: {0}")] DeviceConfigurationManagerError(ButtplugDeviceError), diff --git a/crates/buttplug_server/src/ping_timer.rs b/crates/buttplug_server/src/ping_timer.rs index 4b7ec5a9d..cc1a1c075 100644 --- a/crates/buttplug_server/src/ping_timer.rs +++ b/crates/buttplug_server/src/ping_timer.rs @@ -5,19 +5,19 @@ // Licensed under the BSD 3-Clause license. See LICENSE file in the project root // for full license information. -use buttplug_core::util::async_manager; +use buttplug_core::util::{async_manager, task::TaskScope}; use futures::Future; use std::{sync::Arc, time::Duration}; use tokio::{ select, sync::{Mutex, mpsc}, }; +use tokio_util::sync::CancellationToken; pub enum PingMessage { Ping, StartTimer, StopTimer, - End, } /// Internal ping timer task that monitors for ping timeouts. @@ -26,6 +26,7 @@ async fn ping_timer( max_ping_time: u32, mut ping_msg_receiver: mpsc::Receiver, on_ping_timeout: Arc>>, + token: CancellationToken, ) where F: FnOnce() + Send + 'static, { @@ -33,6 +34,9 @@ async fn ping_timer( let mut pinged = false; loop { select! { + _ = token.cancelled() => { + return; + } _ = async_manager::sleep(Duration::from_millis(max_ping_time.into())) => { if started { if !pinged { @@ -53,7 +57,6 @@ async fn ping_timer( PingMessage::StartTimer => started = true, PingMessage::StopTimer => started = false, PingMessage::Ping => pinged = true, - PingMessage::End => break, } } }; @@ -63,19 +66,8 @@ async fn ping_timer( pub struct PingTimer { max_ping_time: u32, ping_msg_sender: mpsc::Sender, -} - -impl Drop for PingTimer { - fn drop(&mut self) { - // This cannot block, otherwise it will throw in WASM contexts on - // destruction. We must use send(), not blocking_send(). - let sender = self.ping_msg_sender.clone(); - buttplug_core::spawn!("PingTimerDrop", async move { - if sender.send(PingMessage::End).await.is_err() { - debug!("Receiver does not exist, assuming ping timer event loop already dead."); - } - }); - } + // Dropping the timer drops the scope, which cancels the timer task. + _task_scope: TaskScope, } impl PingTimer { @@ -84,20 +76,26 @@ impl PingTimer { /// The callback is called once when the ping timer expires without receiving /// a ping message. If max_ping_time is 0, the timer is disabled and the /// callback will never be called. - pub fn new(max_ping_time: u32, on_ping_timeout: Option) -> Self + pub fn new( + max_ping_time: u32, + on_ping_timeout: Option, + task_scope: TaskScope, + ) -> Result where F: FnOnce() + Send + 'static, { let (sender, receiver) = mpsc::channel(256); if max_ping_time > 0 { let callback = Arc::new(Mutex::new(on_ping_timeout)); - let fut = ping_timer(max_ping_time, receiver, callback); - buttplug_core::spawn!("PingTimer", fut); + task_scope.spawn("timer", move |token| { + ping_timer(max_ping_time, receiver, callback, token) + })?; } - Self { + Ok(Self { max_ping_time, ping_msg_sender: sender, - } + _task_scope: task_scope, + }) } fn send_ping_msg(&self, msg: PingMessage) -> impl Future + use<> { diff --git a/crates/buttplug_server/src/server.rs b/crates/buttplug_server/src/server.rs index 61642068a..f428a1735 100644 --- a/crates/buttplug_server/src/server.rs +++ b/crates/buttplug_server/src/server.rs @@ -35,7 +35,7 @@ use buttplug_core::{ StopCmdV4, StopScanningV0, }, - util::stream::convert_broadcast_receiver_to_stream, + util::{stream::convert_broadcast_receiver_to_stream, task::TaskScope}, }; use futures::{ Stream, @@ -90,6 +90,9 @@ pub struct ButtplugServer { /// Broadcaster for server events. Receivers for this are handed out through the /// [ButtplugServer::event_stream()] method. output_sender: broadcast::Sender, + /// Root scope owning all tasks spawned for this server instance (e.g. the + /// ping timer). Cancelled on shutdown. + task_scope: TaskScope, } impl std::fmt::Debug for ButtplugServer { @@ -110,6 +113,7 @@ impl ButtplugServer { device_manager: Arc, state: Arc>, output_sender: broadcast::Sender, + task_scope: TaskScope, ) -> Self { ButtplugServer { server_name: server_name.to_owned(), @@ -118,6 +122,7 @@ impl ButtplugServer { device_manager, state, output_sender, + task_scope, } } @@ -233,8 +238,24 @@ impl ButtplugServer { pub fn shutdown(&self) -> ButtplugServerResultFuture { let device_manager = self.device_manager.clone(); - //let disconnect_future = self.disconnect(); - async move { device_manager.shutdown().await }.boxed() + // Capture the token and path synchronously, but cancel only after the + // device manager has finished its own cleanup. The server scope only owns + // the ping timer today, so cancelling early is harmless, but we align with + // the device manager's cleanup-before-cancel ordering for consistency. + let token = self.task_scope.token().clone(); + let scope_path = self.task_scope.path().to_owned(); + async move { + let result = device_manager.shutdown().await; + token.cancel(); + // Await only this server's own subtree. A shared device manager + // (with_shared_device_manager) outlives this server, so its subtree is + // NOT awaited here -- device_manager.shutdown() handles its own. + buttplug_core::util::task::registry() + .wait_empty_under(&scope_path) + .await; + result + } + .boxed() } /// Sends a [ButtplugClientMessage] to be parsed by the server (for handshake or ping), or passed diff --git a/crates/buttplug_server/src/server_builder.rs b/crates/buttplug_server/src/server_builder.rs index d5ab30e5b..8ee71b313 100644 --- a/crates/buttplug_server/src/server_builder.rs +++ b/crates/buttplug_server/src/server_builder.rs @@ -14,6 +14,7 @@ use super::{ use buttplug_core::{ errors::*, message::{self, ButtplugServerMessageV4, StopCmdV4}, + util::task::TaskScope, }; use buttplug_server_device_config::DeviceConfigurationManagerBuilder; use std::sync::{Arc, RwLock}; @@ -98,12 +99,16 @@ impl ButtplugServerBuilder { let ping_time = self.max_ping_time.unwrap_or(0); + // Root scope owning all tasks spawned for this server instance. + let task_scope = TaskScope::root("server"); + // Create the ping timeout callback if ping time is configured. // The callback handles: updating state, stopping devices, and sending error. let ping_timeout_callback = if ping_time > 0 { let state_clone = state.clone(); let device_manager_clone = self.device_manager.clone(); let output_sender_clone = output_sender.clone(); + let ping_timeout_scope = task_scope.child("ping-timeout"); Some(move || { error!("Ping out signal received, stopping server"); @@ -112,15 +117,22 @@ impl ButtplugServerBuilder { let mut state_guard = state_clone.write().expect("State lock poisoned"); *state_guard = ConnectionState::PingedOut; } - // Stop all devices (spawn async task since callback is sync) - buttplug_core::spawn!("PingTimeoutStopDevices", async move { + // Stop all devices (spawn async task since callback is sync). The + // callback is FnOnce, so the child scope moves in and is consumed by + // spawn_and_hold, keeping it alive for the duration of the task. + match ping_timeout_scope.spawn_and_hold("stop-devices", move |_token| async move { if let Err(e) = device_manager_clone .stop_devices(&StopCmdV4::default()) .await { error!("Could not stop devices on ping timeout: {:?}", e); } - }); + }) { + Ok(()) => {} + Err(e) => { + debug!("Skipping ping-timeout device stop because teardown closed its scope: {e}"); + } + } // Send error to output channel if output_sender_clone .send(ButtplugServerMessageV4::Error(message::ErrorV0::from( @@ -135,7 +147,11 @@ impl ButtplugServerBuilder { None }; - let ping_timer = Arc::new(PingTimer::new(ping_time, ping_timeout_callback)); + let ping_timer = Arc::new(PingTimer::new( + ping_time, + ping_timeout_callback, + task_scope.child("ping-timer"), + )?); // Assuming everything passed, return the server. Ok(ButtplugServer::new( @@ -145,6 +161,7 @@ impl ButtplugServerBuilder { self.device_manager.clone(), state, output_sender, + task_scope, )) } } diff --git a/crates/buttplug_tests/CLAUDE.md b/crates/buttplug_tests/CLAUDE.md index 31f5b50a1..e5408baa3 100644 --- a/crates/buttplug_tests/CLAUDE.md +++ b/crates/buttplug_tests/CLAUDE.md @@ -54,4 +54,5 @@ Tests run across multiple protocol spec versions (v0–v4) via version-specific - `test_message_downgrades.rs` — Protocol version downgrade path tests - `test_disabled_device_features.rs` — Tests for user config feature disabling - `test_output_observations.rs` — Integration tests for output observability (observation stream, filtering, multi-device, enable/disable) +- `test_task_lifecycle.rs` — Integration test asserting the global Task Registry has no live tasks under the server's scope after server shutdown (verifies `TaskScope` ownership leaves no leaked tasks) - `test_websocket_connectors.rs` / `test_websocket_device_comm_manager.rs` — WebSocket transport integration tests diff --git a/crates/buttplug_tests/tests/test_task_lifecycle.rs b/crates/buttplug_tests/tests/test_task_lifecycle.rs new file mode 100644 index 000000000..d3b7c7d93 --- /dev/null +++ b/crates/buttplug_tests/tests/test_task_lifecycle.rs @@ -0,0 +1,531 @@ +// Buttplug Rust Source Code File - See https://buttplug.io for more info. +// +// Copyright 2016-2026 Nonpolynomial Labs LLC. All rights reserved. +// +// Licensed under the BSD 3-Clause license. See LICENSE file in the project root +// for full license information. + +mod util; + +use buttplug_core::{ + message::{ + BUTTPLUG_CURRENT_API_MAJOR_VERSION, + BUTTPLUG_CURRENT_API_MINOR_VERSION, + ButtplugServerMessageV4, + OutputCmdV4, + OutputCommand, + OutputValue, + RequestServerInfoV4, + StartScanningV0, + StopCmdV4, + }, + util::task::registry, +}; +use buttplug_server::{device::hardware::HardwareCommand, message::ButtplugClientMessageVariant}; +use futures::{StreamExt, pin_mut}; +use std::time::Duration; +use util::TestDeviceChannelHost; +use util::stalling_device_communication_manager::StallingDeviceCommunicationManagerBuilder; +use util::{ + test_server_with_comm_manager, + test_server_with_device, + test_server_with_device_and_message_gap, +}; + +/// The Aneros "Massage Demo" stop write for feature 0: `[0xF1, 0x00]`. A vibrate +/// on the same feature writes `[0xF1, 0x40]`; the stop resets it to zero. See +/// `device_test_case/test_aneros_protocol.yaml` for the full sequence. +const ANEROS_STOP_WRITE_FEATURE_0: [u8; 2] = [0xF1, 0x00]; + +/// Non-blockingly drain every hardware write the test device has recorded so +/// far and return whether any of them is the feature-0 stop write. We use +/// `try_recv` so the check reflects exactly what was on the wire *at the moment +/// stop/shutdown resolved* — a write still sitting in the device io task's batch +/// window has not reached the host channel yet and will not be counted. +fn recorded_a_stop_write(host: &mut TestDeviceChannelHost) -> bool { + let mut saw_stop = false; + while let Ok(command) = host.receiver.try_recv() { + if let HardwareCommand::Write(write) = command + && write.data().as_slice() == ANEROS_STOP_WRITE_FEATURE_0 + { + saw_stop = true; + } + } + saw_stop +} + +/// Bring a "Massage Demo" device online under `server`, returning its device +/// index. Mirrors the handshake/scan/connect dance the other lifecycle tests do. +async fn bring_device_online(server: &buttplug_server::ButtplugServer, client_name: &str) -> u32 { + let recv = server.server_version_event_stream(); + pin_mut!(recv); + server + .parse_message(ButtplugClientMessageVariant::V4( + RequestServerInfoV4::new( + client_name, + BUTTPLUG_CURRENT_API_MAJOR_VERSION, + BUTTPLUG_CURRENT_API_MINOR_VERSION, + ) + .into(), + )) + .await + .expect("server info request should succeed"); + server + .parse_message(ButtplugClientMessageVariant::V4( + StartScanningV0::default().into(), + )) + .await + .expect("start scanning should succeed"); + tokio::time::timeout(Duration::from_secs(5), async { + while let Some(msg) = recv.next().await { + if let ButtplugServerMessageV4::DeviceList(list) = msg + && let Some((&idx, _)) = list.devices().iter().next() + { + return idx; + } + } + panic!("device event stream ended before a device connected"); + }) + .await + .expect("timed out waiting for device to connect") +} + +/// Bring a real (test-harness) device online and confirm that, once the server +/// is shut down and dropped, no tasks remain registered under the scope tree. +/// +/// This exercises the *device task* lifecycle specifically: a connected device +/// spawns an `io` task whose only exit path is its `token.cancelled()` select +/// arm. If scope cancellation were removed, that task would leak and this test +/// would fail with a non-empty leaked-task list. (A server with no device is +/// insufficient — the device-manager event loop also exits on channel drop, so +/// it cannot prove cancellation actually fires.) +#[tokio::test] +async fn test_server_shutdown_leaves_no_tasks() { + let baseline: Vec = registry().snapshot().into_iter().map(|t| t.path).collect(); + + // Hold the channel so the device stays connected. + let (server, _channel) = test_server_with_device("Massage Demo"); + + let recv = server.server_version_event_stream(); + pin_mut!(recv); + + server + .parse_message(ButtplugClientMessageVariant::V4( + RequestServerInfoV4::new( + "Task Lifecycle Test", + BUTTPLUG_CURRENT_API_MAJOR_VERSION, + BUTTPLUG_CURRENT_API_MINOR_VERSION, + ) + .into(), + )) + .await + .expect("server info request should succeed"); + server + .parse_message(ButtplugClientMessageVariant::V4( + StartScanningV0::default().into(), + )) + .await + .expect("start scanning should succeed"); + + // Wait until the device is actually connected — this is what spawns the + // per-device `io` task that we want to prove gets cleaned up. + tokio::time::timeout(Duration::from_secs(5), async { + while let Some(msg) = recv.next().await { + if let ButtplugServerMessageV4::DeviceList(list) = msg { + if !list.devices().is_empty() { + return; + } + } + } + panic!("device event stream ended before a device connected"); + }) + .await + .expect("timed out waiting for device to connect"); + + // Device is up. Scope all subsequent leak checks to THIS server's own + // device-manager subtree: the registry is process-global, so other tests + // running in parallel must not pollute these assertions. We ask the manager + // for its own scope path directly rather than guessing it from the global + // registry snapshot — guessing is racy, because a concurrent test's + // `device-manager-N` tasks are also "new" relative to our baseline and could + // be picked instead of ours. + let scope_prefix: String = server.device_manager().scope_path().to_owned(); + + // Sanity: the registry must now show this server spawned per-device tasks + // under its own subtree. + let new_tasks: Vec = registry() + .snapshot() + .into_iter() + .map(|t| t.path) + .filter(|p| !baseline.contains(p) && p.starts_with(&format!("{scope_prefix}/"))) + .collect(); + assert!( + !new_tasks.is_empty(), + "expected scope-spawned tasks under {scope_prefix} after a device connected" + ); + assert!( + new_tasks.iter().any(|p| p.contains("/device-")), + "expected a per-device task in the registry, got: {new_tasks:?}" + ); + + // `shutdown()` is contractually responsible for draining every task it + // spawned: it cancels the scope tree and awaits the registry going empty + // under its path before returning. We assert emptiness *before* dropping the + // server, so that ordinary drop-of-channels teardown cannot mask a missing + // cancellation arm. If any scope-spawned task fails to observe cancellation, + // it will still be parked on its event stream / receiver here and show up as + // leaked. We hold `_channel` alive across this check precisely so the device + // event stream does not close on its own. + server.shutdown().await.expect("server shutdown errored"); + + let leaked = registry().live_count_under(&scope_prefix); + assert_eq!( + leaked, 0, + "shutdown() returned but {leaked} task(s) are still registered under {scope_prefix}" + ); + + // Dropping the server must not resurrect or strand anything either. Give a + // short grace period and confirm this server's subtree stays empty. + drop(server); + drop(_channel); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if registry().live_count_under(&scope_prefix) == 0 { + return; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .unwrap_or_else(|_| { + let leaked = registry().live_count_under(&scope_prefix); + panic!("leaked {leaked} task(s) under {scope_prefix} after drop"); + }); +} + +/// Shutdown-under-load smoke test: with a device connected, scanning still +/// active, and the device in a non-zero output state, `shutdown()` must drive +/// its cleanup (stop_scanning / stop_devices / per-device disconnect) through +/// the live event loop, drain every task under its scope, and return Ok within +/// a bounded time — it must neither hang nor strand tasks. +/// +/// SCOPE / what this does NOT verify: this is a smoke test, not a regression +/// test for shutdown *ordering*. It does not prove that cleanup runs *before* +/// scope cancellation — reverting the cleanup-before-cancel ordering leaves this +/// test green, because the contract it checks (shutdown completes and its +/// subtree drains) holds under both orderings with this harness. The +/// cleanup-before-cancel ordering is the correct production behavior, but a +/// test that goes RED on that specific regression is not achievable here (see +/// NOTE). This test guards against the coarser failure mode: a shutdown that +/// hangs or leaks tasks when invoked under realistic load. +/// +/// NOTE on why ordering can't be observed here: the stronger "observe the +/// device's actual stop write" assertion is infeasible with this harness. The +/// test hardware sets a 1ms message_gap (see TestHardwareConnector::specialize), +/// so the device io task batches commands; during shutdown the per-device +/// `disconnect()` fires a `Disconnected` hardware event that tears the io task +/// down inside that 1ms batch window, dropping the pending (batched) stop write +/// regardless of cancel ordering. That teardown race is independent of any +/// ordering bug, so a write-observation assertion is inherently flaky here. An +/// instrumented-ordering variant was also attempted and found inherently flaky +/// with this harness, so it is deliberately not pursued. +/// +/// UPDATE: with the stop-command write acknowledgement (DeviceTaskMessage ack) +/// and a configurable per-device message_gap, the write observation IS now +/// possible — see `test_stop_resolves_only_after_stop_write_reaches_hardware` +/// and `test_shutdown_writes_stop_before_resolving` below, which retire the +/// limitation described in the NOTE above for the acked stop path. This test +/// remains as the coarse hang/leak smoke test. +#[tokio::test] +async fn test_shutdown_under_load_drains_subtree() { + // Hold the channel so the device stays connected through shutdown. + let (server, _channel) = test_server_with_device("Massage Demo"); + + let recv = server.server_version_event_stream(); + pin_mut!(recv); + + server + .parse_message(ButtplugClientMessageVariant::V4( + RequestServerInfoV4::new( + "Shutdown Under Load Test", + BUTTPLUG_CURRENT_API_MAJOR_VERSION, + BUTTPLUG_CURRENT_API_MINOR_VERSION, + ) + .into(), + )) + .await + .expect("server info request should succeed"); + + // Start scanning and leave it running: shutdown's stop_scanning must drain + // through the event loop before the scope is cancelled. + server + .parse_message(ButtplugClientMessageVariant::V4( + StartScanningV0::default().into(), + )) + .await + .expect("start scanning should succeed"); + + // Wait for the device to connect so its io / event-forwarding tasks exist + // under the scope alongside the still-running scanning state. + let device_index = tokio::time::timeout(Duration::from_secs(5), async { + while let Some(msg) = recv.next().await { + if let ButtplugServerMessageV4::DeviceList(list) = msg + && let Some((&idx, _)) = list.devices().iter().next() + { + return idx; + } + } + panic!("device event stream ended before a device connected"); + }) + .await + .expect("timed out waiting for device to connect"); + + // Put the device into an actively-running state so there is real cleanup to + // perform (a non-zero output that StopCmd must reset). + server + .parse_message(ButtplugClientMessageVariant::V4( + OutputCmdV4::new( + device_index, + 0, + OutputCommand::Vibrate(OutputValue::new(50)), + ) + .into(), + )) + .await + .expect("vibrate command should succeed"); + + // Identify this server's own device-manager scope prefix so the leak check + // below is isolated from any other test running in parallel against the + // global registry. We ask the manager directly rather than guessing from the + // global registry snapshot, which would race with concurrent tests' own + // `device-manager-N` roots. + let scope_prefix: String = server.device_manager().scope_path().to_owned(); + + // shutdown() must drive cleanup (stop_scanning + stop_devices + disconnect) + // through the live event loop and only then cancel the scope, returning Ok + // within a bounded time. + tokio::time::timeout(Duration::from_secs(10), server.shutdown()) + .await + .expect("shutdown did not resolve in time — cleanup likely raced against cancellation") + .expect("server shutdown errored"); + + // shutdown() is contractually responsible for draining every task under its + // own scope. Inspect only this server's subtree to stay parallel-safe. + let leaked = registry().live_count_under(&scope_prefix); + assert_eq!( + leaked, 0, + "shutdown() returned but {leaked} task(s) are still registered under {scope_prefix}" + ); +} + +/// Regression test for the cancellable-bringup fix (fix 2): `shutdown()` must +/// not hang when a device bringup is stalled in `connect()`. +/// +/// The stalling comm manager emits one `DeviceFound` on scan; the device-manager +/// event loop spawns a bringup task that awaits `connect()`, which never +/// resolves. `shutdown()` cancels the device-manager scope and then +/// `wait_empty_under`s its subtree. The bringup task only deregisters once it +/// observes cancellation via the `biased` select on its token — without that +/// select it would await `connect()` forever and `shutdown()` would never +/// resolve. +/// +/// RED evidence: replacing the bringup's `move |token|` select with the +/// non-cancellable `move |_token|` form makes this test time out at the 10s +/// bound and fail. With the fix in place it resolves promptly. +#[tokio::test] +async fn test_shutdown_resolves_with_stalled_bringup() { + let server = test_server_with_comm_manager(StallingDeviceCommunicationManagerBuilder::default()); + + server + .parse_message(ButtplugClientMessageVariant::V4( + RequestServerInfoV4::new( + "Stalled Bringup Test", + BUTTPLUG_CURRENT_API_MAJOR_VERSION, + BUTTPLUG_CURRENT_API_MINOR_VERSION, + ) + .into(), + )) + .await + .expect("server info request should succeed"); + + // Kick off scanning so a device is found and a bringup task begins — and then + // stalls in connect(). + server + .parse_message(ButtplugClientMessageVariant::V4( + StartScanningV0::default().into(), + )) + .await + .expect("start scanning should succeed"); + + // Give the bringup task time to spawn and enter (and block in) connect(). + tokio::time::sleep(Duration::from_millis(200)).await; + + // shutdown() must cancel the stalled bringup and return within the bound. + // Without the biased select on the bringup token this hangs forever. + tokio::time::timeout(Duration::from_secs(10), server.shutdown()) + .await + .expect("shutdown hung with a stalled device bringup — bringup is not honoring its cancellation token") + .expect("server shutdown errored"); +} + +/// A bring-up that has entered `connect()` is shut down deterministically. The +/// retained device-manager scope handle cannot register a task after shutdown, +/// and the rejected bring-up cannot add a device after shutdown resolves. +#[tokio::test] +async fn test_shutdown_rejects_bringup_and_drains_manager_scope() { + let builder = StallingDeviceCommunicationManagerBuilder::default(); + let state = builder.state(); + let server = test_server_with_comm_manager(builder); + let scope_prefix = server.device_manager().scope_path().to_owned(); + let events = server.server_version_event_stream(); + pin_mut!(events); + + server + .parse_message(ButtplugClientMessageVariant::V4( + RequestServerInfoV4::new( + "Bringup Shutdown Race Test", + BUTTPLUG_CURRENT_API_MAJOR_VERSION, + BUTTPLUG_CURRENT_API_MINOR_VERSION, + ) + .into(), + )) + .await + .expect("server info request should succeed"); + server + .parse_message(ButtplugClientMessageVariant::V4( + StartScanningV0::default().into(), + )) + .await + .expect("start scanning should succeed"); + + tokio::time::timeout(Duration::from_secs(5), state.connect_started.notified()) + .await + .expect("bring-up did not reach connect() in time"); + + tokio::time::timeout(Duration::from_secs(10), server.shutdown()) + .await + .expect("shutdown did not resolve in time") + .expect("server shutdown errored"); + + assert_eq!(registry().live_count_under(&scope_prefix), 0); + + // The stalling bring-up never reaches device construction, so shutdown must + // not be followed by a device-added event. Drain already-ready events only; + // do not sleep or loop to manufacture a race. + while let Ok(Some(message)) = tokio::time::timeout(Duration::from_millis(10), events.next()).await + { + if let ButtplugServerMessageV4::DeviceList(list) = message { + assert!( + list.devices().is_empty(), + "a device was added after shutdown resolved" + ); + } + } + assert_eq!(registry().live_count_under(&scope_prefix), 0); +} + +/// Test A — the stop-write-acknowledgement contract: a `StopCmd` must not +/// resolve until the resulting stop write has actually reached the hardware, +/// even when the device io task is batching commands over a long message gap. +/// +/// This retires the limitation documented on `test_shutdown_under_load_drains_subtree` +/// (the 1ms harness gap made write observation flaky). Here we deliberately give +/// the device a 500ms message gap so the batching window is large and the race +/// is deterministic: an active vibrate write lands, the stop write is queued +/// into that 500ms batch, and we assert the stop write is on the wire *by the +/// time the stop message resolves*. +/// +/// RED (pre-fix, `git stash` Tasks 1-2): `handle_hardware_commands` fire-and-forgets +/// the stop write into the io channel and `parse_message` for the StopCmd resolves +/// immediately, while the write sits unflushed in the 500ms batch. `try_recv` +/// finds no stop write and this assertion fails deterministically. +/// +/// GREEN (with the ack-on-write fix): the stop path requests a write +/// acknowledgement; the io task urgent-flushes the batch and fires the ack, so +/// StopCmd only resolves after the write is on the wire. +#[tokio::test] +async fn test_stop_resolves_only_after_stop_write_reaches_hardware() { + // 500ms gap: large enough that, pre-fix, the stop write provably has not been + // flushed by the time StopCmd resolves (which is ~immediate). + let (server, mut device) = + test_server_with_device_and_message_gap("Massage Demo", Duration::from_millis(500)); + + let device_index = bring_device_online(&server, "Stop Write Ack Test").await; + + // Put the device into an actively-running state so the stop has real work to + // flush. This vibrate write itself enters the batch window. + server + .parse_message(ButtplugClientMessageVariant::V4( + OutputCmdV4::new( + device_index, + 0, + OutputCommand::Vibrate(OutputValue::new(50)), + ) + .into(), + )) + .await + .expect("vibrate command should succeed"); + + // Stop. With ack-on-write, parse_message for StopCmd resolves only after the + // stop write has been flushed to hardware. + server + .parse_message(ButtplugClientMessageVariant::V4( + StopCmdV4::new(Some(device_index), None, true, true).into(), + )) + .await + .expect("stop command should succeed"); + + // The instant stop resolves, the stop write must already be on the wire. No + // sleep here: that is the whole point — we are asserting ordering, not + // eventual delivery. + assert!( + recorded_a_stop_write(&mut device), + "StopCmd resolved but the stop write was not yet recorded on the hardware channel — \ + it is still sitting in the device io task's batch window" + ); +} + +/// Test B — the original incident: with a batched device in an active output +/// state, `server.shutdown()` must drive the per-device stop write all the way +/// to hardware before it resolves. This is the assertion the task-scope work +/// could not make with the 1ms harness gap (the disconnect tore the io task down +/// inside the batch window, dropping the pending stop write). With ack-on-write, +/// stop_devices waits for the write before shutdown proceeds to disconnect. +/// +/// RED (pre-fix): shutdown's stop_devices fire-and-forgets the stop write, then +/// disconnect tears down the io task mid-batch and the write is dropped — no stop +/// write is ever recorded. +/// +/// GREEN (with the fix): the stop write is acknowledged before disconnect, so it +/// is recorded before shutdown resolves. +#[tokio::test] +async fn test_shutdown_writes_stop_before_resolving() { + let (server, mut device) = + test_server_with_device_and_message_gap("Massage Demo", Duration::from_millis(500)); + + let device_index = bring_device_online(&server, "Shutdown Stop Write Test").await; + + server + .parse_message(ButtplugClientMessageVariant::V4( + OutputCmdV4::new( + device_index, + 0, + OutputCommand::Vibrate(OutputValue::new(50)), + ) + .into(), + )) + .await + .expect("vibrate command should succeed"); + + tokio::time::timeout(Duration::from_secs(10), server.shutdown()) + .await + .expect("shutdown did not resolve in time") + .expect("server shutdown errored"); + + // By the time shutdown resolved, the stop write must have reached hardware. + assert!( + recorded_a_stop_write(&mut device), + "shutdown() resolved but the device's stop write was never recorded — \ + the pending stop write was dropped when the io task was torn down mid-batch" + ); +} diff --git a/crates/buttplug_tests/tests/util/mod.rs b/crates/buttplug_tests/tests/util/mod.rs index c19a80b54..a82460353 100644 --- a/crates/buttplug_tests/tests/util/mod.rs +++ b/crates/buttplug_tests/tests/util/mod.rs @@ -15,6 +15,8 @@ pub mod test_device_manager; pub use delay_device_communication_manager::DelayDeviceCommunicationManagerBuilder; #[allow(dead_code)] pub mod channel_transport; +#[allow(dead_code)] +pub mod stalling_device_communication_manager; use buttplug_client::ButtplugClient; use buttplug_client_in_process::ButtplugInProcessClientConnectorBuilder; use buttplug_server::{ @@ -168,6 +170,22 @@ pub fn test_server_with_device(device_type: &str) -> (ButtplugServer, TestDevice (test_server_with_comm_manager(builder), device) } +/// Like `test_server_with_device`, but the device's io task batches commands +/// over `message_gap`. A large gap widens the stop-write batching window so +/// tests can deterministically observe whether stop commands are flushed to +/// hardware before stop/shutdown resolve. +#[allow(dead_code)] +pub fn test_server_with_device_and_message_gap( + device_type: &str, + message_gap: std::time::Duration, +) -> (ButtplugServer, TestDeviceChannelHost) { + let mut builder = TestDeviceCommunicationManagerBuilder::default(); + let device = builder + .add_test_device_with_message_gap(&TestDeviceIdentifier::new(device_type, None), message_gap); + + (test_server_with_comm_manager(builder), device) +} + #[allow(dead_code)] pub fn test_server_v4_with_device(device_type: &str) -> (ButtplugServer, TestDeviceChannelHost) { let mut builder = TestDeviceCommunicationManagerBuilder::default(); diff --git a/crates/buttplug_tests/tests/util/stalling_device_communication_manager.rs b/crates/buttplug_tests/tests/util/stalling_device_communication_manager.rs new file mode 100644 index 000000000..0d3540302 --- /dev/null +++ b/crates/buttplug_tests/tests/util/stalling_device_communication_manager.rs @@ -0,0 +1,136 @@ +// Buttplug Rust Source Code File - See https://buttplug.io for more info. +// +// Copyright 2016-2026 Nonpolynomial Labs LLC. All rights reserved. +// +// Licensed under the BSD 3-Clause license. See LICENSE file in the project root +// for full license information. + +//! A test comm manager whose device bringup stalls forever in `connect()`. +//! +//! On `start_scanning` it emits a single `DeviceFound` event carrying a +//! [StallingHardwareConnector]. The device-manager event loop spawns a bringup +//! task that awaits `connect()` — which never resolves — modelling a real BLE +//! connect that hangs. This is the precise scenario the cancellable-bringup fix +//! guards: without a `biased` select on the bringup token, `shutdown()`'s +//! `wait_empty_under` would block forever waiting for the bringup task to +//! deregister. + +use async_trait::async_trait; +use buttplug_core::{ButtplugResultFuture, errors::ButtplugDeviceError}; +use buttplug_server::device::hardware::{ + HardwareConnector, + HardwareSpecializer, + communication::{ + HardwareCommunicationManager, + HardwareCommunicationManagerBuilder, + HardwareCommunicationManagerEvent, + }, +}; +use buttplug_server_device_config::{BluetoothLESpecifier, ProtocolCommunicationSpecifier}; +use futures::FutureExt; +use log::error; +use std::collections::HashMap; +use tokio::sync::{Notify, mpsc::Sender}; + +/// Shared state used to deterministically observe that bring-up entered connect(). +#[derive(Clone, Debug, Default)] +pub struct StallingDeviceState { + pub connect_started: std::sync::Arc, +} + +/// A `HardwareConnector` whose `connect()` future never resolves. +#[derive(Debug)] +struct StallingHardwareConnector { + specifier: ProtocolCommunicationSpecifier, + state: StallingDeviceState, +} + +#[async_trait] +impl HardwareConnector for StallingHardwareConnector { + fn specifier(&self) -> ProtocolCommunicationSpecifier { + self.specifier.clone() + } + + async fn connect(&mut self) -> Result, ButtplugDeviceError> { + // Block forever: this models a device connect that hangs and never returns. + // The bringup task must drop this future when its cancellation token fires. + self.state.connect_started.notify_waiters(); + std::future::pending::<()>().await; + unreachable!("stalling connector connect() should never resolve"); + } +} + +#[derive(Clone, Default)] +pub struct StallingDeviceCommunicationManagerBuilder { + state: StallingDeviceState, +} + +impl StallingDeviceCommunicationManagerBuilder { + pub fn state(&self) -> StallingDeviceState { + self.state.clone() + } +} + +impl HardwareCommunicationManagerBuilder for StallingDeviceCommunicationManagerBuilder { + fn finish( + &mut self, + sender: Sender, + ) -> Box { + Box::new(StallingDeviceCommunicationManager { + device_sender: sender, + state: self.state.clone(), + }) + } +} + +pub struct StallingDeviceCommunicationManager { + device_sender: Sender, + state: StallingDeviceState, +} + +impl HardwareCommunicationManager for StallingDeviceCommunicationManager { + fn name(&self) -> &'static str { + "StallingDeviceCommunicationManager" + } + + fn start_scanning(&mut self) -> ButtplugResultFuture { + let device_sender = self.device_sender.clone(); + let state = self.state.clone(); + async move { + // "Massage Demo" is a known test device name with a real protocol config, + // so bringup proceeds into connect() (which then stalls). + let specifier = ProtocolCommunicationSpecifier::BluetoothLE( + BluetoothLESpecifier::new_from_device("Massage Demo", &HashMap::new(), &[]), + ); + let connector = StallingHardwareConnector { + specifier, + state: state.clone(), + }; + if device_sender + .send(HardwareCommunicationManagerEvent::DeviceFound { + name: "Massage Demo".to_owned(), + address: "stalling-device-0".to_owned(), + creator: Box::new(connector), + }) + .await + .is_err() + { + error!("Device channel no longer open."); + } + Ok(()) + } + .boxed() + } + + fn stop_scanning(&mut self) -> ButtplugResultFuture { + async { Ok(()) }.boxed() + } + + fn scanning_status(&self) -> bool { + false + } + + fn can_scan(&self) -> bool { + true + } +} diff --git a/crates/buttplug_tests/tests/util/test_device_manager/test_device.rs b/crates/buttplug_tests/tests/util/test_device_manager/test_device.rs index 91916c916..5e4c56654 100644 --- a/crates/buttplug_tests/tests/util/test_device_manager/test_device.rs +++ b/crates/buttplug_tests/tests/util/test_device_manager/test_device.rs @@ -115,12 +115,16 @@ impl HardwareSpecializer for TestHardwareSpecializer { } } } + // The device carries its own message gap (default 1ms for the slight delay + // multi-message protocols expect). Tests that need a large batching window + // to make the stop-write race deterministic override it via + // `add_test_device_with_message_gap`. + let message_gap = device.message_gap(); let hardware = Hardware::new( &device.name(), &device.address(), &endpoints, - // Add slight delay for protocols with multiple messages. - &Some(Duration::from_millis(1)), + &Some(message_gap), false, Box::new(device), ); @@ -153,10 +157,15 @@ pub fn new_device_channel() -> (TestDeviceChannelHost, TestDeviceChannelDevice) ) } +/// Default inter-message gap for test hardware. A slight delay multi-message +/// protocols expect; small enough that ordinary tests are unaffected. +const DEFAULT_MESSAGE_GAP: Duration = Duration::from_millis(1); + pub struct TestDevice { name: String, address: String, endpoints: HashSet, + message_gap: Duration, test_device_channel: mpsc::Sender, event_sender: broadcast::Sender, subscribed_endpoints: Arc>, @@ -166,6 +175,16 @@ pub struct TestDevice { impl TestDevice { #[allow(dead_code)] pub fn new(name: &str, address: &str, test_device_channel: TestDeviceChannelDevice) -> Self { + Self::new_with_message_gap(name, address, DEFAULT_MESSAGE_GAP, test_device_channel) + } + + #[allow(dead_code)] + pub fn new_with_message_gap( + name: &str, + address: &str, + message_gap: Duration, + test_device_channel: TestDeviceChannelDevice, + ) -> Self { let (event_sender, _) = broadcast::channel(256); let event_sender_clone = event_sender.clone(); @@ -210,6 +229,7 @@ impl TestDevice { name: name.to_owned(), address: address.to_owned(), endpoints: HashSet::new(), + message_gap, test_device_channel: command_sender, event_sender, subscribed_endpoints, @@ -221,6 +241,10 @@ impl TestDevice { self.endpoints.insert(*endpoint); } + pub fn message_gap(&self) -> Duration { + self.message_gap + } + pub fn name(&self) -> String { self.name.clone() } diff --git a/crates/buttplug_tests/tests/util/test_device_manager/test_device_comm_manager.rs b/crates/buttplug_tests/tests/util/test_device_manager/test_device_comm_manager.rs index 6e0d18a70..81fe9c276 100644 --- a/crates/buttplug_tests/tests/util/test_device_manager/test_device_comm_manager.rs +++ b/crates/buttplug_tests/tests/util/test_device_manager/test_device_comm_manager.rs @@ -30,7 +30,7 @@ use std::{ Arc, atomic::{AtomicBool, Ordering}, }, - time::{SystemTime, UNIX_EPOCH}, + time::{Duration, SystemTime, UNIX_EPOCH}, }; use tokio::sync::mpsc::Sender; @@ -74,8 +74,11 @@ impl TestDeviceIdentifier { } } +/// Default inter-message gap for test hardware, matching `TestDevice`'s default. +const DEFAULT_MESSAGE_GAP: Duration = Duration::from_millis(1); + pub struct TestDeviceCommunicationManagerBuilder { - devices: Option>, + devices: Option>, } impl Default for TestDeviceCommunicationManagerBuilder { @@ -88,12 +91,24 @@ impl Default for TestDeviceCommunicationManagerBuilder { impl TestDeviceCommunicationManagerBuilder { pub fn add_test_device(&mut self, device: &TestDeviceIdentifier) -> TestDeviceChannelHost { + self.add_test_device_with_message_gap(device, DEFAULT_MESSAGE_GAP) + } + + /// Register a test device whose io task batches commands over `message_gap`. + /// A large gap (e.g. 500ms) makes the stop-write batching race deterministic + /// for tests asserting that stop commands are flushed before stop resolves. + #[allow(dead_code)] + pub fn add_test_device_with_message_gap( + &mut self, + device: &TestDeviceIdentifier, + message_gap: Duration, + ) -> TestDeviceChannelHost { let (host_channel, device_channel) = new_device_channel(); self .devices .as_mut() .expect("Devices vec does not exist, is this running twice?") - .push((device.clone(), device_channel)); + .push((device.clone(), message_gap, device_channel)); host_channel } } @@ -115,26 +130,28 @@ impl HardwareCommunicationManagerBuilder for TestDeviceCommunicationManagerBuild fn new_uninitialized_ble_test_device( identifier: &TestDeviceIdentifier, + message_gap: Duration, device_channel: TestDeviceChannelDevice, ) -> TestHardwareConnector { let address = identifier.address.clone(); let specifier = ProtocolCommunicationSpecifier::BluetoothLE( BluetoothLESpecifier::new_from_device(&identifier.name, &HashMap::new(), &[]), ); - let hardware = TestDevice::new(&identifier.name, &address, device_channel); + let hardware = + TestDevice::new_with_message_gap(&identifier.name, &address, message_gap, device_channel); TestHardwareConnector::new(specifier, hardware) } pub struct TestDeviceCommunicationManager { device_sender: Sender, - devices: Vec<(TestDeviceIdentifier, TestDeviceChannelDevice)>, + devices: Vec<(TestDeviceIdentifier, Duration, TestDeviceChannelDevice)>, is_scanning: Arc, } impl TestDeviceCommunicationManager { pub fn new( device_sender: Sender, - devices: Vec<(TestDeviceIdentifier, TestDeviceChannelDevice)>, + devices: Vec<(TestDeviceIdentifier, Duration, TestDeviceChannelDevice)>, ) -> Self { Self { device_sender, @@ -156,8 +173,8 @@ impl HardwareCommunicationManager for TestDeviceCommunicationManager { let mut events = vec![]; - while let Some((device, test_channel)) = self.devices.pop() { - let device_creator = new_uninitialized_ble_test_device(&device, test_channel); + while let Some((device, message_gap, test_channel)) = self.devices.pop() { + let device_creator = new_uninitialized_ble_test_device(&device, message_gap, test_channel); events.push(HardwareCommunicationManagerEvent::DeviceFound { name: device.name.clone(), diff --git a/crates/intiface_engine/CHANGELOG.md b/crates/intiface_engine/CHANGELOG.md index 4f5e5d67f..d764baad7 100644 --- a/crates/intiface_engine/CHANGELOG.md +++ b/crates/intiface_engine/CHANGELOG.md @@ -1,3 +1,19 @@ +# 5.0.0 (2026-07-19) + +## Breaking Changes + +- Added task lifecycle messages to the public engine frontend API: + - `EngineMessage::TaskStarted` reports a newly registered task. + - `EngineMessage::TaskEnded` reports task completion, cancellation, or panic. + - `EngineMessage::TaskList` returns a snapshot of live tasks. +- Added `IntifaceMessage::RequestTaskList` for requesting a live-task snapshot. +- Added the required `emit_task_events: bool` field to `EngineOptionsExternal`. Downstream code using struct literals must provide this field (or use `..Default::default()`). +- Downstream exhaustive matches over `EngineMessage` or `IntifaceMessage` may need new arms for the task lifecycle variants. + +## Features + +- Task lifecycle event emission is opt-in. Set `emit_task_events` to `true` to receive `TaskStarted` and `TaskEnded` events; task events remain disabled by default. + # 4.0.4 (2026-06-01) ## Features diff --git a/crates/intiface_engine/Cargo.toml b/crates/intiface_engine/Cargo.toml index c0c0754b5..6d4f5423f 100644 --- a/crates/intiface_engine/Cargo.toml +++ b/crates/intiface_engine/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "intiface-engine" -version = "4.0.4" +version = "5.0.0" authors = ["Nonpolynomial Labs, LLC "] description = "CLI and Library frontend for the Buttplug sex toy control library" license = "BSD-3-Clause" diff --git a/crates/intiface_engine/src/engine.rs b/crates/intiface_engine/src/engine.rs index a4a81e44c..81780d02d 100644 --- a/crates/intiface_engine/src/engine.rs +++ b/crates/intiface_engine/src/engine.rs @@ -306,8 +306,15 @@ impl IntifaceEngine { let event_receiver = server.event_stream(); let frontend_clone = frontend.clone(); let stop_child_token = self.stop_token.child_token(); + let emit_task_events = options.emit_task_events(); tokio::spawn(async move { - frontend_server_event_loop(event_receiver, frontend_clone, stop_child_token).await; + frontend_server_event_loop( + event_receiver, + frontend_clone, + stop_child_token, + emit_task_events, + ) + .await; }); } diff --git a/crates/intiface_engine/src/frontend/mod.rs b/crates/intiface_engine/src/frontend/mod.rs index 8f4dd416b..45f45a415 100644 --- a/crates/intiface_engine/src/frontend/mod.rs +++ b/crates/intiface_engine/src/frontend/mod.rs @@ -9,8 +9,9 @@ pub mod process_messages; use crate::error::IntifaceError; use crate::remote_server::ButtplugRemoteServerEvent; use async_trait::async_trait; +use buttplug_core::util::task::{TaskEvent, registry}; use futures::{Stream, StreamExt, pin_mut}; -pub use process_messages::{EngineMessage, IntifaceMessage}; +pub use process_messages::{EngineMessage, IntifaceMessage, TaskListEntry}; use std::sync::Arc; use tokio::{ select, @@ -50,7 +51,19 @@ pub async fn frontend_external_event_loop( connection_cancellation_token.cancel(); info!("Got external stop request"); break; - } + }, + IntifaceMessage::RequestTaskList {} => { + let tasks = registry() + .snapshot() + .into_iter() + .map(|t| TaskListEntry { + id: t.id.value(), + path: t.path, + detached: t.detached, + }) + .collect(); + frontend.send(EngineMessage::TaskList { tasks }).await; + }, }, Err(_) => { info!("Frontend sender dropped, assuming connection lost, breaking."); @@ -70,9 +83,12 @@ pub async fn frontend_server_event_loop( receiver: impl Stream, frontend: Arc, connection_cancellation_token: CancellationToken, + emit_task_events: bool, ) { pin_mut!(receiver); + let mut task_events = emit_task_events.then(|| registry().event_stream()); + loop { select! { maybe_event = receiver.next() => { @@ -113,6 +129,31 @@ pub async fn frontend_server_event_loop( }, } }, + task_event = async { + match task_events.as_mut() { + Some(rx) => rx.recv().await, + None => std::future::pending().await, + } + } => { + match task_event { + Ok(TaskEvent::Started { id, path }) => { + frontend.send(EngineMessage::TaskStarted { id: id.value(), path }).await; + } + Ok(TaskEvent::Ended { id, path, outcome }) => { + frontend.send(EngineMessage::TaskEnded { + id: id.value(), + path, + outcome: format!("{outcome:?}"), + }).await; + } + Err(broadcast::error::RecvError::Lagged(n)) => { + warn!("Task event stream lagged, dropped {} events", n); + } + Err(broadcast::error::RecvError::Closed) => { + // Global registry sender never closes; unreachable in practice. + } + } + }, _ = connection_cancellation_token.cancelled() => { info!("Connection cancellation token activated, breaking from frontend server event loop"); break; diff --git a/crates/intiface_engine/src/frontend/process_messages.rs b/crates/intiface_engine/src/frontend/process_messages.rs index cc9abbacb..2ffc5652d 100644 --- a/crates/intiface_engine/src/frontend/process_messages.rs +++ b/crates/intiface_engine/src/frontend/process_messages.rs @@ -8,6 +8,13 @@ use buttplug_server_device_config::UserDeviceIdentifier; use serde::{Deserialize, Serialize}; +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TaskListEntry { + pub id: u64, + pub path: String, + pub detached: bool, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(tag = "code", rename_all = "snake_case")] pub enum EngineErrorDetail { @@ -62,19 +69,50 @@ pub enum EngineMessage { output_type: String, value: f64, }, + TaskStarted { + id: u64, + path: String, + }, + TaskEnded { + id: u64, + path: String, + /// How the task ended: "Completed" | "Cancelled" | "Panicked" + /// (the Debug rendering of `buttplug_core`'s `TaskOutcome`). + outcome: String, + }, + TaskList { + tasks: Vec, + }, } #[derive(Debug, Clone, Serialize, Deserialize)] pub enum IntifaceMessage { RequestEngineVersion { expected_version: u32 }, Stop {}, + RequestTaskList {}, } #[cfg(test)] mod test { - use super::{EngineErrorDetail, EngineMessage}; + use super::{EngineErrorDetail, EngineMessage, IntifaceMessage}; use serde_json::json; + #[test] + fn test_task_message_serialization() { + let msg = EngineMessage::TaskEnded { + id: 42, + path: "server-1/ping-timer/timer".to_owned(), + outcome: "Cancelled".to_owned(), + }; + let json = serde_json::to_string(&msg).unwrap(); + let back: EngineMessage = serde_json::from_str(&json).unwrap(); + assert!(matches!(back, EngineMessage::TaskEnded { id: 42, .. })); + + let req = r#"{"RequestTaskList":{}}"#; + let parsed: IntifaceMessage = serde_json::from_str(req).unwrap(); + assert!(matches!(parsed, IntifaceMessage::RequestTaskList {})); + } + #[test] fn generic_engine_error_serializes_without_structured_fields() { let message = EngineMessage::EngineError { diff --git a/crates/intiface_engine/src/options.rs b/crates/intiface_engine/src/options.rs index f76e1adaa..ca8fa663a 100644 --- a/crates/intiface_engine/src/options.rs +++ b/crates/intiface_engine/src/options.rs @@ -67,6 +67,8 @@ pub struct EngineOptions { rest_api_port: Option, #[getset(get_copy = "pub")] emit_output_observations: bool, + #[getset(get_copy = "pub")] + emit_task_events: bool, } #[derive(Default, Debug, Clone)] @@ -100,6 +102,7 @@ pub struct EngineOptionsExternal { pub repeater_remote_address: Option, pub rest_api_port: Option, pub emit_output_observations: bool, + pub emit_task_events: bool, } impl From for EngineOptions { @@ -134,6 +137,7 @@ impl From for EngineOptions { repeater_remote_address: other.repeater_remote_address, rest_api_port: other.rest_api_port, emit_output_observations: other.emit_output_observations, + emit_task_events: other.emit_task_events, } } } @@ -297,6 +301,11 @@ impl EngineOptionsBuilder { self } + pub fn emit_task_events(&mut self, value: bool) -> &mut Self { + self.options.emit_task_events = value; + self + } + pub fn finish(&mut self) -> EngineOptions { self.options.clone() }