Data packet acks

Cargo.toml

@@ -18,7 +18,7 @@ members = ["lorawan"]
 byteorder = "1"
 serde = { version = "1", features = ["rc", "derive"] }
 rust_decimal = { version = "1", features = ["serde-with-float"] }
-helium-proto = { git = "https://github.com/helium/proto", branch = "master", features = [
+helium-proto = { git = "https://github.com/helium/proto", branch = "madninja/packet_ack", features = [
     "services",
 ] }
 rand = "0.8"
@@ -62,15 +62,15 @@ thiserror = { workspace = true }
 rand = { workspace = true }
 prost = { workspace = true }
 tonic = "0"
-http = "*"
+http = "0"
 sha2 = { workspace = true }
 base64 = { workspace = true }
 helium-proto = { workspace = true }
 signature = { version = "1", features = ["std"] }
 async-trait = "0"
 angry-purple-tiger = "0"
 lorawan = { package = "lorawan", path = "lorawan" }
-beacon = { git = "https://github.com/helium/proto", branch = "master" }
+beacon = { git = "https://github.com/helium/proto", branch = "madninja/packet_ack" }
 exponential-backoff = { git = "https://github.com/yoshuawuyts/exponential-backoff", branch = "master" }
 semtech-udp = { version = ">=0.11", default-features = false, features = [
     "server",

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,6 +10,7 @@ use crate::{
 use futures::TryFutureExt;
 use helium_proto::services::poc_lora::{self, lora_stream_response_v1};
 use http::Uri;
+use sha2::{Digest, Sha256};
 use std::sync::Arc;
 use time::{Duration, Instant, OffsetDateTime};
 use tracing::{info, warn};
@@ -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,

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> {
+pub struct CacheMessage<T: PartialEq + MessageHash> {
     received: Instant,
     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 {
@@ -60,15 +64,18 @@ impl<T: PartialEq> MessageCache<T> {
 
     /// 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,12 +106,22 @@ 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>> {
@@ -122,7 +139,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 +160,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,19 @@ 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
+        value.packet
     }
 }
+
 impl From<&PacketUp> for PacketRouterPacketUpV1 {
     fn from(value: &PacketUp) -> Self {
-        value.0.clone()
+        value.packet.clone()
     }
 }
 
@@ -52,12 +56,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 +71,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 +111,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 +140,7 @@ impl PacketUp {
     }
 
     pub fn payload(&self) -> &[u8] {
-        &self.0.payload
+        &self.packet.payload
     }
 
     pub fn parse_header(payload: &[u8]) -> Result<MHDR> {
@@ -151,10 +158,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 {