lyquor/

lyquor.rs

use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;

use alloy_node_bindings::{Anvil, AnvilInstance};
use barstrainer_client::SigningIdentity;
use lyquor_primitives::{LyquidNumber, debug_struct_name};
use serde_json::{Map, Value};
use tokio::signal::unix::{SignalKind, signal};
use tokio_util::sync::CancellationToken;

use lyquor_api::{
    anyhow::{self, Context as _Context},
    config::{DB, NodeConfig},
    kvstore::{KVStore, Key, PrefixedKVStore, ShadowKVStore},
    profile::{LyquorProfile, NetworkType},
    subkey_builder,
};
use lyquor_db::{MemDB, RocksDB};
use lyquor_image_store::{DirectoryOciStore, DirectoryRepo, LyquidImageRepo, OciDistributionStore};
use lyquor_jsonrpc::client::{ClientConfig, ClientHandle};
use lyquor_jsonrpc::types::{EthGetChainId, EthGetChainIdResp};
use lyquor_lib::db::{DBFlusher, DBFlusherHandle};
use lyquor_lib::dns_updater::publish_external_host_dns;
use lyquor_lib::image_resolver::ImageResolver;
use lyquor_lib::lyquid_log::LyquidLog;
use lyquor_lib::side_effects::{SideEffectRuntime, build_side_effect_runtime};
use lyquor_lib::{http, lyquid, utils};
use lyquor_primitives::{Address, Bytes, LyteLog};
use lyquor_seq::{SequenceBackend, eth, fco::FCO};

subkey_builder!(RootSubKey(
    ([0x01])-lyquid() => Key,
    ([0x02])-fco() => Key,
    ([0x03])-log() => Key,
    ([0x04])-side_effect_store() => Key,
));

fn subkey() -> &'static RootSubKey {
    static KEY: std::sync::OnceLock<RootSubKey> = std::sync::OnceLock::new();
    KEY.get_or_init(|| RootSubKey::new(Bytes::new().into()))
}

const ANVIL_STATE_FILENAME: &str = "anvil-state.json";

static ANVIL_PID: std::sync::OnceLock<u32> = std::sync::OnceLock::new();

async fn setup_eth_backend(
    profile: &LyquorProfile, reg: lyquor_seq::eth::ContractRegistry, jsonrpc_client: ClientHandle,
    shutdown: CancellationToken,
) -> anyhow::Result<eth::Backend> {
    let finality_tag: lyquor_jsonrpc::types::BlockNumber =
        serde_json::from_str(profile.finality()).context("Invalid finality tag.")?;

    Ok(eth::Backend::new(
        eth::Config::builder()
            .finality_tag(finality_tag)
            .registry(reg)
            .blocks_per_request(std::num::NonZeroUsize::new(10).context("Block per request should be >0.")?)
            .build(),
        jsonrpc_client,
        shutdown,
    ))
}

#[derive(Default)]
struct Subsystems {
    eth_backend: Option<lyquor_seq::eth::Backend>,
    fco: Option<FCO>,
    log: Option<LyquidLog>,
    pool: Option<lyquid::LyquidPool>,
    api: Option<http::HttpServer>,
    db: Option<DBFlusherHandle>,
    discovery: Option<lyquor_lib::discovery::LyquidDiscovery>,
    client: Option<ClientHandle>,
}
debug_struct_name!(Subsystems);

impl Subsystems {
    #[tracing::instrument(level = "trace")]
    async fn stop(&mut self, shutdown: CancellationToken) {
        if let Some(api) = self.api.take() {
            api.stop();
        }
        if let Some(eth_backend) = self.eth_backend.take() {
            eth_backend.shutdown().await;
        }
        if let Some(discovery) = self.discovery.take() {
            discovery.shutdown().await;
        }
        if let Some(pool) = self.pool.take() {
            pool.shutdown().await;
        }
        if let Some(db) = self.db.take() {
            db.stop().await;
        }
        shutdown.cancel();
        if let Some(client) = self.client.take() {
            client.wait_for_shutdown().await;
        }
        if let Some(fco) = self.fco.take() {
            fco.shutdown().await;
        }
        if let Some(log) = self.log.take() {
            log.shutdown().await;
        }
    }
}

