bacnet_server/server/
dcc_disable_rate.rs

1use super::{BipServerBuilder, ServerBuilder, TransportPort, runtime_clock};
2use bacnet_types::error::Error;
3use std::sync::Mutex;
4use std::time::Instant;
5
6/// Optional global authorization budget for DISABLE_INITIATION, not ingress protection.
7/// Use `Some(Default::default())` to enable; server configuration defaults to None.
8#[derive(Clone, Copy, Debug, Eq, PartialEq)]
9pub struct DccDisableRateLimit {
10    /// Initial and maximum whole tokens (1..=65535).
11    pub capacity: u32,
12    /// Time to earn one token, in milliseconds (1..=86400000).
13    pub refill_interval_ms: u64,
14}
15
16impl Default for DccDisableRateLimit {
17    fn default() -> Self {
18        Self {
19            capacity: 3,
20            refill_interval_ms: 20_000,
21        }
22    }
23}
24
25impl DccDisableRateLimit {
26    /// Validate local operator limits before startup/dialing, without time arithmetic.
27    pub fn validate(self) -> Result<(), Error> {
28        if !(1..=65535).contains(&self.capacity)
29            || !(1..=86_400_000).contains(&self.refill_interval_ms)
30        {
31            return Err(Error::Encoding(
32                "DCC disable rate requires capacity 1..=65535 and refill_interval_ms 1..=86400000"
33                    .into(),
34            ));
35        }
36        Ok(())
37    }
38}
39
40pub(super) struct Bucket {
41    interval_ns: u128,
42    maximum: u128,
43    state: Mutex<State>,
44}
45
46struct State {
47    // Credit is measured in nanoseconds: one token costs interval_ns. This
48    // retains fractional progress even across denied checks, without floats.
49    credit: u128,
50    last: Instant,
51}
52
53impl Bucket {
54    pub(super) fn new(config: DccDisableRateLimit) -> Result<Self, Error> {
55        config.validate()?;
56        let interval_ns = u128::from(config.refill_interval_ms) * 1_000_000;
57        let maximum = interval_ns * u128::from(config.capacity);
58        Ok(Self {
59            interval_ns,
60            maximum,
61            state: Mutex::new(State {
62                credit: maximum,
63                last: runtime_clock::now(),
64            }),
65        })
66    }
67
68    pub(super) fn admit(&self) -> bool {
69        let mut state = self.state.lock().unwrap();
70        // Read time under the lock: concurrent callers cannot move last backwards.
71        let now = runtime_clock::now();
72        state.credit = state
73            .credit
74            .saturating_add(now.saturating_duration_since(state.last).as_nanos())
75            .min(self.maximum);
76        state.last = now;
77        if state.credit < self.interval_ns {
78            return false;
79        }
80        state.credit -= self.interval_ns;
81        // Admission is charged now, with no rollback on later cancellation.
82        true
83    }
84}
85
86impl<T: TransportPort + 'static> ServerBuilder<T> {
87    /// Limit authorized DISABLE_INITIATION globally; None (default) disables it.
88    /// ENABLE is exempt. Each new native server starts with a full bucket.
89    pub fn dcc_disable_rate_limit(mut self, limit: Option<DccDisableRateLimit>) -> Self {
90        self.config.dcc_disable_rate_limit = limit;
91        self
92    }
93}
94
95impl BipServerBuilder {
96    /// Limit authorized DISABLE_INITIATION globally; None (default) disables it.
97    pub fn dcc_disable_rate_limit(mut self, limit: Option<DccDisableRateLimit>) -> Self {
98        self.config.dcc_disable_rate_limit = limit;
99        self
100    }
101}
102
103#[cfg(test)]
104mod tests {
105    use super::*;
106    use std::sync::{Arc, Barrier};
107    use tokio::time::{Duration, advance};
108
109    #[tokio::test(start_paused = true)]
110    async fn dcc_disable_rate_fractional_boundaries_and_idle_cap() {
111        let bucket = Bucket::new(DccDisableRateLimit::default()).unwrap();
112        for _ in 0..3 {
113            assert!(bucket.admit());
114        }
115        assert!(!bucket.admit());
116        // Repeated denials must preserve all accumulated fractional credit.
117        for _ in 0..19 {
118            advance(Duration::from_secs(1)).await;
119            assert!(!bucket.admit());
120        }
121        advance(Duration::from_millis(999)).await;
122        assert!(!bucket.admit());
123        advance(Duration::from_micros(999)).await;
124        assert!(!bucket.admit());
125        advance(Duration::from_micros(1)).await;
126        assert!(bucket.admit());
127        assert!(!bucket.admit());
128        advance(Duration::from_millis(45_500)).await;
129        assert!(bucket.admit());
130        assert!(bucket.admit());
131        assert!(!bucket.admit());
132        advance(Duration::from_millis(14_500)).await;
133        assert!(bucket.admit());
134        assert!(!bucket.admit());
135        advance(Duration::from_secs(86400 * 365)).await;
136        for _ in 0..3 {
137            assert!(bucket.admit());
138        }
139        assert!(!bucket.admit());
140    }
141
142    #[test]
143    fn dcc_disable_rate_concurrent_global_burst() {
144        let bucket = Arc::new(
145            Bucket::new(DccDisableRateLimit {
146                capacity: 3,
147                refill_interval_ms: 86_400_000,
148            })
149            .unwrap(),
150        );
151        let barrier = Arc::new(Barrier::new(32));
152        let threads: Vec<_> = (0..32)
153            .map(|_| {
154                let bucket = bucket.clone();
155                let barrier = barrier.clone();
156                std::thread::spawn(move || {
157                    barrier.wait();
158                    bucket.admit()
159                })
160            })
161            .collect();
162        assert_eq!(
163            threads
164                .into_iter()
165                .map(|t| usize::from(t.join().unwrap()))
166                .sum::<usize>(),
167            3
168        );
169        assert!(!bucket.admit());
170    }
171}
bacnet_server/server/dcc_disable_rate.rs

bacnet_server/server/
dcc_disable_rate.rs
