pamoja_lorawan/relay/node.rs
1//! A relay: an end device that also listens for others and forwards what it hears,
2//! TS011-1.0.1 chapters 3, 7 and 8.
3
4use pamoja_lora::region::{Modulation, RelayChannel};
5use pamoja_lora::LinkSettings;
6
7use super::ack::{wor_ack, CadPeriodicity, Forward, StateSync};
8use super::filter::{FilterAction, JoinFilter};
9use super::forward::{ForwardedUplink, UplinkMetadata, WorChannel, FORWARD_OVERHEAD};
10use super::limits::{ForwardLimits, RELOAD_PERIOD_US};
11use super::state::{config_ok, wor_link, RelayConfig, RelaySettings, RelayState};
12use super::timing::{t_offset_ms, unsynchronized_preamble_symbols};
13use super::trusted::TrustedDevices;
14use super::wor::{Carrier, Wor};
15use super::{
16 LA_FPORT_RELAY, RELAY_FWD_DELAY_US, RXR_DELAY_US, WOR_ACK_DELAY_US, WOR_ACK_LEN,
17 WOR_DATA_DELAY_US,
18};
19use crate::device::{Delivery, DeviceError, EndDevice, Heard, Next, ReceiveWindow, Transmission};
20use crate::mac::{MacCommand, CID_NOTIFY_NEW_END_DEVICE};
21use crate::{LorawanError, MAX_FRAME};
22
23/// The length of a join request, TS001-1.0.4 section 6.2.4.
24const JOIN_REQUEST_LEN: usize = 23;
25
26/// The shortest data frame: its header, address, control, counter and integrity code.
27const MIN_DATA_FRAME: usize = 12;
28
29/// A scan for wake-on-radio frames, due next.
30#[derive(Clone, Copy, Debug, PartialEq, Eq)]
31pub struct Scan {
32 /// When to start detecting, in microseconds.
33 pub start_us: u64,
34 /// Which channel.
35 pub channel: WorChannel,
36 /// Its WOR frequency and data rate.
37 pub carrier: Carrier,
38 /// The LoRa settings of a WOR frame: an explicit header and a payload CRC, heard with
39 /// inverted IQ, as RP002-1.0.5 table 124 has.
40 pub link: LinkSettings,
41 /// The longest preamble an end device sends on the channel, in symbols: a second of
42 /// it on the default channel, which a device that has never heard from a relay assumes
43 /// it scans, and a scan period of it on the second.
44 pub preamble_symbols: u16,
45}
46
47/// What a wake-on-radio frame led a relay to do.
48#[derive(Clone, Copy, Debug, PartialEq, Eq)]
49pub enum Wake {
50 /// A join request follows; listen for it.
51 JoinRequest {
52 /// Where and when.
53 listen: Listen,
54 },
55 /// A trusted end device's uplink follows.
56 Uplink {
57 /// The device.
58 dev_addr: u32,
59 /// The WOR frame counter it used.
60 wfcnt: u32,
61 /// Whether the relay forwards the uplink, which the acknowledgment reports.
62 forward: Forward,
63 /// The acknowledgment to send, or `None` when the relay's duty cycle leaves no room
64 /// for one.
65 acknowledgment: Option<Acknowledgment>,
66 /// Where and when to listen for the uplink, or `None` when a forwarding limit
67 /// keeps the relay from forwarding it.
68 listen: Option<Listen>,
69 },
70 /// A device the relay does not trust woke it, and a `NotifyNewEndDeviceReq` for it
71 /// goes out with the relay's next uplink.
72 Notified {
73 /// The address the frame named.
74 dev_addr: u32,
75 },
76}
77
78/// A WOR ACK to send.
79#[derive(Clone, Copy, Debug, PartialEq, Eq)]
80pub struct Acknowledgment {
81 /// The frame.
82 pub frame: [u8; WOR_ACK_LEN],
83 /// When to start sending it, in microseconds.
84 pub start_us: u64,
85 /// The channel's acknowledgment frequency, at the WOR frame's data rate.
86 pub carrier: Carrier,
87 /// The LoRa settings: an eight-symbol preamble, an explicit header and a payload CRC,
88 /// sent with inverted IQ, RP002-1.0.5 table 125.
89 pub link: LinkSettings,
90 /// The power to ask of the radio, the region's default.
91 pub output_dbm: i8,
92 /// How long it holds the air, in microseconds.
93 pub airtime_us: u64,
94}
95
96/// When and where to listen for the uplink a wake-on-radio frame announced.
97#[derive(Clone, Copy, Debug, PartialEq, Eq)]
98pub struct Listen {
99 /// When the uplink starts, in microseconds.
100 pub start_us: u64,
101 /// Where and how fast.
102 pub carrier: Carrier,
103 /// The LoRa settings of an uplink, heard with standard IQ.
104 pub link: LinkSettings,
105 /// The longest frame the relay forwards, in bytes; stop receiving anything longer, as
106 /// TS011-1.0.1 section 3.4 asks.
107 pub max_len: usize,
108}
109
110/// A downlink for an end device, to send in its RXR window, TS011-1.0.1 chapter 7.
111#[derive(Clone, Copy, Debug, PartialEq, Eq)]
112pub struct RxrDownlink {
113 frame: [u8; MAX_FRAME],
114 len: usize,
115 /// When to start sending it: [`RXR_DELAY_US`] after the end device's uplink ended.
116 pub start_us: u64,
117 /// The frequency of the WOR frame, at the uplink's first window data rate with no
118 /// offset, RP002-1.0.5 table 126.
119 pub carrier: Carrier,
120 /// The LoRa settings: an eight-symbol preamble, an explicit header and a payload CRC,
121 /// sent with inverted IQ, RP002-1.0.5 table 127.
122 pub link: LinkSettings,
123 /// The power to ask of the radio, the region's default.
124 pub output_dbm: i8,
125 /// How long it holds the air, in microseconds.
126 pub airtime_us: u64,
127}
128
129impl RxrDownlink {
130 /// Returns the frame to send.
131 ///
132 /// # Returns
133 ///
134 /// The end device's PHYPayload, as the network sent it.
135 #[must_use]
136 pub fn frame(&self) -> &[u8] {
137 &self.frame[..self.len]
138 }
139}
140
141/// What a frame a relay's own device heard turned out to be.
142#[allow(clippy::large_enum_variant)]
143#[derive(Clone, Copy, Debug, PartialEq, Eq)]
144pub enum RelayHeard {
145 /// A join accept, or a downlink for the relay itself.
146 Device(Heard),
147 /// A downlink the network sent for an end device, to send on in its RXR window.
148 Downlink {
149 /// The downlink as the relay's device read it.
150 delivery: Delivery,
151 /// What to send the end device.
152 downlink: RxrDownlink,
153 },
154 /// A downlink for an end device the relay cannot send on.
155 Undeliverable {
156 /// The downlink as the relay's device read it.
157 delivery: Delivery,
158 /// Why.
159 reason: RelayError,
160 },
161}
162
163/// Why a relay did not do what a frame or a call asked.
164#[derive(Clone, Copy, Debug, PartialEq, Eq)]
165pub enum RelayError {
166 /// The relay is not scanning.
167 Stopped,
168 /// A forwarded uplink is still waiting to go out.
169 Busy,
170 /// No wake-on-radio frame announced an uplink to listen for.
171 NotListening,
172 /// No forwarded uplink is waiting to go out.
173 NothingHeld,
174 /// No forwarded uplink is waiting on the network's answer.
175 NothingAwaited,
176 /// The frame did not decode or verify.
177 Frame(LorawanError),
178 /// A frame names a carrier the relay cannot hear or send on.
179 Carrier(Carrier),
180 /// A forwarding limit is reached.
181 Limited,
182 /// The join request filter drops the request.
183 Filtered,
184 /// The uplink is from another device than the WOR frame named.
185 Foreign,
186 /// The relay configuration uses a channel the relay cannot scan, or scans too often
187 /// for its data rates.
188 Configuration,
189 /// The relay's own device refused.
190 Device(DeviceError),
191}
192
193impl core::fmt::Display for RelayError {
194 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
195 match self {
196 RelayError::Stopped => f.write_str("the relay is not scanning"),
197 RelayError::Busy => f.write_str("a forwarded uplink is still waiting to go out"),
198 RelayError::NotListening => f.write_str("no WOR frame announced this uplink"),
199 RelayError::NothingHeld => f.write_str("no forwarded uplink is waiting to go out"),
200 RelayError::NothingAwaited => f.write_str("no forwarded uplink awaits an answer"),
201 RelayError::Frame(error) => write!(f, "the frame did not decode: {error}"),
202 RelayError::Carrier(carrier) => write!(
203 f,
204 "the relay cannot use {} Hz at DR{}",
205 carrier.frequency_hz, carrier.data_rate
206 ),
207 RelayError::Limited => f.write_str("a forwarding limit is reached"),
208 RelayError::Filtered => f.write_str("the join request filter drops it"),
209 RelayError::Foreign => f.write_str("the uplink is not from the device that woke it"),
210 RelayError::Configuration => f.write_str("the relay cannot run that configuration"),
211 RelayError::Device(error) => write!(f, "the relay's device refused: {error}"),
212 }
213 }
214}
215
216impl core::error::Error for RelayError {}
217
218impl From<DeviceError> for RelayError {
219 fn from(error: DeviceError) -> RelayError {
220 RelayError::Device(error)
221 }
222}
223
224/// Which uplink a wake-on-radio frame announced.
225#[derive(Clone, Copy, Debug, PartialEq, Eq)]
226enum Announced {
227 JoinRequest,
228 Uplink { index: u8, dev_addr: u32 },
229}
230
231/// A wake-on-radio exchange waiting on its uplink.
232#[derive(Clone, Copy, Debug, PartialEq, Eq)]
233struct Exchange {
234 announced: Announced,
235 channel: WorChannel,
236 wor_frequency_hz: u32,
237 uplink: Carrier,
238 max_len: usize,
239}
240
241/// Where a forwarded uplink's answer goes: the end device's RXR window.
242#[derive(Clone, Copy, Debug, PartialEq, Eq)]
243struct Route {
244 uplink_end_us: u64,
245 wor_frequency_hz: u32,
246 uplink_data_rate: u8,
247}
248
249/// A forwarded uplink waiting to go out.
250#[derive(Clone, Copy, Debug, PartialEq, Eq)]
251struct Held {
252 payload: [u8; FORWARD_OVERHEAD + MAX_FRAME],
253 len: usize,
254 due_us: u64,
255 route: Route,
256}
257
258/// A LoRaWAN relay: an end device that scans for wake-on-radio frames, acknowledges and
259/// forwards the uplinks behind them, and sends the network's answers on in each end
260/// device's RXR window.
261///
262/// Like [`EndDevice`], it owns no radio and no clock. One forwarded uplink runs like this:
263///
264/// 1. [`next_scan`](Relay::next_scan) says when and where to detect a preamble.
265/// 2. A WOR frame received there goes to [`heard_wor`](Relay::heard_wor), which says
266/// whether to acknowledge it and when to listen for the uplink.
267/// 3. That uplink goes to [`heard_uplink`](Relay::heard_uplink), and
268/// [`forward`](Relay::forward) sends it to the network in the relay's own uplink on port
269/// [`LA_FPORT_RELAY`].
270/// 4. What the relay's receive windows hear goes to [`heard_in`](Relay::heard_in), which
271/// reads the relay commands among its MAC commands and turns a downlink for the end
272/// device into an [`RxrDownlink`].
273///
274/// # Examples
275///
276/// A relay forwards a sensor's reading and sends the network's answer back:
277///
278/// ```
279/// use pamoja_lora::region::Region;
280/// use pamoja_lorawan::device::{EndDevice, ReceiveWindow, Settings};
281/// use pamoja_lorawan::mac::MacCommand;
282/// use pamoja_lorawan::relay::{
283/// root_wor_s_key, wor_uplink, CadToRx, Carrier, ForwardedUplink, Relay, RelayConfig,
284/// RelayHeard, RelaySettings, Wake, WorKeys, XtalAccuracy, LA_FPORT_RELAY,
285/// };
286/// use pamoja_lorawan::{Downlink, Session, Uplink};
287///
288/// let plan = Region::Eu868.plan();
289/// let relay_session = Session::new(0x2601_0001, [0x11; 16], [0x22; 16]);
290/// let device = EndDevice::personalized(plan, relay_session, Settings::new(2, 14))?;
291/// let settings = RelaySettings::new(XtalAccuracy::Ppm10, CadToRx::Symbols2);
292/// let mut relay = Relay::new(device, settings);
293/// relay.start(RelayConfig::region_default(plan).expect("EU868 has relay channels"))?;
294///
295/// // The network trusts a sensor to it, as an UpdateUplinkListReq would.
296/// let sensor = Session::new(0x2601_1BDA, [0x2B; 16], [0x99; 16]);
297/// let root = root_wor_s_key(&[0x2B; 16]);
298/// relay.trust(0, sensor.dev_addr(), &root, 0, 63, 0);
299///
300/// // The first scan starts now, on 865.1 MHz.
301/// let scan = relay.next_scan(0).expect("running");
302/// assert_eq!(scan.carrier, Carrier::new(865_100_000, 3));
303///
304/// // The sensor's WOR frame says its uplink follows on 868.1 MHz at DR5.
305/// let uplink = Carrier::new(868_100_000, 5);
306/// let wor = wor_uplink(&WorKeys::derive(&root, sensor.dev_addr()), sensor.dev_addr(), 0, uplink, scan.carrier)?;
307/// let Wake::Uplink { acknowledgment, listen, .. } = relay.heard_wor(&scan, &wor, -90, 4, 900_000)?
308/// else {
309/// panic!("a trusted device's uplink follows");
310/// };
311/// assert_eq!(acknowledgment.expect("room to acknowledge").carrier, Carrier::new(865_300_000, 3));
312/// let listen = listen.expect("forwarding is available");
313///
314/// // The uplink arrives, and goes out in the relay's own uplink on port 226.
315/// let frame = sensor.encode_uplink(&Uplink::new(0, 1, b"21.5"))?;
316/// let due = relay.heard_uplink(frame.as_bytes(), -88, 6, listen.start_us + 60_000)?;
317/// let transmission = relay.forward(due)?;
318///
319/// // The network reads it back, and answers the sensor through the relay.
320/// let sent = relay_session.decode(transmission.frame.as_bytes(), 0)?;
321/// assert_eq!(sent.fport(), Some(LA_FPORT_RELAY));
322/// assert_eq!(ForwardedUplink::parse(sent.payload())?.phy_payload, frame.as_bytes());
323/// let answer = sensor.encode_downlink(&Downlink::new(0, 1, b"ok"))?;
324/// let reply = relay_session.encode_downlink(&Downlink::new(0, LA_FPORT_RELAY, answer.as_bytes()))?;
325///
326/// let RelayHeard::Downlink { downlink, .. } = relay.heard_in(ReceiveWindow::Rx1, reply.as_bytes(), 7)?
327/// else {
328/// panic!("a downlink for the sensor");
329/// };
330/// assert_eq!(downlink.frame(), answer.as_bytes());
331/// assert_eq!(downlink.carrier.frequency_hz, 865_100_000, "on the WOR frequency");
332/// assert_eq!(downlink.start_us, listen.start_us + 60_000 + 18_000_000);
333/// # Ok::<(), Box<dyn std::error::Error>>(())
334/// ```
335pub struct Relay<'p> {
336 device: EndDevice<'p>,
337 state: RelayState,
338 exchange: Option<Exchange>,
339 held: Option<Held>,
340 awaiting: Option<Route>,
341}
342
343impl<'p> Relay<'p> {
344 /// Makes a relay of an end device.
345 ///
346 /// It does not scan until [`start`](Relay::start) or its network's `RelayConfReq` gives
347 /// it channels, and it forwards every join request and trusts no one until its network
348 /// says otherwise.
349 ///
350 /// # Arguments
351 ///
352 /// * `device` - the relay's own device, joined or to join.
353 /// * `settings` - what its radio can do.
354 ///
355 /// # Returns
356 ///
357 /// The relay.
358 #[must_use]
359 pub const fn new(device: EndDevice<'p>, settings: RelaySettings) -> Relay<'p> {
360 Relay {
361 device,
362 state: RelayState::new(settings),
363 exchange: None,
364 held: None,
365 awaiting: None,
366 }
367 }
368
369 /// Returns the relay's own device.
370 ///
371 /// # Returns
372 ///
373 /// The device.
374 #[must_use]
375 pub const fn device(&self) -> &EndDevice<'p> {
376 &self.device
377 }
378
379 /// Returns the relay's own device, to join, send its own uplinks, and hear what comes
380 /// back for them.
381 ///
382 /// Read downlinks through [`heard_in`](Relay::heard_in) instead, so the relay commands
383 /// among their MAC commands reach the relay.
384 ///
385 /// # Returns
386 ///
387 /// The device.
388 pub fn device_mut(&mut self) -> &mut EndDevice<'p> {
389 &mut self.device
390 }
391
392 /// Gives the relay's device back.
393 ///
394 /// # Returns
395 ///
396 /// The device.
397 #[must_use]
398 pub fn into_device(self) -> EndDevice<'p> {
399 self.device
400 }
401
402 /// Returns what the relay's radio can do.
403 ///
404 /// # Returns
405 ///
406 /// The settings.
407 #[must_use]
408 pub const fn settings(&self) -> RelaySettings {
409 self.state.settings
410 }
411
412 /// Returns the channels the relay scans.
413 ///
414 /// # Returns
415 ///
416 /// The configuration, or `None` while the relay is stopped.
417 #[must_use]
418 pub const fn config(&self) -> Option<RelayConfig> {
419 self.state.config
420 }
421
422 /// Returns one of the region's wake-on-radio channels, TS011-1.0.1 section 3.2.2.
423 ///
424 /// # Arguments
425 ///
426 /// * `index` - which, 0 or 1.
427 ///
428 /// # Returns
429 ///
430 /// The channel, or `None` where the region defines none.
431 #[must_use]
432 pub fn region_channel(&self, index: u8) -> Option<RelayChannel> {
433 self.device.plan().relay_channel(index)
434 }
435
436 /// Starts scanning, or changes what a running relay scans from its next scan on.
437 ///
438 /// # Arguments
439 ///
440 /// * `config` - the channels and how often.
441 ///
442 /// # Errors
443 ///
444 /// Returns [`RelayError::Configuration`] for a channel that is not LoRa or not in the
445 /// radio's range, or scans so frequent that detecting a preamble and starting to receive
446 /// on one channel does not fit before the next.
447 pub fn start(&mut self, config: RelayConfig) -> Result<(), RelayError> {
448 if !config_ok(&self.device, &config, self.state.settings.cad_to_rx) {
449 return Err(RelayError::Configuration);
450 }
451 self.state.run(config);
452 Ok(())
453 }
454
455 /// Stops scanning. A forwarded uplink already waiting still goes out.
456 pub fn stop(&mut self) {
457 self.state.stop();
458 self.exchange = None;
459 }
460
461 /// Returns the join request filter.
462 ///
463 /// # Returns
464 ///
465 /// The filter.
466 #[must_use]
467 pub const fn join_filter(&self) -> &JoinFilter {
468 &self.state.filter
469 }
470
471 /// Returns the join request filter to change.
472 ///
473 /// # Returns
474 ///
475 /// The filter.
476 pub fn join_filter_mut(&mut self) -> &mut JoinFilter {
477 &mut self.state.filter
478 }
479
480 /// Returns the end devices the relay trusts.
481 ///
482 /// # Returns
483 ///
484 /// The list.
485 #[must_use]
486 pub const fn trusted(&self) -> &TrustedDevices {
487 &self.state.trusted
488 }
489
490 /// Returns the end devices the relay trusts, to change.
491 ///
492 /// # Returns
493 ///
494 /// The list.
495 pub fn trusted_mut(&mut self) -> &mut TrustedDevices {
496 &mut self.state.trusted
497 }
498
499 /// Trusts an end device, as an `UpdateUplinkListReq` with the same fields does.
500 ///
501 /// # Arguments
502 ///
503 /// * `index` - the entry, 0 to 15.
504 /// * `dev_addr` - the device's address.
505 /// * `root_wor_s_key` - its root relay session key.
506 /// * `next_wfcnt` - the WOR frame counter to expect from it next.
507 /// * `reload_rate` - its uplinks forwarded an hour, 63 for no limit.
508 /// * `bucket_size` - the coded bucket size multiplier, TS011-1.0.1 table 55.
509 ///
510 /// # Returns
511 ///
512 /// `false` for an index past 15.
513 pub fn trust(
514 &mut self,
515 index: u8,
516 dev_addr: u32,
517 root_wor_s_key: &[u8; 16],
518 next_wfcnt: u32,
519 reload_rate: u8,
520 bucket_size: u8,
521 ) -> bool {
522 index < 16
523 && self
524 .state
525 .command(
526 &MacCommand::UpdateUplinkListReq {
527 index,
528 reload_rate,
529 bucket_size,
530 dev_addr,
531 wfcnt: next_wfcnt,
532 root_wor_s_key: *root_wor_s_key,
533 },
534 &self.device,
535 )
536 .is_some()
537 }
538
539 /// Returns the relay-wide forwarding limits.
540 ///
541 /// # Returns
542 ///
543 /// The limits.
544 #[must_use]
545 pub const fn limits(&self) -> &ForwardLimits {
546 &self.state.limits
547 }
548
549 /// Returns the relay-wide forwarding limits, to change.
550 ///
551 /// # Returns
552 ///
553 /// The limits.
554 pub fn limits_mut(&mut self) -> &mut ForwardLimits {
555 &mut self.state.limits
556 }
557
558 /// Says when and where to scan next.
559 ///
560 /// The first call after the relay starts begins its scan at `now_us`, with every
561 /// forwarding limit at one hour's reload, as TS011-1.0.1 section 8.8 has. Scans fall on
562 /// whole periods from then, and a second channel's half a period between, so a relay
563 /// that was busy picks the schedule up at the next of them, section 3.2.1.
564 ///
565 /// # Arguments
566 ///
567 /// * `now_us` - the time, in microseconds.
568 ///
569 /// # Returns
570 ///
571 /// The scan, or `None` while the relay is stopped.
572 pub fn next_scan(&mut self, now_us: u64) -> Option<Scan> {
573 let config = self.state.config?;
574 let anchor = match self.state.anchor_us {
575 Some(anchor) => anchor,
576 None => {
577 self.state.anchor_us = Some(now_us);
578 self.state.reloads = 0;
579 self.state.limits.restart();
580 now_us
581 }
582 };
583 let period = config.cad_periodicity.period_us();
584 let default_at = next_multiple(anchor, period, now_us);
585 let (start_us, channel) = match config.second_channel {
586 Some(_) => {
587 let second_at = next_multiple(anchor + period / 2, period, now_us);
588 if second_at < default_at {
589 (second_at, WorChannel::Second)
590 } else {
591 (default_at, WorChannel::Default)
592 }
593 }
594 None => (default_at, WorChannel::Default),
595 };
596 let relay_channel = config.channel(channel)?;
597 let link = wor_link(self.device.plan(), relay_channel.data_rate)?;
598 let longest = match channel {
599 WorChannel::Default => CadPeriodicity::Ms1000,
600 WorChannel::Second => config.cad_periodicity,
601 };
602 Some(Scan {
603 start_us,
604 channel,
605 carrier: Carrier::new(relay_channel.wor_frequency_hz, relay_channel.data_rate),
606 link,
607 preamble_symbols: unsynchronized_preamble_symbols(
608 longest,
609 link.symbol_time_us(),
610 self.state.settings.cad_to_rx,
611 ),
612 })
613 }
614
615 /// Reads a wake-on-radio frame a scan heard, TS011-1.0.1 sections 3.2.4 and 3.3.
616 ///
617 /// A join request's frame has the relay listen for the request, if the join request
618 /// limit allows. A trusted device's frame is verified against the counter the relay
619 /// expects from it, and answered with a WOR ACK saying whether the relay forwards the
620 /// uplink, which it listens for only if it does. A frame from a device the relay does
621 /// not trust queues a notification of it for the network, once while one is waiting,
622 /// if the notification limit allows.
623 ///
624 /// # Arguments
625 ///
626 /// * `scan` - the scan that heard it.
627 /// * `frame` - the frame.
628 /// * `rssi_dbm` - its signal strength.
629 /// * `snr_db` - its signal-to-noise ratio.
630 /// * `ended_us` - when it finished arriving, in microseconds.
631 ///
632 /// # Returns
633 ///
634 /// What to do next.
635 ///
636 /// # Errors
637 ///
638 /// Returns [`RelayError::Stopped`] while stopped, [`RelayError::Busy`] while a forwarded
639 /// uplink waits to go out, [`RelayError::Frame`] for a frame that does not decode or, from
640 /// a trusted device's address, does not verify, [`RelayError::Carrier`] for an uplink
641 /// on a carrier the relay cannot hear, and [`RelayError::Limited`] when a limit keeps it
642 /// from listening or notifying.
643 pub fn heard_wor(
644 &mut self,
645 scan: &Scan,
646 frame: &[u8],
647 rssi_dbm: i16,
648 snr_db: i8,
649 ended_us: u64,
650 ) -> Result<Wake, RelayError> {
651 let config = self.state.config.ok_or(RelayError::Stopped)?;
652 if self.held.is_some() {
653 return Err(RelayError::Busy);
654 }
655 self.exchange = None;
656 self.state.reload(ended_us);
657 let channel = config
658 .channel(scan.channel)
659 .ok_or(RelayError::Carrier(scan.carrier))?;
660
661 match Wor::parse(frame).map_err(RelayError::Frame)? {
662 Wor::JoinRequest { uplink } => {
663 let link = self.uplink_link(uplink)?;
664 let limits = &self.state.limits;
665 if !(limits.join_request.has_token() && limits.overall.has_token()) {
666 return Err(RelayError::Limited);
667 }
668 if self.forwardable_len(uplink)? < JOIN_REQUEST_LEN {
669 return Err(RelayError::Frame(LorawanError::PayloadTooLong));
670 }
671 let listen = Listen {
672 start_us: ended_us + u64::from(WOR_DATA_DELAY_US),
673 carrier: uplink,
674 link,
675 max_len: JOIN_REQUEST_LEN,
676 };
677 self.exchange = Some(Exchange {
678 announced: Announced::JoinRequest,
679 channel: scan.channel,
680 wor_frequency_hz: scan.carrier.frequency_hz,
681 uplink,
682 max_len: JOIN_REQUEST_LEN,
683 });
684 Ok(Wake::JoinRequest { listen })
685 }
686 Wor::Uplink(sealed) => {
687 let dev_addr = sealed.dev_addr();
688 let (index, wfcnt, uplink) = match self.state.trusted.open(&sealed, scan.carrier) {
689 None => return self.notify(dev_addr, rssi_dbm, snr_db),
690 Some(opened) => opened.map_err(RelayError::Frame)?,
691 };
692 let link = self.uplink_link(uplink)?;
693 let max_len = self.forwardable_len(uplink)?;
694 let mut forward = self.forward_status(index, ended_us);
695 if forward == Forward::Available && max_len < MIN_DATA_FRAME {
696 forward = Forward::Disabled;
697 }
698
699 let ack_carrier = Carrier::new(channel.ack_frequency_hz, channel.data_rate);
700 let ack_link = scan.link.with_preamble(8);
701 let ack_start = ended_us + u64::from(WOR_ACK_DELAY_US);
702 let ack_airtime = ack_link.airtime_us(WOR_ACK_LEN);
703 let preamble_end =
704 ended_us.saturating_sub(scan.link.with_preamble(0).airtime_us(frame.len()));
705 let t_offset = t_offset_ms(scan.start_us, preamble_end).unwrap_or(
706 if preamble_end < scan.start_us {
707 0
708 } else {
709 StateSync::MAX_T_OFFSET_MS
710 },
711 );
712 let state = StateSync {
713 cad_to_rx: self.state.settings.cad_to_rx,
714 forward,
715 relay_data_rate: self.device.data_rate(),
716 xtal_accuracy: self.state.settings.xtal_accuracy,
717 cad_periodicity: config.cad_periodicity,
718 t_offset_ms: t_offset,
719 };
720 let keys = *self
721 .state
722 .trusted
723 .get(index)
724 .ok_or(RelayError::NotListening)?
725 .keys();
726 let acknowledgment = if self.device.free_at(ack_carrier.frequency_hz) <= ack_start {
727 let frame = wor_ack(&keys, dev_addr, wfcnt, ack_carrier, uplink, state)
728 .map_err(RelayError::Frame)?;
729 self.device
730 .record_air(ack_start, ack_airtime, ack_carrier.frequency_hz);
731 Some(Acknowledgment {
732 frame,
733 start_us: ack_start,
734 carrier: ack_carrier,
735 link: ack_link,
736 output_dbm: self.device.output_dbm(ack_carrier.frequency_hz, 0),
737 airtime_us: ack_airtime,
738 })
739 } else {
740 None
741 };
742
743 let listen = (forward == Forward::Available).then(|| Listen {
744 start_us: ack_start + ack_airtime + u64::from(WOR_DATA_DELAY_US),
745 carrier: uplink,
746 link,
747 max_len,
748 });
749 if listen.is_some() {
750 self.exchange = Some(Exchange {
751 announced: Announced::Uplink { index, dev_addr },
752 channel: scan.channel,
753 wor_frequency_hz: scan.carrier.frequency_hz,
754 uplink,
755 max_len,
756 });
757 }
758 Ok(Wake::Uplink {
759 dev_addr,
760 wfcnt,
761 forward,
762 acknowledgment,
763 listen,
764 })
765 }
766 }
767 }
768
769 /// Reads the uplink a wake-on-radio frame announced, and holds it to forward,
770 /// TS011-1.0.1 sections 3.5, 8.1 and 8.6.
771 ///
772 /// A join request must be one, and pass the join request filter. A data uplink must
773 /// come from the device whose WOR frame announced it. Either spends a token from every
774 /// limit it counts against.
775 ///
776 /// # Arguments
777 ///
778 /// * `frame` - the uplink.
779 /// * `rssi_dbm` - its signal strength, which the forwarded metadata carries.
780 /// * `snr_db` - its signal-to-noise ratio, likewise.
781 /// * `ended_us` - when it finished arriving, in microseconds.
782 ///
783 /// # Returns
784 ///
785 /// When to [`forward`](Relay::forward) it: [`RELAY_FWD_DELAY_US`] after it ended.
786 ///
787 /// # Errors
788 ///
789 /// Returns [`RelayError::NotListening`] with no uplink announced, [`RelayError::Frame`]
790 /// for a frame longer than the relay forwards or not of the kind announced,
791 /// [`RelayError::Foreign`] for another device's uplink, [`RelayError::Filtered`] for a
792 /// join request the filter drops, and [`RelayError::Limited`] when a limit is reached.
793 pub fn heard_uplink(
794 &mut self,
795 frame: &[u8],
796 rssi_dbm: i16,
797 snr_db: i8,
798 ended_us: u64,
799 ) -> Result<u64, RelayError> {
800 let exchange = self.exchange.take().ok_or(RelayError::NotListening)?;
801 if frame.len() > exchange.max_len {
802 return Err(RelayError::Frame(LorawanError::PayloadTooLong));
803 }
804 self.state.reload(ended_us);
805 let limits = &mut self.state.limits;
806 match exchange.announced {
807 Announced::JoinRequest => {
808 if frame.len() != JOIN_REQUEST_LEN || frame[0] & 0xE3 != 0x00 {
809 return Err(RelayError::Frame(LorawanError::MalformedFrame));
810 }
811 let mut join_eui = [0u8; 8];
812 let mut dev_eui = [0u8; 8];
813 for (at, byte) in frame[1..9].iter().rev().enumerate() {
814 join_eui[at] = *byte;
815 }
816 for (at, byte) in frame[9..17].iter().rev().enumerate() {
817 dev_eui[at] = *byte;
818 }
819 if self.state.filter.decide(&join_eui, &dev_eui) == FilterAction::Filter {
820 return Err(RelayError::Filtered);
821 }
822 if !(limits.join_request.has_token() && limits.overall.has_token()) {
823 return Err(RelayError::Limited);
824 }
825 limits.join_request.take();
826 limits.overall.take();
827 }
828 Announced::Uplink { index, dev_addr } => {
829 if frame.len() < MIN_DATA_FRAME {
830 return Err(RelayError::Frame(LorawanError::FrameTooShort));
831 }
832 if !matches!(frame[0] & 0xE3, 0x40 | 0x80) {
833 return Err(RelayError::Frame(LorawanError::MalformedFrame));
834 }
835 if u32::from_le_bytes([frame[1], frame[2], frame[3], frame[4]]) != dev_addr {
836 return Err(RelayError::Foreign);
837 }
838 let device = self
839 .state
840 .trusted
841 .get_mut(index)
842 .filter(|device| device.dev_addr() == dev_addr)
843 .ok_or(RelayError::NotListening)?;
844 if !(device.bucket().has_token()
845 && limits.global_uplink.has_token()
846 && limits.overall.has_token())
847 {
848 return Err(RelayError::Limited);
849 }
850 device.bucket_mut().take();
851 limits.global_uplink.take();
852 limits.overall.take();
853 }
854 }
855
856 let forwarded = ForwardedUplink {
857 metadata: UplinkMetadata {
858 wor_channel: exchange.channel,
859 rssi_dbm,
860 snr_db,
861 data_rate: exchange.uplink.data_rate,
862 },
863 frequency_hz: exchange.uplink.frequency_hz,
864 phy_payload: frame,
865 };
866 let mut payload = [0u8; FORWARD_OVERHEAD + MAX_FRAME];
867 let len = forwarded.encode(&mut payload).map_err(RelayError::Frame)?;
868 let due_us = ended_us + u64::from(RELAY_FWD_DELAY_US);
869 self.held = Some(Held {
870 payload,
871 len,
872 due_us,
873 route: Route {
874 uplink_end_us: ended_us,
875 wor_frequency_hz: exchange.wor_frequency_hz,
876 uplink_data_rate: exchange.uplink.data_rate,
877 },
878 });
879 Ok(due_us)
880 }
881
882 /// Clears the uplink a wake-on-radio frame announced, once listening for it heard
883 /// nothing.
884 pub fn uplink_missed(&mut self) {
885 self.exchange = None;
886 }
887
888 /// Returns when the forwarded uplink waiting to go out is due.
889 ///
890 /// # Returns
891 ///
892 /// The time, or `None` with nothing waiting.
893 #[must_use]
894 pub fn forward_due(&self) -> Option<u64> {
895 self.held.map(|held| held.due_us)
896 }
897
898 /// Sends the uplink [`heard_uplink`](Relay::heard_uplink) holds, in the relay's own
899 /// uplink on port [`LA_FPORT_RELAY`], TS011-1.0.1 section 9.1.
900 ///
901 /// It stays held when the relay's device cannot send yet, and when the answers the
902 /// device owes crowd it out of this frame, which
903 /// [`Transmission::carries_payload`] reports; call again once the device is free.
904 ///
905 /// # Arguments
906 ///
907 /// * `now_us` - the time, in microseconds.
908 ///
909 /// # Returns
910 ///
911 /// What the relay's device transmits, with its receive windows.
912 ///
913 /// # Errors
914 ///
915 /// Returns [`RelayError::NothingHeld`] with nothing held, and [`RelayError::Device`] with
916 /// [`DeviceError::Wait`] before it is due or whatever else keeps the device from
917 /// sending.
918 pub fn forward(&mut self, now_us: u64) -> Result<Transmission, RelayError> {
919 let held = self.held.ok_or(RelayError::NothingHeld)?;
920 if now_us < held.due_us {
921 return Err(RelayError::Device(DeviceError::Wait {
922 until_us: held.due_us,
923 }));
924 }
925 let transmission =
926 self.device
927 .send(LA_FPORT_RELAY, &held.payload[..held.len], false, now_us)?;
928 if transmission.carries_payload {
929 self.held = None;
930 self.awaiting = Some(held.route);
931 }
932 Ok(transmission)
933 }
934
935 /// Reads a frame heard in a given receive window of the relay's last transmission.
936 ///
937 /// Relay commands among its MAC commands reconfigure the relay, answered in order with
938 /// the rest. A downlink on port [`LA_FPORT_RELAY`] answering a forwarded uplink becomes
939 /// an [`RxrDownlink`] for the end device, TS011-1.0.1 section 9.2.
940 ///
941 /// # Arguments
942 ///
943 /// * `window` - the window the radio heard the frame in.
944 /// * `frame` - the bytes the radio received.
945 /// * `snr_db` - the frame's signal-to-noise ratio.
946 ///
947 /// # Returns
948 ///
949 /// What the frame was.
950 ///
951 /// # Errors
952 ///
953 /// Returns [`RelayError::Device`] for whatever the relay's device refuses, as
954 /// [`EndDevice::heard_in`] does.
955 pub fn heard_in(
956 &mut self,
957 window: ReceiveWindow,
958 frame: &[u8],
959 snr_db: i8,
960 ) -> Result<RelayHeard, RelayError> {
961 self.heard_frame(Some(window), frame, snr_db)
962 }
963
964 /// Reads a frame heard in one of the receive windows of the relay's last transmission,
965 /// without saying which, as [`heard_in`](Relay::heard_in) otherwise does.
966 ///
967 /// # Arguments
968 ///
969 /// * `frame` - the bytes the radio received.
970 /// * `snr_db` - the frame's signal-to-noise ratio.
971 ///
972 /// # Returns
973 ///
974 /// What the frame was.
975 ///
976 /// # Errors
977 ///
978 /// As [`heard_in`](Relay::heard_in).
979 pub fn heard(&mut self, frame: &[u8], snr_db: i8) -> Result<RelayHeard, RelayError> {
980 self.heard_frame(None, frame, snr_db)
981 }
982
983 /// Says what comes next once the relay's receive windows closed with nothing for it, as
984 /// [`EndDevice::nothing_heard`] does. A forwarded uplink not repeated no longer waits on
985 /// an answer.
986 ///
987 /// # Arguments
988 ///
989 /// * `now_us` - the time the second window closed, in microseconds.
990 ///
991 /// # Returns
992 ///
993 /// Whether to repeat the frame, join again, or move on.
994 ///
995 /// # Errors
996 ///
997 /// Returns [`RelayError::Device`] with [`DeviceError::NothingPending`] with no
998 /// transmission waiting on its windows.
999 pub fn nothing_heard(&mut self, now_us: u64) -> Result<Next, RelayError> {
1000 let next = self.device.nothing_heard(now_us)?;
1001 if !matches!(next, Next::Repeat { .. }) {
1002 self.awaiting = None;
1003 }
1004 Ok(next)
1005 }
1006
1007 fn heard_frame(
1008 &mut self,
1009 window: Option<ReceiveWindow>,
1010 frame: &[u8],
1011 snr_db: i8,
1012 ) -> Result<RelayHeard, RelayError> {
1013 let heard = self
1014 .device
1015 .heard_frame(window, frame, snr_db, Some(&mut self.state))?;
1016 if self.state.config.is_none() {
1017 self.exchange = None;
1018 }
1019 let Heard::Data(delivery) = heard else {
1020 return Ok(RelayHeard::Device(heard));
1021 };
1022 if delivery.port() != Some(LA_FPORT_RELAY) {
1023 self.awaiting = None;
1024 return Ok(RelayHeard::Device(heard));
1025 }
1026 let Some(route) = self.awaiting.take() else {
1027 return Ok(RelayHeard::Undeliverable {
1028 delivery,
1029 reason: RelayError::NothingAwaited,
1030 });
1031 };
1032 Ok(match self.rxr(route, delivery.payload()) {
1033 Ok(downlink) => RelayHeard::Downlink { delivery, downlink },
1034 Err(reason) => RelayHeard::Undeliverable { delivery, reason },
1035 })
1036 }
1037
1038 /// Builds the RXR downlink for a forwarded uplink's answer.
1039 fn rxr(&mut self, route: Route, payload: &[u8]) -> Result<RxrDownlink, RelayError> {
1040 let plan = self.device.plan();
1041 let data_rate =
1042 plan.rx1_data_rate(route.uplink_data_rate, 0)
1043 .ok_or(RelayError::Carrier(Carrier::new(
1044 route.wor_frequency_hz,
1045 route.uplink_data_rate,
1046 )))?;
1047 let carrier = Carrier::new(route.wor_frequency_hz, data_rate);
1048 let link = match plan
1049 .downlink_data_rate(data_rate)
1050 .map(|rate| rate.modulation)
1051 {
1052 Some(Modulation::LoRa {
1053 spreading_factor,
1054 bandwidth_hz,
1055 }) => LinkSettings::new(spreading_factor, bandwidth_hz),
1056 _ => return Err(RelayError::Carrier(carrier)),
1057 };
1058 if payload.len() < MIN_DATA_FRAME {
1059 return Err(RelayError::Frame(LorawanError::FrameTooShort));
1060 }
1061 let limit = plan
1062 .downlink_max_payload(data_rate, true)
1063 .map_or(MAX_FRAME, |limit| usize::from(limit.mac_payload) + 5);
1064 if payload.len() > limit.min(MAX_FRAME) {
1065 return Err(RelayError::Frame(LorawanError::PayloadTooLong));
1066 }
1067 let start_us = route.uplink_end_us + u64::from(RXR_DELAY_US);
1068 let free_at = self.device.free_at(carrier.frequency_hz);
1069 if free_at > start_us {
1070 return Err(RelayError::Device(DeviceError::Wait { until_us: free_at }));
1071 }
1072 let airtime_us = link.airtime_us(payload.len());
1073 self.device
1074 .record_air(start_us, airtime_us, carrier.frequency_hz);
1075 let mut frame = [0u8; MAX_FRAME];
1076 frame[..payload.len()].copy_from_slice(payload);
1077 Ok(RxrDownlink {
1078 frame,
1079 len: payload.len(),
1080 start_us,
1081 carrier,
1082 link,
1083 output_dbm: self.device.output_dbm(carrier.frequency_hz, 0),
1084 airtime_us,
1085 })
1086 }
1087
1088 /// The LoRa settings of the uplink a WOR frame announced, if the relay can hear it.
1089 fn uplink_link(&self, carrier: Carrier) -> Result<LinkSettings, RelayError> {
1090 match self
1091 .device
1092 .plan()
1093 .uplink_data_rate(carrier.data_rate)
1094 .map(|rate| rate.modulation)
1095 {
1096 Some(Modulation::LoRa {
1097 spreading_factor,
1098 bandwidth_hz,
1099 }) if self.device.tunes(carrier.frequency_hz) => {
1100 Ok(LinkSettings::new(spreading_factor, bandwidth_hz))
1101 }
1102 _ => Err(RelayError::Carrier(carrier)),
1103 }
1104 }
1105
1106 /// The longest end device frame the relay forwards at a data rate: what the end device
1107 /// may send behind a repeater, and what fits the relay's own uplink after the forwarded
1108 /// metadata, TS011-1.0.1 section 8.3.
1109 fn forwardable_len(&self, uplink: Carrier) -> Result<usize, RelayError> {
1110 let plan = self.device.plan();
1111 let device = plan
1112 .max_payload(uplink.data_rate, true)
1113 .map(|limit| usize::from(limit.mac_payload) + 5)
1114 .ok_or(RelayError::Carrier(uplink))?;
1115 let room = self.device.application_room(self.device.data_rate())?;
1116 Ok(device
1117 .min(room.saturating_sub(FORWARD_OVERHEAD))
1118 .min(MAX_FRAME))
1119 }
1120
1121 /// What a WOR ACK tells a trusted device about forwarding its uplink.
1122 fn forward_status(&self, index: u8, now_us: u64) -> Forward {
1123 let Some(device) = self.state.trusted.get(index) else {
1124 return Forward::Disabled;
1125 };
1126 let limits = &self.state.limits;
1127 let buckets = [device.bucket(), &limits.global_uplink, &limits.overall];
1128 if buckets.iter().all(|bucket| bucket.has_token()) {
1129 return Forward::Available;
1130 }
1131 if buckets.iter().any(|bucket| bucket.disabled()) {
1132 return Forward::Disabled;
1133 }
1134 let into_hour = self
1135 .state
1136 .anchor_us
1137 .map_or(0, |anchor| now_us.saturating_sub(anchor) % RELOAD_PERIOD_US);
1138 if RELOAD_PERIOD_US - into_hour <= RELOAD_PERIOD_US / 2 {
1139 Forward::RetryIn30Minutes
1140 } else {
1141 Forward::RetryIn60Minutes
1142 }
1143 }
1144
1145 /// Queues a notification of a device the relay does not trust, once while one for it
1146 /// is still to go out, TS011-1.0.1 section 8.7.
1147 fn notify(&mut self, dev_addr: u32, rssi_dbm: i16, snr_db: i8) -> Result<Wake, RelayError> {
1148 let addr = dev_addr.to_le_bytes();
1149 let key = [
1150 CID_NOTIFY_NEW_END_DEVICE,
1151 addr[0],
1152 addr[1],
1153 addr[2],
1154 addr[3],
1155 ];
1156 if self.device.starts_command(&key) {
1157 return Ok(Wake::Notified { dev_addr });
1158 }
1159 let limits = &mut self.state.limits;
1160 if !(limits.notify.has_token() && limits.overall.has_token()) {
1161 return Err(RelayError::Limited);
1162 }
1163 let queued = self
1164 .device
1165 .start_command(MacCommand::NotifyNewEndDeviceReq {
1166 dev_addr,
1167 rssi_dbm,
1168 snr_db,
1169 });
1170 if !queued {
1171 return Err(RelayError::Limited);
1172 }
1173 let limits = &mut self.state.limits;
1174 limits.notify.take();
1175 limits.overall.take();
1176 Ok(Wake::Notified { dev_addr })
1177 }
1178}
1179
1180/// The first `base + k * period` at or after `now`.
1181fn next_multiple(base: u64, period: u64, now: u64) -> u64 {
1182 if now <= base {
1183 base
1184 } else {
1185 base + (now - base).div_ceil(period) * period
1186 }
1187}