Update for latest SDK; Add constructors for DaService and DaVerifier (#…

…19)

* Make compatible with risc0.

Introduce the CompactHeader type, which allows us to circumvent
informalsystems/tendermint-rs#1309.
Update to Tendermint 0.32, which allows removal of unnecessary data
copies.

Feature gate `CelestiaService` and its netowkr/io dependencies
behind the "native" feature, re-organizing files as necessary.

* Update to latest sdk. Add CI

* Update for latest sdk

* Update to user prover-demo branch of sdk

* Bump commit version for sdk

* Update to latest sdk

* Fix after Encode/Decode removals (#18)

* Clippy fixes

* Add header_respons.json

* Remove unneeded patch

---------

Co-authored-by: Nikolai Golub <nikolai@sovlabs.io>

Cargo.toml

@@ -6,31 +6,30 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-sovereign-sdk =  { git = "https://github.com/Sovereign-Labs/sovereign.git", rev = "e848baea380880d7c95d3d5cfbf7486c51fe3b79" }
-tendermint = "0.32"
-tendermint-proto = "0.32"
-
+borsh = { version = "0.10.3", features = ["bytes"] }
 prost = "0.11"
 prost-types = "0.11"
-borsh = { version = "0.10.3", features = ["bytes"] }
+tendermint = "0.32"
+tendermint-proto = "0.32"
 
 # Convenience
-reqwest = { version = "0.11.13", features = ["blocking"], optional = true }
-tokio = { version = "1", features = ["full"], optional = true }
-serde = { version = "1", features = ["derive"] }
-serde_json = "1"
-serde_cbor = "0.11.2"
+anyhow = "1.0.62"
+base64 = "0.13.1"
 hex = { version = "0.4.3", features = ["serde"] }
 hex-literal = "0.3.4"
-base64 = "0.13.1"
-anyhow = "1.0.62"
 jsonrpsee = { version = "0.16.2", features = ["http-client"], optional = true }
+reqwest = { version = "0.11.13", features = ["blocking"], optional = true }
+serde = { version = "1", features = ["derive"] }
+serde_cbor = "0.11.2"
+serde_json = "1"
+tokio = { version = "1", features = ["full"], optional = true }
 tracing = "0.1.37"
 
-nmt-rs = { git = "https://github.com/Sovereign-Labs/nmt-rs.git", rev = "aec2dcdc279b381162537f5b20ce43d1d46dc42f", features = ["serde", "borsh"] }
+sovereign-sdk = { git = "https://github.com/Sovereign-Labs/sovereign.git", rev = "3f948c098cf37040de3aaf2a86a18a1d91d082be" }
+nmt-rs = { git = "https://github.com/Sovereign-Labs/nmt-rs.git", rev = "dd37588444fca72825d11fe4a46838f66525c49f", features = ["serde", "borsh"] }
 
 [dev-dependencies]
-postcard = { version = "1", features = ["use-std"]}
+postcard = { version = "1", features = ["use-std"] }
 
 [build-dependencies]
 prost-build = { version = "0.11" }

src/celestia.rs

@@ -1,3 +1,4 @@
+use std::fmt::{Display, Formatter};
 use std::{cell::RefCell, ops::Range};
 
 use borsh::{BorshDeserialize, BorshSerialize};
@@ -43,7 +44,7 @@ pub struct PartialBlockId {
 /// a tendermint::block::Header from being deserialized in most formats except JSON. However
 /// it also provides a significant efficiency benefit over the standard tendermint type, which
 /// performs a complete protobuf serialization every time `.hash()` is called.
-#[derive(PartialEq, Eq, Debug, Clone, Serialize, Deserialize)]
+#[derive(PartialEq, Eq, Debug, Clone, Serialize, Deserialize, BorshDeserialize, BorshSerialize)]
 pub struct CompactHeader {
     /// Header version
     pub version: Vec<u8>,
@@ -182,9 +183,7 @@ impl CompactHeader {
     }
 }
 
-#[derive(
-    PartialEq, Debug, Clone, Deserialize, serde::Serialize, BorshDeserialize, BorshSerialize,
-)]
+#[derive(PartialEq, Debug, Clone, Serialize, Deserialize, BorshSerialize, BorshDeserialize)]
 pub struct DataAvailabilityHeader {
     pub row_roots: Vec<NamespacedHash>,
     pub column_roots: Vec<NamespacedHash>,
@@ -224,16 +223,17 @@ pub struct CelestiaHeaderResponse {
     pub dah: MarshalledDataAvailabilityHeader,
 }
 
-#[derive(Deserialize, Serialize, PartialEq, Debug, Clone)]
+#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
 pub struct NamespacedSharesResponse {
     pub shares: Option<Vec<String>>,
     pub height: u64,
 }
 
-#[derive(Debug, PartialEq, Clone, Deserialize, serde::Serialize)]
+#[derive(Debug, PartialEq, Clone, Serialize, Deserialize, BorshSerialize, BorshDeserialize)]
 pub struct CelestiaHeader {
     pub dah: DataAvailabilityHeader,
     pub header: CompactHeader,
+    #[borsh_skip]
     #[serde(skip)]
     cached_prev_hash: RefCell<Option<TmHash>>,
 }
@@ -260,7 +260,7 @@ impl CanonicalHash for CelestiaHeader {
     }
 }
 
