refactor: remove unsafe code in rocksdb

Use rocksdb txn write for all write operations, this may slightly impact performance

Signed-off-by: bsbds <[email protected]>

crates/engine/src/api/engine_api.rs

@@ -8,10 +8,10 @@ pub trait StorageEngine: Send + Sync + 'static + std::fmt::Debug {
     /// The snapshot type
     type Snapshot: SnapshotApi;
     /// The transaction type
-    type Transaction: TransactionApi;
+    type Transaction<'db>: TransactionApi;
 
     /// Creates a transaction
-    fn transaction(&self) -> Self::Transaction;
+    fn transaction(&self) -> Self::Transaction<'_>;
 
     /// Get the value associated with a key value and the given table
     ///

crates/engine/src/api/operation.rs

@@ -1,6 +1,7 @@
 use crate::EngineError;
 
 /// Storage operations
+/// TODO: refactor this trait, require `&mut self` for write operations
 pub trait StorageOps {
     /// Write an op to the transaction
     ///

crates/engine/src/memory_engine/mod.rs

@@ -58,7 +58,7 @@ impl MemoryEngine {
 #[async_trait::async_trait]
 impl StorageEngine for MemoryEngine {
     type Snapshot = MemorySnapshot;
-    type Transaction = MemoryTransaction;
+    type Transaction<'db> = MemoryTransaction;
 
     #[inline]
     fn transaction(&self) -> MemoryTransaction {

crates/engine/src/metrics.rs

@@ -60,10 +60,10 @@ where
 {
     /// The snapshot type
     type Snapshot = Layer<E::Snapshot>;
-    type Transaction = Layer<E::Transaction>;
+    type Transaction<'db> = Layer<E::Transaction<'db>>;
 
     /// Creates a transaction
-    fn transaction(&self) -> Self::Transaction {
+    fn transaction(&self) -> Self::Transaction<'_> {
         Layer::new(self.engine.transaction())
     }
 

crates/engine/src/proxy.rs

@@ -68,10 +68,10 @@ impl Engine {
 #[async_trait::async_trait]
 impl StorageEngine for Engine {
     type Snapshot = Snapshot;
-    type Transaction = Transaction;
+    type Transaction<'db> = Transaction<'db>;
 
     #[inline]
-    fn transaction(&self) -> Transaction {
+    fn transaction(&self) -> Transaction<'_> {
         match *self {
             Engine::Memory(ref e) => Transaction::Memory(e.transaction()),
             Engine::Rocks(ref e) => Transaction::Rocks(e.transaction()),
@@ -166,14 +166,14 @@ impl StorageEngine for Engine {
 /// NOTE: Currently multiple concurrent transactions is not supported
 #[derive(Debug)]
 #[non_exhaustive]
-pub enum Transaction {
+pub enum Transaction<'a> {
     /// Memory transaction
     Memory(MemoryTransaction),
     /// Rocks transaction
-    Rocks(metrics::Layer<RocksTransaction>),
+    Rocks(metrics::Layer<RocksTransaction<'a>>),
 }
 
-impl TransactionApi for Transaction {
+impl TransactionApi for Transaction<'_> {
     #[inline]
     fn commit(self) -> Result<(), EngineError> {
         match self {

crates/engine/src/rocksdb_engine/mod.rs

@@ -151,11 +151,12 @@ impl RocksEngine {
 #[async_trait::async_trait]
 impl StorageEngine for RocksEngine {
     type Snapshot = RocksSnapshot;
-    type Transaction = RocksTransaction;
+    type Transaction<'db> = RocksTransaction<'db>;
 
     #[inline]
-    fn transaction(&self) -> RocksTransaction {
-        RocksTransaction::new(Arc::clone(&self.inner), Arc::clone(&self.size))
+    fn transaction(&self) -> RocksTransaction<'_> {
+        let txn = self.inner.transaction();
+        RocksTransaction::new(Arc::clone(&self.inner), txn, Arc::clone(&self.size))
     }
 
     #[inline]