bacnet_server/
cov.rs

1//! COV subscription engine — tracks SubscribeCOV subscriptions and their lifetimes.
2
3use std::collections::HashMap;
4use std::sync::Arc;
5use std::sync::atomic::Ordering;
6use std::time::Instant;
7
8use bacnet_encoding::npdu::NpduAddress;
9use bacnet_types::MacAddr;
10use bacnet_types::enums::{ErrorClass, ErrorCode, PropertyIdentifier};
11use bacnet_types::error::Error;
12use bacnet_types::primitives::ObjectIdentifier;
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        match self.subs.remove(key) {
223            Some(sub) => {
224                self.timed.lock().remove(key);
225                self.revisits.forget(key);
226                let peer = sub.recipient();
227                if let Some(count) = self.peer_counts.get_mut(&peer) {
228                    *count = count.saturating_sub(1);
229                    if *count == 0 {
230                        self.peer_counts.remove(&peer);
231                    }
232                }
233                if sub.expires_at.is_none()
234                    && let Some(count) = self.peer_indefinite_counts.get_mut(&peer)
235                {
236                    *count = count.saturating_sub(1);
237                    if *count == 0 {
238                        self.peer_indefinite_counts.remove(&peer);
239                    }
240                }
241                if was_cancelled {
242                    self.counters
243                        .subscriptions_cancelled
244                        .fetch_add(1, Ordering::Relaxed);
245                }
246                self.counters
247                    .subscriptions_active
248                    .store(self.subs.len() as u64, Ordering::Relaxed);
249                true
250            }
251            _ => false,
252        }
253    }
254
255    /// Cancel only the exact family/form/property/index identity.
256    pub fn unsubscribe(&mut self, key: &CovSubscriptionKey) -> bool {
257        self.remove_internal(key, true)
258    }
259
260    /// Cancel every reference in the exact Multiple context.
261    pub fn unsubscribe_cov_multiple_context(&mut self, context: &MultipleContextKey) {
262        let keys: Vec<_> = self
263            .subs
264            .keys()
265            .filter(|key| key.multiple_context() == Some(context))
266            .cloned()
267            .collect();
268        for key in keys {
269            self.remove_internal(&key, true);
270        }
271    }
272
273    /// Remove all subscriptions for a given object (used on DeleteObject).
274    pub fn remove_for_object(&mut self, oid: ObjectIdentifier) {
275        let to_remove: Vec<_> = self
276            .subs
277            .iter()
278            .filter(|(k, _)| k.object() == oid)
279            .map(|(k, _)| k.clone())
280            .collect();
281        for key in to_remove {
282            self.remove_internal(&key, false);
283        }
284    }
285
286    /// Remove subscriptions using this current route and release shared peer quotas.
287    /// An obsolete router cannot remove a subscription migrated elsewhere.
288    pub fn remove_peer_subscriptions(
289        &mut self,
290        mac: &[u8],
291        network: Option<&NpduAddress>,
292    ) -> usize {
293        let target = SubscriberEndpoint::new(mac, network);
294        let to_remove: Vec<_> = self
295            .subs
296            .iter()
297            .filter(|(_, entry)| entry.endpoint() == target)
298            .map(|(k, _)| k.clone())
299            .collect();
300        let count = to_remove.len();
301        for key in to_remove {
302            self.remove_internal(&key, true);
303        }
304        count
305    }
306
307    /// Remove all expired subscriptions. Returns the number removed.
308    pub fn purge_expired(&mut self) -> usize {
309        let now = runtime_clock::now();
310        let mut purged_count = 0;
311        let mut to_remove = Vec::new();
312        for (k, sub) in &self.subs {
313            if sub.expires_at.is_some_and(|exp| exp <= now) {
314                to_remove.push(k.clone());
315            }
316        }
317        for key in to_remove {
318            purged_count += usize::from(self.remove_internal(&key, false));
319        }
320        if purged_count > 0 {
321            self.counters
322                .subscriptions_purged
323                .fetch_add(purged_count as u64, Ordering::Relaxed);
324            self.counters
325                .subscriptions_active
326                .store(self.subs.len() as u64, Ordering::Relaxed);
327        }
328        purged_count
329    }
330
331    /// Test fixture: let every subscription's lifetime run out now, without
332    /// waiting for the clock that lifetimes use to reach it.
333    #[cfg(test)]
334    pub(crate) fn expire_all_for_test(&mut self) {
335        let now = runtime_clock::now();
336        for entry in self.subs.values_mut() {
337            entry.subscription.expires_at = Some(now);
338        }
339    }
340
341    /// Get all active (non-expired) subscriptions for a given object.
342    pub fn subscriptions_for(&mut self, oid: &ObjectIdentifier) -> Vec<&CovSubscriptionSnapshot> {
343        self.purge_expired();
344        self.subs
345            .values()
346            .filter(|sub| sub.monitored_object_identifier == *oid)
347            .collect()
348    }
349
350    /// The finest COV increment each object's live subscriptions to
351    /// `property` ask for (#1510): SubscribeCOVProperty subscriptions and
352    /// Multiple references that give one. A negative increment reports any
353    /// change, as zero does, so it counts as zero; a NaN or infinite one
354    /// reports nothing finer, so it's left out.
355    pub(crate) fn finest_increments(
356        &self,
357        property: PropertyIdentifier,
358    ) -> HashMap<ObjectIdentifier, f64> {
359        let now = runtime_clock::now();
360        let mut finest = HashMap::new();
361        for sub in self.subs.values() {
362            if sub.monitored_property != Some(property)
363                || sub.expires_at.is_some_and(|expires| expires <= now)
364            {
365                continue;
366            }
367            let Some(increment) = sub
368                .cov_increment
369                .map(f64::from)
370                .filter(|increment| increment.is_finite())
371            else {
372                continue;
373            };
374            let increment = increment.max(0.0);
375            finest
376                .entry(sub.monitored_object_identifier)
377                .and_modify(|finest: &mut f64| *finest = finest.min(increment))
378                .or_insert(increment);
379        }
380        finest
381    }
382
383    /// Whether a snapshot still owns a live entry in this table.
384    pub fn is_current(&self, snapshot: &CovSubscriptionSnapshot) -> bool {
385        self.remaining_lifetime(snapshot, runtime_clock::now())
386            .and_then(CovTimeRemaining::wire_seconds)
387            .is_some()
388    }
389
390    /// Ordinary whole-object trigger policy: a numeric Present_Value must move
391    /// by the increment, any other value must change, and the first report
392    /// always fires. Status_Flags and other Table 13-1 trigger values (such as
393    /// Staging's Present_Stage) are checked separately by the caller.
394    /// An unchanged object therefore reports nothing, however often it is
395    /// fanned out. Property subscriptions compare their prepared sample instead.
396    pub fn should_notify(
397        sub: &CovSubscription,
398        current_value: Option<&CovSample>,
399        cov_increment: Option<f64>,
400    ) -> bool {
401        let Some(current) = current_value else {
402            return true;
403        };
404        current.reports(
405            sub.last_notified_observation
406                .as_ref()
407                .map(|observation| observation.sample()),
408            cov_increment,
409            true,
410        )
411    }
412
413    /// Number of active subscriptions.
414    pub fn len(&self) -> usize {
415        self.subs.len()
416    }
417
418    /// Whether the table is empty.
419    pub fn is_empty(&self) -> bool {
420        self.subs.is_empty()
421    }
422}
bacnet_server/cov.rs

bacnet_server/
cov.rs
