feat(exex): write ahead log

crates/exex/exex/Cargo.toml

@@ -16,7 +16,8 @@ workspace = true
 reth-chainspec.workspace = true
 reth-config.workspace = true
 reth-evm.workspace = true
-reth-exex-types.workspace = true
+reth-exex-types = { workspace = true, features = ["serde"] }
+reth-fs-util.workspace = true
 reth-metrics.workspace = true
 reth-node-api.workspace = true
 reth-node-core.workspace = true
@@ -38,6 +39,8 @@ tokio.workspace = true
 ## misc
 eyre.workspace = true
 metrics.workspace = true
+rmp-serde = "1.3"
+tracing.workspace = true
 
 [dev-dependencies]
 reth-blockchain-tree.workspace = true
@@ -48,8 +51,10 @@ reth-node-api.workspace = true
 reth-primitives-traits = { workspace = true, features = ["test-utils"] }
 reth-provider = { workspace = true, features = ["test-utils"] }
 reth-testing-utils.workspace = true
+reth-trie.workspace = true
 
 secp256k1.workspace = true
+tempfile.workspace = true
 
 [features]
 default = []

crates/exex/exex/src/lib.rs

@@ -46,6 +46,8 @@ pub use event::*;
 mod manager;
 pub use manager::*;
 
+mod wal;
+
 // Re-export exex types
 #[doc(inline)]
 pub use reth_exex_types::*;

crates/exex/exex/src/manager.rs

@@ -26,7 +26,7 @@ use tokio::sync::{
     mpsc::{self, error::SendError, Receiver, UnboundedReceiver, UnboundedSender},
     watch,
 };
-use tokio_util::sync::{PollSendError, PollSender, ReusableBoxFuture};
+use tokio_util::sync::{PollSender, ReusableBoxFuture};
 
 /// Metrics for an `ExEx`.
 #[derive(Metrics)]
@@ -99,7 +99,7 @@ impl ExExHandle {
         &mut self,
         cx: &mut Context<'_>,
         (notification_id, notification): &(usize, ExExNotification),
-    ) -> Poll<Result<(), PollSendError<ExExNotification>>> {
+    ) -> Poll<eyre::Result<()>> {
         if let Some(finished_height) = self.finished_height {
             match notification {
                 ExExNotification::ChainCommitted { new } => {
@@ -132,24 +132,25 @@ impl ExExHandle {
             %notification_id,
             "Reserving slot for notification"
         );
-        match self.sender.poll_reserve(cx) {
-            Poll::Ready(Ok(())) => (),
-            other => return other,
-        }
+        ready!(self.sender.poll_reserve(cx)?);
+
+        // debug!(
+        //     exex_id = %self.id,
+        //     %notification_id,
+        //     "Appending notification to WAL"
+        // );
+        // self.wal.commit(notification)?;
 
         debug!(
             exex_id = %self.id,
             %notification_id,
             "Sending notification"
         );
-        match self.sender.send_item(notification.clone()) {
-            Ok(()) => {
-                self.next_notification_id = notification_id + 1;
-                self.metrics.notifications_sent_total.increment(1);
-                Poll::Ready(Ok(()))
-            }
-            Err(err) => Poll::Ready(Err(err)),
-        }
+        self.sender.send_item(notification.clone())?;
+        self.next_notification_id = notification_id + 1;
+        self.metrics.notifications_sent_total.increment(1);
+
+        Poll::Ready(Ok(()))
     }
 }
 
@@ -647,10 +648,7 @@ impl Future for ExExManager {
                 .checked_sub(self.min_id)
                 .expect("exex expected notification ID outside the manager's range");
             if let Some(notification) = self.buffer.get(notification_index) {
-                if let Poll::Ready(Err(err)) = exex.send(cx, notification) {
-                    // the channel was closed, which is irrecoverable for the manager
-                    return Poll::Ready(Err(err.into()))
-                }
+                let _ = exex.send(cx, notification)?;
             }
             min_id = min_id.min(exex.next_notification_id);
             self.exex_handles.push(exex);