bacnet_client/client/
routed_path_limits.rs

1use std::collections::HashMap;
2use std::sync::{Arc, Mutex as StdMutex, MutexGuard as StdMutexGuard};
3use std::time::{Duration, Instant as StdInstant};
4
5use bacnet_network::layer::ReceivedNetworkControl;
6use bacnet_npdu::{NetworkMessageType, RejectMessageReason, RejectMessageToNetwork};
7use tokio::sync::{Mutex as AsyncMutex, OwnedMutexGuard};
8use tokio::time::Instant as TokioInstant;
9
10use super::*;
11use crate::tsm::CompletionOutcome;
12
13/// Smallest routed NPDU envelope required across the standard data links.
14const CONSERVATIVE_ROUTED_PATH_MAX_NPDU: u16 = 228;
15/// Hard bound covering gates, configured evidence, and learned evidence.
16pub(super) const MAX_ROUTED_PATH_ENTRIES: usize = 256;
17
18/// Whether a router MAC or forwarded source MAC fits a routed path: 1 to
19/// [`MAX_MAC_LEN`](bacnet_npdu::MAX_MAC_LEN) octets. The NPDU codec carries no longer SADR
20/// or DADR (#1141), and the network layer drops a frame from a longer
21/// link-layer source (#1198), so a longer MAC could never complete a routed
22/// request (#1267).
23fn routed_mac_fits(length: usize) -> bool {
24    (1..=bacnet_npdu::MAX_MAC_LEN).contains(&length)
25}
26
27#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
28struct RoutedPathKey {
29    immediate_router_mac: MacAddr,
30    dnet: u16,
31}
32
33impl RoutedPathKey {
34    fn new(immediate_router_mac: &[u8], dnet: u16) -> Self {
35        Self {
36            immediate_router_mac: MacAddr::from_slice(immediate_router_mac),
37            dnet,
38        }
39    }
40}
41
42#[derive(Clone, Copy, Debug)]
43enum ConfiguredLimitProvenance {
44    ExplicitApi,
45}
46
47struct ConfiguredPathLimit {
48    max_npdu: u16,
49    provenance: ConfiguredLimitProvenance,
50    observed_at: StdInstant,
51}
52
53struct LearnedPathLimit {
54    exclusive_max_npdu: u16,
55    observed_at: StdInstant,
56}
57
58struct ActiveRoutedSend {
59    path: RoutedPathKey,
60    tsm_mac: MacAddr,
61    invoke_id: u8,
62    owner: TransactionOwner,
63    forwarded_npci_len: u16,
64    outgoing_apdu_len: Option<u16>,
65    ingress_floor: u64,
66}
67
68struct RoutedPathEntry {
69    gate: Arc<AsyncMutex<()>>,
70    configured: Option<ConfiguredPathLimit>,
71    learned: Option<LearnedPathLimit>,
72    active: Option<ActiveRoutedSend>,
73    last_used: u64,
74    quarantine_started: Option<TokioInstant>,
75    generation_attempts: u32,
76    generation_terminal_observed: bool,
77}
78
79impl RoutedPathEntry {
80    fn new(last_used: u64) -> Self {
81        Self {
82            gate: Arc::new(AsyncMutex::new(())),
83            configured: None,
84            learned: None,
85            active: None,
86            last_used,
87            quarantine_started: None,
88            generation_attempts: 0,
89            generation_terminal_observed: false,
90        }
91    }
92}
93
94struct RoutedPathState {
95    entries: HashMap<RoutedPathKey, RoutedPathEntry>,
96    next_use: u64,
97    capacity: usize,
98}
99
100pub(super) struct RoutedPathLimits {
101    state: StdMutex<RoutedPathState>,
102    quarantine_horizon: Duration,
103}
104
105impl RoutedPathLimits {
106    pub(super) fn new(quarantine_horizon: Duration) -> Self {
107        Self::with_capacity(MAX_ROUTED_PATH_ENTRIES, quarantine_horizon)
108    }
109
110    fn with_capacity(capacity: usize, quarantine_horizon: Duration) -> Self {
111        assert!(capacity > 0, "routed path capacity must be nonzero");
112        Self {
113            state: StdMutex::new(RoutedPathState {
114                entries: HashMap::new(),
115                next_use: 0,
116                capacity,
117            }),
118            quarantine_horizon,
119        }
120    }
121
122    fn state(&self) -> StdMutexGuard<'_, RoutedPathState> {
123        self.state
124            .lock()
125            .unwrap_or_else(std::sync::PoisonError::into_inner)
126    }
127
128    /// How many path entries exist, in use or idle.
129    #[cfg(test)]
130    pub(super) fn entry_count(&self) -> usize {
131        self.state().entries.len()
132    }
133
134    /// Reserve and lock the path through `router_mac` to `dnet`. A router MAC
135    /// outside [`routed_mac_fits`] is refused before any entry is reserved.
136    pub(super) async fn acquire(
137        self: &Arc<Self>,
138        router_mac: &[u8],
139        dnet: u16,
140    ) -> Result<RoutedPathLease, Error> {
141        if !routed_mac_fits(router_mac.len()) {
142            return Err(Error::Encoding(format!(
143                "immediate router MAC address must contain 1..={} octets",
144                bacnet_npdu::MAX_MAC_LEN
145            )));
146        }
147        let key = RoutedPathKey::new(router_mac, dnet);
148        let gate = self.reserve_gate(&key)?;
149        let gate = gate.lock_owned().await;
150        self.wait_for_quarantine(&key).await;
151        Ok(RoutedPathLease {
152            limits: Arc::clone(self),
153            key,
154            _gate: gate,
155        })
156    }
157
158    fn reserve_gate(&self, key: &RoutedPathKey) -> Result<Arc<AsyncMutex<()>>, Error> {
159        let mut state = self.state();
160        let last_used = bump_use(&mut state);
161        if let Some(entry) = state.entries.get_mut(key) {
162            entry.last_used = last_used;
163            return Ok(Arc::clone(&entry.gate));
164        }
165
166        if state.entries.len() >= state.capacity {
167            let reclaim = state
168                .entries
169                .iter()
170                .filter(|(_, entry)| self.is_safely_reclaimable(entry))
171                .min_by(|(key_a, entry_a), (key_b, entry_b)| {
172                    entry_a
173                        .last_used
174                        .cmp(&entry_b.last_used)
175                        .then_with(|| key_a.cmp(key_b))
176                })
177                .map(|(key, _)| key.clone());
178            if let Some(reclaim) = reclaim {
179                state.entries.remove(&reclaim);
180            }
181        }
182        if state.entries.len() >= state.capacity {
183            return Err(Error::RoutedPathCapacityExceeded {
184                capacity: state.capacity,
185            });
186        }
187
188        let entry = RoutedPathEntry::new(last_used);
189        let gate = Arc::clone(&entry.gate);
190        state.entries.insert(key.clone(), entry);
191        Ok(gate)
192    }
193
194    fn is_safely_reclaimable(&self, entry: &RoutedPathEntry) -> bool {
195        entry.configured.is_none()
196            && entry.learned.is_none()
197            && entry.active.is_none()
198            && Arc::strong_count(&entry.gate) == 1
199            && entry
200                .quarantine_started
201                .is_none_or(|started| started.elapsed() >= self.quarantine_horizon)
202    }
203
204    async fn wait_for_quarantine(&self, key: &RoutedPathKey) {
205        loop {
206            let remaining = {
207                let mut state = self.state();
208                let last_used = bump_use(&mut state);
209                let entry = state
210                    .entries
211                    .get_mut(key)
212                    .expect("a held or waiting gate keeps its path entry resident");
213                entry.last_used = last_used;
214                match entry.quarantine_started {
215                    Some(started) => {
216                        let remaining = self.quarantine_horizon.saturating_sub(started.elapsed());
217                        if remaining.is_zero() {
218                            entry.quarantine_started = None;
219                            entry.generation_attempts = 0;
220                            entry.generation_terminal_observed = false;
221                            None
222                        } else {
223                            Some(remaining)
224                        }
225                    }
226                    None => {
227                        entry.generation_attempts = 0;
228                        entry.generation_terminal_observed = false;
229                        None
230                    }
231                }
232            };
233            let Some(remaining) = remaining else {
234                return;
235            };
236            tokio::time::sleep(remaining).await;
237        }
238    }
239
240    pub(super) fn install_active(
241        &self,
242        target: ConfirmedTarget<'_>,
243        tsm_mac: MacAddr,
244        invoke_id: u8,
245        owner: TransactionOwner,
246        forwarded_npci_len: u16,
247        ingress_floor: u64,
248    ) {
249        let ConfirmedTarget::Routed {
250            router_mac,
251            dest_network,
252            ..
253        } = target
254        else {
255            return;
256        };
257        let key = RoutedPathKey::new(router_mac, dest_network);
258        let mut state = self.state();
259        let entry = state
260            .entries
261            .get_mut(&key)
262            .expect("the routed path lease keeps its entry resident");
263        debug_assert!(
264            entry.active.is_none(),
265            "the per-path gate permits only one active routed request"
266        );
267        entry.active = Some(ActiveRoutedSend {
268            path: key,
269            tsm_mac,
270            invoke_id,
271            owner,
272            forwarded_npci_len,
273            outgoing_apdu_len: None,
274            ingress_floor,
275        });
276    }
277
278    pub(super) fn authorize_attempt(
279        &self,
280        target: ConfirmedTarget<'_>,
281        tsm_mac: &MacAddr,
282        invoke_id: u8,
283        owner: &TransactionOwner,
284        outgoing_apdu_len: usize,
285    ) -> bool {
286        let ConfirmedTarget::Routed {
287            router_mac,
288            dest_network,
289            ..
290        } = target
291        else {
292            return true;
293        };
294        let Ok(outgoing_apdu_len) = u16::try_from(outgoing_apdu_len) else {
295            return false;
296        };
297        let key = RoutedPathKey::new(router_mac, dest_network);
298        let mut state = self.state();
299        let Some(entry) = state.entries.get_mut(&key) else {
300            return false;
301        };
302        let Some(active) = entry.active.as_mut() else {
303            return false;
304        };
305        if active.tsm_mac != *tsm_mac
306            || active.invoke_id != invoke_id
307            || !active.owner.same_as(owner)
308        {
309            return false;
310        }
311        active.outgoing_apdu_len = Some(outgoing_apdu_len);
312        entry.generation_attempts = entry.generation_attempts.saturating_add(1);
313        true
314    }
315
316    pub(super) async fn handle_network_control(
317        &self,
318        tsm: &Arc<Mutex<Tsm>>,
319        control: ReceivedNetworkControl,
320    ) {
321        if control.npdu.message_type != Some(NetworkMessageType::REJECT_MESSAGE_TO_NETWORK) {
322            return;
323        }
324        let Ok(reject) = RejectMessageToNetwork::decode(&control.npdu.payload) else {
325            return;
326        };
327        if reject.reason != RejectMessageReason::MESSAGE_TOO_LONG {
328            return;
329        }
330
331        let Some(active) = self.claim_active(
332            &control.source_mac,
333            reject.network,
334            control.ingress_sequence,
335        ) else {
336            return;
337        };
338        let outcome = tsm.lock().await.complete_network_path_too_long_for_owner(
339            &active.tsm_mac,
340            active.invoke_id,
341            &active.owner,
342            reject.network,
343        );
344        if outcome == CompletionOutcome::Delivered {
345            self.commit_learned_limit(active);
346        }
347    }
348
349    fn claim_active(
350        &self,
351        router_mac: &[u8],
352        dnet: u16,
353        ingress_sequence: u64,
354    ) -> Option<ActiveRoutedSend> {
355        let key = RoutedPathKey::new(router_mac, dnet);
356        let mut state = self.state();
357        let entry = state.entries.get_mut(&key)?;
358        if entry.active.as_ref().is_none_or(|active| {
359            active.outgoing_apdu_len.is_none()
360                || active.path != key
361                || ingress_sequence <= active.ingress_floor
362        }) {
363            return None;
364        }
365        entry.active.take()
366    }
367
368    fn commit_learned_limit(&self, active: ActiveRoutedSend) {
369        let Some(attempted_npdu) = active
370            .outgoing_apdu_len
371            .and_then(|apdu| active.forwarded_npci_len.checked_add(apdu))
372        else {
373            return;
374        };
375        let mut state = self.state();
376        let entry = state
377            .entries
378            .get_mut(&active.path)
379            .expect("the reason-4 claimant still holds its routed path gate");
380        let exclusive_max_npdu = entry.learned.as_ref().map_or(attempted_npdu, |learned| {
381            learned.exclusive_max_npdu.min(attempted_npdu)
382        });
383        entry.learned = Some(LearnedPathLimit {
384            exclusive_max_npdu,
385            observed_at: StdInstant::now(),
386        });
387    }
388}
389
390pub(super) struct RoutedPathLease {
391    limits: Arc<RoutedPathLimits>,
392    key: RoutedPathKey,
393    _gate: OwnedMutexGuard<()>,
394}
395
396impl RoutedPathLease {
397    /// The APDU allowance left on this path after the forwarded NPCI that
398    /// [`forwarded_npci_len`] measured.
399    pub(super) fn max_apdu(&self, forwarded_npci_len: u16) -> u16 {
400        let state = self.limits.state();
401        let entry = state
402            .entries
403            .get(&self.key)
404            .expect("path entry remains present while its gate is held");
405        let configured_or_conservative =
406            entry
407                .configured
408                .as_ref()
409                .map_or(CONSERVATIVE_ROUTED_PATH_MAX_NPDU, |configured| {
410                    tracing::trace!(
411                        max_npdu = configured.max_npdu,
412                        provenance = ?configured.provenance,
413                        age = ?configured.observed_at.elapsed(),
414                        "Using configured routed-path NPDU limit"
415                    );
416                    configured.max_npdu
417                });
418        let effective_npdu = entry
419            .learned
420            .as_ref()
421            .map_or(configured_or_conservative, |learned| {
422                tracing::trace!(
423                    exclusive_max_npdu = learned.exclusive_max_npdu,
424                    age = ?learned.observed_at.elapsed(),
425                    "Applying learned routed-path upper bound"
426                );
427                configured_or_conservative.min(learned.exclusive_max_npdu.saturating_sub(1))
428            });
429        effective_npdu.saturating_sub(forwarded_npci_len)
430    }
431
432    /// Mark a source-correlated terminal response for the current generation.
433    /// One attempted frame plus its terminal response proves there cannot be a
434    /// second reason-4 still in flight for that generation. Multi-frame or
435    /// retried generations remain ambiguous and are quarantined on drop.
436    pub(super) fn mark_terminal_observed(&self) {
437        let mut state = self.limits.state();
438        let entry = state
439            .entries
440            .get_mut(&self.key)
441            .expect("path entry remains present while its gate is held");
442        entry.generation_terminal_observed = entry.generation_attempts == 1;
443    }
444
445    fn configure(&self, max_npdu: u16) {
446        let mut state = self.limits.state();
447        let entry = state
448            .entries
449            .get_mut(&self.key)
450            .expect("path entry remains present while its gate is held");
451        entry.configured = Some(ConfiguredPathLimit {
452            max_npdu,
453            provenance: ConfiguredLimitProvenance::ExplicitApi,
454            observed_at: StdInstant::now(),
455        });
456        entry.learned = None;
457    }
458
459    fn clear(&self) {
460        let mut state = self.limits.state();
461        let entry = state
462            .entries
463            .get_mut(&self.key)
464            .expect("path entry remains present while its gate is held");
465        entry.configured = None;
466        entry.learned = None;
467    }
468}
469
470impl Drop for RoutedPathLease {
471    fn drop(&mut self) {
472        let mut state = self.limits.state();
473        if let Some(entry) = state.entries.get_mut(&self.key) {
474            entry.active = None;
475            if entry.generation_attempts > 0 && !entry.generation_terminal_observed {
476                entry.quarantine_started = Some(TokioInstant::now());
477            }
478            entry.generation_attempts = 0;
479            entry.generation_terminal_observed = false;
480        }
481    }
482}
483
484fn bump_use(state: &mut RoutedPathState) -> u64 {
485    state.next_use = state.next_use.saturating_add(1);
486    state.next_use
487}
488
489pub(super) fn routed_path_quarantine_horizon(config: &ClientConfig) -> Duration {
490    // Reuse the configured complete request/retry horizon instead of adding a
491    // separate arbitrary stale-control timer. A zero timeout still receives
492    // one millisecond, the existing configuration's millisecond granularity,
493    // so ambiguous generations never skip quarantine entirely.
494    let attempts = u64::from(config.apdu_retries).saturating_add(1);
495    Duration::from_millis(config.apdu_timeout_ms.saturating_mul(attempts).max(1))
496}
497
498/// The NPCI a router adds when it forwards a request to the DADR: the fixed
499/// fields plus DADR and the SADR it fills with the local source MAC. A DADR or
500/// source MAC outside [`routed_mac_fits`] is refused (#1267): an empty DADR
501/// would have the remote router broadcast the request (#1278).
502pub(super) fn forwarded_npci_len(
503    dadr_len: usize,
504    local_source_mac_len: usize,
505) -> Result<u16, Error> {
506    check_routed_dadr(dadr_len)?;
507    if !routed_mac_fits(local_source_mac_len) {
508        return Err(Error::Encoding(format!(
509            "local source MAC address for routed forwarding must contain 1..={} octets",
510            bacnet_npdu::MAX_MAC_LEN
511        )));
512    }
513    Ok(u16::try_from(9 + dadr_len + local_source_mac_len)
514        .expect("two addresses within bacnet_npdu::MAX_MAC_LEN fit a u16"))
515}
516
517impl<T: TransportPort + 'static> BACnetClient<T> {
518    /// Configure the maximum NPDU for one immediate-router/DNET path.
519    ///
520    /// The update waits for any request already active on the same path. It
521    /// replaces the previous configured value and deliberately clears learned
522    /// reason-4 evidence for that path. A router MAC outside 1 to
523    /// [`MAX_MAC_LEN`](bacnet_npdu::MAX_MAC_LEN) octets or a DNET outside 1..=65534 is
524    /// refused with [`Error::Encoding`].
525    pub async fn configure_routed_path_max_npdu(
526        &self,
527        router_mac: &[u8],
528        dnet: u16,
529        max_npdu: u16,
530    ) -> Result<(), Error> {
531        check_remote_dnet(dnet)?;
532        let lease = self.routed_path_limits.acquire(router_mac, dnet).await?;
533        lease.configure(max_npdu);
534        Ok(())
535    }
536
537    /// Restore one immediate-router/DNET path to the conservative default.
538    ///
539    /// The update waits for an active request and clears both configured and
540    /// learned reason-4 evidence. Later requests therefore use the 228-octet
541    /// conservative routed NPDU envelope until new evidence is supplied. The
542    /// router MAC and DNET are checked as in
543    /// [`configure_routed_path_max_npdu`](Self::configure_routed_path_max_npdu).
544    pub async fn clear_routed_path_limit(&self, router_mac: &[u8], dnet: u16) -> Result<(), Error> {
545        check_remote_dnet(dnet)?;
546        let lease = self.routed_path_limits.acquire(router_mac, dnet).await?;
547        lease.clear();
548        Ok(())
549    }
550}
551
552/// Whether `dnet` names one remote network: 1..=65534. DNET 0 names no
553/// network and 65535 is the global broadcast, so neither can key a routed path
554/// or address a confirmed request (Clauses 6.2.2.1 and 6.3, #1278). The path
555/// APIs and every routed confirmed request share this check;
556/// [`RoutedPathLimits::acquire`] checks the router MAC.
557pub(super) fn check_remote_dnet(dnet: u16) -> Result<(), Error> {
558    if dnet == 0 || dnet == u16::MAX {
559        return Err(Error::Encoding("routed DNET must be in 1..=65534".into()));
560    }
561    Ok(())
562}
563
564/// A DADR that names one station: 1 to [`MAX_MAC_LEN`](bacnet_npdu::MAX_MAC_LEN) octets.
565/// With no DADR the router on DNET broadcasts the NPDU, and a broadcast
566/// network address carries only unconfirmed requests (Clause 6.3).
567fn check_routed_dadr(dadr_len: usize) -> Result<(), Error> {
568    if !routed_mac_fits(dadr_len) {
569        return Err(Error::Encoding(format!(
570            "routed destination MAC address must contain 1..={} octets",
571            bacnet_npdu::MAX_MAC_LEN
572        )));
573    }
574    Ok(())
575}
576
577/// Refuse a routed confirmed destination that does not name exactly one
578/// device: a DNET [`check_remote_dnet`] refuses or a DADR outside
579/// [`routed_mac_fits`]. The client's own request path runs it on the target
580/// as named, before a target on the local network is sent locally, and
581/// measures the forwarded NPCI with [`forwarded_npci_len`] only for a target
582/// still routed.
583pub(crate) fn check_routed_unicast(dnet: u16, dadr_len: usize) -> Result<(), Error> {
584    check_remote_dnet(dnet)?;
585    check_routed_dadr(dadr_len)
586}
587
588#[cfg(test)]
589#[path = "routed_path_limits_tests.rs"]
590mod tests;
bacnet_client/client/routed_path_limits.rs

bacnet_client/client/
routed_path_limits.rs
