Remove TransactionLedgerDB

Author: kpp

Cargo.lock

@@ -25,7 +25,6 @@ anyhow = { workspace = true, default-features = true }
 bincode = { workspace = true }
 borsh = { workspace = true, default-features = true, features = ["bytes", "rc"] }
 byteorder = { workspace = true, default-features = true }
-metrics = { workspace = true }
 rlimit = { workspace = true }
 rocksdb = { workspace = true }
 serde = { workspace = true, default-features = true, features = ["rc"] }

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

@@ -1,11 +1,7 @@
-use std::mem::ManuallyDrop;
 use std::ops::RangeInclusive;
 use std::path::Path;
 use std::sync::Arc;
-use std::time::Instant;
 
-use anyhow::Context;
-use metrics::histogram;
 use rocksdb::ReadOptions;
 use sov_rollup_interface::block::L2Block;
 use sov_rollup_interface::da::SequencerCommitment;
@@ -14,9 +10,7 @@ use sov_rollup_interface::stf::StateDiff;
 use sov_rollup_interface::zk::{Proof, StorageRootHash};
 use sov_schema_db::schema::{KeyCodec, ValueCodec};
 pub use sov_schema_db::SchemaBatch;
-use sov_schema_db::{
-    ScanDirection, Schema, SchemaIterator, SchemaIteratorTx, SeekKeyEncoder, TransactionDB, DB,
-};
+use sov_schema_db::{ScanDirection, Schema, SchemaIterator, SeekKeyEncoder, DB};
 use tracing::instrument;
 use uuid::Uuid;
 
@@ -786,270 +780,3 @@ impl ForkMigration for LedgerDB {
         Ok(())
     }
 }
