Implement request cancellation

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<P: RadioPhy, R: Runtime> ZigbeeStack<P, R> {
         }
     }
 
+    /// 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<SendRequest> = 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

crates/ziggurat-ncp-api/src/protocol.rs

@@ -148,6 +148,10 @@ async fn dispatch<P: RadioPhy>(
             proto::set_tunable(&**configured(app)?, &payload)?;
             Ok(Response::Empty)
         }
+        Request::CancelRequest(payload) => Ok(Response::CancelResult(proto::cancel_request(
+            &**running(app)?,
+            &payload,
+        ))),
     }
 }
 

crates/ziggurat-protocol/src/bridge.rs

@@ -272,6 +272,16 @@ pub fn send_aps<P: RadioPhy, R: Runtime>(
         .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<P: RadioPhy, R: Runtime>(
+    stack: &ZigbeeStack<P, R>,
+    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<P: RadioPhy, R: Runtime>(

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<u8>,
 }
 
+/// 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")

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,
+            ))),
         }
     }