-#[derive(PartialEq, Clone, Debug, BorshDeserialize, BorshSerialize)]
+#[derive(PartialEq, Clone, Debug, Serialize, Deserialize, BorshDeserialize, BorshSerialize)]
 pub struct BlobWithSender {
     pub blob: Blob,
     pub sender: CelestiaAddress,
@@ -283,7 +283,7 @@ impl BlockHeader for CelestiaHeader {
             )
             .expect("must not call prev_hash on block with no predecessor")
             .hash;
-        *cached_hash = Some(TmHash(hash.clone()));
+        *cached_hash = Some(TmHash(hash));
         TmHash(hash)
     }
 }
@@ -329,6 +329,21 @@ impl<'a> TryFrom<&'a [u8]> for H160 {
         anyhow::bail!("Adress is not exactly 20 bytes");
     }
 }
+
+impl From<[u8; 32]> for H160 {
+    fn from(value: [u8; 32]) -> Self {
+        let mut addr = [0u8; 20];
+        addr.copy_from_slice(&value[12..]);
+        Self(addr)
+    }
+}
+
+impl Display for H160 {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "0x{}", hex::encode(self.0))
+    }
+}
+
 impl Address for H160 {}
 
 pub fn parse_pfb_namespace(

src/da_service.rs

@@ -1,12 +1,16 @@
 use std::{collections::HashMap, future::Future, pin::Pin};
 
-use jsonrpsee::{core::client::ClientT, http_client::HttpClient};
+use jsonrpsee::{
+    core::client::ClientT,
+    http_client::{HeaderMap, HttpClient},
+};
 use nmt_rs::NamespaceId;
 use sovereign_sdk::services::da::DaService;
 use tracing::{debug, info, span, Level};
 
 // 0x736f762d74657374 = b"sov-test"
-// pub const ROLLUP_NAMESPACE: NamespaceId = NamespaceId(b"sov-test");
+// For testing, use this NamespaceId (b"sov-test"):
+// pub const ROLLUP_NAMESPACE: NamespaceId = NamespaceId([115, 111, 118, 45, 116, 101, 115, 116]);
 
 use crate::{
     parse_pfb_namespace,
@@ -17,33 +21,36 @@ use crate::{
     verifier::{
         address::CelestiaAddress,
         proofs::{CompletenessProof, CorrectnessProof},
-        CelestiaSpec, PFB_NAMESPACE, ROLLUP_NAMESPACE,
+        CelestiaSpec, RollupParams, PFB_NAMESPACE,
     },
     BlobWithSender, CelestiaHeader, CelestiaHeaderResponse, DataAvailabilityHeader,
 };
 
 #[derive(Debug, Clone)]
 pub struct CelestiaService {
     client: HttpClient,
+    rollup_namespace: NamespaceId,
 }
 
 impl CelestiaService {
-    pub fn with_client(client: HttpClient) -> Self {
-        Self { client }
+    pub fn with_client(client: HttpClient, nid: NamespaceId) -> Self {
+        Self {
+            client,
+            rollup_namespace: nid,
+        }
     }
 }
 
-impl CelestiaService {}
-
 /// Fetch the rollup namespace shares and etx data. Returns a tuple `(rollup_shares, etx_shares)`
-pub async fn fetch_needed_shares_by_header(
+async fn fetch_needed_shares_by_header(
+    rollup_namespace: NamespaceId,
     client: &HttpClient,
     header: &serde_json::Value,
 ) -> Result<(NamespaceGroup, NamespaceGroup), BoxError> {
     let dah = header
         .get("dah")
         .ok_or(BoxError::msg("missing dah in block header"))?;
-    let rollup_namespace_str = base64::encode(ROLLUP_NAMESPACE).into();
+    let rollup_namespace_str = base64::encode(rollup_namespace).into();
     let rollup_shares_future = {
         let params: Vec<&serde_json::Value> = vec![dah, &rollup_namespace_str];
         client.request::<RpcNamespacedSharesResponse, _>("share.GetSharesByNamespace", params)
@@ -77,6 +84,26 @@ pub async fn fetch_needed_shares_by_header(
 
     Ok((rollup_shares, tx_data))
 }
+/// Runtime configuration for the DA service
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+pub struct DaServiceConfig {
+    /// The jwt used to authenticate with the Celestia rpc server
+    pub celestia_rpc_auth_token: String,
+    /// The address of the Celestia rpc server
+    #[serde(default = "default_rpc_addr")]
+    pub celestia_rpc_address: String,
+    /// The maximum size of a Celestia RPC response, in bytes
+    #[serde(default = "default_max_response_size")]
+    pub max_celestia_response_body_size: u32,
+}
+
+fn default_rpc_addr() -> String {
+    "http://localhost:11111/".into()
+}
+
+fn default_max_response_size() -> u32 {
+    1024 * 1024 * 100 // 100 MB
+}
 
 impl DaService for CelestiaService {
     type FilteredBlock = FilteredCelestiaBlock;
@@ -87,8 +114,30 @@ impl DaService for CelestiaService {
 
     type Error = BoxError;
 
+    type RuntimeConfig = DaServiceConfig;
+
+    fn new(config: Self::RuntimeConfig, chain_params: RollupParams) -> Self {
+        let client = {
+            let mut headers = HeaderMap::new();
+            headers.insert(
+                "Authorization",
+                format!("Bearer {}", config.celestia_rpc_auth_token)
+                    .parse()
+                    .unwrap(),
+            );
+            jsonrpsee::http_client::HttpClientBuilder::default()
+                .set_headers(headers)
+                .max_request_body_size(config.max_celestia_response_body_size) // 100 MB
+                .build(&config.celestia_rpc_address)
+        }
+        .expect("Client initialization is valid");
+
+        Self::with_client(client, chain_params.namespace)
+    }
+
     fn get_finalized_at(&self, height: u64) -> Self::Future<Self::FilteredBlock> {
         let client = self.client.clone();
+        let rollup_namespace = self.rollup_namespace.clone();
         Box::pin(async move {
             let _span = span!(Level::TRACE, "fetching finalized block", height = height);
             // Fetch the header and relevant shares via RPC
@@ -98,7 +147,8 @@ impl DaService for CelestiaService {
                 .await?;
             debug!(header_result = ?header);
             info!("Fetching shares...");
-            let (rollup_shares, tx_data) = fetch_needed_shares_by_header(&client, &header).await?;
+            let (rollup_shares, tx_data) =
+                fetch_needed_shares_by_header(rollup_namespace, &client, &header).await?;
 
             info!("Fetching EDS...");
             // Fetch entire extended data square
@@ -120,7 +170,7 @@ impl DaService for CelestiaService {
                     .await?;
             // Parse out all of the rows containing rollup data
             let rollup_rows = get_rows_containing_namespace(
-                ROLLUP_NAMESPACE,
+                rollup_namespace,
                 &dah,
                 data_square.rows()?.into_iter(),
             )
@@ -132,7 +182,7 @@ impl DaService for CelestiaService {
             let mut pfd_map = HashMap::new();
             for tx in pfds {
                 for (idx, nid) in tx.0.namespace_ids.iter().enumerate() {
-                    if nid == &ROLLUP_NAMESPACE.0[..] {
+                    if nid == &rollup_namespace.0[..] {
                         // TODO: Retool this map to avoid cloning txs
                         pfd_map.insert(tx.0.share_commitments[idx].clone(), tx.clone());
                     }
@@ -190,7 +240,8 @@ impl DaService for CelestiaService {
     ) {
         let relevant_txs = self.extract_relevant_txs(block.clone());
         let etx_proofs = CorrectnessProof::for_block(&block, &relevant_txs);
-        let rollup_row_proofs = CompletenessProof::from_filtered_block(&block);
+        let rollup_row_proofs =
+            CompletenessProof::from_filtered_block(&block, self.rollup_namespace);
 
         (relevant_txs, etx_proofs.0, rollup_row_proofs.0)
     }