struct Node {
    sys: Subsystems,
    config: Arc<NodeConfig>,
    profile: Arc<LyquorProfile>,
    seq_eth_url: String,
    http_tls: Option<http::httptls::HttpTls>,
    started_anvil: bool,
    network: Option<Arc<lyquor_net::hub::Hub>>,
    shutdown: CancellationToken,
}

lyquor_primitives::debug_struct_name!(Node);

impl Node {
    #[tracing::instrument(level = "trace", skip_all)]
    async fn start(mut self) -> anyhow::Result<Self> {
        self.sys = Self::init(
            self.config.clone(),
            self.profile.clone(),
            self.seq_eth_url.clone(),
            self.http_tls.clone(),
            self.started_anvil,
            self.network.take().expect("Node: Network not found."),
            self.shutdown.clone(),
        )
        .await?;
        Ok(self)
    }

    #[tracing::instrument(level = "trace", skip_all)]
    async fn shutdown(&mut self) {
        self.sys.stop(self.shutdown.clone()).await;
    }

    #[tracing::instrument(level = "trace", skip_all)]
    async fn init(
        config: Arc<NodeConfig>, profile: Arc<LyquorProfile>, seq_eth_url: String,
        http_tls: Option<http::httptls::HttpTls>, started_anvil: bool, network: Arc<lyquor_net::hub::Hub>,
        shutdown: CancellationToken,
    ) -> anyhow::Result<Subsystems> {
        let mut sys = Subsystems::default();

        // Set up Ethereum sequence backend.
        let jsonrpc_client = ClientConfig::builder()
            .url(seq_eth_url.parse()?)
            .build()
            .into_client(shutdown.child_token());
        sys.client = Some(jsonrpc_client.clone());
        let sequence_backend_chain_id: u64 = jsonrpc_client
            .request::<EthGetChainId, EthGetChainIdResp>(EthGetChainId)
            .await
            .context("Failed to query sequence backend chain ID.")?
            .0
            .to();
        let sequence_backend_bartender_addr =
            Address::from_str(profile.bartender_addr()).context("Invalid bartender address in profile.")?;
        let resolved_db_dir = config.resolved_db_dir();
        let resolved_image_repo_dir = config.resolved_image_repo_dir();

        //// 1. DB & Storage
        let (store, lyq_store): (Arc<ShadowKVStore<Box<dyn KVStore>>>, Arc<dyn utils::LVMStoreFactory>) =
            match &config.storage.db {
                DB::MemDb => {
                    tracing::warn!("using MemDB, all data will be lost upon exit");
                    let store = Arc::new(ShadowKVStore::new(Box::new(MemDB::new()) as Box<dyn KVStore>));
                    (store, Arc::new(utils::LVMMemStore))
                }
                DB::RocksDb { .. } => {
                    let base_dir = resolved_db_dir.clone();
                    let store = Arc::new(ShadowKVStore::new(
                        Box::new(RocksDB::new(&base_dir.join(profile.id()))?) as Box<dyn KVStore>,
                    ));
                    let lyq_store = Arc::new(utils::LVMDBStore(Arc::new(PrefixedKVStore::new(
                        store.clone(),
                        subkey().lyquid(),
                    ))));

                    if started_anvil {
                        let state_file = anvil_state_path(&base_dir, profile.id());
                        if state_file.exists() {
                            tracing::info!("Using Anvil state file {}.", state_file.display());
                        } else {
                            tracing::info!(
                                "No existing Anvil state file found at {}, starting with fresh state.",
                                state_file.display()
                            );
                        }
                    }

                    (store, lyq_store)
                }
            };
        ////

        //// 2. Fate-Constrainted Ordering Subsystem.
        // Start discovering Lyquid and sequencing contracts.
        let discovery =
            lyquor_lib::discovery::LyquidDiscovery::new(*profile.bartender_id(), sequence_backend_bartender_addr);
        let discovery_handle = discovery.handle();
        sys.discovery = Some(discovery);

        let eth_backend = setup_eth_backend(
            &profile,
            discovery_handle.contract_registry_service(),
            jsonrpc_client.clone(),
            shutdown.child_token(),
        )
        .await?;
        sys.eth_backend = Some(eth_backend.clone());
        let SideEffectRuntime {
            store: side_effect_store,
            provider: side_effect_provider,
        } = build_side_effect_runtime(config.as_ref(), store.clone(), subkey().side_effect_store())?;
        let archive_inbound_store = (!side_effect_provider.is_selective_hosting()).then(|| side_effect_store.clone());

        let fco_store = PrefixedKVStore::new(store.clone(), subkey().fco());
        let fco = FCO::new(
            eth_backend.clone(),
            Arc::new(fco_store),
            Some(profile.init_chain_position()),
            side_effect_provider,
            archive_inbound_store,
        )?;
        let fco_handle = fco.handle();
        sys.fco = Some(fco);
        ////

        //// 3. Lyquor Virtual Machine Subsystem.
        // Configure VM log persistence (store) and runtime log hub (delivery).
        let log_store = PrefixedKVStore::new(store.clone(), subkey().log());
        let vm_log_store: Arc<dyn lyquor_state::SimpleStateStore> =
            Arc::new(lyquor_state::DBSimpleStateStore::new(log_store));
        let log = LyquidLog::new(vm_log_store.clone());
        let log_hub = log.handle();
        sys.log = Some(log);

        let mut lvm = lyquid::LVM::new(
            vm_log_store,
            network.clone(),
            Some(lyquor_vm::Sequencer {
                inter: lyquor_api::interface::InterCallService::new(fco_handle.clone()),
                submit: lyquor_api::interface::SubmitService::new(fco_handle.clone()),
                fetch_oracle_info: lyquor_api::interface::OracleInfoService::new(fco_handle.clone()),
            }),
        );
        ////

        //// 4. UPC peers
        for peer in &config.peers {
            lvm.upc_mut()
                .add_remote(peer.node_id, Some(peer.upc_addr.clone()))
                .await
                .with_context(|| format!("Failed to add UPC peer {}", peer.node_id))?;
        }
        ////

        //// 5. Pool of all Lyquids
        let image_repo = Arc::new(
            DirectoryRepo::new(&resolved_image_repo_dir)
                .await
                .context("Failed to initialize image store.")?,
        );
        let image_repo_for_resolver: Arc<dyn LyquidImageRepo> = image_repo.clone();
        let image_repo_for_http: Arc<dyn OciDistributionStore> = Arc::new(DirectoryOciStore::new(image_repo.clone()));
        let image_resolver = Arc::new(ImageResolver::new(
            image_repo_for_resolver,
            config.image.fallback_repos.clone(),
        ));
        let bartender_id = *profile.bartender_id();
        let pool = lyquid::LyquidPool::new(
            bartender_id,
            lvm,
            lyquid::LyquidPoolSetup {
                fco: fco_handle.clone(),
                store_factory: lyq_store,
                image_resolver,
                side_effect_store,
                log_hub: log_hub.clone(),
                sequence_backend_chain_id,
                sequence_backend_bartender_addr,
                config: lyquid::LyquidConfig::builder().build(),
            },
        )
        .await?;
        let pool_handle = pool.handle();
        sys.pool = Some(pool);
        ////

        // Configure bartender.
        let register_topic = LyteLog::tagged_value_topic("Register");
        let register_count = log_hub.get_number_of_records(bartender_id, register_topic);

        let node_service_context = http::NodeServiceContext::new(config.clone(), network.clone());
        // Start the API server - use the same jsonrpc client as sequencer
        let mut api = lyquor_lib::http::HttpServer::new(
            &config.network.api_addr,
            jsonrpc_client.clone(),
            fco_handle.clone(),
            pool_handle.clone(),
            node_service_context,
            image_repo_for_http,
            shutdown.child_token(),
            http_tls,
        )
        .await?;
        api.start();
        sys.api = Some(api);

        // Wait for bartender to be registered.
        if register_count == 0 {
            tracing::info!("Waiting for bartender to be registered.");
            log_hub
                .wait_next(bartender_id, register_topic)
                .await
                .context("Failed to wait for bartender registration.")?;
        }
        // Bartender, as a normal Lyquid, may have some re-execution, let's make sure it is done.
        let bartender_ins = pool_handle
            .get_bartender()
            .await
            .context("Failed to obtain bartender.")?;
        let bartender_ln = bartender_ins.instance().max_number().unwrap_or(LyquidNumber::ZERO);
        tracing::debug!("Waiting for bartender to reach {bartender_ln}.");
        bartender_ins
            .wait_until(bartender_ln)
            .await
            .context("Failed to wait for the initial re-execution of bartender.")?;

        // Finish up LyquidDiscovery.
        discovery_handle.bind_log_stream(
            log_hub
                .subscribe(bartender_id, register_topic)
                .await
                .context("Failed to subscribe to log system for LyquidDiscovery.")?,
        );
        discovery_handle
            .set_pool(pool_handle.clone())
            .await
            .context("Failed to set pool in LyquidDiscovery.")?;

        tracing::info!("INIT COMPLETE (bartender={bartender_id})");

        // Start a DB writer that persist when node is idle.
        let db = DBFlusher::new(0, store, fco_handle, Duration::from_secs(config.flush_interval_secs)).start();
        sys.db = Some(db);

        Ok(sys)
    }
}

