fix(storage): keep !Send MDBX write tx out of async state machine (#41)

* fix(storage): keep !Send MDBX write tx out of async state machine

Extract synchronous hot-storage unwind logic from `drain_above` into
`unwind_hot_above` so the `!Send` MDBX write transaction never appears
in the async generator's state machine. This makes the future returned
by `drain_above` `Send`, unblocking use from `Send`-bounded executors
like `reth::install_exex`.

Adds compile-time `Send` canaries for `drain_above` and `cold_lag`
parameterized over `DatabaseEnv` to prevent regressions.

Closes ENG-2080

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* chore(storage): add Send compile canaries to builder and replay_to_cold

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: prestwich

Cargo.toml

@@ -3,7 +3,7 @@ members = ["crates/*"]
 resolver = "2"
 
 [workspace.package]
-version = "0.6.6"
+version = "0.6.7"
 edition = "2024"
 rust-version = "1.92"
 authors = ["init4"]
@@ -35,13 +35,13 @@ incremental = false
 
 [workspace.dependencies]
 # internal
-signet-hot = { version = "0.6.6", path = "./crates/hot" }
-signet-hot-mdbx = { version = "0.6.6", path = "./crates/hot-mdbx" }
-signet-cold = { version = "0.6.6", path = "./crates/cold" }
-signet-cold-mdbx = { version = "0.6.6", path = "./crates/cold-mdbx" }
-signet-cold-sql = { version = "0.6.6", path = "./crates/cold-sql" }
-signet-storage = { version = "0.6.6", path = "./crates/storage" }
-signet-storage-types = { version = "0.6.6", path = "./crates/types" }
+signet-hot = { version = "0.6.7", path = "./crates/hot" }
+signet-hot-mdbx = { version = "0.6.7", path = "./crates/hot-mdbx" }
+signet-cold = { version = "0.6.7", path = "./crates/cold" }
+signet-cold-mdbx = { version = "0.6.7", path = "./crates/cold-mdbx" }
+signet-cold-sql = { version = "0.6.7", path = "./crates/cold-sql" }
+signet-storage = { version = "0.6.7", path = "./crates/storage" }
+signet-storage-types = { version = "0.6.7", path = "./crates/types" }
 
 # External, in-house
 signet-libmdbx = { version = "0.8.0" }

crates/storage/src/builder.rs

@@ -165,6 +165,13 @@ mod tests {
     use super::*;
     use serial_test::serial;
 
+    /// Compile-time canary: `StorageBuilder::build` must return a `Send`
+    /// future so it can be used from `Send`-bounded executors.
+    fn _assert_send<T: Send>(_: T) {}
+    fn _build_is_send(b: StorageBuilder<MdbxConnector, MdbxConnector>) {
+        _assert_send(b.build());
+    }
+
     #[test]
     #[serial]
     fn from_env_missing_hot_path() {

crates/storage/src/unified.rs

@@ -255,17 +255,13 @@ impl<H: HotKv> UnifiedStorage<H> {
     /// [`Hot`]: crate::StorageError::Hot
     /// [`Cold`]: crate::StorageError::Cold
     pub async fn drain_above(&self, block: BlockNumber) -> StorageResult<Vec<DrainedBlock>> {
-        // 1–2. Read headers then unwind hot storage in a single write tx
-        //       to avoid TOCTOU races between reading and unwinding.
-        let writer = self.hot.writer()?;
-        let last = match writer.get_execution_range().map_err(|e| e.into_hot_kv_error())? {
-            Some((_, last)) if last > block => last,
-            _ => return Ok(Vec::new()),
-        };
-        let headers =
-            writer.get_headers_range(block + 1, last).map_err(|e| e.into_hot_kv_error())?;
-        writer.unwind_above(block).map_err(|e| e.map_db(|e| e.into_hot_kv_error()))?;
-        writer.raw_commit().map_err(|e| e.into_hot_kv_error())?;
+        // 1–2. Read headers and unwind hot storage synchronously.
+        //      Extracted to a sync helper so the `!Send` write transaction
+        //      does not appear in the async state machine.
+        let headers = self.unwind_hot_above(block)?;
+        if headers.is_empty() {
+            return Ok(Vec::new());
+        }
 
         // 3. Atomically drain cold (best-effort — failure = normal cold lag)
         let cold_receipts = self.cold.drain_above(block).await.unwrap_or_default();
@@ -283,6 +279,22 @@ impl<H: HotKv> UnifiedStorage<H> {
         Ok(drained)
     }
 
+    /// Read headers above `block` and unwind hot storage in a single write
+    /// transaction to avoid TOCTOU races. Returns an empty vec if there is
+    /// nothing to unwind.
+    fn unwind_hot_above(&self, block: BlockNumber) -> StorageResult<Vec<SealedHeader>> {
+        let writer = self.hot.writer()?;
+        let last = match writer.get_execution_range().map_err(|e| e.into_hot_kv_error())? {
+            Some((_, last)) if last > block => last,
+            _ => return Ok(Vec::new()),
+        };
+        let headers =
+            writer.get_headers_range(block + 1, last).map_err(|e| e.into_hot_kv_error())?;
+        writer.unwind_above(block).map_err(|e| e.map_db(|e| e.into_hot_kv_error()))?;
+        writer.raw_commit().map_err(|e| e.into_hot_kv_error())?;
+        Ok(headers)
+    }
+
     /// Unwind storage above the given block number (reorg handling).
     ///
     /// This method:
@@ -350,3 +362,24 @@ impl<H: HotKv> UnifiedStorage<H> {
         self.cold.append_blocks(cold_data).await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use signet_hot_mdbx::DatabaseEnv;
+
+    /// Compile-time canaries: all async methods on `UnifiedStorage<DatabaseEnv>`
+    /// must return `Send` futures, even though MDBX write transactions are
+    /// `!Send`. If a `!Send` type leaks into the async state machine, these
+    /// will fail to compile.
+    fn _assert_send<T: Send>(_: T) {}
+    fn _drain_above_is_send(s: &UnifiedStorage<DatabaseEnv>) {
+        _assert_send(s.drain_above(0));
+    }
+    fn _cold_lag_is_send(s: &UnifiedStorage<DatabaseEnv>) {
+        _assert_send(s.cold_lag());
+    }
+    fn _replay_to_cold_is_send(s: &UnifiedStorage<DatabaseEnv>) {
+        _assert_send(s.replay_to_cold(Vec::new()));
+    }
+}