feat: live collector integration

crates/optimism/bin/src/main.rs

@@ -77,20 +77,23 @@ async fn launch_node_with_storage<S>(
 where
     S: OpProofsStore + Clone + 'static,
 {
-    let storage_clone = storage.clone();
+    let storage_exec = storage.clone();
+    let storage_rpc = storage.clone();
     let proofs_history_enabled = args.proofs_history;
     let handle = builder
         .node(OpNode::new(args.rollup_args))
         .install_exex_if(proofs_history_enabled, "proofs-history", async move |exex_context| {
-            Ok(OpProofsExEx::new(exex_context, storage, args.proofs_history_window).run().boxed())
+            Ok(OpProofsExEx::new(exex_context, storage_exec, args.proofs_history_window)
+                .run()
+                .boxed())
         })
         .extend_rpc_modules(move |ctx| {
             if proofs_history_enabled {
-                let api_ext = EthApiExt::new(ctx.registry.eth_api().clone(), storage_clone.clone());
+                let api_ext = EthApiExt::new(ctx.registry.eth_api().clone(), storage_rpc.clone());
                 let debug_ext = DebugApiExt::new(
                     ctx.node().provider().clone(),
                     ctx.registry.eth_api().clone(),
-                    storage_clone,
+                    storage_rpc,
                     Box::new(ctx.node().task_executor().clone()),
                     ctx.node().evm_config().clone(),
                 );

crates/optimism/exex/Cargo.toml

@@ -23,10 +23,14 @@ reth-chainspec.workspace = true
 # proofs exex handles `TrieUpdates` in notifications
 reth-optimism-trie = { workspace = true, features = ["serde-bincode-compat"] }
 
+# alloy
+alloy-consensus.workspace = true
+
 # misc
 eyre.workspace = true
 futures-util.workspace = true
 derive_more.workspace = true
+tracing.workspace = true
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["test-util", "rt-multi-thread", "macros"] }

crates/optimism/exex/src/lib.rs

@@ -8,42 +8,43 @@
 #![cfg_attr(docsrs, feature(doc_cfg))]
 #![cfg_attr(not(test), warn(unused_crate_dependencies))]
 
+use alloy_consensus::BlockHeader;
 use derive_more::Constructor;
 use futures_util::TryStreamExt;
 use reth_chainspec::ChainInfo;
-use reth_exex::{ExExContext, ExExEvent};
+use reth_exex::{ExExContext, ExExEvent, ExExNotification};
 use reth_node_api::{FullNodeComponents, NodePrimitives};
 use reth_node_types::NodeTypes;
-use reth_optimism_trie::{BackfillJob, OpProofsStore};
+use reth_optimism_trie::{live::LiveTrieCollector, BackfillJob, OpProofsStorage, OpProofsStore};
 use reth_provider::{BlockNumReader, DBProvider, DatabaseProviderFactory};
+use tracing::{debug, error};
 
 /// OP Proofs ExEx - processes blocks and tracks state changes within fault proof window.
 ///
 /// Saves and serves trie nodes to make proofs faster. This handles the process of
 /// saving the current state, new blocks as they're added, and serving proof RPCs
 /// based on the saved data.
 #[derive(Debug, Constructor)]
-pub struct OpProofsExEx<Node, S>
+pub struct OpProofsExEx<Node, Storage>
 where
     Node: FullNodeComponents,
-    S: OpProofsStore + Clone,
 {
     /// The ExEx context containing the node related utilities e.g. provider, notifications,
     /// events.
     ctx: ExExContext<Node>,
     /// The type of storage DB.
-    storage: S,
+    storage: OpProofsStorage<Storage>,
     /// The window to span blocks for proofs history. Value is the number of blocks, received as
     /// cli arg.
     #[expect(dead_code)]
     proofs_history_window: u64,
 }
 
-impl<Node, S, Primitives> OpProofsExEx<Node, S>
+impl<Node, Storage, Primitives> OpProofsExEx<Node, Storage>
 where
     Node: FullNodeComponents<Types: NodeTypes<Primitives = Primitives>>,
     Primitives: NodePrimitives,
-    S: OpProofsStore + Clone,
+    Storage: OpProofsStore + Clone + 'static,
 {
     /// Main execution loop for the ExEx
     pub async fn run(mut self) -> eyre::Result<()> {
@@ -53,10 +54,84 @@ where
         let ChainInfo { best_number, best_hash } = self.ctx.provider().chain_info()?;
         BackfillJob::new(self.storage.clone(), &db_tx).run(best_number, best_hash).await?;
 
+        let collector = LiveTrieCollector::new(
+            self.ctx.evm_config().clone(),
+            self.ctx.provider().clone(),
+            &self.storage,
+        );
+
         while let Some(notification) = self.ctx.notifications.try_next().await? {
-            // match &notification {
-            //     _ => {}
-            // };
+            match &notification {
+                ExExNotification::ChainCommitted { new } => {
+                    debug!(
+                        block_number = new.tip().number(),
+                        block_hash = ?new.tip().hash(),
+                        "ChainCommitted notification received",
+                    );
+
+                    // Get latest stored number (ignore stored hash for now)
+                    let latest_stored_block_number =
+                        match self.storage.get_latest_block_number().await? {
+                            Some((n, _)) => n,
+                            None => {
+                                return Err(eyre::eyre!("No blocks stored in proofs storage"));
+                            }
+                        };
+
+                    // If tip is not newer than what we have, nothing to do.
+                    if new.tip().number() <= latest_stored_block_number {
+                        debug!(
+                            block_number = new.tip().number(),
+                            latest_stored = latest_stored_block_number,
+                            "Tip number is less than or equal to latest stored, skipping"
+                        );
+                        continue;
+                    }
+
+                    // Start from the next block after the latest stored one.
+                    let start = latest_stored_block_number.saturating_add(1);
+                    debug!(
+                        start,
+                        end = new.tip().number(),
+                        "Applying updates for blocks in committed chain"
+                    );
+                    for block_number in start..=new.tip().number() {
+                        match new.blocks().get(&block_number) {
+                            Some(block) => {
+                                collector.execute_and_store_block_updates(block).await?;
+                            }
+                            None => {
+                                error!(
+                                    block_number,
+                                    "Missing block in committed chain; stopping incremental application",
+                                );
+                                return Err(eyre::eyre!(
+                                    "Missing block {} in committed chain",
+                                    block_number
+                                ));
+                            }
+                        }
+                    }
+                }
+                ExExNotification::ChainReorged { old, new } => {
+                    debug!(
+                        old_block_number = old.tip().number(),
+                        old_block_hash = ?old.tip().hash(),
+                        new_block_number = new.tip().number(),
+                        new_block_hash = ?new.tip().hash(),
+                        "ChainReorged notification received",
+                    );
+                    // handle reorg logic here
+                }
+                ExExNotification::ChainReverted { old } => {
+                    debug!(
+                        old_block_number = old.tip().number(),
+                        old_block_hash = ?old.tip().hash(),
+                        "ChainReverted notification received",
+                    );
+                    // handle revert logic here
+                }
+            };
 
             if let Some(committed_chain) = notification.committed_chain() {
                 self.ctx

crates/optimism/trie/src/api.rs

@@ -1,5 +1,6 @@
 //! Storage API for external storage of intermediary trie nodes.
 
+use alloy_eips::eip1898::BlockWithParent;
 use alloy_primitives::{map::HashMap, B256, U256};
 use auto_impl::auto_impl;
 use reth_db::DatabaseError;
@@ -17,6 +18,9 @@ pub enum OpProofsStorageError {
     /// Parent block number is less than earliest stored block number
     #[error("Parent block number is less than earliest stored block number")]
     UnknownParent,
+    /// Block is out of order
+    #[error("Block {0} is out of order (parent: {1}, latest stored block number: {2})")]
+    OutOfOrder(u64, B256, B256),
     /// Block update failed since parent state
     #[error("Cannot execute block updates for block {0} without parent state {1} (latest stored block number: {2})")]
     BlockUpdateFailed(u64, u64, u64),
@@ -187,7 +191,7 @@ pub trait OpProofsStore: Send + Sync + Debug {
     /// so should only happen for legacy reasons.
     fn store_trie_updates(
         &self,
-        block_number: u64,
+        block_ref: BlockWithParent,
         block_state_diff: BlockStateDiff,
     ) -> impl Future<Output = OpProofsStorageResult<()>> + Send;