From ace5d45d3abbb0fa9451f5d90f22da66f9ad2309 Mon Sep 17 00:00:00 2001 From: Kyle Machulis Date: Wed, 10 Jun 2026 13:55:57 -0700 Subject: [PATCH 1/6] fix(server): stop commands flush batches and resolve on hardware write Stop commands previously fire-and-forgot into the device io channel, where message_gap batching could leave the stop write sitting in pending_commands until the batch deadline. Since shutdown order is stop then disconnect, the disconnect routinely beat the deadline and dropped the pending stop write unflushed, leaving the device running. The io-channel payload is now DeviceTaskMessage, carrying the commands plus an optional oneshot write-acknowledgement sender. A message with an ack is urgent: the io task merges it into any pending batch (existing dedupe), flushes everything to hardware immediately regardless of the batch deadline, then fires the ack. Messages without an ack keep the exact prior batching behaviour, so normal output is unchanged. The stop path accumulates the hardware commands from every per-feature stop OutputCmd into a single write-acknowledged batch and awaits the ack, so stop() (and therefore stop_devices()/shutdown) resolves only once the stop write has reached hardware. The wait is bounded by a runtime-agnostic 1s timeout (select! against async_manager::sleep, not tokio::time, for WASM parity); a wedged or dead device resolves Ok rather than hanging shutdown. The io task's token-cancel and channel-close exits now best-effort flush pending_commands first so a stop in the batch window still lands; the hardware-Disconnected exit deliberately does not flush, since the hardware is gone. Co-Authored-By: Claude Fable 5 --- .../src/device/device_handle.rs | 139 ++++++++++++++---- .../buttplug_server/src/device/device_task.rs | 104 ++++++++++--- 2 files changed, 194 insertions(+), 49 deletions(-) diff --git a/crates/buttplug_server/src/device/device_handle.rs b/crates/buttplug_server/src/device/device_handle.rs index 759dc4824..16b139cfb 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, task::TaskScope}, + 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,7 +112,7 @@ 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. @@ -127,7 +130,7 @@ impl DeviceHandle { 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 { @@ -272,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 @@ -293,37 +317,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)| { @@ -536,7 +617,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)) diff --git a/crates/buttplug_server/src/device/device_task.rs b/crates/buttplug_server/src/device/device_task.rs index 8664ba35d..15e2fb1cf 100644 --- a/crates/buttplug_server/src/device/device_task.rs +++ b/crates/buttplug_server/src/device/device_task.rs @@ -34,6 +34,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: @@ -48,7 +57,7 @@ pub fn spawn_device_task( hardware: Arc, _handler: Arc, config: DeviceTaskConfig, - mut command_receiver: Receiver>, + mut command_receiver: Receiver, ) { task_scope.spawn("io", move |token| async move { run_device_task(hardware, config, &mut command_receiver, token).await; @@ -62,7 +71,7 @@ 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(); @@ -93,6 +102,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 { @@ -121,17 +147,54 @@ async fn run_device_task( // 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) @@ -147,28 +210,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; } @@ -208,6 +270,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; } } From f10dbe166dbfadd5ab0bd782f7560db7de5935f8 Mon Sep 17 00:00:00 2001 From: Kyle Machulis Date: Wed, 10 Jun 2026 14:07:54 -0700 Subject: [PATCH 2/6] test: stop commands verified written before stop/shutdown resolve MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two RED-verified lifecycle tests pinning the stop-command write-ack contract, plus the minimal test-util change they need. Test util: TestDevice now carries a configurable message_gap (default 1ms, unchanged for all existing tests). add_test_device_with_message_gap and test_server_with_device_and_message_gap let a test widen the device io task's batch window so the stop-write race is deterministic rather than relying on the 1ms default. test_stop_resolves_only_after_stop_write_reaches_hardware (Test A): with a 500ms message gap and an active vibrate, a StopCmd must not resolve until the resulting stop write ([0xF1, 0x00]) has reached the hardware channel. Pre-fix this fails deterministically — StopCmd fire-and-forgets and resolves while the write sits in the batch window. test_shutdown_writes_stop_before_resolving (Test B): the original incident. With the same batched device in an active output state, server.shutdown() must record the device's stop write before it resolves. Pre-fix the disconnect tore the io task down mid-batch and dropped the pending stop write. Both retire the write-observation limitation documented on test_shutdown_under_load_drains_subtree (1ms harness gap made it flaky). RED verified by reverting the two production files to HEAD~1 (2 failed / 0 passed across three runs); restored exactly (empty production diff); GREEN and stable across five runs. Full buttplug_tests suite green incl. 808 conformance. Co-Authored-By: Claude Fable 5 --- .../tests/test_task_lifecycle.rs | 174 +++++++++++++++++- crates/buttplug_tests/tests/util/mod.rs | 16 ++ .../util/test_device_manager/test_device.rs | 28 ++- .../test_device_comm_manager.rs | 33 +++- 4 files changed, 239 insertions(+), 12 deletions(-) diff --git a/crates/buttplug_tests/tests/test_task_lifecycle.rs b/crates/buttplug_tests/tests/test_task_lifecycle.rs index 4425c4443..86d296f62 100644 --- a/crates/buttplug_tests/tests/test_task_lifecycle.rs +++ b/crates/buttplug_tests/tests/test_task_lifecycle.rs @@ -17,14 +17,78 @@ use buttplug_core::{ OutputValue, RequestServerInfoV4, StartScanningV0, + StopCmdV4, }, util::task::registry, }; -use buttplug_server::message::ButtplugClientMessageVariant; +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}; +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. @@ -296,3 +360,109 @@ async fn test_shutdown_resolves_with_stalled_bringup() { .expect("shutdown hung with a stalled device bringup — bringup is not honoring its cancellation token") .expect("server shutdown errored"); } + +/// 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 9f4e07e8c..a82460353 100644 --- a/crates/buttplug_tests/tests/util/mod.rs +++ b/crates/buttplug_tests/tests/util/mod.rs @@ -170,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/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(), From e7a431f55dd6cc2e366978bb2499864de83a9fbc Mon Sep 17 00:00:00 2001 From: Kyle Machulis Date: Wed, 10 Jun 2026 14:18:37 -0700 Subject: [PATCH 3/6] docs: clean stale comments flagged in stop-ack review Co-Authored-By: Claude Fable 5 --- crates/buttplug_server/src/device/device_handle.rs | 1 - crates/buttplug_tests/tests/test_task_lifecycle.rs | 7 +++++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/crates/buttplug_server/src/device/device_handle.rs b/crates/buttplug_server/src/device/device_handle.rs index 16b139cfb..d994ae60b 100644 --- a/crates/buttplug_server/src/device/device_handle.rs +++ b/crates/buttplug_server/src/device/device_handle.rs @@ -308,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(), diff --git a/crates/buttplug_tests/tests/test_task_lifecycle.rs b/crates/buttplug_tests/tests/test_task_lifecycle.rs index 86d296f62..251e17df2 100644 --- a/crates/buttplug_tests/tests/test_task_lifecycle.rs +++ b/crates/buttplug_tests/tests/test_task_lifecycle.rs @@ -229,6 +229,13 @@ async fn test_server_shutdown_leaves_no_tasks() { /// 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. From 509be614f786ab4e803db1ca45ef03ed7e42acb9 Mon Sep 17 00:00:00 2001 From: Kyle Machulis Date: Sun, 19 Jul 2026 20:08:54 -0700 Subject: [PATCH 4/6] fix(core): harden task scope lifecycle --- crates/buttplug_core/src/util/task/mod.rs | 10 +- .../buttplug_core/src/util/task/registry.rs | 54 ++++ crates/buttplug_core/src/util/task/scope.rs | 288 ++++++++++++++++-- 3 files changed, 323 insertions(+), 29 deletions(-) diff --git a/crates/buttplug_core/src/util/task/mod.rs b/crates/buttplug_core/src/util/task/mod.rs index 0391a9f0e..847a917cc 100644 --- a/crates/buttplug_core/src/util/task/mod.rs +++ b/crates/buttplug_core/src/util/task/mod.rs @@ -17,7 +17,15 @@ mod registry; mod scope; -pub use registry::{TaskEvent, TaskId, TaskInfo, TaskOutcome, TaskRegistry, registry}; +pub use registry::{ + TaskEvent, + TaskId, + TaskInfo, + TaskOutcome, + TaskRegistry, + TaskSpawnError, + registry, +}; pub use scope::TaskScope; use crate::util::async_manager; diff --git a/crates/buttplug_core/src/util/task/registry.rs b/crates/buttplug_core/src/util/task/registry.rs index 97f986e28..361cce292 100644 --- a/crates/buttplug_core/src/util/task/registry.rs +++ b/crates/buttplug_core/src/util/task/registry.rs @@ -7,10 +7,20 @@ 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. /// @@ -71,6 +81,8 @@ pub enum TaskEvent { #[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, @@ -86,6 +98,7 @@ 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, @@ -98,6 +111,47 @@ impl TaskRegistry { } 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, diff --git a/crates/buttplug_core/src/util/task/scope.rs b/crates/buttplug_core/src/util/task/scope.rs index 035a2518b..1027781b4 100644 --- a/crates/buttplug_core/src/util/task/scope.rs +++ b/crates/buttplug_core/src/util/task/scope.rs @@ -5,7 +5,7 @@ // Licensed under the BSD 3-Clause license. See LICENSE file in the project root // for full license information. -use super::registry::{TaskId, TaskOutcome, registry}; +use super::registry::{TaskId, TaskOutcome, TaskSpawnError, registry}; use crate::util::async_manager; use std::future::Future; use tokio_util::sync::CancellationToken; @@ -56,40 +56,44 @@ impl TaskScope { /// Request cancellation of every task in this scope's subtree. pub fn cancel(&self) { - self.token.cancel(); + 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) { - self.token.cancel(); + 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) + 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 (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, id, task_token), span); + 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) + 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 (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, id, task_token), span); + async_manager::spawn(finish_task(fut, guard), span); + Ok(()) } /// Consume the scope and spawn a task that holds it alive for its own @@ -97,47 +101,53 @@ impl TaskScope { /// 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) + 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 (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, id, task_token).await; + 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) + 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 (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, id, task_token).await; + finish_task(fut, guard).await; }, span, ); + Ok(()) } - fn register_task(&self, name: &str) -> (TaskId, CancellationToken, tracing::Span) { + fn register_task( + &self, + name: &str, + ) -> Result<(TaskId, CancellationToken, tracing::Span), TaskSpawnError> { let path = format!("{}/{}", self.path, name); - let id = registry().register(path.clone(), false); - let task_token = self.token.child_token(); + 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); - (id, task_token, span) + Ok((id, task_token, span)) } } @@ -174,14 +184,13 @@ impl Drop for DeregisterGuard { } } -async fn finish_task(fut: impl Future, id: TaskId, token: CancellationToken) { - let _guard = DeregisterGuard::new(id, Some(token)); +async fn finish_task(fut: impl Future, _guard: DeregisterGuard) { fut.await; } impl Drop for TaskScope { fn drop(&mut self) { - self.token.cancel(); + registry().cancel(&self.token); } } @@ -189,6 +198,7 @@ impl Drop for TaskScope { mod test { use super::*; use crate::util::task::registry; + use crate::util::task::registry::TaskEvent; use std::time::Duration; #[test] @@ -206,12 +216,132 @@ mod test { 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(); - root.spawn("worker", |_token| async move { + let _ = root.spawn("worker", |_token| async move { let _ = tx.send(()); }); rx.await.unwrap(); @@ -225,7 +355,7 @@ mod test { let root = TaskScope::root("canceltest"); let path = root.path().to_owned(); let child = root.child("inner"); - child.spawn("worker", |token| async move { + let _ = child.spawn("worker", |token| async move { token.cancelled().await; }); root.cancel(); @@ -238,7 +368,7 @@ mod test { async fn test_drop_cancels() { let root = TaskScope::root("droptest"); let path = root.path().to_owned(); - root.spawn("worker", |token| async move { + let _ = root.spawn("worker", |token| async move { token.cancelled().await; }); drop(root); @@ -250,7 +380,7 @@ mod test { #[tokio::test] async fn test_shutdown_awaits_subtree() { let root = TaskScope::root("shutdowntest"); - root.spawn("worker", |token| async move { + let _ = root.spawn("worker", |token| async move { token.cancelled().await; // Simulate cleanup work after observing cancellation. tokio::time::sleep(Duration::from_millis(20)).await; @@ -268,7 +398,7 @@ mod test { // spawned panic. Without the guard, this wait would time out. let root = TaskScope::root("panictest"); let path = root.path().to_owned(); - root.spawn("panicker", |_token| async move { + 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)) @@ -276,6 +406,108 @@ mod test { .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; @@ -288,7 +520,7 @@ mod test { 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. - sub_scope.spawn_and_hold("worker", |_token| async move { + let _ = sub_scope.spawn_and_hold("worker", |_token| async move { tokio::time::sleep(Duration::from_millis(20)).await; let _ = tx.send(()); }); From e4705e6343f0d1fb886de8a454fc7c69a919b002 Mon Sep 17 00:00:00 2001 From: Kyle Machulis Date: Sun, 19 Jul 2026 20:09:06 -0700 Subject: [PATCH 5/6] fix(server): handle closed task scopes --- .../src/device/device_handle.rs | 84 +++++----- .../buttplug_server/src/device/device_task.rs | 17 +- .../src/device/protocol_impl/kgoal_boost.rs | 146 ++++++++++-------- .../src/device/server_device_manager.rs | 36 +++-- .../server_device_manager_event_loop.rs | 10 +- crates/buttplug_server/src/lib.rs | 3 + crates/buttplug_server/src/ping_timer.rs | 12 +- crates/buttplug_server/src/server_builder.rs | 11 +- .../tests/test_task_lifecycle.rs | 56 +++++++ .../stalling_device_communication_manager.rs | 30 +++- 10 files changed, 267 insertions(+), 138 deletions(-) diff --git a/crates/buttplug_server/src/device/device_handle.rs b/crates/buttplug_server/src/device/device_handle.rs index d994ae60b..50a9f86ce 100644 --- a/crates/buttplug_server/src/device/device_handle.rs +++ b/crates/buttplug_server/src/device/device_handle.rs @@ -634,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![]; @@ -717,51 +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(); - 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; - } - 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; + 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; + } + 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 - ); + 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 15e2fb1cf..014bf351a 100644 --- a/crates/buttplug_server/src/device/device_task.rs +++ b/crates/buttplug_server/src/device/device_task.rs @@ -14,7 +14,10 @@ use std::{collections::VecDeque, sync::Arc, time::Duration}; -use buttplug_core::util::{async_manager, task::TaskScope}; +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; @@ -58,10 +61,14 @@ pub fn spawn_device_task( _handler: Arc, config: DeviceTaskConfig, mut command_receiver: Receiver, -) { - task_scope.spawn("io", move |token| async move { - run_device_task(hardware, config, &mut command_receiver, token).await; - }); +) -> 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). 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 c21f1a796..b2fbb1a4f 100644 --- a/crates/buttplug_server/src/device/protocol_impl/kgoal_boost.rs +++ b/crates/buttplug_server/src/device/protocol_impl/kgoal_boost.rs @@ -91,79 +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. - 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 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 ddb36fbf6..437f1cffd 100644 --- a/crates/buttplug_server/src/device/server_device_manager.rs +++ b/crates/buttplug_server/src/device/server_device_manager.rs @@ -190,20 +190,28 @@ impl ServerDeviceManagerBuilder { 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; - }); + 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, 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 0635bffbc..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 @@ -330,9 +330,11 @@ impl ServerDeviceManagerEventLoop { // 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. - self.devices_scope.spawn( + 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() => { @@ -371,6 +373,12 @@ impl ServerDeviceManagerEventLoop { connecting_devices.remove(&address); }, ); + 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 2fece8dba..cc1a1c075 100644 --- a/crates/buttplug_server/src/ping_timer.rs +++ b/crates/buttplug_server/src/ping_timer.rs @@ -76,7 +76,11 @@ 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, task_scope: TaskScope) -> Self + pub fn new( + max_ping_time: u32, + on_ping_timeout: Option, + task_scope: TaskScope, + ) -> Result where F: FnOnce() + Send + 'static, { @@ -85,13 +89,13 @@ impl PingTimer { let callback = Arc::new(Mutex::new(on_ping_timeout)); 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_builder.rs b/crates/buttplug_server/src/server_builder.rs index 5763f384e..8ee71b313 100644 --- a/crates/buttplug_server/src/server_builder.rs +++ b/crates/buttplug_server/src/server_builder.rs @@ -120,14 +120,19 @@ impl ButtplugServerBuilder { // 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. - ping_timeout_scope.spawn_and_hold("stop-devices", move |_token| async move { + 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( @@ -146,7 +151,7 @@ impl ButtplugServerBuilder { ping_time, ping_timeout_callback, task_scope.child("ping-timer"), - )); + )?); // Assuming everything passed, return the server. Ok(ButtplugServer::new( diff --git a/crates/buttplug_tests/tests/test_task_lifecycle.rs b/crates/buttplug_tests/tests/test_task_lifecycle.rs index 251e17df2..d3b7c7d93 100644 --- a/crates/buttplug_tests/tests/test_task_lifecycle.rs +++ b/crates/buttplug_tests/tests/test_task_lifecycle.rs @@ -368,6 +368,62 @@ async fn test_shutdown_resolves_with_stalled_bringup() { .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. diff --git a/crates/buttplug_tests/tests/util/stalling_device_communication_manager.rs b/crates/buttplug_tests/tests/util/stalling_device_communication_manager.rs index 7b21868c4..0d3540302 100644 --- a/crates/buttplug_tests/tests/util/stalling_device_communication_manager.rs +++ b/crates/buttplug_tests/tests/util/stalling_device_communication_manager.rs @@ -30,12 +30,19 @@ use buttplug_server_device_config::{BluetoothLESpecifier, ProtocolCommunicationS use futures::FutureExt; use log::error; use std::collections::HashMap; -use tokio::sync::mpsc::Sender; +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] @@ -47,13 +54,22 @@ impl HardwareConnector for StallingHardwareConnector { 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(Default)] -pub struct StallingDeviceCommunicationManagerBuilder {} +#[derive(Clone, Default)] +pub struct StallingDeviceCommunicationManagerBuilder { + state: StallingDeviceState, +} + +impl StallingDeviceCommunicationManagerBuilder { + pub fn state(&self) -> StallingDeviceState { + self.state.clone() + } +} impl HardwareCommunicationManagerBuilder for StallingDeviceCommunicationManagerBuilder { fn finish( @@ -62,12 +78,14 @@ impl HardwareCommunicationManagerBuilder for StallingDeviceCommunicationManagerB ) -> Box { Box::new(StallingDeviceCommunicationManager { device_sender: sender, + state: self.state.clone(), }) } } pub struct StallingDeviceCommunicationManager { device_sender: Sender, + state: StallingDeviceState, } impl HardwareCommunicationManager for StallingDeviceCommunicationManager { @@ -77,13 +95,17 @@ impl HardwareCommunicationManager for 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 }; + let connector = StallingHardwareConnector { + specifier, + state: state.clone(), + }; if device_sender .send(HardwareCommunicationManagerEvent::DeviceFound { name: "Massage Demo".to_owned(), From 2696330a0f9357ab2e23017da4be5f13c263edaf Mon Sep 17 00:00:00 2001 From: Kyle Machulis Date: Sun, 19 Jul 2026 20:09:16 -0700 Subject: [PATCH 6/6] chore(release): prepare Intiface Engine 5.0.0 --- crates/buttplug_server/CHANGELOG.md | 7 +++++++ crates/intiface_engine/CHANGELOG.md | 16 ++++++++++++++++ crates/intiface_engine/Cargo.toml | 2 +- 3 files changed, 24 insertions(+), 1 deletion(-) 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/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"