lmdb: implement automatic event batching with flume channels

Update the LMDB ingester to support automatic event batching for improved write performance. The existing ingester has been enhanced to process multiple operations in batches using single LMDB transactions, significantly reducing write overhead.

Key changes:
- Modify ingester to process events in batches rather than one-by-one, using a single RwTxn per batch;
- Add support for delete operations alongside save operations in the ingester.

Pull-Request: #1010
Signed-off-by: Yuki Kishimoto <[email protected]>

Author: dcadenas

Cargo.lock

@@ -28,6 +28,7 @@
 ### Changed
 
 - Consolidate deletion logic and improve transactional consistency (https://github.com/rust-nostr/nostr/pull/1010)
+- Add automatic event batching for better write performance (https://github.com/rust-nostr/nostr/pull/1010)
 
 ## v0.43.0 - 2025/07/28
 

database/nostr-lmdb/CHANGELOG.md

@@ -13,6 +13,7 @@ keywords = ["nostr", "database", "lmdb"]
 
 [dependencies]
 async-utility.workspace = true
+flume = "0.11"
 nostr = { workspace = true, features = ["std"] }
 nostr-database = { workspace = true, features = ["flatbuf"] }
 tokio = { workspace = true, features = ["sync"] }
@@ -26,5 +27,6 @@ heed = { version = "0.20", default-features = false, features = ["read-txn-no-tl
 heed = { version = "0.20", default-features = false, features = ["read-txn-no-tls", "posix-sem"] }
 
 [dev-dependencies]
+futures = "0.3"
 tempfile.workspace = true
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }

database/nostr-lmdb/Cargo.toml

@@ -22,12 +22,14 @@ pub enum Error {
     Key(key::Error),
     Secp256k1(secp256k1::Error),
     OneshotRecv(oneshot::error::RecvError),
-    /// MPSC send error
-    MpscSend,
+    /// Flume channel send error
+    FlumeSend,
     /// The event kind is wrong
     WrongEventKind,
     /// Not found
     NotFound,
+    /// Batched transaction failed - sent to operations that didn't cause the error
+    BatchTransactionFailed,
 }
 
 impl std::error::Error for Error {}
@@ -42,9 +44,10 @@ impl fmt::Display for Error {
             Self::Key(e) => write!(f, "{e}"),
             Self::Secp256k1(e) => write!(f, "{e}"),
             Self::OneshotRecv(e) => write!(f, "{e}"),
-            Self::MpscSend => write!(f, "mpsc channel send error"),
+            Self::FlumeSend => write!(f, "flume channel send error"),
             Self::NotFound => write!(f, "Not found"),
             Self::WrongEventKind => write!(f, "Wrong event kind"),
+            Self::BatchTransactionFailed => write!(f, "Batched transaction failed"),
         }
     }
 }