Cancel pending requests when a device leave or rejoins

crates/ziggurat-driver/src/zigbee_stack/joining.rs

@@ -1122,6 +1122,18 @@ impl<P: RadioPhy, R: Runtime> ZigbeeStack<P, R> {
             return;
         };
 
+        // Spec 4.7.3.6: an unsecured rejoin re-delivers the network key encrypted
+        // with the well-known key unless the device holds a unique link key. Reject
+        // when it has neither, so the network key is never exposed to anyone who
+        // knows the well-known key. Nobody authenticated this frame, so it is rejected
+        // before it changes any state.
+        if !secured && !self.may_rejoin_unsecured(source_ieee) {
+            tracing::warn!(
+                "Rejecting unsecured rejoin from {source_ieee:?} without a unique link key"
+            );
+            return;
+        }
+
         let requested_nwk = nwk_frame.nwk_header.source;
         let capability = &rejoin_request.capability_information;
 
@@ -1211,16 +1223,6 @@ impl<P: RadioPhy, R: Runtime> ZigbeeStack<P, R> {
 
         // A trust center rejoin means the device no longer has the network key
         if !secured {
-            // Spec 4.7.3.6: an unsecured rejoin re-delivers the network key encrypted
-            // with the well-known key unless the device holds a unique link key. Reject
-            // when it has neither, so the network key is never exposed to anyone who
-            // knows the well-known key.
-            if !self.may_rejoin_unsecured(source_ieee) {
-                tracing::warn!(
-                    "Rejecting unsecured rejoin from {source_ieee:?} without a unique link key"
-                );
-                return;
-            }
             // The join is announced only once the key transport lands
             let stack = self
                 .self_weak
@@ -1305,7 +1307,7 @@ impl<P: RadioPhy, R: Runtime> ZigbeeStack<P, R> {
         // The address map entry and any negotiated link key are kept around so that the
         // device can rejoin later
         self.drop_indirect_transactions(source_ieee, source);
-        self.core().nib.routing.remove_route(source);
+        self.forget_nwk_state(source);
 
         self.push_notification(ZigbeeNotification::DeviceLeft {
             nwk: source,

crates/ziggurat-driver/src/zigbee_stack/nwk.rs

@@ -32,12 +32,33 @@ use super::{
 };
 
 impl<P: RadioPhy, R: Runtime> ZigbeeStack<P, R> {
-    pub fn update_nwk_eui64_mapping(&self, nwk: Nwk, eui64: Eui64) {
-        let conflict = self.core().nib.address_map.update_mapping(eui64, nwk);
+    /// A device joined or rejoined as `nwk`: the address starts a new session. The
+    /// mapping and the state it replaces change under one lock, so no send can be
+    /// enqueued against the old session in between.
+    pub(super) fn update_nwk_eui64_mapping(&self, nwk: Nwk, eui64: Eui64) {
+        let mut core = self.core();
 
-        if conflict {
+        // `nwk` belongs to another device, whose state must stay
+        if core.nib.address_map.update_mapping(eui64, nwk) {
+            drop(core);
             self.handle_address_conflict(nwk, AddrConflictSource::Local);
+            return;
         }
+
+        // The device can be reached again, so sends held for it are kept. The route
+        // request they wait on went out before it joined. With that discovery gone, the
+        // timer pass starts a new one.
+        core.nib.routing.forget(self.state.network_address, nwk);
+        self.commit_sends(core, SendActions::new());
+    }
+
+    /// The device at `nwk` left, so what we know about the address belongs to a session
+    /// that is over. Held sends fail: nothing will answer a route request for it.
+    pub(super) fn forget_nwk_state(&self, nwk: Nwk) {
+        let mut core = self.core();
+        core.nib.routing.forget(self.state.network_address, nwk);
+        let actions = core.sends.fail_unrouted(nwk);
+        self.commit_sends(core, actions);
     }
 
     /// Filter broadcast frames based on the NWK broadcast transaction table
@@ -89,8 +110,8 @@ impl<P: RadioPhy, R: Runtime> ZigbeeStack<P, R> {
         nwk_frame: NwkFrame,
         attempts: u8,
         initial_delay: Duration,
-    ) -> SendActions {
-        let (_, actions) = self.enqueue_broadcast_in(
+    ) -> (SendId, SendActions) {
+        let (handle, actions) = self.enqueue_broadcast_in(
             core,
             Frame::nwk(nwk_frame, NwkSecurityMode::NetworkKey),
             TxPolicy {
@@ -104,7 +125,7 @@ impl<P: RadioPhy, R: Runtime> ZigbeeStack<P, R> {
             },
         );
 
-        actions
+        (handle.id(), actions)
     }
 
     /// A random retransmission jitter in `[0, max_broadcast_jitter)` (spec 3.6.6).
@@ -153,9 +174,19 @@ impl<P: RadioPhy, R: Runtime> ZigbeeStack<P, R> {
             );
         }
 
-        // Update the address cache
+        // Update the address cache. Hearing a device is not a join, so the address keeps
+        // its session state.
         if let Some(src_eui64) = nwk_frame.nwk_header.source_ieee {
-            self.update_nwk_eui64_mapping(nwk_frame.nwk_header.source, src_eui64);
+            let source = nwk_frame.nwk_header.source;
+            let conflict = self
+                .core()
+                .nib
+                .address_map
+                .update_mapping(src_eui64, source);
+
+            if conflict {
+                self.handle_address_conflict(source, AddrConflictSource::Local);
+            }
         }
 
         // Spec 3.6.1.10.2: a frame addressed to our network address but a different

crates/ziggurat-driver/src/zigbee_stack/route.rs

@@ -112,6 +112,13 @@ impl<P: RadioPhy, R: Runtime> ZigbeeStack<P, R> {
         let network_address = self.state.network_address;
         let many_to_one = route_request_cmd.many_to_one != NwkRouteRequestManyToOne::NotManyToOne;
 
+        // Neighbors relay our own route requests back to us. Our discovery entry may
+        // already be gone, so it cannot be what filters them.
+        if nwk_frame.nwk_header.source == network_address {
+            tracing::debug!("Ignoring our own route request relayed by {sender_nwk:?}");
+            return;
+        }
+
         // We need to know who sent the frame
         let sender_link = self.core().nib.neighbors.link(sender_nwk);
 
@@ -251,7 +258,7 @@ impl<P: RadioPhy, R: Runtime> ZigbeeStack<P, R> {
 
         let attempts = self.tunables.rreq_retries() + 1;
         let mut core = self.core();
-        let actions =
+        let (_, actions) =
             self.broadcast_route_request(&mut core, relayed_route_request_cmd, attempts, jitter);
         self.commit_sends(core, actions);
     }
@@ -474,11 +481,14 @@ impl<P: RadioPhy, R: Runtime> ZigbeeStack<P, R> {
 
         // Spec 3.6.4.5.1: the initial broadcast is repeated `nwkcInitialRREQRetries`
         // times, separated by the retry interval
-        self.broadcast_route_request(
+        let (request, mut actions) = self.broadcast_route_request(
             core,
             route_request_frame,
             self.tunables.initial_rreq_retries() + 1,
             Duration::ZERO,
-        )
+        );
+        actions.extend(core.sends.route_request_sent(destination, request));
+
+        actions
     }
 }

crates/ziggurat-zigbee/src/nwk/routing.rs

@@ -284,6 +284,22 @@ impl Routing {
         self.route_table.remove(&destination).is_some()
     }
 
+    /// The device at `nwk` left or joined again, clear associated state.
+    pub fn forget(&mut self, own_address: Nwk, nwk: Nwk) {
+        // TODO: a discovery toward `nwk` that we relay for another router loses its
+        // route entry here, so `accept_route_reply` drops the reply as coming from an
+        // unknown responder. That router gets no route through us for that discovery.
+        self.route_table.remove(&nwk);
+        self.route_record_table.remove(&nwk);
+
+        // Only our own discoveries toward it are dropped. Discoveries we relay stay
+        // until they expire: they are the only duplicate check for late copies of
+        // their route requests.
+        self.discovery_table.retain(|_, entry| {
+            entry.source_address != own_address || entry.destination_address != nwk
+        });
+    }
+
     /// Seed an active routing entry restored by the client on startup. Unlike a route
     /// learned at runtime, this does not record a change (the client already has it).
     pub fn restore_route(&mut self, destination: Nwk, next_hop: Nwk, path_cost: u8) {

crates/ziggurat-zigbee/src/sends/broadcasts.rs

@@ -306,6 +306,34 @@ impl<F, W: Clone, E: From<SendFailure>> Broadcasts<F, W, E> {
         Some(actions)
     }
 
+    /// Send no further copies of a broadcast nobody waits on any more. A copy in the
+    /// radio right now cannot be recalled: its schedule is emptied instead, so the
+    /// record settles when it next comes due (a `Once` schedule settles on the copy's
+    /// own report).
+    pub(super) fn stop(&mut self, id: SendId) -> Actions<F, W, E> {
+        let mut actions = Actions::new();
+
+        let Some(record) = self.records.get_mut(&id) else {
+            return actions;
+        };
+
+        if record.copy == CopyState::Transmitting {
+            match &mut record.schedule {
+                Schedule::Once => {}
+                Schedule::PassiveAck {
+                    copies_remaining, ..
+                }
+                | Schedule::FixedInterval {
+                    copies_remaining, ..
+                } => *copies_remaining = 0,
+            }
+        } else {
+            self.settle(id, Err(SendFailure::Cancelled.into()), &mut actions);
+        }
+
+        actions
+    }
+
     /// Queue copies that are due, complete or fail schedules that ran out, and fail
     /// records past their deadline.
     pub(super) fn tick(&mut self, now: Instant) -> Actions<F, W, E> {

crates/ziggurat-zigbee/src/sends/mod.rs

@@ -416,10 +416,28 @@ impl<F, W: Clone, E: From<SendFailure>> Sends<F, W, E> {
         self.unrouted_unicasts.retry_discovery(destination)
     }
 
+    /// The driver broadcast `request`, the route request of `destination`'s discovery.
+    /// Copies of the request it replaces still to come are not sent.
+    pub fn route_request_sent(&mut self, destination: Nwk, request: SendId) -> Actions<F, W, E> {
+        match self
+            .unrouted_unicasts
+            .route_request_sent(destination, request)
+        {
+            Some(previous) => self.broadcasts.stop(previous),
+            None => Actions::new(),
+        }
+    }
+
     /// Fail every unrouted send to `destination`: its discovery is over without a
-    /// route.
+    /// route. Copies of its route request still to come are not sent.
     pub fn fail_unrouted(&mut self, destination: Nwk) -> Actions<F, W, E> {
-        self.unrouted_unicasts.fail(destination)
+        let (request, mut actions) = self.unrouted_unicasts.fail(destination);
+
+        if let Some(request) = request {
+            actions.extend(self.broadcasts.stop(request));
+        }
+
+        actions
     }
 
     /// A route to `destination` is known: move every unrouted send to it to the unicasts,