Da service backend

nomos-core/src/blob/mod.rs

@@ -0,0 +1,13 @@
+use bytes::Bytes;
+use std::hash::Hash;
+
+pub type BlobHasher<T> = fn(&T) -> <T as Blob>::Hash;
+
+pub trait Blob {
+    const HASHER: BlobHasher<Self>;
+    type Hash: Hash + Eq + Clone;
+    fn hash(&self) -> Self::Hash {
+        Self::HASHER(self)
+    }
+    fn as_bytes(&self) -> Bytes;
+}

nomos-core/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod account;
+pub mod blob;
 pub mod block;
 pub mod crypto;
 pub mod fountain;

nomos-services/data-availability/Cargo.toml

@@ -8,6 +8,8 @@ edition = "2021"
 [dependencies]
 async-trait = "0.1"
 futures = "0.3"
+moka = { version = "0.11", features = ["future"] }
+nomos-core = { path = "../../nomos-core" }
 nomos-network = { path = "../network" }
 overwatch-rs = { git = "https://github.com/logos-co/Overwatch", branch = "main" }
 tracing = "0.1"

nomos-services/data-availability/src/backend/memory_cache.rs

@@ -0,0 +1,73 @@
+use crate::backend::{DaBackend, DaError};
+use moka::future::{Cache, CacheBuilder};
+use nomos_core::blob::Blob;
+use std::time::Duration;
+
+#[derive(Clone, Copy)]
+pub struct BlobCacheSettings {
+    max_capacity: usize,
+    evicting_period: Duration,
+}
+
+pub struct BlobCache<H, B>(Cache<H, B>);
+
+impl<B> BlobCache<B::Hash, B>
+where
+    B: Clone + Blob + Send + Sync + 'static,
+    B::Hash: Send + Sync + 'static,
+{
+    pub fn new(settings: BlobCacheSettings) -> Self {
+        let BlobCacheSettings {
+            max_capacity,
+            evicting_period,
+        } = settings;
+        let cache = CacheBuilder::new(max_capacity as u64)
+            .time_to_live(evicting_period)
+            // can we leverage this to evict really old blobs?
+            .time_to_idle(evicting_period)
+            .build();
+        Self(cache)
+    }
+
+    pub async fn add(&self, blob: B) {
+        self.0.insert(blob.hash(), blob).await
+    }
+
+    pub async fn remove(&self, hash: &B::Hash) {
+        self.0.remove(hash).await;
+    }
+
+    pub fn pending_blobs(&self) -> Box<dyn Iterator<Item = B> + Send> {
+        // bypass lifetime
+        let blobs: Vec<_> = self.0.iter().map(|t| t.1).collect();
+        Box::new(blobs.into_iter())
+    }
+}
+
+#[async_trait::async_trait]
+impl<B> DaBackend for BlobCache<B::Hash, B>
+where
+    B: Clone + Blob + Send + Sync + 'static,
+    B::Hash: Send + Sync + 'static,
+{
+    type Settings = BlobCacheSettings;
+    type Blob = B;
+
+    fn new(settings: Self::Settings) -> Self {
+        BlobCache::new(settings)
+    }
+
+    async fn add_blob(&self, blob: Self::Blob) -> Result<(), DaError> {
+        self.add(blob).await;
+        Ok(())
+    }
+
+    async fn remove_blob(&self, blob: &<Self::Blob as Blob>::Hash) -> Result<(), DaError> {
+        self.remove(blob).await;
+        Ok(())
+    }
+
+    fn pending_blobs(&self) -> Box<dyn Iterator<Item = Self::Blob> + Send> {
+        BlobCache::pending_blobs(self)
+    }
+}

nomos-services/data-availability/src/backend/mod.rs

@@ -1,18 +1,24 @@
+mod memory_cache;
+
+use nomos_core::blob::Blob;
 use overwatch_rs::DynError;
 
 #[derive(Debug)]
 pub enum DaError {
     Dyn(DynError),
 }
 
+#[async_trait::async_trait]
 pub trait DaBackend {
     type Settings: Clone;
 
-    type Blob;
+    type Blob: Blob;
 
     fn new(settings: Self::Settings) -> Self;
 
-    fn add_blob(&mut self, blob: Self::Blob) -> Result<(), DaError>;
+    async fn add_blob(&self, blob: Self::Blob) -> Result<(), DaError>;
+
+    async fn remove_blob(&self, blob: &<Self::Blob as Blob>::Hash) -> Result<(), DaError>;
 
     fn pending_blobs(&self) -> Box<dyn Iterator<Item = Self::Blob> + Send>;
 }

