storage: refactor and rocksdb implementation

Cargo.toml

@@ -14,15 +14,22 @@ keywords = ["crypto", "key-transparency"]
 readme = "README.md"
 
 [workspace]
+default-members = [
+    "crates/prism",
+    "crates/common",
+    "crates/errors",
+    "crates/storage",
+    "crates/da",
+]
+
 members = [
     "crates/prism",
     "crates/common",
-    "crates/nova",
-    "crates/groth16",
     "crates/errors",
     "crates/sp1",
+    "crates/storage",
+    "crates/da",
 ]
-default-members = ["crates/prism", "crates/common", "crates/errors"]
 resolver = "2"
 
 [workspace.dependencies]
@@ -72,10 +79,14 @@ secp256k1 = "0.29.0"
 sp1-zkvm = { version = "1.2.0" }
 sp1-sdk = { version = "1.2.0" }
 prism-common = { path = "crates/common" }
+prism-storage = { path = "crates/storage" }
 prism-nova = { path = "crates/nova" }
+prism-da = { path = "crates/da" }
 prism-errors = { path = "crates/errors" }
 prism-main = { path = "crates/prism" }
 prism-groth16 = { path = "crates/groth16" }
+rocksdb = { version = "0.21.0", features = ["multi-threaded-cf"] }
+
 
 [patch.crates-io]
 sha2-v0-10-8 = { git = "https://github.com/sp1-patches/RustCrypto-hashes", package = "sha2", branch = "patch-sha2-v0.10.8" }

crates/da/Cargo.toml

@@ -0,0 +1,52 @@
+[package]
+name = "prism-da"
+version.workspace = true
+authors.workspace = true
+edition.workspace = true
+description.workspace = true
+homepage.workspace = true
+repository.workspace = true
+license.workspace = true
+keywords.workspace = true
+readme.workspace = true
+
+[dependencies]
+axum = { workspace = true }
+borsh = { workspace = true }
+tower-http = { workspace = true }
+utoipa = { workspace = true }
+utoipa-swagger-ui = { workspace = true }
+async-trait = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+redis = { workspace = true }
+ed25519-dalek = { workspace = true }
+base64 = { workspace = true }
+tokio = { workspace = true }
+bellman = { workspace = true }
+bincode = { workspace = true }
+bls12_381 = { workspace = true }
+rand = { workspace = true }
+hex = { workspace = true }
+ff = { workspace = true }
+log = { workspace = true }
+pretty_env_logger = { workspace = true }
+clap = { workspace = true }
+config = { workspace = true }
+thiserror = { workspace = true }
+indexed-merkle-tree = { workspace = true }
+dotenvy = { workspace = true }
+celestia-rpc = { workspace = true }
+celestia-types = { workspace = true }
+mockall = { workspace = true }
+keystore-rs = { workspace = true }
+toml = { workspace = true }
+dirs = { workspace = true }
+anyhow = { workspace = true }
+jmt = { workspace = true }
+sha2 = { workspace = true }
+auto_impl = { workspace = true }
+prism-common = { workspace = true }
+prism-errors = { workspace = true }
+prism-groth16 = { workspace = true }
+sp1-sdk = { workspace = true }

crates/prism/src/da/celestia.rs → crates/da/src/celestia.rs

@@ -1,13 +1,12 @@
-use crate::{
-    cfg::CelestiaConfig,
-    da::{DataAvailabilityLayer, FinalizedEpoch},
-};
+use crate::{DataAvailabilityLayer, FinalizedEpoch};
 use anyhow::{anyhow, Context, Result};
 use async_trait::async_trait;
 use celestia_rpc::{BlobClient, Client, HeaderClient};
 use celestia_types::{blob::GasPrice, nmt::Namespace, Blob};
+use log::{debug, error, trace, warn};
 use prism_common::operation::Operation;
 use prism_errors::{DataAvailabilityError, GeneralError};