-
-/// A transaction for batching multiple ledger operations together.
-pub struct LedgerTx {
-    /// The batch of schema changes to apply.
-    /// Using ManuallyDrop to avoid silent drop of SchemaBatch which would lose the changes.
-    /// The batch must be explicitly written to the DB using `::commit` method.
-    batch: ManuallyDrop<SchemaBatch>,
-}
-
-impl Default for LedgerTx {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl LedgerTx {
-    /// Create a new ledger transaction.
-    pub fn new() -> Self {
-        Self {
-            batch: ManuallyDrop::new(SchemaBatch::new()),
-        }
-    }
-
-    /// Put a state diff into the transaction to be saved.
-    pub fn put_state_diff(
-        &mut self,
-        l2_height: L2BlockNumber,
-        state_diff: &StateDiff,
-    ) -> Result<&mut Self, anyhow::Error> {
-        self.batch
-            .put::<StateDiffByBlockNumber>(&l2_height, state_diff)
-            .context("Failed to add StateDiffByBlockNumber")?;
-        Ok(self)
-    }
-
-    /// Put an L2 block into the transaction to be saved.
-    pub fn put_l2_block(&mut self, l2_block: &StoredL2Block) -> Result<&mut Self, anyhow::Error> {
-        let l2_block_number = L2BlockNumber(l2_block.height);
-        self.batch
-            .put::<L2BlockByNumber>(&l2_block_number, l2_block)
-            .context("Failed to add L2BlockByNumber")?;
-        self.batch
-            .put::<L2BlockByHash>(&l2_block.hash, &l2_block_number)
-            .context("Failed to add L2BlockByHash")?;
-
-        Ok(self)
-    }
-
-    /// Commit the transaction to the given ledger DB.
-    #[must_use = "LedgerTx must be committed to apply changes"]
-    pub fn commit(self, db: &LedgerDB) -> Result<(), anyhow::Error> {
-        let Self { batch } = self;
-        let batch = ManuallyDrop::into_inner(batch);
-        db.db
-            .write_schemas(batch)
-            .context("Failed to write LedgerTx to DB")
-    }
-
-    /// Reject the transaction, dropping all changes.
-    pub fn reject(self) {
-        let Self { batch } = self;
-        let _ = ManuallyDrop::into_inner(batch);
-    }
-}
-
-#[derive(Clone)]
-/// An instance of the ledger database capable of transactional operations.
-pub struct TransactionLedgerDB {
-    /// The underlying database instance.
-    pub(crate) db: Arc<TransactionDB>,
-}
-
-/// A transaction for batching multiple ledger operations together.
-pub struct LedgerDBTransaction<'a> {
-    db: Arc<TransactionDB>,
-    tx: rocksdb::Transaction<'a, rocksdb::TransactionDB>,
-}
-
-impl TransactionLedgerDB {
-    /// LedgerDB path suffix
-    pub const DB_PATH_SUFFIX: &'static str = "ledger";
-
-    /// Open a [`LedgerDB`] (backed by RocksDB) at the specified path.
-    /// Will take optional column families, used for migration purposes.
-    /// The returned instance will be at the path `{path}/ledger`.
-    #[instrument(level = "trace", skip_all, err)]
-    pub fn with_config(cfg: &RocksdbConfig) -> Result<Self, anyhow::Error> {
-        let path = cfg.path.join(Self::DB_PATH_SUFFIX);
-        let raw_options = cfg.as_raw_options(false);
-        let tables = cfg
-            .column_families
-            .clone()
-            .unwrap_or_else(|| LEDGER_TABLES.iter().map(|e| e.to_string()).collect());
-        let inner = DB::open_transaction_db(path, tables, &raw_options)?;
-
-        Ok(Self {
-            db: Arc::new(inner),
-        })
-    }
-
-    /// Create a new transaction
-    pub fn transaction(&self) -> LedgerDBTransaction {
-        LedgerDBTransaction {
-            db: Arc::clone(&self.db),
-            tx: self.db.transaction(),
-        }
-    }
-}
-
-/// Low-level operations
-/// 1. to put/get/delete data using KeyCodec/ValueCodec.
-/// 2. to commit the transaction.
-impl LedgerDBTransaction<'_> {
-    /// Put an instance by key and value.
-    pub fn put<S: Schema>(
-        &self,
-        key: &impl KeyCodec<S>,
-        value: &impl ValueCodec<S>,
-    ) -> anyhow::Result<()> {
-        let start = Instant::now();
-
-        let cf_handle = self.get_cf_handle(S::COLUMN_FAMILY_NAME)?;
-        let key = key.encode_key()?;
-        let value = value.encode_value()?;
-
-        self.tx.put_cf(cf_handle, key, value)?;
-
-        histogram!("ledger_tx_put_latency_seconds").record(
-            Instant::now()
-                .saturating_duration_since(start)
-                .as_secs_f64(),
-        );
-        Ok(())
-    }
-
-    /// Returns the handle for a rocksdb column family.
-    fn get_cf_handle(&self, cf_name: &str) -> anyhow::Result<&rocksdb::ColumnFamily> {
-        self.db.get_cf_handle(cf_name)
-    }
-
-    /// Get an instance by key.
-    pub fn get<S: Schema>(
-        &self,
-        schema_key: &impl KeyCodec<S>,
-    ) -> anyhow::Result<Option<S::Value>> {
-        let start = Instant::now();
-
-        let k = schema_key.encode_key()?;
-        let cf_handle = self.get_cf_handle(S::COLUMN_FAMILY_NAME)?;
-
-        let result = self.tx.get_pinned_cf(cf_handle, k)?;
-
-        histogram!("schemadb_get_bytes", "cf_name" => S::COLUMN_FAMILY_NAME)
-            .record(result.as_ref().map_or(0.0, |v| v.len() as f64));
-
-        let result = result
-            .map(|raw_value| S::Value::decode_value(&raw_value))
-            .transpose()
-            .map_err(|err| err.into());
-
-        histogram!("schemadb_get_latency_seconds", "cf_name" => S::COLUMN_FAMILY_NAME).record(
-            Instant::now()
-                .saturating_duration_since(start)
-                .as_secs_f64(),
-        );
-        result
-    }
-
-    /// Delete an instance by key.
-    pub fn delete<S: Schema>(&self, key: &impl KeyCodec<S>) -> anyhow::Result<()> {
-        let cf_handle = self.get_cf_handle(S::COLUMN_FAMILY_NAME)?;
-        let key = key.encode_key()?;
-        self.tx.delete_cf(cf_handle, key)?;
-
-        Ok(())
-    }
-
-    /// Returns a [`SchemaIteratorTx`] on a certain schema with the provided read options and direction.
-    pub fn iter_with_direction<S: Schema>(
-        &self,
-        opts: ReadOptions,
-        direction: ScanDirection,
-    ) -> anyhow::Result<SchemaIteratorTx<S>> {
-        let cf_handle = self.get_cf_handle(S::COLUMN_FAMILY_NAME)?;
-        Ok(SchemaIteratorTx::new(
-            self.tx.raw_iterator_cf_opt(cf_handle, opts),
-            direction,
-        ))
-    }
-
-    /// Returns a forward [`SchemaIteratorTx`] on a certain schema with the default read options.
-    pub fn iter<S: Schema>(&self) -> anyhow::Result<SchemaIteratorTx<S>> {
-        let mut read_options = ReadOptions::default();
-        read_options.set_async_io(true);
-        self.iter_with_direction::<S>(read_options, ScanDirection::Forward)
-    }
-
-    /// Commit the transaction
-    pub fn commit(self) -> anyhow::Result<()> {
-        let start = Instant::now();
-        self.tx.commit()?;
-
-        histogram!("ledger_tx_commit_latency_seconds").record(
-            Instant::now()
-                .saturating_duration_since(start)
-                .as_secs_f64(),
-        );
-        Ok(())
-    }
-}
-
-/// Implementation of high-level ledger operations for LedgerDBTransaction.
-/// TODO: consider removing it and use raw put/get/delete methods instead.
-impl LedgerDBTransaction<'_> {
-    /// Sets the state diff by block number
-    #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn set_state_diff(
-        &self,
-        l2_height: L2BlockNumber,
-        state_diff: &StateDiff,
-    ) -> anyhow::Result<()> {
-        self.put::<StateDiffByBlockNumber>(&l2_height, state_diff)?;
-
-        Ok(())
-    }
-
-    /// Removes the state diff by block range
-    #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn delete_state_diff_by_range(
-        &self,
-        l2_height_range: RangeInclusive<L2BlockNumber>,
-    ) -> anyhow::Result<()> {
-        for l2_height in l2_height_range.start().0..=l2_height_range.end().0 {
-            self.delete::<StateDiffByBlockNumber>(&L2BlockNumber(l2_height))?;
-        }
-
-        Ok(())
-    }
-
-    /// Gets the state diff by block number
-    #[instrument(level = "trace", skip(self), err, ret)]
-    pub fn get_state_diff(&self, l2_height: L2BlockNumber) -> Result<StateDiff, anyhow::Error> {
-        self.get::<StateDiffByBlockNumber>(&l2_height)
-            .map(|diff| diff.unwrap_or_default())
-    }
-
-    /// Put an L2 block into the transaction to be saved.
-    #[instrument(level = "trace", skip(self), err)]
-    pub fn set_l2_block(&self, l2_block: &StoredL2Block) -> Result<(), anyhow::Error> {
-        let l2_block_number = L2BlockNumber(l2_block.height);
-        self.put::<L2BlockByNumber>(&l2_block_number, l2_block)
-            .context("Failed to add L2BlockByNumber")?;
-        self.put::<L2BlockByHash>(&l2_block.hash, &l2_block_number)
-            .context("Failed to add L2BlockByHash")?;
-
-        Ok(())
-    }
-
-    /// Gets l2 block by number
-    #[instrument(level = "trace", skip(self), err)]
-    fn get_l2_block_by_number(
-        &self,
-        number: &L2BlockNumber,
-    ) -> Result<Option<StoredL2Block>, anyhow::Error> {
-        self.get::<L2BlockByNumber>(number)
-    }
-}

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

