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::{Apdu, UnconfirmedRequest as UnconfirmedRequestPdu, encode_apdu};
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::MacAddr;
22use bacnet_types::enums::{ErrorClass, ErrorCode, UnconfirmedServiceChoice};
23use bacnet_types::error::Error;
24use bacnet_types::primitives::ObjectIdentifier;
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            && (1..=0xFFFE).contains(&net.network)
205            && !net.mac_address.is_empty()
206        {
207            return Self {
208                network: net.network,
209                mac: MacAddr::from_slice(&net.mac_address),
210            };
211        }
212        Self {
213            network: 0,
214            mac: source_mac.clone(),
215        }
216    }
217
218    pub(crate) fn from_received(received: &ReceivedApdu) -> Self {
219        Self::from_parts(&received.source_mac, received.source_network.as_ref())
220    }
221}
222
223#[derive(Debug)]
224struct SourceState {
225    tokens: f64,
226    byte_tokens: f64,
227    last_refill: Instant,
228    last_activity: Instant,
229    last_who_is: Option<Instant>,
230    last_who_has: HashMap<WhoHasTarget, Instant>,
231}
232
233fn refill_bucket(
234    tokens: &mut f64,
235    byte_tokens: &mut f64,
236    last_refill: &mut Instant,
237    rate: u32,
238    byte_rate: usize,
239    burst: u32,
240    now: Instant,
241) {
242    let elapsed = now.saturating_duration_since(*last_refill).as_secs_f64();
243    if elapsed > 0.0 {
244        *tokens = if rate == u32::MAX {
245            burst as f64
246        } else {
247            (*tokens + elapsed * (rate as f64)).min(burst as f64)
248        };
249        *byte_tokens = if byte_rate == usize::MAX {
250            byte_rate as f64
251        } else {
252            (*byte_tokens + elapsed * (byte_rate as f64)).min(byte_rate as f64)
253        };
254        *last_refill = now;
255    }
256}
257
258impl SourceState {
259    fn refill(&mut self, policy: &DiscoveryPolicy, now: Instant) {
260        refill_bucket(
261            &mut self.tokens,
262            &mut self.byte_tokens,
263            &mut self.last_refill,
264            policy.max_responses_per_sec_per_source,
265            policy.max_bytes_per_sec_per_source,
266            policy.source_burst_capacity,
267            now,
268        );
269    }
270
271    fn has_capacity(&self, policy: &DiscoveryPolicy, bytes: usize) -> bool {
272        (policy.max_responses_per_sec_per_source == u32::MAX || self.tokens >= 1.0)
273            && (policy.max_bytes_per_sec_per_source == usize::MAX
274                || self.byte_tokens >= bytes as f64)
275    }
276
277    fn deduct(&mut self, policy: &DiscoveryPolicy, bytes: usize) {
278        if policy.max_responses_per_sec_per_source != u32::MAX {
279            self.tokens -= 1.0;
280        }
281        if policy.max_bytes_per_sec_per_source != usize::MAX {
282            self.byte_tokens -= bytes as f64;
283        }
284    }
285}
286
287#[derive(Debug)]
288struct DiscoveryState {
289    global_tokens: f64,
290    global_byte_tokens: f64,
291    global_last_refill: Instant,
292    last_broadcast_who_is: Option<Instant>,
293    last_broadcast_who_has: HashMap<WhoHasTarget, Instant>,
294    negative_who_has: HashMap<WhoHasTarget, Instant>,
295    sources: HashMap<SourceKey, SourceState>,
296}
297
298impl DiscoveryState {
299    fn refill_global(&mut self, policy: &DiscoveryPolicy, now: Instant) {
300        refill_bucket(
301            &mut self.global_tokens,
302            &mut self.global_byte_tokens,
303            &mut self.global_last_refill,
304            policy.max_responses_per_sec_global,
305            policy.max_bytes_per_sec_global,
306            policy.global_burst_capacity,
307            now,
308        );
309    }
310
311    fn has_global_capacity(
312        &self,
313        policy: &DiscoveryPolicy,
314        is_reserved: bool,
315        bytes: usize,
316    ) -> bool {
317        let count_ok = policy.max_responses_per_sec_global == u32::MAX || {
318            let req = if is_reserved {
319                1.0
320            } else {
321                1.0 + policy.reserved_capacity as f64
322            };
323            self.global_tokens >= req
324        };
325        count_ok
326            && (policy.max_bytes_per_sec_global == usize::MAX
327                || self.global_byte_tokens >= bytes as f64)
328    }
329
330    fn deduct_global(&mut self, policy: &DiscoveryPolicy, bytes: usize) {
331        if policy.max_responses_per_sec_global != u32::MAX {
332            self.global_tokens -= 1.0;
333        }
334        if policy.max_bytes_per_sec_global != usize::MAX {
335            self.global_byte_tokens -= bytes as f64;
336        }
337    }
338
339    fn get_or_create_source_mut(
340        &mut self,
341        policy: &DiscoveryPolicy,
342        key: &SourceKey,
343        now: Instant,
344    ) -> &mut SourceState {
345        if !self.sources.contains_key(key) {
346            ensure_source_capacity(self, policy, now);
347        }
348        let src = self
349            .sources
350            .entry(key.clone())
351            .or_insert_with(|| SourceState {
352                tokens: policy.source_burst_capacity as f64,
353                byte_tokens: policy.max_bytes_per_sec_per_source as f64,
354                last_refill: now,
355                last_activity: now,
356                last_who_is: None,
357                last_who_has: HashMap::new(),
358            });
359        src.refill(policy, now);
360        src
361    }
362
363    fn is_coalesced_who_has(
364        &self,
365        policy: &DiscoveryPolicy,
366        target: &WhoHasTarget,
367        key: &SourceKey,
368        is_broadcast: bool,
369        now: Instant,
370    ) -> bool {
371        if policy.coalesce_window.is_zero() {
372            return false;
373        }
374        let is_recent = |ts: &Instant| now.saturating_duration_since(*ts) < policy.coalesce_window;
375        self.negative_who_has.get(target).is_some_and(is_recent)
376            || (is_broadcast
377                && self
378                    .last_broadcast_who_has
379                    .get(target)
380                    .is_some_and(is_recent))
381            || self
382                .sources
383                .get(key)
384                .and_then(|s| s.last_who_has.get(target))
385                .is_some_and(is_recent)
386    }
387}
388
389#[derive(Debug)]
390struct AtomicDeviceInstance(AtomicU64);
391
392impl AtomicDeviceInstance {
393    const PRESENT_BIT: u64 = 0x1_0000_0000;
394
395    fn new(opt: Option<u32>) -> Self {
396        Self(AtomicU64::new(
397            opt.map_or(0, |i| Self::PRESENT_BIT | (i as u64)),
398        ))
399    }
400
401    fn load(&self, order: Ordering) -> Option<u32> {
402        let val = self.0.load(order);
403        (val & Self::PRESENT_BIT != 0).then_some(val as u32)
404    }
405}
406
407#[derive(Debug, Clone, Copy, PartialEq, Eq)]
408pub(crate) enum PreCheckDecision {
409    Admit,
410    OutOfRange,
411    Coalesced,
412    ThrottledSource,
413    ThrottledGlobal,
414    DecodeError,
415}
416
417/// Thread-safe rate limiter, duplicate coalescer, and negative query cache.
418#[derive(Debug)]
419pub(crate) struct DiscoveryLimiter {
420    policy: DiscoveryPolicy,
421    counters: AtomicDiscoveryCounters,
422    device_instance: AtomicDeviceInstance,
423    state: Mutex<DiscoveryState>,
424}
425
426impl DiscoveryLimiter {
427    /// Every throttle and coalescing window here is measured on
428    /// [`runtime_clock::now`](super::runtime_clock::now), as its callers'
429    /// instants are.
430    pub(crate) fn new(policy: DiscoveryPolicy, device_instance: Option<u32>) -> Self {
431        let now = super::runtime_clock::now();
432        let state = DiscoveryState {
433            global_tokens: policy.global_burst_capacity as f64,
434            global_byte_tokens: policy.max_bytes_per_sec_global as f64,
435            global_last_refill: now,
436            last_broadcast_who_is: None,
437            last_broadcast_who_has: HashMap::new(),
438            negative_who_has: HashMap::new(),
439            sources: HashMap::new(),
440        };
441        Self {
442            policy,
443            counters: AtomicDiscoveryCounters::default(),
444            device_instance: AtomicDeviceInstance::new(device_instance),
445            state: Mutex::new(state),
446        }
447    }
448
449    pub(crate) fn counters(&self) -> DiscoveryCounters {
450        self.counters.snapshot()
451    }
452
453    #[allow(dead_code)]
454    pub(crate) fn policy(&self) -> &DiscoveryPolicy {
455        &self.policy
456    }
457
458    #[cfg(test)]
459    pub(crate) fn negative_who_has_count(&self) -> usize {
460        self.state.lock().unwrap().negative_who_has.len()
461    }
462
463    #[cfg(test)]
464    pub(crate) fn source_byte_tokens(
465        &self,
466        source_mac: &MacAddr,
467        source_network: Option<&NpduAddress>,
468    ) -> Option<f64> {
469        let key = SourceKey::from_parts(source_mac, source_network);
470        self.state
471            .lock()
472            .unwrap()
473            .sources
474            .get(&key)
475            .map(|s| s.byte_tokens)
476    }
477
478    fn check_instance_in_range(&self, range: Option<DeviceInstanceRange>) -> bool {
479        self.device_instance
480            .load(Ordering::Acquire)
481            .is_some_and(|instance| range.is_none_or(|range| range.contains(instance)))
482    }
483
484    pub(crate) fn pre_check_who_is(
485        &self,
486        req_bytes: &[u8],
487        received: &ReceivedApdu,
488        now: Instant,
489    ) -> PreCheckDecision {
490        // RB-07 compat mode: provenance threaded here for RB-09, no decision.
491        let _ = received.provenance;
492        let who_is = match WhoIsRequest::decode(req_bytes) {
493            Ok(w) => w,
494            Err(_) => {
495                self.counters.inc(&self.counters.malformed_dropped);
496                return PreCheckDecision::DecodeError;
497            }
498        };
499
500        self.counters.inc(&self.counters.who_is_received);
501
502        if !self.check_instance_in_range(who_is.range) {
503            return PreCheckDecision::OutOfRange;
504        }
505
506        let is_broadcast = received.is_group || received.link_layer_group;
507        let source_key = SourceKey::from_received(received);
508        let mut state = self.state.lock().unwrap();
509
510        // 1. Coalescing check
511        if !self.policy.coalesce_window.is_zero() {
512            let is_recent =
513                |ts: &Instant| now.saturating_duration_since(*ts) < self.policy.coalesce_window;
514            let bcast = is_broadcast && state.last_broadcast_who_is.as_ref().is_some_and(is_recent);
515            let src = state
516                .sources
517                .get(&source_key)
518                .and_then(|s| s.last_who_is.as_ref())
519                .is_some_and(is_recent);
520            if bcast || src {
521                self.counters.inc(&self.counters.requests_coalesced);
522                return PreCheckDecision::Coalesced;
523            }
524        }
525
526        // 2. Refill & Rate limits
527        state.refill_global(&self.policy, now);
528        let is_reserved = is_source_reserved(&self.policy, &source_key, received);
529        const ESTIMATED_I_AM_BYTES: usize = 64;
530
531        if !state.has_global_capacity(&self.policy, is_reserved, ESTIMATED_I_AM_BYTES) {
532            self.counters.inc(&self.counters.responses_throttled_global);
533            return PreCheckDecision::ThrottledGlobal;
534        }
535
536        let src = state.get_or_create_source_mut(&self.policy, &source_key, now);
537        if !src.has_capacity(&self.policy, ESTIMATED_I_AM_BYTES) {
538            self.counters.inc(&self.counters.responses_throttled_source);
539            return PreCheckDecision::ThrottledSource;
540        }
541
542        src.deduct(&self.policy, ESTIMATED_I_AM_BYTES);
543        src.last_activity = now;
544        src.last_who_is = Some(now);
545        state.deduct_global(&self.policy, ESTIMATED_I_AM_BYTES);
546        if is_broadcast && !self.policy.prefer_directed_responses {
547            state.last_broadcast_who_is = Some(now);
548        }
549
550        PreCheckDecision::Admit
551    }
552
553    pub(crate) fn record_i_am_sent(
554        &self,
555        actual_bytes: usize,
556        directed: bool,
557        source_mac: &MacAddr,
558        source_network: Option<&NpduAddress>,
559        now: Instant,
560    ) {
561        self.counters.record_sent(actual_bytes, directed, true);
562
563        const ESTIMATED_I_AM_BYTES: usize = 64;
564        let delta = (actual_bytes as f64) - (ESTIMATED_I_AM_BYTES as f64);
565        let mut state = self.state.lock().unwrap();
566        if delta != 0.0 {
567            if self.policy.max_bytes_per_sec_global != usize::MAX {
568                state.global_byte_tokens = (state.global_byte_tokens - delta).max(0.0);
569            }
570            let key = SourceKey::from_parts(source_mac, source_network);
571            if let Some(src) = state.sources.get_mut(&key)
572                && self.policy.max_bytes_per_sec_per_source != usize::MAX
573            {
574                src.byte_tokens = (src.byte_tokens - delta).max(0.0);
575            }
576        }
577        if !directed {
578            state.last_broadcast_who_is = Some(now);
579        }
580    }
581
582    pub(crate) fn pre_check_who_has(
583        &self,
584        req_bytes: &[u8],
585        received: &ReceivedApdu,
586        now: Instant,
587    ) -> PreCheckDecision {
588        // RB-07 compat mode: provenance threaded here for RB-09, no decision.
589        let _ = received.provenance;
590        let who_has = match WhoHasRequest::decode(req_bytes) {
591            Ok(w) => w,
592            Err(_) => {
593                self.counters.inc(&self.counters.malformed_dropped);
594                return PreCheckDecision::DecodeError;
595            }
596        };
597
598        self.counters.inc(&self.counters.who_has_received);
599
600        if !self.check_instance_in_range(who_has.range) {
601            return PreCheckDecision::OutOfRange;
602        }
603
604        let target = match who_has.object {
605            WhoHasObject::Identifier(oid) => WhoHasTarget::Id(oid),
606            WhoHasObject::Name(name) => WhoHasTarget::Name(name),
607        };
608
609        let is_broadcast = received.is_group || received.link_layer_group;
610        let source_key = SourceKey::from_received(received);
611        let mut state = self.state.lock().unwrap();
612
613        if state.is_coalesced_who_has(&self.policy, &target, &source_key, is_broadcast, now) {
614            self.counters.inc(&self.counters.requests_coalesced);
615            return PreCheckDecision::Coalesced;
616        }
617
618        state.refill_global(&self.policy, now);
619        let is_reserved = is_source_reserved(&self.policy, &source_key, received);
620        const ESTIMATED_I_HAVE_BYTES: usize = 64;
621
622        if !state.has_global_capacity(&self.policy, is_reserved, ESTIMATED_I_HAVE_BYTES) {
623            self.counters.inc(&self.counters.responses_throttled_global);
624            return PreCheckDecision::ThrottledGlobal;
625        }
626
627        let src = state.get_or_create_source_mut(&self.policy, &source_key, now);
628        if !src.has_capacity(&self.policy, ESTIMATED_I_HAVE_BYTES) {
629            self.counters.inc(&self.counters.responses_throttled_source);
630            return PreCheckDecision::ThrottledSource;
631        }
632
633        src.last_activity = now;
634        PreCheckDecision::Admit
635    }
636
637    pub(crate) fn is_negative_who_has(&self, target: &WhoHasTarget, now: Instant) -> bool {
638        if self.policy.coalesce_window.is_zero() {
639            return false;
640        }
641        let state = self.state.lock().unwrap();
642        if let Some(ts) = state.negative_who_has.get(target) {
643            now.saturating_duration_since(*ts) < self.policy.coalesce_window
644        } else {
645            false
646        }
647    }
648
649    pub(crate) fn record_negative_who_has(&self, target: WhoHasTarget, now: Instant) {
650        const MAX_NEGATIVE_WHO_HAS: usize = 512;
651        let mut state = self.state.lock().unwrap();
652        if state.negative_who_has.len() >= MAX_NEGATIVE_WHO_HAS {
653            let window = self.policy.coalesce_window;
654            state
655                .negative_who_has
656                .retain(|_, ts| now.saturating_duration_since(*ts) < window);
657        }
658        while state.negative_who_has.len() >= MAX_NEGATIVE_WHO_HAS {
659            let Some(oldest) = state
660                .negative_who_has
661                .iter()
662                .min_by_key(|(_, ts)| *ts)
663                .map(|(k, _)| k.clone())
664            else {
665                break;
666            };
667            state.negative_who_has.remove(&oldest);
668        }
669        state.negative_who_has.insert(target, now);
670    }
671
672    pub(crate) fn try_consume_who_has_response(
673        &self,
674        target: &WhoHasTarget,
675        received: &ReceivedApdu,
676        response_bytes: usize,
677        now: Instant,
678    ) -> bool {
679        let source_key = SourceKey::from_received(received);
680        let mut state = self.state.lock().unwrap();
681        let is_broadcast = received.is_group || received.link_layer_group;
682
683        if state.is_coalesced_who_has(&self.policy, target, &source_key, is_broadcast, now) {
684            self.counters.inc(&self.counters.requests_coalesced);
685            return false;
686        }
687
688        state.refill_global(&self.policy, now);
689        let is_reserved = is_source_reserved(&self.policy, &source_key, received);
690
691        if !state.has_global_capacity(&self.policy, is_reserved, response_bytes) {
692            self.counters.inc(&self.counters.responses_throttled_global);
693            return false;
694        }
695
696        let src = state.get_or_create_source_mut(&self.policy, &source_key, now);
697        if !src.has_capacity(&self.policy, response_bytes) {
698            self.counters.inc(&self.counters.responses_throttled_source);
699            return false;
700        }
701
702        src.deduct(&self.policy, response_bytes);
703        src.last_activity = now;
704        src.last_who_has.insert(target.clone(), now);
705        state.deduct_global(&self.policy, response_bytes);
706        if is_broadcast && !self.policy.prefer_directed_responses {
707            state.last_broadcast_who_has.insert(target.clone(), now);
708        }
709
710        true
711    }
712
713    pub(crate) fn record_i_have_sent(
714        &self,
715        actual_bytes: usize,
716        directed: bool,
717        target: &WhoHasTarget,
718        _source_mac: &MacAddr,
719        _source_network: Option<&NpduAddress>,
720        now: Instant,
721    ) {
722        self.counters.record_sent(actual_bytes, directed, false);
723
724        if !directed {
725            let mut state = self.state.lock().unwrap();
726            state.last_broadcast_who_has.insert(target.clone(), now);
727        }
728    }
729}
730
731fn is_source_reserved(policy: &DiscoveryPolicy, key: &SourceKey, recv: &ReceivedApdu) -> bool {
732    policy
733        .reserved_sources
734        .iter()
735        .any(|r| r == &key.mac || r == &recv.source_mac)
736}
737
738fn ensure_source_capacity(state: &mut DiscoveryState, policy: &DiscoveryPolicy, now: Instant) {
739    if state.sources.len() < policy.max_tracked_sources {
740        return;
741    }
742    state.sources.retain(|_, src| {
743        now.saturating_duration_since(src.last_activity) < Duration::from_secs(10)
744    });
745    while state.sources.len() >= policy.max_tracked_sources {
746        let Some(key) = state
747            .sources
748            .iter()
749            .min_by_key(|(_, src)| src.last_activity)
750            .map(|(k, _)| k.clone())
751        else {
752            break;
753        };
754        state.sources.remove(&key);
755    }
756}
757
758/// Broadcast this device's I-Am unless DeviceCommunicationControl restricts
759/// initiation. This call announces independently of any received Who-Is;
760/// under DISABLE_INITIATION it sends nothing and fails with
761/// `SERVICES` / `COMMUNICATION_DISABLED`, the code Clause 18.6 keeps for local
762/// work that could not initiate a service for that reason.
763pub(crate) async fn broadcast_i_am_from<T: TransportPort + 'static>(
764    config: &ServerConfig,
765    db: &Arc<RwLock<ObjectDatabase>>,
766    network: &Arc<NetworkLayer<T>>,
767    comm_state: &CommState,
768    limiter: Option<&Arc<DiscoveryLimiter>>,
769) -> Result<(), Error> {
770    let guard = db.read().await;
771    crate::local_device::announcement_device(&guard, &config.device)
772        .ok_or_else(|| Error::Encoding("no Device object in database".into()))?;
773
774    // Both discovery paths encode the same frozen, resolved declarations.
775    let mut service_buf = BytesMut::new();
776    iam_request_for(config).encode(&mut service_buf);
777
778    let mut buf = BytesMut::new();
779    encode_apdu(
780        &mut buf,
781        &Apdu::UnconfirmedRequest(UnconfirmedRequestPdu {
782            service_choice: UnconfirmedServiceChoice::I_AM,
783            service_request: service_buf.freeze(),
784        }),
785    )?;
786
787    drop(guard);
788    if comm_state.initiation_restricted() {
789        return Err(Error::protocol(
790            ErrorClass::SERVICES,
791            ErrorCode::COMMUNICATION_DISABLED,
792        ));
793    }
794    if let Some(limiter) = limiter {
795        limiter.record_i_am_sent(
796            buf.len(),
797            false,
798            &MacAddr::new(),
799            None,
800            super::runtime_clock::now(),
801        );
802    }
803
804    network
805        .broadcast_apdu(&buf, false, NetworkPriority::Normal)
806        .await
807}
808
809/// I-Am from the server's frozen, resolved Device declarations.
810#[doc(hidden)]
811pub fn iam_request_for(config: &ServerConfig) -> IAmRequest {
812    IAmRequest {
813        object_identifier: config.device.device_oid(),
814        max_apdu_length: u32::from(config.device.max_apdu_length()),
815        segmentation_supported: config.device.segmentation(),
816        vendor_id: config.device.vendor_id(),
817    }
818}
819
820impl<T: TransportPort + 'static> BACnetServer<T> {
821    /// Broadcast I-Am through this server's bounded local-send owner.
822    /// Cancelling the waiter leaves admitted work owned until completion or stop.
823    /// While DeviceCommunicationControl restricts initiation nothing is sent
824    /// and this fails with `SERVICES` / `COMMUNICATION_DISABLED` (Clause 16.1).
825    pub async fn broadcast_i_am(&self) -> Result<(), Error> {
826        self.broadcaster
827            .send(Some(Arc::clone(&self.discovery_limiter)))
828            .await
829    }
830}
bacnet_server/server/discovery.rs

bacnet_server/server/
discovery.rs
