umsh_ulcp_runtime/
duty_gate.rs

1//! Duty-cycle admission gate for the device node's radio path.
2//!
3//! `PROP_PHY_DUTY_LIMIT` bounds the *combined* airtime of every radio
4//! client (device-node plan increment 4). The session already prices
5//! and records its own transmissions against the shared
6//! [`DutyLedger`]; this wrapper is the node-side counterpart: it sits
7//! between the node's MAC and its virtual mux bundle, admits each
8//! transmit against the combined budget, and records the airtime once
9//! the transmit completes.
10//!
11//! A refused transmit is reported as [`TxError::CadTimeout`] — the one
12//! transmit error the MAC treats as "channel unavailable right now":
13//! it backs off with jitter, retries a bounded number of times, and
14//! then drops the frame. Any other error would surface as a fatal
15//! `MacError::Transmit` and kill the node pump, which must never be a
16//! consequence of duty limiting. A budget exhausted this hour rarely
17//! recovers within the MAC's short backoff horizon, so a refused frame
18//! is effectively shed — exactly the spec's posture (transmits never
19//! wait for duty-cycle allowance).
20
21use umsh_hal::{Clock, Radio, RxInfo, TxError, TxOptions};
22use umsh_ulcp_device::DutyLedger;
23
24/// A [`Radio`] decorator enforcing the shared duty budget.
25pub struct DutyGatedRadio<R, C> {
26    inner: R,
27    ledger: &'static DutyLedger,
28    clock: C,
29    /// Board coupling invoked after each successful transmit (e.g. the
30    /// T-1000E marks the load for its battery level estimator: voltage
31    /// sampled near a transmission is sagged, not resting OCV).
32    load_hook: fn(),
33}
34
35impl<R, C> DutyGatedRadio<R, C> {
36    pub fn new(inner: R, ledger: &'static DutyLedger, clock: C) -> Self {
37        Self::with_load_hook(inner, ledger, clock, || {})
38    }
39
40    pub fn with_load_hook(
41        inner: R,
42        ledger: &'static DutyLedger,
43        clock: C,
44        load_hook: fn(),
45    ) -> Self {
46        Self {
47            inner,
48            ledger,
49            clock,
50            load_hook,
51        }
52    }
53}
54
55impl<R: Radio, C: Clock> Radio for DutyGatedRadio<R, C> {
56    type Error = R::Error;
57
58    async fn transmit(
59        &mut self,
60        data: &[u8],
61        options: TxOptions,
62    ) -> Result<(), TxError<Self::Error>> {
63        let airtime_ms = self
64            .ledger
65            .admit(self.clock.now_ms(), data.len())
66            .map_err(|_| TxError::CadTimeout)?;
67        self.inner.transmit(data, options).await?;
68        // Record with the completion timestamp, mirroring the
69        // session's on_tx_result accounting. Refusals and failed
70        // transmits consume no budget.
71        self.ledger.record(self.clock.now_ms(), airtime_ms);
72        (self.load_hook)();
73        Ok(())
74    }
75
76    fn poll_receive(
77        &mut self,
78        cx: &mut core::task::Context<'_>,
79        buf: &mut [u8],
80    ) -> core::task::Poll<Result<RxInfo, Self::Error>> {
81        self.inner.poll_receive(cx, buf)
82    }
83
84    fn max_frame_size(&self) -> usize {
85        self.inner.max_frame_size()
86    }
87
88    fn t_frame_ms(&self) -> u32 {
89        self.inner.t_frame_ms()
90    }
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96    use core::future::Future;
97    use core::task::{Context, Poll, Waker};
98    use std::cell::Cell;
99    use std::rc::Rc;
100
101    fn block_on<F: Future>(future: F) -> F::Output {
102        let mut future = core::pin::pin!(future);
103        let waker = Waker::noop();
104        let mut context = Context::from_waker(&waker);
105        loop {
106            if let Poll::Ready(output) = future.as_mut().poll(&mut context) {
107                return output;
108            }
109        }
110    }
111
112    fn ledger() -> &'static DutyLedger {
113        Box::leak(Box::new(DutyLedger::new()))
114    }
115
116    /// Records transmit lengths; can be told to fail.
117    struct MockRadio {
118        sent: std::vec::Vec<usize>,
119        fail_next: bool,
120    }
121
122    impl Radio for MockRadio {
123        type Error = u8;
124
125        async fn transmit(
126            &mut self,
127            data: &[u8],
128            _options: TxOptions,
129        ) -> Result<(), TxError<Self::Error>> {
130            if self.fail_next {
131                self.fail_next = false;
132                return Err(TxError::Io(0xEE));
133            }
134            self.sent.push(data.len());
135            Ok(())
136        }
137
138        fn poll_receive(
139            &mut self,
140            _cx: &mut Context<'_>,
141            _buf: &mut [u8],
142        ) -> Poll<Result<RxInfo, Self::Error>> {
143            Poll::Pending
144        }
145
146        fn max_frame_size(&self) -> usize {
147            255
148        }
149
150        fn t_frame_ms(&self) -> u32 {
151            1_000
152        }
153    }
154
155    #[derive(Clone)]
156    struct TestClock(Rc<Cell<u64>>);
157
158    impl Clock for TestClock {
159        fn now_ms(&self) -> u64 {
160            self.0.get()
161        }
162    }
163
164    /// The acceptance shape for the shared ledger: session airtime and
165    /// node airtime interleave against one combined budget. The node's
166    /// transmit is admitted while the combined figure is under the
167    /// limit, refused as CadTimeout once session traffic exhausts it,
168    /// and node traffic symmetrically starves the session's own check.
169    #[test]
170    fn interleaved_session_and_node_tx_share_one_budget() {
171        let ledger = ledger();
172        // Fixture modulation (906.875 MHz profile): SF9/BW250k/CR5.
173        ledger.set_phy(9, 250_000, 5);
174        ledger.set_limit(655); // ≈1%: 36 s per hour.
175        let now = Rc::new(Cell::new(0u64));
176        let mut node = DutyGatedRadio::new(
177            MockRadio {
178                sent: vec![],
179                fail_next: false,
180            },
181            ledger,
182            TestClock(now.clone()),
183        );
184
185        // Node beacon goes out while the budget is fresh.
186        block_on(node.transmit(&[0u8; 32], TxOptions::default())).unwrap();
187        assert_eq!(node.inner.sent, [32]);
188        assert!(ledger.usage(now.get()) > 0);
189
190        // The session records a burst of its own completed TX
191        // (exactly what Session::on_tx_result does), exhausting the
192        // combined budget...
193        for _ in 0..36 {
194            ledger.record(now.get(), 1_000);
195        }
196        // ...so the session's own pre-check refuses...
197        assert!(ledger.would_exceed(now.get(), 100));
198        // ...and the node's next transmit is shed as CadTimeout
199        // without reaching the radio or consuming budget.
200        let refused = block_on(node.transmit(&[0u8; 32], TxOptions::default()));
201        assert!(matches!(refused, Err(TxError::CadTimeout)));
202        assert_eq!(node.inner.sent, [32]);
203        let usage_after_refusal = ledger.usage(now.get());
204
205        // Symmetrically: after the window ages out, node traffic alone
206        // starves the session's check.
207        now.set(now.get() + 60 * 60 * 1_000);
208        assert_eq!(ledger.usage(now.get()), 0);
209        for _ in 0..36 {
210            ledger.record(now.get(), 1_000);
211        }
212        assert!(ledger.would_exceed(now.get(), 100));
213        assert!(matches!(
214            block_on(node.transmit(&[0u8; 32], TxOptions::default())),
215            Err(TxError::CadTimeout)
216        ));
217        let _ = usage_after_refusal;
218    }
219
220    /// A failed inner transmit consumes no budget, and the admitted
221    /// airtime matches the ledger's modulation pricing.
222    #[test]
223    fn failed_transmit_records_nothing() {
224        let ledger = ledger();
225        ledger.set_phy(9, 250_000, 5);
226        ledger.set_limit(655);
227        let now = Rc::new(Cell::new(0u64));
228        let mut node = DutyGatedRadio::new(
229            MockRadio {
230                sent: vec![],
231                fail_next: true,
232            },
233            ledger,
234            TestClock(now.clone()),
235        );
236        let failed = block_on(node.transmit(&[0u8; 48], TxOptions::default()));
237        assert!(matches!(failed, Err(TxError::Io(0xEE))));
238        assert_eq!(ledger.usage(0), 0);
239
240        block_on(node.transmit(&[0u8; 48], TxOptions::default())).unwrap();
241        let expected = umsh_ulcp::airtime::lora_airtime_ms(9, 250_000, 5, 48);
242        // One recorded frame: usage reflects exactly its priced airtime
243        // (rounded up to 5 ms units and rescaled).
244        assert_eq!(
245            u64::from(ledger.usage(0)),
246            u64::from(expected.div_ceil(5)) * 65_535 / 720_000
247        );
248    }
249}
umsh_ulcp_runtime/duty_gate.rs

umsh_ulcp_runtime/
duty_gate.rs
