bacnet_server/
cov.rs

1//! COV subscription engine — tracks SubscribeCOV subscriptions and their lifetimes.
2
3use std::collections::HashMap;
4use std::sync::atomic::Ordering;
5use std::sync::Arc;
6use std::time::Instant;
7
8use bacnet_encoding::npdu::NpduAddress;
9use bacnet_types::enums::{ErrorClass, ErrorCode, PropertyIdentifier};
10use bacnet_types::error::Error;
11use bacnet_types::primitives::ObjectIdentifier;
12use bacnet_types::MacAddr;
13
14use crate::runtime_clock;
15
16mod identity;
17pub use identity::*;
18pub(crate) mod active;
19mod admission;
20pub use admission::MultipleRefusal;
21mod confirmed;
22pub(crate) use confirmed::{BeginRefusal, CovRevisits};
23mod sample;
24pub use sample::CovSample;
25mod observation;
26mod observation_order;
27pub use observation::CovObservation;
28use observation_order::ObservationOwner;
29pub(crate) use observation_order::PreparedCovCompletion;
30pub(crate) mod flags;
31mod lifetime;
32pub(crate) mod multiple_reads;
33pub(crate) mod prepare;
34pub(crate) mod reported;
35pub(crate) mod timed;
36mod timed_capture;
37pub(crate) use timed_capture::TimedWriteCapture;
38pub(crate) mod value_source;
39pub use lifetime::CovTimeRemaining;
40
41mod policy;
42pub use policy::*;
43
44#[cfg(test)]
45mod identity_tests;
46#[cfg(test)]
47mod policy_tests;
48#[cfg(test)]
49mod tests;
50
51/// Largest B/IP APDU; the default history bound until a server sets its own.
52const DEFAULT_TIMED_APDU_LENGTH: usize = 1476;
53
54/// Proposed COV subscription data. Table acceptance validates its canonical
55/// identity and returns an immutable [`CovSubscriptionSnapshot`] for delivery.
56#[derive(Debug, Clone)]
57pub struct CovSubscription {
58    /// Immediate delivery MAC: the local subscriber or its current router.
59    pub subscriber_mac: MacAddr,
60    /// Routed source address when the subscriber is behind a BACnet router.
61    pub subscriber_network: Option<NpduAddress>,
62    /// Process identifier chosen by the subscriber.
63    pub subscriber_process_identifier: u32,
64    /// The object being monitored.
65    pub monitored_object_identifier: ObjectIdentifier,
66    /// Notification form. Mutable renewal data for ordinary/Single subscriptions;
67    /// part of the canonical identity for Multiple references.
68    pub issue_confirmed_notifications: bool,
69    /// When this subscription expires (None = infinite lifetime), on tokio's
70    /// clock: `tokio::time::Instant::now().into_std()` plus the lifetime,
71    /// which is `Instant::now()` plus it unless a test has paused the clock.
72    pub expires_at: Option<Instant>,
73    /// Last delivered bounded observation, including specialized command fields.
74    pub last_notified_observation: Option<CovObservation>,
75    /// Monitored property for Single-property and Multiple-reference subscriptions.
76    pub monitored_property: Option<PropertyIdentifier>,
77    /// Accepted property index; absent, zero and element indexes are independent.
78    pub monitored_property_array_index: Option<u32>,
79    /// COV increment override (SubscribeCOVProperty only).
80    pub cov_increment: Option<f32>,
81    /// Notification service family used for this subscription.
82    pub notification_kind: CovNotificationKind,
83    /// Whether COVNotificationMultiple values should include timeOfChange.
84    pub timestamped: bool,
85}
86
87impl CovSubscription {
88    /// Original client address shared by subscription identity and accounting.
89    /// Process, family and monitored coordinates additionally distinguish subscriptions.
90    pub fn recipient(&self) -> CovRecipient {
91        CovRecipient::from_endpoint(&self.subscriber_mac, self.subscriber_network.as_ref())
92    }
93}
94
95/// COV notification service family for a stored subscription.
96#[derive(Debug, Clone, Copy, PartialEq, Eq)]
97pub enum CovNotificationKind {
98    /// COVNotification / ConfirmedCOVNotification.
99    Single,
100    /// COVNotificationMultiple / ConfirmedCOVNotificationMultiple.
101    Multiple,
102}
103
104/// Table of active COV subscriptions.
105#[derive(Debug)]
106pub struct CovSubscriptionTable {
107    subs: HashMap<CovSubscriptionKey, CovSubscriptionSnapshot>,
108    generation: u64,
109    owner: Arc<ObservationOwner>,
110    peer_counts: HashMap<CovRecipient, usize>,
111    peer_indefinite_counts: HashMap<CovRecipient, usize>,
112    policy: CovPolicy,
113    counters: Arc<AtomicCovCounters>,
114    in_flight: Arc<CovInFlightTracker>,
115    revisits: Arc<CovRevisits>,
116    dispatch_turn: usize,
117    timed: timed::TimedStore,
118    /// Live list samples taken for read requests, so tests can see which
119    /// requests snapshot the table (#1213).
120    #[cfg(test)]
121    live_samples: std::sync::atomic::AtomicUsize,
122}
123
124impl Default for CovSubscriptionTable {
125    fn default() -> Self {
126        Self::new()
127    }
128}
129
130impl CovSubscriptionTable {
131    /// Create a new COV subscription table with default policy and counters.
132    pub fn new() -> Self {
133        Self::with_policy(CovPolicy::default(), Arc::new(AtomicCovCounters::default()))
134    }
135
136    /// Create a new COV subscription table with a custom policy and counters.
137    pub fn with_policy(policy: CovPolicy, counters: Arc<AtomicCovCounters>) -> Self {
138        let timed = timed::TimedStore::new(DEFAULT_TIMED_APDU_LENGTH, Arc::clone(&counters));
139        Self {
140            subs: HashMap::new(),
141            generation: 0,
142            owner: Arc::new(ObservationOwner::default()),
143            peer_counts: HashMap::new(),
144            peer_indefinite_counts: HashMap::new(),
145            policy: policy.sanitized(),
146            counters,
147            in_flight: Arc::new(CovInFlightTracker::default()),
148            revisits: Arc::default(),
149            dispatch_turn: 0,
150            timed,
151            #[cfg(test)]
152            live_samples: std::sync::atomic::AtomicUsize::new(0),
153        }
154    }
155
156    /// Bound each Multiple context's pending timestamped changes by what one
157    /// notification of this maximum APDU length can carry.
158    pub fn with_max_apdu_length(mut self, max_apdu_length: usize) -> Self {
159        self.timed = timed::TimedStore::new(max_apdu_length, Arc::clone(&self.counters));
160        self
161    }
162
163    /// Pending timestamped COV-multiple changes of this table's references.
164    pub(crate) fn timed(&self) -> &timed::TimedStore {
165        &self.timed
166    }
167
168    /// Return the next notification dispatch turn counter (wrapping).
169    pub(crate) fn next_dispatch_turn(&mut self) -> usize {
170        let turn = self.dispatch_turn;
171        self.dispatch_turn = self.dispatch_turn.wrapping_add(1);
172        turn
173    }
174
175    /// Get a reference to the active COV policy.
176    pub fn policy(&self) -> &CovPolicy {
177        &self.policy
178    }
179
180    /// Get a reference to the atomic counters.
181    pub fn counters(&self) -> &Arc<AtomicCovCounters> {
182        &self.counters
183    }
184
185    /// Get a reference to the in-flight confirmed notification tracker.
186    pub fn in_flight_tracker(&self) -> &Arc<CovInFlightTracker> {
187        &self.in_flight
188    }
189
190    /// Get the number of active subscriptions for a peer.
191    pub fn peer_subscription_count(&self, peer: &CovRecipient) -> usize {
192        self.peer_counts.get(peer).copied().unwrap_or(0)
193    }
194
195    /// Get the number of active indefinite subscriptions for a peer.
196    pub fn peer_indefinite_count(&self, peer: &CovRecipient) -> usize {
197        self.peer_indefinite_counts.get(peer).copied().unwrap_or(0)
198    }
199
200    /// Get an accepted entry by its complete typed identity.
201    pub fn get_subscription(&self, key: &CovSubscriptionKey) -> Option<&CovSubscriptionSnapshot> {
202        self.subs.get(key)
203    }
204
205    /// Every accepted reference of one Multiple context.
206    pub(crate) fn multiple_context_references<'a>(
207        &'a self,
208        context: &'a MultipleContextKey,
209    ) -> impl Iterator<Item = &'a CovSubscriptionSnapshot> + 'a {
210        self.subs
211            .iter()
212            .filter(move |(key, _)| key.multiple_context() == Some(context))
213            .map(|(_, sub)| sub)
214    }
215
216    /// Whether an exact subscription identity is present.
217    pub fn contains(&self, key: &CovSubscriptionKey) -> bool {
218        self.subs.contains_key(key)
219    }
220
221    fn remove_internal(&mut self, key: &CovSubscriptionKey, was_cancelled: bool) -> bool {
222        if let Some(sub) = self.subs.remove(key) {
223            self.timed.lock().remove(key);
224            self.revisits.forget(key);
225            let peer = sub.recipient();
226            if let Some(count) = self.peer_counts.get_mut(&peer) {
227                *count = count.saturating_sub(1);
228                if *count == 0 {
229                    self.peer_counts.remove(&peer);
230                }
231            }
232            if sub.expires_at.is_none() {
233                if let Some(count) = self.peer_indefinite_counts.get_mut(&peer) {
234                    *count = count.saturating_sub(1);
235                    if *count == 0 {
236                        self.peer_indefinite_counts.remove(&peer);
237                    }
238                }
239            }
240            if was_cancelled {
241                self.counters
242                    .subscriptions_cancelled
243                    .fetch_add(1, Ordering::Relaxed);
244            }
245            self.counters
246                .subscriptions_active
247                .store(self.subs.len() as u64, Ordering::Relaxed);
248            true
249        } else {
250            false
251        }
252    }
253
254    /// Cancel only the exact family/form/property/index identity.
255    pub fn unsubscribe(&mut self, key: &CovSubscriptionKey) -> bool {
256        self.remove_internal(key, true)
257    }
258
259    /// Cancel every reference in the exact Multiple context.
260    pub fn unsubscribe_cov_multiple_context(&mut self, context: &MultipleContextKey) {
261        let keys: Vec<_> = self
262            .subs
263            .keys()
264            .filter(|key| key.multiple_context() == Some(context))
265            .cloned()
266            .collect();
267        for key in keys {
268            self.remove_internal(&key, true);
269        }
270    }
271
272    /// Remove all subscriptions for a given object (used on DeleteObject).
273    pub fn remove_for_object(&mut self, oid: ObjectIdentifier) {
274        let to_remove: Vec<_> = self
275            .subs
276            .iter()
277            .filter(|(k, _)| k.object() == oid)
278            .map(|(k, _)| k.clone())
279            .collect();
280        for key in to_remove {
281            self.remove_internal(&key, false);
282        }
283    }
284
285    /// Remove subscriptions using this current route and release shared peer quotas.
286    /// An obsolete router cannot remove a subscription migrated elsewhere.
287    pub fn remove_peer_subscriptions(
288        &mut self,
289        mac: &[u8],
290        network: Option<&NpduAddress>,
291    ) -> usize {
292        let target = SubscriberEndpoint::new(mac, network);
293        let to_remove: Vec<_> = self
294            .subs
295            .iter()
296            .filter(|(_, entry)| entry.endpoint() == target)
297            .map(|(k, _)| k.clone())
298            .collect();
299        let count = to_remove.len();
300        for key in to_remove {
301            self.remove_internal(&key, true);
302        }
303        count
304    }
305
306    /// Remove all expired subscriptions. Returns the number removed.
307    pub fn purge_expired(&mut self) -> usize {
308        let now = runtime_clock::now();
309        let mut purged_count = 0;
310        let mut to_remove = Vec::new();
311        for (k, sub) in &self.subs {
312            if sub.expires_at.is_some_and(|exp| exp <= now) {
313                to_remove.push(k.clone());
314            }
315        }
316        for key in to_remove {
317            purged_count += usize::from(self.remove_internal(&key, false));
318        }
319        if purged_count > 0 {
320            self.counters
321                .subscriptions_purged
322                .fetch_add(purged_count as u64, Ordering::Relaxed);
323            self.counters
324                .subscriptions_active
325                .store(self.subs.len() as u64, Ordering::Relaxed);
326        }
327        purged_count
328    }
329
330    /// Test fixture: let every subscription's lifetime run out now, without
331    /// waiting for the clock that lifetimes use to reach it.
332    #[cfg(test)]
333    pub(crate) fn expire_all_for_test(&mut self) {
334        let now = runtime_clock::now();
335        for entry in self.subs.values_mut() {
336            entry.subscription.expires_at = Some(now);
337        }
338    }
339
340    /// Get all active (non-expired) subscriptions for a given object.
341    pub fn subscriptions_for(&mut self, oid: &ObjectIdentifier) -> Vec<&CovSubscriptionSnapshot> {
342        self.purge_expired();
343        self.subs
344            .values()
345            .filter(|sub| sub.monitored_object_identifier == *oid)
346            .collect()
347    }
348
349    /// The finest COV increment each object's live subscriptions to
350    /// `property` ask for (#1510): SubscribeCOVProperty subscriptions and
351    /// Multiple references that give one. A negative increment reports any
352    /// change, as zero does, so it counts as zero; a NaN or infinite one
353    /// reports nothing finer, so it's left out.
354    pub(crate) fn finest_increments(
355        &self,
356        property: PropertyIdentifier,
357    ) -> HashMap<ObjectIdentifier, f64> {
358        let now = runtime_clock::now();
359        let mut finest = HashMap::new();
360        for sub in self.subs.values() {
361            if sub.monitored_property != Some(property)
362                || sub.expires_at.is_some_and(|expires| expires <= now)
363            {
364                continue;
365            }
366            let Some(increment) = sub
367                .cov_increment
368                .map(f64::from)
369                .filter(|increment| increment.is_finite())
370            else {
371                continue;
372            };
373            let increment = increment.max(0.0);
374            finest
375                .entry(sub.monitored_object_identifier)
376                .and_modify(|finest: &mut f64| *finest = finest.min(increment))
377                .or_insert(increment);
378        }
379        finest
380    }
381
382    /// Whether a snapshot still owns a live entry in this table.
383    pub fn is_current(&self, snapshot: &CovSubscriptionSnapshot) -> bool {
384        self.remaining_lifetime(snapshot, runtime_clock::now())
385            .and_then(CovTimeRemaining::wire_seconds)
386            .is_some()
387    }
388
389    /// Ordinary whole-object trigger policy: a numeric Present_Value must move
390    /// by the increment, any other value must change, and the first report
391    /// always fires. Status_Flags and other Table 13-1 trigger values (such as
392    /// Staging's Present_Stage) are checked separately by the caller.
393    /// An unchanged object therefore reports nothing, however often it is
394    /// fanned out. Property subscriptions compare their prepared sample instead.
395    pub fn should_notify(
396        sub: &CovSubscription,
397        current_value: Option<&CovSample>,
398        cov_increment: Option<f64>,
399    ) -> bool {
400        let Some(current) = current_value else {
401            return true;
402        };
403        current.reports(
404            sub.last_notified_observation
405                .as_ref()
406                .map(|observation| observation.sample()),
407            cov_increment,
408            true,
409        )
410    }
411
412    /// Number of active subscriptions.
413    pub fn len(&self) -> usize {
414        self.subs.len()
415    }
416
417    /// Whether the table is empty.
418    pub fn is_empty(&self) -> bool {
419        self.subs.is_empty()
420    }
421}
bacnet_server/cov.rs

bacnet_server/
cov.rs
