diff --git a/crates/ziggurat-driver/src/zigbee_stack/nwk.rs b/crates/ziggurat-driver/src/zigbee_stack/nwk.rs index bfedec9..87653b9 100644 --- a/crates/ziggurat-driver/src/zigbee_stack/nwk.rs +++ b/crates/ziggurat-driver/src/zigbee_stack/nwk.rs @@ -6,6 +6,7 @@ use crate::ziggurat_ieee_802154::{ Ieee802154FrameControl, Ieee802154FrameHeader, Ieee802154FrameType, }; use alloc::boxed::Box; +use alloc::collections::BinaryHeap; use alloc::string::ToString; use alloc::vec::Vec; use core::sync::atomic::Ordering as AtomicOrdering; @@ -1425,6 +1426,108 @@ impl ZigbeeStack { } } + /// Best-effort cancellation of an application send by the `request_id` it was + /// issued under. Tears the send out of whichever pre-delivery queue still holds + /// it. + pub fn cancel_send(&self, request_id: RequestId) -> bool { + self.cancel_queued_send(request_id) + || self.cancel_pending_route(request_id) + || self.cancel_pending_unicast_retry(request_id) + || self.cancel_pending_broadcast(request_id) + } + + /// Whether a transmit outcome is the confirmation of `request_id`. + const fn confirms_request(outcome: &TxOutcome, request_id: RequestId) -> bool { + matches!(outcome, TxOutcome::Confirm { request_id: rid, .. } if *rid == request_id) + } + + /// Remove a send still waiting in the sender priority queue (not yet transmitted). + fn cancel_queued_send(&self, request_id: RequestId) -> bool { + let mut queue = self.send_queue.lock(); + if !queue + .iter() + .any(|send| Self::confirms_request(&send.outcome, request_id)) + { + return false; + } + + // The heap has no keyed removal, so drain, drop the one match (freeing its + // token), and rebuild from the rest. Cancellation is rare, so the rebuild is + // fine. + let mut removed = false; + let kept: Vec = queue + .drain() + .filter(|send| { + let drop_it = !removed && Self::confirms_request(&send.outcome, request_id); + removed |= drop_it; + !drop_it + }) + .collect(); + *queue = BinaryHeap::from(kept); + true + } + + /// Remove a frame parked awaiting route discovery. When it was the last frame + /// waiting on that destination, drop the empty bucket so the reactor stops + /// inspecting it; the shared route request (critical-class, a fixed short burst) + /// and the routing table's `DiscoveryUnderway` entry are left to run out / expire + /// on their own. + fn cancel_pending_route(&self, request_id: RequestId) -> bool { + let mut pending = self.state.pending_routes.lock(); + + let mut emptied = None; + let mut found = false; + + for (destination, bucket) in pending.iter_mut() { + let before = bucket.frames.len(); + bucket + .frames + .retain(|frame| !Self::confirms_request(&frame.outcome, request_id)); + + if bucket.frames.len() != before { + found = true; + + if bucket.frames.is_empty() { + emptied = Some(*destination); + } + + break; + } + } + + if let Some(destination) = emptied { + pending.remove(&destination); + } + found + } + + /// Remove a unicast parked in the retry backoff before its next attempt is due. + fn cancel_pending_unicast_retry(&self, request_id: RequestId) -> bool { + let mut pending = self.state.pending_unicast_retries.lock(); + + pending + .iter() + .position(|retry| Self::confirms_request(&retry.outcome, request_id)) + .is_some_and(|index| { + pending.swap_remove(index); + true + }) + } + + /// Remove a data/group broadcast still being retransmitted, stopping its remaining + /// transmissions. Route-request broadcasts carry no `request_id` and are never matched. + fn cancel_pending_broadcast(&self, request_id: RequestId) -> bool { + let mut pending = self.state.pending_broadcasts.lock(); + + pending + .iter() + .find_map(|(key, broadcast)| (broadcast.request_id == Some(request_id)).then_some(*key)) + .is_some_and(|key| { + pending.remove(&key); + true + }) + } + /// A unicast exhausted its retries at the sender. The next hop is dead: invalidate /// routes through it. A frame we originated also drops any stored source route and /// pushes the MTORR scheduler; a frame we were relaying reports the failure back diff --git a/crates/ziggurat-ncp-api/src/protocol.rs b/crates/ziggurat-ncp-api/src/protocol.rs index 700797c..804afd3 100644 --- a/crates/ziggurat-ncp-api/src/protocol.rs +++ b/crates/ziggurat-ncp-api/src/protocol.rs @@ -148,6 +148,10 @@ async fn dispatch( proto::set_tunable(&**configured(app)?, &payload)?; Ok(Response::Empty) } + Request::CancelRequest(payload) => Ok(Response::CancelResult(proto::cancel_request( + &**running(app)?, + &payload, + ))), } } diff --git a/crates/ziggurat-protocol/src/bridge.rs b/crates/ziggurat-protocol/src/bridge.rs index 381074b..89f018e 100644 --- a/crates/ziggurat-protocol/src/bridge.rs +++ b/crates/ziggurat-protocol/src/bridge.rs @@ -272,6 +272,16 @@ pub fn send_aps( .map_err(|e| Error::new(Status::TransmitFailed, &e.to_string())) } +/// Cancel an in-flight send by the `request_id` it was issued under. Best-effort: the +/// reply reports whether a still-cancellable (pre-delivery) send was found and removed. +pub fn cancel_request( + stack: &ZigbeeStack, + payload: &CancelRequestPayload, +) -> CancelResultPayload { + let cancelled = stack.cancel_send(StackRequestId::from(payload.request_id)); + CancelResultPayload { cancelled } +} + /// Apply a `set_tunable` to the stack: the name is matched against the Rust field /// names of `Tunables`, the raw value decoded per the field's type. pub fn set_tunable( diff --git a/crates/ziggurat-protocol/src/wire.rs b/crates/ziggurat-protocol/src/wire.rs index 217f76f..98eda48 100644 --- a/crates/ziggurat-protocol/src/wire.rs +++ b/crates/ziggurat-protocol/src/wire.rs @@ -50,6 +50,7 @@ pub enum CommandId { PacketCapture = 0x27, PacketCaptureChannel = 0x28, SetTunable = 0x29, + CancelRequest = 0x2A, // More notifications. ReceivedAps = 0x30, SendConfirm = 0x31, @@ -343,6 +344,22 @@ pub struct SendApsPayload { pub asdu: Vec, } +/// Cancels an in-flight send by the `request_id` it was issued under. Best-effort: +/// the send is torn down only if it is still in a pre-delivery state (queued, awaiting +/// route discovery, or between retries). The reply reports whether anything was caught. +#[abstract_bits] +#[derive(Debug, Clone)] +pub struct CancelRequestPayload { + pub request_id: RequestId, +} + +#[abstract_bits] +#[derive(Debug, Clone)] +pub struct CancelResultPayload { + pub cancelled: bool, + pub reserved: u7, +} + #[abstract_bits] #[derive(Debug, Clone)] pub struct PermitJoinsPayload { @@ -571,6 +588,7 @@ pub enum Request { PacketCapture(ChannelPayload), PacketCaptureChannel(ChannelPayload), SetTunable(SetTunablePayload), + CancelRequest(CancelRequestPayload), } impl Request { @@ -605,6 +623,7 @@ impl Request { Self::PacketCaptureChannel(require(payload, "channel")?) } CommandId::SetTunable => Self::SetTunable(require(payload, "set_tunable")?), + CommandId::CancelRequest => Self::CancelRequest(require(payload, "cancel_request")?), // Everything else (the device -> host notification opcodes) is not a request. _ => return Err(Error::new(Status::UnknownCommand, "not a request")), }) @@ -618,6 +637,7 @@ pub enum Response { HwAddress(HwAddressPayload), NetworkInfo(NetworkInfoPayload), ScanCount(ScanCountPayload), + CancelResult(CancelResultPayload), } impl Response { @@ -630,6 +650,7 @@ impl Response { Self::HwAddress(payload) => append(&mut bytes, payload), Self::NetworkInfo(payload) => append(&mut bytes, payload), Self::ScanCount(payload) => append(&mut bytes, payload), + Self::CancelResult(payload) => append(&mut bytes, payload), }; if !fits { return Error::new(Status::InvalidRequest, "reply too large") diff --git a/crates/ziggurat-server/src/main.rs b/crates/ziggurat-server/src/main.rs index 7aca069..95cbf38 100644 --- a/crates/ziggurat-server/src/main.rs +++ b/crates/ziggurat-server/src/main.rs @@ -527,6 +527,10 @@ impl ZigguratServer { proto::set_tunable(&*self.configured()?, &payload)?; Ok(proto::Response::Empty) } + R::CancelRequest(payload) => Ok(proto::Response::CancelResult(proto::cancel_request( + &*self.running()?, + &payload, + ))), } }