nomos-services/data-availability/src/lib.rs

@@ -10,6 +10,7 @@ use tokio::sync::oneshot::Sender;
 // internal
 use crate::backend::{DaBackend, DaError};
 use crate::network::NetworkAdapter;
+use nomos_core::blob::Blob;
 use nomos_network::NetworkService;
 use overwatch_rs::services::handle::ServiceStateHandle;
 use overwatch_rs::services::relay::{Relay, RelayMessage};
@@ -27,23 +28,29 @@ where
     network_relay: Relay<NetworkService<N::Backend>>,
 }
 
-pub enum DaMsg<Blob> {
+pub enum DaMsg<B: Blob> {
     PendingBlobs {
-        reply_channel: Sender<Box<dyn Iterator<Item = Blob> + Send>>,
+        reply_channel: Sender<Box<dyn Iterator<Item = B> + Send>>,
+    },
+    RemoveBlobs {
+        blobs: Box<dyn Iterator<Item = <B as Blob>::Hash> + Send>,
     },
 }
 
-impl<Blob: 'static> Debug for DaMsg<Blob> {
+impl<B: Blob + 'static> Debug for DaMsg<B> {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match self {
             DaMsg::PendingBlobs { .. } => {
                 write!(f, "DaMsg::PendingBlobs")
             }
+            DaMsg::RemoveBlobs { .. } => {
+                write!(f, "DaMsg::RemoveBlobs")
+            }
         }
     }
 }
 
-impl<Blob: 'static> RelayMessage for DaMsg<Blob> {}
+impl<B: Blob + 'static> RelayMessage for DaMsg<B> {}
 
 impl<B, N> ServiceData for DataAvailabilityService<B, N>
 where
@@ -61,9 +68,10 @@ where
 #[async_trait::async_trait]
 impl<B, N> ServiceCore for DataAvailabilityService<B, N>
 where
-    B: DaBackend + Send,
+    B: DaBackend + Send + Sync,
     B::Settings: Clone + Send + Sync + 'static,
     B::Blob: Send,
+    <B::Blob as Blob>::Hash: Debug + Send + Sync,
     // TODO: Reply type must be piped together, for now empty array.
     N: NetworkAdapter<Blob = B::Blob, Reply = [u8; 32]> + Send + Sync,
 {
@@ -115,21 +123,34 @@ async fn handle_new_blob<B: DaBackend, A: NetworkAdapter<Blob = B::Blob, Reply =
     blob: B::Blob,
 ) -> Result<(), DaError> {
     // we need to handle the reply (verification + signature)
-    backend.add_blob(blob)?;
+    backend.add_blob(blob).await?;
     adapter
         .send_attestation([0u8; 32])
         .await
         .map_err(DaError::Dyn)
 }
 
-async fn handle_da_msg<B: DaBackend>(backend: &mut B, msg: DaMsg<B::Blob>) -> Result<(), DaError> {
+async fn handle_da_msg<B: DaBackend>(backend: &mut B, msg: DaMsg<B::Blob>) -> Result<(), DaError>
+where
+    <B::Blob as Blob>::Hash: Debug,
+{
     match msg {
         DaMsg::PendingBlobs { reply_channel } => {
             let pending_blobs = backend.pending_blobs();
             if reply_channel.send(pending_blobs).is_err() {
                 tracing::debug!("Could not send pending blobs");
             }
         }
+        DaMsg::RemoveBlobs { blobs } => {
+            let backend = &*backend;
+            futures::stream::iter(blobs)
+                .for_each_concurrent(None, |blob| async move {
+                    if let Err(e) = backend.remove_blob(&blob).await {
+                        tracing::debug!("Could not remove blob {blob:?} due to: {e:?}");
+                    }
+                })
+                .await;
+        }
     }
     Ok(())
 }

nomos-services/data-availability/src/network/mod.rs

@@ -1,12 +1,12 @@
 // std
 // crates
 use futures::Stream;
-use overwatch_rs::DynError;
 // internal
 use nomos_network::backends::NetworkBackend;
 use nomos_network::NetworkService;
 use overwatch_rs::services::relay::OutboundRelay;
 use overwatch_rs::services::ServiceData;
+use overwatch_rs::DynError;
 
 #[async_trait::async_trait]
 pub trait NetworkAdapter {