prns_interfaces_embassy/
lora.rs

1//! Embedded SX126x LoRa transport with bounded spectrum access.
2//!
3//! [`LoRaInterfaceInput`] requires an [`AirtimePolicy`], a
4//! [`LoRaSpectrumStatus`], and caller-owned transmit queue storage, with
5//! [`LORA_TX_QUEUE_BYTES`] as the general-purpose default. Construction validates
6//! frequency, transmit power, preamble,
7//! and any fixed airtime limit before the radio task can start.
8//! [`AirtimePolicy::Regional`] is the normal choice; a fixed policy may tighten
9//! a regional limit but cannot weaken one.
10//!
11//! The radio runs in continuous receive and combines preamble/header IRQ
12//! evidence, an adaptive RSSI noise floor, real two-slot DIFS, a fine-grained
13//! randomized ticket, and a final IRQ-plus-RSSI check immediately before
14//! transmit. Channel activity restarts DIFS but permanently freezes completed
15//! ticket progress. Successfully decoded peer airtime earns bounded `1x..=3x`
16//! countdown acceleration; preamble, RSSI, CRC-error, and header-error evidence
17//! remains conservatively busy without earning priority.
18//!
19//! A winner drains complete FIFO packets into an airtime-bounded opportunity.
20//! The profile-derived limit is an integral number of maximum split packets and
21//! at least 42 contention slots. The radio releases immediately when the FIFO
22//! empties or the next logical packet would cross that limit. Split Reticulum
23//! packets remain indivisible and contiguous on air for RNode interoperability.
24//! A capped winner re-contends from band zero with one quantum of earned age;
25//! existing waiters retain their smaller residual tickets.
26//!
27//! [`LoRaSpectrumStatus::snapshot`] exposes sampled channel occupancy, noise and
28//! CCA levels, deferrals, false preambles, contention and duty drops, and radio
29//! recoveries. These diagnostics are observational; they do not provide a
30//! listen-before-talk bypass.
31//!
32//! The active packet remains separate from the packed 6 KiB FIFO, and its
33//! contention timeout begins only when it becomes active. Scheduler additions
34//! are scalar state only: no heap allocation or additional packet buffers. A
35//! one-slot manifold lane provides the ingress handoff. ESP32-S3 Hopspots place
36//! the FIFO in PSRAM, while T-Echo supplies static SRAM storage.
37
38use embassy_futures::select::{select, select4, select5, Either, Either4, Either5};
39use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
40use embassy_sync::channel::DynamicSender;
41use embassy_sync::signal::Signal;
42use embassy_time::{Duration, Instant, Timer};
43use embedded_hal::digital::OutputPin;
44use embedded_hal_async::delay::DelayNs;
45use embedded_hal_async::digital::Wait;
46use embedded_hal_async::spi::SpiDevice;
47use heapless::Vec as HeaplessVec;
48use portable_atomic::{AtomicU32, Ordering};
49
50use prns_core::engine::InstantMillis;
51use prns_core::interfaces::lora::{
52    self, air_frame_count, encode_air_frame_part, AirtimePolicy, AirtimePolicyError, CodingRate,
53    LoRaReassembler, LoraBandwidth, Modulation, RadioProfile, RadioProfileError, SpreadingFactor,
54    CHANNEL_TAG_CAP, LORA_MAX_PAYLOAD, LORA_SINGLE_FRAME_MAX, RNODE_LORA_SYNC_WORD,
55};
56use prns_core::interfaces::{
57    AirtimeDutyCycle, ConnectionState, InterfaceDescriptor, InterfaceId, InterfaceKind,
58    PacketPhyStats,
59};
60use prns_runtime::manifold::airtime::AirtimeLedger;
61use prns_runtime::manifold::driver::{EmbassyInterfaceStatus, InterfaceLifecycle};
62use prns_runtime::manifold::interface_seam::{
63    Interface, InterfaceSeam, OutboundDisposition, OutboundDropReason,
64};
65use prns_runtime::manifold::throughput::ThroughputLedger;
66
67use crate::radios::sx126x::{self, RadioEvent, Sx126x};
68
69mod airtime_quantum;
70mod channel_access;
71#[cfg(test)]
72mod simulation_tests;
73mod transmit_queue;
74
75use airtime_quantum::ServiceAge;
76use channel_access::{
77    ChannelAccess, ChannelAccessAction, ChannelObservation, ChannelTiming, ContentionPriority,
78    DemodulatorActivity, NoiseFloor,
79};
80use transmit_queue::{TransmitQueue, TransmitQueueError};
81
82const IDLE_TICK: Duration = Duration::from_millis(250);
83const SENSING_UNPUBLISHED: u32 = u32::MAX;
84pub const LORA_TX_QUEUE_BYTES: usize = 6 * 1024;
85
86#[derive(Debug, PartialEq, Eq)]
87enum PacketPlacement {
88    Active,
89    Queued,
90}
91
92struct ActivePacket {
93    bytes: [u8; LORA_MAX_PAYLOAD],
94    len: Option<usize>,
95    airtime_us: u64,
96    activated_at_ms: u64,
97}
98
99impl ActivePacket {
100    const fn new() -> Self {
101        Self {
102            bytes: [0; LORA_MAX_PAYLOAD],
103            len: None,
104            airtime_us: 0,
105            activated_at_ms: 0,
106        }
107    }
108
109    fn activate(&mut self, packet: &[u8], profile: &RadioProfile, now_ms: u64) {
110        self.bytes[..packet.len()].copy_from_slice(packet);
111        self.len = Some(packet.len());
112        self.airtime_us = packet_airtime(packet, profile);
113        self.activated_at_ms = now_ms;
114    }
115
116    fn clear(&mut self) -> bool {
117        self.len.take().is_some()
118    }
119
120    fn recompute_airtime(&mut self, profile: &RadioProfile) {
121        let Some(len) = self.len else {
122            return;
123        };
124        self.airtime_us = packet_airtime(&self.bytes[..len], profile);
125    }
126
127    fn channel_access(
128        &self,
129        profile: RadioProfile,
130        now_ms: u64,
131        priority: ContentionPriority,
132    ) -> Option<ChannelAccess> {
133        self.len.map(|_| {
134            ChannelAccess::new_at(
135                profile,
136                now_ms,
137                self.activated_at_ms,
138                self.airtime_us,
139                priority,
140            )
141        })
142    }
143}
144
145struct TransmitBacklog<'a> {
146    queue: TransmitQueue<'a>,
147    active: ActivePacket,
148}
149
150impl<'a> TransmitBacklog<'a> {
151    fn new(storage: &'a mut [u8]) -> Self {
152        Self {
153            queue: TransmitQueue::new(storage),
154            active: ActivePacket::new(),
155        }
156    }
157
158    fn accept(
159        &mut self,
160        packet: &[u8],
161        profile: &RadioProfile,
162        now_ms: u64,
163    ) -> Result<PacketPlacement, TransmitQueueError> {
164        if packet.len() > LORA_MAX_PAYLOAD {
165            return Err(TransmitQueueError::PacketTooLarge);
166        }
167        if self.active.len.is_none() && self.queue.is_empty() {
168            self.active.activate(packet, profile, now_ms);
169            return Ok(PacketPlacement::Active);
170        }
171        self.queue.push(packet)?;
172        Ok(PacketPlacement::Queued)
173    }
174
175    fn activate_next(&mut self, profile: &RadioProfile, now_ms: u64) -> bool {
176        if self.active.len.is_some() {
177            return false;
178        }
179        let Some(len) = self.queue.pop(&mut self.active.bytes) else {
180            return false;
181        };
182        self.active.len = Some(len);
183        self.active.airtime_us = packet_airtime(&self.active.bytes[..len], profile);
184        self.active.activated_at_ms = now_ms;
185        true
186    }
187
188    const fn can_accept_outbound(&self) -> bool {
189        if self.active.len.is_none() && self.queue.is_empty() {
190            true
191        } else {
192            self.queue.can_push_max_packet()
193        }
194    }
195
196    const fn has_pending(&self) -> bool {
197        self.active.len.is_some() || !self.queue.is_empty()
198    }
199}
200
201#[derive(Debug, Clone, Copy, PartialEq, Eq)]
202pub struct LoRaSpectrumSnapshot {
203    pub channel_busy_per_mille: u16,
204    pub noise_floor_dbm: Option<i16>,
205    pub cca_threshold_dbm: Option<i16>,
206    pub deferrals: u32,
207    pub false_preambles: u32,
208    pub contention_timeouts: u32,
209    pub duty_holds: u32,
210    pub duty_timeouts: u32,
211    pub radio_recoveries: u32,
212}
213
214/// Lock-free spectrum-stewardship diagnostics for one LoRa interface.
215pub struct LoRaSpectrumStatus {
216    channel_observations: AtomicU32,
217    busy_observations: AtomicU32,
218    sensing: AtomicU32,
219    deferrals: AtomicU32,
220    false_preambles: AtomicU32,
221    contention_timeouts: AtomicU32,
222    duty_holds: AtomicU32,
223    duty_timeouts: AtomicU32,
224    radio_recoveries: AtomicU32,
225}
226
227impl Default for LoRaSpectrumStatus {
228    fn default() -> Self {
229        Self::new()
230    }
231}
232
233impl LoRaSpectrumStatus {
234    #[must_use]
235    pub const fn new() -> Self {
236        Self {
237            channel_observations: AtomicU32::new(0),
238            busy_observations: AtomicU32::new(0),
239            sensing: AtomicU32::new(SENSING_UNPUBLISHED),
240            deferrals: AtomicU32::new(0),
241            false_preambles: AtomicU32::new(0),
242            contention_timeouts: AtomicU32::new(0),
243            duty_holds: AtomicU32::new(0),
244            duty_timeouts: AtomicU32::new(0),
245            radio_recoveries: AtomicU32::new(0),
246        }
247    }
248
249    #[must_use]
250    pub fn snapshot(&self) -> LoRaSpectrumSnapshot {
251        let observations = self.channel_observations.load(Ordering::Relaxed);
252        let busy = self.busy_observations.load(Ordering::Relaxed);
253        let channel_busy_per_mille = if observations == 0 {
254            0
255        } else {
256            busy.saturating_mul(1_000).saturating_div(observations) as u16
257        };
258        let sensing = self.sensing.load(Ordering::Relaxed);
259        let (noise_floor_dbm, cca_threshold_dbm) = if sensing == SENSING_UNPUBLISHED {
260            (None, None)
261        } else {
262            (
263                Some((sensing >> 16) as u16 as i16),
264                Some(sensing as u16 as i16),
265            )
266        };
267        LoRaSpectrumSnapshot {
268            channel_busy_per_mille,
269            noise_floor_dbm,
270            cca_threshold_dbm,
271            deferrals: self.deferrals.load(Ordering::Relaxed),
272            false_preambles: self.false_preambles.load(Ordering::Relaxed),
273            contention_timeouts: self.contention_timeouts.load(Ordering::Relaxed),
274            duty_holds: self.duty_holds.load(Ordering::Relaxed),
275            duty_timeouts: self.duty_timeouts.load(Ordering::Relaxed),
276            radio_recoveries: self.radio_recoveries.load(Ordering::Relaxed),
277        }
278    }
279
280    fn record_channel(&self, observation: ChannelObservation, noise: Option<&NoiseFloor>) {
281        match observation {
282            ChannelObservation::Clear => {
283                self.channel_observations.fetch_add(1, Ordering::Relaxed);
284            }
285            ChannelObservation::Busy => {
286                self.channel_observations.fetch_add(1, Ordering::Relaxed);
287                self.busy_observations.fetch_add(1, Ordering::Relaxed);
288            }
289            ChannelObservation::Unknown => {}
290        }
291        if let Some((floor, threshold)) =
292            noise.and_then(|noise| noise.noise_floor_dbm().zip(noise.cca_threshold_dbm()))
293        {
294            let packed = (u32::from(floor as u16) << 16) | u32::from(threshold as u16);
295            self.sensing.store(packed, Ordering::Relaxed);
296        }
297    }
298
299    fn add_deferrals(&self, count: u32) {
300        self.deferrals.fetch_add(count, Ordering::Relaxed);
301    }
302
303    fn add_false_preamble(&self) {
304        self.false_preambles.fetch_add(1, Ordering::Relaxed);
305    }
306
307    fn add_contention_timeout(&self) {
308        self.contention_timeouts.fetch_add(1, Ordering::Relaxed);
309    }
310
311    fn add_duty_hold(&self) {
312        self.duty_holds.fetch_add(1, Ordering::Relaxed);
313    }
314
315    fn add_duty_timeout(&self) {
316        self.duty_timeouts.fetch_add(1, Ordering::Relaxed);
317    }
318
319    fn add_radio_recovery(&self) {
320        self.radio_recoveries.fetch_add(1, Ordering::Relaxed);
321    }
322}
323
324struct ObservedAirFrame<'a> {
325    bytes: &'a [u8],
326    phy: PacketPhyStats,
327    spreading_factor: SpreadingFactor,
328    arrived_at: InstantMillis,
329}
330
331#[derive(Debug, Clone, Copy, PartialEq, Eq)]
332struct ChannelEvidence {
333    observation: ChannelObservation,
334    decoded_airtime_us: Option<u64>,
335}
336
337struct ReceivePath<'a, Seam> {
338    profile: &'a RadioProfile,
339    activity: &'a mut DemodulatorActivity,
340    spectrum: &'a LoRaSpectrumStatus,
341    rx_buf: &'a [u8],
342    status: &'a EmbassyInterfaceStatus,
343    throughput: &'a mut ThroughputLedger,
344    reassembler: &'a mut LoRaReassembler<LORA_MAX_PAYLOAD>,
345    seam: &'a mut Seam,
346}
347
348fn decoded_peer_airtime_us(event: RadioEvent, profile: &RadioProfile) -> Option<u64> {
349    match event {
350        RadioEvent::Frame(received) => Some(profile.time_on_air_us(received.len)),
351        RadioEvent::PreambleDetected
352        | RadioEvent::HeaderValid
353        | RadioEvent::HeaderError
354        | RadioEvent::CrcError
355        | RadioEvent::Timeout
356        | RadioEvent::Other => None,
357    }
358}
359
360async fn deliver_rx<Seam: InterfaceSeam>(
361    frame: ObservedAirFrame<'_>,
362    status: &EmbassyInterfaceStatus,
363    throughput: &mut ThroughputLedger,
364    reassembler: &mut LoRaReassembler<LORA_MAX_PAYLOAD>,
365    seam: &mut Seam,
366) {
367    status.add_rx(frame.bytes.len() as u64);
368    throughput.record_rx(frame.arrived_at, frame.bytes.len() as u64);
369    status.set_transfer_rates(throughput.rates());
370    if let Some(packet) = reassembler.feed_with_phy(frame.bytes, frame.phy) {
371        if !packet.bytes.is_empty() && packet.bytes.len() <= LORA_MAX_PAYLOAD {
372            let mut phy = packet.phy;
373            if let Some(snr) = phy.snr {
374                phy.quality = frame.spreading_factor.signal_quality(snr);
375            }
376            seam.next_inbound_with_phy(packet.bytes, phy).await;
377        }
378    }
379}
380
381fn choose_backoff_entropy<Seam: InterfaceSeam>(
382    access: &mut ChannelAccess,
383    seam: &mut Seam,
384) -> ChannelAccessAction {
385    let mut entropy = [0u8; 2];
386    loop {
387        seam.fill_entropy(&mut entropy);
388        if access.choose_backoff(u16::from_le_bytes(entropy)) {
389            return access.after_entropy();
390        }
391    }
392}
393
394async fn observe_radio_event<Seam: InterfaceSeam>(
395    event: RadioEvent,
396    now: InstantMillis,
397    receive: ReceivePath<'_, Seam>,
398) -> ChannelEvidence {
399    let decoded_airtime_us = decoded_peer_airtime_us(event, receive.profile);
400    let observation = match event {
401        RadioEvent::PreambleDetected => {
402            receive.activity.preamble_detected(now.0, *receive.profile);
403            ChannelObservation::Busy
404        }
405        RadioEvent::HeaderValid => {
406            receive.activity.header_valid(now.0, *receive.profile);
407            ChannelObservation::Busy
408        }
409        RadioEvent::Frame(received) => {
410            deliver_rx(
411                ObservedAirFrame {
412                    bytes: &receive.rx_buf[..received.len],
413                    phy: received.phy,
414                    spreading_factor: receive.profile.modulation.spreading_factor(),
415                    arrived_at: now,
416                },
417                receive.status,
418                receive.throughput,
419                receive.reassembler,
420                receive.seam,
421            )
422            .await;
423            receive.activity.frame_finished();
424            ChannelObservation::Busy
425        }
426        RadioEvent::HeaderError => {
427            receive.spectrum.add_false_preamble();
428            receive.activity.frame_finished();
429            ChannelObservation::Busy
430        }
431        RadioEvent::CrcError => {
432            receive.activity.frame_finished();
433            ChannelObservation::Busy
434        }
435        RadioEvent::Timeout => {
436            receive.activity.frame_finished();
437            ChannelObservation::Unknown
438        }
439        RadioEvent::Other => ChannelObservation::Unknown,
440    };
441    receive.spectrum.record_channel(observation, None);
442    ChannelEvidence {
443        observation,
444        decoded_airtime_us,
445    }
446}
447
448async fn sample_channel<SPI, BUSY, DIO1, RST, DLY>(
449    radio: &mut Sx126x<SPI, BUSY, DIO1, RST, DLY>,
450    now: InstantMillis,
451    activity: &mut DemodulatorActivity,
452    spectrum: &LoRaSpectrumStatus,
453    noise: &mut NoiseFloor,
454) -> Result<ChannelObservation, sx126x::Error>
455where
456    SPI: SpiDevice,
457    BUSY: Wait,
458    DIO1: Wait,
459    RST: OutputPin,
460    DLY: DelayNs,
461{
462    let (demodulator_busy, false_preamble) = activity.observe(now.0);
463    if false_preamble {
464        spectrum.add_false_preamble();
465    }
466    if demodulator_busy {
467        let observation = ChannelObservation::Busy;
468        spectrum.record_channel(observation, Some(noise));
469        return Ok(observation);
470    }
471    let rssi_dbm = radio.channel_rssi_dbm().await?;
472    let observation = noise.observe(now.0, rssi_dbm, false);
473    spectrum.record_channel(observation, Some(noise));
474    Ok(observation)
475}
476
477#[derive(Debug, Clone, Copy, PartialEq, Eq)]
478pub enum LoRaApplyOutcome {
479    Applied,
480    Rejected,
481}
482
483#[derive(Debug, Clone, Copy, PartialEq, Eq)]
484struct LoRaApplyRequest {
485    id: u32,
486    profile: RadioProfile,
487}
488
489#[derive(Debug, Clone, Copy, PartialEq, Eq)]
490struct LoRaApplyResult {
491    id: u32,
492    outcome: LoRaApplyOutcome,
493}
494
495/// The app-side control for reconfiguring a running radio.
496///
497/// [`signal`](Self::signal) preserves the original fire-and-forget behavior. An app that must
498/// persist only an accepted profile uses [`apply`](Self::apply) and awaits the matching worker
499/// result. Callers serialize awaited requests; Hopspot's UI event loop naturally does so.
500pub struct LoRaControl {
501    requests: Signal<CriticalSectionRawMutex, LoRaApplyRequest>,
502    results: Signal<CriticalSectionRawMutex, LoRaApplyResult>,
503    next_id: AtomicU32,
504}
505
506impl LoRaControl {
507    #[must_use]
508    pub const fn new() -> Self {
509        Self {
510            requests: Signal::new(),
511            results: Signal::new(),
512            next_id: AtomicU32::new(1),
513        }
514    }
515
516    pub fn signal(&self, profile: RadioProfile) {
517        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
518        self.requests.signal(LoRaApplyRequest { id, profile });
519    }
520
521    pub async fn apply(&self, profile: RadioProfile) -> LoRaApplyOutcome {
522        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
523        self.requests.signal(LoRaApplyRequest { id, profile });
524        loop {
525            let result = self.results.wait().await;
526            if result.id == id {
527                return result.outcome;
528            }
529        }
530    }
531
532    async fn wait(&self) -> LoRaApplyRequest {
533        self.requests.wait().await
534    }
535
536    fn complete(&self, id: u32, applied: bool) {
537        self.results.signal(LoRaApplyResult {
538            id,
539            outcome: if applied {
540                LoRaApplyOutcome::Applied
541            } else {
542                LoRaApplyOutcome::Rejected
543            },
544        });
545    }
546}
547
548impl Default for LoRaControl {
549    fn default() -> Self {
550        Self::new()
551    }
552}
553
554fn sx126x_config(profile: &RadioProfile) -> sx126x::RadioConfig {
555    let Modulation::Lora {
556        spreading_factor,
557        bandwidth,
558        coding_rate,
559    } = profile.modulation;
560    let spreading_factor = match spreading_factor {
561        SpreadingFactor::Sf5 => sx126x::SpreadingFactor::Sf5,
562        SpreadingFactor::Sf6 => sx126x::SpreadingFactor::Sf6,
563        SpreadingFactor::Sf7 => sx126x::SpreadingFactor::Sf7,
564        SpreadingFactor::Sf8 => sx126x::SpreadingFactor::Sf8,
565        SpreadingFactor::Sf9 => sx126x::SpreadingFactor::Sf9,
566        SpreadingFactor::Sf10 => sx126x::SpreadingFactor::Sf10,
567        SpreadingFactor::Sf11 => sx126x::SpreadingFactor::Sf11,
568        SpreadingFactor::Sf12 => sx126x::SpreadingFactor::Sf12,
569    };
570    let bandwidth = match bandwidth {
571        LoraBandwidth::Bw125kHz => sx126x::Bandwidth::Bw125,
572        LoraBandwidth::Bw250kHz => sx126x::Bandwidth::Bw250,
573        LoraBandwidth::Bw500kHz => sx126x::Bandwidth::Bw500,
574    };
575    let coding_rate = match coding_rate {
576        CodingRate::Cr45 => sx126x::CodingRate::Cr4_5,
577        CodingRate::Cr46 => sx126x::CodingRate::Cr4_6,
578        CodingRate::Cr47 => sx126x::CodingRate::Cr4_7,
579        CodingRate::Cr48 => sx126x::CodingRate::Cr4_8,
580    };
581    sx126x::RadioConfig {
582        frequency_hz: profile.frequency.hz(),
583        modulation: sx126x::Modulation::Lora {
584            spreading_factor,
585            bandwidth,
586            coding_rate,
587        },
588        packet: sx126x::LoraPacket {
589            preamble_symbols: profile.preamble.count(),
590            explicit_header: true,
591            crc_on: true,
592            invert_iq: false,
593        },
594        sync_word: RNODE_LORA_SYNC_WORD,
595        tx_power_dbm: profile.tx_power.dbm(),
596    }
597}
598
599/// The [`Retag`](InterfaceLifecycle::Retag) a reconfigure to `new_profile` warrants, or `None` when the change leaves the channel identity untouched — a local knob like transmit power or preamble. The channel_tag (frequency + modulation) is what mints the id, so only a change to it re-keys.
600fn retag_message(
601    current_id: InterfaceId,
602    new_profile: &RadioProfile,
603    duty: Option<AirtimeDutyCycle>,
604) -> Option<InterfaceLifecycle> {
605    let new_id =
606        InterfaceId::from_channel_tag(InterfaceKind::LoRa, &lora::channel_tag(new_profile));
607    (new_id != current_id).then(|| InterfaceLifecycle::Retag {
608        old_id: current_id,
609        new_id,
610        descriptor: lora::descriptor(new_id, new_profile, duty),
611    })
612}
613
614fn packet_airtime(packet: &[u8], profile: &RadioProfile) -> u64 {
615    let mut scratch = [0u8; LORA_SINGLE_FRAME_MAX];
616    let mut total = 0;
617    for index in 0..air_frame_count(packet.len()) {
618        if let Ok(n) = encode_air_frame_part(packet, 0, index, &mut scratch) {
619            total += profile.time_on_air_us(n);
620        }
621    }
622    total
623}
624
625fn take_contention_priority(
626    continuation: &mut bool,
627    airtime: &mut AirtimeLedger,
628    now: InstantMillis,
629) -> ContentionPriority {
630    if core::mem::take(continuation) {
631        ContentionPriority::Continuation
632    } else {
633        ContentionPriority::Fresh {
634            short_airtime_per_mille: airtime.utilization(now).short_per_mille,
635        }
636    }
637}
638
639#[expect(
640    clippy::too_many_arguments,
641    reason = "embedded serve-loop internals pass the loop's split-borrowed locals; bundling awaits an on-hardware validation pass"
642)]
643async fn transmit_packet<SPI, BUSY, DIO1, RST, DLY>(
644    radio: &mut Sx126x<SPI, BUSY, DIO1, RST, DLY>,
645    packet: &[u8],
646    seq: &mut u8,
647    airtime: &mut AirtimeLedger,
648    throughput: &mut ThroughputLedger,
649    status: &EmbassyInterfaceStatus,
650    profile: &RadioProfile,
651    started: &Instant,
652    tx_frame: &mut [u8; LORA_SINGLE_FRAME_MAX],
653) -> Result<(), sx126x::Error>
654where
655    SPI: SpiDevice,
656    BUSY: Wait,
657    DIO1: Wait,
658    RST: OutputPin,
659    DLY: DelayNs,
660{
661    for index in 0..air_frame_count(packet.len()) {
662        let n = match encode_air_frame_part(packet, *seq, index, tx_frame) {
663            Ok(n) => n,
664            Err(e) => {
665                crate::diagnostic_log::debug!("RNS_LORA frame {index} encode failed: {e:?}");
666                *seq = seq.wrapping_add(0x10);
667                return Err(sx126x::Error::BufferTooSmall);
668            }
669        };
670        if let Err(e) = radio.transmit(&tx_frame[..n]).await {
671            crate::diagnostic_log::debug!("RNS_LORA tx failed: {e:?}");
672            *seq = seq.wrapping_add(0x10);
673            return Err(e);
674        }
675        let completed_at = InstantMillis(started.elapsed().as_millis());
676        status.add_tx(n as u64);
677        throughput.record_tx(completed_at, n as u64);
678        status.set_transfer_rates(throughput.rates());
679        status.set_airtime(airtime.record_tx(completed_at, profile.time_on_air_us(n)));
680    }
681    *seq = seq.wrapping_add(0x10);
682    Ok(())
683}
684
685fn is_radio_fault(e: &sx126x::Error) -> bool {
686    matches!(
687        e,
688        sx126x::Error::Busy
689            | sx126x::Error::Dio1
690            | sx126x::Error::Spi
691            | sx126x::Error::Timeout
692            | sx126x::Error::Reset
693    )
694}
695
696async fn reinit_radio<SPI, BUSY, DIO1, RST, DLY>(
697    radio: &mut Sx126x<SPI, BUSY, DIO1, RST, DLY>,
698    profile: &RadioProfile,
699    spectrum: &LoRaSpectrumStatus,
700) -> bool
701where
702    SPI: SpiDevice,
703    BUSY: Wait,
704    DIO1: Wait,
705    RST: OutputPin,
706    DLY: DelayNs,
707{
708    if let Err(e) = radio.init(sx126x_config(profile)).await {
709        crate::diagnostic_log::warn!("RNS_LORA hard re-init failed: {e:?}");
710        return false;
711    }
712    if let Err(e) = radio.arm_rx().await {
713        crate::diagnostic_log::warn!("RNS_LORA re-init RX arm failed: {e:?}");
714        return false;
715    }
716    crate::diagnostic_log::warn!("RNS_LORA radio recovered via hard re-init");
717    spectrum.add_radio_recovery();
718    true
719}
720
721#[expect(
722    clippy::too_many_arguments,
723    reason = "transactional radio reconfiguration owns the full old/new policy boundary"
724)]
725async fn apply_profile<SPI, BUSY, DIO1, RST, DLY>(
726    radio: &mut Sx126x<SPI, BUSY, DIO1, RST, DLY>,
727    requested: RadioProfile,
728    airtime_policy: AirtimePolicy,
729    profile: &mut RadioProfile,
730    duty: &mut Option<AirtimeDutyCycle>,
731    current_id: &mut InterfaceId,
732    status: &EmbassyInterfaceStatus,
733    spectrum: &LoRaSpectrumStatus,
734    lifecycle: DynamicSender<'_, InterfaceLifecycle>,
735) -> bool
736where
737    SPI: SpiDevice,
738    BUSY: Wait,
739    DIO1: Wait,
740    RST: OutputPin,
741    DLY: DelayNs,
742{
743    if requested.validate().is_err() {
744        crate::diagnostic_log::warn!("RNS_LORA rejected invalid profile");
745        return false;
746    }
747    let requested_duty = match airtime_policy.resolve(requested.region) {
748        Ok(duty) => duty,
749        Err(_) => {
750            crate::diagnostic_log::warn!("RNS_LORA rejected airtime policy");
751            return false;
752        }
753    };
754    if requested == *profile && requested_duty == *duty {
755        return true;
756    }
757
758    let previous = *profile;
759    if let Err(error) = radio.init(sx126x_config(&requested)).await {
760        crate::diagnostic_log::warn!(
761            "RNS_LORA reconfigure init failed: {error:?}; restoring prior profile"
762        );
763        if !reinit_radio(radio, &previous, spectrum).await {
764            status.set_connection(ConnectionState::Disconnected);
765        }
766        return false;
767    }
768    if let Err(error) = radio.arm_rx().await {
769        crate::diagnostic_log::warn!(
770            "RNS_LORA reconfigure RX arm failed: {error:?}; restoring prior profile"
771        );
772        if !reinit_radio(radio, &previous, spectrum).await {
773            status.set_connection(ConnectionState::Disconnected);
774        }
775        return false;
776    }
777
778    *profile = requested;
779    *duty = requested_duty;
780    if let Some(message) = retag_message(*current_id, profile, requested_duty) {
781        if let InterfaceLifecycle::Retag { new_id, .. } = &message {
782            *current_id = *new_id;
783            status.set_id(*new_id);
784        }
785        lifecycle.send(message).await;
786    } else {
787        lifecycle
788            .send(InterfaceLifecycle::Update {
789                descriptor: lora::descriptor(*current_id, profile, requested_duty),
790            })
791            .await;
792    }
793    true
794}
795
796#[derive(Debug, Clone, Copy, PartialEq, Eq)]
797pub enum LoRaConfigError {
798    Profile(RadioProfileError),
799    AirtimePolicy(AirtimePolicyError),
800}
801
802pub struct LoRaInterfaceInput<'a, SPI, BUSY, DIO1, RST, DLY> {
803    pub radio: Sx126x<SPI, BUSY, DIO1, RST, DLY>,
804    pub profile: RadioProfile,
805    pub airtime_policy: AirtimePolicy,
806    pub tx_queue: &'a mut [u8],
807    pub control: &'a LoRaControl,
808    pub status: &'a EmbassyInterfaceStatus,
809    pub spectrum: &'a LoRaSpectrumStatus,
810    pub lifecycle: DynamicSender<'a, InterfaceLifecycle>,
811}
812
813pub struct LoRaInterface<'a, SPI, BUSY, DIO1, RST, DLY> {
814    id: InterfaceId,
815    radio: Sx126x<SPI, BUSY, DIO1, RST, DLY>,
816    profile: RadioProfile,
817    airtime_policy: AirtimePolicy,
818    duty: Option<AirtimeDutyCycle>,
819    tag: HeaplessVec<u8, CHANNEL_TAG_CAP>,
820    tx_queue: &'a mut [u8],
821    control: &'a LoRaControl,
822    status: &'a EmbassyInterfaceStatus,
823    spectrum: &'a LoRaSpectrumStatus,
824    lifecycle: DynamicSender<'a, InterfaceLifecycle>,
825}
826
827impl<'a, SPI, BUSY, DIO1, RST, DLY> LoRaInterface<'a, SPI, BUSY, DIO1, RST, DLY> {
828    /// The id a radio on `profile` will carry — for the caller that stands its [`EmbassyInterfaceStatus`] up under the same key before building the interface.
829    #[must_use]
830    pub fn interface_id(profile: &RadioProfile) -> InterfaceId {
831        InterfaceId::from_channel_tag(InterfaceKind::LoRa, &lora::channel_tag(profile))
832    }
833
834    pub fn new(
835        input: LoRaInterfaceInput<'a, SPI, BUSY, DIO1, RST, DLY>,
836    ) -> Result<Self, LoRaConfigError> {
837        let LoRaInterfaceInput {
838            radio,
839            profile,
840            airtime_policy,
841            tx_queue,
842            control,
843            status,
844            spectrum,
845            lifecycle,
846        } = input;
847        profile.validate().map_err(LoRaConfigError::Profile)?;
848        let tag = lora::channel_tag(&profile);
849        let id = Self::interface_id(&profile);
850        let duty = airtime_policy
851            .resolve(profile.region)
852            .map_err(LoRaConfigError::AirtimePolicy)?;
853        Ok(Self {
854            id,
855            radio,
856            profile,
857            airtime_policy,
858            duty,
859            tag,
860            tx_queue,
861            control,
862            status,
863            spectrum,
864            lifecycle,
865        })
866    }
867
868    #[must_use]
869    pub fn id(&self) -> InterfaceId {
870        self.id
871    }
872}
873
874impl<SPI, BUSY, DIO1, RST, DLY> Interface for LoRaInterface<'_, SPI, BUSY, DIO1, RST, DLY>
875where
876    SPI: SpiDevice,
877    BUSY: Wait,
878    DIO1: Wait,
879    RST: OutputPin,
880    DLY: DelayNs,
881{
882    const HW_MTU: usize = LORA_MAX_PAYLOAD;
883    const KIND: InterfaceKind = InterfaceKind::LoRa;
884
885    fn descriptor(&self) -> InterfaceDescriptor {
886        lora::descriptor(self.id, &self.profile, self.duty)
887    }
888
889    fn channel_tag(&self) -> &[u8] {
890        &self.tag
891    }
892
893    async fn run<Seam: InterfaceSeam>(self, mut seam: Seam) {
894        let LoRaInterface {
895            id,
896            mut radio,
897            mut profile,
898            airtime_policy,
899            duty,
900            tag: _,
901            tx_queue,
902            control,
903            status,
904            spectrum,
905            lifecycle,
906        } = self;
907        let mut current_id = id;
908
909        if let Err(e) = radio.init(sx126x_config(&profile)).await {
910            crate::diagnostic_log::error!("RNS_LORA radio init failed: {e:?}; interface offline");
911            status.set_connection(ConnectionState::Disconnected);
912            return;
913        }
914
915        let mut reassembler = LoRaReassembler::<LORA_MAX_PAYLOAD>::new();
916        let mut rx_buf = [0u8; LORA_SINGLE_FRAME_MAX];
917        let mut tx_frame = [0u8; LORA_SINGLE_FRAME_MAX];
918        let mut seq: u8 = 0;
919        let mut airtime = AirtimeLedger::new();
920        let mut throughput = ThroughputLedger::new();
921        let mut duty_cycle = duty;
922        let mut noise = NoiseFloor::new();
923        let mut activity = DemodulatorActivity::new();
924        let started = Instant::now();
925        status.set_connection(ConnectionState::Connected);
926        if let Err(e) = radio.arm_rx().await {
927            crate::diagnostic_log::debug!("RNS_LORA initial RX arm failed: {e:?}");
928        }
929
930        let mut backlog = TransmitBacklog::new(tx_queue);
931        let mut access: Option<ChannelAccess> = None;
932        let mut service_age = ServiceAge::new(profile);
933        let mut continuation = false;
934        let mut access_suspended = true;
935        let mut duty_was_held = false;
936        let mut reported_deferrals = 0u32;
937
938        loop {
939            if !status.is_enabled() {
940                status.set_connection(ConnectionState::Disabled);
941                reassembler = LoRaReassembler::new();
942                activity.frame_finished();
943                noise = NoiseFloor::new();
944                service_age.reset(profile);
945                continuation = false;
946                if backlog.active.clear() {
947                    access = None;
948                    seam.complete_outbound(OutboundDisposition::Dropped(
949                        OutboundDropReason::Disabled,
950                    ));
951                }
952                status.wait_until_enabled().await;
953                status.set_connection(ConnectionState::Connected);
954                if let Err(e) = radio.arm_rx().await {
955                    crate::diagnostic_log::debug!("RNS_LORA RX re-arm after enable failed: {e:?}");
956                    if is_radio_fault(&e) {
957                        reinit_radio(&mut radio, &profile, spectrum).await;
958                    }
959                }
960            }
961
962            let activation_time = InstantMillis(started.elapsed().as_millis());
963            if backlog.activate_next(&profile, activation_time.0) {
964                let priority =
965                    take_contention_priority(&mut continuation, &mut airtime, activation_time);
966                access = backlog
967                    .active
968                    .channel_access(profile, activation_time.0, priority);
969                access_suspended = true;
970                duty_was_held = false;
971                reported_deferrals = 0;
972            }
973
974            if backlog.active.len.is_some() {
975                let before_wait = InstantMillis(started.elapsed().as_millis());
976                let projected =
977                    airtime.projected_utilization(before_wait, backlog.active.airtime_us);
978                let duty_permits = duty_cycle.is_none_or(|duty| duty.permits(projected));
979                if !duty_permits && !duty_was_held {
980                    spectrum.add_duty_hold();
981                }
982                duty_was_held = !duty_permits;
983                let suspended_now = !duty_permits;
984                if access_suspended && !suspended_now {
985                    if let Some(access) = access.as_mut() {
986                        access.restart_contention(before_wait.0);
987                    }
988                }
989                access_suspended = suspended_now;
990
991                let expired = access
992                    .as_ref()
993                    .is_some_and(|access| access.is_expired(before_wait.0));
994                if expired {
995                    let reason = if duty_permits {
996                        spectrum.add_contention_timeout();
997                        OutboundDropReason::ContentionTimeout
998                    } else {
999                        spectrum.add_duty_timeout();
1000                        OutboundDropReason::DutyLimited
1001                    };
1002                    backlog.active.clear();
1003                    access = None;
1004                    seam.complete_outbound(OutboundDisposition::Dropped(reason));
1005                    if backlog.queue.is_empty() {
1006                        service_age.consume();
1007                        continuation = false;
1008                    }
1009                    continue;
1010                }
1011
1012                let ordinary_tick_ms = if access_suspended {
1013                    IDLE_TICK.as_millis()
1014                } else {
1015                    access.as_ref().map_or(IDLE_TICK.as_millis(), |access| {
1016                        access.next_poll_ms(service_age.backoff_rate())
1017                    })
1018                };
1019                let tick_ms = activity.next_poll_ms(before_wait.0, ordinary_tick_ms);
1020
1021                let mut next_action = None;
1022                let can_accept_outbound = backlog.can_accept_outbound();
1023                match select5(
1024                    control.wait(),
1025                    status.wait_until_disabled(),
1026                    radio.read_event(&mut rx_buf),
1027                    Timer::after(Duration::from_millis(tick_ms)),
1028                    async {
1029                        if can_accept_outbound {
1030                            return Some(seam.next_outbound().await);
1031                        }
1032                        core::future::pending::<Option<&[u8]>>().await
1033                    },
1034                )
1035                .await
1036                {
1037                    Either5::First(request) => {
1038                        let changed = apply_profile(
1039                            &mut radio,
1040                            request.profile,
1041                            airtime_policy,
1042                            &mut profile,
1043                            &mut duty_cycle,
1044                            &mut current_id,
1045                            status,
1046                            spectrum,
1047                            lifecycle,
1048                        )
1049                        .await;
1050                        if changed {
1051                            let now = InstantMillis(started.elapsed().as_millis());
1052                            reassembler = LoRaReassembler::new();
1053                            activity.frame_finished();
1054                            noise = NoiseFloor::new();
1055                            service_age.reset(profile);
1056                            continuation = false;
1057                            backlog.active.recompute_airtime(&profile);
1058                            let priority =
1059                                take_contention_priority(&mut continuation, &mut airtime, now);
1060                            access = backlog.active.channel_access(profile, now.0, priority);
1061                            access_suspended = true;
1062                            duty_was_held = false;
1063                            reported_deferrals = 0;
1064                        }
1065                        control.complete(request.id, changed);
1066                    }
1067                    Either5::Second(()) => continue,
1068                    Either5::Third(Ok(event)) => {
1069                        let now = InstantMillis(started.elapsed().as_millis());
1070                        let evidence = observe_radio_event(
1071                            event,
1072                            now,
1073                            ReceivePath {
1074                                profile: &profile,
1075                                activity: &mut activity,
1076                                spectrum,
1077                                rx_buf: &rx_buf,
1078                                status,
1079                                throughput: &mut throughput,
1080                                reassembler: &mut reassembler,
1081                                seam: &mut seam,
1082                            },
1083                        )
1084                        .await;
1085                        if backlog.has_pending() {
1086                            if let Some(decoded_airtime_us) = evidence.decoded_airtime_us {
1087                                service_age.record_peer_airtime(decoded_airtime_us);
1088                            }
1089                        }
1090                        if !access_suspended {
1091                            if let Some(access) = access.as_mut() {
1092                                let action = access.observe(
1093                                    now.0,
1094                                    evidence.observation,
1095                                    service_age.backoff_rate(),
1096                                );
1097                                next_action = Some(
1098                                    if matches!(action, ChannelAccessAction::NeedBackoffEntropy) {
1099                                        choose_backoff_entropy(access, &mut seam)
1100                                    } else {
1101                                        action
1102                                    },
1103                                );
1104                            }
1105                        }
1106                    }
1107                    Either5::Third(Err(error)) => {
1108                        crate::diagnostic_log::debug!("RNS_LORA rx event error: {error:?}");
1109                        activity.frame_finished();
1110                        if is_radio_fault(&error) {
1111                            reinit_radio(&mut radio, &profile, spectrum).await;
1112                        }
1113                        if !access_suspended {
1114                            let now = InstantMillis(started.elapsed().as_millis());
1115                            if let Some(access) = access.as_mut() {
1116                                next_action = Some(access.observe(
1117                                    now.0,
1118                                    noise.fail_closed(),
1119                                    service_age.backoff_rate(),
1120                                ));
1121                            }
1122                        }
1123                    }
1124                    Either5::Fourth(()) => {
1125                        let now = InstantMillis(started.elapsed().as_millis());
1126                        let observation = match sample_channel(
1127                            &mut radio,
1128                            now,
1129                            &mut activity,
1130                            spectrum,
1131                            &mut noise,
1132                        )
1133                        .await
1134                        {
1135                            Ok(observation) => observation,
1136                            Err(error) => {
1137                                crate::diagnostic_log::debug!(
1138                                    "RNS_LORA channel sample failed: {error:?}"
1139                                );
1140                                activity.frame_finished();
1141                                if is_radio_fault(&error) {
1142                                    reinit_radio(&mut radio, &profile, spectrum).await;
1143                                }
1144                                noise.fail_closed()
1145                            }
1146                        };
1147                        if !access_suspended {
1148                            if let Some(access) = access.as_mut() {
1149                                let action =
1150                                    access.observe(now.0, observation, service_age.backoff_rate());
1151                                next_action = Some(
1152                                    if matches!(action, ChannelAccessAction::NeedBackoffEntropy) {
1153                                        choose_backoff_entropy(access, &mut seam)
1154                                    } else {
1155                                        action
1156                                    },
1157                                );
1158                            }
1159                        }
1160                    }
1161                    Either5::Fifth(Some(outbound)) => {
1162                        let now = InstantMillis(started.elapsed().as_millis());
1163                        match backlog.accept(outbound, &profile, now.0) {
1164                            Ok(PacketPlacement::Queued) => seam.accept_outbound_custody(),
1165                            Ok(PacketPlacement::Active) => {
1166                                seam.accept_outbound_custody();
1167                                let priority =
1168                                    take_contention_priority(&mut continuation, &mut airtime, now);
1169                                access = backlog.active.channel_access(profile, now.0, priority);
1170                                access_suspended = true;
1171                                duty_was_held = false;
1172                                reported_deferrals = 0;
1173                            }
1174                            Err(TransmitQueueError::Full | TransmitQueueError::PacketTooLarge) => {
1175                                seam.complete_outbound(OutboundDisposition::Dropped(
1176                                    OutboundDropReason::Rejected,
1177                                ));
1178                            }
1179                        }
1180                    }
1181                    Either5::Fifth(None) => {}
1182                }
1183
1184                if matches!(next_action, Some(ChannelAccessAction::ReadyForFinalCheck)) {
1185                    let now = InstantMillis(started.elapsed().as_millis());
1186                    let evidence = match radio.poll_event(&mut rx_buf).await {
1187                        Ok(Some(event)) => {
1188                            observe_radio_event(
1189                                event,
1190                                now,
1191                                ReceivePath {
1192                                    profile: &profile,
1193                                    activity: &mut activity,
1194                                    spectrum,
1195                                    rx_buf: &rx_buf,
1196                                    status,
1197                                    throughput: &mut throughput,
1198                                    reassembler: &mut reassembler,
1199                                    seam: &mut seam,
1200                                },
1201                            )
1202                            .await
1203                        }
1204                        Ok(None) => match sample_channel(
1205                            &mut radio,
1206                            now,
1207                            &mut activity,
1208                            spectrum,
1209                            &mut noise,
1210                        )
1211                        .await
1212                        {
1213                            Ok(observation) => ChannelEvidence {
1214                                observation,
1215                                decoded_airtime_us: None,
1216                            },
1217                            Err(error) => {
1218                                crate::diagnostic_log::debug!(
1219                                    "RNS_LORA final channel check failed: {error:?}"
1220                                );
1221                                activity.frame_finished();
1222                                if is_radio_fault(&error) {
1223                                    reinit_radio(&mut radio, &profile, spectrum).await;
1224                                }
1225                                ChannelEvidence {
1226                                    observation: noise.fail_closed(),
1227                                    decoded_airtime_us: None,
1228                                }
1229                            }
1230                        },
1231                        Err(error) => {
1232                            crate::diagnostic_log::debug!(
1233                                "RNS_LORA final IRQ check failed: {error:?}"
1234                            );
1235                            activity.frame_finished();
1236                            if is_radio_fault(&error) {
1237                                reinit_radio(&mut radio, &profile, spectrum).await;
1238                            }
1239                            ChannelEvidence {
1240                                observation: noise.fail_closed(),
1241                                decoded_airtime_us: None,
1242                            }
1243                        }
1244                    };
1245                    if let Some(decoded_airtime_us) = evidence.decoded_airtime_us {
1246                        service_age.record_peer_airtime(decoded_airtime_us);
1247                    }
1248                    if let Some(access) = access.as_mut() {
1249                        next_action = Some(access.final_check(now.0, evidence.observation));
1250                    }
1251                }
1252
1253                if let Some(access) = access.as_ref() {
1254                    let deferrals = access.deferrals();
1255                    if deferrals > reported_deferrals {
1256                        spectrum.add_deferrals(deferrals - reported_deferrals);
1257                        reported_deferrals = deferrals;
1258                    }
1259                }
1260
1261                match next_action {
1262                    Some(ChannelAccessAction::Transmit) => {
1263                        service_age.consume();
1264                        let quantum = service_age.quantum();
1265                        let mut txop_airtime_us = 0u64;
1266                        let mut quantum_limited = false;
1267                        let mut transmission_failed = false;
1268
1269                        while let Some(active_len) = backlog.active.len {
1270                            let packet_airtime_us = backlog.active.airtime_us;
1271                            if !quantum.permits(txop_airtime_us, packet_airtime_us) {
1272                                quantum_limited = true;
1273                                break;
1274                            }
1275
1276                            let packet_start = InstantMillis(started.elapsed().as_millis());
1277                            let projected =
1278                                airtime.projected_utilization(packet_start, packet_airtime_us);
1279                            if duty_cycle.is_some_and(|duty| !duty.permits(projected)) {
1280                                if !duty_was_held {
1281                                    spectrum.add_duty_hold();
1282                                }
1283                                duty_was_held = true;
1284                                break;
1285                            }
1286
1287                            let tx = transmit_packet(
1288                                &mut radio,
1289                                &backlog.active.bytes[..active_len],
1290                                &mut seq,
1291                                &mut airtime,
1292                                &mut throughput,
1293                                status,
1294                                &profile,
1295                                &started,
1296                                &mut tx_frame,
1297                            )
1298                            .await;
1299                            let disposition = match tx {
1300                                Ok(()) => {
1301                                    txop_airtime_us =
1302                                        txop_airtime_us.saturating_add(packet_airtime_us);
1303                                    OutboundDisposition::Sent
1304                                }
1305                                Err(error) => {
1306                                    transmission_failed = true;
1307                                    if is_radio_fault(&error) {
1308                                        reinit_radio(&mut radio, &profile, spectrum).await;
1309                                    }
1310                                    OutboundDisposition::Dropped(
1311                                        OutboundDropReason::TransportFailure,
1312                                    )
1313                                }
1314                            };
1315                            backlog.active.clear();
1316                            seam.complete_outbound(disposition);
1317
1318                            if transmission_failed {
1319                                break;
1320                            }
1321                            let activated_at = InstantMillis(started.elapsed().as_millis());
1322                            if !backlog.activate_next(&profile, activated_at.0) {
1323                                break;
1324                            }
1325                        }
1326
1327                        if let Err(error) = radio.arm_rx().await {
1328                            crate::diagnostic_log::debug!(
1329                                "RNS_LORA RX re-arm after tx failed: {error:?}"
1330                            );
1331                            if is_radio_fault(&error) {
1332                                reinit_radio(&mut radio, &profile, spectrum).await;
1333                            }
1334                        }
1335                        activity.frame_finished();
1336                        access = None;
1337                        access_suspended = true;
1338                        reported_deferrals = 0;
1339
1340                        if backlog.active.len.is_none() {
1341                            let activated_at = InstantMillis(started.elapsed().as_millis());
1342                            let _ = backlog.activate_next(&profile, activated_at.0);
1343                        }
1344                        if backlog.active.len.is_some() {
1345                            if quantum_limited {
1346                                service_age.seed_continuation();
1347                                continuation = true;
1348                            }
1349                            let now = InstantMillis(started.elapsed().as_millis());
1350                            let priority =
1351                                take_contention_priority(&mut continuation, &mut airtime, now);
1352                            access = backlog.active.channel_access(profile, now.0, priority);
1353                        } else {
1354                            service_age.consume();
1355                            continuation = false;
1356                            duty_was_held = false;
1357                        }
1358                    }
1359                    Some(ChannelAccessAction::Expired) => {
1360                        spectrum.add_contention_timeout();
1361                        backlog.active.clear();
1362                        access = None;
1363                        seam.complete_outbound(OutboundDisposition::Dropped(
1364                            OutboundDropReason::ContentionTimeout,
1365                        ));
1366                        if backlog.queue.is_empty() {
1367                            service_age.consume();
1368                            continuation = false;
1369                        }
1370                    }
1371                    Some(
1372                        ChannelAccessAction::Wait
1373                        | ChannelAccessAction::NeedBackoffEntropy
1374                        | ChannelAccessAction::ReadyForFinalCheck,
1375                    )
1376                    | None => {}
1377                }
1378            } else {
1379                let ordinary_idle_tick_ms = if noise.is_calibrated() {
1380                    IDLE_TICK.as_millis()
1381                } else {
1382                    ChannelTiming::for_profile(profile).sample_ms()
1383                };
1384                let now_ms = started.elapsed().as_millis();
1385                let idle_tick =
1386                    Duration::from_millis(activity.next_poll_ms(now_ms, ordinary_idle_tick_ms));
1387                match select4(
1388                    control.wait(),
1389                    status.wait_until_disabled(),
1390                    radio.read_event(&mut rx_buf),
1391                    select(seam.next_outbound(), Timer::after(idle_tick)),
1392                )
1393                .await
1394                {
1395                    Either4::First(request) => {
1396                        let changed = apply_profile(
1397                            &mut radio,
1398                            request.profile,
1399                            airtime_policy,
1400                            &mut profile,
1401                            &mut duty_cycle,
1402                            &mut current_id,
1403                            status,
1404                            spectrum,
1405                            lifecycle,
1406                        )
1407                        .await;
1408                        if changed {
1409                            reassembler = LoRaReassembler::new();
1410                            activity.frame_finished();
1411                            noise = NoiseFloor::new();
1412                            service_age.reset(profile);
1413                            continuation = false;
1414                        }
1415                        control.complete(request.id, changed);
1416                    }
1417                    Either4::Second(()) => continue,
1418                    Either4::Third(Ok(event)) => {
1419                        let now = InstantMillis(started.elapsed().as_millis());
1420                        let _ = observe_radio_event(
1421                            event,
1422                            now,
1423                            ReceivePath {
1424                                profile: &profile,
1425                                activity: &mut activity,
1426                                spectrum,
1427                                rx_buf: &rx_buf,
1428                                status,
1429                                throughput: &mut throughput,
1430                                reassembler: &mut reassembler,
1431                                seam: &mut seam,
1432                            },
1433                        )
1434                        .await;
1435                    }
1436                    Either4::Third(Err(e)) => {
1437                        crate::diagnostic_log::debug!("RNS_LORA rx event error: {e:?}");
1438                        activity.frame_finished();
1439                        if is_radio_fault(&e) {
1440                            reinit_radio(&mut radio, &profile, spectrum).await;
1441                        }
1442                    }
1443                    Either4::Fourth(Either::First(outbound)) => {
1444                        let now = InstantMillis(started.elapsed().as_millis());
1445                        match backlog.accept(outbound, &profile, now.0) {
1446                            Ok(PacketPlacement::Active) => {
1447                                seam.accept_outbound_custody();
1448                                let priority =
1449                                    take_contention_priority(&mut continuation, &mut airtime, now);
1450                                access = backlog.active.channel_access(profile, now.0, priority);
1451                                access_suspended = true;
1452                                duty_was_held = false;
1453                                reported_deferrals = 0;
1454                            }
1455                            Ok(PacketPlacement::Queued) => seam.accept_outbound_custody(),
1456                            Err(TransmitQueueError::Full | TransmitQueueError::PacketTooLarge) => {
1457                                seam.complete_outbound(OutboundDisposition::Dropped(
1458                                    OutboundDropReason::Rejected,
1459                                ));
1460                            }
1461                        }
1462                    }
1463                    Either4::Fourth(Either::Second(())) => {
1464                        let now = InstantMillis(started.elapsed().as_millis());
1465                        if let Err(error) =
1466                            sample_channel(&mut radio, now, &mut activity, spectrum, &mut noise)
1467                                .await
1468                        {
1469                            crate::diagnostic_log::debug!(
1470                                "RNS_LORA idle channel sample failed: {error:?}"
1471                            );
1472                            activity.frame_finished();
1473                            if is_radio_fault(&error) {
1474                                reinit_radio(&mut radio, &profile, spectrum).await;
1475                            }
1476                        }
1477                    }
1478                }
1479            }
1480        }
1481    }
1482}
1483
1484#[cfg(test)]
1485mod tests {
1486    use super::*;
1487    use core::future::Future;
1488    use core::task::{Context, Poll};
1489    use embassy_futures::join::join;
1490    use prns_core::interfaces::lora::{PreambleSymbols, TxPower, DEFAULT_915_PROFILE};
1491    use std::boxed::Box;
1492    use std::task::Waker;
1493
1494    fn block_on<F: Future>(future: F) -> F::Output {
1495        let mut context = Context::from_waker(Waker::noop());
1496        let mut future = Box::pin(future);
1497        loop {
1498            match future.as_mut().poll(&mut context) {
1499                Poll::Ready(output) => return output,
1500                Poll::Pending => std::thread::yield_now(),
1501            }
1502        }
1503    }
1504
1505    fn id_of(profile: &RadioProfile) -> InterfaceId {
1506        InterfaceId::from_channel_tag(InterfaceKind::LoRa, &lora::channel_tag(profile))
1507    }
1508
1509    #[test]
1510    fn awaited_control_returns_only_the_matching_radio_result() {
1511        let control = LoRaControl::new();
1512        control.signal(DEFAULT_915_PROFILE);
1513        block_on(async {
1514            let stale = control.wait().await;
1515            control.complete(stale.id, true);
1516        });
1517
1518        let requested = RadioProfile {
1519            tx_power: TxPower::new(12),
1520            ..DEFAULT_915_PROFILE
1521        };
1522        let (outcome, ()) = block_on(join(control.apply(requested), async {
1523            let request = control.wait().await;
1524            assert_eq!(request.profile, requested);
1525            control.complete(request.id, false);
1526        }));
1527        assert_eq!(outcome, LoRaApplyOutcome::Rejected);
1528    }
1529
1530    #[test]
1531    fn a_channel_change_re_keys_to_the_new_id() {
1532        let current = id_of(&DEFAULT_915_PROFILE);
1533        let mut next = DEFAULT_915_PROFILE;
1534        next.modulation = Modulation::Lora {
1535            spreading_factor: SpreadingFactor::Sf10,
1536            bandwidth: LoraBandwidth::Bw125kHz,
1537            coding_rate: CodingRate::Cr45,
1538        };
1539        let message = retag_message(current, &next, None).expect("a channel change re-keys");
1540        let InterfaceLifecycle::Retag { old_id, new_id, .. } = message else {
1541            panic!("expected a Retag");
1542        };
1543        assert_eq!(old_id, current);
1544        assert_eq!(new_id, id_of(&next));
1545        assert_ne!(new_id, current);
1546    }
1547
1548    #[test]
1549    fn a_local_only_change_does_not_re_key() {
1550        let current = id_of(&DEFAULT_915_PROFILE);
1551        let mut next = DEFAULT_915_PROFILE;
1552        next.tx_power = TxPower::new(2);
1553        next.preamble = PreambleSymbols::new(24);
1554        assert!(
1555            retag_message(current, &next, None).is_none(),
1556            "transmit power and preamble are local knobs, not channel identity"
1557        );
1558    }
1559
1560    #[test]
1561    fn packet_airtime_sums_both_frames_of_a_split() {
1562        let profile = lora::DEFAULT_915_PROFILE;
1563        let one_frame = packet_airtime(&[0u8; 100], &profile);
1564        let two_frames = packet_airtime(&[0u8; 400], &profile);
1565        assert_eq!(
1566            one_frame,
1567            profile.time_on_air_us(101),
1568            "one frame: the header plus 100 payload bytes on air"
1569        );
1570        assert_eq!(
1571            two_frames,
1572            profile.time_on_air_us(255) + profile.time_on_air_us(147),
1573            "two frames: a full 255-byte frame plus the 147-byte remainder"
1574        );
1575        assert!(two_frames > one_frame);
1576    }
1577
1578    #[test]
1579    fn only_successfully_decoded_frames_earn_service_age() {
1580        for event in [
1581            RadioEvent::PreambleDetected,
1582            RadioEvent::HeaderValid,
1583            RadioEvent::HeaderError,
1584            RadioEvent::CrcError,
1585            RadioEvent::Timeout,
1586            RadioEvent::Other,
1587        ] {
1588            assert_eq!(decoded_peer_airtime_us(event, &DEFAULT_915_PROFILE), None);
1589        }
1590        assert_eq!(
1591            decoded_peer_airtime_us(
1592                RadioEvent::Frame(sx126x::ReceivedAirFrame {
1593                    len: 100,
1594                    phy: PacketPhyStats::default(),
1595                }),
1596                &DEFAULT_915_PROFILE,
1597            ),
1598            Some(DEFAULT_915_PROFILE.time_on_air_us(100))
1599        );
1600    }
1601
1602    #[test]
1603    fn an_active_packet_accepts_a_fifo_burst_until_maximum_record_backpressure() {
1604        let mut storage = [0; LORA_TX_QUEUE_BYTES];
1605        let mut backlog = TransmitBacklog::new(&mut storage);
1606        let profile = DEFAULT_915_PROFILE;
1607
1608        assert_eq!(
1609            backlog.accept(&[0; LORA_MAX_PAYLOAD], &profile, 10),
1610            Ok(PacketPlacement::Active)
1611        );
1612        for value in 1..=12 {
1613            assert_eq!(
1614                backlog.accept(&[value; LORA_MAX_PAYLOAD], &profile, 10),
1615                Ok(PacketPlacement::Queued)
1616            );
1617        }
1618        assert!(!backlog.can_accept_outbound());
1619
1620        for value in 0..=12 {
1621            assert_eq!(
1622                backlog.active.bytes[..backlog.active.len.unwrap()],
1623                [value; LORA_MAX_PAYLOAD]
1624            );
1625            backlog.active.clear();
1626            if value < 12 {
1627                assert!(backlog.activate_next(&profile, 20 + u64::from(value)));
1628            }
1629        }
1630        assert!(backlog.queue.is_empty());
1631    }
1632
1633    #[test]
1634    fn disable_drops_only_the_active_packet_and_activation_starts_a_fresh_profile_timeout() {
1635        let mut storage = [0; LORA_TX_QUEUE_BYTES];
1636        let mut backlog = TransmitBacklog::new(&mut storage);
1637        let original_profile = DEFAULT_915_PROFILE;
1638        let queued = [0xA5; 400];
1639        backlog.accept(b"active", &original_profile, 1_000).unwrap();
1640        backlog.accept(&queued, &original_profile, 1_001).unwrap();
1641
1642        assert!(backlog.active.clear());
1643        assert!(!backlog.queue.is_empty());
1644
1645        let mut current_profile = original_profile;
1646        current_profile.modulation = Modulation::Lora {
1647            spreading_factor: SpreadingFactor::Sf12,
1648            bandwidth: LoraBandwidth::Bw125kHz,
1649            coding_rate: CodingRate::Cr48,
1650        };
1651        let activated_at_ms = 200_000;
1652        assert!(backlog.activate_next(&current_profile, activated_at_ms));
1653        assert_eq!(
1654            backlog.active.airtime_us,
1655            packet_airtime(&queued, &current_profile)
1656        );
1657        assert_eq!(backlog.active.activated_at_ms, activated_at_ms);
1658
1659        let access = backlog
1660            .active
1661            .channel_access(
1662                current_profile,
1663                activated_at_ms,
1664                ContentionPriority::Fresh {
1665                    short_airtime_per_mille: 0,
1666                },
1667            )
1668            .unwrap();
1669        let ttl_ms = channel_access::pending_ttl_ms(backlog.active.airtime_us);
1670        assert!(!access.is_expired(activated_at_ms + ttl_ms - 1));
1671        assert!(access.is_expired(activated_at_ms + ttl_ms));
1672        assert!(backlog.active.clear());
1673        assert!(!backlog.activate_next(&current_profile, activated_at_ms + ttl_ms));
1674    }
1675
1676    #[test]
1677    fn sx126x_packet_uses_the_rnode_sync_word() {
1678        let config = sx126x_config(&DEFAULT_915_PROFILE);
1679        assert_eq!(config.sync_word, RNODE_LORA_SYNC_WORD);
1680    }
1681
1682    #[test]
1683    fn spectrum_status_publishes_sensing_and_stewardship_counters() {
1684        let status = LoRaSpectrumStatus::new();
1685        let mut noise = NoiseFloor::new();
1686        for index in 0..32 {
1687            let observation = noise.observe(index, -120, false);
1688            status.record_channel(observation, Some(&noise));
1689        }
1690        status.record_channel(ChannelObservation::Busy, Some(&noise));
1691        status.add_deferrals(3);
1692        status.add_false_preamble();
1693        status.add_contention_timeout();
1694        status.add_duty_hold();
1695        status.add_duty_timeout();
1696        status.add_radio_recovery();
1697
1698        assert_eq!(
1699            status.snapshot(),
1700            LoRaSpectrumSnapshot {
1701                channel_busy_per_mille: 500,
1702                noise_floor_dbm: Some(-120),
1703                cca_threshold_dbm: Some(-109),
1704                deferrals: 3,
1705                false_preambles: 1,
1706                contention_timeouts: 1,
1707                duty_holds: 1,
1708                duty_timeouts: 1,
1709                radio_recoveries: 1,
1710            }
1711        );
1712    }
1713}
prns_interfaces_embassy/lora.rs

prns_interfaces_embassy/
lora.rs
