diff --git a/bin/citrea/src/rollup/mod.rs b/bin/citrea/src/rollup/mod.rs
index a36085ebb..8e9d3b6b6 100644
--- a/bin/citrea/src/rollup/mod.rs
+++ b/bin/citrea/src/rollup/mod.rs
@@ -4,6 +4,7 @@ use citrea_fullnode::{CitreaFullnode, FullNode};
 use citrea_prover::{CitreaProver, Prover};
 use citrea_sequencer::{CitreaSequencer, Sequencer, SequencerConfig};
 pub use mock::*;
+use sov_db::ledger_db::SharedLedgerOps;
 use sov_modules_api::storage::HierarchicalStorageManager;
 use sov_modules_api::Spec;
 use sov_modules_rollup_blueprint::RollupBlueprint;
diff --git a/bin/citrea/src/test_rpc.rs b/bin/citrea/src/test_rpc.rs
index cead7b8cf..d6d2b2c4a 100644
--- a/bin/citrea/src/test_rpc.rs
+++ b/bin/citrea/src/test_rpc.rs
@@ -8,7 +8,7 @@ use proptest::{prop_compose, proptest};
 use reqwest::header::CONTENT_TYPE;
 use serde_json::json;
 use sha2::Digest;
-use sov_db::ledger_db::{LedgerDB, SlotCommit};
+use sov_db::ledger_db::{LedgerDB, SequencerLedgerOps, SharedLedgerOps, SlotCommit};
 use sov_mock_da::MockDaSpec;
 #[cfg(test)]
 use sov_mock_da::{MockBlock, MockBlockHeader, MockHash};
diff --git a/crates/fullnode/src/lib.rs b/crates/fullnode/src/lib.rs
index 7af6b108e..bfbbe26b5 100644
--- a/crates/fullnode/src/lib.rs
+++ b/crates/fullnode/src/lib.rs
@@ -1,6 +1,7 @@
 use std::net::SocketAddr;
 
 pub use runner::*;
+use sov_db::ledger_db::LedgerDB;
 use sov_modules_rollup_blueprint::RollupBlueprint;
 use sov_modules_stf_blueprint::StfBlueprint;
 use tokio::sync::oneshot;
