Data packet acks

Cargo.toml

@@ -25,7 +25,7 @@ rand = "0.8"
 base64 = ">=0.21"
 sha2 = "0"
 thiserror = "1.0"
-prost = "0"
+prost = "0.12"
 
 [dependencies]
 clap = { version = "4", default-features = false, features = [
@@ -61,8 +61,8 @@ tracing-appender = "0"
 thiserror = { workspace = true }
 rand = { workspace = true }
 prost = { workspace = true }
-tonic = "0"
-http = "*"
+tonic = "0.10"
+http = "0.2" # match with tonic's version
 sha2 = { workspace = true }
 base64 = { workspace = true }
 helium-proto = { workspace = true }

config/settings.toml

@@ -54,5 +54,13 @@ uri = "http://mainnet-config.helium.io:6080/"
 [router]
 uri = "http://mainnet-router.helium.io:8080/"
 # Maximum number of packets to queue up for the packet router
+# Do NOT set hiher than 25 to avoid hpr rate limiting restrictions
 queue = 20
+# When non 0, the numnber of seconds to expect an ack for uplink packets
+#
+# ack_timeout = 0
 
+# maximum time a packet is queued for before considered to stale to uplink. 
+# Default is 60s
+#
+# gc_timeout = 60

src/api/server.rs

@@ -37,7 +37,7 @@ impl LocalServer {
 
     pub async fn run(self, shutdown: &triggered::Listener) -> Result {
         let listen_addr = self.listen_addr;
-        tracing::Span::current().record("listen", &listen_addr.to_string());
+        tracing::Span::current().record("listen", listen_addr.to_string());
         info!(listen = %listen_addr, "starting");
         TransportServer::builder()
             .add_service(Server::new(self))

src/base64.rs

@@ -1,16 +1,6 @@
-use base64::{
-    engine::general_purpose::{STANDARD, URL_SAFE_NO_PAD},
-    Engine,
-};
+use base64::{engine::general_purpose::STANDARD, Engine};
 
 pub trait Base64 {
-    fn to_b64url(&self) -> String
-    where
-        Self: AsRef<[u8]>,
-    {
-        URL_SAFE_NO_PAD.encode(self.as_ref())
-    }
-
     fn to_b64(&self) -> String
     where
         Self: AsRef<[u8]>,

src/beaconer.rs

@@ -1,7 +1,7 @@
 //! This module provides proof-of-coverage (PoC) beaconing support.
 use crate::{
     gateway::{self, BeaconResp},
-    message_cache::MessageCache,
+    message_cache::{MessageCache, MessageHash},
     region_watcher,
     service::{entropy::EntropyService, poc::PocIotService, Reconnect},
     settings::Settings,
@@ -10,8 +10,9 @@ use crate::{
 use futures::TryFutureExt;
 use helium_proto::services::poc_lora::{self, lora_stream_response_v1};
 use http::Uri;
-use std::sync::Arc;
-use time::{Duration, Instant, OffsetDateTime};
+use sha2::{Digest, Sha256};
+use std::{sync::Arc, time::Instant};
+use time::{Duration, OffsetDateTime};
 use tracing::{info, warn};
 
 /// Message types that can be sent to `Beaconer`'s inbox.
@@ -57,6 +58,12 @@ pub struct Beaconer {
     entropy_uri: Uri,
 }
 
+impl MessageHash for Vec<u8> {
+    fn hash(&self) -> Vec<u8> {
+        Sha256::digest(self).to_vec()
+    }
+}
+
 impl Beaconer {
     pub fn new(
         settings: &Settings,
@@ -106,7 +113,7 @@ impl Beaconer {
                     info!("shutting down");
                     return Ok(())
                 },
-                _ = tokio::time::sleep_until(next_beacon_instant.into_inner().into()) => {
+                _ = tokio::time::sleep_until(next_beacon_instant.into()) => {
                     // Check if beaconing is enabled and we have valid region params
                     if !self.disabled && self.region_params.check_valid().is_ok() {
                         self.handle_beacon_tick().await;

src/message_cache.rs

@@ -4,19 +4,23 @@ use std::{
     time::{Duration, Instant},
 };
 
+pub trait MessageHash {
+    fn hash(&self) -> Vec<u8>;
+}
+
 #[derive(Debug)]
-pub struct MessageCache<T: PartialEq> {
+pub struct MessageCache<T: PartialEq + MessageHash> {
     cache: VecDeque<CacheMessage<T>>,
     max_messages: u16,
 }
 
 #[derive(Debug, Clone)]
-pub struct CacheMessage<T: PartialEq> {
-    received: Instant,
-    message: T,
+pub struct CacheMessage<T: PartialEq + MessageHash> {
+    pub received: Instant,
+    pub message: T,
 }
 
-impl<T: PartialEq> CacheMessage<T> {
+impl<T: PartialEq + MessageHash> CacheMessage<T> {
     pub fn new(message: T, received: Instant) -> Self {
         Self { message, received }
     }
@@ -26,15 +30,15 @@ impl<T: PartialEq> CacheMessage<T> {
     }
 }
 
-impl<T: PartialEq> Deref for CacheMessage<T> {
+impl<T: PartialEq + MessageHash> Deref for CacheMessage<T> {
     type Target = T;
 
     fn deref(&self) -> &Self::Target {
         &self.message
     }
 }
 
-impl<T: PartialEq> MessageCache<T> {
+impl<T: PartialEq + MessageHash> MessageCache<T> {
     pub fn new(max_messages: u16) -> Self {
         let waiting = VecDeque::new();
         Self {
@@ -49,26 +53,30 @@ impl<T: PartialEq> MessageCache<T> {
     ///
     /// Pushing a packet onto the back of a full cache will cause the oldest
     /// (first) message in the cache to be dropped.
-    pub fn push_back(&mut self, message: T, received: Instant) -> Option<CacheMessage<T>> {
-        self.cache.push_back(CacheMessage::new(message, received));
+    pub fn push_back(&mut self, message: T, received: Instant) -> &CacheMessage<T> {
+        let message = CacheMessage::new(message, received);
+        self.cache.push_back(message);
         if self.len() > self.max_messages as usize {
-            self.cache.pop_front()
-        } else {
-            None
+            self.cache.pop_front();
         }
+        // safe to unwrap given that the message we just pushed to the back
+        self.cache.back().unwrap()
     }
 
     /// Returns the index of the first matching message in the cache or None if
     /// not present
-    pub fn index_of(&self, message: &T) -> Option<usize> {
-        self.cache.iter().position(|m| m.message == *message)
+    pub fn index_of<P>(&self, pred: P) -> Option<usize>
+    where
+        P: Fn(&T) -> bool,
+    {
+        self.cache.iter().position(|entry| pred(&entry.message))
     }
 
     /// Promotes the given message to the back of the queue, effectively
     /// recreating an LRU cache. Returns true if a cache hit was found
     pub fn tag(&mut self, message: T, received: Instant) -> bool {
         let result = self
-            .index_of(&message)
+            .index_of(|msg| *msg == message)
             .and_then(|index| self.cache.remove(index))
             .is_some();
         self.push_back(message, received);
@@ -99,18 +107,32 @@ impl<T: PartialEq> MessageCache<T> {
                 front = Some(msg);
                 break;
             }
-            // held for too long, count as dropped and move on
             dropped += 1;
         }
         (dropped, front)
     }
 
+    /// Removes all items from the cache up to and including the given index.
+    ///
+    /// The index is bounds checked and an index beyond the length of the cache
+    /// is ignored
+    pub fn remove_to(&mut self, index: usize) {
+        if index >= self.len() {
+            return;
+        }
+        self.cache = self.cache.split_off(index + 1);
+    }
+
     /// Returns a reference to the first (and oldest/first to be removed)
     /// message in the cache
     pub fn peek_front(&self) -> Option<&CacheMessage<T>> {
         self.cache.front()
     }
 
+    pub fn peek_back(&self) -> Option<&CacheMessage<T>> {
+        self.cache.back()
+    }
+
     pub fn len(&self) -> usize {
         self.cache.len()
     }
@@ -122,7 +144,8 @@ impl<T: PartialEq> MessageCache<T> {
 
 #[cfg(test)]
 mod test {
-    use super::MessageCache;
+    use super::{Instant, MessageCache};
+    use sha2::{Digest, Sha256};
 
     #[test]
     fn test_cache_tagging() {
@@ -142,8 +165,36 @@ mod test {
 
         // Third tag should evict the least recently used entry (2)
         assert!(!cache.tag_now(vec![3]));
-        assert_eq!(Some(0), cache.index_of(&vec![1u8]));
-        assert_eq!(Some(1), cache.index_of(&vec![3u8]));
-        assert!(cache.index_of(&vec![2u8]).is_none());
+        assert_eq!(Some(0), cache.index_of(|msg| msg.as_slice() == &[1u8]));
+        assert_eq!(Some(1), cache.index_of(|msg| msg.as_slice() == &[3u8]));
+        assert!(cache.index_of(|msg| msg.as_slice() == &[2u8]).is_none());
+    }
+
+    #[test]
+    fn test_remove_to() {
+        let mut cache = MessageCache::<Vec<u8>>::new(5);
+        cache.push_back(vec![1], Instant::now());
+        cache.push_back(vec![2], Instant::now());
+        cache.push_back(vec![3], Instant::now());
+
+        let ack = Sha256::digest(vec![2]).to_vec();
+
+        // Find entry by hash as an example
+        let ack_index = cache.index_of(|msg| Sha256::digest(msg).to_vec() == ack);
+        assert_eq!(Some(1), ack_index);
+        // Can't find non existing
+        assert_eq!(None, cache.index_of(|_| false));
+
+        // remove and check inclusion of remove_to
+        cache.remove_to(1);
+        assert_eq!(1, cache.len());
+
+        // remove past last index
+        cache.remove_to(5);
+        assert_eq!(1, cache.len());
+
+        // remove last element
+        cache.remove_to(0);
+        assert!(cache.is_empty());
     }
 }

src/packet.rs

@@ -18,7 +18,10 @@ use std::{
 };
 
 #[derive(Debug, Clone, PartialEq)]
-pub struct PacketUp(PacketRouterPacketUpV1);
+pub struct PacketUp {
+    packet: PacketRouterPacketUpV1,
+    pub(crate) hash: Vec<u8>,
+}
 
 #[derive(Debug, Clone)]
 pub struct PacketDown(PacketRouterPacketDownV1);
@@ -27,18 +30,13 @@ impl Deref for PacketUp {
     type Target = PacketRouterPacketUpV1;
 
     fn deref(&self) -> &Self::Target {
-        &self.0
+        &self.packet
     }
 }
 
 impl From<PacketUp> for PacketRouterPacketUpV1 {
     fn from(value: PacketUp) -> Self {
-        value.0
-    }
-}
-impl From<&PacketUp> for PacketRouterPacketUpV1 {
-    fn from(value: &PacketUp) -> Self {
-        value.0.clone()
+        value.packet
     }
 }
 
@@ -52,12 +50,12 @@ impl fmt::Display for PacketUp {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_fmt(format_args!(
             "@{} us, {:.2} MHz, {:?}, snr: {}, rssi: {}, len: {}",
-            self.0.timestamp,
-            self.0.frequency,
-            self.0.datarate(),
-            self.0.snr,
-            self.0.rssi,
-            self.0.payload.len()
+            self.packet.timestamp,
+            self.packet.frequency,
+            self.packet.datarate(),
+            self.packet.snr,
+            self.packet.rssi,
+            self.packet.payload.len()
         ))
     }
 }
@@ -67,15 +65,15 @@ impl TryFrom<PacketUp> for poc_lora::LoraWitnessReportReqV1 {
     fn try_from(value: PacketUp) -> Result<Self> {
         let report = poc_lora::LoraWitnessReportReqV1 {
             data: vec![],
-            tmst: value.0.timestamp as u32,
+            tmst: value.packet.timestamp as u32,
             timestamp: SystemTime::now()
                 .duration_since(UNIX_EPOCH)
                 .map_err(Error::from)?
                 .as_nanos() as u64,
-            signal: value.0.rssi * 10,
-            snr: (value.0.snr * 10.0) as i32,
-            frequency: value.0.frequency as u64,
-            datarate: value.0.datarate,
+            signal: value.packet.rssi * 10,
+            snr: (value.packet.snr * 10.0) as i32,
+            frequency: value.packet.frequency as u64,
+            datarate: value.packet.datarate,
             pub_key: vec![],
             signature: vec![],
         };
@@ -107,7 +105,10 @@ impl PacketUp {
             gateway: gateway.into(),
             signature: vec![],
         };
-        Ok(Self(packet))
+        Ok(Self {
+            hash: Sha256::digest(&packet.payload).to_vec(),
+            packet,
+        })
     }
 
     pub fn is_potential_beacon(&self) -> bool {
@@ -133,7 +134,7 @@ impl PacketUp {
     }
 
     pub fn payload(&self) -> &[u8] {
-        &self.0.payload
+        &self.packet.payload
     }
 
     pub fn parse_header(payload: &[u8]) -> Result<MHDR> {
@@ -151,10 +152,6 @@ impl PacketUp {
             .map(|p| p.payload)
             .map_err(Error::from)
     }
-
-    pub fn hash(&self) -> Vec<u8> {
-        Sha256::digest(&self.0.payload).to_vec()
-    }
 }
 
 impl PacketDown {