Implement ledger rpc endpoints for getting commitments on slot (#527)

* Implement ledger rpc endpoints for getting commitments on slot

* Lint

* Use commitment ledger rpcs in full node instead of sequencer

* review fixes

* performance improvements

* Fix commitment bug

* change response for sequencer commitment

---------

Co-authored-by: eyusufatik <[email protected]>

bin/citrea/tests/e2e/mod.rs

@@ -1072,8 +1072,8 @@ async fn test_soft_confirmations_status_two_l1() -> Result<(), anyhow::Error> {
 
     // publish new da block
     da_service.publish_test_block().await.unwrap();
-    seq_test_client.send_publish_batch_request().await; // TODO https://github.com/chainwayxyz/citrea/issues/214
-    seq_test_client.send_publish_batch_request().await; // TODO https://github.com/chainwayxyz/citrea/issues/214
+    seq_test_client.send_publish_batch_request().await;
+    seq_test_client.send_publish_batch_request().await;
 
     sleep(Duration::from_secs(2)).await;
 

bin/citrea/tests/sequencer_commitments/mod.rs

@@ -259,3 +259,109 @@ async fn check_commitment_in_offchain_db() {
 
     seq_task.abort();
 }
+
+#[tokio::test]
+async fn test_ledger_get_commitments_on_slot() {
+    // citrea::initialize_logging();
+
+    let (seq_port_tx, seq_port_rx) = tokio::sync::oneshot::channel();
+
+    let seq_task = tokio::spawn(async {
+        start_rollup(
+            seq_port_tx,
+            GenesisPaths::from_dir("../test-data/genesis/integration-tests"),
+            BasicKernelGenesisPaths {
+                chain_state:
+                    "../test-data/genesis/integration-tests-low-limiting-number/chain_state.json"
+                        .into(),
+            },
+            RollupProverConfig::Execute,
+            NodeMode::SequencerNode,
+            None,
+            4,
+            true,
+            None,
+            None,
+            Some(true),
+            DEFAULT_DEPOSIT_MEMPOOL_FETCH_LIMIT,
+        )
+        .await;
+    });
+
+    let seq_port = seq_port_rx.await.unwrap();
+    let test_client = make_test_client(seq_port).await;
+    let da_service = MockDaService::new(MockAddress::from([0; 32]));
+
+    let (full_node_port_tx, full_node_port_rx) = tokio::sync::oneshot::channel();
+
+    let full_node_task = tokio::spawn(async move {
+        start_rollup(
+            full_node_port_tx,
+            GenesisPaths::from_dir("../test-data/genesis/integration-tests"),
+            BasicKernelGenesisPaths {
+                chain_state:
+                    "../test-data/genesis/integration-tests-low-limiting-number/chain_state.json"
+                        .into(),
+            },
+            RollupProverConfig::Execute,
+            NodeMode::FullNode(seq_port),
+            None,
+            4,
+            true,
+            None,
+            None,
+            Some(true),
+            DEFAULT_DEPOSIT_MEMPOOL_FETCH_LIMIT,
+        )
+        .await;
+    });
+
+    let full_node_port = full_node_port_rx.await.unwrap();
+
+    let full_node_test_client = make_test_client(full_node_port).await;
+    da_service.publish_test_block().await.unwrap();
+    sleep(Duration::from_secs(1)).await;
+
+    test_client.send_publish_batch_request().await;
+    test_client.send_publish_batch_request().await;
+    test_client.send_publish_batch_request().await;
+    test_client.send_publish_batch_request().await;
+    da_service.publish_test_block().await.unwrap();
+    // submits with new da block
+    test_client.send_publish_batch_request().await;
+    // full node gets the commitment
+    test_client.send_publish_batch_request().await;
+    // da_service.publish_test_block().await.unwrap();
+    sleep(Duration::from_secs(4)).await;
+
+    let commitments = full_node_test_client
+        .ledger_get_sequencer_commitments_on_slot_by_number(4)
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(commitments.len(), 1);
+
+    let second_hash = da_service.get_block_at(2).await.unwrap().header.hash;
+    assert_eq!(
+        commitments[0].l1_start_block_hash.to_vec(),
+        second_hash.0.to_vec()
+    );
+    assert_eq!(
+        commitments[0].l1_end_block_hash.to_vec(),
+        second_hash.0.to_vec()
+    );
+
+    assert_eq!(commitments[0].found_in_l1, 4);
+
+    let fourth_block_hash = da_service.get_block_at(4).await.unwrap().header.hash;
+
+    let commitments_hash = full_node_test_client
+        .ledger_get_sequencer_commitments_on_slot_by_hash(fourth_block_hash.0)
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(commitments_hash, commitments);
+
+    seq_task.abort();
+    full_node_task.abort();
+}

bin/citrea/tests/test_client/mod.rs

@@ -17,7 +17,7 @@ use jsonrpsee::rpc_params;
 use reth_primitives::BlockNumberOrTag;
 use reth_rpc_types::trace::geth::{GethDebugTracingOptions, GethTrace};
 use sequencer_client::GetSoftBatchResponse;
-use sov_rollup_interface::rpc::SoftConfirmationStatus;
+use sov_rollup_interface::rpc::{SequencerCommitmentResponse, SoftConfirmationStatus};
 
 pub const MAX_FEE_PER_GAS: u64 = 1000000001;
 
@@ -584,6 +584,32 @@ impl TestClient {
             .map_err(|e| e.into())
     }
 
+    pub(crate) async fn ledger_get_sequencer_commitments_on_slot_by_number(
+        &self,
+        height: u64,
+    ) -> Result<Option<Vec<SequencerCommitmentResponse>>, Box<dyn std::error::Error>> {
+        self.http_client
+            .request(
+                "ledger_getSequencerCommitmentsOnSlotByNumber",
+                rpc_params![height],
+            )
+            .await
+            .map_err(|e| e.into())
+    }
+
+    pub(crate) async fn ledger_get_sequencer_commitments_on_slot_by_hash(
+        &self,
+        hash: [u8; 32],
+    ) -> Result<Option<Vec<SequencerCommitmentResponse>>, Box<dyn std::error::Error>> {
+        self.http_client
+            .request(
+                "ledger_getSequencerCommitmentsOnSlotByHash",
+                rpc_params![hash],
+            )
+            .await
+            .map_err(|e| e.into())
+    }
+
     pub(crate) async fn get_limiting_number(&self) -> u64 {
         self.http_client
             .request(

crates/sequencer/src/sequencer.rs

@@ -788,6 +788,7 @@ where
                             commitment_info.l1_height_range.end().0,
                         ))
                         .expect("Sequencer: Failed to set last sequencer commitment L1 height");
+
                     warn!("Commitment info: {:?}", commitment_info);
                     if let Some(db_config) = db_config {
                         match PostgresConnector::new(db_config).await {

crates/sovereign-sdk/full-node/db/sov-db/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "sov-db"
 description = "A high-level DB interface for the Sovereign SDK"
-license = "Apache-2.0" # This license is inherited from Aptos 
+license = "Apache-2.0"                                          # This license is inherited from Aptos 
 edition = { workspace = true }
 authors = { workspace = true }
 homepage = { workspace = true }
@@ -17,20 +17,26 @@ resolver = "2"
 # Maintained by sovereign labs
 jmt = { workspace = true }
 sov-schema-db = { path = "../sov-schema-db", version = "0.3" }
-sov-rollup-interface = { path = "../../../rollup-interface", version = "0.3", features = ["native"] }
+sov-rollup-interface = { path = "../../../rollup-interface", version = "0.3", features = [
+    "native",
+] }
 
 # External
 anyhow = { workspace = true, default-features = true }
 arbitrary = { workspace = true, optional = true }
 byteorder = { workspace = true, default-features = true }
-borsh = { workspace = true, default-features = true, features = ["bytes", "rc"] }
+borsh = { workspace = true, default-features = true, features = [
+    "bytes",
+    "rc",
+] }
 proptest = { workspace = true, optional = true, default-features = true }
 proptest-derive = { workspace = true, optional = true }
 serde = { workspace = true, default-features = true, features = ["rc"] }
 tempfile = { workspace = true, optional = true }
 rocksdb = { workspace = true }
 bincode = { workspace = true }
 tokio = { workspace = true }
+hex = { workspace = true }
 
 
 [dev-dependencies]

crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/mod.rs

@@ -2,14 +2,14 @@ use std::path::Path;
 use std::sync::{Arc, Mutex};
 
 use serde::Serialize;
-use sov_rollup_interface::da::DaSpec;
+use sov_rollup_interface::da::{DaSpec, SequencerCommitment};
 use sov_rollup_interface::services::da::SlotData;
 use sov_rollup_interface::stf::{BatchReceipt, Event, SoftBatchReceipt};
 use sov_schema_db::{Schema, SchemaBatch, SeekKeyEncoder, DB};
 
 use crate::rocks_db_config::gen_rocksdb_options;
 use crate::schema::tables::{
-    BatchByHash, BatchByNumber, EventByKey, EventByNumber, L2RangeByL1Height,
+    BatchByHash, BatchByNumber, CommitmentsByNumber, EventByKey, EventByNumber, L2RangeByL1Height,
     LastSequencerCommitmentSent, SlotByHash, SlotByNumber, SoftBatchByHash, SoftBatchByNumber,
     SoftConfirmationStatus, TxByHash, TxByNumber, LEDGER_TABLES,
 };
@@ -519,4 +519,33 @@ impl LedgerDB {
     ) -> anyhow::Result<Option<L2HeightRange>> {
         self.db.get::<L2RangeByL1Height>(&l1_height)
     }
+
+    /// Gets the commitments in the da slot with given height if any
+    /// Adds the new coming commitment info
+    pub fn update_commitments_on_da_slot(
+        &self,
+        height: u64,
+        commitment: SequencerCommitment,
+    ) -> anyhow::Result<()> {
+        // get commitments
+        let commitments = self.db.get::<CommitmentsByNumber>(&SlotNumber(height))?;
+
+        match commitments {
+            // If there were other commitments, upsert
+            Some(mut commitments) => {
+                commitments.push(commitment);
+                self.db
+                    .put::<CommitmentsByNumber>(&SlotNumber(height), &commitments)
+            }
+            // Else insert
+            None => self
+                .db
+                .put::<CommitmentsByNumber>(&SlotNumber(height), &vec![commitment]),
+        }
+    }
+
+    /// Sets l1 height of l1 hash
+    pub fn set_l1_height_of_l1_hash(&self, hash: [u8; 32], height: u64) -> anyhow::Result<()> {
+        self.db.put::<SlotByHash>(&hash, &SlotNumber(height))
+    }
 }

crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/rpc.rs

@@ -1,15 +1,16 @@
 use serde::de::DeserializeOwned;
 use sov_rollup_interface::rpc::{
-    BatchIdAndOffset, BatchIdentifier, BatchResponse, EventIdentifier, ItemOrHash,
-    LedgerRpcProvider, QueryMode, SlotIdAndOffset, SlotIdentifier, SlotResponse,
-    SoftBatchIdentifier, SoftBatchResponse, TxIdAndOffset, TxIdentifier, TxResponse,
+    sequencer_commitment_to_response, BatchIdAndOffset, BatchIdentifier, BatchResponse,
+    EventIdentifier, ItemOrHash, LedgerRpcProvider, QueryMode, SequencerCommitmentResponse,
+    SlotIdAndOffset, SlotIdentifier, SlotResponse, SoftBatchIdentifier, SoftBatchResponse,
+    TxIdAndOffset, TxIdentifier, TxResponse,
 };
 use sov_rollup_interface::stf::Event;
 use tokio::sync::broadcast::Receiver;
 
 use crate::schema::tables::{
-    BatchByHash, BatchByNumber, EventByNumber, SlotByHash, SlotByNumber, SoftBatchByHash,
-    SoftBatchByNumber, SoftConfirmationStatus, TxByHash, TxByNumber,
+    BatchByHash, BatchByNumber, CommitmentsByNumber, EventByNumber, SlotByHash, SlotByNumber,
+    SoftBatchByHash, SoftBatchByNumber, SoftConfirmationStatus, TxByHash, TxByNumber,
 };
 use crate::schema::types::{
     BatchNumber, EventNumber, SlotNumber, StoredBatch, StoredSlot, TxNumber,
@@ -364,6 +365,24 @@ impl LedgerRpcProvider for LedgerDB {
             None => Ok(sov_rollup_interface::rpc::SoftConfirmationStatus::Trusted),
         }
     }
+    fn get_slot_number_by_hash(&self, hash: [u8; 32]) -> Result<Option<u64>, anyhow::Error> {
+        self.db.get::<SlotByHash>(&hash).map(|v| v.map(|a| a.0))
+    }
+
+    fn get_sequencer_commitments_on_slot_by_number(
+        &self,
+        height: u64,
+    ) -> Result<Option<Vec<SequencerCommitmentResponse>>, anyhow::Error> {
+        match self.db.get::<CommitmentsByNumber>(&SlotNumber(height))? {
+            Some(commitments) => Ok(Some(
+                commitments
+                    .into_iter()
+                    .map(|commitment| sequencer_commitment_to_response(commitment, height))
+                    .collect(),
+            )),
+            None => Ok(None),
+        }
+    }
 
     fn subscribe_slots(&self) -> Result<Receiver<u64>, anyhow::Error> {
         Ok(self.slot_subscriptions.subscribe())

crates/sovereign-sdk/full-node/db/sov-db/src/schema/tables.rs

@@ -29,6 +29,7 @@ use borsh::{maybestd, BorshDeserialize, BorshSerialize};
 use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
 use jmt::storage::{NibblePath, Node, NodeKey};
 use jmt::Version;
+use sov_rollup_interface::da::SequencerCommitment;
 use sov_rollup_interface::stf::{Event, EventKey};
 use sov_schema_db::schema::{KeyDecoder, KeyEncoder, ValueCodec};
 use sov_schema_db::{CodecError, SeekKeyEncoder};
@@ -62,6 +63,7 @@ pub const LEDGER_TABLES: &[&str] = &[
     TxByNumber::table_name(),
     EventByKey::table_name(),
     EventByNumber::table_name(),
+    CommitmentsByNumber::table_name(),
 ];
 
 /// A list of all tables used by the NativeDB. These tables store
@@ -223,6 +225,11 @@ define_table_with_default_codec!(
     (SlotByHash) DbHash => SlotNumber
 );
 
+define_table_with_default_codec!(
+    /// The primary source for sequencer commitment data
+    (CommitmentsByNumber) SlotNumber => Vec<SequencerCommitment>
+);
+
 define_table_with_seek_key_codec!(
     /// The primary source for soft batch data
     (SoftBatchByNumber) BatchNumber => StoredSoftBatch

crates/sovereign-sdk/full-node/sov-ledger-rpc/src/server.rs

@@ -173,6 +173,37 @@ where
         },
     )?;
 
+    rpc.register_async_method(
+        "ledger_getSequencerCommitmentsOnSlotByNumber",
+        |params, ledger| async move {
+            // Returns commitments on DA slot with given height.
+            let height: u64 = params.one()?;
+
+            ledger
+                .get_sequencer_commitments_on_slot_by_number(height)
+                .map_err(|e| to_jsonrpsee_error_object(LEDGER_RPC_ERROR, e))
+        },
+    )?;
+
+    rpc.register_async_method(
+        "ledger_getSequencerCommitmentsOnSlotByHash",
+        |params, ledger| async move {
+            // Returns commitments on DA slot with given hash.
+            let hash: [u8; 32] = params.one()?;
+            println!("hash:{:?}", hash);
+            let height = ledger
+                .get_slot_number_by_hash(hash)
+                .map_err(|e| to_jsonrpsee_error_object(LEDGER_RPC_ERROR, e))?;
+            println!("height:{:?}", height);
+            match height {
+                Some(height) => ledger
+                    .get_sequencer_commitments_on_slot_by_number(height)
+                    .map_err(|e| to_jsonrpsee_error_object(LEDGER_RPC_ERROR, e)),
+                None => Ok(None),
+            }
+        },
+    )?;
+
     rpc.register_subscription(
         "ledger_subscribeSlots",
         "ledger_slotProcessed",