+use serde::{Deserialize, Serialize};
 use std::{
     self,
     sync::{
@@ -17,6 +16,8 @@ use std::{
 };
 use tokio::{sync::broadcast, task::spawn};
 
+use tokio::{sync::broadcast, task::spawn};
+
 use bincode;
 
 impl TryFrom<&Blob> for FinalizedEpoch {
@@ -29,6 +30,25 @@ impl TryFrom<&Blob> for FinalizedEpoch {
     }
 }
 
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct CelestiaConfig {
+    pub connection_string: String,
+    pub start_height: u64,
+    pub snark_namespace_id: String,
+    pub operation_namespace_id: Option<String>,
+}
+
+impl Default for CelestiaConfig {
+    fn default() -> Self {
+        CelestiaConfig {
+            connection_string: "ws://localhost:26658".to_string(),
+            start_height: 0,
+            snark_namespace_id: "00000000000000de1008".to_string(),
+            operation_namespace_id: Some("00000000000000de1009".to_string()),
+        }
+    }
+}
+
 pub struct CelestiaConnection {
     pub client: celestia_rpc::Client,
     pub snark_namespace: Namespace,

crates/da/src/lib.rs

@@ -0,0 +1,75 @@
+use anyhow::Result;
+use async_trait::async_trait;
+use ed25519_dalek::{Signature, Signer, SigningKey, VerifyingKey};
+use prism_common::{operation::Operation, tree::Digest};
+use serde::{Deserialize, Serialize};
+use sp1_sdk::SP1ProofWithPublicValues;
+use tokio::sync::broadcast;
+
+pub mod celestia;
+pub mod consts;
+pub mod memory;
+
+// FinalizedEpoch is the data structure that represents the finalized epoch data, and is posted to the DA layer.
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct FinalizedEpoch {
+    pub height: u64,
+    pub prev_commitment: Digest,
+    pub current_commitment: Digest,
+    pub proof: SP1ProofWithPublicValues,
+    pub signature: Option<String>,
+}
+
+impl FinalizedEpoch {
+    pub fn insert_signature(&mut self, key: &SigningKey) {
+        let plaintext = bincode::serialize(&self).unwrap();
+        let signature = key.sign(&plaintext);
+        self.signature = Some(hex::encode(signature.to_bytes()));
+    }
+
+    pub fn verify_signature(&self, vk: VerifyingKey) -> Result<()> {
+        let epoch_without_signature = FinalizedEpoch {
+            height: self.height,
+            prev_commitment: self.prev_commitment,
+            current_commitment: self.current_commitment,
+            proof: self.proof.clone(),
+            signature: None,
+        };
+
+        let message = bincode::serialize(&epoch_without_signature)
+            .map_err(|e| anyhow::anyhow!("Failed to serialize epoch: {}", e))?;
+
+        let signature = self
+            .signature
+            .as_ref()
+            .ok_or_else(|| anyhow::anyhow!("No signature present"))?;
+
+        let signature_bytes = hex::decode(signature)
+            .map_err(|e| anyhow::anyhow!("Failed to decode signature: {}", e))?;
+
+        if signature_bytes.len() != 64 {
+            return Err(anyhow::anyhow!("Invalid signature length"));
+        }
+
+        let signature: Signature = signature_bytes
+            .as_slice()
+            .try_into()
+            .map_err(|_| anyhow::anyhow!("Invalid signature length"))?;
+
+        vk.verify_strict(&message, &signature)
+            .map_err(|e| anyhow::anyhow!("Signature verification failed: {}", e))?;
+        Ok(())
+    }
+}
+
+#[async_trait]
+pub trait DataAvailabilityLayer: Send + Sync {
+    async fn get_latest_height(&self) -> Result<u64>;
+    async fn initialize_sync_target(&self) -> Result<u64>;
+    async fn get_finalized_epoch(&self, height: u64) -> Result<Option<FinalizedEpoch>>;
+    async fn submit_finalized_epoch(&self, epoch: FinalizedEpoch) -> Result<u64>;
+    async fn get_operations(&self, height: u64) -> Result<Vec<Operation>>;
+    async fn submit_operations(&self, operations: Vec<Operation>) -> Result<u64>;
+    async fn start(&self) -> Result<()>;
+    fn subscribe_to_heights(&self) -> broadcast::Receiver<u64>;
+}

crates/prism/src/da/memory.rs → crates/da/src/memory.rs

@@ -1,6 +1,7 @@
-use crate::da::{DataAvailabilityLayer, FinalizedEpoch};
+use crate::{DataAvailabilityLayer, FinalizedEpoch};
 use anyhow::Result;
 use async_trait::async_trait;
+use log::debug;
 use prism_common::operation::Operation;
 use std::{collections::VecDeque, sync::Arc};
 use tokio::{

crates/prism/Cargo.toml

@@ -49,7 +49,9 @@ jmt = { workspace = true }
 sha2 = { workspace = true }
 auto_impl = { workspace = true }
 prism-common = { workspace = true, features = ["test_utils"] }
+prism-storage = { workspace = true }
 prism-errors = { workspace = true }
+prism-da = { workspace = true }
 prism-groth16 = { workspace = true }
 sp1-sdk = { workspace = true }
 

crates/prism/src/cfg.rs

@@ -1,17 +1,20 @@
-use crate::{
-    consts::{DA_RETRY_COUNT, DA_RETRY_INTERVAL},
-    da::memory::InMemoryDataAvailabilityLayer,
-};
 use anyhow::{anyhow, Context, Result};
 use clap::{Parser, Subcommand};
 use config::{builder::DefaultState, ConfigBuilder, File};
 use dirs::home_dir;
 use dotenvy::dotenv;
+use log::{error, warn};
 use prism_errors::{DataAvailabilityError, GeneralError, PrismError};
+use prism_storage::redis::RedisConfig;
 use serde::{Deserialize, Serialize};
 use std::{fs, path::Path, sync::Arc};
 
-use crate::da::{celestia::CelestiaConnection, DataAvailabilityLayer};
+use prism_da::{
+    celestia::{CelestiaConfig, CelestiaConnection},
+    consts::{DA_RETRY_COUNT, DA_RETRY_INTERVAL},
+    memory::InMemoryDataAvailabilityLayer,
+    DataAvailabilityLayer,
+};
 
 #[derive(Clone, Debug, Subcommand, Deserialize)]
 pub enum Commands {
@@ -97,38 +100,6 @@ impl Default for WebServerConfig {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct RedisConfig {
-    pub connection_string: String,
-}
-
-impl Default for RedisConfig {
-    fn default() -> Self {
-        RedisConfig {
-            connection_string: "redis://127.0.0.1/".to_string(),
-        }
-    }
-}
-
-#[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct CelestiaConfig {
-    pub connection_string: String,
-    pub start_height: u64,
-    pub snark_namespace_id: String,
-    pub operation_namespace_id: Option<String>,
-}
-
-impl Default for CelestiaConfig {
-    fn default() -> Self {
-        CelestiaConfig {
-            connection_string: "ws://localhost:26658".to_string(),
-            start_height: 0,
-            snark_namespace_id: "00000000000000de1008".to_string(),
-            operation_namespace_id: Some("00000000000000de1009".to_string()),
-        }
-    }
-}
-
 impl Default for Config {
     fn default() -> Self {
         Config {
@@ -300,7 +271,7 @@ pub async fn initialize_da_layer(
             unreachable!() // This line should never be reached due to the return in the last iteration
         }
         DALayerOption::InMemory => {
-            let (da_layer, _height_rx, _block_rx) = InMemoryDataAvailabilityLayer::new(1);
+            let (da_layer, _height_rx, _block_rx) = InMemoryDataAvailabilityLayer::new(30);
             Ok(Arc::new(da_layer) as Arc<dyn DataAvailabilityLayer + 'static>)
         }
         DALayerOption::None => Err(anyhow!(PrismError::ConfigError(

crates/prism/src/lib.rs

@@ -1,8 +1,5 @@
 pub mod cfg;
-pub mod consts;
-pub mod da;
 pub mod node_types;
-pub mod storage;
 pub mod utils;
 pub mod webserver;
 #[macro_use]

crates/prism/src/main.rs

@@ -1,19 +1,14 @@
 mod cfg;
-pub mod consts;
-pub mod da;
 mod node_types;
-pub mod storage;
 mod utils;
 mod webserver;
 
-use cfg::{initialize_da_layer, load_config};
+use cfg::{initialize_da_layer, load_config, CommandLineArgs, Commands};
 use clap::Parser;
 use keystore_rs::{KeyChain, KeyStore, KeyStoreType};
-
-use crate::cfg::{CommandLineArgs, Commands};
 use node_types::{lightclient::LightClient, sequencer::Sequencer, NodeType};
+use prism_storage::RedisConnection;
 use std::sync::Arc;
-use storage::RedisConnection;
 
 #[macro_use]
 extern crate log;

crates/prism/src/node_types/lightclient.rs

@@ -1,15 +1,14 @@
-use crate::cfg::CelestiaConfig;
+use crate::node_types::NodeType;
 use anyhow::{Context, Result};
 use async_trait::async_trait;
 use ed25519_dalek::VerifyingKey;
 use prism_common::tree::Digest;
+use prism_da::{celestia::CelestiaConfig, DataAvailabilityLayer};
 use prism_errors::{DataAvailabilityError, GeneralError};
 use sp1_sdk::{ProverClient, SP1VerifyingKey};
 use std::{self, sync::Arc};
 use tokio::{sync::broadcast, task::spawn};
 
-use crate::{da::DataAvailabilityLayer, node_types::NodeType};
-
 pub const PRISM_ELF: &[u8] = include_bytes!("../../../../elf/riscv32im-succinct-zkvm-elf");
 
 pub struct LightClient {