@@ -18,6 +19,7 @@ pub struct FullNode<S: RollupBlueprint> {
         S::DaService,
         S::Vm,
         S::NativeContext,
+        LedgerDB,
     >,
     /// Rpc methods for the rollup.
     pub rpc_methods: jsonrpsee::RpcModule<()>,
diff --git a/crates/fullnode/src/runner.rs b/crates/fullnode/src/runner.rs
index 9eca791d3..e54e1d4e7 100644
--- a/crates/fullnode/src/runner.rs
+++ b/crates/fullnode/src/runner.rs
@@ -14,7 +14,7 @@ use jsonrpsee::RpcModule;
 use rs_merkle::algorithms::Sha256;
 use rs_merkle::MerkleTree;
 use sequencer_client::{GetSoftBatchResponse, SequencerClient};
-use sov_db::ledger_db::{LedgerDB, SlotCommit};
+use sov_db::ledger_db::{NodeLedgerOps, SlotCommit};
 use sov_db::schema::types::{BatchNumber, SlotNumber, StoredSoftBatch, StoredStateTransition};
 use sov_modules_api::Context;
 use sov_modules_stf_blueprint::StfBlueprintTrait;
@@ -36,7 +36,7 @@ use tracing::{debug, error, info, instrument, warn};
 type StateRoot<ST, Vm, Da> = <ST as StateTransitionFunction<Vm, Da>>::StateRoot;
 
 /// Citrea's own STF runner implementation.
-pub struct CitreaFullnode<Stf, Sm, Da, Vm, C>
+pub struct CitreaFullnode<Stf, Sm, Da, Vm, C, DB>
 where
     Da: DaService,
     Vm: ZkvmHost + Zkvm,
@@ -44,14 +44,14 @@ where
     Stf: StateTransitionFunction<Vm, Da::Spec, Condition = <Da::Spec as DaSpec>::ValidityCondition>
         + StfBlueprintTrait<C, Da::Spec, Vm>,
     C: Context,
+    DB: NodeLedgerOps,
 {
     start_l2_height: u64,
     start_l1_height: u64,
     da_service: Da,
     stf: Stf,
     storage_manager: Sm,
-    /// made pub so that sequencer can clone it
-    pub ledger_db: LedgerDB,
+    ledger_db: DB,
     state_root: StateRoot<Stf, Vm, Da::Spec>,
     batch_hash: SoftConfirmationHash,
     rpc_config: RpcConfig,
@@ -68,7 +68,7 @@ where
     soft_confirmation_tx: broadcast::Sender<u64>,
 }
 
-impl<Stf, Sm, Da, Vm, C> CitreaFullnode<Stf, Sm, Da, Vm, C>
+impl<Stf, Sm, Da, Vm, C, DB> CitreaFullnode<Stf, Sm, Da, Vm, C, DB>
 where
     Da: DaService<Error = anyhow::Error> + Clone + Send + Sync + 'static,
     Vm: ZkvmHost + Zkvm,
@@ -81,6 +81,7 @@ where
             ChangeSet = Sm::NativeChangeSet,
         > + StfBlueprintTrait<C, Da::Spec, Vm>,
     C: Context,
+    DB: NodeLedgerOps,
 {
     /// Creates a new `StateTransitionRunner`.
     ///
@@ -93,7 +94,7 @@ where
         public_keys: RollupPublicKeys,
         rpc_config: RpcConfig,
         da_service: Da,
-        ledger_db: LedgerDB,
+        ledger_db: DB,
         stf: Stf,
         mut storage_manager: Sm,
         init_variant: InitVariant<Stf, Vm, Da::Spec>,
diff --git a/crates/fullnode/tests/runner_initialization_tests.rs b/crates/fullnode/tests/runner_initialization_tests.rs
index be40a096e..54db3fdbe 100644
--- a/crates/fullnode/tests/runner_initialization_tests.rs
+++ b/crates/fullnode/tests/runner_initialization_tests.rs
@@ -49,6 +49,7 @@ fn initialize_runner(
     MockDaService,
     MockZkvm<MockValidityCond>,
     sov_modules_api::default_context::DefaultContext,
+    LedgerDB,
 > {
     let da_storage_path = storage_path.join("da").to_path_buf();
     let rollup_storage_path = storage_path.join("rollup").to_path_buf();
diff --git a/crates/fullnode/tests/runner_reorg_tests.rs b/crates/fullnode/tests/runner_reorg_tests.rs
index 60cb5bdb5..0df95281a 100644
--- a/crates/fullnode/tests/runner_reorg_tests.rs
+++ b/crates/fullnode/tests/runner_reorg_tests.rs
@@ -12,7 +12,7 @@ use sov_stf_runner::{
 mod hash_stf;
 
 use hash_stf::{get_result_from_blocks, HashStf, Q, S};
-use sov_db::ledger_db::LedgerDB;
+use sov_db::ledger_db::{LedgerDB, NodeLedgerOps};
 use sov_mock_zkvm::MockCodeCommitment;
 use sov_prover_storage_manager::ProverStorageManager;
 use sov_rollup_interface::services::da::DaService;
@@ -159,7 +159,7 @@ async fn runner_execution(
     };
     let storage_manager = ProverStorageManager::new(storage_config).unwrap();
 
-    let mut runner: CitreaFullnode<_, _, _, _, DefaultContext> = CitreaFullnode::new(
+    let mut runner: CitreaFullnode<_, _, _, _, DefaultContext, _> = CitreaFullnode::new(
         rollup_config.runner.unwrap(),
         rollup_config.public_keys,
         rollup_config.rpc,
diff --git a/crates/prover/src/lib.rs b/crates/prover/src/lib.rs
index ebc730ece..3f334fb09 100644
--- a/crates/prover/src/lib.rs
+++ b/crates/prover/src/lib.rs
@@ -1,5 +1,6 @@
 use std::net::SocketAddr;
 
+use sov_db::ledger_db::LedgerDB;
 use sov_modules_rollup_blueprint::RollupBlueprint;
 use sov_modules_stf_blueprint::StfBlueprint;
 use tokio::sync::oneshot;
@@ -20,6 +21,7 @@ pub struct Prover<S: RollupBlueprint> {
         S::Vm,
         StfBlueprint<S::NativeContext, S::DaSpec, S::Vm, S::NativeRuntime>,
         S::ProverService,
+        LedgerDB,
     >,
     /// Rpc methods for the rollup.
     pub rpc_methods: jsonrpsee::RpcModule<()>,
diff --git a/crates/prover/src/runner.rs b/crates/prover/src/runner.rs
index 5bb085639..56f505452 100644
--- a/crates/prover/src/runner.rs
+++ b/crates/prover/src/runner.rs
@@ -16,7 +16,7 @@ use jsonrpsee::RpcModule;
 use rand::Rng;
 use sequencer_client::{GetSoftBatchResponse, SequencerClient};
 use shared_backup_db::{DbPoolError, PostgresConnector, ProofType};
-use sov_db::ledger_db::{LedgerDB, SlotCommit};
+use sov_db::ledger_db::{ProverLedgerOps, SlotCommit};
 use sov_db::schema::types::{BatchNumber, SlotNumber, StoredStateTransition};
 use sov_modules_api::storage::HierarchicalStorageManager;
 use sov_modules_api::{BlobReaderTrait, Context, SignedSoftConfirmationBatch, SlotData};
@@ -42,7 +42,7 @@ type CommitmentStateTransitionData<Stf, Vm, Da> = (
     VecDeque<Vec<<<Da as DaService>::Spec as DaSpec>::BlockHeader>>,
 );
 
-pub struct CitreaProver<C, Da, Sm, Vm, Stf, Ps>
+pub struct CitreaProver<C, Da, Sm, Vm, Stf, Ps, DB>
 where
     C: Context,
     Da: DaService,
@@ -52,13 +52,13 @@ where
         + StfBlueprintTrait<C, Da::Spec, Vm>,
 
     Ps: ProverService<Vm>,
+    DB: ProverLedgerOps + Clone,
 {
     start_l2_height: u64,
     da_service: Da,
     stf: Stf,
     storage_manager: Sm,
-    /// made pub so that sequencer can clone it
-    pub ledger_db: LedgerDB,
+    ledger_db: DB,
     state_root: StateRoot<Stf, Vm, Da::Spec>,
     batch_hash: SoftConfirmationHash,
     rpc_config: RpcConfig,
@@ -75,7 +75,7 @@ where
     soft_confirmation_tx: broadcast::Sender<u64>,
 }
 
-impl<C, Da, Sm, Vm, Stf, Ps> CitreaProver<C, Da, Sm, Vm, Stf, Ps>
+impl<C, Da, Sm, Vm, Stf, Ps, DB> CitreaProver<C, Da, Sm, Vm, Stf, Ps, DB>
 where
     C: Context,
     Da: DaService<Error = anyhow::Error> + Clone + Send + Sync + 'static,
@@ -89,6 +89,7 @@ where
             ChangeSet = Sm::NativeChangeSet,
         > + StfBlueprintTrait<C, Da::Spec, Vm>,
     Ps: ProverService<Vm, StateRoot = Stf::StateRoot, Witness = Stf::Witness, DaService = Da>,
+    DB: ProverLedgerOps + Clone,
 {
     /// Creates a new `StateTransitionRunner`.
     ///
@@ -101,7 +102,7 @@ where
         public_keys: RollupPublicKeys,
         rpc_config: RpcConfig,
         da_service: Da,
-        ledger_db: LedgerDB,
+        ledger_db: DB,
         stf: Stf,
         mut storage_manager: Sm,
         init_variant: InitVariant<Stf, Vm, Da::Spec>,
diff --git a/crates/sequencer/src/commitment_controller.rs b/crates/sequencer/src/commitment_controller.rs
index 8548cfae7..bc675c292 100644
--- a/crates/sequencer/src/commitment_controller.rs
+++ b/crates/sequencer/src/commitment_controller.rs
@@ -4,7 +4,7 @@ use std::ops::RangeInclusive;
 use anyhow::anyhow;
 use rs_merkle::algorithms::Sha256;
 use rs_merkle::MerkleTree;
-use sov_db::ledger_db::LedgerDB;
+use sov_db::ledger_db::SequencerLedgerOps;
 use sov_db::schema::types::BatchNumber;
 use sov_rollup_interface::da::SequencerCommitment;
 use tracing::{debug, instrument};
@@ -19,8 +19,8 @@ pub struct CommitmentInfo {
 /// Returns none if the commitable L2 block range is shorter than `min_soft_confirmations_per_commitment`
 /// Returns `CommitmentInfo` if the sequencer should commit
 #[instrument(level = "debug", skip_all, fields(prev_l1_height), err)]
-pub fn get_commitment_info(
-    ledger_db: &LedgerDB,
+pub fn get_commitment_info<T: SequencerLedgerOps>(
+    ledger_db: &T,
     min_soft_confirmations_per_commitment: u64,
     state_diff_threshold_reached: bool,
 ) -> anyhow::Result<Option<CommitmentInfo>> {
diff --git a/crates/sequencer/src/lib.rs b/crates/sequencer/src/lib.rs
index 2957b2261..8c8940595 100644
--- a/crates/sequencer/src/lib.rs
+++ b/crates/sequencer/src/lib.rs
@@ -11,6 +11,7 @@ use std::net::SocketAddr;
 
 pub use config::{SequencerConfig, SequencerMempoolConfig};
 pub use sequencer::CitreaSequencer;
+use sov_db::ledger_db::LedgerDB;
 use sov_modules_rollup_blueprint::RollupBlueprint;
 use sov_modules_stf_blueprint::StfBlueprint;
 use tokio::sync::oneshot;
@@ -26,6 +27,7 @@ pub struct Sequencer<S: RollupBlueprint> {
         S::StorageManager,
         S::Vm,
         StfBlueprint<S::NativeContext, S::DaSpec, S::Vm, S::NativeRuntime>,
+        LedgerDB,
     >,
     /// Rpc methods for the rollup.
     pub rpc_methods: jsonrpsee::RpcModule<()>,
diff --git a/crates/sequencer/src/sequencer.rs b/crates/sequencer/src/sequencer.rs
index 1a89af74e..6e6d37ce3 100644
--- a/crates/sequencer/src/sequencer.rs
+++ b/crates/sequencer/src/sequencer.rs
@@ -27,7 +27,7 @@ use shared_backup_db::{CommitmentStatus, PostgresConnector};
 use soft_confirmation_rule_enforcer::SoftConfirmationRuleEnforcer;
 use sov_accounts::Accounts;
 use sov_accounts::Response::{AccountEmpty, AccountExists};
-use sov_db::ledger_db::{LedgerDB, SlotCommit};
+use sov_db::ledger_db::{SequencerLedgerOps, SlotCommit};
 use sov_db::schema::types::{BatchNumber, SlotNumber};
 use sov_modules_api::hooks::HookSoftConfirmationInfo;
 use sov_modules_api::transaction::Transaction;
@@ -64,7 +64,7 @@ type StateRoot<ST, Vm, Da> = <ST as StateTransitionFunction<Vm, Da>>::StateRoot;
 /// Contains previous height, latest finalized block and fee rate.
 type L1Data<Da> = (<Da as DaService>::FilteredBlock, u128);
 
-pub struct CitreaSequencer<C, Da, Sm, Vm, Stf>
+pub struct CitreaSequencer<C, Da, Sm, Vm, Stf, DB>
 where
     C: Context,
     Da: DaService,
@@ -72,6 +72,7 @@ where
     Vm: ZkvmHost,
     Stf: StateTransitionFunction<Vm, Da::Spec, Condition = <Da::Spec as DaSpec>::ValidityCondition>
         + StfBlueprintTrait<C, Da::Spec, Vm>,
+    DB: SequencerLedgerOps + Send + Clone + 'static,
 {
     da_service: Da,
     mempool: Arc<CitreaMempool<C>>,
@@ -80,7 +81,7 @@ where
     l2_force_block_rx: UnboundedReceiver<()>,
     db_provider: DbProvider<C>,
     storage: C::Storage,
-    ledger_db: LedgerDB,
+    ledger_db: DB,
     config: SequencerConfig,
     stf: Stf,
     deposit_mempool: Arc<Mutex<DepositDataMempool>>,
@@ -99,7 +100,7 @@ enum L2BlockMode {
     NotEmpty,
 }
 
-impl<C, Da, Sm, Vm, Stf> CitreaSequencer<C, Da, Sm, Vm, Stf>
+impl<C, Da, Sm, Vm, Stf, DB> CitreaSequencer<C, Da, Sm, Vm, Stf, DB>
 where
     C: Context,
     Da: DaService + Clone,
@@ -112,6 +113,7 @@ where
             PreState = Sm::NativeStorage,
             ChangeSet = Sm::NativeChangeSet,
         > + StfBlueprintTrait<C, Da::Spec, Vm>,
+    DB: SequencerLedgerOps + Send + Clone + 'static,
 {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
@@ -122,7 +124,7 @@ where
         mut storage_manager: Sm,
         init_variant: InitVariant<Stf, Vm, Da::Spec>,
         public_keys: RollupPublicKeys,
-        ledger_db: LedgerDB,
+        ledger_db: DB,
         rpc_config: RpcConfig,
         soft_confirmation_tx: broadcast::Sender<u64>,
     ) -> anyhow::Result<Self> {
diff --git a/crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/mod.rs b/crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/mod.rs
index 40833dbef..4fd35b811 100644
--- a/crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/mod.rs
+++ b/crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/mod.rs
@@ -26,6 +26,9 @@ use crate::schema::types::{
 };
 
 mod rpc;
+mod traits;
+
+pub use traits::*;
 
 const LEDGER_DB_PATH_SUFFIX: &str = "ledger";
 
@@ -128,69 +131,6 @@ impl LedgerDB {
         })
     }
 
-    /// Get the next slot, block, transaction, and event numbers
-    #[instrument(level = "trace", skip(self), ret)]
-    pub fn get_next_items_numbers(&self) -> ItemNumbers {
-        self.next_item_numbers.lock().unwrap().clone()
-    }
-
-    /// Gets all slots with numbers `range.start` to `range.end`. If `range.end` is outside
-    /// the range of the database, the result will smaller than the requested range.
-    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
-    /// directly via rpc.
-    #[instrument(level = "trace", skip(self), err)]
-    pub(crate) fn _get_slot_range(
-        &self,
-        range: &std::ops::Range<SlotNumber>,
-    ) -> Result<Vec<StoredSlot>, anyhow::Error> {
-        self.get_data_range::<SlotByNumber, _, _>(range)
-    }
-
-    /// Gets all batches with numbers `range.start` to `range.end`. If `range.end` is outside
-    /// the range of the database, the result will smaller than the requested range.
-    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
-    /// directly via rpc.
-    #[instrument(level = "trace", skip(self), err)]
-    pub(crate) fn get_batch_range(
-        &self,
-        range: &std::ops::Range<BatchNumber>,
-    ) -> Result<Vec<StoredBatch>, anyhow::Error> {
-        self.get_data_range::<BatchByNumber, _, _>(range)
-    }
-
-    /// Gets all soft confirmations by numbers
-    #[instrument(level = "trace", skip(self), err)]
-    pub fn get_soft_batch_by_number(
-        &self,
-        number: &BatchNumber,
-    ) -> Result<Option<StoredSoftBatch>, anyhow::Error> {
-        self.db.get::<SoftBatchByNumber>(number)
-    }
-
-    /// Gets all soft confirmations with numbers `range.start` to `range.end`. If `range.end` is outside
-    /// the range of the database, the result will smaller than the requested range.
-    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
-    /// directly via rpc.
-    #[instrument(level = "trace", skip(self), err)]
-    pub fn get_soft_batch_range(
-        &self,
-        range: &std::ops::Range<BatchNumber>,
-    ) -> Result<Vec<StoredSoftBatch>, anyhow::Error> {
-        self.get_data_range::<SoftBatchByNumber, _, _>(range)
-    }
-
-    /// Gets all transactions with numbers `range.start` to `range.end`. If `range.end` is outside
-    /// the range of the database, the result will smaller than the requested range.
-    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
-    /// directly via rpc.
-    #[instrument(level = "trace", skip(self), err)]
-    pub(crate) fn get_tx_range(
-        &self,
-        range: &std::ops::Range<TxNumber>,
-    ) -> Result<Vec<StoredTransaction>, anyhow::Error> {
-        self.get_data_range::<TxByNumber, _, _>(range)
-    }
-
     /// Gets all data with identifier in `range.start` to `range.end`. If `range.end` is outside
     /// the range of the database, the result will smaller than the requested range.
     /// Note that this method blindly preallocates for the requested range, so it should not be exposed
@@ -213,17 +153,22 @@ impl LedgerDB {
         Ok(out)
     }
 
-    #[instrument(level = "trace", skip(self, schema_batch), err, ret)]
-    fn put_slot(
-        &self,
-        slot: &StoredSlot,
-        slot_number: &SlotNumber,
-        schema_batch: &mut SchemaBatch,
-    ) -> Result<(), anyhow::Error> {
-        schema_batch.put::<SlotByNumber>(slot_number, slot)?;
-        schema_batch.put::<SlotByHash>(&slot.hash, slot_number)
+    fn last_version_written<T: Schema<Key = U>, U: Into<u64>>(
+        db: &DB,
+        _schema: T,
+    ) -> anyhow::Result<Option<u64>> {
+        let mut iter = db.iter::<T>()?;
+        iter.seek_to_last();
+
+        match iter.next() {
+            Some(Ok(item)) => Ok(Some(item.key.into())),
+            Some(Err(e)) => Err(e),
+            _ => Ok(None),
+        }
     }
+}
 
+impl SharedLedgerOps for LedgerDB {
     #[instrument(level = "trace", skip(self, schema_batch), err, ret)]
     fn put_soft_batch(
         &self,
@@ -270,7 +215,7 @@ impl LedgerDB {
     }
 
     /// Commits a soft batch to the database by inserting its transactions and batches before
-    pub fn commit_soft_batch<B: Serialize, T: Serialize, DS: DaSpec>(
+    fn commit_soft_batch<B: Serialize, T: Serialize, DS: DaSpec>(
         &self,
         batch_receipt: SoftBatchReceipt<B, T, DS>,
         include_tx_body: bool,
@@ -357,93 +302,9 @@ impl LedgerDB {
         Ok(())
     }
 
-    /// Commits a slot to the database by inserting its events, transactions, and batches before
-    /// inserting the slot metadata.
-    #[instrument(level = "trace", skip_all, err, ret)]
-    pub fn commit_slot<S: SlotData, B: Serialize, T: Serialize>(
-        &self,
-        data_to_commit: SlotCommit<S, B, T>,
-    ) -> Result<(), anyhow::Error> {
-        // Create a scope to ensure that the lock is released before we commit to the db
-        let mut current_item_numbers = {
-            let mut next_item_numbers = self.next_item_numbers.lock().unwrap();
-            let item_numbers = next_item_numbers.clone();
-            next_item_numbers.slot_number += 1;
-            next_item_numbers.batch_number += data_to_commit.batch_receipts.len() as u64;
-            next_item_numbers.tx_number += data_to_commit.num_txs as u64;
-            next_item_numbers.event_number += data_to_commit.num_events as u64;
-            item_numbers
-            // The lock is released here
-        };
-
-        let mut schema_batch = SchemaBatch::new();
-
-        let first_batch_number = current_item_numbers.batch_number;
-        let last_batch_number = first_batch_number + data_to_commit.batch_receipts.len() as u64;
-        // Insert data from "bottom up" to ensure consistency if the application crashes during insertion
-        for batch_receipt in data_to_commit.batch_receipts.into_iter() {
-            let first_tx_number = current_item_numbers.tx_number;
-            let last_tx_number = first_tx_number + batch_receipt.tx_receipts.len() as u64;
-            // Insert transactions and events from each batch before inserting the batch
-            for tx in batch_receipt.tx_receipts.into_iter() {
-                let (tx_to_store, events) =
-                    split_tx_for_storage(tx, current_item_numbers.event_number);
-                for event in events.into_iter() {
-                    self.put_event(
-                        &event,
-                        &EventNumber(current_item_numbers.event_number),
-                        TxNumber(current_item_numbers.tx_number),
-                        &mut schema_batch,
-                    )?;
-                    current_item_numbers.event_number += 1;
-                }
-                self.put_transaction(
-                    &tx_to_store,
-                    &TxNumber(current_item_numbers.tx_number),
-                    &mut schema_batch,
-                )?;
-                current_item_numbers.tx_number += 1;
-            }
-
-            // Insert batch
-            let batch_to_store = StoredBatch {
-                hash: batch_receipt.hash,
-                txs: TxNumber(first_tx_number)..TxNumber(last_tx_number),
-            };
-            self.put_batch(
-                &batch_to_store,
-                &BatchNumber(current_item_numbers.batch_number),
-                &mut schema_batch,
-            )?;
-            current_item_numbers.batch_number += 1;
-        }
-
-        // Once all batches are inserted, Insert slot
-        let slot_to_store = StoredSlot {
-            hash: data_to_commit.slot_data.hash(),
-            // TODO: Add a method to the slot data trait allowing additional data to be stored
-            extra_data: vec![].into(),
-            batches: BatchNumber(first_batch_number)..BatchNumber(last_batch_number),
-        };
-        self.put_slot(
-            &slot_to_store,
-            &SlotNumber(current_item_numbers.slot_number),
-            &mut schema_batch,
-        )?;
-
-        self.db.write_schemas(schema_batch)?;
-
-        // Notify subscribers. This call returns an error IFF there are no subscribers, so we don't need to check the result
-        let _ = self
-            .slot_subscriptions
-            .send(current_item_numbers.slot_number);
-
-        Ok(())
-    }
-
     /// Records the L2 height that was created as a soft confirmaiton of an L1 height
     #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn extend_l2_range_of_l1_slot(
+    fn extend_l2_range_of_l1_slot(
         &self,
         l1_height: SlotNumber,
         l2_height: BatchNumber,
@@ -457,77 +318,114 @@ impl LedgerDB {
 
         let mut schema_batch = SchemaBatch::new();
 
-        schema_batch
-            .put::<L2RangeByL1Height>(&l1_height, &new_range)
-            .unwrap();
+        schema_batch.put::<L2RangeByL1Height>(&l1_height, &new_range)?;
         self.db.write_schemas(schema_batch)?;
 
         Ok(())
     }
 
-    /// Used by the sequencer to record that it has committed to soft confirmations on a given L2 height
-    #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn set_last_sequencer_commitment_l2_height(
+    /// Get the next slot, block, transaction, and event numbers
+    #[instrument(level = "trace", skip(self), ret)]
+    fn get_next_items_numbers(&self) -> ItemNumbers {
+        self.next_item_numbers.lock().unwrap().clone()
+    }
+
+    /// Gets all slots with numbers `range.start` to `range.end`. If `range.end` is outside
+    /// the range of the database, the result will smaller than the requested range.
+    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
+    /// directly via rpc.
+    #[instrument(level = "trace", skip(self), err)]
+    fn _get_slot_range(
         &self,
-        l2_height: BatchNumber,
-    ) -> Result<(), anyhow::Error> {
-        let mut schema_batch = SchemaBatch::new();
+        range: &std::ops::Range<SlotNumber>,
+    ) -> Result<Vec<StoredSlot>, anyhow::Error> {
+        self.get_data_range::<SlotByNumber, _, _>(range)
+    }
 
-        schema_batch
-            .put::<LastSequencerCommitmentSent>(&(), &l2_height)
-            .unwrap();
-        self.db.write_schemas(schema_batch)?;
+    /// Gets all batches with numbers `range.start` to `range.end`. If `range.end` is outside
+    /// the range of the database, the result will smaller than the requested range.
+    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
+    /// directly via rpc.
+    #[instrument(level = "trace", skip(self), err)]
+    fn get_batch_range(
+        &self,
+        range: &std::ops::Range<BatchNumber>,
+    ) -> Result<Vec<StoredBatch>, anyhow::Error> {
+        self.get_data_range::<BatchByNumber, _, _>(range)
+    }
 
-        Ok(())
+    /// Gets l1 height of l1 hash
+    #[instrument(level = "trace", skip(self), err, ret)]
+    fn get_state_diff(&self) -> Result<StateDiff, anyhow::Error> {
+        self.db
+            .get::<LastStateDiff>(&())
+            .map(|diff| diff.unwrap_or_default())
+    }
+
+    /// Sets l1 height of l1 hash
+    #[instrument(level = "trace", skip(self), err, ret)]
+    fn set_l1_height_of_l1_hash(&self, hash: [u8; 32], height: u64) -> anyhow::Result<()> {
+        self.db.put::<SlotByHash>(&hash, &SlotNumber(height))
     }
 
     /// Saves a soft confirmation status for a given L1 height
     #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn put_soft_confirmation_status(
+    fn put_soft_confirmation_status(
         &self,
         height: BatchNumber,
         status: sov_rollup_interface::rpc::SoftConfirmationStatus,
     ) -> Result<(), anyhow::Error> {
         let mut schema_batch = SchemaBatch::new();
 
-        schema_batch
-            .put::<SoftConfirmationStatus>(&height, &status)
-            .unwrap();
+        schema_batch.put::<SoftConfirmationStatus>(&height, &status)?;
         self.db.write_schemas(schema_batch)?;
 
         Ok(())
     }
 
-    fn last_version_written<T: Schema<Key = U>, U: Into<u64>>(
-        db: &DB,
-        _schema: T,
-    ) -> anyhow::Result<Option<u64>> {
-        let mut iter = db.iter::<T>()?;
-        iter.seek_to_last();
+    /// Gets the commitments in the da slot with given height if any
+    /// Adds the new coming commitment info
+    #[instrument(level = "trace", skip(self, commitment), err, ret)]
+    fn update_commitments_on_da_slot(
+        &self,
+        height: u64,
+        commitment: SequencerCommitment,
+    ) -> anyhow::Result<()> {
+        // get commitments
+        let commitments = self.db.get::<CommitmentsByNumber>(&SlotNumber(height))?;
 
-        match iter.next() {
-            Some(Ok(item)) => Ok(Some(item.key.into())),
-            Some(Err(e)) => Err(e),
-            _ => Ok(None),
+        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]),
         }
     }
 
-    /// Get the most recent committed slot, if any
-    #[instrument(level = "trace", skip(self), err)]
-    pub fn get_head_slot(&self) -> anyhow::Result<Option<(SlotNumber, StoredSlot)>> {
-        let mut iter = self.db.iter::<SlotByNumber>()?;
-        iter.seek_to_last();
+    /// Set the genesis state root
+    #[instrument(level = "trace", skip_all, err, ret)]
+    fn set_l2_genesis_state_root<StateRoot: Serialize>(
+        &self,
+        state_root: &StateRoot,
+    ) -> anyhow::Result<()> {
+        let buf = bincode::serialize(state_root)?;
+        let mut schema_batch = SchemaBatch::new();
+        schema_batch.put::<L2GenesisStateRoot>(&(), &buf)?;
 
-        match iter.next() {
-            Some(Ok(item)) => Ok(Some(item.into_tuple())),
-            Some(Err(e)) => Err(e),
-            _ => Ok(None),
-        }
+        self.db.write_schemas(schema_batch)?;
+
+        Ok(())
     }
 
     /// Get the most recent committed soft batch, if any
     #[instrument(level = "trace", skip(self), err)]
-    pub fn get_head_soft_batch(&self) -> anyhow::Result<Option<(BatchNumber, StoredSoftBatch)>> {
+    fn get_head_soft_batch(&self) -> anyhow::Result<Option<(BatchNumber, StoredSoftBatch)>> {
         let mut iter = self.db.iter::<SoftBatchByNumber>()?;
         iter.seek_to_last();
 
@@ -538,71 +436,61 @@ impl LedgerDB {
         }
     }
 
-    /// Gets all pending commitments' l2 ranges.
-    /// Returns start-end L2 heights.
-    #[instrument(level = "trace", skip(self), err)]
-    pub fn get_pending_commitments_l2_range(&self) -> anyhow::Result<Vec<L2HeightRange>> {
-        let mut iter = self.db.iter::<PendingSequencerCommitmentL2Range>()?;
-        iter.seek_to_first();
-
-        let mut l2_ranges = iter
-            .map(|item| item.map(|item| item.key))
-            .collect::<Result<Vec<_>, _>>()?;
-        // Sort ascending
-        l2_ranges.sort();
-
-        Ok(l2_ranges)
-    }
-
-    /// Put a pending commitment l2 range
+    /// Gets all soft confirmations with numbers `range.start` to `range.end`. If `range.end` is outside
+    /// the range of the database, the result will smaller than the requested range.
+    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
+    /// directly via rpc.
     #[instrument(level = "trace", skip(self), err)]
-    pub fn put_pending_commitment_l2_range(&self, l2_range: &L2HeightRange) -> anyhow::Result<()> {
-        self.db
-            .put::<PendingSequencerCommitmentL2Range>(l2_range, &())
+    fn get_soft_batch_range(
+        &self,
+        range: &std::ops::Range<BatchNumber>,
+    ) -> Result<Vec<StoredSoftBatch>, anyhow::Error> {
+        self.get_data_range::<SoftBatchByNumber, _, _>(range)
     }
 
-    /// Delete a pending commitment l2 range
+    /// Gets all soft confirmations by numbers
     #[instrument(level = "trace", skip(self), err)]
-    pub fn delete_pending_commitment_l2_range(
+    fn get_soft_batch_by_number(
         &self,
-        l2_range: &L2HeightRange,
-    ) -> anyhow::Result<()> {
-        self.db
-            .delete::<PendingSequencerCommitmentL2Range>(l2_range)
-    }
-
-    /// Get the most recent committed batch
-    /// Returns L2 height.
-    #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn get_last_sequencer_commitment_l2_height(&self) -> anyhow::Result<Option<BatchNumber>> {
-        self.db.get::<LastSequencerCommitmentSent>(&())
+        number: &BatchNumber,
+    ) -> Result<Option<StoredSoftBatch>, anyhow::Error> {
+        self.db.get::<SoftBatchByNumber>(number)
     }
+}
 
-    /// Get L2 height range for a given L1 height.
-    /// This means L2 heights in that range were soft confirmations for L1 height.
-    #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn get_l2_range_by_l1_height(
+impl ProverLedgerOps for LedgerDB {
+    /// Get the state root by L2 height
+    #[instrument(level = "trace", skip_all, err)]
+    fn get_l2_state_root<StateRoot: DeserializeOwned>(
         &self,
-        l1_height: SlotNumber,
-    ) -> anyhow::Result<Option<L2HeightRange>> {
-        self.db.get::<L2RangeByL1Height>(&l1_height)
+        l2_height: u64,
+    ) -> anyhow::Result<Option<StateRoot>> {
+        if l2_height == 0 {
+            self.db
+                .get::<L2GenesisStateRoot>(&())?
+                .map(|state_root| bincode::deserialize(&state_root).map_err(Into::into))
+                .transpose()
+        } else {
+            self.db
+                .get::<SoftBatchByNumber>(&BatchNumber(l2_height))?
+                .map(|soft_batch| bincode::deserialize(&soft_batch.state_root).map_err(Into::into))
+                .transpose()
+        }
     }
 
     /// Get the last scanned slot by the prover
     #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn get_prover_last_scanned_l1_height(&self) -> anyhow::Result<Option<SlotNumber>> {
+    fn get_prover_last_scanned_l1_height(&self) -> anyhow::Result<Option<SlotNumber>> {
         self.db.get::<ProverLastScannedSlot>(&())
     }
 
     /// Set the last scanned slot by the prover
     /// Called by the prover.
     #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn set_prover_last_scanned_l1_height(&self, l1_height: SlotNumber) -> anyhow::Result<()> {
+    fn set_prover_last_scanned_l1_height(&self, l1_height: SlotNumber) -> anyhow::Result<()> {
         let mut schema_batch = SchemaBatch::new();
 
-        schema_batch
-            .put::<ProverLastScannedSlot>(&(), &l1_height)
-            .unwrap();
+        schema_batch.put::<ProverLastScannedSlot>(&(), &l1_height)?;
         self.db.write_schemas(schema_batch)?;
 
         Ok(())
@@ -610,7 +498,7 @@ impl LedgerDB {
 
     /// Get the witness by L2 height
     #[instrument(level = "trace", skip_all, err)]
-    pub fn get_l2_witness<Witness: DeserializeOwned>(
+    fn get_l2_witness<Witness: DeserializeOwned>(
         &self,
         l2_height: u64,
     ) -> anyhow::Result<Option<Witness>> {
@@ -623,9 +511,27 @@ impl LedgerDB {
         }
     }
 
+    /// Stores proof related data on disk, accessible via l1 slot height
+    #[instrument(level = "trace", skip(self, proof, state_transition), err, ret)]
+    fn put_proof_data(
+        &self,
+        l1_height: u64,
+        l1_tx_id: [u8; 32],
+        proof: Proof,
+        state_transition: StoredStateTransition,
+    ) -> anyhow::Result<()> {
+        let data_to_store = StoredProof {
+            l1_tx_id,
+            proof,
+            state_transition,
+        };
+        self.db
+            .put::<ProofBySlotNumber>(&SlotNumber(l1_height), &data_to_store)
+    }
+
     /// Set the witness by L2 height
     #[instrument(level = "trace", skip_all, err, ret)]
-    pub fn set_l2_witness<Witness: Serialize>(
+    fn set_l2_witness<Witness: Serialize>(
         &self,
         l2_height: u64,
         witness: &Witness,
@@ -638,96 +544,172 @@ impl LedgerDB {
 
         Ok(())
     }
+}
 
-    /// Set the genesis state root
+impl SequencerLedgerOps for LedgerDB {
+    /// Put slots
+    #[instrument(level = "trace", skip(self, schema_batch), err, ret)]
+    fn put_slot(
+        &self,
+        slot: &StoredSlot,
+        slot_number: &SlotNumber,
+        schema_batch: &mut SchemaBatch,
+    ) -> Result<(), anyhow::Error> {
+        schema_batch.put::<SlotByNumber>(slot_number, slot)?;
+        schema_batch.put::<SlotByHash>(&slot.hash, slot_number)
+    }
+
+    /// Commits a slot to the database by inserting its events, transactions, and batches before
+    /// inserting the slot metadata.
     #[instrument(level = "trace", skip_all, err, ret)]
-    pub fn set_l2_genesis_state_root<StateRoot: Serialize>(
+    fn commit_slot<S: SlotData, B: Serialize, T: Serialize>(
         &self,
-        state_root: &StateRoot,
-    ) -> anyhow::Result<()> {
-        let buf = bincode::serialize(state_root)?;
+        data_to_commit: SlotCommit<S, B, T>,
+    ) -> Result<(), anyhow::Error> {
+        // Create a scope to ensure that the lock is released before we commit to the db
+        let mut current_item_numbers = {
+            let mut next_item_numbers = self.next_item_numbers.lock().unwrap();
+            let item_numbers = next_item_numbers.clone();
+            next_item_numbers.slot_number += 1;
+            next_item_numbers.batch_number += data_to_commit.batch_receipts.len() as u64;
+            next_item_numbers.tx_number += data_to_commit.num_txs as u64;
+            next_item_numbers.event_number += data_to_commit.num_events as u64;
+            item_numbers
+            // The lock is released here
+        };
+
         let mut schema_batch = SchemaBatch::new();
-        schema_batch.put::<L2GenesisStateRoot>(&(), &buf)?;
+
+        let first_batch_number = current_item_numbers.batch_number;
+        let last_batch_number = first_batch_number + data_to_commit.batch_receipts.len() as u64;
+        // Insert data from "bottom up" to ensure consistency if the application crashes during insertion
+        for batch_receipt in data_to_commit.batch_receipts.into_iter() {
+            let first_tx_number = current_item_numbers.tx_number;
+            let last_tx_number = first_tx_number + batch_receipt.tx_receipts.len() as u64;
+            // Insert transactions and events from each batch before inserting the batch
+            for tx in batch_receipt.tx_receipts.into_iter() {
+                let (tx_to_store, events) =
+                    split_tx_for_storage(tx, current_item_numbers.event_number);
+                for event in events.into_iter() {
+                    self.put_event(
+                        &event,
+                        &EventNumber(current_item_numbers.event_number),
+                        TxNumber(current_item_numbers.tx_number),
+                        &mut schema_batch,
+                    )?;
+                    current_item_numbers.event_number += 1;
+                }
+                self.put_transaction(
+                    &tx_to_store,
+                    &TxNumber(current_item_numbers.tx_number),
+                    &mut schema_batch,
+                )?;
+                current_item_numbers.tx_number += 1;
+            }
+
+            // Insert batch
+            let batch_to_store = StoredBatch {
+                hash: batch_receipt.hash,
+                txs: TxNumber(first_tx_number)..TxNumber(last_tx_number),
+            };
+            self.put_batch(
+                &batch_to_store,
+                &BatchNumber(current_item_numbers.batch_number),
+                &mut schema_batch,
+            )?;
+            current_item_numbers.batch_number += 1;
+        }
+
+        // Once all batches are inserted, Insert slot
+        let slot_to_store = StoredSlot {
+            hash: data_to_commit.slot_data.hash(),
+            // TODO: Add a method to the slot data trait allowing additional data to be stored
+            extra_data: vec![].into(),
+            batches: BatchNumber(first_batch_number)..BatchNumber(last_batch_number),
+        };
+        self.put_slot(
+            &slot_to_store,
+            &SlotNumber(current_item_numbers.slot_number),
+            &mut schema_batch,
+        )?;
 
         self.db.write_schemas(schema_batch)?;
 
+        // Notify subscribers. This call returns an error IFF there are no subscribers, so we don't need to check the result
+        let _ = self
+            .slot_subscriptions
+            .send(current_item_numbers.slot_number);
+
         Ok(())
     }
 
-    /// Get the state root by L2 height
-    #[instrument(level = "trace", skip_all, err)]
-    pub fn get_l2_state_root<StateRoot: DeserializeOwned>(
+    /// Used by the sequencer to record that it has committed to soft confirmations on a given L2 height
+    #[instrument(level = "trace", skip(self), err, ret)]
+    fn set_last_sequencer_commitment_l2_height(
         &self,
-        l2_height: u64,
-    ) -> anyhow::Result<Option<StateRoot>> {
-        if l2_height == 0 {
-            self.db
-                .get::<L2GenesisStateRoot>(&())?
-                .map(|state_root| bincode::deserialize(&state_root).map_err(Into::into))
-                .transpose()
-        } else {
-            self.db
-                .get::<SoftBatchByNumber>(&BatchNumber(l2_height))?
-                .map(|soft_batch| bincode::deserialize(&soft_batch.state_root).map_err(Into::into))
-                .transpose()
-        }
+        l2_height: BatchNumber,
+    ) -> Result<(), anyhow::Error> {
+        let mut schema_batch = SchemaBatch::new();
+
+        schema_batch.put::<LastSequencerCommitmentSent>(&(), &l2_height)?;
+        self.db.write_schemas(schema_batch)?;
+
+        Ok(())
     }
 
-    /// Gets the commitments in the da slot with given height if any
-    /// Adds the new coming commitment info
-    #[instrument(level = "trace", skip(self, commitment), err, ret)]
-    pub fn update_commitments_on_da_slot(
-        &self,
-        height: u64,
-        commitment: SequencerCommitment,
-    ) -> anyhow::Result<()> {
-        // get commitments
-        let commitments = self.db.get::<CommitmentsByNumber>(&SlotNumber(height))?;
+    /// Gets all pending commitments' l2 ranges.
+    /// Returns start-end L2 heights.
+    #[instrument(level = "trace", skip(self), err)]
+    fn get_pending_commitments_l2_range(&self) -> anyhow::Result<Vec<L2HeightRange>> {
+        let mut iter = self.db.iter::<PendingSequencerCommitmentL2Range>()?;
+        iter.seek_to_first();
 
-        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]),
-        }
+        let mut l2_ranges = iter
+            .map(|item| item.map(|item| item.key))
+            .collect::<Result<Vec<_>, _>>()?;
+        // Sort ascending
+        l2_ranges.sort();
+
+        Ok(l2_ranges)
     }
 
-    /// Gets the commitments in the da slot with given height if any
+    /// Put a pending commitment l2 range
     #[instrument(level = "trace", skip(self), err)]
-    pub fn get_commitments_on_da_slot(
-        &self,
-        height: u64,
-    ) -> anyhow::Result<Option<Vec<SequencerCommitment>>> {
-        self.db.get::<CommitmentsByNumber>(&SlotNumber(height))
+    fn put_pending_commitment_l2_range(&self, l2_range: &L2HeightRange) -> anyhow::Result<()> {
+        self.db
+            .put::<PendingSequencerCommitmentL2Range>(l2_range, &())
     }
 
-    /// Stores proof related data on disk, accessible via l1 slot height
-    #[instrument(level = "trace", skip(self, proof, state_transition), err, ret)]
-    pub fn put_proof_data(
-        &self,
-        l1_height: u64,
-        l1_tx_id: [u8; 32],
-        proof: Proof,
-        state_transition: StoredStateTransition,
-    ) -> anyhow::Result<()> {
-        let data_to_store = StoredProof {
-            l1_tx_id,
-            proof,
-            state_transition,
-        };
+    /// Delete a pending commitment l2 range
+    #[instrument(level = "trace", skip(self), err)]
+    fn delete_pending_commitment_l2_range(&self, l2_range: &L2HeightRange) -> anyhow::Result<()> {
         self.db
-            .put::<ProofBySlotNumber>(&SlotNumber(l1_height), &data_to_store)
+            .delete::<PendingSequencerCommitmentL2Range>(l2_range)
+    }
+
+    /// Sets the latest state diff
+    #[instrument(level = "trace", skip(self), err, ret)]
+    fn set_state_diff(&self, state_diff: StateDiff) -> anyhow::Result<()> {
+        let mut schema_batch = SchemaBatch::new();
+        schema_batch.put::<LastStateDiff>(&(), &state_diff)?;
+
+        self.db.write_schemas(schema_batch)?;
+
+        Ok(())
     }
 
+    /// Get the most recent committed batch
+    /// Returns L2 height.
+    #[instrument(level = "trace", skip(self), err, ret)]
+    fn get_last_sequencer_commitment_l2_height(&self) -> anyhow::Result<Option<BatchNumber>> {
+        self.db.get::<LastSequencerCommitmentSent>(&())
+    }
+}
+
+impl NodeLedgerOps for LedgerDB {
     /// Stores proof related data on disk, accessible via l1 slot height
     #[instrument(level = "trace", skip(self, proof, state_transition), err, ret)]
-    pub fn update_verified_proof_data(
+    fn update_verified_proof_data(
         &self,
         l1_height: u64,
         proof: Proof,
@@ -757,34 +739,43 @@ impl LedgerDB {
         }
     }
 
-    /// Sets l1 height of l1 hash
-    #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn set_l1_height_of_l1_hash(&self, hash: [u8; 32], height: u64) -> anyhow::Result<()> {
-        self.db.put::<SlotByHash>(&hash, &SlotNumber(height))
-    }
+    /// Get the most recent committed slot, if any
+    #[instrument(level = "trace", skip(self), err)]
+    fn get_head_slot(&self) -> anyhow::Result<Option<(SlotNumber, StoredSlot)>> {
+        let mut iter = self.db.iter::<SlotByNumber>()?;
+        iter.seek_to_last();
 
-    /// Gets l1 height of l1 hash
-    #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn get_l1_height_of_l1_hash(&self, hash: [u8; 32]) -> Result<Option<u64>, anyhow::Error> {
-        self.db.get::<SlotByHash>(&hash).map(|v| v.map(|a| a.0))
+        match iter.next() {
+            Some(Ok(item)) => Ok(Some(item.into_tuple())),
+            Some(Err(e)) => Err(e),
+            _ => Ok(None),
+        }
     }
 
-    /// Sets the latest state diff
-    #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn set_state_diff(&self, state_diff: StateDiff) -> anyhow::Result<()> {
-        let mut schema_batch = SchemaBatch::new();
-        schema_batch.put::<LastStateDiff>(&(), &state_diff)?;
-
-        self.db.write_schemas(schema_batch)?;
+    /// Gets all transactions with numbers `range.start` to `range.end`. If `range.end` is outside
+    /// the range of the database, the result will smaller than the requested range.
+    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
+    /// directly via rpc.
+    #[instrument(level = "trace", skip(self), err)]
+    fn get_tx_range(
+        &self,
+        range: &std::ops::Range<TxNumber>,
+    ) -> Result<Vec<StoredTransaction>, anyhow::Error> {
+        self.get_data_range::<TxByNumber, _, _>(range)
+    }
 
-        Ok(())
+    /// Gets the commitments in the da slot with given height if any
+    #[instrument(level = "trace", skip(self), err)]
+    fn get_commitments_on_da_slot(
+        &self,
+        height: u64,
+    ) -> anyhow::Result<Option<Vec<SequencerCommitment>>> {
+        self.db.get::<CommitmentsByNumber>(&SlotNumber(height))
     }
 
     /// Gets l1 height of l1 hash
     #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn get_state_diff(&self) -> Result<StateDiff, anyhow::Error> {
-        self.db
-            .get::<LastStateDiff>(&())
-            .map(|diff| diff.unwrap_or_default())
+    fn get_l1_height_of_l1_hash(&self, hash: [u8; 32]) -> Result<Option<u64>, anyhow::Error> {
+        self.db.get::<SlotByHash>(&hash).map(|v| v.map(|a| a.0))
     }
 }
diff --git a/crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/rpc.rs b/crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/rpc.rs
index d0a2cb951..e5b6e87f2 100644
--- a/crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/rpc.rs
+++ b/crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/rpc.rs
@@ -29,7 +29,7 @@ const MAX_TRANSACTIONS_PER_REQUEST: u64 = 100;
 /// The maximum number of events that can be requested in a single RPC range query
 const MAX_EVENTS_PER_REQUEST: u64 = 500;
 
-use super::LedgerDB;
+use super::{LedgerDB, NodeLedgerOps, ProverLedgerOps, SharedLedgerOps};
 
 impl LedgerRpcProvider for LedgerDB {
     fn get_slots<B: DeserializeOwned, T: DeserializeOwned>(
@@ -394,7 +394,7 @@ impl LedgerRpcProvider for LedgerDB {
     }
 
     fn get_prover_last_scanned_l1_height(&self) -> Result<u64, anyhow::Error> {
-        match self.get_prover_last_scanned_l1_height()? {
+        match ProverLedgerOps::get_prover_last_scanned_l1_height(self)? {
             Some(height) => Ok(height.0),
             None => Ok(0),
         }
@@ -625,7 +625,7 @@ mod tests {
     use sov_mock_da::{MockBlob, MockBlock};
     use sov_rollup_interface::rpc::LedgerRpcProvider;
 
-    use crate::ledger_db::{LedgerDB, SlotCommit};
+    use crate::ledger_db::{LedgerDB, SequencerLedgerOps, SlotCommit};
     #[test]
     fn test_slot_subscription() {
         let temp_dir = tempfile::tempdir().unwrap();
diff --git a/crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/traits.rs b/crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/traits.rs
new file mode 100644
index 000000000..33e800328
--- /dev/null
+++ b/crates/sovereign-sdk/full-node/db/sov-db/src/ledger_db/traits.rs
@@ -0,0 +1,216 @@
+use anyhow::Result;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use sov_rollup_interface::da::{DaSpec, SequencerCommitment};
+use sov_rollup_interface::services::da::SlotData;
+use sov_rollup_interface::stf::{Event, SoftBatchReceipt, StateDiff};
+use sov_rollup_interface::zk::Proof;
+use sov_schema_db::SchemaBatch;
+
+use super::{ItemNumbers, SlotCommit};
+use crate::schema::types::{
+    BatchNumber, EventNumber, L2HeightRange, SlotNumber, StoredBatch, StoredSlot, StoredSoftBatch,
+    StoredStateTransition, StoredTransaction, TxNumber,
+};
+
+/// Shared ledger operations
+pub trait SharedLedgerOps {
+    /// Put soft batch to db
+    fn put_soft_batch(
+        &self,
+        batch: &StoredSoftBatch,
+        batch_number: &BatchNumber,
+        schema_batch: &mut SchemaBatch,
+    ) -> Result<()>;
+
+    /// Put batch to db
+    fn put_batch(
+        &self,
+        batch: &StoredBatch,
+        batch_number: &BatchNumber,
+        schema_batch: &mut SchemaBatch,
+    ) -> Result<()>;
+
+    /// Put transaction to db
+    fn put_transaction(
+        &self,
+        tx: &StoredTransaction,
+        tx_number: &TxNumber,
+        schema_batch: &mut SchemaBatch,
+    ) -> Result<()>;
+
+    /// Put event to db
+    fn put_event(
+        &self,
+        event: &Event,
+        event_number: &EventNumber,
+        tx_number: TxNumber,
+        schema_batch: &mut SchemaBatch,
+    ) -> Result<()>;
+
+    /// Commits a soft batch to the database by inserting its transactions and batches before
+    fn commit_soft_batch<B: Serialize, T: Serialize, DS: DaSpec>(
+        &self,
+        batch_receipt: SoftBatchReceipt<B, T, DS>,
+        include_tx_body: bool,
+    ) -> Result<()>;
+
+    /// Records the L2 height that was created as a soft confirmaiton of an L1 height
+    fn extend_l2_range_of_l1_slot(
+        &self,
+        l1_height: SlotNumber,
+        l2_height: BatchNumber,
+    ) -> Result<()>;
+
+    /// Get the next slot, block, transaction, and event numbers
+    fn get_next_items_numbers(&self) -> ItemNumbers;
+
+    /// Gets all slots with numbers `range.start` to `range.end`. If `range.end` is outside
+    /// the range of the database, the result will smaller than the requested range.
+    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
+    /// directly via rpc.
+    fn _get_slot_range(&self, range: &std::ops::Range<SlotNumber>) -> Result<Vec<StoredSlot>>;
+
+    /// Gets all batches with numbers `range.start` to `range.end`. If `range.end` is outside
+    /// the range of the database, the result will smaller than the requested range.
+    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
+    /// directly via rpc.
+    fn get_batch_range(&self, range: &std::ops::Range<BatchNumber>) -> Result<Vec<StoredBatch>>;
+
+    /// Gets l1 height of l1 hash
+    fn get_state_diff(&self) -> Result<StateDiff>;
+
+    /// Sets l1 height of l1 hash
+    fn set_l1_height_of_l1_hash(&self, hash: [u8; 32], height: u64) -> Result<()>;
+
+    /// Saves a soft confirmation status for a given L1 height
+    fn put_soft_confirmation_status(
+        &self,
+        height: BatchNumber,
+        status: sov_rollup_interface::rpc::SoftConfirmationStatus,
+    ) -> Result<()>;
+
+    /// Gets the commitments in the da slot with given height if any
+    /// Adds the new coming commitment info
+    fn update_commitments_on_da_slot(
+        &self,
+        height: u64,
+        commitment: SequencerCommitment,
+    ) -> Result<()>;
+
+    /// Set the genesis state root
+    fn set_l2_genesis_state_root<StateRoot: Serialize>(
+        &self,
+        state_root: &StateRoot,
+    ) -> anyhow::Result<()>;
+
+    /// Get the most recent committed soft batch, if any
+    fn get_head_soft_batch(&self) -> Result<Option<(BatchNumber, StoredSoftBatch)>>;
+
+    /// Gets all soft confirmations with numbers `range.start` to `range.end`. If `range.end` is outside
+    /// the range of the database, the result will smaller than the requested range.
+    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
+    /// directly via rpc.
+    fn get_soft_batch_range(
+        &self,
+        range: &std::ops::Range<BatchNumber>,
+    ) -> Result<Vec<StoredSoftBatch>>;
+
+    /// Gets all soft confirmations by numbers
+    fn get_soft_batch_by_number(&self, number: &BatchNumber) -> Result<Option<StoredSoftBatch>>;
+}
+
+/// Node ledger operations
+pub trait NodeLedgerOps: SharedLedgerOps {
+    /// Stores proof related data on disk, accessible via l1 slot height
+    fn update_verified_proof_data(
+        &self,
+        l1_height: u64,
+        proof: Proof,
+        state_transition: StoredStateTransition,
+    ) -> Result<()>;
+
+    /// Get the most recent committed slot, if any
+    fn get_head_slot(&self) -> Result<Option<(SlotNumber, StoredSlot)>>;
+
+    /// Gets all transactions with numbers `range.start` to `range.end`. If `range.end` is outside
+    /// the range of the database, the result will smaller than the requested range.
+    /// Note that this method blindly preallocates for the requested range, so it should not be exposed
+    /// directly via rpc.
+    fn get_tx_range(&self, range: &std::ops::Range<TxNumber>) -> Result<Vec<StoredTransaction>>;
+
+    /// Gets the commitments in the da slot with given height if any
+    fn get_commitments_on_da_slot(&self, height: u64) -> Result<Option<Vec<SequencerCommitment>>>;
+
+    /// Gets l1 height of l1 hash
+    fn get_l1_height_of_l1_hash(&self, hash: [u8; 32]) -> Result<Option<u64>>;
+}
+
+/// Prover ledger operations
+pub trait ProverLedgerOps: SharedLedgerOps {
+    /// Get the state root by L2 height
+    fn get_l2_state_root<StateRoot: DeserializeOwned>(
+        &self,
+        l2_height: u64,
+    ) -> anyhow::Result<Option<StateRoot>>;
+
+    /// Get the last scanned slot by the prover
+    fn get_prover_last_scanned_l1_height(&self) -> Result<Option<SlotNumber>>;
+
+    /// Set the last scanned slot by the prover
+    /// Called by the prover.
+    fn set_prover_last_scanned_l1_height(&self, l1_height: SlotNumber) -> Result<()>;
+
+    /// Get the witness by L2 height
+    fn get_l2_witness<Witness: DeserializeOwned>(&self, l2_height: u64) -> Result<Option<Witness>>;
+
+    /// Stores proof related data on disk, accessible via l1 slot height
+    fn put_proof_data(
+        &self,
+        l1_height: u64,
+        l1_tx_id: [u8; 32],
+        proof: Proof,
+        state_transition: StoredStateTransition,
+    ) -> Result<()>;
+
+    /// Set the witness by L2 height
+    fn set_l2_witness<Witness: Serialize>(&self, l2_height: u64, witness: &Witness) -> Result<()>;
+}
+
+/// Sequencer ledger operations
+pub trait SequencerLedgerOps: SharedLedgerOps {
+    /// Put slots
+    fn put_slot(
+        &self,
+        slot: &StoredSlot,
+        slot_number: &SlotNumber,
+        schema_batch: &mut SchemaBatch,
+    ) -> Result<()>;
+
+    /// Commits a slot to the database by inserting its events, transactions, and batches before
+    /// inserting the slot metadata.
+    fn commit_slot<S: SlotData, B: Serialize, T: Serialize>(
+        &self,
+        data_to_commit: SlotCommit<S, B, T>,
+    ) -> Result<()>;
+
+    /// Used by the sequencer to record that it has committed to soft confirmations on a given L2 height
+    fn set_last_sequencer_commitment_l2_height(&self, l2_height: BatchNumber) -> Result<()>;
+
+    /// Gets all pending commitments' l2 ranges.
+    /// Returns start-end L2 heights.
+    fn get_pending_commitments_l2_range(&self) -> Result<Vec<L2HeightRange>>;
+
+    /// Put a pending commitment l2 range
+    fn put_pending_commitment_l2_range(&self, l2_range: &L2HeightRange) -> Result<()>;
+
+    /// Delete a pending commitment l2 range
+    fn delete_pending_commitment_l2_range(&self, l2_range: &L2HeightRange) -> Result<()>;
+
+    /// Sets the latest state diff
+    fn set_state_diff(&self, state_diff: StateDiff) -> Result<()>;
+
+    /// Get the most recent committed batch
+    /// Returns L2 height.
+    fn get_last_sequencer_commitment_l2_height(&self) -> anyhow::Result<Option<BatchNumber>>;
+}