fn generate_unused_port() -> anyhow::Result<u16> {
    let listener = std::net::TcpListener::bind("0.0.0.0:0").context("Failed to bind to local address.")?;
    let port = listener.local_addr().context("Failed to get local address.")?.port();
    Ok(port)
}

fn anvil_state_path(data_dir: &Path, profile_id: &str) -> PathBuf {
    data_dir.join(profile_id).join(ANVIL_STATE_FILENAME)
}

fn start_devnet_anvil(state_file: Option<&Path>, state_interval_secs: u64) -> anyhow::Result<AnvilInstance> {
    let anvil_port = generate_unused_port()?;
    // Mask the SIGINT signal so anvil process won't terminate on SIGINT (it'll only
    // terminate when dropped)
    use nix::sys::signal::{self, SigSet, SigmaskHow, Signal};
    let mut sigset = SigSet::empty();
    sigset.add(Signal::SIGINT);
    let mut old_mask = SigSet::empty();
    signal::sigprocmask(SigmaskHow::SIG_BLOCK, Some(&sigset), Some(&mut old_mask))
        .context("Failed to mask SIGINT for anvil.")?;

    let mut anvil_builder = Anvil::new().port(anvil_port).keep_stdout();
    if let Some(state_file) = state_file {
        if let Some(state_dir) = state_file.parent() {
            std::fs::create_dir_all(state_dir)?;
        }
        anvil_builder = anvil_builder.arg("--state").arg(state_file.as_os_str());
        if state_interval_secs > 0 {
            anvil_builder = anvil_builder
                .arg("--state-interval")
                .arg(state_interval_secs.to_string());
        }
    }
    let mut anvil = anvil_builder.try_spawn()?;

    let stdout_reader = std::io::BufReader::new(
        anvil
            .child_mut()
            .stdout
            .take()
            .context("Failed to read from anvil stdout.")?,
    );

    tokio::task::spawn_blocking(move || {
        use std::io::BufRead;
        let mut stdout_lines = stdout_reader.lines();
        while let Some(line) = stdout_lines.next() {
            tracing::debug!(target: "lyquor_anvil", "{}", line.unwrap_or_else(|_| "".to_string()));
        }
    });

    tracing::info!("Anvil started at port {}.", anvil_port);
    signal::sigprocmask(SigmaskHow::SIG_SETMASK, Some(&old_mask), None)
        .context("Failed to restore old signal mask for anvil.")?;

    Ok(anvil)
}

