From 449203ea709a2a1123364e1d059ceed6df324be9 Mon Sep 17 00:00:00 2001 From: Andrus Salumets Date: Tue, 13 May 2025 20:05:38 +0200 Subject: [PATCH] New storage abstraction layer --- .cargo-deny.toml | 1 - Cargo.toml | 2 - dependencies_graph.dot | 97 ++++++------ nomos-core/chain-defs/src/block/mod.rs | 25 +++ nomos-da/storage/Cargo.toml | 8 - nomos-da/storage/src/lib.rs | 1 - nomos-services/api/Cargo.toml | 1 - nomos-services/api/src/http/da_shares.rs | 7 +- nomos-services/api/src/http/storage.rs | 7 +- .../cryptarchia-consensus/Cargo.toml | 1 - .../cryptarchia-consensus/src/lib.rs | 19 ++- .../cryptarchia-consensus/src/relays.rs | 13 +- .../src/storage/adapters/storage.rs | 82 +++++----- .../cryptarchia-consensus/src/storage/mod.rs | 8 +- .../data-availability/indexer/Cargo.toml | 1 - .../indexer/src/storage/adapters/rocksdb.rs | 7 +- .../data-availability/sampling/Cargo.toml | 1 - .../sampling/src/storage/adapters/rocksdb.rs | 33 ++-- .../data-availability/verifier/Cargo.toml | 1 - .../verifier/src/storage/adapters/rocksdb.rs | 104 +++---------- nomos-services/storage/Cargo.toml | 5 +- nomos-services/storage/src/api/backend/mod.rs | 2 + .../storage/src/api/backend/rocksdb/blocks.rs | 30 ++++ .../storage/src/api/backend/rocksdb/da.rs | 95 ++++++++++++ .../storage/src/api/backend/rocksdb/mod.rs | 10 ++ .../storage/src/api/backend/rocksdb/utils.rs | 9 +- nomos-services/storage/src/api/chain/mod.rs | 20 +++ .../storage/src/api/chain/requests.rs | 92 +++++++++++ nomos-services/storage/src/api/da/mod.rs | 33 ++++ nomos-services/storage/src/api/da/requests.rs | 136 +++++++++++++++++ nomos-services/storage/src/api/mod.rs | 35 +++++ nomos-services/storage/src/backends/mock.rs | 85 ++++++++++- nomos-services/storage/src/backends/mod.rs | 28 ++-- .../storage/src/backends/rocksdb.rs | 31 ++-- nomos-services/storage/src/backends/sled.rs | 143 ------------------ nomos-services/storage/src/lib.rs | 83 ++++++---- 36 files changed, 828 insertions(+), 428 deletions(-) delete mode 100644 nomos-da/storage/Cargo.toml delete mode 100644 nomos-da/storage/src/lib.rs create mode 100644 nomos-services/storage/src/api/backend/mod.rs create mode 100644 nomos-services/storage/src/api/backend/rocksdb/blocks.rs create mode 100644 nomos-services/storage/src/api/backend/rocksdb/da.rs create mode 100644 nomos-services/storage/src/api/backend/rocksdb/mod.rs rename nomos-da/storage/src/rocksdb/mod.rs => nomos-services/storage/src/api/backend/rocksdb/utils.rs (66%) create mode 100644 nomos-services/storage/src/api/chain/mod.rs create mode 100644 nomos-services/storage/src/api/chain/requests.rs create mode 100644 nomos-services/storage/src/api/da/mod.rs create mode 100644 nomos-services/storage/src/api/da/requests.rs create mode 100644 nomos-services/storage/src/api/mod.rs delete mode 100644 nomos-services/storage/src/backends/sled.rs diff --git a/.cargo-deny.toml b/.cargo-deny.toml index 4f76d5c20..2fec72a6e 100644 --- a/.cargo-deny.toml +++ b/.cargo-deny.toml @@ -10,7 +10,6 @@ ignore = [ { id = "RUSTSEC-2020-0095", reason = "`difference` is unmaintained; consider using an alternative. Use `cargo tree -p difference -i > tmp.txt` to check the dependency tree." }, { id = "RUSTSEC-2023-0071", reason = "Marvin Attack: potential key recovery through timing sidechannels" }, { id = "RUSTSEC-2024-0370", reason = "`proc-macro-error` is unmaintained; consider using an alternative. Use `cargo tree -p proc-macro-error -i > tmp.txt` to check the dependency tree." }, - { id = "RUSTSEC-2024-0384", reason = "`instant` is unmaintained; consider using an alternative. Use `cargo tree -p instant -i > tmp.txt` to check the dependency tree." }, { id = "RUSTSEC-2024-0388", reason = "`derivative` is unmaintained; consider using an alternative. Use `cargo tree -p derivative -i > tmp.txt` to check the dependency tree." }, { id = "RUSTSEC-2024-0436", reason = "`paste` has a security vulnerability; consider using an alternative. Use `cargo tree -p paste -i > tmp.txt` to check the dependency tree." }, ] diff --git a/Cargo.toml b/Cargo.toml index b26c6fd1a..1536f51a8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,7 +22,6 @@ members = [ "nomos-da/network/core", "nomos-da/network/messages", "nomos-da/network/subnetworks-assignations", - "nomos-da/storage", "nomos-libp2p", "nomos-sdp", "nomos-services/api", @@ -75,7 +74,6 @@ nomos-da-messages = { path = "./nomos-da/network/messages" } nomos-da-network-core = { path = "./nomos-da/network/core" } nomos-da-network-service = { path = "./nomos-services/data-availability/network" } nomos-da-sampling = { path = "./nomos-services/data-availability/sampling" } -nomos-da-storage = { path = "./nomos-da/storage" } nomos-da-verifier = { path = "./nomos-services/data-availability/verifier" } nomos-executor = { path = "./nodes/nomos-executor/executor", default-features = false } nomos-http-api-common = { path = "./nodes/api-common" } diff --git a/dependencies_graph.dot b/dependencies_graph.dot index 85634c090..7b5f703d1 100644 --- a/dependencies_graph.dot +++ b/dependencies_graph.dot @@ -25,38 +25,37 @@ digraph { 23 [ label = "subnetworks-assignations" shape = box] 24 [ label = "nomos-da-network-service" shape = box] 25 [ label = "nomos-tracing" shape = box] - 26 [ label = "nomos-da-storage" shape = box] - 27 [ label = "nomos-da-verifier" shape = box] - 28 [ label = "nomos-storage" shape = box] - 29 [ label = "nomos-mempool" shape = box] - 30 [ label = "nomos-tracing-service" shape = box] - 31 [ label = "nomos-time" shape = box] - 32 [ label = "nomos-da-dispersal" shape = box] - 33 [ label = "nomos-da-indexer" shape = box] - 34 [ label = "nomos-node" shape = box] - 35 [ label = "nomos-system-sig" shape = box] - 36 [ label = "executor-http-client" shape = box] - 37 [ label = "bundler" shape = box] - 38 [ label = "nomos-cli" shape = box] - 39 [ label = "full-replication" shape = box] - 40 [ label = "nomos-sdp-core" shape = box] - 41 [ label = "key-management-system" shape = box] - 42 [ label = "nomos-sdp" shape = box] - 43 [ label = "cfgsync" shape = box] - 44 [ label = "tests" shape = box] + 26 [ label = "nomos-da-verifier" shape = box] + 27 [ label = "nomos-storage" shape = box] + 28 [ label = "nomos-mempool" shape = box] + 29 [ label = "nomos-tracing-service" shape = box] + 30 [ label = "nomos-time" shape = box] + 31 [ label = "nomos-da-dispersal" shape = box] + 32 [ label = "nomos-da-indexer" shape = box] + 33 [ label = "nomos-node" shape = box] + 34 [ label = "nomos-system-sig" shape = box] + 35 [ label = "executor-http-client" shape = box] + 36 [ label = "bundler" shape = box] + 37 [ label = "nomos-cli" shape = box] + 38 [ label = "full-replication" shape = box] + 39 [ label = "nomos-sdp-core" shape = box] + 40 [ label = "key-management-system" shape = box] + 41 [ label = "nomos-sdp" shape = box] + 42 [ label = "cfgsync" shape = box] + 43 [ label = "tests" shape = box] 0 -> 1 [ style = dotted] 2 -> 0 [ ] 2 -> 4 [ ] 4 -> 3 [ ] - 6 -> 34 [ ] + 6 -> 33 [ ] 7 -> 8 [ ] 7 -> 9 [ ] 9 -> 2 [ ] + 10 -> 31 [ ] 10 -> 32 [ ] - 10 -> 33 [ ] 11 -> 12 [ ] - 11 -> 29 [ ] - 11 -> 31 [ ] + 11 -> 28 [ ] + 11 -> 30 [ ] 12 -> 15 [ ] 12 -> 9 [ ] 12 -> 17 [ ] @@ -66,7 +65,7 @@ digraph { 17 -> 16 [ style = dotted] 17 -> 18 [ ] 19 -> 20 [ ] - 19 -> 27 [ ] + 19 -> 26 [ ] 20 -> 21 [ ] 20 -> 5 [ ] 21 -> 7 [ ] @@ -77,31 +76,31 @@ digraph { 24 -> 16 [ ] 24 -> 25 [ ] 24 -> 18 [ ] - 27 -> 24 [ ] - 27 -> 26 [ ] - 27 -> 28 [ ] - 28 -> 18 [ ] - 29 -> 19 [ ] - 29 -> 29 [ color = blue] - 29 -> 17 [ ] - 29 -> 30 [ color = blue] - 30 -> 25 [ ] - 31 -> 0 [ ] - 31 -> 18 [ ] - 32 -> 29 [ ] - 33 -> 11 [ ] - 34 -> 10 [ ] - 34 -> 35 [ ] - 35 -> 18 [ ] - 36 -> 20 [ ] - 38 -> 36 [ ] - 38 -> 34 [ ] - 39 -> 9 [ ] + 26 -> 24 [ ] + 26 -> 27 [ ] + 27 -> 9 [ ] + 27 -> 18 [ ] + 28 -> 19 [ ] + 28 -> 28 [ color = blue] + 28 -> 17 [ ] + 28 -> 29 [ color = blue] + 29 -> 25 [ ] + 30 -> 0 [ ] + 30 -> 18 [ ] + 31 -> 28 [ ] + 32 -> 11 [ ] + 33 -> 10 [ ] + 33 -> 34 [ ] + 34 -> 18 [ ] + 35 -> 20 [ ] + 37 -> 35 [ ] + 37 -> 33 [ ] + 38 -> 9 [ ] + 40 -> 18 [ ] + 41 -> 39 [ ] 41 -> 18 [ ] - 42 -> 40 [ ] - 42 -> 18 [ ] - 43 -> 44 [ ] - 44 -> 36 [ ] - 44 -> 6 [ ] + 42 -> 43 [ ] + 43 -> 35 [ ] + 43 -> 6 [ ] } diff --git a/nomos-core/chain-defs/src/block/mod.rs b/nomos-core/chain-defs/src/block/mod.rs index 8c00129c5..5f0212348 100644 --- a/nomos-core/chain-defs/src/block/mod.rs +++ b/nomos-core/chain-defs/src/block/mod.rs @@ -59,3 +59,28 @@ impl< self.bl_blobs.len() } } + +impl< + Tx: Clone + Eq + Hash + Serialize + DeserializeOwned, + BlobCertificate: Clone + Eq + Hash + Serialize + DeserializeOwned, + > TryFrom for Block +{ + type Error = wire::Error; + + fn try_from(bytes: Bytes) -> Result { + wire::deserialize(&bytes) + } +} + +impl< + Tx: Clone + Eq + Hash + Serialize + DeserializeOwned, + BlobCertificate: Clone + Eq + Hash + Serialize + DeserializeOwned, + > TryFrom> for Bytes +{ + type Error = wire::Error; + + fn try_from(block: Block) -> Result { + let serialized = wire::serialize(&block)?; + Ok(serialized.into()) + } +} diff --git a/nomos-da/storage/Cargo.toml b/nomos-da/storage/Cargo.toml deleted file mode 100644 index d6c0bb658..000000000 --- a/nomos-da/storage/Cargo.toml +++ /dev/null @@ -1,8 +0,0 @@ -[package] -edition = "2021" -license = { workspace = true } -name = "nomos-da-storage" -version = "0.1.0" - -[dependencies] -bytes = "1.2" diff --git a/nomos-da/storage/src/lib.rs b/nomos-da/storage/src/lib.rs deleted file mode 100644 index 36917b1a6..000000000 --- a/nomos-da/storage/src/lib.rs +++ /dev/null @@ -1 +0,0 @@ -pub mod rocksdb; diff --git a/nomos-services/api/Cargo.toml b/nomos-services/api/Cargo.toml index e6538161c..a75725afe 100644 --- a/nomos-services/api/Cargo.toml +++ b/nomos-services/api/Cargo.toml @@ -22,7 +22,6 @@ nomos-da-indexer = { workspace = true, features = ["rocksdb-backend"] } nomos-da-network-core = { workspace = true } nomos-da-network-service = { workspace = true } nomos-da-sampling = { workspace = true } -nomos-da-storage = { workspace = true } nomos-da-verifier = { workspace = true, features = ["libp2p", "rocksdb-backend"] } nomos-libp2p = { workspace = true } nomos-mempool = { workspace = true, features = ["libp2p", "mock", "openapi"] } diff --git a/nomos-services/api/src/http/da_shares.rs b/nomos-services/api/src/http/da_shares.rs index 8234f242d..fc7fa666f 100644 --- a/nomos-services/api/src/http/da_shares.rs +++ b/nomos-services/api/src/http/da_shares.rs @@ -9,10 +9,11 @@ use std::{ use bytes::Bytes; use futures::{stream, Stream, StreamExt as _}; use nomos_core::da::blob::{LightShare, Share}; -use nomos_da_storage::rocksdb::{ - create_share_idx, key_bytes, DA_BLOB_SHARES_INDEX_PREFIX, DA_SHARE_PREFIX, -}; use nomos_storage::{ + api::backend::rocksdb::{ + da::{DA_BLOB_SHARES_INDEX_PREFIX, DA_SHARE_PREFIX}, + utils::{create_share_idx, key_bytes}, + }, backends::{rocksdb::RocksBackend, StorageSerde}, StorageMsg, StorageService, }; diff --git a/nomos-services/api/src/http/storage.rs b/nomos-services/api/src/http/storage.rs index 15163dd6d..3ac9cdb6a 100644 --- a/nomos-services/api/src/http/storage.rs +++ b/nomos-services/api/src/http/storage.rs @@ -2,10 +2,11 @@ use std::fmt::{Debug, Display}; use bytes::Bytes; use nomos_core::{block::Block, da::blob::Share, header::HeaderId}; -use nomos_da_storage::rocksdb::{ - create_share_idx, key_bytes, DA_SHARED_COMMITMENTS_PREFIX, DA_SHARE_PREFIX, -}; use nomos_storage::{ + api::backend::rocksdb::{ + da::{DA_SHARED_COMMITMENTS_PREFIX, DA_SHARE_PREFIX}, + utils::{create_share_idx, key_bytes}, + }, backends::{rocksdb::RocksBackend, StorageSerde}, StorageMsg, StorageService, }; diff --git a/nomos-services/cryptarchia-consensus/Cargo.toml b/nomos-services/cryptarchia-consensus/Cargo.toml index 45ef4e3fd..5d5558209 100644 --- a/nomos-services/cryptarchia-consensus/Cargo.toml +++ b/nomos-services/cryptarchia-consensus/Cargo.toml @@ -8,7 +8,6 @@ version = "0.1.0" [dependencies] async-trait = "0.1" -bytes = "1" cl = { workspace = true } cryptarchia-engine = { workspace = true, features = ["serde"] } futures = "0.3" diff --git a/nomos-services/cryptarchia-consensus/src/lib.rs b/nomos-services/cryptarchia-consensus/src/lib.rs index c392019ba..5efe31bd0 100644 --- a/nomos-services/cryptarchia-consensus/src/lib.rs +++ b/nomos-services/cryptarchia-consensus/src/lib.rs @@ -9,7 +9,6 @@ pub mod storage; use core::fmt::Debug; use std::{collections::BTreeSet, fmt::Display, hash::Hash, path::PathBuf}; -use bytes::Bytes; use cryptarchia_engine::Slot; use futures::StreamExt as _; pub use leadership::LeaderConfig; @@ -31,7 +30,7 @@ use nomos_mempool::{ MempoolMsg, TxMempoolService, }; use nomos_network::NetworkService; -use nomos_storage::{backends::StorageBackend, StorageMsg, StorageService}; +use nomos_storage::{api::chain::StorageChainApi, backends::StorageBackend, StorageService}; use nomos_time::{SlotTick, TimeService, TimeServiceMessage}; use overwatch::{ services::{relay::OutboundRelay, AsServiceId, ServiceCore, ServiceData}, @@ -468,6 +467,8 @@ where BS: BlobSelect + Clone + Send + Sync + 'static, BS::Settings: Send + Sync + 'static, Storage: StorageBackend + Send + Sync + 'static, + ::Block: + TryFrom> + TryInto>, SamplingBackend: DaSamplingServiceBackend + Send, SamplingBackend::Settings: Clone, SamplingBackend::Share: Debug + Send + 'static, @@ -756,6 +757,8 @@ where BS: BlobSelect + Clone + Send + Sync + 'static, BS::Settings: Send, Storage: StorageBackend + Send + Sync + 'static, + ::Block: + TryFrom> + TryInto>, SamplingBackend: DaSamplingServiceBackend + Send, SamplingBackend::Settings: Clone, SamplingBackend::Share: Debug + 'static, @@ -950,12 +953,12 @@ where ) .await; - // store block - let key: [u8; 32] = header.id().into(); - let msg = - >::new_store_message(Bytes::copy_from_slice(&key), block.clone()); - if let Err((e, _msg)) = relays.storage_adapter().storage_relay.send(msg).await { - tracing::error!("Could not send block to storage: {e}"); + if let Err(e) = relays + .storage_adapter() + .store_block(header.id(), block.clone()) + .await + { + error!("Could not store block {e}"); } if let Err(e) = block_broadcaster.send(block) { diff --git a/nomos-services/cryptarchia-consensus/src/relays.rs b/nomos-services/cryptarchia-consensus/src/relays.rs index e172e4d26..640c56b48 100644 --- a/nomos-services/cryptarchia-consensus/src/relays.rs +++ b/nomos-services/cryptarchia-consensus/src/relays.rs @@ -8,6 +8,7 @@ use nomos_blend_service::{ network::NetworkAdapter as BlendNetworkAdapter, BlendService, ServiceMessage, }; use nomos_core::{ + block::Block, da::blob::{info::DispersedBlobInfo, BlobSelect}, header::HeaderId, tx::TxSelect, @@ -19,7 +20,9 @@ use nomos_mempool::{ DaMempoolService, TxMempoolService, }; use nomos_network::{NetworkMsg, NetworkService}; -use nomos_storage::{backends::StorageBackend, StorageMsg, StorageService}; +use nomos_storage::{ + api::chain::StorageChainApi, backends::StorageBackend, StorageMsg, StorageService, +}; use nomos_time::{backends::TimeBackend as TimeBackendTrait, TimeService, TimeServiceMessage}; use overwatch::{ services::{relay::OutboundRelay, AsServiceId}, @@ -74,7 +77,7 @@ pub struct CryptarchiaConsensusRelays< DaPool: MemPool, DaPoolAdapter: MempoolAdapter, NetworkAdapter: network::NetworkAdapter, - Storage: StorageBackend + Send + Sync, + Storage: StorageBackend + Send + Sync + 'static, SamplingRng: SeedableRng + RngCore, SamplingBackend: DaSamplingServiceBackend, TxS: TxSelect, @@ -136,13 +139,13 @@ where BS::Settings: Send, ClPool: RecoverableMempool, ClPool::RecoveryState: Serialize + for<'de> Deserialize<'de>, - ClPool::Item: Debug + DeserializeOwned + Eq + Hash + Clone + Send + Sync + 'static, + ClPool::Item: Debug + Serialize + DeserializeOwned + Eq + Hash + Clone + Send + Sync + 'static, ClPool::Key: Debug + 'static, ClPool::Settings: Clone, ClPoolAdapter: MempoolAdapter, DaPool: RecoverableMempool, DaPool::BlockId: Debug, - DaPool::Item: Debug + DeserializeOwned + Eq + Hash + Clone + Send + Sync + 'static, + DaPool::Item: Debug + Serialize + DeserializeOwned + Eq + Hash + Clone + Send + Sync + 'static, DaPool::Key: Debug + 'static, DaPool::RecoveryState: Serialize + for<'de> Deserialize<'de>, DaPool::Settings: Clone, @@ -155,6 +158,8 @@ where SamplingBackend::Share: Debug + 'static, SamplingRng: SeedableRng + RngCore, Storage: StorageBackend + Send + Sync + 'static, + ::Block: + TryFrom> + TryInto>, TxS: TxSelect, TxS::Settings: Send, DaVerifierBackend: nomos_da_verifier::backend::VerifierBackend + Send + Sync + 'static, diff --git a/nomos-services/cryptarchia-consensus/src/storage/adapters/storage.rs b/nomos-services/cryptarchia-consensus/src/storage/adapters/storage.rs index a0f72b888..97366807b 100644 --- a/nomos-services/cryptarchia-consensus/src/storage/adapters/storage.rs +++ b/nomos-services/cryptarchia-consensus/src/storage/adapters/storage.rs @@ -1,16 +1,18 @@ use std::{hash::Hash, marker::PhantomData}; -use bytes::Bytes; use nomos_core::{block::Block, header::HeaderId}; -use nomos_storage::{backends::StorageBackend, StorageMsg, StorageService}; +use nomos_storage::{ + api::chain::StorageChainApi, backends::StorageBackend, StorageMsg, StorageService, +}; use overwatch::services::{relay::OutboundRelay, ServiceData}; -use serde::de::DeserializeOwned; +use serde::{de::DeserializeOwned, Serialize}; +use tokio::sync::oneshot; use crate::storage::StorageAdapter as StorageAdapterTrait; pub struct StorageAdapter where - Storage: StorageBackend + Send + Sync, + Storage: StorageBackend + Send + Sync + 'static, { pub storage_relay: OutboundRelay< as ServiceData>::Message>, @@ -18,40 +20,15 @@ where _blob_certificate: PhantomData, } -impl - StorageAdapter -where - Storage: StorageBackend + Send + Sync, - Tx: Sync, - BlobCertificate: Sync, -{ - /// Sends a store message to the storage service to retrieve a value by its - /// key - /// - /// # Arguments - /// - /// * `key` - The key to retrieve the value for - /// - /// # Returns - /// - /// The value for the given key. If no value is found, returns None. - pub async fn get_value(&self, key: Bytes) -> Option - where - Value: DeserializeOwned, - { - let (msg, receiver) = >::new_load_message(key); - self.storage_relay.send(msg).await.unwrap(); - receiver.recv().await.unwrap() - } -} - #[async_trait::async_trait] impl StorageAdapterTrait for StorageAdapter where - Storage: StorageBackend + Send + Sync, - Tx: Clone + Eq + Hash + DeserializeOwned + Send + Sync, - BlobCertificate: Clone + Eq + Hash + DeserializeOwned + Send + Sync, + Storage: StorageBackend + Send + Sync + 'static, + ::Block: + TryFrom> + TryInto>, + Tx: Clone + Eq + Hash + Serialize + DeserializeOwned + Send + Sync + 'static, + BlobCertificate: Clone + Eq + Hash + Serialize + DeserializeOwned + Send + Sync + 'static, { type Backend = Storage; type Block = Block; @@ -68,8 +45,39 @@ where } } - async fn get_block(&self, key: &HeaderId) -> Option { - let key: [u8; 32] = (*key).into(); - self.get_value(Bytes::copy_from_slice(&key)).await + async fn get_block(&self, header_id: &HeaderId) -> Option { + let (sender, receiver) = oneshot::channel(); + + self.storage_relay + .send(StorageMsg::get_block_request(*header_id, sender)) + .await + .unwrap(); + + if let Ok(maybe_block) = receiver.await { + let block = maybe_block?; + block.try_into().ok() + } else { + tracing::error!("Failed to receive block from storage relay"); + return None; + }; + + None + } + + async fn store_block( + &self, + header_id: HeaderId, + block: Self::Block, + ) -> Result<(), overwatch::DynError> { + let block = block + .try_into() + .map_err(|_| "Failed to convert block to storage format")?; + + self.storage_relay + .send(StorageMsg::store_block_request(header_id, block)) + .await + .map_err(|_| "Failed to send store block request to storage relay")?; + + Ok(()) } } diff --git a/nomos-services/cryptarchia-consensus/src/storage/mod.rs b/nomos-services/cryptarchia-consensus/src/storage/mod.rs index 7e59d8b1d..874283c34 100644 --- a/nomos-services/cryptarchia-consensus/src/storage/mod.rs +++ b/nomos-services/cryptarchia-consensus/src/storage/mod.rs @@ -6,7 +6,7 @@ use overwatch::services::{relay::OutboundRelay, ServiceData}; #[async_trait::async_trait] pub trait StorageAdapter { - type Backend: StorageBackend + Send + Sync; + type Backend: StorageBackend + Send + Sync + 'static; type Block; async fn new( @@ -22,4 +22,10 @@ pub trait StorageAdapter { /// /// The block with the given header id. If no block is found, returns None. async fn get_block(&self, key: &HeaderId) -> Option; + + async fn store_block( + &self, + header_id: HeaderId, + block: Self::Block, + ) -> Result<(), overwatch::DynError>; } diff --git a/nomos-services/data-availability/indexer/Cargo.toml b/nomos-services/data-availability/indexer/Cargo.toml index 1b400a8b4..53c37d26e 100644 --- a/nomos-services/data-availability/indexer/Cargo.toml +++ b/nomos-services/data-availability/indexer/Cargo.toml @@ -14,7 +14,6 @@ futures = "0.3" kzgrs-backend = { workspace = true } nomos-core = { workspace = true } nomos-da-sampling = { workspace = true } -nomos-da-storage = { workspace = true } nomos-da-verifier = { workspace = true } nomos-mempool = { workspace = true } nomos-storage = { workspace = true } diff --git a/nomos-services/data-availability/indexer/src/storage/adapters/rocksdb.rs b/nomos-services/data-availability/indexer/src/storage/adapters/rocksdb.rs index 4cc47d3db..5b9fd4717 100644 --- a/nomos-services/data-availability/indexer/src/storage/adapters/rocksdb.rs +++ b/nomos-services/data-availability/indexer/src/storage/adapters/rocksdb.rs @@ -11,10 +11,11 @@ use nomos_core::da::{ }, BlobId, }; -use nomos_da_storage::rocksdb::{ - key_bytes, DA_SHARED_COMMITMENTS_PREFIX, DA_SHARE_PREFIX, DA_VID_KEY_PREFIX, -}; use nomos_storage::{ + api::backend::rocksdb::{ + da::{DA_SHARED_COMMITMENTS_PREFIX, DA_SHARE_PREFIX, DA_VID_KEY_PREFIX}, + utils::key_bytes, + }, backends::{rocksdb::RocksBackend, StorageSerde}, StorageMsg, StorageService, }; diff --git a/nomos-services/data-availability/sampling/Cargo.toml b/nomos-services/data-availability/sampling/Cargo.toml index 2845e9aff..2fb11188a 100644 --- a/nomos-services/data-availability/sampling/Cargo.toml +++ b/nomos-services/data-availability/sampling/Cargo.toml @@ -15,7 +15,6 @@ multiaddr = "0.18" nomos-core = { workspace = true } nomos-da-network-core = { workspace = true } nomos-da-network-service = { workspace = true } -nomos-da-storage = { workspace = true } nomos-da-verifier = { workspace = true } nomos-storage = { workspace = true } nomos-tracing = { workspace = true } diff --git a/nomos-services/data-availability/sampling/src/storage/adapters/rocksdb.rs b/nomos-services/data-availability/sampling/src/storage/adapters/rocksdb.rs index 30db7f1b0..7def7b1d4 100644 --- a/nomos-services/data-availability/sampling/src/storage/adapters/rocksdb.rs +++ b/nomos-services/data-availability/sampling/src/storage/adapters/rocksdb.rs @@ -2,10 +2,8 @@ use std::{marker::PhantomData, path::PathBuf}; use kzgrs_backend::common::ShareIndex; use nomos_core::da::blob::Share; -use nomos_da_storage::rocksdb::{ - create_share_idx, key_bytes, DA_SHARED_COMMITMENTS_PREFIX, DA_SHARE_PREFIX, -}; use nomos_storage::{ + api::backend::rocksdb::{da::DA_SHARED_COMMITMENTS_PREFIX, utils::key_bytes}, backends::{rocksdb::RocksBackend, StorageSerde}, StorageMsg, StorageService, }; @@ -75,21 +73,26 @@ where blob_id: ::BlobId, share_idx: ShareIndex, ) -> Result::LightShare>, DynError> { - let share_idx = create_share_idx(blob_id.as_ref(), share_idx.to_be_bytes().as_ref()); - let share_key = key_bytes(DA_SHARE_PREFIX, share_idx); - let (share_reply_tx, share_reply_rx) = tokio::sync::oneshot::channel(); + let blob_id = blob_id.as_ref().try_into().unwrap(); + let share_idx = share_idx.to_be_bytes(); + + let (reply_channel, reply_rx) = tokio::sync::oneshot::channel(); self.storage_relay - .send(StorageMsg::Load { - key: share_key, - reply_channel: share_reply_tx, - }) + .send(StorageMsg::get_light_share_request( + blob_id, + share_idx, + reply_channel, + )) .await - .expect("Failed to send load request to storage relay"); + .expect("Failed to send request to storage relay"); - let share = share_reply_rx.await?; - let share = share.map(|share| S::deserialize(share).expect("Failed to deserialize blob")); - - Ok(share) + reply_rx + .await + .map(|maybe_share| { + maybe_share + .map(|share| S::deserialize(share).expect("Failed to deserialize light share")) + }) + .map_err(|_| "Failed to receive response from storage".into()) } } diff --git a/nomos-services/data-availability/verifier/Cargo.toml b/nomos-services/data-availability/verifier/Cargo.toml index 8922f1186..0faac2162 100644 --- a/nomos-services/data-availability/verifier/Cargo.toml +++ b/nomos-services/data-availability/verifier/Cargo.toml @@ -12,7 +12,6 @@ libp2p = { workspace = true, features = ["ed25519"] } nomos-core = { workspace = true } nomos-da-network-core = { workspace = true } nomos-da-network-service = { workspace = true } -nomos-da-storage = { workspace = true } nomos-storage = { workspace = true } nomos-tracing = { workspace = true } overwatch = { workspace = true } diff --git a/nomos-services/data-availability/verifier/src/storage/adapters/rocksdb.rs b/nomos-services/data-availability/verifier/src/storage/adapters/rocksdb.rs index eb8bb9667..0372fcfc1 100644 --- a/nomos-services/data-availability/verifier/src/storage/adapters/rocksdb.rs +++ b/nomos-services/data-availability/verifier/src/storage/adapters/rocksdb.rs @@ -1,11 +1,7 @@ -use std::{collections::HashSet, fmt::Debug, hash::Hash, marker::PhantomData, path::PathBuf}; +use std::{fmt::Debug, hash::Hash, marker::PhantomData, path::PathBuf}; use futures::try_join; use nomos_core::da::blob::Share; -use nomos_da_storage::rocksdb::{ - create_share_idx, key_bytes, DA_BLOB_SHARES_INDEX_PREFIX, DA_SHARED_COMMITMENTS_PREFIX, - DA_SHARE_PREFIX, -}; use nomos_storage::{ backends::{rocksdb::RocksBackend, StorageSerde}, StorageMsg, StorageService, @@ -15,7 +11,6 @@ use overwatch::{ DynError, }; use serde::{de::DeserializeOwned, Deserialize, Serialize}; -use tokio::sync::oneshot; use crate::storage::DaStorageAdapter; @@ -59,100 +54,51 @@ where shared_commitments: ::SharesCommitments, light_share: ::LightShare, ) -> Result<(), DynError> { - let blob_share_idx = create_share_idx(blob_id.as_ref(), share_idx.as_ref()); - let share_key = key_bytes(DA_SHARE_PREFIX, blob_share_idx); - let shared_commitments_key = key_bytes(DA_SHARED_COMMITMENTS_PREFIX, &blob_id); - let index_key = key_bytes(DA_BLOB_SHARES_INDEX_PREFIX, &blob_id); + let blob_id: [u8; 32] = blob_id.as_ref().try_into().unwrap(); + let share_idx: [u8; 2] = share_idx.as_ref().try_into().unwrap(); - let mut shares_index = self - .try_load_shares_index(&blob_id) - .await? - .unwrap_or_default(); - shares_index.insert(share_idx); + let store_share_msg = + StorageMsg::store_light_share_request(blob_id, share_idx, S::serialize(light_share)); + + let store_commitments_msg = StorageMsg::store_shared_commitments_request( + blob_id, + S::serialize(shared_commitments.clone()), + ); try_join!( - self.storage_relay.send(StorageMsg::Store { - key: share_key, - value: S::serialize(light_share), - }), - self.storage_relay.send(StorageMsg::Store { - key: shared_commitments_key, - value: S::serialize(shared_commitments), - }), - self.storage_relay.send(StorageMsg::Store { - key: index_key, - value: S::serialize(shares_index), - }), + self.storage_relay.send(store_share_msg), + self.storage_relay.send(store_commitments_msg), ) .map_err(|(e, _)| DynError::from(e))?; Ok(()) } + async fn get_share( &self, blob_id: ::BlobId, share_idx: ::ShareIndex, ) -> Result::LightShare>, DynError> { - let share_idx = create_share_idx(blob_id.as_ref(), share_idx.as_ref()); - let share_key = key_bytes(DA_SHARE_PREFIX, share_idx); + let blob_id: [u8; 32] = blob_id.as_ref().try_into().unwrap(); + let share_idx: [u8; 2] = share_idx.as_ref().try_into().unwrap(); + let (reply_channel, reply_rx) = tokio::sync::oneshot::channel(); self.storage_relay - .send(StorageMsg::Load { - key: share_key, + .send(StorageMsg::get_light_share_request( + blob_id, + share_idx, reply_channel, - }) + )) .await - .expect("Failed to send load request to storage relay"); + .expect("Failed to send request to storage relay"); - // TODO: Use storage backend ser/de functionality. - // - // Storage backend already handles ser/de, but lacks the ability to seperate - // storage domains using prefixed keys. Once implemented Indexer and - // Verifier can be simplified. reply_rx .await - .map(|maybe_bytes| { - maybe_bytes.map(|bytes| { - S::deserialize(bytes).expect("Blob should be deserialized from bytes") - }) + .map(|maybe_share| { + maybe_share + .map(|share| S::deserialize(share).expect("Failed to deserialize light share")) }) - .map_err(|_| "".into()) - } -} - -impl RocksAdapter -where - B: Share + Clone + Send + Sync + 'static, - B::BlobId: AsRef<[u8]> + Send + Sync + 'static, - B::ShareIndex: AsRef<[u8]> + DeserializeOwned + Eq + Hash + Send + Sync + 'static, - B::LightShare: Clone + DeserializeOwned + Send + Serialize + Sync + 'static, - B::SharesCommitments: Clone + DeserializeOwned + Send + Serialize + Sync + 'static, - S: Send + StorageSerde + Sync + 'static, -{ - async fn try_load_shares_index( - &self, - blob_id: &::BlobId, - ) -> Result>, DynError> { - // Load the blob index. This should be most of the time only read from memory - // because all shares are added around the same time. So we probably - // don't need to keep additional state in the adapter to avoid reads per - // each share - let index_key = key_bytes(DA_BLOB_SHARES_INDEX_PREFIX, blob_id); - let (reply_tx, reply_rx) = oneshot::channel(); - self.storage_relay - .send(StorageMsg::Load { - key: index_key, - reply_channel: reply_tx, - }) - .await - .map_err(|(e, _)| e)?; - - let shares = reply_rx.await?.map_or_else( - || None, - |bytes| Some(S::deserialize(bytes).expect("Failed to deserialize shares")), - ); - - Ok(shares) + .map_err(|_| "Failed to receive response from storage".into()) } } diff --git a/nomos-services/storage/Cargo.toml b/nomos-services/storage/Cargo.toml index cf4adafd8..00062aa10 100644 --- a/nomos-services/storage/Cargo.toml +++ b/nomos-services/storage/Cargo.toml @@ -8,13 +8,13 @@ version = "0.1.0" [dependencies] async-trait = "0.1" -bytes = "1.2" +bytes = "1.3" futures = "0.3" +nomos-core = { workspace = true } overwatch = { workspace = true } rocksdb = { version = "0.22", optional = true } serde = { version = "1.0", features = ["derive"] } services-utils = { workspace = true } -sled = { version = "0.34", optional = true } thiserror = "1.0" tokio = { version = "1", features = ["macros", "sync"] } tracing = "0.1" @@ -27,7 +27,6 @@ tokio = { version = "1", features = ["macros", "sync", "time"] } default = [] mock = [] rocksdb-backend = ["dep:rocksdb"] -sled-backend = ["dep:sled"] [[bin]] name = "rocks" diff --git a/nomos-services/storage/src/api/backend/mod.rs b/nomos-services/storage/src/api/backend/mod.rs new file mode 100644 index 000000000..663a2ff35 --- /dev/null +++ b/nomos-services/storage/src/api/backend/mod.rs @@ -0,0 +1,2 @@ +#[cfg(feature = "rocksdb-backend")] +pub mod rocksdb; diff --git a/nomos-services/storage/src/api/backend/rocksdb/blocks.rs b/nomos-services/storage/src/api/backend/rocksdb/blocks.rs new file mode 100644 index 000000000..ebe858693 --- /dev/null +++ b/nomos-services/storage/src/api/backend/rocksdb/blocks.rs @@ -0,0 +1,30 @@ +use async_trait::async_trait; +use bytes::Bytes; +use nomos_core::header::HeaderId; +use rocksdb::Error; + +use crate::{ + api::chain::StorageChainApi, + backends::{rocksdb::RocksBackend, StorageBackend as _, StorageSerde}, +}; + +#[async_trait] +impl StorageChainApi for RocksBackend { + type Error = Error; + type Block = Bytes; + async fn get_block(&mut self, header_id: HeaderId) -> Result, Self::Error> { + let header_id: [u8; 32] = header_id.into(); + let key = Bytes::copy_from_slice(&header_id); + self.load(&key).await + } + + async fn store_block( + &mut self, + header_id: HeaderId, + block: Self::Block, + ) -> Result<(), Self::Error> { + let header_id: [u8; 32] = header_id.into(); + let key = Bytes::copy_from_slice(&header_id); + self.store(key, block).await + } +} diff --git a/nomos-services/storage/src/api/backend/rocksdb/da.rs b/nomos-services/storage/src/api/backend/rocksdb/da.rs new file mode 100644 index 000000000..25c5d834f --- /dev/null +++ b/nomos-services/storage/src/api/backend/rocksdb/da.rs @@ -0,0 +1,95 @@ +use std::collections::HashSet; + +use async_trait::async_trait; +use bytes::Bytes; +use rocksdb::Error; +use tracing::{debug, error}; + +use crate::{ + api::{ + backend::rocksdb::utils::{create_share_idx, key_bytes}, + da::StorageDaApi, + }, + backends::{rocksdb::RocksBackend, StorageBackend as _, StorageSerde}, +}; + +pub const DA_VID_KEY_PREFIX: &str = "da/vid/"; +pub const DA_BLOB_SHARES_INDEX_PREFIX: &str = concat!("da/verified/", "si"); +pub const DA_SHARED_COMMITMENTS_PREFIX: &str = concat!("da/verified/", "sc"); +pub const DA_SHARE_PREFIX: &str = concat!("da/verified/", "bl"); + +#[async_trait] +impl StorageDaApi for RocksBackend { + type Error = Error; + type BlobId = [u8; 32]; + type Share = Bytes; + type Commitments = Bytes; + type ShareIndex = [u8; 2]; + + async fn get_light_share( + &mut self, + blob_id: Self::BlobId, + share_idx: Self::ShareIndex, + ) -> Result, Self::Error> { + let share_idx_bytes = create_share_idx(blob_id.as_ref(), share_idx.as_ref()); + let share_key = key_bytes(DA_SHARE_PREFIX, share_idx_bytes); + let share_bytes = self.load(&share_key).await?; + Ok(share_bytes) + } + + async fn store_light_share( + &mut self, + blob_id: Self::BlobId, + share_idx: Self::ShareIndex, + light_share: Self::Share, + ) -> Result<(), Self::Error> { + let share_idx_bytes = create_share_idx(blob_id.as_ref(), share_idx.as_ref()); + let share_key = key_bytes(DA_SHARE_PREFIX, share_idx_bytes); + let index_key = key_bytes(DA_BLOB_SHARES_INDEX_PREFIX, blob_id.as_ref()); + + let txn = self.txn(move |db| { + if let Err(e) = db.put(&share_key, &light_share) { + error!("Failed to store share data: {:?}", e); + return Err(e); + } + + let mut indices = db.get(&index_key)?.map_or_else(HashSet::new, |bytes| { + SerdeOp::deserialize::>(bytes.into()).unwrap_or_else(|e| { + error!("Failed to deserialize indices: {:?}", e); + HashSet::new() + }) + }); + + indices.insert(share_idx); + + let serialized_indices = SerdeOp::serialize(indices); + + if let Err(e) = db.put(&index_key, &serialized_indices) { + error!("Failed to store indices: {:?}", e); + return Err(e); + } + + Ok(None) + }); + + match self.execute(txn).await { + Ok(_) => { + debug!("Successfully stored light share and updated indices"); + Ok(()) + } + Err(e) => { + error!("Failed to execute transaction: {:?}", e); + Err(e) + } + } + } + + async fn store_shared_commitments( + &mut self, + blob_id: Self::BlobId, + shared_commitments: Self::Commitments, + ) -> Result<(), Self::Error> { + let commitments_key = key_bytes(DA_SHARED_COMMITMENTS_PREFIX, blob_id.as_ref()); + self.store(commitments_key, shared_commitments).await + } +} diff --git a/nomos-services/storage/src/api/backend/rocksdb/mod.rs b/nomos-services/storage/src/api/backend/rocksdb/mod.rs new file mode 100644 index 000000000..b0fab40c3 --- /dev/null +++ b/nomos-services/storage/src/api/backend/rocksdb/mod.rs @@ -0,0 +1,10 @@ +use crate::{ + api::StorageBackendApi, + backends::{rocksdb::RocksBackend, StorageSerde}, +}; + +pub mod blocks; +pub mod da; +pub mod utils; + +impl StorageBackendApi for RocksBackend {} diff --git a/nomos-da/storage/src/rocksdb/mod.rs b/nomos-services/storage/src/api/backend/rocksdb/utils.rs similarity index 66% rename from nomos-da/storage/src/rocksdb/mod.rs rename to nomos-services/storage/src/api/backend/rocksdb/utils.rs index 7344d9404..c76a559b1 100644 --- a/nomos-da/storage/src/rocksdb/mod.rs +++ b/nomos-services/storage/src/api/backend/rocksdb/utils.rs @@ -1,13 +1,6 @@ use bytes::{Bytes, BytesMut}; -// "DA/VID/" PREFIX -pub const DA_VID_KEY_PREFIX: &str = "da/vid/"; - -// "DA/VERIFIED/" PREFIX -pub const DA_BLOB_SHARES_INDEX_PREFIX: &str = concat!("da/verified/", "si"); -pub const DA_SHARED_COMMITMENTS_PREFIX: &str = concat!("da/verified/", "sc"); -pub const DA_SHARE_PREFIX: &str = concat!("da/verified/", "bl"); - +#[must_use] pub fn key_bytes(prefix: &str, id: impl AsRef<[u8]>) -> Bytes { let mut buffer = BytesMut::new(); diff --git a/nomos-services/storage/src/api/chain/mod.rs b/nomos-services/storage/src/api/chain/mod.rs new file mode 100644 index 000000000..44925bb05 --- /dev/null +++ b/nomos-services/storage/src/api/chain/mod.rs @@ -0,0 +1,20 @@ +pub mod requests; + +use std::error::Error; + +use async_trait::async_trait; +use nomos_core::header::HeaderId; + +#[async_trait] +pub trait StorageChainApi { + type Error: Error + Send + Sync + 'static; + type Block: Send + Sync; + + async fn get_block(&mut self, header_id: HeaderId) -> Result, Self::Error>; + + async fn store_block( + &mut self, + header_id: HeaderId, + block: Self::Block, + ) -> Result<(), Self::Error>; +} diff --git a/nomos-services/storage/src/api/chain/requests.rs b/nomos-services/storage/src/api/chain/requests.rs new file mode 100644 index 000000000..a06f867cb --- /dev/null +++ b/nomos-services/storage/src/api/chain/requests.rs @@ -0,0 +1,92 @@ +use nomos_core::header::HeaderId; +use tokio::sync::oneshot::Sender; + +use crate::{ + api::{chain::StorageChainApi, StorageApiRequest, StorageBackendApi, StorageOperation}, + backends::StorageBackend, + StorageMsg, StorageServiceError, +}; + +pub enum ChainApiRequest { + GetBlock { + header_id: HeaderId, + response_tx: Sender::Block>>, + }, + StoreBlock { + header_id: HeaderId, + block: ::Block, + }, +} + +impl StorageOperation for ChainApiRequest +where + B: StorageBackend + StorageBackendApi, +{ + async fn execute(self, backend: &mut B) -> Result<(), StorageServiceError> { + match self { + Self::GetBlock { + header_id, + response_tx, + } => handle_get_block(backend, header_id, response_tx).await, + Self::StoreBlock { header_id, block } => { + handle_store_block(backend, header_id, block).await + } + } + } +} + +async fn handle_get_block( + backend: &mut B, + header_id: HeaderId, + response_tx: Sender>, +) -> Result<(), StorageServiceError> { + let result = backend + .get_block(header_id) + .await + .map_err(|e| StorageServiceError::BackendError(e.into()))?; + + if response_tx.send(result).is_err() { + return Err(StorageServiceError::ReplyError { + message: format!( + "Failed to send reply for get block request by header_id: {header_id}" + ), + }); + } + + Ok(()) +} + +async fn handle_store_block( + backend: &mut B, + header_id: HeaderId, + block: B::Block, +) -> Result<(), StorageServiceError> { + backend + .store_block(header_id, block) + .await + .map_err(|e| StorageServiceError::BackendError(e.into())) +} + +impl StorageMsg { + #[must_use] + pub const fn get_block_request( + header_id: HeaderId, + response_tx: Sender::Block>>, + ) -> Self { + Self::Api { + request: StorageApiRequest::Chain(ChainApiRequest::GetBlock { + header_id, + response_tx, + }), + } + } + + pub const fn store_block_request( + header_id: HeaderId, + block: ::Block, + ) -> Self { + Self::Api { + request: StorageApiRequest::Chain(ChainApiRequest::StoreBlock { header_id, block }), + } + } +} diff --git a/nomos-services/storage/src/api/da/mod.rs b/nomos-services/storage/src/api/da/mod.rs new file mode 100644 index 000000000..60434324b --- /dev/null +++ b/nomos-services/storage/src/api/da/mod.rs @@ -0,0 +1,33 @@ +use std::error::Error; + +use async_trait::async_trait; + +pub mod requests; + +#[async_trait] +pub trait StorageDaApi { + type Error: Error + Send + Sync + 'static; + type BlobId: Send + Sync; + type Share: Send + Sync; + type Commitments: Send + Sync; + type ShareIndex: Send + Sync; + + async fn get_light_share( + &mut self, + blob_id: Self::BlobId, + share_idx: Self::ShareIndex, + ) -> Result, Self::Error>; + + async fn store_light_share( + &mut self, + blob_id: Self::BlobId, + share_idx: Self::ShareIndex, + light_share: Self::Share, + ) -> Result<(), Self::Error>; + + async fn store_shared_commitments( + &mut self, + blob_id: Self::BlobId, + shared_commitments: Self::Commitments, + ) -> Result<(), Self::Error>; +} diff --git a/nomos-services/storage/src/api/da/requests.rs b/nomos-services/storage/src/api/da/requests.rs new file mode 100644 index 000000000..7271e5724 --- /dev/null +++ b/nomos-services/storage/src/api/da/requests.rs @@ -0,0 +1,136 @@ +use tokio::sync::oneshot::Sender; + +use crate::{ + api::{da::StorageDaApi, StorageApiRequest, StorageBackendApi, StorageOperation}, + backends::StorageBackend, + StorageMsg, StorageServiceError, +}; + +pub enum DaApiRequest { + GetLightShare { + blob_id: ::BlobId, + share_idx: ::ShareIndex, + response_tx: Sender::Share>>, + }, + StoreLightShare { + blob_id: ::BlobId, + share_idx: ::ShareIndex, + light_share: ::Share, + }, + StoreSharedCommitments { + blob_id: ::BlobId, + shared_commitments: ::Commitments, + }, +} + +impl StorageOperation for DaApiRequest +where + B: StorageBackend + StorageBackendApi, +{ + async fn execute(self, backend: &mut B) -> Result<(), StorageServiceError> { + match self { + Self::GetLightShare { + blob_id, + share_idx, + response_tx, + } => handle_get_light_share(backend, blob_id, share_idx, response_tx).await, + Self::StoreLightShare { + blob_id, + share_idx, + light_share, + } => handle_store_light_share(backend, blob_id, share_idx, light_share).await, + Self::StoreSharedCommitments { + blob_id, + shared_commitments, + } => handle_store_shared_commitments(backend, blob_id, shared_commitments).await, + } + } +} + +async fn handle_get_light_share( + backend: &mut B, + blob_id: B::BlobId, + share_idx: B::ShareIndex, + response_tx: Sender>, +) -> Result<(), StorageServiceError> { + let result = backend + .get_light_share(blob_id, share_idx) + .await + .map_err(|e| StorageServiceError::BackendError(e.into()))?; + + if response_tx.send(result).is_err() { + return Err(StorageServiceError::ReplyError { + message: "Failed to send reply for get light share request".to_owned(), + }); + } + + Ok(()) +} + +async fn handle_store_light_share( + backend: &mut B, + blob_id: B::BlobId, + share_idx: B::ShareIndex, + light_share: B::Share, +) -> Result<(), StorageServiceError> { + backend + .store_light_share(blob_id, share_idx, light_share) + .await + .map_err(|e| StorageServiceError::BackendError(e.into())) +} + +async fn handle_store_shared_commitments( + backend: &mut B, + blob_id: B::BlobId, + shared_commitments: B::Commitments, +) -> Result<(), StorageServiceError> { + backend + .store_shared_commitments(blob_id, shared_commitments) + .await + .map_err(|e| StorageServiceError::BackendError(e.into())) +} + +impl StorageMsg { + #[must_use] + pub const fn get_light_share_request( + blob_id: ::BlobId, + share_idx: ::ShareIndex, + response_tx: Sender::Share>>, + ) -> Self { + Self::Api { + request: StorageApiRequest::Da(DaApiRequest::GetLightShare { + blob_id, + share_idx, + response_tx, + }), + } + } + + #[must_use] + pub const fn store_light_share_request( + blob_id: ::BlobId, + share_idx: ::ShareIndex, + light_share: ::Share, + ) -> Self { + Self::Api { + request: StorageApiRequest::Da(DaApiRequest::StoreLightShare { + blob_id, + share_idx, + light_share, + }), + } + } + + #[must_use] + pub const fn store_shared_commitments_request( + blob_id: ::BlobId, + shared_commitments: ::Commitments, + ) -> Self { + Self::Api { + request: StorageApiRequest::Da(DaApiRequest::StoreSharedCommitments { + blob_id, + shared_commitments, + }), + } + } +} diff --git a/nomos-services/storage/src/api/mod.rs b/nomos-services/storage/src/api/mod.rs new file mode 100644 index 000000000..a2d155465 --- /dev/null +++ b/nomos-services/storage/src/api/mod.rs @@ -0,0 +1,35 @@ +use async_trait::async_trait; + +use crate::{ + api::{ + chain::{requests::ChainApiRequest, StorageChainApi}, + da::{requests::DaApiRequest, StorageDaApi}, + }, + backends::StorageBackend, + StorageServiceError, +}; + +pub mod backend; +pub mod chain; +pub mod da; + +#[async_trait] +pub trait StorageBackendApi: StorageChainApi + StorageDaApi {} + +pub(crate) trait StorageOperation { + async fn execute(self, api: &mut B) -> Result<(), StorageServiceError>; +} + +pub enum StorageApiRequest { + Chain(ChainApiRequest), + Da(DaApiRequest), +} + +impl StorageOperation for StorageApiRequest { + async fn execute(self, backend: &mut B) -> Result<(), StorageServiceError> { + match self { + Self::Chain(request) => request.execute(backend).await, + Self::Da(request) => request.execute(backend).await, + } + } +} diff --git a/nomos-services/storage/src/backends/mock.rs b/nomos-services/storage/src/backends/mock.rs index 28b175a4e..e6a0b841a 100644 --- a/nomos-services/storage/src/backends/mock.rs +++ b/nomos-services/storage/src/backends/mock.rs @@ -2,9 +2,11 @@ use std::{collections::HashMap, marker::PhantomData}; use async_trait::async_trait; use bytes::Bytes; +use nomos_core::header::HeaderId; use thiserror::Error; use super::{StorageBackend, StorageSerde, StorageTransaction}; +use crate::api::{chain::StorageChainApi, da::StorageDaApi, StorageBackendApi}; #[derive(Debug, Error)] #[error("Errors in MockStorage should not happen")] @@ -36,32 +38,103 @@ impl StorageBackend for MockStora type Transaction = MockStorageTransaction; type SerdeOperator = SerdeOp; - fn new(_config: Self::Settings) -> Result { + fn new(_config: Self::Settings) -> Result::Error> { Ok(Self { inner: HashMap::new(), _serde_op: PhantomData, }) } - async fn store(&mut self, key: Bytes, value: Bytes) -> Result<(), Self::Error> { + async fn store( + &mut self, + key: Bytes, + value: Bytes, + ) -> Result<(), ::Error> { let _ = self.inner.insert(key, value); Ok(()) } - async fn load(&mut self, key: &[u8]) -> Result, Self::Error> { + async fn load(&mut self, key: &[u8]) -> Result, ::Error> { Ok(self.inner.get(key).cloned()) } - async fn load_prefix(&mut self, _key: &[u8]) -> Result, Self::Error> { + async fn load_prefix( + &mut self, + _key: &[u8], + ) -> Result, ::Error> { unimplemented!() } - async fn remove(&mut self, key: &[u8]) -> Result, Self::Error> { + async fn remove( + &mut self, + key: &[u8], + ) -> Result, ::Error> { Ok(self.inner.remove(key)) } - async fn execute(&mut self, transaction: Self::Transaction) -> Result<(), Self::Error> { + async fn execute( + &mut self, + transaction: Self::Transaction, + ) -> Result<(), ::Error> { transaction(&mut self.inner); Ok(()) } } + +#[async_trait] +impl StorageChainApi for MockStorage { + type Error = MockStorageError; + type Block = Bytes; + + async fn get_block( + &mut self, + _header_id: HeaderId, + ) -> Result, Self::Error> { + unimplemented!() + } + + async fn store_block( + &mut self, + _header_id: HeaderId, + _block: Self::Block, + ) -> Result<(), Self::Error> { + unimplemented!() + } +} + +#[async_trait] +impl StorageDaApi for MockStorage { + type Error = MockStorageError; + type BlobId = [u8; 32]; + type Share = Bytes; + type Commitments = Bytes; + type ShareIndex = [u8; 2]; + + async fn get_light_share( + &mut self, + _blob_id: Self::BlobId, + _share_idx: Self::ShareIndex, + ) -> Result, Self::Error> { + unimplemented!() + } + + async fn store_light_share( + &mut self, + _blob_id: Self::BlobId, + _share_idx: Self::ShareIndex, + _light_share: Self::Share, + ) -> Result<(), Self::Error> { + unimplemented!() + } + + async fn store_shared_commitments( + &mut self, + _blob_id: Self::BlobId, + _shared_commitments: Self::Commitments, + ) -> Result<(), Self::Error> { + unimplemented!() + } +} + +#[async_trait] +impl StorageBackendApi for MockStorage {} diff --git a/nomos-services/storage/src/backends/mod.rs b/nomos-services/storage/src/backends/mod.rs index d48d7aa0b..e4e5ee0ed 100644 --- a/nomos-services/storage/src/backends/mod.rs +++ b/nomos-services/storage/src/backends/mod.rs @@ -1,7 +1,5 @@ #[cfg(feature = "mock")] pub mod mock; -#[cfg(feature = "sled-backend")] -pub mod sled; #[cfg(feature = "rocksdb-backend")] pub mod rocksdb; @@ -12,6 +10,8 @@ use async_trait::async_trait; use bytes::Bytes; use serde::{de::DeserializeOwned, Serialize}; +use crate::api::StorageBackendApi; + /// Trait that defines how to translate from user types to the storage buffer /// type pub trait StorageSerde { @@ -30,7 +30,7 @@ pub trait StorageTransaction: Send + Sync { /// Main storage functionality trait #[async_trait] -pub trait StorageBackend: Sized { +pub trait StorageBackend: StorageBackendApi + Sized { /// Backend settings type Settings: Clone + Send + Sync + 'static; /// Backend operations error type @@ -42,17 +42,27 @@ pub trait StorageBackend: Sized { /// Operator to dump/load custom types into the defined backend store type /// [`Bytes`] type SerdeOperator: StorageSerde + Send + Sync + 'static; - fn new(config: Self::Settings) -> Result; - async fn store(&mut self, key: Bytes, value: Bytes) -> Result<(), Self::Error>; - async fn load(&mut self, key: &[u8]) -> Result, Self::Error>; + fn new(config: Self::Settings) -> Result::Error>; + async fn store( + &mut self, + key: Bytes, + value: Bytes, + ) -> Result<(), ::Error>; + async fn load(&mut self, key: &[u8]) -> Result, ::Error>; /// Loads all values whose keys start with the given prefix. - async fn load_prefix(&mut self, prefix: &[u8]) -> Result, Self::Error>; - async fn remove(&mut self, key: &[u8]) -> Result, Self::Error>; + async fn load_prefix( + &mut self, + prefix: &[u8], + ) -> Result, ::Error>; + async fn remove( + &mut self, + key: &[u8], + ) -> Result, ::Error>; /// Execute a transaction in the current backend async fn execute( &mut self, transaction: Self::Transaction, - ) -> Result<::Result, Self::Error>; + ) -> Result<::Result, ::Error>; } #[cfg(test)] diff --git a/nomos-services/storage/src/backends/rocksdb.rs b/nomos-services/storage/src/backends/rocksdb.rs index d5a6e1d5d..eb5ebf770 100644 --- a/nomos-services/storage/src/backends/rocksdb.rs +++ b/nomos-services/storage/src/backends/rocksdb.rs @@ -2,8 +2,7 @@ use std::{marker::PhantomData, path::PathBuf, sync::Arc}; use async_trait::async_trait; use bytes::Bytes; -pub use rocksdb::Error; -use rocksdb::{Options, DB}; +use rocksdb::{Error, Options, DB}; use serde::{Deserialize, Serialize}; use super::{StorageBackend, StorageSerde, StorageTransaction}; @@ -64,13 +63,16 @@ impl core::fmt::Debug for RocksBackend { } #[async_trait] -impl StorageBackend for RocksBackend { +impl StorageBackend for RocksBackend +where + SerdeOp: StorageSerde + Send + Sync + 'static, +{ type Settings = RocksBackendSettings; type Error = rocksdb::Error; type Transaction = Transaction; type SerdeOperator = SerdeOp; - fn new(config: Self::Settings) -> Result { + fn new(config: Self::Settings) -> Result::Error> { let RocksBackendSettings { db_path, read_only, @@ -108,17 +110,24 @@ impl StorageBackend for RocksBack }) } - async fn store(&mut self, key: Bytes, value: Bytes) -> Result<(), Self::Error> { + async fn store( + &mut self, + key: Bytes, + value: Bytes, + ) -> Result<(), ::Error> { self.rocks.put(key, value) } - async fn load(&mut self, key: &[u8]) -> Result, Self::Error> { + async fn load(&mut self, key: &[u8]) -> Result, ::Error> { self.rocks .get(key) .map(|opt| opt.map(std::convert::Into::into)) } - async fn load_prefix(&mut self, prefix: &[u8]) -> Result, Self::Error> { + async fn load_prefix( + &mut self, + prefix: &[u8], + ) -> Result, ::Error> { let mut values = Vec::new(); let iter = self.rocks.prefix_iterator(prefix); @@ -134,7 +143,10 @@ impl StorageBackend for RocksBack Ok(values) } - async fn remove(&mut self, key: &[u8]) -> Result, Self::Error> { + async fn remove( + &mut self, + key: &[u8], + ) -> Result, ::Error> { let val = self.load(key).await?; if val.is_some() { self.rocks.delete(key).map(|()| val) @@ -146,7 +158,8 @@ impl StorageBackend for RocksBack async fn execute( &mut self, transaction: Self::Transaction, - ) -> Result<::Result, Self::Error> { + ) -> Result<::Result, ::Error> + { Ok(transaction.execute()) } } diff --git a/nomos-services/storage/src/backends/sled.rs b/nomos-services/storage/src/backends/sled.rs deleted file mode 100644 index 90aaedf0f..000000000 --- a/nomos-services/storage/src/backends/sled.rs +++ /dev/null @@ -1,143 +0,0 @@ -use std::{marker::PhantomData, path::PathBuf}; - -use async_trait::async_trait; -use bytes::Bytes; -use sled::transaction::{ - ConflictableTransactionResult, TransactionError, TransactionResult, TransactionalTree, -}; - -use super::{StorageBackend, StorageSerde, StorageTransaction}; - -#[derive(Debug, thiserror::Error)] -pub enum Error { - #[error(transparent)] - Transaction(#[from] TransactionError), - #[error(transparent)] - Error(#[from] sled::Error), -} - -/// Sled backend setting -#[derive(Clone, Debug)] -pub struct SledBackendSettings { - /// File path to the db file - pub db_path: PathBuf, -} - -/// Sled transaction type -/// Function that takes a reference to the transactional tree. No `&mut` needed -/// as sled operations work over simple `&`. -pub type SledTransaction = Box< - dyn Fn(&TransactionalTree) -> ConflictableTransactionResult, sled::Error> - + Send - + Sync, ->; - -impl StorageTransaction for SledTransaction { - type Result = TransactionResult, sled::Error>; - type Transaction = Self; -} - -/// Sled storage backend -pub struct SledBackend { - sled: sled::Db, - _serde_op: PhantomData, -} - -impl core::fmt::Debug for SledBackend { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - format!("SledBackend {{ sled: {:?} }}", self.sled).fmt(f) - } -} - -#[async_trait] -impl StorageBackend for SledBackend { - type Settings = SledBackendSettings; - type Error = Error; - type Transaction = SledTransaction; - type SerdeOperator = SerdeOp; - - fn new(config: Self::Settings) -> Result { - Ok(Self { - sled: sled::open(config.db_path)?, - _serde_op: PhantomData, - }) - } - - async fn store(&mut self, key: Bytes, value: Bytes) -> Result<(), Self::Error> { - let _ = self.sled.insert(key, value.to_vec())?; - Ok(()) - } - - async fn load(&mut self, key: &[u8]) -> Result, Self::Error> { - Ok(self.sled.get(key)?.map(|ivec| ivec.to_vec().into())) - } - - async fn load_prefix(&mut self, _key: &[u8]) -> Result, Self::Error> { - unimplemented!() - } - - async fn remove(&mut self, key: &[u8]) -> Result, Self::Error> { - Ok(self.sled.remove(key)?.map(|ivec| ivec.to_vec().into())) - } - - async fn execute( - &mut self, - transaction: Self::Transaction, - ) -> Result<::Result, Self::Error> { - Ok(self.sled.transaction(transaction)) - } -} - -#[cfg(test)] -mod test { - use tempfile::TempDir; - - use super::{super::testing::NoStorageSerde, *}; - - #[tokio::test] - async fn test_store_load_remove( - ) -> Result<(), as StorageBackend>::Error> { - let temp_path = TempDir::new().unwrap(); - let sled_settings = SledBackendSettings { - db_path: temp_path.path().to_path_buf(), - }; - let key = "foo"; - let value = "bar"; - - let mut sled_db: SledBackend = SledBackend::new(sled_settings)?; - sled_db - .store(key.as_bytes().into(), value.as_bytes().into()) - .await?; - let load_value = sled_db.load(key.as_bytes()).await?; - assert_eq!(load_value, Some(value.as_bytes().into())); - let removed_value = sled_db.remove(key.as_bytes()).await?; - assert_eq!(removed_value, Some(value.as_bytes().into())); - - Ok(()) - } - - #[tokio::test] - async fn test_transaction() -> Result<(), as StorageBackend>::Error> - { - let temp_path = TempDir::new().unwrap(); - - let sled_settings = SledBackendSettings { - db_path: temp_path.path().to_path_buf(), - }; - - let mut sled_db: SledBackend = SledBackend::new(sled_settings)?; - let result = sled_db - .execute(Box::new(move |tx| { - let key = "foo"; - let value = "bar"; - tx.insert(key, value)?; - let result = tx.get(key)?; - tx.remove(key)?; - Ok(result.map(|ivec| ivec.to_vec().into())) - })) - .await??; - assert_eq!(result, Some(b"bar".as_ref().into())); - - Ok(()) - } -} diff --git a/nomos-services/storage/src/lib.rs b/nomos-services/storage/src/lib.rs index 9037eeae5..56ad811a3 100644 --- a/nomos-services/storage/src/lib.rs +++ b/nomos-services/storage/src/lib.rs @@ -1,3 +1,4 @@ +pub mod api; pub mod backends; use std::{ @@ -14,12 +15,14 @@ use overwatch::{ state::{NoOperator, NoState}, AsServiceId, ServiceCore, ServiceData, }, - OpaqueServiceStateHandle, + DynError, OpaqueServiceStateHandle, }; use serde::{de::DeserializeOwned, Serialize}; use services_utils::overwatch::lifecycle; use tracing::error; +use crate::api::{StorageApiRequest, StorageOperation}; + /// Storage message that maps to [`StorageBackend`] trait pub enum StorageMsg { Load { @@ -43,6 +46,9 @@ pub enum StorageMsg { reply_channel: tokio::sync::oneshot::Sender<::Result>, }, + Api { + request: StorageApiRequest, + }, } /// Reply channel for storage messages @@ -144,6 +150,9 @@ impl Debug for StorageMsg { write!(f, "Remove {{ {key:?} }}") } Self::Execute { .. } => write!(f, "Execute transaction"), + Self::Api { .. } => { + write!(f, "Api {{ .. }}") + } } } } @@ -151,21 +160,25 @@ impl Debug for StorageMsg { /// Storage error /// Errors that may happen when performing storage operations #[derive(Debug, thiserror::Error)] -enum StorageServiceError { - #[error("Couldn't send a reply for operation `{operation}` with key [{key:?}]")] - ReplyError { operation: String, key: Bytes }, +pub enum StorageServiceError { + #[error("Couldn't send a reply [{message:?}]")] + ReplyError { message: String }, #[error("Storage backend error")] - BackendError(#[source] Backend::Error), + BackendError(Box), } /// Storage service that wraps a [`StorageBackend`] -pub struct StorageService { +pub struct StorageService +where + Backend: StorageBackend + Send + Sync + 'static, +{ backend: Backend, service_state: OpaqueServiceStateHandle, } -impl - StorageService +impl StorageService +where + Backend: StorageBackend + Send + Sync + 'static, { async fn handle_storage_message(msg: StorageMsg, backend: &mut Backend) { if let Err(e) = match msg { @@ -184,6 +197,7 @@ impl transaction, reply_channel, } => Self::handle_execute(backend, transaction, reply_channel).await, + StorageMsg::Api { request: api_call } => Self::handle_api_call(api_call, backend).await, } { // TODO: add proper logging println!("{e}"); @@ -194,16 +208,15 @@ impl backend: &mut Backend, key: Bytes, reply_channel: tokio::sync::oneshot::Sender>, - ) -> Result<(), StorageServiceError> { + ) -> Result<(), StorageServiceError> { let result: Option = backend .load(&key) .await - .map_err(StorageServiceError::BackendError)?; + .map_err(|e| StorageServiceError::BackendError(e.into()))?; reply_channel .send(result) .map_err(|_| StorageServiceError::ReplyError { - operation: "Load".to_owned(), - key, + message: format!("Load {key:?}"), }) } @@ -212,16 +225,15 @@ impl backend: &mut Backend, prefix: Bytes, reply_channel: tokio::sync::oneshot::Sender>, - ) -> Result<(), StorageServiceError> { + ) -> Result<(), StorageServiceError> { let result: Vec = backend .load_prefix(&prefix) .await - .map_err(StorageServiceError::BackendError)?; + .map_err(|e| StorageServiceError::BackendError(e.into()))?; reply_channel .send(result) .map_err(|_| StorageServiceError::ReplyError { - operation: "LoadPrefix".to_owned(), - key: prefix, + message: format!("LoadPrefix {prefix:?}"), }) } @@ -230,16 +242,15 @@ impl backend: &mut Backend, key: Bytes, reply_channel: tokio::sync::oneshot::Sender>, - ) -> Result<(), StorageServiceError> { + ) -> Result<(), StorageServiceError> { let result: Option = backend .remove(&key) .await - .map_err(StorageServiceError::BackendError)?; + .map_err(|e| StorageServiceError::BackendError(e.into()))?; reply_channel .send(result) .map_err(|_| StorageServiceError::ReplyError { - operation: "Remove".to_owned(), - key, + message: format!("Remove {key:?}"), }) } @@ -248,11 +259,11 @@ impl backend: &mut Backend, key: Bytes, value: Bytes, - ) -> Result<(), StorageServiceError> { + ) -> Result<(), StorageServiceError> { backend .store(key, value) .await - .map_err(StorageServiceError::BackendError) + .map_err(|e| StorageServiceError::BackendError(e.into())) } /// Handle execute message @@ -262,30 +273,39 @@ impl reply_channel: tokio::sync::oneshot::Sender< ::Result, >, - ) -> Result<(), StorageServiceError> { + ) -> Result<(), StorageServiceError> { let result = backend .execute(transaction) .await - .map_err(StorageServiceError::BackendError)?; + .map_err(|e| StorageServiceError::BackendError(e.into()))?; reply_channel .send(result) .map_err(|_| StorageServiceError::ReplyError { - operation: "Execute".to_owned(), - key: Bytes::new(), + message: "Execute transaction".to_owned(), }) } + + async fn handle_api_call( + api_call: StorageApiRequest, + api_backend: &mut Backend, + ) -> Result<(), StorageServiceError> { + as StorageOperation>::execute(api_call, api_backend) + .await + .map_err(|e| StorageServiceError::BackendError(e.into())) + } } #[async_trait] -impl - ServiceCore for StorageService +impl ServiceCore + for StorageService where + Backend: StorageBackend + Send + Sync + 'static, RuntimeServiceId: AsServiceId + Display + Send, { fn init( service_state: OpaqueServiceStateHandle, _init_state: Self::State, - ) -> Result { + ) -> Result { Ok(Self { backend: Backend::new(service_state.settings_reader.get_updated_settings())?, service_state, @@ -321,8 +341,9 @@ where } } -impl ServiceData - for StorageService +impl ServiceData for StorageService +where + Backend: StorageBackend + Send + Sync + 'static, { type Settings = Backend::Settings; type State = NoState;