@@ -49,29 +49,6 @@ pub struct DB {
     inner: rocksdb::DB,
 }
 
-/// asd
-pub struct TransactionDB {
-    // name: &'static str, // for logging
-    inner: rocksdb::TransactionDB,
-}
-
-impl TransactionDB {
-    /// asd
-    pub fn transaction(&self) -> rocksdb::Transaction<'_, rocksdb::TransactionDB> {
-        self.inner.transaction()
-    }
-
-    /// Returns the handle for a rocksdb column family.
-    pub fn get_cf_handle(&self, cf_name: &str) -> anyhow::Result<&rocksdb::ColumnFamily> {
-        self.inner.cf_handle(cf_name).ok_or_else(|| {
-            format_err!(
-                "DB::cf_handle not found for column family name: {}",
-                cf_name
-            )
-        })
-    }
-}
-
 impl DB {
     /// Opens the DB with a tempdir. Should only be used in tests
     #[cfg(feature = "test-utils")]
@@ -120,28 +97,6 @@ impl DB {
         Ok(db)
     }
 
-    /// Opens a database backed by RocksDB, using the provided column family names and default
-    /// column family options.
-    pub fn open_transaction_db(
-        path: impl AsRef<Path>,
-        column_families: impl IntoIterator<Item = impl Into<String>>,
-        options: &RawRocksdbOptions,
-    ) -> anyhow::Result<TransactionDB> {
-        let txn_db_opts = rocksdb::TransactionDBOptions::default();
-        let inner = rocksdb::TransactionDB::open_cf_descriptors(
-            &options.db_options,
-            &txn_db_opts,
-            path,
-            column_families.into_iter().map(|cf_name| {
-                let mut cf_opts = rocksdb::Options::default();
-                cf_opts.set_compression_type(rocksdb::DBCompressionType::Lz4);
-                cf_opts.set_block_based_table_factory(&options.block_options);
-                rocksdb::ColumnFamilyDescriptor::new(cf_name, cf_opts)
-            }),
-        )?;
-        Ok(TransactionDB { inner })
-    }
-
     /// Returns the path of the DB.
     pub fn path(&self) -> &Path {
         self.inner.path()

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

@@ -10,7 +10,7 @@ use citrea_common::{FullNodeConfig, ProverGuestRunConfig, RpcConfig};
 use citrea_stf::runtime::CitreaRuntime;
 use prover_services::ParallelProverService;
 use reth_tasks::TaskExecutor;
-use sov_db::ledger_db::{LedgerDB, TransactionLedgerDB};
+use sov_db::ledger_db::LedgerDB;
 use sov_db::rocks_db_config::RocksdbConfig;
 use sov_modules_api::default_context::DefaultContext;
 use sov_modules_api::DaSpec;
@@ -130,9 +130,4 @@ pub trait RollupBlueprint: Sized + Send + Sync {
     fn create_ledger_db(&self, rocksdb_config: &RocksdbConfig) -> LedgerDB {
         LedgerDB::with_config(rocksdb_config).expect("Ledger DB failed to open")
     }
-
-    /// Creates instance of a transactional LedgerDB.
-    fn create_ledger_db_tx(&self, rocksdb_config: &RocksdbConfig) -> TransactionLedgerDB {
-        TransactionLedgerDB::with_config(rocksdb_config).expect("Ledger DB failed to open")
-    }
 }