bacnet_server/server/
mod.rs

1//! BACnetServer: builder, APDU dispatch, and lifecycle management.
2//!
3//! The server wraps a NetworkLayer behind Arc (shared with the dispatch task),
4//! owns an ObjectDatabase via Arc<Mutex>, and spawns a dispatch task that
5//! routes incoming APDUs to service handlers.
6
7use std::collections::HashMap;
8use std::net::Ipv4Addr;
9use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU8, Ordering};
10use std::sync::Arc;
11use std::time::Instant;
12
13use bytes::{Bytes, BytesMut};
14use tokio::sync::{mpsc, oneshot, watch, Mutex, RwLock, Semaphore};
15use tokio::task::JoinHandle;
16use tokio::time::Duration;
17use tracing::{debug, warn};
18
19use bacnet_encoding::apdu::{
20    self, encode_apdu, validate_max_apdu_length, AbortPdu, Apdu, ComplexAck,
21    ConfirmedRequest as ConfirmedRequestPdu, ErrorPdu, RejectPdu, SegmentAck as SegmentAckPdu,
22    SimpleAck, UnconfirmedRequest as UnconfirmedRequestPdu,
23};
24use bacnet_encoding::npdu::NpduAddress;
25use bacnet_encoding::primitives::encode_property_value;
26use bacnet_encoding::segmentation::{
27    duplicate_in_window, max_segment_payload, split_payload, SegmentReceiver, SegmentedPduType,
28};
29use bacnet_network::layer::NetworkLayer;
30use bacnet_objects::database::ObjectDatabase;
31use bacnet_objects::notification_class::{
32    lookup_notification_recipients, resolve_transition_priority_ack,
33};
34use bacnet_services::alarm_event::EventNotificationRequest;
35use bacnet_services::common::BACnetPropertyValue;
36use bacnet_services::cov::COVNotificationRequest;
37use bacnet_services::cov_multiple::{
38    COVNotificationItem, COVNotificationMultipleRequest, COVNotificationValue,
39};
40use bacnet_services::who_is::{IAmRequest, WhoIsRequest};
41use bacnet_transport::bip::BipTransport;
42use bacnet_transport::port::TransportPort;
43use bacnet_types::enums::{
44    AbortReason, ConfirmedServiceChoice, ErrorClass, ErrorCode, LifeSafetyOperation,
45    NetworkPriority, NotifyType, ObjectType, PropertyIdentifier, RejectReason, Segmentation,
46    UnconfirmedServiceChoice,
47};
48use bacnet_types::error::Error;
49use bacnet_types::primitives::{ObjectIdentifier, PropertyValue};
50use bacnet_types::MacAddr;
51
52use crate::audit_notification::{
53    AuditNotificationAuthorizationContext, AuditNotificationAuthorizer,
54    UnconfirmedAuditNotificationAuthorizationContext, UnconfirmedAuditNotificationAuthorizer,
55    MAX_AUDIT_NOTIFICATIONS, MAX_AUDIT_NOTIFICATION_BYTES,
56};
57use crate::cov::{CovNotificationKind, CovSubscription, CovSubscriptionTable};
58use crate::handlers;
59use crate::life_safety::{LifeSafetyOperationAuthorizationContext, LifeSafetyOperationAuthorizer};
60use confirmed_request_tracker::{ConfirmedRequestAdmission, ConfirmedRequestTracker};
61pub use device_bindings::DeviceBinding;
62use device_bindings::{register_configured_binding, DeviceBindingTable};
63use notification_transactions::{
64    canonical_direct_peer, canonical_routed_peer, run_notification_worker,
65    NotificationTransactions, NotificationWorkerResult,
66};
67
68/// Maximum number of concurrent segmented reassembly sessions.
69const MAX_SEG_RECEIVERS: usize = 128;
70
71/// Hard per-request reassembly ceiling: the sequence-number space (#364).
72///
73/// A local storage bound, not a protocol one. Clause 20.1.2.7 makes the
74/// request sequence number modulo 256, so a longer request is entirely
75/// representable on the wire — this server simply keys its segment store by
76/// that `u8` and cannot tell segment 256 from segment 0. A Device object
77/// configured to receive segments does publish a tighter advertisement — its
78/// `Max_Segments_Accepted` (Clause 12.11) defaults to `Unsigned(65)` — but
79/// enforcing it here is deliberately not done: accepting more segments than
80/// advertised is permissive, not a violation, while the sequence space is the
81/// line past which acceptance silently corrupts. Exactly 256 segments
82/// reassemble correctly and must keep working; 257 is the first that would
83/// corrupt the payload.
84const MAX_REQUEST_SEGMENTS: usize = 256;
85
86/// Maximum number of concurrent segmented response send sessions.
87const MAX_SEG_SENDERS: usize = 128;
88
89/// Timeout for idle segmented reassembly sessions.
90const SEG_RECEIVER_TIMEOUT: Duration = Duration::from_secs(4);
91
92/// Maximum negative SegmentAck retries during segmented response send.
93const MAX_NEG_SEGMENT_ACK_RETRIES: u8 = 3;
94
95/// Default timeout while waiting for SegmentACK during segmented response send.
96const DEFAULT_APDU_SEGMENT_TIMEOUT: Duration = Duration::from_secs(5);
97
98/// Default retransmission budget for segmented response segments.
99const DEFAULT_APDU_SEGMENT_RETRIES: u8 = MAX_NEG_SEGMENT_ACK_RETRIES;
100
101/// Default number of APDU retries for confirmed COV notifications.
102const DEFAULT_APDU_RETRIES: u8 = 3;
103
104type TsmPeer = (MacAddr, Option<NpduAddress>);
105type TsmKey = (MacAddr, Option<NpduAddress>, u8);
106
107// ---------------------------------------------------------------------------
108// Server-side Transaction State Machine (TSM) for outgoing confirmed requests
109// ---------------------------------------------------------------------------
110
111/// Result of a confirmed COV notification from the subscriber's perspective.
112#[derive(Debug, Clone, Copy, PartialEq, Eq)]
113pub enum CovAckResult {
114    /// SimpleAck received — subscriber accepted the notification.
115    Ack,
116    /// Error or Reject/Abort received — subscriber rejected the notification.
117    Error,
118}
119
120/// Legacy server transaction state and learned-router cache.
121///
122/// The allocation and pending-result methods remain available to existing
123/// server internals and tests. Standalone confirmed notification paths use the
124/// private endpoint-core adapter instead.
125pub struct ServerTsm {
126    #[allow(dead_code)]
127    next_invoke_id: u8,
128    /// Oneshot senders keyed by peer MAC and invoke ID. When a result arrives
129    /// from the dispatch loop, we send it directly — no polling needed.
130    #[allow(dead_code)]
131    pending: HashMap<TsmKey, oneshot::Sender<CovAckResult>>,
132    /// Router MACs learned per remote network, Clause 6.5.3 method 4: "using
133    /// the local broadcast MAC address in the initial transmission to a device
134    /// on a remote DNET and noting the SA associated with any subsequent
135    /// responses from the remote device" (#375). Consulted so later confirmed
136    /// sends to that DNET can unicast to the router instead of broadcasting.
137    routers: HashMap<u16, MacAddr>,
138}
139
140/// Cap on learned router entries; a full cache just means later networks keep
141/// using the (always-correct) broadcast form of Clause 6.5.3.
142const MAX_LEARNED_ROUTERS: usize = 64;
143
144impl ServerTsm {
145    fn new() -> Self {
146        Self {
147            next_invoke_id: 0,
148            pending: HashMap::new(),
149            routers: HashMap::new(),
150        }
151    }
152
153    /// Allocate the next invoke ID and register a oneshot channel for the result.
154    /// Returns (invoke_id, receiver).
155    #[allow(dead_code)]
156    fn allocate(&mut self, peer: TsmPeer) -> Option<(u8, oneshot::Receiver<CovAckResult>)> {
157        for offset in 0..=u8::MAX {
158            let id = self.next_invoke_id.wrapping_add(offset);
159            if !self
160                .pending
161                .contains_key(&(peer.0.clone(), peer.1.clone(), id))
162            {
163                self.next_invoke_id = id.wrapping_add(1);
164                let rx = self.register(peer, id);
165                return Some((id, rx));
166            }
167        }
168        None
169    }
170
171    /// Register or replace the pending receiver for a peer/invoke-id pair.
172    #[allow(dead_code)]
173    fn register(&mut self, peer: TsmPeer, invoke_id: u8) -> oneshot::Receiver<CovAckResult> {
174        let (tx, rx) = oneshot::channel();
175        self.pending.insert((peer.0, peer.1, invoke_id), tx);
176        rx
177    }
178
179    /// Record a result from the dispatch loop (SimpleAck, Error, etc.).
180    /// Sends immediately through the oneshot channel.
181    #[allow(dead_code)]
182    fn record_result(
183        &mut self,
184        peer: &MacAddr,
185        network: Option<&NpduAddress>,
186        invoke_id: u8,
187        result: CovAckResult,
188    ) -> bool {
189        if let Some(tx) = self
190            .pending
191            .remove(&(peer.clone(), network.cloned(), invoke_id))
192        {
193            let _ = tx.send(result);
194            true
195        } else {
196            false
197        }
198    }
199
200    /// Remove a pending entry (cleanup on completion or exhaustion).
201    #[allow(dead_code)]
202    fn remove(&mut self, peer: &TsmPeer, invoke_id: u8) {
203        self.pending
204            .remove(&(peer.0.clone(), peer.1.clone(), invoke_id));
205    }
206
207    /// Correlate an inbound response with the transaction awaiting it (#375).
208    ///
209    /// Three key shapes are tried, most specific first:
210    /// 1. exactly as the sender registered it — the immediate MAC plus any
211    ///    routed identity;
212    /// 2. the router-unknown form — an empty local half with the routed
213    ///    identity, used when the request went out via the Clause 6.5.3
214    ///    broadcast DA and the delivering router's MAC was unknowable at
215    ///    registration;
216    /// 3. the legacy wildcard `(empty, None)`, which nothing registers today
217    ///    but which older callers may still expect.
218    ///
219    /// A hit that carries a routed identity also teaches the router cache:
220    /// the response's immediate MAC is "the SA associated with [a] subsequent
221    /// response from the remote device" (Clause 6.5.3 method 4).
222    #[allow(dead_code)]
223    fn record_result_correlated(
224        &mut self,
225        source_mac: &MacAddr,
226        source_network: Option<&NpduAddress>,
227        invoke_id: u8,
228        result: CovAckResult,
229    ) -> bool {
230        let hit = self.record_result(source_mac, source_network, invoke_id, result)
231            || (source_network.is_some()
232                && self.record_result(&MacAddr::new(), source_network, invoke_id, result))
233            || self.record_result(&MacAddr::new(), None, invoke_id, result);
234        if hit {
235            if let Some(address) = source_network {
236                self.learn_router(address.network, source_mac);
237            }
238        }
239        hit
240    }
241
242    /// Cache `router` as the way to reach `network`, bounded by
243    /// [`MAX_LEARNED_ROUTERS`].
244    fn learn_router(&mut self, network: u16, router: &MacAddr) {
245        if router.is_empty() {
246            return;
247        }
248        if self.routers.len() >= MAX_LEARNED_ROUTERS && !self.routers.contains_key(&network) {
249            return;
250        }
251        self.routers.insert(network, router.clone());
252    }
253
254    /// The learned router MAC for `network`, if any.
255    fn cached_router(&self, network: u16) -> Option<MacAddr> {
256        self.routers.get(&network).cloned()
257    }
258}
259
260/// Data from a TimeSynchronization request.
261#[derive(Debug, Clone)]
262pub struct TimeSyncData {
263    /// Raw service request bytes (caller can decode if needed).
264    pub raw_service_data: Bytes,
265    /// Whether this was a UTC time sync (vs. local).
266    pub is_utc: bool,
267}
268
269/// Server configuration.
270#[derive(Clone)]
271pub struct ServerConfig {
272    /// Local interface to bind.
273    pub interface: Ipv4Addr,
274    /// UDP port (default 0xBAC0 = 47808).
275    pub port: u16,
276    /// Directed broadcast address.
277    pub broadcast_address: Ipv4Addr,
278    /// Maximum APDU length accepted.
279    pub max_apdu_length: u32,
280    /// Segmentation support level.
281    ///
282    /// Enforced, not just advertised: the dispatch loop reassembles inbound
283    /// segmented requests only under `BOTH`/`RECEIVE` and transmits
284    /// segmented responses only under `BOTH`/`TRANSMIT` (Clauses 5.4.5.1 and
285    /// 5.4.5.3); anything else draws a SEGMENTATION_NOT_SUPPORTED Abort. The
286    /// default is `NONE`, so a default-configured server refuses segmented
287    /// traffic in both directions — set this to what the device should
288    /// actually honor.
289    pub segmentation_supported: Segmentation,
290    /// Vendor identifier.
291    pub vendor_id: u16,
292    /// Timeout in ms before retrying a failed confirmed COV notification send (default 3000ms).
293    pub cov_retry_timeout_ms: u64,
294    /// Optional observer invoked after a time-synchronization request is accepted.
295    pub on_time_sync: Option<Arc<dyn Fn(TimeSyncData) + Send + Sync>>,
296    /// Optional LifeSafetyOperation authorization policy.
297    ///
298    /// Absence is fail-closed: requests receive SERVICES /
299    /// SERVICE_REQUEST_DENIED before object mutation.
300    pub life_safety_operation_authorizer: Option<LifeSafetyOperationAuthorizer>,
301    /// Exactly one explicitly configured Audit Log notification sink.
302    ///
303    /// Absence is fail-closed; the server never selects a sink by database
304    /// iteration order.
305    pub audit_notification_sink: Option<ObjectIdentifier>,
306    /// Optional fast, nonblocking ConfirmedAuditNotification authorizer.
307    ///
308    /// Absence, `false`, or a panic denies the request before mutation.
309    pub audit_notification_authorizer: Option<AuditNotificationAuthorizer>,
310    /// Optional fast, nonblocking UnconfirmedAuditNotification authorizer.
311    ///
312    /// Absence, `false`, or a panic silently denies the request before mutation.
313    pub unconfirmed_audit_notification_authorizer: Option<UnconfirmedAuditNotificationAuthorizer>,
314    /// Optional password required for DeviceCommunicationControl.
315    pub dcc_password: Option<String>,
316    /// Optional password required for ReinitializeDevice.
317    pub reinit_password: Option<String>,
318    /// Enable periodic fault detection / reliability evaluation.
319    /// When true, the server invokes every object's opt-in, object-owned
320    /// reliability evaluation hook every 10 seconds. Stock objects currently
321    /// inherit the no-op default.
322    ///
323    /// This governs reliability evaluation only. Event Enrollment evaluation
324    /// is configured separately via [`enable_event_enrollment`](Self::enable_event_enrollment).
325    pub enable_fault_detection: bool,
326    /// Enable periodic Event Enrollment evaluation (default `true`).
327    ///
328    /// When true, the server re-reads the property each Event Enrollment object
329    /// names in its `Object_Property_Reference` and applies the configured event
330    /// algorithm. Startup: the task is spawned by [`start`](BACnetServer::start)
331    /// and its first pass runs immediately, then once per interval. Shutdown:
332    /// [`stop`](BACnetServer::stop) aborts it and awaits the abort.
333    ///
334    /// This switch governs the evaluation task; it is not the per-object
335    /// `Event_Detection_Enable` property of ASHRAE 135-2020 Clause 13.2.2.1.
336    /// Setting it false stops evaluation without performing the reset that
337    /// clause requires of a disabled detector (`Event_State` to NORMAL, with the
338    /// corresponding timestamp and acknowledgment state), so a device carrying
339    /// active enrollments will hold whatever state it last detected.
340    ///
341    /// Evaluation is a no-op on databases holding no Event Enrollment objects,
342    /// so the default is on.
343    ///
344    /// Successful enabled transitions commit `Event_State`,
345    /// `Acked_Transitions`, and `Event_Time_Stamps` atomically and are then
346    /// routed through the shared EventNotification sender. Event Enrollment
347    /// message text remains intentionally absent, and exact event-specific
348    /// notification values are deferred to the payload projection work.
349    pub enable_event_enrollment: bool,
350    /// Interval in seconds between Event Enrollment evaluation passes (default 10).
351    ///
352    /// This is a sampling cadence with no basis in ASHRAE 135-2020, which
353    /// prescribes no evaluation frequency and leaves acquisition of a monitored
354    /// value a local matter (Clause 12.12). It is not the `Time_Delay` of an
355    /// event algorithm, which is how long a condition must persist before a
356    /// transition is indicated (Clause 13.3) — a coarse interval delays
357    /// detection and can miss a condition that both appears and clears between
358    /// two passes.
359    ///
360    /// A value of `0` is clamped to one second. Ignored when
361    /// [`enable_event_enrollment`](Self::enable_event_enrollment) is false.
362    pub event_enrollment_interval_secs: u64,
363}
364
365impl std::fmt::Debug for ServerConfig {
366    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
367        f.debug_struct("ServerConfig")
368            .field("interface", &self.interface)
369            .field("port", &self.port)
370            .field("broadcast_address", &self.broadcast_address)
371            .field("max_apdu_length", &self.max_apdu_length)
372            .field("segmentation_supported", &self.segmentation_supported)
373            .field("vendor_id", &self.vendor_id)
374            .field("cov_retry_timeout_ms", &self.cov_retry_timeout_ms)
375            .field(
376                "on_time_sync",
377                &self.on_time_sync.as_ref().map(|_| "<callback>"),
378            )
379            .field(
380                "life_safety_operation_authorizer",
381                &self
382                    .life_safety_operation_authorizer
383                    .as_ref()
384                    .map(|_| "<callback>"),
385            )
386            .field("audit_notification_sink", &self.audit_notification_sink)
387            .field(
388                "audit_notification_authorizer",
389                &self
390                    .audit_notification_authorizer
391                    .as_ref()
392                    .map(|_| "<callback>"),
393            )
394            .field(
395                "unconfirmed_audit_notification_authorizer",
396                &self
397                    .unconfirmed_audit_notification_authorizer
398                    .as_ref()
399                    .map(|_| "<callback>"),
400            )
401            .field("dcc_password", &self.dcc_password.as_ref().map(|_| "***"))
402            .field(
403                "reinit_password",
404                &self.reinit_password.as_ref().map(|_| "***"),
405            )
406            .field("enable_fault_detection", &self.enable_fault_detection)
407            .field("enable_event_enrollment", &self.enable_event_enrollment)
408            .field(
409                "event_enrollment_interval_secs",
410                &self.event_enrollment_interval_secs,
411            )
412            .finish()
413    }
414}
415
416impl Default for ServerConfig {
417    fn default() -> Self {
418        Self {
419            interface: Ipv4Addr::UNSPECIFIED,
420            port: 0xBAC0,
421            broadcast_address: Ipv4Addr::BROADCAST,
422            max_apdu_length: 1476,
423            segmentation_supported: Segmentation::NONE,
424            vendor_id: 0,
425            cov_retry_timeout_ms: 3000,
426            on_time_sync: None,
427            life_safety_operation_authorizer: None,
428            audit_notification_sink: None,
429            audit_notification_authorizer: None,
430            unconfirmed_audit_notification_authorizer: None,
431            dcc_password: None,
432            reinit_password: None,
433            enable_fault_detection: false,
434            enable_event_enrollment: true,
435            event_enrollment_interval_secs: 10,
436        }
437    }
438}
439
440/// Generic builder for BACnetServer with a pre-built transport.
441pub struct ServerBuilder<T: TransportPort> {
442    config: ServerConfig,
443    db: ObjectDatabase,
444    transport: Option<T>,
445    configured_device_bindings: Vec<DeviceBinding>,
446}
447
448impl<T: TransportPort + 'static> ServerBuilder<T> {
449    /// Set the object database (transfers ownership).
450    pub fn database(mut self, db: ObjectDatabase) -> Self {
451        self.db = db;
452        self
453    }
454
455    /// Set the pre-built transport.
456    pub fn transport(mut self, transport: T) -> Self {
457        self.transport = Some(transport);
458        self
459    }
460
461    /// Register one explicit unicast route for a Device recipient.
462    pub fn device_binding(mut self, binding: DeviceBinding) -> Result<Self, Error> {
463        register_configured_binding(&mut self.configured_device_bindings, binding)?;
464        Ok(self)
465    }
466
467    /// Set the password required for DeviceCommunicationControl requests.
468    pub fn dcc_password(mut self, password: impl Into<String>) -> Self {
469        self.config.dcc_password = Some(password.into());
470        self
471    }
472
473    /// Set the password required for ReinitializeDevice requests.
474    pub fn reinit_password(mut self, password: impl Into<String>) -> Self {
475        self.config.reinit_password = Some(password.into());
476        self
477    }
478
479    /// Set the policy that authorizes inbound LifeSafetyOperation requests.
480    pub fn life_safety_operation_authorizer<F>(mut self, authorizer: F) -> Self
481    where
482        F: Fn(&LifeSafetyOperationAuthorizationContext) -> bool + Send + Sync + 'static,
483    {
484        self.config.life_safety_operation_authorizer = Some(Arc::new(authorizer));
485        self
486    }
487
488    /// Select the only local Audit Log that receives authorized notifications.
489    pub fn audit_notification_sink(mut self, sink: ObjectIdentifier) -> Self {
490        self.config.audit_notification_sink = Some(sink);
491        self
492    }
493
494    /// Set the fail-closed ConfirmedAuditNotification authorization policy.
495    pub fn audit_notification_authorizer<F>(mut self, authorizer: F) -> Self
496    where
497        F: Fn(&AuditNotificationAuthorizationContext) -> bool + Send + Sync + 'static,
498    {
499        self.config.audit_notification_authorizer = Some(Arc::new(authorizer));
500        self
501    }
502
503    /// Set the fail-closed UnconfirmedAuditNotification authorization policy.
504    pub fn unconfirmed_audit_notification_authorizer<F>(mut self, authorizer: F) -> Self
505    where
506        F: Fn(&UnconfirmedAuditNotificationAuthorizationContext) -> bool + Send + Sync + 'static,
507    {
508        self.config.unconfirmed_audit_notification_authorizer = Some(Arc::new(authorizer));
509        self
510    }
511
512    /// Enable periodic fault detection / reliability evaluation.
513    ///
514    /// When enabled, every object's opt-in reliability hook runs every 10
515    /// seconds; the default hook is a no-op.
516    ///
517    /// Reliability evaluation only; Event Enrollment evaluation is configured
518    /// by [`enable_event_enrollment`](Self::enable_event_enrollment).
519    pub fn enable_fault_detection(mut self, enabled: bool) -> Self {
520        self.config.enable_fault_detection = enabled;
521        self
522    }
523
524    /// Enable periodic Event Enrollment evaluation (default `true`).
525    pub fn enable_event_enrollment(mut self, enabled: bool) -> Self {
526        self.config.enable_event_enrollment = enabled;
527        self
528    }
529
530    /// Set the interval in seconds between Event Enrollment evaluation passes
531    /// (default 10).
532    pub fn event_enrollment_interval_secs(mut self, secs: u64) -> Self {
533        self.config.event_enrollment_interval_secs = secs;
534        self
535    }
536
537    /// Set the segmentation support this device advertises and enforces.
538    ///
539    /// The dispatch loop honors the advertisement (Clause 5.4.5.1): inbound
540    /// segmented requests are reassembled only under `BOTH` or `RECEIVE`, and
541    /// draw a SEGMENTATION_NOT_SUPPORTED Abort otherwise. The default is
542    /// `NONE`.
543    pub fn segmentation_supported(mut self, segmentation: Segmentation) -> Self {
544        self.config.segmentation_supported = segmentation;
545        self
546    }
547
548    /// Set the vendor identifier (used in IAm responses and protocol operations).
549    pub fn vendor_id(mut self, id: u16) -> Self {
550        self.config.vendor_id = id;
551        self
552    }
553
554    /// Build and start the server.
555    pub async fn build(self) -> Result<BACnetServer<T>, Error> {
556        let transport = self
557            .transport
558            .ok_or_else(|| Error::Encoding("transport not set on ServerBuilder".into()))?;
559        BACnetServer::start_with_clock_mode_and_bindings(
560            self.config,
561            self.db,
562            transport,
563            Some(ClockConfig::default()),
564            self.configured_device_bindings,
565        )
566        .await
567    }
568}
569
570/// BIP-specific builder that constructs `BipTransport` from interface/port/broadcast fields.
571pub struct BipServerBuilder {
572    config: ServerConfig,
573    db: ObjectDatabase,
574    configured_device_bindings: Vec<DeviceBinding>,
575}
576
577impl BipServerBuilder {
578    /// Set the local interface IP.
579    pub fn interface(mut self, ip: Ipv4Addr) -> Self {
580        self.config.interface = ip;
581        self
582    }
583
584    /// Set the UDP port.
585    pub fn port(mut self, port: u16) -> Self {
586        self.config.port = port;
587        self
588    }
589
590    /// Set the directed broadcast address.
591    pub fn broadcast_address(mut self, addr: Ipv4Addr) -> Self {
592        self.config.broadcast_address = addr;
593        self
594    }
595
596    /// Set the object database (transfers ownership).
597    pub fn database(mut self, db: ObjectDatabase) -> Self {
598        self.db = db;
599        self
600    }
601
602    /// Register one explicit unicast route for a Device recipient.
603    pub fn device_binding(mut self, binding: DeviceBinding) -> Result<Self, Error> {
604        register_configured_binding(&mut self.configured_device_bindings, binding)?;
605        Ok(self)
606    }
607
608    /// Set the password required for DeviceCommunicationControl requests.
609    pub fn dcc_password(mut self, password: impl Into<String>) -> Self {
610        self.config.dcc_password = Some(password.into());
611        self
612    }
613
614    /// Set the password required for ReinitializeDevice requests.
615    pub fn reinit_password(mut self, password: impl Into<String>) -> Self {
616        self.config.reinit_password = Some(password.into());
617        self
618    }
619
620    /// Set the policy that authorizes inbound LifeSafetyOperation requests.
621    pub fn life_safety_operation_authorizer<F>(mut self, authorizer: F) -> Self
622    where
623        F: Fn(&LifeSafetyOperationAuthorizationContext) -> bool + Send + Sync + 'static,
624    {
625        self.config.life_safety_operation_authorizer = Some(Arc::new(authorizer));
626        self
627    }
628
629    /// Select the only local Audit Log that receives authorized notifications.
630    pub fn audit_notification_sink(mut self, sink: ObjectIdentifier) -> Self {
631        self.config.audit_notification_sink = Some(sink);
632        self
633    }
634
635    /// Set the fail-closed ConfirmedAuditNotification authorization policy.
636    pub fn audit_notification_authorizer<F>(mut self, authorizer: F) -> Self
637    where
638        F: Fn(&AuditNotificationAuthorizationContext) -> bool + Send + Sync + 'static,
639    {
640        self.config.audit_notification_authorizer = Some(Arc::new(authorizer));
641        self
642    }
643
644    /// Set the fail-closed UnconfirmedAuditNotification authorization policy.
645    pub fn unconfirmed_audit_notification_authorizer<F>(mut self, authorizer: F) -> Self
646    where
647        F: Fn(&UnconfirmedAuditNotificationAuthorizationContext) -> bool + Send + Sync + 'static,
648    {
649        self.config.unconfirmed_audit_notification_authorizer = Some(Arc::new(authorizer));
650        self
651    }
652
653    /// Enable periodic fault detection / reliability evaluation.
654    ///
655    /// When enabled, every object's opt-in reliability hook runs every 10
656    /// seconds; the default hook is a no-op.
657    ///
658    /// Reliability evaluation only; Event Enrollment evaluation is configured
659    /// by [`enable_event_enrollment`](Self::enable_event_enrollment).
660    pub fn enable_fault_detection(mut self, enabled: bool) -> Self {
661        self.config.enable_fault_detection = enabled;
662        self
663    }
664
665    /// Enable periodic Event Enrollment evaluation (default `true`).
666    pub fn enable_event_enrollment(mut self, enabled: bool) -> Self {
667        self.config.enable_event_enrollment = enabled;
668        self
669    }
670
671    /// Set the interval in seconds between Event Enrollment evaluation passes
672    /// (default 10).
673    pub fn event_enrollment_interval_secs(mut self, secs: u64) -> Self {
674        self.config.event_enrollment_interval_secs = secs;
675        self
676    }
677
678    /// Set the segmentation support this device advertises and enforces.
679    ///
680    /// The dispatch loop honors the advertisement (Clause 5.4.5.1): inbound
681    /// segmented requests are reassembled only under `BOTH` or `RECEIVE`, and
682    /// draw a SEGMENTATION_NOT_SUPPORTED Abort otherwise. The default is
683    /// `NONE`.
684    pub fn segmentation_supported(mut self, segmentation: Segmentation) -> Self {
685        self.config.segmentation_supported = segmentation;
686        self
687    }
688
689    /// Set the vendor identifier advertised in I-Am responses.
690    pub fn vendor_id(mut self, id: u16) -> Self {
691        self.config.vendor_id = id;
692        self
693    }
694
695    /// Build and start the server, constructing a BipTransport from the config.
696    pub async fn build(self) -> Result<BACnetServer<BipTransport>, Error> {
697        let transport = BipTransport::new(
698            self.config.interface,
699            self.config.port,
700            self.config.broadcast_address,
701        );
702        BACnetServer::start_with_clock_mode_and_bindings(
703            self.config,
704            self.db,
705            transport,
706            Some(ClockConfig::default()),
707            self.configured_device_bindings,
708        )
709        .await
710    }
711}
712
713/// Key for tracking segmented transactions by peer and invoke ID.
714type SegKey = (MacAddr, Option<NpduAddress>, u8);
715
716fn segmented_transaction_key(
717    source_mac: &[u8],
718    source_network: Option<&NpduAddress>,
719    invoke_id: u8,
720) -> SegKey {
721    match source_network {
722        Some(address)
723            if (1..=0xFFFE).contains(&address.network) && !address.mac_address.is_empty() =>
724        {
725            (MacAddr::new(), Some(address.clone()), invoke_id)
726        }
727        _ => (
728            MacAddr::from_slice(source_mac),
729            source_network.cloned(),
730            invoke_id,
731        ),
732    }
733}
734
735#[derive(Debug)]
736enum SegmentedSendEvent {
737    SegmentAck(SegmentAckPdu),
738    Abort(AbortPdu),
739}
740
741#[derive(Debug, Clone)]
742enum SegmentedSendControlEvent {
743    Abort(AbortPdu),
744    Cancel,
745}
746
747struct SegmentedSendHandle {
748    segment_ack_tx: mpsc::Sender<SegmentAckPdu>,
749    control_tx: watch::Sender<Option<SegmentedSendControlEvent>>,
750    closed: AtomicBool,
751    current_sequence: AtomicU16,
752    total_segments: usize,
753}
754
755#[derive(Debug, Clone, Copy, PartialEq, Eq)]
756enum SegmentAckDisposition {
757    Advance,
758    Retransmit,
759}
760
761fn segment_ack_disposition(
762    ack: &SegmentAckPdu,
763    current: usize,
764    total_segments: usize,
765) -> Option<SegmentAckDisposition> {
766    if current >= total_segments {
767        return None;
768    }
769
770    let ack_seq = ack.sequence_number as usize;
771    if ack_seq >= total_segments {
772        return None;
773    }
774
775    // Clause 5.4.4.2 treats either ACK flavor's sequence number as the last
776    // segment accepted. A NAK for the preceding segment asks for `current`
777    // again; a NAK for `current` confirms it and advances the send window.
778    if ack_seq == current {
779        Some(SegmentAckDisposition::Advance)
780    } else if ack.negative_ack && current.checked_sub(1) == Some(ack_seq) {
781        Some(SegmentAckDisposition::Retransmit)
782    } else {
783        None
784    }
785}
786
787impl SegmentedSendHandle {
788    fn new(
789        segment_ack_tx: mpsc::Sender<SegmentAckPdu>,
790        control_tx: watch::Sender<Option<SegmentedSendControlEvent>>,
791        total_segments: usize,
792    ) -> Self {
793        Self {
794            segment_ack_tx,
795            control_tx,
796            closed: AtomicBool::new(false),
797            current_sequence: AtomicU16::new(u16::MAX),
798            total_segments,
799        }
800    }
801
802    fn accepts_segment_ack(&self, ack: &SegmentAckPdu) -> bool {
803        if ack.sent_by_server || self.closed.load(Ordering::Acquire) {
804            return false;
805        }
806
807        let current = self.current_sequence.load(Ordering::Acquire) as usize;
808        if current >= self.total_segments {
809            return false;
810        }
811
812        segment_ack_disposition(ack, current, self.total_segments).is_some()
813    }
814
815    fn send_control(&self, event: SegmentedSendControlEvent) {
816        self.closed.store(true, Ordering::Release);
817        self.control_tx.send_replace(Some(event));
818    }
819
820    fn same_channel(&self, sender: &mpsc::Sender<SegmentAckPdu>) -> bool {
821        self.segment_ack_tx.same_channel(sender)
822    }
823}
824
825#[derive(Debug, Clone, Copy)]
826struct SegmentedSendOptions {
827    segment_timeout: Duration,
828    max_retries: u8,
829}
830
831impl Default for SegmentedSendOptions {
832    fn default() -> Self {
833        Self {
834            segment_timeout: DEFAULT_APDU_SEGMENT_TIMEOUT,
835            max_retries: DEFAULT_APDU_SEGMENT_RETRIES,
836        }
837    }
838}
839
840struct SegmentedRequestState {
841    payload: segmented_receive::RequestPayload,
842    last_activity: Instant,
843    /// Last successfully saved new in-order segment, independent of SegmentTimer.
844    last_progress: Instant,
845    expected_seq: u8,
846    /// Last sequence number in the previously completed receive window.
847    initial_sequence_number: u8,
848    /// Duplicates silently discarded in the current receive window.
849    duplicate_count: u8,
850    /// Last segment accepted in order (Clause 5.4.2 LastSequenceNumber).
851    last_acked_seq: u8,
852    window_pos: u8,
853    actual_window_size: u8,
854    /// Monotonic count of segments accepted in order (#364).
855    ///
856    /// The reassembly total. `expected_seq` cannot serve: Clause 20.1.2.7
857    /// makes the sequence number modulo 256, so a 260-segment request ends at
858    /// sequence 3 and `seq + 1` names a four-segment total. This counter also
859    /// carries the overrun cap — acceptance is strictly in order, so it
860    /// reaches [`MAX_REQUEST_SEGMENTS`] exactly when the sequence number is
861    /// about to wrap onto stored segment 0.
862    accepted_segments: usize,
863}
864
865/// BACnet server with APDU dispatch and service handling.
866pub struct BACnetServer<T: TransportPort> {
867    config: ServerConfig,
868    /// Server-owned clock controller; absent in explicit clockless mode.
869    _clock: Option<Arc<ServerClock>>,
870    /// Shared network layer (also held by dispatch task; read by
871    /// [`write_local`](Self::write_local) for post-write COV/event sends).
872    network: Arc<NetworkLayer<T>>,
873    /// Shared object database.
874    db: Arc<RwLock<ObjectDatabase>>,
875    /// COV subscription table (also held by dispatch task; read by
876    /// [`write_local`](Self::write_local) to fire post-write notifications).
877    cov_table: Arc<RwLock<CovSubscriptionTable>>,
878    /// Channels for routing segmented-send events to in-progress segmented sends.
879    #[allow(dead_code)]
880    seg_ack_senders: Arc<Mutex<HashMap<SegKey, Arc<SegmentedSendHandle>>>>,
881    /// Permits that cap live segmented response sender tasks, including
882    /// cancelled senders that have not yet exited a transport send.
883    #[allow(dead_code)]
884    seg_send_permits: Arc<Semaphore>,
885    /// Operational cap of 255 concurrent confirmed COV notification workers.
886    /// Invoke-ID ownership is handled by `notification_transactions`.
887    cov_in_flight: Arc<Semaphore>,
888    /// Legacy public TSM state and the learned DNET-to-router cache.
889    server_tsm: Arc<Mutex<ServerTsm>>,
890    /// Invoke-ID ownership and terminal admission for confirmed notifications.
891    notification_transactions: Arc<NotificationTransactions>,
892    /// Server-lifetime exact inbound ConfirmedRequest duplicate state.
893    #[allow(dead_code)]
894    confirmed_request_tracker: Arc<ConfirmedRequestTracker>,
895    /// Shared configured and passively observed Device recipient authority.
896    device_bindings: Arc<RwLock<DeviceBindingTable>>,
897    /// Communication state: 0 = Enable, 1 = Disable, 2 = DisableInitiation.
898    comm_state: Arc<AtomicU8>,
899    /// Handle for the DCC auto-re-enable timer. A new DCC request aborts
900    /// any previous timer.
901    #[allow(dead_code)]
902    dcc_timer: Arc<Mutex<Option<JoinHandle<()>>>>,
903    dispatch_task: Option<JoinHandle<()>>,
904    cov_purge_task: Option<JoinHandle<()>>,
905    fault_detection_task: Option<JoinHandle<()>>,
906    event_enrollment_task: Option<JoinHandle<()>>,
907    trend_log_task: Option<JoinHandle<()>>,
908    schedule_tick_task: Option<JoinHandle<()>>,
909    /// One-second `Time_Delay` confirmation task for intrinsic reporting.
910    intrinsic_reporting_task: Option<JoinHandle<()>>,
911    /// Monotonic Binary Lighting Output WARN_OFF/WARN_RELINQUISH task.
912    binary_lighting_operation_task: Option<JoinHandle<()>>,
913    local_mac: MacAddr,
914}
915
916/// Cloneable handle for sending unsolicited I-Am announcements.
917pub struct IAmBroadcaster<T: TransportPort> {
918    config: ServerConfig,
919    network: Arc<NetworkLayer<T>>,
920    db: Arc<RwLock<ObjectDatabase>>,
921}
922
923impl<T: TransportPort> Clone for IAmBroadcaster<T> {
924    fn clone(&self) -> Self {
925        Self {
926            config: self.config.clone(),
927            network: Arc::clone(&self.network),
928            db: Arc::clone(&self.db),
929        }
930    }
931}
932
933impl BACnetServer<BipTransport> {
934    /// Create a BIP-specific builder with interface/port/broadcast fields.
935    pub fn bip_builder() -> BipServerBuilder {
936        BipServerBuilder {
937            config: ServerConfig::default(),
938            db: ObjectDatabase::new(),
939            configured_device_bindings: Vec::new(),
940        }
941    }
942
943    /// Create a BIP-specific builder (alias for backward compatibility).
944    pub fn builder() -> BipServerBuilder {
945        Self::bip_builder()
946    }
947}
948
949mod clock;
950#[cfg(test)]
951pub(crate) use clock::clocked_test_database;
952pub use clock::ClockConfig;
953use clock::ServerClock;
954mod binary_lighting_lifecycle;
955mod confirmed_request_tracker;
956mod cov_clock;
957mod cov_encoding;
958mod cov_notifications;
959mod cov_snapshot;
960mod device_bindings;
961mod dispatch;
962mod event_enrollment_lifecycle;
963mod event_message_policy;
964pub(crate) mod event_notification_payload;
965mod event_notifications;
966mod event_recipient_route;
967pub(crate) mod event_timestamp;
968mod lifecycle;
969mod local_writes;
970mod notification_transactions;
971mod requests;
972#[cfg(feature = "sc-tls")]
973mod sc_builder;
974#[cfg(test)]
975pub(crate) use requests::{EXECUTED_CONFIRMED, EXECUTED_UNCONFIRMED};
976#[cfg(test)]
977mod audit_notification_tests;
978#[cfg(test)]
979mod unconfirmed_audit_notification_tests;
980#[cfg(feature = "sc-tls")]
981pub use sc_builder::ScServerBuilder;
982mod responses;
983mod segmentation;
984mod segmented_receive;
985mod shutdown;
986
987#[cfg(test)]
988mod acknowledge_alarm_tests;
989#[cfg(test)]
990mod audit_log_query_tests;
991#[cfg(test)]
992mod binary_lighting_task_tests;
993#[cfg(test)]
994mod cov_notifications_tests;
995#[cfg(test)]
996mod dcc_event_detection_tests;
997#[cfg(test)]
998mod device_bindings_tests;
999#[cfg(test)]
1000mod device_recipient_routing_tests;
1001#[cfg(test)]
1002mod event_confirmed_routing_tests;
1003#[cfg(test)]
1004mod event_enable_distribution_tests;
1005#[cfg(test)]
1006mod event_enrollment_task_tests;
1007#[cfg(test)]
1008mod event_network_priority_tests;
1009#[cfg(test)]
1010mod event_notifications_tests;
1011#[cfg(test)]
1012mod event_recipient_routing_tests;
1013#[cfg(test)]
1014mod life_safety_cov_tests;
1015#[cfg(test)]
1016mod life_safety_operation_tests;
1017#[cfg(test)]
1018mod notification_transactions_tests;
1019#[cfg(test)]
1020mod segmentation_tests;
1021#[cfg(test)]
1022mod tests;
1023
1024impl<T: TransportPort + 'static> BACnetServer<T> {
1025    pub fn generic_builder() -> ServerBuilder<T> {
1026        ServerBuilder {
1027            config: ServerConfig::default(),
1028            db: ObjectDatabase::new(),
1029            transport: None,
1030            configured_device_bindings: Vec::new(),
1031        }
1032    }
1033}
bacnet_server/server/mod.rs

bacnet_server/server/
mod.rs
