lyquor/
lyquor.rs

1use std::path::{Path, PathBuf};
2use std::sync::Arc;
3
4use alloy_node_bindings::{Anvil, AnvilInstance};
5use serde_json::{Map, Value};
6use tokio::signal::unix::{SignalKind, signal};
7
8use lyquor_api::anyhow::{self, Context as _};
9use lyquor_config::{
10    config::{DB, NodeConfig},
11    profile::NetworkType,
12};
13use lyquor_lib::node::{LyquorNode, NodeArgs, anvil_state_path};
14
15static ANVIL_PID: std::sync::OnceLock<u32> = std::sync::OnceLock::new();
16
17fn generate_unused_port() -> anyhow::Result<u16> {
18    let listener = std::net::TcpListener::bind("0.0.0.0:0").context("Failed to bind to local address.")?;
19    let port = listener.local_addr().context("Failed to get local address.")?.port();
20    Ok(port)
21}
22
23fn start_devnet_anvil(state_file: Option<&Path>, state_interval_secs: u64) -> anyhow::Result<AnvilInstance> {
24    let anvil_port = generate_unused_port()?;
25    // Mask SIGINT so Anvil is terminated only when this owner drops it.
26    use nix::sys::signal::{self, SigSet, SigmaskHow, Signal};
27    let mut sigset = SigSet::empty();
28    sigset.add(Signal::SIGINT);
29    let mut old_mask = SigSet::empty();
30    signal::sigprocmask(SigmaskHow::SIG_BLOCK, Some(&sigset), Some(&mut old_mask))
31        .context("Failed to mask SIGINT for anvil.")?;
32
33    let mut anvil_builder = Anvil::new().port(anvil_port).keep_stdout();
34    if let Some(state_file) = state_file {
35        if let Some(state_dir) = state_file.parent() {
36            std::fs::create_dir_all(state_dir)?;
37        }
38        anvil_builder = anvil_builder.arg("--state").arg(state_file.as_os_str());
39        if state_interval_secs > 0 {
40            anvil_builder = anvil_builder
41                .arg("--state-interval")
42                .arg(state_interval_secs.to_string());
43        }
44    }
45    let mut anvil = anvil_builder.try_spawn()?;
46
47    let stdout_reader = std::io::BufReader::new(
48        anvil
49            .child_mut()
50            .stdout
51            .take()
52            .context("Failed to read from anvil stdout.")?,
53    );
54    tokio::task::spawn_blocking(move || {
55        use std::io::BufRead;
56        for line in stdout_reader.lines() {
57            tracing::debug!(target: "lyquor_anvil", "{}", line.unwrap_or_else(|_| String::new()));
58        }
59    });
60
61    tracing::info!("Anvil started at port {}.", anvil_port);
62    signal::sigprocmask(SigmaskHow::SIG_SETMASK, Some(&old_mask), None)
63        .context("Failed to restore old signal mask for anvil.")?;
64    Ok(anvil)
65}
66
67fn parse_override_value(raw: &str) -> Value {
68    if let Ok(value) = raw.parse::<bool>() {
69        return Value::from(value);
70    }
71    if let Ok(value) = raw.parse::<u64>() {
72        return Value::from(value);
73    }
74    if let Ok(value) = raw.parse::<i64>() {
75        return Value::from(value);
76    }
77    if let Ok(value) = raw.parse::<f64>() {
78        return Value::from(value);
79    }
80    Value::from(raw.to_string())
81}
82
83fn insert_override_path(overrides: &mut Map<String, Value>, key: &str, value: Value) -> anyhow::Result<()> {
84    let mut segments = key.split('.').peekable();
85    let mut current = overrides;
86
87    while let Some(segment) = segments.next() {
88        if segments.peek().is_none() {
89            current.insert(segment.to_string(), value);
90            return Ok(());
91        }
92
93        let entry = current
94            .entry(segment.to_string())
95            .or_insert_with(|| Value::Object(Map::<String, Value>::new()));
96        match entry {
97            Value::Object(dict) => current = dict,
98            _ => anyhow::bail!("Override path '{key}' conflicts with non-table value."),
99        }
100    }
101
102    Ok(())
103}
104
105fn build_config_overrides(matches: &clap::ArgMatches) -> anyhow::Result<Map<String, Value>> {
106    let mut overrides = Map::new();
107    if let Some(entries) = matches.get_many::<String>("config-override") {
108        for entry in entries {
109            let (key, value) = entry
110                .split_once('=')
111                .ok_or_else(|| anyhow::anyhow!("Invalid --config-override '{entry}', expected key=value"))?;
112            insert_override_path(&mut overrides, key, parse_override_value(value))?;
113        }
114    }
115    Ok(overrides)
116}
117
118const NODE_RUNTIME_THREAD_NAME: &str = "lyquor-node";
119const VM_RUNTIME_THREAD_NAME: &str = "lyquor-vm";
120
121fn build_node_runtime(worker_threads: usize) -> std::io::Result<tokio::runtime::Runtime> {
122    tokio::runtime::Builder::new_multi_thread()
123        .enable_all()
124        .worker_threads(worker_threads)
125        .thread_name(NODE_RUNTIME_THREAD_NAME)
126        .build()
127}
128
129fn build_vm_runtime(worker_threads: usize) -> std::io::Result<tokio::runtime::Runtime> {
130    tokio::runtime::Builder::new_multi_thread()
131        .enable_all()
132        .worker_threads(worker_threads)
133        .thread_name(VM_RUNTIME_THREAD_NAME)
134        .build()
135}
136
137fn load_config_from_args() -> anyhow::Result<NodeConfig> {
138    let matches = clap::command!()
139        .version(lyquor_cli::build_version!())
140        .propagate_version(true)
141        .arg(clap::arg!(--config <PATH> "Path to the Lyquor config file (toml, yaml, json)."))
142        .arg(
143            clap::arg!(--"config-override" <KEY_VALUE> "Override config with key=value (repeatable).")
144                .action(clap::ArgAction::Append)
145                .value_parser(clap::builder::NonEmptyStringValueParser::new()),
146        )
147        .get_matches();
148
149    let config_path = matches.get_one::<String>("config").map(PathBuf::from);
150    NodeConfig::load(config_path, build_config_overrides(&matches)?)
151}
152
153fn main() -> anyhow::Result<()> {
154    std::panic::set_hook(Box::new(|_| {
155        if let Some(&pid) = ANVIL_PID.get() {
156            tracing::warn!("Panic detected, killing anvil process {pid}");
157            use nix::sys::signal::{Signal, kill};
158            let _ = kill(nix::unistd::Pid::from_raw(pid as i32), Signal::SIGTERM);
159        }
160    }));
161
162    lyquor_cli::setup_tracing()?;
163    println!("{}", lyquor_cli::format_logo_banner(lyquor_cli::build_version!()));
164
165    let config = load_config_from_args()?;
166    let runtime = build_node_runtime(config.runtime.node_threads).context("Failed to build node Tokio runtime")?;
167    let vm_runtime = build_vm_runtime(config.runtime.execution_threads).context("Failed to build VM Tokio runtime")?;
168    let vm_runtime_handle = vm_runtime.handle().clone();
169
170    let result = runtime.block_on(run_node(config, vm_runtime_handle));
171    vm_runtime.shutdown_background();
172    result
173}
174
175async fn run_node(config: NodeConfig, vm_runtime: tokio::runtime::Handle) -> anyhow::Result<()> {
176    let mut anvil: Option<AnvilInstance> = None;
177    let seq_eth_url = match config.profile.base {
178        NetworkType::Devnet => match config.profile.sequencer.as_deref() {
179            None => {
180                let state_file = match &config.storage.db {
181                    DB::RocksDb { .. } => Some(anvil_state_path(
182                        &config.resolved_db_dir(),
183                        config.profile.base.as_str(),
184                    )),
185                    DB::MemDb => None,
186                };
187                let instance = start_devnet_anvil(state_file.as_deref(), config.flush_interval_secs)?;
188                ANVIL_PID.set(instance.child().id()).ok();
189                let endpoint = instance.ws_endpoint();
190                anvil = Some(instance);
191                endpoint
192            }
193            Some(url) => url.to_string(),
194        },
195        _ => anyhow::bail!("Network is not supported."),
196    };
197    let profile = Arc::new(config.profile.resolve(seq_eth_url.clone())?);
198    let node_seed = config.node_key.seed_bytes()?;
199    let (_, tls_config) = lyquor_tls::generator::single_node_config_with_seed(&node_seed, profile.network_domain());
200
201    // Install handlers before startup so a signal received while the node is becoming ready is
202    // observed as soon as startup completes.
203    let mut sigint = signal(SignalKind::interrupt()).context("Failed to register SIGINT handler")?;
204    let mut sigterm = signal(SignalKind::terminate()).context("Failed to register SIGTERM handler")?;
205
206    let node = LyquorNode::start(NodeArgs {
207        config: Arc::new(config),
208        profile,
209        seq_eth_url,
210        tls_config,
211        started_anvil: anvil.is_some(),
212        vm_runtime,
213    })
214    .await?;
215
216    tokio::select! {
217        _ = sigint.recv() => {
218            tracing::info!("received SIGINT");
219        }
220        _ = sigterm.recv() => {
221            tracing::info!("received SIGTERM");
222        }
223        () = node.finished() => {
224            tracing::info!("all node subsystems finished");
225        }
226    }
227    node.shutdown().await;
228    Ok(())
229}
lyquor/lyquor.rs

lyquor/
lyquor.rs
