reth_node_builder/launch/
exex.rs

1//! Support for launching execution extensions.
2
3use std::{fmt, fmt::Debug};
4
5use futures::future;
6use reth_chain_state::ForkChoiceSubscriptions;
7use reth_chainspec::EthChainSpec;
8use reth_exex::{
9    ExExContext, ExExHandle, ExExManager, ExExManagerHandle, ExExNotificationSource, Wal,
10    DEFAULT_EXEX_MANAGER_CAPACITY,
11};
12use reth_node_api::{FullNodeComponents, NodeTypes};
13use reth_primitives::Head;
14use reth_provider::CanonStateSubscriptions;
15use reth_tracing::tracing::{debug, info};
16use tracing::Instrument;
17
18use crate::{common::WithConfigs, exex::BoxedLaunchExEx};
19
20/// Can launch execution extensions.
21pub struct ExExLauncher<Node: FullNodeComponents> {
22    head: Head,
23    extensions: Vec<(String, Box<dyn BoxedLaunchExEx<Node>>)>,
24    components: Node,
25    config_container: WithConfigs<<Node::Types as NodeTypes>::ChainSpec>,
26}
27
28impl<Node: FullNodeComponents + Clone> ExExLauncher<Node> {
29    /// Create a new `ExExLauncher` with the given extensions.
30    pub const fn new(
31        head: Head,
32        components: Node,
33        extensions: Vec<(String, Box<dyn BoxedLaunchExEx<Node>>)>,
34        config_container: WithConfigs<<Node::Types as NodeTypes>::ChainSpec>,
35    ) -> Self {
36        Self { head, extensions, components, config_container }
37    }
38
39    /// Launches all execution extensions.
40    ///
41    /// Spawns all extensions and returns the handle to the exex manager if any extensions are
42    /// installed.
43    pub async fn launch(
44        self,
45    ) -> eyre::Result<Option<ExExManagerHandle<<Node::Types as NodeTypes>::Primitives>>> {
46        let Self { head, extensions, components, config_container } = self;
47
48        if extensions.is_empty() {
49            // nothing to launch
50            return Ok(None)
51        }
52
53        info!(target: "reth::cli", "Loading ExEx Write-Ahead Log...");
54        let exex_wal = Wal::new(
55            config_container
56                .config
57                .datadir
58                .clone()
59                .resolve_datadir(config_container.config.chain.chain())
60                .exex_wal(),
61        )?;
62
63        let mut exex_handles = Vec::with_capacity(extensions.len());
64        let mut exexes = Vec::with_capacity(extensions.len());
65
66        for (id, exex) in extensions {
67            // create a new exex handle
68            let (handle, events, notifications) = ExExHandle::new(
69                id.clone(),
70                head,
71                components.provider().clone(),
72                components.block_executor().clone(),
73                exex_wal.handle(),
74            );
75            exex_handles.push(handle);
76
77            // create the launch context for the exex
78            let context = ExExContext {
79                head,
80                config: config_container.config.clone(),
81                reth_config: config_container.toml_config.clone(),
82                components: components.clone(),
83                events,
84                notifications,
85            };
86
87            let executor = components.task_executor().clone();
88            exexes.push(async move {
89                debug!(target: "reth::cli", id, "spawning exex");
90                let span = reth_tracing::tracing::info_span!("exex", id);
91
92                // init the exex
93                let exex = exex.launch(context).instrument(span.clone()).await.unwrap();
94
95                // spawn it as a crit task
96                executor.spawn_critical(
97                    "exex",
98                    async move {
99                        info!(target: "reth::cli", "ExEx started");
100                        match exex.await {
101                            Ok(_) => panic!("ExEx {id} finished. ExExes should run indefinitely"),
102                            Err(err) => panic!("ExEx {id} crashed: {err}"),
103                        }
104                    }
105                    .instrument(span),
106                );
107            });
108        }
109
110        future::join_all(exexes).await;
111
112        // spawn exex manager
113        debug!(target: "reth::cli", "spawning exex manager");
114        let exex_manager = ExExManager::new(
115            components.provider().clone(),
116            exex_handles,
117            DEFAULT_EXEX_MANAGER_CAPACITY,
118            exex_wal,
119            components.provider().finalized_block_stream(),
120        );
121        let exex_manager_handle = exex_manager.handle();
122        components.task_executor().spawn_critical("exex manager", async move {
123            exex_manager.await.expect("exex manager crashed");
124        });
125
126        // send notifications from the blockchain tree to exex manager
127        let mut canon_state_notifications = components.provider().subscribe_to_canonical_state();
128        let mut handle = exex_manager_handle.clone();
129        components.task_executor().spawn_critical(
130            "exex manager blockchain tree notifications",
131            async move {
132                while let Ok(notification) = canon_state_notifications.recv().await {
133                    handle
134                        .send_async(ExExNotificationSource::BlockchainTree, notification.into())
135                        .await
136                        .expect("blockchain tree notification could not be sent to exex manager");
137                }
138            },
139        );
140
141        info!(target: "reth::cli", "ExEx Manager started");
142
143        Ok(Some(exex_manager_handle))
144    }
145}
146
147impl<Node: FullNodeComponents> Debug for ExExLauncher<Node> {
148    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
149        f.debug_struct("ExExLauncher")
150            .field("head", &self.head)
151            .field("extensions", &self.extensions.iter().map(|(id, _)| id).collect::<Vec<_>>())
152            .field("components", &"...")
153            .field("config_container", &self.config_container)
154            .finish()
155    }
156}
reth_node_builder/launch/exex.rs

reth_node_builder/launch/
exex.rs