fn parse_override_value(raw: &str) -> Value {
    if let Ok(v) = raw.parse::<bool>() {
        return Value::from(v);
    }
    if let Ok(v) = raw.parse::<u64>() {
        return Value::from(v);
    }
    if let Ok(v) = raw.parse::<i64>() {
        return Value::from(v);
    }
    if let Ok(v) = raw.parse::<f64>() {
        return Value::from(v);
    }
    Value::from(raw.to_string())
}

fn insert_override_path(overrides: &mut Map<String, Value>, key: &str, value: Value) -> anyhow::Result<()> {
    let mut segments = key.split('.').peekable();
    let mut current = overrides;

    while let Some(segment) = segments.next() {
        if segments.peek().is_none() {
            current.insert(segment.to_string(), value);
            return Ok(());
        }

        let entry = current
            .entry(segment.to_string())
            .or_insert_with(|| Value::Object(Map::<String, Value>::new()));

        match entry {
            Value::Object(dict) => {
                current = dict;
            }
            _ => anyhow::bail!("Override path '{key}' conflicts with non-table value."),
        }
    }

    Ok(())
}

fn build_config_overrides(matches: &clap::ArgMatches) -> anyhow::Result<Map<String, Value>> {
    let mut overrides: Map<String, Value> = Map::new();

    if let Some(entries) = matches.get_many::<String>("config-override") {
        for entry in entries {
            let (key, value) = entry
                .split_once('=')
                .ok_or_else(|| anyhow::anyhow!("Invalid --config-override '{entry}', expected key=value"))?;
            insert_override_path(&mut overrides, key, parse_override_value(value))?;
        }
    }

    Ok(overrides)
}

