feat(core): Store::release and is_released (0.3.0)

tines-core/src/exchange.rs

@@ -1820,6 +1820,12 @@ mod tests {
         fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) {
             self.inner.for_each_addressed_id(f)
         }
+        fn release(&mut self, id: &ObservationId) {
+            self.inner.release(id)
+        }
+        fn is_released(&self, id: &ObservationId) -> bool {
+            self.inner.is_released(id)
+        }
         fn for_each_topic(&self, prefix: &[u8], now: Tick, f: &mut dyn FnMut(&Observation)) {
             self.inner.for_each_topic(prefix, now, f)
         }

tines-core/src/mem_store.rs

@@ -21,6 +21,7 @@ use alloc::vec::Vec;
 struct Held {
     obs: Observation,
     stored_at: Tick,
+    released: bool,
 }
 
 /// An unbounded [`Store`] backed by a `BTreeMap` keyed by content-derived
@@ -77,6 +78,7 @@ impl Store for MemStore {
             Held {
                 obs,
                 stored_at: now,
+                released: false,
             },
         );
         Ok(true)
@@ -112,12 +114,24 @@ impl Store for MemStore {
 
     fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) {
         for (id, held) in self.map.iter() {
-            if held.obs.to.is_some() {
+            // Skipping released ids is an optimization; the exchange predicate
+            // is the load-bearing check.
+            if held.obs.to.is_some() && !held.released {
                 f(id);
             }
         }
     }
 
+    fn release(&mut self, id: &ObservationId) {
+        if let Some(held) = self.map.get_mut(id) {
+            held.released = true;
+        }
+    }
+
+    fn is_released(&self, id: &ObservationId) -> bool {
+        self.map.get(id).map(|held| held.released).unwrap_or(false)
+    }
+
     fn for_each_topic(&self, prefix: &[u8], now: Tick, f: &mut dyn FnMut(&Observation)) {
         for held in self.map.values() {
             if held.obs.topic.starts_with(prefix) {
@@ -438,4 +452,24 @@ mod tests {
             std::collections::BTreeSet::from_iter([0u64, 1, 2, 100])
         );
     }
+
+    #[test]
+    fn release_marks_and_is_released_reports() {
+        let mut s = MemStore::new();
+        let o = obs([1; 16], 1, 0, None, b"door/1");
+        let id = o.id();
+        assert!(s.insert(o, 1000).unwrap());
+        assert!(!s.is_released(&id));
+        s.release(&id);
+        assert!(s.is_released(&id));
+    }
+
+    #[test]
+    fn release_unknown_is_noop() {
+        let mut s = MemStore::new();
+        // Releasing an id that was never inserted must not panic...
+        s.release(&[0u8; 16]);
+        // ...and must not report anything as released.
+        assert!(!s.is_released(&[0u8; 16]));
+    }
 }

tines-core/src/store.rs

@@ -74,6 +74,17 @@ pub trait Store {
         self.for_each_id(f)
     }
 
+    /// Mark the observation with `id` as released: it has been sent and must
+    /// not be offered for exchange again.
+    ///
+    /// Releasing an id that is not currently stored is a no-op. The mark
+    /// persists for the lifetime of the entry (until it is trimmed); a
+    /// re-inserted observation starts un-released.
+    fn release(&mut self, id: &ObservationId);
+
+    /// Returns `true` if the observation with `id` has been released.
+    fn is_released(&self, id: &ObservationId) -> bool;
+
     /// Visit every stored observation whose topic starts with `prefix`
     /// (an empty prefix matches all), each age-adjusted to `now`.
     fn for_each_topic(&self, prefix: &[u8], now: Tick, f: &mut dyn FnMut(&Observation));