logos-co/nomos-node
1
0
Fork 0
mirror of https://github.com/logos-co/nomos-node.git synced 2026-08-31 03:21:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

New storage abstraction layer

Browse Source
This commit is contained in:
Andrus Salumets
2025-05-13 19:05:38 +01:00
committed by GitHub
parent ff67598188
commit 449203ea70
36 changed files with 828 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.cargo-deny.toml
-1
View File
@@ -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." },
]
Cargo.toml
-2
View File
@@ -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" }
dependencies_graph.dot
+48 -49
View File
@@ -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 [ ]
}
nomos-core/chain-defs/src/block/mod.rs
+25
View File
@@ -59,3 +59,28 @@ impl<
self.bl_blobs.len()
}
}
impl<
Tx: Clone + Eq + Hash + Serialize + DeserializeOwned,
BlobCertificate: Clone + Eq + Hash + Serialize + DeserializeOwned,
> TryFrom<Bytes> for Block<Tx, BlobCertificate>
{
type Error = wire::Error;
fn try_from(bytes: Bytes) -> Result<Self, Self::Error> {
wire::deserialize(&bytes)
}
}
impl<
Tx: Clone + Eq + Hash + Serialize + DeserializeOwned,
BlobCertificate: Clone + Eq + Hash + Serialize + DeserializeOwned,
> TryFrom<Block<Tx, BlobCertificate>> for Bytes
{
type Error = wire::Error;
fn try_from(block: Block<Tx, BlobCertificate>) -> Result<Self, Self::Error> {
let serialized = wire::serialize(&block)?;
Ok(serialized.into())
}
}
nomos-da/storage/Cargo.toml
-8
View File
@@ -1,8 +0,0 @@
[package]
edition = "2021"
license = { workspace = true }
name = "nomos-da-storage"
version = "0.1.0"
[dependencies]
bytes = "1.2"
nomos-da/storage/src/lib.rs
-1
View File
@@ -1 +0,0 @@
pub mod rocksdb;
nomos-services/api/Cargo.toml
-1
View File
@@ -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"] }
nomos-services/api/src/http/da_shares.rs
+4 -3
View File
@@ -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,
};
nomos-services/api/src/http/storage.rs
+4 -3
View File
@@ -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,
};
nomos-services/cryptarchia-consensus/Cargo.toml
-1
View File
@@ -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"
nomos-services/cryptarchia-consensus/src/lib.rs
+11 -8
View File
@@ -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<BlobId = DaPool::Item> + Clone + Send + Sync + 'static,
BS::Settings: Send + Sync + 'static,
Storage: StorageBackend + Send + Sync + 'static,
<Storage as StorageChainApi>::Block:
TryFrom<Block<ClPool::Item, DaPool::Item>> + TryInto<Block<ClPool::Item, DaPool::Item>>,
SamplingBackend: DaSamplingServiceBackend<SamplingRng> + Send,
SamplingBackend::Settings: Clone,
SamplingBackend::Share: Debug + Send + 'static,
@@ -756,6 +757,8 @@ where
BS: BlobSelect<BlobId = DaPool::Item> + Clone + Send + Sync + 'static,
BS::Settings: Send,
Storage: StorageBackend + Send + Sync + 'static,
<Storage as StorageChainApi>::Block:
TryFrom<Block<ClPool::Item, DaPool::Item>> + TryInto<Block<ClPool::Item, DaPool::Item>>,
SamplingBackend: DaSamplingServiceBackend<SamplingRng> + 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 =
<StorageMsg<_>>::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) {
nomos-services/cryptarchia-consensus/src/relays.rs
+9 -4
View File
@@ -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<RuntimeServiceId>,
NetworkAdapter: network::NetworkAdapter<RuntimeServiceId>,
Storage: StorageBackend + Send + Sync,
Storage: StorageBackend + Send + Sync + 'static,
SamplingRng: SeedableRng + RngCore,
SamplingBackend: DaSamplingServiceBackend<SamplingRng>,
TxS: TxSelect,
@@ -136,13 +139,13 @@ where
BS::Settings: Send,
ClPool: RecoverableMempool<BlockId = HeaderId>,
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<RuntimeServiceId, Payload = ClPool::Item, Key = ClPool::Key>,
DaPool: RecoverableMempool<BlockId = HeaderId>,
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,
<Storage as StorageChainApi>::Block:
TryFrom<Block<ClPool::Item, DaPool::Item>> + TryInto<Block<ClPool::Item, DaPool::Item>>,
TxS: TxSelect<Tx = ClPool::Item>,
TxS::Settings: Send,
DaVerifierBackend: nomos_da_verifier::backend::VerifierBackend + Send + Sync + 'static,
nomos-services/cryptarchia-consensus/src/storage/adapters/storage.rs
+45 -37
View File
@@ -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<Storage, Tx, BlobCertificate, RuntimeServiceId>
where
Storage: StorageBackend + Send + Sync,
Storage: StorageBackend + Send + Sync + 'static,
{
pub storage_relay:
OutboundRelay<<StorageService<Storage, RuntimeServiceId> as ServiceData>::Message>,
@@ -18,40 +20,15 @@ where
_blob_certificate: PhantomData<BlobCertificate>,
}
impl<Storage, Tx, BlobCertificate, RuntimeServiceId>
StorageAdapter<Storage, Tx, BlobCertificate, RuntimeServiceId>
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<Value>(&self, key: Bytes) -> Option<Value>
where
Value: DeserializeOwned,
{
let (msg, receiver) = <StorageMsg<Storage>>::new_load_message(key);
self.storage_relay.send(msg).await.unwrap();
receiver.recv().await.unwrap()
}
}
#[async_trait::async_trait]
impl<Storage, Tx, BlobCertificate, RuntimeServiceId> StorageAdapterTrait<RuntimeServiceId>
for StorageAdapter<Storage, Tx, BlobCertificate, RuntimeServiceId>
where
Storage: StorageBackend + Send + Sync,
Tx: Clone + Eq + Hash + DeserializeOwned + Send + Sync,
BlobCertificate: Clone + Eq + Hash + DeserializeOwned + Send + Sync,
Storage: StorageBackend + Send + Sync + 'static,
<Storage as StorageChainApi>::Block:
TryFrom<Block<Tx, BlobCertificate>> + TryInto<Block<Tx, BlobCertificate>>,
Tx: Clone + Eq + Hash + Serialize + DeserializeOwned + Send + Sync + 'static,
BlobCertificate: Clone + Eq + Hash + Serialize + DeserializeOwned + Send + Sync + 'static,
{
type Backend = Storage;
type Block = Block<Tx, BlobCertificate>;
@@ -68,8 +45,39 @@ where
}
}
async fn get_block(&self, key: &HeaderId) -> Option<Self::Block> {
let key: [u8; 32] = (*key).into();
self.get_value(Bytes::copy_from_slice(&key)).await
async fn get_block(&self, header_id: &HeaderId) -> Option<Self::Block> {
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(())
}
}
nomos-services/cryptarchia-consensus/src/storage/mod.rs
+7 -1
View File
@@ -6,7 +6,7 @@ use overwatch::services::{relay::OutboundRelay, ServiceData};
#[async_trait::async_trait]
pub trait StorageAdapter<RuntimeServiceId> {
type Backend: StorageBackend + Send + Sync;
type Backend: StorageBackend + Send + Sync + 'static;
type Block;
async fn new(
@@ -22,4 +22,10 @@ pub trait StorageAdapter<RuntimeServiceId> {
///
/// The block with the given header id. If no block is found, returns None.
async fn get_block(&self, key: &HeaderId) -> Option<Self::Block>;
async fn store_block(
&self,
header_id: HeaderId,
block: Self::Block,
) -> Result<(), overwatch::DynError>;
}
nomos-services/data-availability/indexer/Cargo.toml
-1
View File
@@ -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 }
nomos-services/data-availability/indexer/src/storage/adapters/rocksdb.rs
+4 -3
View File
@@ -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,
};
nomos-services/data-availability/sampling/Cargo.toml
-1
View File
@@ -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 }
nomos-services/data-availability/sampling/src/storage/adapters/rocksdb.rs
+18 -15
View File
@@ -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: <Self::Share as Share>::BlobId,
share_idx: ShareIndex,
) -> Result<Option<<Self::Share as Share>::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())
}
}
nomos-services/data-availability/verifier/Cargo.toml
-1
View File
@@ -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 }
nomos-services/data-availability/verifier/src/storage/adapters/rocksdb.rs
+25 -79
View File
@@ -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: <Self::Share as Share>::SharesCommitments,
light_share: <Self::Share as 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: <Self::Share as Share>::BlobId,
share_idx: <Self::Share as Share>::ShareIndex,
) -> Result<Option<<Self::Share as Share>::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<B, S> RocksAdapter<B, S>
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: &<B as Share>::BlobId,
) -> Result<Option<HashSet<B::ShareIndex>>, 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())
}
}
nomos-services/storage/Cargo.toml
+2 -3
View File
@@ -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"
nomos-services/storage/src/api/backend/mod.rs
+2
View File
@@ -0,0 +1,2 @@
#[cfg(feature = "rocksdb-backend")]
pub mod rocksdb;
nomos-services/storage/src/api/backend/rocksdb/blocks.rs
+30
View File
@@ -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<SerdeOp: StorageSerde + Send + Sync + 'static> StorageChainApi for RocksBackend<SerdeOp> {
type Error = Error;
type Block = Bytes;
async fn get_block(&mut self, header_id: HeaderId) -> Result<Option<Self::Block>, 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
}
}
nomos-services/storage/src/api/backend/rocksdb/da.rs
+95
View File
@@ -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<SerdeOp: StorageSerde + Send + Sync + 'static> StorageDaApi for RocksBackend<SerdeOp> {
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<Option<Self::Share>, 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::<HashSet<[u8; 2]>>(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
}
}
nomos-services/storage/src/api/backend/rocksdb/mod.rs
+10
View File
@@ -0,0 +1,10 @@
use crate::{
api::StorageBackendApi,
backends::{rocksdb::RocksBackend, StorageSerde},
};
pub mod blocks;
pub mod da;
pub mod utils;
impl<SerdeOp: StorageSerde + Send + Sync + 'static> StorageBackendApi for RocksBackend<SerdeOp> {}
nomos-da/storage/src/rocksdb/mod.rs → nomos-services/storage/src/api/backend/rocksdb/utils.rs
+1 -8
View File
@@ -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();
nomos-services/storage/src/api/chain/mod.rs
+20
View File
@@ -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<Option<Self::Block>, Self::Error>;
async fn store_block(
&mut self,
header_id: HeaderId,
block: Self::Block,
) -> Result<(), Self::Error>;
}
nomos-services/storage/src/api/chain/requests.rs
+92
View File
@@ -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<B: StorageBackend> {
GetBlock {
header_id: HeaderId,
response_tx: Sender<Option<<B as StorageChainApi>::Block>>,
},
StoreBlock {
header_id: HeaderId,
block: <B as StorageChainApi>::Block,
},
}
impl<B> StorageOperation<B> for ChainApiRequest<B>
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<B: StorageBackend>(
backend: &mut B,
header_id: HeaderId,
response_tx: Sender<Option<B::Block>>,
) -> 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<B: StorageBackend>(
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<Api: StorageBackend> StorageMsg<Api> {
#[must_use]
pub const fn get_block_request(
header_id: HeaderId,
response_tx: Sender<Option<<Api as StorageChainApi>::Block>>,
) -> Self {
Self::Api {
request: StorageApiRequest::Chain(ChainApiRequest::GetBlock {
header_id,
response_tx,
}),
}
}
pub const fn store_block_request(
header_id: HeaderId,
block: <Api as StorageChainApi>::Block,
) -> Self {
Self::Api {
request: StorageApiRequest::Chain(ChainApiRequest::StoreBlock { header_id, block }),
}
}
}
nomos-services/storage/src/api/da/mod.rs
+33
View File
@@ -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<Option<Self::Share>, 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>;
}
nomos-services/storage/src/api/da/requests.rs
+136
View File
@@ -0,0 +1,136 @@
use tokio::sync::oneshot::Sender;
use crate::{
api::{da::StorageDaApi, StorageApiRequest, StorageBackendApi, StorageOperation},
backends::StorageBackend,
StorageMsg, StorageServiceError,
};
pub enum DaApiRequest<B: StorageBackend> {
GetLightShare {
blob_id: <B as StorageDaApi>::BlobId,
share_idx: <B as StorageDaApi>::ShareIndex,
response_tx: Sender<Option<<B as StorageDaApi>::Share>>,
},
StoreLightShare {
blob_id: <B as StorageDaApi>::BlobId,
share_idx: <B as StorageDaApi>::ShareIndex,
light_share: <B as StorageDaApi>::Share,
},
StoreSharedCommitments {
blob_id: <B as StorageDaApi>::BlobId,
shared_commitments: <B as StorageDaApi>::Commitments,
},
}
impl<B> StorageOperation<B> for DaApiRequest<B>
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<B: StorageBackend>(
backend: &mut B,
blob_id: B::BlobId,
share_idx: B::ShareIndex,
response_tx: Sender<Option<B::Share>>,
) -> 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<B: StorageBackend>(
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<B: StorageBackend>(
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<Api: StorageBackend> StorageMsg<Api> {
#[must_use]
pub const fn get_light_share_request(
blob_id: <Api as StorageDaApi>::BlobId,
share_idx: <Api as StorageDaApi>::ShareIndex,
response_tx: Sender<Option<<Api as StorageDaApi>::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: <Api as StorageDaApi>::BlobId,
share_idx: <Api as StorageDaApi>::ShareIndex,
light_share: <Api as StorageDaApi>::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: <Api as StorageDaApi>::BlobId,
shared_commitments: <Api as StorageDaApi>::Commitments,
) -> Self {
Self::Api {
request: StorageApiRequest::Da(DaApiRequest::StoreSharedCommitments {
blob_id,
shared_commitments,
}),
}
}
}
nomos-services/storage/src/api/mod.rs
+35
View File
@@ -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<B: StorageBackend> {
async fn execute(self, api: &mut B) -> Result<(), StorageServiceError>;
}
pub enum StorageApiRequest<B: StorageBackend> {
Chain(ChainApiRequest<B>),
Da(DaApiRequest<B>),
}
impl<B: StorageBackend> StorageOperation<B> for StorageApiRequest<B> {
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,
}
}
}
nomos-services/storage/src/backends/mock.rs
+79 -6
View File
@@ -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<SerdeOp: StorageSerde + Send + Sync + 'static> StorageBackend for MockStora
type Transaction = MockStorageTransaction;
type SerdeOperator = SerdeOp;
fn new(_config: Self::Settings) -> Result<Self, Self::Error> {
fn new(_config: Self::Settings) -> Result<Self, <Self as StorageBackend>::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<(), <Self as StorageBackend>::Error> {
let _ = self.inner.insert(key, value);
Ok(())
}
async fn load(&mut self, key: &[u8]) -> Result<Option<Bytes>, Self::Error> {
async fn load(&mut self, key: &[u8]) -> Result<Option<Bytes>, <Self as StorageBackend>::Error> {
Ok(self.inner.get(key).cloned())
}
async fn load_prefix(&mut self, _key: &[u8]) -> Result<Vec<Bytes>, Self::Error> {
async fn load_prefix(
&mut self,
_key: &[u8],
) -> Result<Vec<Bytes>, <Self as StorageBackend>::Error> {
unimplemented!()
}
async fn remove(&mut self, key: &[u8]) -> Result<Option<Bytes>, Self::Error> {
async fn remove(
&mut self,
key: &[u8],
) -> Result<Option<Bytes>, <Self as StorageBackend>::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<(), <Self as StorageBackend>::Error> {
transaction(&mut self.inner);
Ok(())
}
}
#[async_trait]
impl<SerdeOp: StorageSerde + Send + Sync + 'static> StorageChainApi for MockStorage<SerdeOp> {
type Error = MockStorageError;
type Block = Bytes;
async fn get_block(
&mut self,
_header_id: HeaderId,
) -> Result<Option<Self::Block>, Self::Error> {
unimplemented!()
}
async fn store_block(
&mut self,
_header_id: HeaderId,
_block: Self::Block,
) -> Result<(), Self::Error> {
unimplemented!()
}
}
#[async_trait]
impl<SerdeOp: StorageSerde + Send + Sync + 'static> StorageDaApi for MockStorage<SerdeOp> {
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<Option<Self::Share>, 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<SerdeOp: StorageSerde + Send + Sync + 'static> StorageBackendApi for MockStorage<SerdeOp> {}
nomos-services/storage/src/backends/mod.rs
+19 -9
View File
@@ -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<Self, Self::Error>;
async fn store(&mut self, key: Bytes, value: Bytes) -> Result<(), Self::Error>;
async fn load(&mut self, key: &[u8]) -> Result<Option<Bytes>, Self::Error>;
fn new(config: Self::Settings) -> Result<Self, <Self as StorageBackend>::Error>;
async fn store(
&mut self,
key: Bytes,
value: Bytes,
) -> Result<(), <Self as StorageBackend>::Error>;
async fn load(&mut self, key: &[u8]) -> Result<Option<Bytes>, <Self as StorageBackend>::Error>;
/// Loads all values whose keys start with the given prefix.
async fn load_prefix(&mut self, prefix: &[u8]) -> Result<Vec<Bytes>, Self::Error>;
async fn remove(&mut self, key: &[u8]) -> Result<Option<Bytes>, Self::Error>;
async fn load_prefix(
&mut self,
prefix: &[u8],
) -> Result<Vec<Bytes>, <Self as StorageBackend>::Error>;
async fn remove(
&mut self,
key: &[u8],
) -> Result<Option<Bytes>, <Self as StorageBackend>::Error>;
/// Execute a transaction in the current backend
async fn execute(
&mut self,
transaction: Self::Transaction,
) -> Result<<Self::Transaction as StorageTransaction>::Result, Self::Error>;
) -> Result<<Self::Transaction as StorageTransaction>::Result, <Self as StorageBackend>::Error>;
}
#[cfg(test)]
nomos-services/storage/src/backends/rocksdb.rs
+22 -9
View File
@@ -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<SerdeOp> core::fmt::Debug for RocksBackend<SerdeOp> {
}
#[async_trait]
impl<SerdeOp: StorageSerde + Send + Sync + 'static> StorageBackend for RocksBackend<SerdeOp> {
impl<SerdeOp> StorageBackend for RocksBackend<SerdeOp>
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<Self, Self::Error> {
fn new(config: Self::Settings) -> Result<Self, <Self as StorageBackend>::Error> {
let RocksBackendSettings {
db_path,
read_only,
@@ -108,17 +110,24 @@ impl<SerdeOp: StorageSerde + Send + Sync + 'static> 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<(), <Self as StorageBackend>::Error> {
self.rocks.put(key, value)
}
async fn load(&mut self, key: &[u8]) -> Result<Option<Bytes>, Self::Error> {
async fn load(&mut self, key: &[u8]) -> Result<Option<Bytes>, <Self as StorageBackend>::Error> {
self.rocks
.get(key)
.map(|opt| opt.map(std::convert::Into::into))
}
async fn load_prefix(&mut self, prefix: &[u8]) -> Result<Vec<Bytes>, Self::Error> {
async fn load_prefix(
&mut self,
prefix: &[u8],
) -> Result<Vec<Bytes>, <Self as StorageBackend>::Error> {
let mut values = Vec::new();
let iter = self.rocks.prefix_iterator(prefix);
@@ -134,7 +143,10 @@ impl<SerdeOp: StorageSerde + Send + Sync + 'static> StorageBackend for RocksBack
Ok(values)
}
async fn remove(&mut self, key: &[u8]) -> Result<Option<Bytes>, Self::Error> {
async fn remove(
&mut self,
key: &[u8],
) -> Result<Option<Bytes>, <Self as StorageBackend>::Error> {
let val = self.load(key).await?;
if val.is_some() {
self.rocks.delete(key).map(|()| val)
@@ -146,7 +158,8 @@ impl<SerdeOp: StorageSerde + Send + Sync + 'static> StorageBackend for RocksBack
async fn execute(
&mut self,
transaction: Self::Transaction,
) -> Result<<Self::Transaction as StorageTransaction>::Result, Self::Error> {
) -> Result<<Self::Transaction as StorageTransaction>::Result, <Self as StorageBackend>::Error>
{
Ok(transaction.execute())
}
}
nomos-services/storage/src/backends/sled.rs
-143
View File
@@ -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<Option<Bytes>, sled::Error>
+ Send
+ Sync,
>;
impl StorageTransaction for SledTransaction {
type Result = TransactionResult<Option<Bytes>, sled::Error>;
type Transaction = Self;
}
/// Sled storage backend
pub struct SledBackend<SerdeOp> {
sled: sled::Db,
_serde_op: PhantomData<SerdeOp>,
}
impl<SerdeOp> core::fmt::Debug for SledBackend<SerdeOp> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
format!("SledBackend {{ sled: {:?} }}", self.sled).fmt(f)
}
}
#[async_trait]
impl<SerdeOp: StorageSerde + Send + Sync + 'static> StorageBackend for SledBackend<SerdeOp> {
type Settings = SledBackendSettings;
type Error = Error;
type Transaction = SledTransaction;
type SerdeOperator = SerdeOp;
fn new(config: Self::Settings) -> Result<Self, Self::Error> {
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<Option<Bytes>, Self::Error> {
Ok(self.sled.get(key)?.map(|ivec| ivec.to_vec().into()))
}
async fn load_prefix(&mut self, _key: &[u8]) -> Result<Vec<Bytes>, Self::Error> {
unimplemented!()
}
async fn remove(&mut self, key: &[u8]) -> Result<Option<Bytes>, Self::Error> {
Ok(self.sled.remove(key)?.map(|ivec| ivec.to_vec().into()))
}
async fn execute(
&mut self,
transaction: Self::Transaction,
) -> Result<<Self::Transaction as StorageTransaction>::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<(), <SledBackend<NoStorageSerde> 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<NoStorageSerde> = 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<(), <SledBackend<NoStorageSerde> 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<NoStorageSerde> = 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(())
}
}
nomos-services/storage/src/lib.rs
+52 -31
View File
@@ -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<Backend: StorageBackend> {
Load {
@@ -43,6 +46,9 @@ pub enum StorageMsg<Backend: StorageBackend> {
reply_channel:
tokio::sync::oneshot::Sender<<Backend::Transaction as StorageTransaction>::Result>,
},
Api {
request: StorageApiRequest<Backend>,
},
}
/// Reply channel for storage messages
@@ -144,6 +150,9 @@ impl<Backend: StorageBackend> Debug for StorageMsg<Backend> {
write!(f, "Remove {{ {key:?} }}")
}
Self::Execute { .. } => write!(f, "Execute transaction"),
Self::Api { .. } => {
write!(f, "Api {{ .. }}")
}
}
}
}
@@ -151,21 +160,25 @@ impl<Backend: StorageBackend> Debug for StorageMsg<Backend> {
/// Storage error
/// Errors that may happen when performing storage operations
#[derive(Debug, thiserror::Error)]
enum StorageServiceError<Backend: StorageBackend> {
#[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<dyn std::error::Error + Send + Sync>),
}
/// Storage service that wraps a [`StorageBackend`]
pub struct StorageService<Backend: StorageBackend + Send + Sync + 'static, RuntimeServiceId> {
pub struct StorageService<Backend, RuntimeServiceId>
where
Backend: StorageBackend + Send + Sync + 'static,
{
backend: Backend,
service_state: OpaqueServiceStateHandle<Self, RuntimeServiceId>,
}
impl<Backend: StorageBackend + Send + Sync + 'static, RuntimeServiceId>
StorageService<Backend, RuntimeServiceId>
impl<Backend, RuntimeServiceId> StorageService<Backend, RuntimeServiceId>
where
Backend: StorageBackend + Send + Sync + 'static,
{
async fn handle_storage_message(msg: StorageMsg<Backend>, backend: &mut Backend) {
if let Err(e) = match msg {
@@ -184,6 +197,7 @@ impl<Backend: StorageBackend + Send + Sync + 'static, RuntimeServiceId>
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: StorageBackend + Send + Sync + 'static, RuntimeServiceId>
backend: &mut Backend,
key: Bytes,
reply_channel: tokio::sync::oneshot::Sender<Option<Bytes>>,
) -> Result<(), StorageServiceError<Backend>> {
) -> Result<(), StorageServiceError> {
let result: Option<Bytes> = 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: StorageBackend + Send + Sync + 'static, RuntimeServiceId>
backend: &mut Backend,
prefix: Bytes,
reply_channel: tokio::sync::oneshot::Sender<Vec<Bytes>>,
) -> Result<(), StorageServiceError<Backend>> {
) -> Result<(), StorageServiceError> {
let result: Vec<Bytes> = 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: StorageBackend + Send + Sync + 'static, RuntimeServiceId>
backend: &mut Backend,
key: Bytes,
reply_channel: tokio::sync::oneshot::Sender<Option<Bytes>>,
) -> Result<(), StorageServiceError<Backend>> {
) -> Result<(), StorageServiceError> {
let result: Option<Bytes> = 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: StorageBackend + Send + Sync + 'static, RuntimeServiceId>
backend: &mut Backend,
key: Bytes,
value: Bytes,
) -> Result<(), StorageServiceError<Backend>> {
) -> 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<Backend: StorageBackend + Send + Sync + 'static, RuntimeServiceId>
reply_channel: tokio::sync::oneshot::Sender<
<Backend::Transaction as StorageTransaction>::Result,
>,
) -> Result<(), StorageServiceError<Backend>> {
) -> 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<Backend>,
api_backend: &mut Backend,
) -> Result<(), StorageServiceError> {
<StorageApiRequest<Backend> as StorageOperation<Backend>>::execute(api_call, api_backend)
.await
.map_err(|e| StorageServiceError::BackendError(e.into()))
}
}
#[async_trait]
impl<Backend: StorageBackend + Send + Sync + 'static, RuntimeServiceId>
ServiceCore<RuntimeServiceId> for StorageService<Backend, RuntimeServiceId>
impl<Backend, RuntimeServiceId> ServiceCore<RuntimeServiceId>
for StorageService<Backend, RuntimeServiceId>
where
Backend: StorageBackend + Send + Sync + 'static,
RuntimeServiceId: AsServiceId<Self> + Display + Send,
{
fn init(
service_state: OpaqueServiceStateHandle<Self, RuntimeServiceId>,
_init_state: Self::State,
) -> Result<Self, overwatch::DynError> {
) -> Result<Self, DynError> {
Ok(Self {
backend: Backend::new(service_state.settings_reader.get_updated_settings())?,
service_state,
@@ -321,8 +341,9 @@ where
}
}
impl<Backend: StorageBackend + Send + Sync, RuntimeServiceId> ServiceData
for StorageService<Backend, RuntimeServiceId>
impl<Backend, RuntimeServiceId> ServiceData for StorageService<Backend, RuntimeServiceId>
where
Backend: StorageBackend + Send + Sync + 'static,
{
type Settings = Backend::Settings;
type State = NoState<Self::Settings>;