const NODE_RUNTIME_THREAD_NAME: &str = "lyquor-node";

fn build_node_runtime() -> std::io::Result<tokio::runtime::Runtime> {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .thread_name(NODE_RUNTIME_THREAD_NAME)
        .build()
}

fn main() -> anyhow::Result<()> {
    // Install panic hook to cleanup anvil on panic
    std::panic::set_hook(Box::new(|_| {
        if let Some(&pid) = ANVIL_PID.get() {
            tracing::warn!("Panic detected, killing anvil process {}", pid);
            use nix::sys::signal::{Signal, kill};
            let _ = kill(nix::unistd::Pid::from_raw(pid as i32), Signal::SIGTERM);
        }
    }));

    lyquor_cli::setup_tracing()?;

    println!("{}", lyquor_cli::format_logo_banner());

    let runtime = build_node_runtime().context("Failed to build node Tokio runtime")?;

    runtime.block_on(run_node())
}

async fn run_node() -> anyhow::Result<()> {
    let matches = clap::command!()
        .version(lyquor_cli::build_version())
        .propagate_version(true)
        .arg(clap::arg!(--config <PATH> "Path to the Lyquor config file (toml, yaml, json).").required(false))
        .arg(
            clap::arg!(--"config-override" <KEY_VALUE> "Override config with key=value (repeatable).")
                .required(false)
                .action(clap::ArgAction::Append)
                .value_parser(clap::builder::NonEmptyStringValueParser::new()),
        )
        .get_matches();

    let config_path = matches.get_one::<String>("config").map(PathBuf::from);
    let overrides = build_config_overrides(&matches)?;
    let config = NodeConfig::load(config_path, overrides)?;

    let mut _anvil: Option<AnvilInstance> = None;
    let seq_eth_url = match config.profile.base {
        NetworkType::Devnet => {
            match config.profile.sequencer.as_deref() {
                None => {
                    let state_file = match &config.storage.db {
                        DB::RocksDb { .. } => Some(anvil_state_path(
                            &config.resolved_db_dir(),
                            config.profile.base.as_str(),
                        )),
                        DB::MemDb => None,
                    };
                    let anvil = start_devnet_anvil(state_file.as_deref(), config.flush_interval_secs)?;
                    // Store anvil PID for panic hook
                    ANVIL_PID.set(anvil.child().id()).ok();
                    let ep = anvil.ws_endpoint();
                    _anvil = Some(anvil);
                    ep
                }
                Some(url) => url.to_string(),
            }
        }
        _ => return Err(anyhow::anyhow!("Network is not supported.")),
    };
    let profile: Arc<LyquorProfile> = Arc::new(config.profile.resolve(seq_eth_url.clone())?);
    let node_seed = config.node_key.seed_bytes()?;
    let (_, tls_config) = lyquor_tls::generator::single_node_config_with_seed(&node_seed, profile.dns_suffix());

    if let (Some(external_host), Some(barstrainer_addr)) = (
        config.network.external_host.as_deref(),
        config.network.barstrainer_addr.as_deref(),
    ) {
        match SigningIdentity::from_tls_config(&tls_config) {
            Ok(barstrainer_identity) => {
                let dns_suffix = profile.dns_suffix().to_string();
                let external_host = external_host.to_string();
                let barstrainer_addr = barstrainer_addr.to_string();
                tokio::spawn(async move {
                    if let Err(err) =
                        publish_external_host_dns(&barstrainer_identity, &dns_suffix, &external_host, &barstrainer_addr)
                            .await
                    {
                        tracing::warn!(
                            error = ?err,
                            external_host = %external_host,
                            barstrainer_addr = %barstrainer_addr,
                            "Failed to publish node DNS record"
                        );
                    }
                });
            }
            Err(err) => {
                tracing::warn!(
                    error = ?err,
                    "Skipping external DNS publication because signing identity could not be built"
                );
            }
        }
    } else {
        if config.network.external_host.is_some() {
            tracing::warn!("Skipping external DNS publication because network.barstrainer_addr is not configured");
        }
    }

    let shutdown = CancellationToken::new();

    // Grab TLS cert via ACME DNS-01 and barstrainer
    // This still works without barstrainer_addr because user may supply local certs that is valid
    let http_tls = if let Some(mut acme_config) = config.network.acme.clone() {
        acme_config.storage_path = config
            .resolved_acme_storage_path()
            .expect("resolved ACME storage path should exist when ACME is configured");
        let acme = lyquor_lib::acme::Acme::new_from_acme_config(
            acme_config,
            config.network.barstrainer_addr.clone().unwrap_or_default(),
            profile.dns_suffix().to_owned(),
            &tls_config,
        );

        let http_tls = if let Ok(acme) = acme {
            let domain = barstrainer_client::fqdn_for_node(&tls_config.node_id(), profile.dns_suffix(), None);
            let domains = vec![domain.clone()];

            Ok(http::httptls::HttpTls::new(acme, domains).await)
        } else {
            Err(anyhow::anyhow!("Failed to construct ACME config"))
        };

        match http_tls {
            Ok(t) => Some(t),
            Err(err) => {
                tracing::warn!("Failed to initialize HTTP TLS for api server: {:?}", err);
                None
            }
        }
    } else {
        None
    };

    // Start network hub
    let config = Arc::new(config);
    let mut network = lyquor_net::hub::HubBuilder::new(tls_config, shutdown.child_token())
        .listen_addr(config.network.upc_addr.clone())
        .build()
        .context("Failed to setup network.")?;
    network.start().await.context("Failed to start network.")?;
    let network = Arc::new(network);

    // Signal handlers should be registered before starting any actors, so they'll be stopped
    // gracefully.
    let mut sigint = signal(SignalKind::interrupt()).context("Failed to register SIGINT handler")?;
    let mut sigterm = signal(SignalKind::terminate()).context("Failed to register SIGTERM handler")?;

    let mut node = Node {
        sys: Subsystems::default(),
        config,
        profile: profile.clone(),
        seq_eth_url,
        http_tls,
        started_anvil: _anvil.is_some(),
        network: Some(network.clone()),
        shutdown,
    }
    .start()
    .await?;

    tokio::select! {
        _ = sigint.recv() => {
            tracing::info!("received SIGINT");
        }
        _ = sigterm.recv() => {
            tracing::info!("received SIGTERM");
        }
    }

    node.shutdown().await;
    network.wait_for_shutdown().await;
    Ok(())
}