bacnet_server/server/
discovery.rs

1//! Discovery rate-limiting, duplicate suppression, and directed response handling.
2//!
3//! Controls response amplification for unconfirmed discovery services (Who-Is
4//! and Who-Has) per ASHRAE 135-2020 Clauses 16.9 and 16.10.
5
6use std::collections::HashMap;
7use std::sync::atomic::{AtomicU64, Ordering};
8use std::sync::{Arc, Mutex};
9use std::time::{Duration, Instant};
10
11use bytes::BytesMut;
12
13use bacnet_encoding::apdu::{encode_apdu, Apdu, UnconfirmedRequest as UnconfirmedRequestPdu};
14use bacnet_encoding::npdu::NpduAddress;
15use bacnet_network::layer::{NetworkLayer, ReceivedApdu};
16use bacnet_npdu::NetworkPriority;
17use bacnet_objects::database::ObjectDatabase;
18use bacnet_services::who_has::{WhoHasObject, WhoHasRequest};
19use bacnet_services::who_is::{DeviceInstanceRange, IAmRequest, WhoIsRequest};
20use bacnet_transport::port::TransportPort;
21use bacnet_types::enums::{ErrorClass, ErrorCode, UnconfirmedServiceChoice};
22use bacnet_types::error::Error;
23use bacnet_types::primitives::ObjectIdentifier;
24use bacnet_types::MacAddr;
25use tokio::sync::RwLock;
26
27use super::{BACnetServer, CommState, ServerConfig};
28
29/// Configuration policy for discovery rate limiting and duplicate suppression.
30#[derive(Debug, Clone)]
31pub struct DiscoveryPolicy {
32    /// Maximum discovery responses emitted globally per second (default 64).
33    pub max_responses_per_sec_global: u32,
34    /// Maximum discovery response bytes emitted globally per second (default 65,536).
35    pub max_bytes_per_sec_global: usize,
36    /// Maximum discovery responses emitted per source per second (default 16).
37    pub max_responses_per_sec_per_source: u32,
38    /// Maximum discovery response bytes emitted per source per second (default 16,384).
39    pub max_bytes_per_sec_per_source: usize,
40    /// Maximum burst capacity for global discovery responses (default 64).
41    pub global_burst_capacity: u32,
42    /// Maximum burst capacity for per-source discovery responses (default 16).
43    pub source_burst_capacity: u32,
44    /// Headroom reserved exclusively for configured reserved sources (default 8).
45    pub reserved_capacity: u32,
46    /// Sources permitted to consume from reserved capacity.
47    pub reserved_sources: Vec<MacAddr>,
48    /// Coalescing window for duplicate discovery request suppression (default 200ms).
49    pub coalesce_window: Duration,
50    /// Prefer directed unicast responses over network broadcast (default true).
51    pub prefer_directed_responses: bool,
52    /// Maximum number of distinct discovery sources tracked concurrently (default 256).
53    pub max_tracked_sources: usize,
54}
55
56impl Default for DiscoveryPolicy {
57    fn default() -> Self {
58        Self {
59            max_responses_per_sec_global: 64,
60            max_bytes_per_sec_global: 65_536,
61            max_responses_per_sec_per_source: 16,
62            max_bytes_per_sec_per_source: 16_384,
63            global_burst_capacity: 64,
64            source_burst_capacity: 16,
65            reserved_capacity: 8,
66            reserved_sources: Vec::new(),
67            coalesce_window: Duration::from_millis(200),
68            prefer_directed_responses: true,
69            max_tracked_sources: 256,
70        }
71    }
72}
73
74impl DiscoveryPolicy {
75    /// Create an unlimited discovery policy with all rate limits disabled.
76    pub fn unlimited() -> Self {
77        Self {
78            max_responses_per_sec_global: u32::MAX,
79            max_bytes_per_sec_global: usize::MAX,
80            max_responses_per_sec_per_source: u32::MAX,
81            max_bytes_per_sec_per_source: usize::MAX,
82            global_burst_capacity: u32::MAX,
83            source_burst_capacity: u32::MAX,
84            reserved_capacity: 0,
85            coalesce_window: Duration::ZERO,
86            prefer_directed_responses: false,
87            ..Default::default()
88        }
89    }
90
91    /// Return a sanitized copy with valid burst and capacity bounds.
92    pub fn sanitized(&self) -> Self {
93        let mut policy = self.clone();
94        if (1..u32::MAX).contains(&policy.max_responses_per_sec_global) {
95            policy.global_burst_capacity = policy.global_burst_capacity.max(1);
96        }
97        if (1..u32::MAX).contains(&policy.max_responses_per_sec_per_source) {
98            policy.source_burst_capacity = policy.source_burst_capacity.max(1);
99        }
100        policy.reserved_capacity = policy.reserved_capacity.min(policy.global_burst_capacity);
101        policy.max_tracked_sources = policy.max_tracked_sources.max(1);
102        policy
103    }
104}
105
106/// Operational counters for discovery requests and responses.
107///
108/// Non-exhaustive, so a counter can be added without breaking callers: read
109/// it from [`BACnetServer::discovery_counters`](crate::server::BACnetServer::discovery_counters)
110/// and its fields rather than building one.
111#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
112#[non_exhaustive]
113pub struct DiscoveryCounters {
114    /// Total Who-Is requests received.
115    pub who_is_received: u64,
116    /// Total Who-Has requests received.
117    pub who_has_received: u64,
118    /// Total I-Am responses sent.
119    pub i_am_sent: u64,
120    /// Total I-Have responses sent.
121    pub i_have_sent: u64,
122    /// Total response bytes sent for discovery.
123    pub response_bytes_sent: u64,
124    /// Total duplicate or cached requests coalesced without sending responses.
125    pub requests_coalesced: u64,
126    /// Responses throttled by per-source rate or burst limits.
127    pub responses_throttled_source: u64,
128    /// Responses throttled by global rate or reserved capacity limits.
129    pub responses_throttled_global: u64,
130    /// Total directed unicast discovery responses sent.
131    pub directed_responses_sent: u64,
132    /// Who-Is and Who-Has requests dropped because they didn't decode, such
133    /// as one with a single limit or octets after its last member. They are
134    /// dropped before any other work and count in neither received total.
135    pub malformed_dropped: u64,
136}
137
138#[derive(Debug, Default)]
139pub(crate) struct AtomicDiscoveryCounters {
140    pub(crate) who_is_received: AtomicU64,
141    pub(crate) who_has_received: AtomicU64,
142    pub(crate) i_am_sent: AtomicU64,
143    pub(crate) i_have_sent: AtomicU64,
144    pub(crate) response_bytes_sent: AtomicU64,
145    pub(crate) requests_coalesced: AtomicU64,
146    pub(crate) responses_throttled_source: AtomicU64,
147    pub(crate) responses_throttled_global: AtomicU64,
148    pub(crate) directed_responses_sent: AtomicU64,
149    pub(crate) malformed_dropped: AtomicU64,
150}
151
152impl AtomicDiscoveryCounters {
153    pub(crate) fn snapshot(&self) -> DiscoveryCounters {
154        DiscoveryCounters {
155            who_is_received: self.who_is_received.load(Ordering::Relaxed),
156            who_has_received: self.who_has_received.load(Ordering::Relaxed),
157            i_am_sent: self.i_am_sent.load(Ordering::Relaxed),
158            i_have_sent: self.i_have_sent.load(Ordering::Relaxed),
159            response_bytes_sent: self.response_bytes_sent.load(Ordering::Relaxed),
160            requests_coalesced: self.requests_coalesced.load(Ordering::Relaxed),
161            responses_throttled_source: self.responses_throttled_source.load(Ordering::Relaxed),
162            responses_throttled_global: self.responses_throttled_global.load(Ordering::Relaxed),
163            directed_responses_sent: self.directed_responses_sent.load(Ordering::Relaxed),
164            malformed_dropped: self.malformed_dropped.load(Ordering::Relaxed),
165        }
166    }
167
168    #[inline]
169    pub(crate) fn inc(&self, c: &AtomicU64) {
170        c.fetch_add(1, Ordering::Relaxed);
171    }
172
173    pub(crate) fn record_sent(&self, bytes: usize, directed: bool, is_who_is: bool) {
174        if is_who_is {
175            self.i_am_sent.fetch_add(1, Ordering::Relaxed);
176        } else {
177            self.i_have_sent.fetch_add(1, Ordering::Relaxed);
178        }
179        self.response_bytes_sent
180            .fetch_add(bytes as u64, Ordering::Relaxed);
181        if directed {
182            self.directed_responses_sent.fetch_add(1, Ordering::Relaxed);
183        }
184    }
185}
186
187/// Target of a Who-Has query.
188#[derive(Debug, Clone, PartialEq, Eq, Hash)]
189pub(crate) enum WhoHasTarget {
190    Id(ObjectIdentifier),
191    Name(String),
192}
193
194/// Network-aware identity of a discovery requester.
195#[derive(Debug, Clone, PartialEq, Eq, Hash)]
196pub(crate) struct SourceKey {
197    pub(crate) network: u16,
198    pub(crate) mac: MacAddr,
199}
200
201impl SourceKey {
202    pub(crate) fn from_parts(source_mac: &MacAddr, source_network: Option<&NpduAddress>) -> Self {
203        if let Some(net) = source_network {
204            if (1..=0xFFFE).contains(&net.network) && !net.mac_address.is_empty() {
205                return Self {
206                    network: net.network,
207                    mac: MacAddr::from_slice(&net.mac_address),
208                };
209            }
210        }
211        Self {
212            network: 0,
213            mac: source_mac.clone(),
214        }
215    }
216
217    pub(crate) fn from_received(received: &ReceivedApdu) -> Self {
218        Self::from_parts(&received.source_mac, received.source_network.as_ref())
219    }
220}
221
222#[derive(Debug)]
223struct SourceState {
224    tokens: f64,
225    byte_tokens: f64,
226    last_refill: Instant,
227    last_activity: Instant,
228    last_who_is: Option<Instant>,
229    last_who_has: HashMap<WhoHasTarget, Instant>,
230}
231
232fn refill_bucket(
233    tokens: &mut f64,
234    byte_tokens: &mut f64,
235    last_refill: &mut Instant,
236    rate: u32,
237    byte_rate: usize,
238    burst: u32,
239    now: Instant,
240) {
241    let elapsed = now.saturating_duration_since(*last_refill).as_secs_f64();
242    if elapsed > 0.0 {
243        *tokens = if rate == u32::MAX {
244            burst as f64
245        } else {
246            (*tokens + elapsed * (rate as f64)).min(burst as f64)
247        };
248        *byte_tokens = if byte_rate == usize::MAX {
249            byte_rate as f64
250        } else {
251            (*byte_tokens + elapsed * (byte_rate as f64)).min(byte_rate as f64)
252        };
253        *last_refill = now;
254    }
255}
256
257impl SourceState {
258    fn refill(&mut self, policy: &DiscoveryPolicy, now: Instant) {
259        refill_bucket(
260            &mut self.tokens,
261            &mut self.byte_tokens,
262            &mut self.last_refill,
263            policy.max_responses_per_sec_per_source,
264            policy.max_bytes_per_sec_per_source,
265            policy.source_burst_capacity,
266            now,
267        );
268    }
269
270    fn has_capacity(&self, policy: &DiscoveryPolicy, bytes: usize) -> bool {
271        (policy.max_responses_per_sec_per_source == u32::MAX || self.tokens >= 1.0)
272            && (policy.max_bytes_per_sec_per_source == usize::MAX
273                || self.byte_tokens >= bytes as f64)
274    }
275
276    fn deduct(&mut self, policy: &DiscoveryPolicy, bytes: usize) {
277        if policy.max_responses_per_sec_per_source != u32::MAX {
278            self.tokens -= 1.0;
279        }
280        if policy.max_bytes_per_sec_per_source != usize::MAX {
281            self.byte_tokens -= bytes as f64;
282        }
283    }
284}
285
286#[derive(Debug)]
287struct DiscoveryState {
288    global_tokens: f64,
289    global_byte_tokens: f64,
290    global_last_refill: Instant,
291    last_broadcast_who_is: Option<Instant>,
292    last_broadcast_who_has: HashMap<WhoHasTarget, Instant>,
293    negative_who_has: HashMap<WhoHasTarget, Instant>,
294    sources: HashMap<SourceKey, SourceState>,
295}
296
297impl DiscoveryState {
298    fn refill_global(&mut self, policy: &DiscoveryPolicy, now: Instant) {
299        refill_bucket(
300            &mut self.global_tokens,
301            &mut self.global_byte_tokens,
302            &mut self.global_last_refill,
303            policy.max_responses_per_sec_global,
304            policy.max_bytes_per_sec_global,
305            policy.global_burst_capacity,
306            now,
307        );
308    }
309
310    fn has_global_capacity(
311        &self,
312        policy: &DiscoveryPolicy,
313        is_reserved: bool,
314        bytes: usize,
315    ) -> bool {
316        let count_ok = policy.max_responses_per_sec_global == u32::MAX || {
317            let req = if is_reserved {
318                1.0
319            } else {
320                1.0 + policy.reserved_capacity as f64
321            };
322            self.global_tokens >= req
323        };
324        count_ok
325            && (policy.max_bytes_per_sec_global == usize::MAX
326                || self.global_byte_tokens >= bytes as f64)
327    }
328
329    fn deduct_global(&mut self, policy: &DiscoveryPolicy, bytes: usize) {
330        if policy.max_responses_per_sec_global != u32::MAX {
331            self.global_tokens -= 1.0;
332        }
333        if policy.max_bytes_per_sec_global != usize::MAX {
334            self.global_byte_tokens -= bytes as f64;
335        }
336    }
337
338    fn get_or_create_source_mut(
339        &mut self,
340        policy: &DiscoveryPolicy,
341        key: &SourceKey,
342        now: Instant,
343    ) -> &mut SourceState {
344        if !self.sources.contains_key(key) {
345            ensure_source_capacity(self, policy, now);
346        }
347        let src = self
348            .sources
349            .entry(key.clone())
350            .or_insert_with(|| SourceState {
351                tokens: policy.source_burst_capacity as f64,
352                byte_tokens: policy.max_bytes_per_sec_per_source as f64,
353                last_refill: now,
354                last_activity: now,
355                last_who_is: None,
356                last_who_has: HashMap::new(),
357            });
358        src.refill(policy, now);
359        src
360    }
361
362    fn is_coalesced_who_has(
363        &self,
364        policy: &DiscoveryPolicy,
365        target: &WhoHasTarget,
366        key: &SourceKey,
367        is_broadcast: bool,
368        now: Instant,
369    ) -> bool {
370        if policy.coalesce_window.is_zero() {
371            return false;
372        }
373        let is_recent = |ts: &Instant| now.saturating_duration_since(*ts) < policy.coalesce_window;
374        self.negative_who_has.get(target).is_some_and(is_recent)
375            || (is_broadcast
376                && self
377                    .last_broadcast_who_has
378                    .get(target)
379                    .is_some_and(is_recent))
380            || self
381                .sources
382                .get(key)
383                .and_then(|s| s.last_who_has.get(target))
384                .is_some_and(is_recent)
385    }
386}
387
388#[derive(Debug)]
389struct AtomicDeviceInstance(AtomicU64);
390
391impl AtomicDeviceInstance {
392    const PRESENT_BIT: u64 = 0x1_0000_0000;
393
394    fn new(opt: Option<u32>) -> Self {
395        Self(AtomicU64::new(
396            opt.map_or(0, |i| Self::PRESENT_BIT | (i as u64)),
397        ))
398    }
399
400    fn load(&self, order: Ordering) -> Option<u32> {
401        let val = self.0.load(order);
402        (val & Self::PRESENT_BIT != 0).then_some(val as u32)
403    }
404}
405
406#[derive(Debug, Clone, Copy, PartialEq, Eq)]
407pub(crate) enum PreCheckDecision {
408    Admit,
409    OutOfRange,
410    Coalesced,
411    ThrottledSource,
412    ThrottledGlobal,
413    DecodeError,
414}
415
416/// Thread-safe rate limiter, duplicate coalescer, and negative query cache.
417#[derive(Debug)]
418pub(crate) struct DiscoveryLimiter {
419    policy: DiscoveryPolicy,
420    counters: AtomicDiscoveryCounters,
421    device_instance: AtomicDeviceInstance,
422    state: Mutex<DiscoveryState>,
423}
424
425impl DiscoveryLimiter {
426    /// Every throttle and coalescing window here is measured on
427    /// [`runtime_clock::now`](super::runtime_clock::now), as its callers'
428    /// instants are.
429    pub(crate) fn new(policy: DiscoveryPolicy, device_instance: Option<u32>) -> Self {
430        let now = super::runtime_clock::now();
431        let state = DiscoveryState {
432            global_tokens: policy.global_burst_capacity as f64,
433            global_byte_tokens: policy.max_bytes_per_sec_global as f64,
434            global_last_refill: now,
435            last_broadcast_who_is: None,
436            last_broadcast_who_has: HashMap::new(),
437            negative_who_has: HashMap::new(),
438            sources: HashMap::new(),
439        };
440        Self {
441            policy,
442            counters: AtomicDiscoveryCounters::default(),
443            device_instance: AtomicDeviceInstance::new(device_instance),
444            state: Mutex::new(state),
445        }
446    }
447
448    pub(crate) fn counters(&self) -> DiscoveryCounters {
449        self.counters.snapshot()
450    }
451
452    #[allow(dead_code)]
453    pub(crate) fn policy(&self) -> &DiscoveryPolicy {
454        &self.policy
455    }
456
457    #[cfg(test)]
458    pub(crate) fn negative_who_has_count(&self) -> usize {
459        self.state.lock().unwrap().negative_who_has.len()
460    }
461
462    #[cfg(test)]
463    pub(crate) fn source_byte_tokens(
464        &self,
465        source_mac: &MacAddr,
466        source_network: Option<&NpduAddress>,
467    ) -> Option<f64> {
468        let key = SourceKey::from_parts(source_mac, source_network);
469        self.state
470            .lock()
471            .unwrap()
472            .sources
473            .get(&key)
474            .map(|s| s.byte_tokens)
475    }
476
477    fn check_instance_in_range(&self, range: Option<DeviceInstanceRange>) -> bool {
478        self.device_instance
479            .load(Ordering::Acquire)
480            .is_some_and(|instance| range.is_none_or(|range| range.contains(instance)))
481    }
482
483    pub(crate) fn pre_check_who_is(
484        &self,
485        req_bytes: &[u8],
486        received: &ReceivedApdu,
487        now: Instant,
488    ) -> PreCheckDecision {
489        // RB-07 compat mode: provenance threaded here for RB-09, no decision.
490        let _ = received.provenance;
491        let who_is = match WhoIsRequest::decode(req_bytes) {
492            Ok(w) => w,
493            Err(_) => {
494                self.counters.inc(&self.counters.malformed_dropped);
495                return PreCheckDecision::DecodeError;
496            }
497        };
498
499        self.counters.inc(&self.counters.who_is_received);
500
501        if !self.check_instance_in_range(who_is.range) {
502            return PreCheckDecision::OutOfRange;
503        }
504
505        let is_broadcast = received.is_group || received.link_layer_group;
506        let source_key = SourceKey::from_received(received);
507        let mut state = self.state.lock().unwrap();
508
509        // 1. Coalescing check
510        if !self.policy.coalesce_window.is_zero() {
511            let is_recent =
512                |ts: &Instant| now.saturating_duration_since(*ts) < self.policy.coalesce_window;
513            let bcast = is_broadcast && state.last_broadcast_who_is.as_ref().is_some_and(is_recent);
514            let src = state
515                .sources
516                .get(&source_key)
517                .and_then(|s| s.last_who_is.as_ref())
518                .is_some_and(is_recent);
519            if bcast || src {
520                self.counters.inc(&self.counters.requests_coalesced);
521                return PreCheckDecision::Coalesced;
522            }
523        }
524
525        // 2. Refill & Rate limits
526        state.refill_global(&self.policy, now);
527        let is_reserved = is_source_reserved(&self.policy, &source_key, received);
528        const ESTIMATED_I_AM_BYTES: usize = 64;
529
530        if !state.has_global_capacity(&self.policy, is_reserved, ESTIMATED_I_AM_BYTES) {
531            self.counters.inc(&self.counters.responses_throttled_global);
532            return PreCheckDecision::ThrottledGlobal;
533        }
534
535        let src = state.get_or_create_source_mut(&self.policy, &source_key, now);
536        if !src.has_capacity(&self.policy, ESTIMATED_I_AM_BYTES) {
537            self.counters.inc(&self.counters.responses_throttled_source);
538            return PreCheckDecision::ThrottledSource;
539        }
540
541        src.deduct(&self.policy, ESTIMATED_I_AM_BYTES);
542        src.last_activity = now;
543        src.last_who_is = Some(now);
544        state.deduct_global(&self.policy, ESTIMATED_I_AM_BYTES);
545        if is_broadcast && !self.policy.prefer_directed_responses {
546            state.last_broadcast_who_is = Some(now);
547        }
548
549        PreCheckDecision::Admit
550    }
551
552    pub(crate) fn record_i_am_sent(
553        &self,
554        actual_bytes: usize,
555        directed: bool,
556        source_mac: &MacAddr,
557        source_network: Option<&NpduAddress>,
558        now: Instant,
559    ) {
560        self.counters.record_sent(actual_bytes, directed, true);
561
562        const ESTIMATED_I_AM_BYTES: usize = 64;
563        let delta = (actual_bytes as f64) - (ESTIMATED_I_AM_BYTES as f64);
564        let mut state = self.state.lock().unwrap();
565        if delta != 0.0 {
566            if self.policy.max_bytes_per_sec_global != usize::MAX {
567                state.global_byte_tokens = (state.global_byte_tokens - delta).max(0.0);
568            }
569            let key = SourceKey::from_parts(source_mac, source_network);
570            if let Some(src) = state.sources.get_mut(&key) {
571                if self.policy.max_bytes_per_sec_per_source != usize::MAX {
572                    src.byte_tokens = (src.byte_tokens - delta).max(0.0);
573                }
574            }
575        }
576        if !directed {
577            state.last_broadcast_who_is = Some(now);
578        }
579    }
580
581    pub(crate) fn pre_check_who_has(
582        &self,
583        req_bytes: &[u8],
584        received: &ReceivedApdu,
585        now: Instant,
586    ) -> PreCheckDecision {
587        // RB-07 compat mode: provenance threaded here for RB-09, no decision.
588        let _ = received.provenance;
589        let who_has = match WhoHasRequest::decode(req_bytes) {
590            Ok(w) => w,
591            Err(_) => {
592                self.counters.inc(&self.counters.malformed_dropped);
593                return PreCheckDecision::DecodeError;
594            }
595        };
596
597        self.counters.inc(&self.counters.who_has_received);
598
599        if !self.check_instance_in_range(who_has.range) {
600            return PreCheckDecision::OutOfRange;
601        }
602
603        let target = match who_has.object {
604            WhoHasObject::Identifier(oid) => WhoHasTarget::Id(oid),
605            WhoHasObject::Name(name) => WhoHasTarget::Name(name),
606        };
607
608        let is_broadcast = received.is_group || received.link_layer_group;
609        let source_key = SourceKey::from_received(received);
610        let mut state = self.state.lock().unwrap();
611
612        if state.is_coalesced_who_has(&self.policy, &target, &source_key, is_broadcast, now) {
613            self.counters.inc(&self.counters.requests_coalesced);
614            return PreCheckDecision::Coalesced;
615        }
616
617        state.refill_global(&self.policy, now);
618        let is_reserved = is_source_reserved(&self.policy, &source_key, received);
619        const ESTIMATED_I_HAVE_BYTES: usize = 64;
620
621        if !state.has_global_capacity(&self.policy, is_reserved, ESTIMATED_I_HAVE_BYTES) {
622            self.counters.inc(&self.counters.responses_throttled_global);
623            return PreCheckDecision::ThrottledGlobal;
624        }
625
626        let src = state.get_or_create_source_mut(&self.policy, &source_key, now);
627        if !src.has_capacity(&self.policy, ESTIMATED_I_HAVE_BYTES) {
628            self.counters.inc(&self.counters.responses_throttled_source);
629            return PreCheckDecision::ThrottledSource;
630        }
631
632        src.last_activity = now;
633        PreCheckDecision::Admit
634    }
635
636    pub(crate) fn is_negative_who_has(&self, target: &WhoHasTarget, now: Instant) -> bool {
637        if self.policy.coalesce_window.is_zero() {
638            return false;
639        }
640        let state = self.state.lock().unwrap();
641        if let Some(ts) = state.negative_who_has.get(target) {
642            now.saturating_duration_since(*ts) < self.policy.coalesce_window
643        } else {
644            false
645        }
646    }
647
648    pub(crate) fn record_negative_who_has(&self, target: WhoHasTarget, now: Instant) {
649        const MAX_NEGATIVE_WHO_HAS: usize = 512;
650        let mut state = self.state.lock().unwrap();
651        if state.negative_who_has.len() >= MAX_NEGATIVE_WHO_HAS {
652            let window = self.policy.coalesce_window;
653            state
654                .negative_who_has
655                .retain(|_, ts| now.saturating_duration_since(*ts) < window);
656        }
657        while state.negative_who_has.len() >= MAX_NEGATIVE_WHO_HAS {
658            let Some(oldest) = state
659                .negative_who_has
660                .iter()
661                .min_by_key(|(_, ts)| *ts)
662                .map(|(k, _)| k.clone())
663            else {
664                break;
665            };
666            state.negative_who_has.remove(&oldest);
667        }
668        state.negative_who_has.insert(target, now);
669    }
670
671    pub(crate) fn try_consume_who_has_response(
672        &self,
673        target: &WhoHasTarget,
674        received: &ReceivedApdu,
675        response_bytes: usize,
676        now: Instant,
677    ) -> bool {
678        let source_key = SourceKey::from_received(received);
679        let mut state = self.state.lock().unwrap();
680        let is_broadcast = received.is_group || received.link_layer_group;
681
682        if state.is_coalesced_who_has(&self.policy, target, &source_key, is_broadcast, now) {
683            self.counters.inc(&self.counters.requests_coalesced);
684            return false;
685        }
686
687        state.refill_global(&self.policy, now);
688        let is_reserved = is_source_reserved(&self.policy, &source_key, received);
689
690        if !state.has_global_capacity(&self.policy, is_reserved, response_bytes) {
691            self.counters.inc(&self.counters.responses_throttled_global);
692            return false;
693        }
694
695        let src = state.get_or_create_source_mut(&self.policy, &source_key, now);
696        if !src.has_capacity(&self.policy, response_bytes) {
697            self.counters.inc(&self.counters.responses_throttled_source);
698            return false;
699        }
700
701        src.deduct(&self.policy, response_bytes);
702        src.last_activity = now;
703        src.last_who_has.insert(target.clone(), now);
704        state.deduct_global(&self.policy, response_bytes);
705        if is_broadcast && !self.policy.prefer_directed_responses {
706            state.last_broadcast_who_has.insert(target.clone(), now);
707        }
708
709        true
710    }
711
712    pub(crate) fn record_i_have_sent(
713        &self,
714        actual_bytes: usize,
715        directed: bool,
716        target: &WhoHasTarget,
717        _source_mac: &MacAddr,
718        _source_network: Option<&NpduAddress>,
719        now: Instant,
720    ) {
721        self.counters.record_sent(actual_bytes, directed, false);
722
723        if !directed {
724            let mut state = self.state.lock().unwrap();
725            state.last_broadcast_who_has.insert(target.clone(), now);
726        }
727    }
728}
729
730fn is_source_reserved(policy: &DiscoveryPolicy, key: &SourceKey, recv: &ReceivedApdu) -> bool {
731    policy
732        .reserved_sources
733        .iter()
734        .any(|r| r == &key.mac || r == &recv.source_mac)
735}
736
737fn ensure_source_capacity(state: &mut DiscoveryState, policy: &DiscoveryPolicy, now: Instant) {
738    if state.sources.len() < policy.max_tracked_sources {
739        return;
740    }
741    state.sources.retain(|_, src| {
742        now.saturating_duration_since(src.last_activity) < Duration::from_secs(10)
743    });
744    while state.sources.len() >= policy.max_tracked_sources {
745        let Some(key) = state
746            .sources
747            .iter()
748            .min_by_key(|(_, src)| src.last_activity)
749            .map(|(k, _)| k.clone())
750        else {
751            break;
752        };
753        state.sources.remove(&key);
754    }
755}
756
757/// Broadcast this device's I-Am unless DeviceCommunicationControl restricts
758/// initiation. This call announces independently of any received Who-Is;
759/// under DISABLE_INITIATION it sends nothing and fails with
760/// `SERVICES` / `COMMUNICATION_DISABLED`, the code Clause 18.6 keeps for local
761/// work that could not initiate a service for that reason.
762pub(crate) async fn broadcast_i_am_from<T: TransportPort + 'static>(
763    config: &ServerConfig,
764    db: &Arc<RwLock<ObjectDatabase>>,
765    network: &Arc<NetworkLayer<T>>,
766    comm_state: &CommState,
767    limiter: Option<&Arc<DiscoveryLimiter>>,
768) -> Result<(), Error> {
769    let guard = db.read().await;
770    crate::local_device::announcement_device(&guard, &config.device)
771        .ok_or_else(|| Error::Encoding("no Device object in database".into()))?;
772
773    // Both discovery paths encode the same frozen, resolved declarations.
774    let mut service_buf = BytesMut::new();
775    iam_request_for(config).encode(&mut service_buf);
776
777    let mut buf = BytesMut::new();
778    encode_apdu(
779        &mut buf,
780        &Apdu::UnconfirmedRequest(UnconfirmedRequestPdu {
781            service_choice: UnconfirmedServiceChoice::I_AM,
782            service_request: service_buf.freeze(),
783        }),
784    )?;
785
786    drop(guard);
787    if comm_state.initiation_restricted() {
788        return Err(Error::Protocol {
789            class: ErrorClass::SERVICES,
790            code: ErrorCode::COMMUNICATION_DISABLED,
791        });
792    }
793    if let Some(limiter) = limiter {
794        limiter.record_i_am_sent(
795            buf.len(),
796            false,
797            &MacAddr::new(),
798            None,
799            super::runtime_clock::now(),
800        );
801    }
802
803    network
804        .broadcast_apdu(&buf, false, NetworkPriority::Normal)
805        .await
806}
807
808/// I-Am from the server's frozen, resolved Device declarations.
809#[doc(hidden)]
810pub fn iam_request_for(config: &ServerConfig) -> IAmRequest {
811    IAmRequest {
812        object_identifier: config.device.device_oid(),
813        max_apdu_length: u32::from(config.device.max_apdu_length()),
814        segmentation_supported: config.device.segmentation(),
815        vendor_id: config.device.vendor_id(),
816    }
817}
818
819impl<T: TransportPort + 'static> BACnetServer<T> {
820    /// Broadcast I-Am through this server's bounded local-send owner.
821    /// Cancelling the waiter leaves admitted work owned until completion or stop.
822    /// While DeviceCommunicationControl restricts initiation nothing is sent
823    /// and this fails with `SERVICES` / `COMMUNICATION_DISABLED` (Clause 16.1).
824    pub async fn broadcast_i_am(&self) -> Result<(), Error> {
825        self.broadcaster
826            .send(Some(Arc::clone(&self.discovery_limiter)))
827            .await
828    }
829}
bacnet_server/server/discovery.rs

bacnet_server/server/
discovery.rs
