bacnet_server/server/
confirmed_request_tracker.rs

1use std::sync::atomic::{AtomicU64, Ordering};
2use std::sync::{Arc, Mutex};
3
4use super::request_peer::{CanonicalRequester, canonical_requester};
5use bacnet_encoding::apdu::ConfirmedRequest;
6use bacnet_encoding::npdu::NpduAddress;
7use bacnet_transport::port::{DirectScIdentity, TransportProvenance};
8
9// Local detection bounds, not transaction lifetimes. Above either bound,
10// Clause 5.3.5.3 permits normal service when exact duplicate detection is unavailable.
11const MAX_ENTRIES: usize = 256;
12const MAX_TRACKED_SERVICE_REQUEST_BYTES: usize = 64 * 1024;
13
14struct Entry {
15    id: u64,
16    requester: CanonicalRequester,
17    direct_identity: Option<DirectScIdentity>,
18    invoke_id: u8,
19    request: ConfirmedRequest,
20}
21
22#[derive(Default)]
23struct TrackerState {
24    next_id: u64,
25    entries: Vec<Entry>,
26}
27
28/// Exact duplicate detection only while the ordinary server transaction is live.
29/// No completed response or TTL is retained. Accepted direct SC keys also carry
30/// the original verified leaf/incarnation; non-direct canonicalization is unchanged.
31/// LSO's completed replay is an independent service policy and budget.
32#[derive(Default)]
33pub(super) struct ConfirmedRequestTracker {
34    state: Mutex<TrackerState>,
35    pub(super) lso: Arc<super::lso_replay::LsoReplayCache>,
36    /// See [`super::BACnetServer::group_source_request_drops`].
37    group_source_drops: AtomicU64,
38}
39
40pub(super) enum ConfirmedRequestAdmission {
41    Duplicate,
42    New(PendingConfirmedRequest),
43}
44
45/// Non-cloneable ownership of one pending transaction. Move into its response
46/// owner: unsegmented local encoded issuance/handoff, or the segmented child
47/// through final ACK/terminal outcome. Every cancellation/drop removes exactly
48/// this entry; request-task and segmented-send permits have separate lifetimes.
49pub(super) struct PendingConfirmedRequest {
50    tracker: Arc<ConfirmedRequestTracker>,
51    id: Option<u64>,
52}
53
54impl ConfirmedRequestTracker {
55    pub(super) fn begin(
56        self: &Arc<Self>,
57        source_mac: &[u8],
58        source_network: Option<&NpduAddress>,
59        provenance: TransportProvenance,
60        request: ConfirmedRequest,
61    ) -> ConfirmedRequestAdmission {
62        if request.service_request.len() > MAX_TRACKED_SERVICE_REQUEST_BYTES {
63            return ConfirmedRequestAdmission::New(PendingConfirmedRequest::untracked(self));
64        }
65        let requester = canonical_requester(source_mac, source_network);
66        let direct_identity = provenance.direct_sc_identity();
67        let invoke_id = request.invoke_id;
68        let mut state = self
69            .state
70            .lock()
71            .unwrap_or_else(|poisoned| poisoned.into_inner());
72        if state.entries.iter().any(|entry| {
73            entry.requester == requester
74                && entry.direct_identity == direct_identity
75                && entry.invoke_id == invoke_id
76                && entry.request == request
77        }) {
78            return ConfirmedRequestAdmission::Duplicate;
79        }
80        if state.entries.len() >= MAX_ENTRIES {
81            return ConfirmedRequestAdmission::New(PendingConfirmedRequest::untracked(self));
82        }
83        let id = state.next_id;
84        let Some(next_id) = id.checked_add(1) else {
85            // Never alias a surviving guard after counter exhaustion.
86            return ConfirmedRequestAdmission::New(PendingConfirmedRequest::untracked(self));
87        };
88        state.next_id = next_id;
89        state.entries.push(Entry {
90            id,
91            requester,
92            direct_identity,
93            invoke_id,
94            request,
95        });
96        ConfirmedRequestAdmission::New(PendingConfirmedRequest {
97            tracker: Arc::clone(self),
98            id: Some(id),
99        })
100    }
101}
102impl ConfirmedRequestTracker {
103    /// Whether to ignore a confirmed request from link-layer `source_mac`
104    /// because that address reaches a group of nodes (`is_group`, the
105    /// transport's `is_group_destination`). Its answer, any segment ACK, and
106    /// the confirmed COV notifications of a subscription it makes would all
107    /// go back there, to every node in the group (#1504). Each one is
108    /// counted.
109    pub(super) fn refuse_group_source(
110        &self,
111        source_mac: &[u8],
112        is_group: impl FnOnce(&[u8]) -> bool,
113    ) -> bool {
114        if !is_group(source_mac) {
115            return false;
116        }
117        self.group_source_drops.fetch_add(1, Ordering::Relaxed);
118        tracing::debug!(
119            ?source_mac,
120            "Ignoring a ConfirmedRequest from a group address"
121        );
122        true
123    }
124}
125
126impl<T: bacnet_transport::port::TransportPort + 'static> super::BACnetServer<T> {
127    /// Confirmed requests ignored since start because the link-layer address
128    /// they came from is a group destination of the transport
129    /// ([`TransportPort::is_group_destination`](bacnet_transport::port::TransportPort::is_group_destination)),
130    /// such as a B/IP broadcast or multicast address (#1504). The answer, and
131    /// the confirmed COV notifications of a subscription such a request
132    /// makes, would go back to that address, so the request is neither
133    /// executed nor answered. No built-in transport hands one up; a custom
134    /// one can.
135    pub fn group_source_request_drops(&self) -> u64 {
136        let tracker = &self.confirmed_request_tracker;
137        tracker.group_source_drops.load(Ordering::Relaxed)
138    }
139}
140
141impl PendingConfirmedRequest {
142    fn untracked(tracker: &Arc<ConfirmedRequestTracker>) -> Self {
143        Self {
144            tracker: Arc::clone(tracker),
145            id: None,
146        }
147    }
148}
149impl Drop for PendingConfirmedRequest {
150    fn drop(&mut self) {
151        if let Some(id) = self.id {
152            self.tracker
153                .state
154                .lock()
155                .unwrap_or_else(|poisoned| poisoned.into_inner())
156                .entries
157                .retain(|entry| entry.id != id);
158        }
159    }
160}
161
162#[cfg(test)]
163#[path = "confirmed_tracker_tests.rs"]
164mod tests;
165
166#[cfg(test)]
167#[path = "group_source_request_tests.rs"]
168mod group_source_tests;
bacnet_server/server/confirmed_request_tracker.rs

bacnet_server/server/
confirmed_request_tracker.rs
