pub mod bundle;
pub mod db_util;
pub mod main;
use crate::auth::{create_token, generate_priv_key, ADMIN, JWT_IDENTIFIER};
use crate::blocks::Tipset;
use crate::chain::ChainStore;
use crate::chain_sync::ChainMuxer;
use crate::cli_shared::{car_db_path, snapshot};
use crate::cli_shared::{
chain_path,
cli::{CliOpts, Config},
};
use crate::daemon::db_util::{
import_chain_as_forest_car, load_all_forest_cars, populate_eth_mappings,
};
use crate::db::car::ManyCar;
use crate::db::db_engine::{db_root, open_db};
use crate::db::{ttl::EthMappingCollector, MarkAndSweep, MemoryDB, SettingsExt, CAR_DB_DIR_NAME};
use crate::genesis::{get_network_name_from_genesis, read_genesis_header};
use crate::key_management::{
KeyStore, KeyStoreConfig, ENCRYPTED_KEYSTORE_NAME, FOREST_KEYSTORE_PHRASE_ENV,
};
use crate::libp2p::{Libp2pConfig, Libp2pService, PeerManager};
use crate::message_pool::{MessagePool, MpoolConfig, MpoolRpcProvider};
use crate::networks::{self, ChainConfig, NetworkChain};
use crate::rpc::eth::filter::EthEventHandler;
use crate::rpc::start_rpc;
use crate::rpc::RPCState;
use crate::shim::address::{CurrentNetwork, Network};
use crate::shim::clock::ChainEpoch;
use crate::shim::version::NetworkVersion;
use crate::state_manager::StateManager;
use crate::utils::{
monitoring::MemStatsTracker, proofs_api::ensure_params_downloaded,
version::FOREST_VERSION_STRING,
};
use anyhow::{bail, Context as _};
use bundle::load_actor_bundles;
use dialoguer::console::Term;
use dialoguer::theme::ColorfulTheme;
use futures::{select, Future, FutureExt};
use fvm_ipld_blockstore::Blockstore;
use once_cell::sync::Lazy;
use raw_sync_2::events::{Event, EventInit as _, EventState};
use shared_memory::ShmemConf;
use std::path::Path;
use std::time::Duration;
use std::{cell::RefCell, cmp, path::PathBuf, sync::Arc};
use tempfile::{Builder, TempPath};
use tokio::{
net::TcpListener,
signal::{
ctrl_c,
unix::{signal, SignalKind},
},
sync::{mpsc, RwLock},
task::JoinSet,
};
use tracing::{debug, info, warn};
static IPC_PATH: Lazy<TempPath> = Lazy::new(|| {
Builder::new()
.prefix("forest-ipc")
.tempfile()
.expect("tempfile must succeed")
.into_temp_path()
});
pub fn ipc_shmem_conf() -> ShmemConf {
ShmemConf::new()
.size(Event::size_of(None))
.force_create_flink()
.flink(IPC_PATH.as_os_str())
}
fn unblock_parent_process() -> anyhow::Result<()> {
let shmem = ipc_shmem_conf().open()?;
let (event, _) =
unsafe { Event::from_existing(shmem.as_ptr()).map_err(|err| anyhow::anyhow!("{err}")) }?;
event
.set(EventState::Signaled)
.map_err(|err| anyhow::anyhow!("{err}"))
}
fn maybe_increase_fd_limit() -> anyhow::Result<()> {
static DESIRED_SOFT_LIMIT: u64 = 8192;
let (soft_before, _) = rlimit::Resource::NOFILE.get()?;
let soft_after = rlimit::increase_nofile_limit(DESIRED_SOFT_LIMIT)?;
if soft_before < soft_after {
debug!("Increased file descriptor limit from {soft_before} to {soft_after}");
}
if soft_after < DESIRED_SOFT_LIMIT {
warn!(
"File descriptor limit is too low: {soft_after} < {DESIRED_SOFT_LIMIT}. \
You may encounter 'too many open files' errors.",
);
}
Ok(())
}
pub async fn start_interruptable(opts: CliOpts, config: Config) -> anyhow::Result<()> {
let mut terminate = signal(SignalKind::terminate())?;
let (shutdown_send, mut shutdown_recv) = mpsc::channel(1);
let result = tokio::select! {
ret = start(opts, config, shutdown_send) => ret,
_ = ctrl_c() => {
info!("Keyboard interrupt.");
Ok(())
},
_ = terminate.recv() => {
info!("Received SIGTERM.");
Ok(())
},
_ = shutdown_recv.recv() => {
info!("Client requested a shutdown.");
Ok(())
},
};
crate::utils::io::terminal_cleanup();
result
}
const GC_INTERVAL: Duration = Duration::from_secs(60 * 60 * 10);
pub(super) async fn start(
opts: CliOpts,
config: Config,
shutdown_send: mpsc::Sender<()>,
) -> anyhow::Result<()> {
let chain_config = Arc::new(ChainConfig::from_chain(&config.chain));
if chain_config.is_testnet() {
CurrentNetwork::set_global(Network::Testnet);
}
info!(
"Starting Forest daemon, version {}",
FOREST_VERSION_STRING.as_str()
);
maybe_increase_fd_limit()?;
let start_time = chrono::Utc::now();
let path: PathBuf = config.client.data_dir.join("libp2p");
let net_keypair = crate::libp2p::keypair::get_or_create_keypair(&path)?;
let mut keystore = load_or_create_keystore(&config).await?;
if keystore.get(JWT_IDENTIFIER).is_err() {
keystore.put(JWT_IDENTIFIER, generate_priv_key())?;
}
handle_admin_token(&opts, &keystore)?;
let keystore = Arc::new(RwLock::new(keystore));
let chain_data_path = chain_path(&config);
let db_migration = crate::db::migration::DbMigration::new(&config);
if let Err(e) = db_migration.migrate() {
warn!("Failed to migrate database: {e}");
}
let db_root_dir = db_root(&chain_data_path)?;
let db_writer = Arc::new(open_db(db_root_dir.clone(), config.db_config().clone())?);
let db = Arc::new(ManyCar::new(db_writer.clone()));
let forest_car_db_dir = db_root_dir.join(CAR_DB_DIR_NAME);
load_all_forest_cars(&db, &forest_car_db_dir)?;
if config.client.load_actors && !opts.stateless {
load_actor_bundles(&db, &config.chain).await?;
}
let mut services = JoinSet::new();
if opts.track_peak_rss {
let mem_stats_tracker = MemStatsTracker::default();
services.spawn(async move {
mem_stats_tracker.run_loop().await;
Ok(())
});
}
let genesis_header = read_genesis_header(
config.client.genesis_file.as_deref(),
chain_config.genesis_bytes(&db).await?.as_deref(),
&db,
)
.await?;
if config.client.enable_metrics_endpoint {
let prometheus_listener = TcpListener::bind(config.client.metrics_address)
.await
.with_context(|| format!("could not bind to {}", config.client.metrics_address))?;
info!(
"Prometheus server started at {}",
config.client.metrics_address
);
let db_directory = crate::db::db_engine::db_root(&chain_path(&config))?;
let db = db.writer().clone();
services.spawn(async {
crate::metrics::init_prometheus(prometheus_listener, db_directory, db)
.await
.context("Failed to initiate prometheus server")
});
crate::metrics::default_registry().register_collector(Box::new(
networks::metrics::NetworkHeightCollector::new(
chain_config.block_delay_secs,
genesis_header.timestamp,
),
));
}
let chain_store = Arc::new(ChainStore::new(
Arc::clone(&db),
db.writer().clone(),
db.writer().clone(),
chain_config.clone(),
genesis_header.clone(),
)?);
if !opts.no_gc {
let mut db_garbage_collector = {
let chain_store = chain_store.clone();
let depth = cmp::max(
chain_config.policy.chain_finality * 2,
config.sync.recent_state_roots,
);
let get_heaviest_tipset = Box::new(move || chain_store.heaviest_tipset());
MarkAndSweep::new(
db_writer,
get_heaviest_tipset,
depth,
Duration::from_secs(chain_config.block_delay_secs as u64),
)
};
services.spawn(async move { db_garbage_collector.gc_loop(GC_INTERVAL).await });
}
if let Some(ttl) = config.client.eth_mapping_ttl {
let chain_store = chain_store.clone();
let chain_config = chain_config.clone();
services.spawn(async move {
tracing::info!("Starting collector for eth_mappings");
let mut collector = EthMappingCollector::new(
chain_store.db.clone(),
chain_config.eth_chain_id,
Duration::from_secs(ttl.into()),
);
collector.run().await
});
}
let publisher = chain_store.publisher();
let sm = StateManager::new(
Arc::clone(&chain_store),
Arc::clone(&chain_config),
Arc::new(config.sync.clone()),
)?;
let state_manager = Arc::new(sm);
let network_name = get_network_name_from_genesis(&genesis_header, &state_manager)?;
info!("Using network :: {}", get_actual_chain_name(&network_name));
display_chain_logo(&config.chain);
let (tipset_sender, tipset_receiver) = flume::bounded(20);
let config = if config.network.bootstrap_peers.is_empty() {
let bootstrap_peers = chain_config.bootstrap_peers.clone();
Config {
network: Libp2pConfig {
bootstrap_peers,
..config.network
},
..config
}
} else {
config
};
if opts.exit_after_init {
return Ok(());
}
let epoch = chain_store.heaviest_tipset().epoch();
let peer_manager = Arc::new(PeerManager::default());
services.spawn(peer_manager.clone().peer_operation_event_loop_task());
let genesis_cid = *genesis_header.cid();
let p2p_service = Libp2pService::new(
config.network.clone(),
Arc::clone(&chain_store),
peer_manager.clone(),
net_keypair,
&network_name,
genesis_cid,
)
.await?;
let network_rx = p2p_service.network_receiver();
let network_send = p2p_service.network_sender();
let provider = MpoolRpcProvider::new(publisher.clone(), Arc::clone(&state_manager));
let mpool = MessagePool::new(
provider,
network_name.clone(),
network_send.clone(),
MpoolConfig::load_config(db.writer().as_ref())?,
state_manager.chain_config().clone(),
&mut services,
)?;
let mpool = Arc::new(mpool);
let chain_muxer = ChainMuxer::new(
Arc::clone(&state_manager),
peer_manager.clone(),
mpool.clone(),
network_send.clone(),
network_rx,
Arc::new(Tipset::from(&genesis_header)),
tipset_sender.clone(),
tipset_receiver,
opts.stateless,
)?;
let bad_blocks = chain_muxer.bad_blocks_cloned();
let sync_state = chain_muxer.sync_state_cloned();
services.spawn(async { Err(anyhow::anyhow!("{}", chain_muxer.await)) });
if config.client.enable_health_check {
let forest_state = crate::health::ForestState {
config: config.clone(),
chain_config: chain_config.clone(),
genesis_timestamp: genesis_header.timestamp,
sync_state: sync_state.clone(),
peer_manager,
settings_store: chain_store.settings(),
};
let listener =
tokio::net::TcpListener::bind(forest_state.config.client.healthcheck_address).await?;
services.spawn(async move {
crate::health::init_healthcheck_server(forest_state, listener)
.await
.context("Failed to initiate healthcheck server")
});
}
if config.client.enable_rpc {
let keystore_rpc = Arc::clone(&keystore);
let rpc_state_manager = Arc::clone(&state_manager);
let rpc_address = config.client.rpc_address;
info!("JSON-RPC endpoint will listen at {rpc_address}");
services.spawn(async move {
start_rpc(
RPCState {
state_manager: Arc::clone(&rpc_state_manager),
keystore: keystore_rpc,
mpool,
bad_blocks,
sync_state,
eth_event_handler: Arc::new(EthEventHandler::new()),
network_send,
network_name,
start_time,
shutdown: shutdown_send,
tipset_send: tipset_sender,
},
rpc_address,
)
.await
});
} else {
debug!("RPC disabled.");
};
if opts.detach {
unblock_parent_process()?;
}
crate::utils::proofs_api::set_proofs_parameter_cache_dir_env(&config.client.data_dir);
let mut config = config;
if config.client.snapshot_path.is_none() && !opts.stateless {
set_snapshot_path_if_needed(
&mut config,
&chain_config,
epoch,
opts.auto_download_snapshot,
&db_root_dir,
)
.await?;
}
if !opts.skip_load.unwrap_or_default() {
if let Some(path) = &config.client.snapshot_path {
let (car_db_path, ts) =
import_chain_as_forest_car(path, &forest_car_db_dir, config.client.import_mode)
.await?;
db.read_only_files(std::iter::once(car_db_path.clone()))?;
debug!("Loaded car DB at {}", car_db_path.display());
state_manager
.chain_store()
.set_heaviest_tipset(Arc::new(ts.clone()))?;
}
}
if let Some(validate_from) = config.client.snapshot_height {
ensure_params_downloaded().await?;
let current_height = config
.client
.snapshot_head
.unwrap_or(state_manager.chain_store().heaviest_tipset().epoch());
assert!(current_height.is_positive());
match validate_from.is_negative() {
true => {
state_manager.validate_range((current_height + validate_from)..=current_height)?
}
false => state_manager.validate_range(validate_from..=current_height)?,
}
}
if opts.halt_after_import {
services.shutdown().await;
return Ok(());
}
if !opts.stateless && !chain_config.is_devnet() {
let state_manager = Arc::clone(&state_manager);
services.spawn(async move {
if let Err(err) = init_ethereum_mapping(state_manager, &config) {
tracing::warn!("Init Ethereum mapping failed: {}", err)
}
Ok(())
});
}
if !opts.stateless {
ensure_params_downloaded().await?;
}
services.spawn(p2p_service.run());
propagate_error(&mut services)
.await
.context("services failure")
.map(|_| {})
}
async fn set_snapshot_path_if_needed(
config: &mut Config,
chain_config: &ChainConfig,
epoch: ChainEpoch,
auto_download_snapshot: bool,
download_directory: &Path,
) -> anyhow::Result<()> {
if !download_directory.is_dir() {
anyhow::bail!(
"`download_directory` does not exist: {}",
download_directory.display()
);
}
let vendor = snapshot::TrustedVendor::default();
let chain = &config.chain;
let network_version = chain_config.network_version(epoch);
let network_version_is_small = network_version < NetworkVersion::V16;
let require_a_snapshot = network_version_is_small;
let have_a_snapshot = config.client.snapshot_path.is_some();
match (require_a_snapshot, have_a_snapshot, auto_download_snapshot) {
(false, _, _) => {} (true, true, _) => {} (true, false, true) => {
let url = crate::cli_shared::snapshot::stable_url(vendor, chain)?;
config.client.snapshot_path = Some(url.to_string().into());
}
(true, false, false) => {
let (url, num_bytes, _path) = crate::cli_shared::snapshot::peek(vendor, chain)
.await
.context("couldn't get snapshot size")?;
println!("Forest requires a snapshot to sync with the network, but automatic fetching is disabled.");
let message = format!(
"Fetch a {} snapshot to the current directory? (denying will exit the program). ",
indicatif::HumanBytes(num_bytes)
);
let have_permission = asyncify(|| {
dialoguer::Confirm::with_theme(&ColorfulTheme::default())
.with_prompt(message)
.default(false)
.interact()
.unwrap_or(false)
})
.await;
if !have_permission {
bail!("Forest requires a snapshot to sync with the network, but automatic fetching is disabled.")
}
config.client.snapshot_path = Some(url.to_string().into());
}
};
Ok(())
}
fn handle_admin_token(opts: &CliOpts, keystore: &KeyStore) -> anyhow::Result<()> {
let ki = keystore.get(JWT_IDENTIFIER)?;
let token_exp = chrono::Duration::days(365 * 100);
let token = create_token(
ADMIN.iter().map(ToString::to_string).collect(),
ki.private_key(),
token_exp,
)?;
info!("Admin token: {token}");
if let Some(path) = opts.save_token.as_ref() {
std::fs::write(path, token)?;
}
Ok(())
}
async fn propagate_error(
services: &mut JoinSet<Result<(), anyhow::Error>>,
) -> anyhow::Result<std::convert::Infallible> {
while !services.is_empty() {
select! {
option = services.join_next().fuse() => {
if let Some(Ok(Err(error_message))) = option {
return Err(error_message)
}
},
}
}
std::future::pending().await
}
pub fn get_actual_chain_name(internal_network_name: &str) -> &str {
match internal_network_name {
"testnetnet" => "mainnet",
"calibrationnet" => "calibnet",
_ => internal_network_name,
}
}
async fn load_or_create_keystore(config: &Config) -> anyhow::Result<KeyStore> {
use std::env::VarError;
let passphrase_from_env = std::env::var(FOREST_KEYSTORE_PHRASE_ENV);
let require_encryption = config.client.encrypt_keystore;
let keystore_already_exists = config
.client
.data_dir
.join(ENCRYPTED_KEYSTORE_NAME)
.is_dir();
match (require_encryption, passphrase_from_env) {
(false, maybe_passphrase) => {
warn!("Forest has encryption disabled");
if let Ok(_) | Err(VarError::NotUnicode(_)) = maybe_passphrase {
warn!(
"Ignoring passphrase provided in {} - encryption is disabled",
FOREST_KEYSTORE_PHRASE_ENV
)
}
KeyStore::new(KeyStoreConfig::Persistent(config.client.data_dir.clone()))
.map_err(anyhow::Error::new)
}
(true, Ok(passphrase)) => KeyStore::new(KeyStoreConfig::Encrypted(
config.client.data_dir.clone(),
passphrase,
))
.map_err(anyhow::Error::new),
(true, Err(error)) => {
if let VarError::NotUnicode(_) = error {
warn!(
"Ignoring passphrase provided in {} - it's not utf-8",
FOREST_KEYSTORE_PHRASE_ENV
)
}
let data_dir = config.client.data_dir.clone();
match keystore_already_exists {
true => asyncify(move || input_password_to_load_encrypted_keystore(data_dir))
.await
.context("Couldn't load keystore"),
false => {
let password =
asyncify(|| create_password("Create a password for Forest's keystore"))
.await?;
KeyStore::new(KeyStoreConfig::Encrypted(data_dir, password))
.context("Couldn't create keystore")
}
}
}
}
}
fn asyncify<T>(f: impl FnOnce() -> T + Send + 'static) -> impl Future<Output = T>
where
T: Send + 'static,
{
tokio::task::spawn_blocking(f).then(|res| async { res.expect("spawned task panicked") })
}
fn input_password_to_load_encrypted_keystore(data_dir: PathBuf) -> dialoguer::Result<KeyStore> {
let keystore = RefCell::new(None);
let term = Term::stderr();
if !term.is_term() {
return Err(std::io::Error::new(
std::io::ErrorKind::NotConnected,
"cannot read password from non-terminal",
)
.into());
}
dialoguer::Password::new()
.with_prompt("Enter the password for Forest's keystore")
.allow_empty_password(true) .validate_with(|input: &String| {
KeyStore::new(KeyStoreConfig::Encrypted(data_dir.clone(), input.clone()))
.map(|created| *keystore.borrow_mut() = Some(created))
.context(
"Error: couldn't load keystore with this password. Try again or press Ctrl+C to abort.",
)
})
.interact_on(&term)?;
Ok(keystore
.into_inner()
.expect("validation succeeded, so keystore must be emplaced"))
}
fn create_password(prompt: &str) -> dialoguer::Result<String> {
let term = Term::stderr();
if !term.is_term() {
return Err(std::io::Error::new(
std::io::ErrorKind::NotConnected,
"cannot read password from non-terminal",
)
.into());
}
dialoguer::Password::new()
.with_prompt(prompt)
.allow_empty_password(false)
.with_confirmation(
"Confirm password",
"Error: the passwords do not match. Try again or press Ctrl+C to abort.",
)
.interact_on(&term)
}
fn display_chain_logo(chain: &NetworkChain) {
let logo = match chain {
NetworkChain::Butterflynet => {
Some(include_str!("../../build/ascii-art/butterfly").to_string())
}
_ => None,
};
if let Some(logo) = logo {
info!("\n{logo}");
}
}
fn init_ethereum_mapping<DB: Blockstore>(
state_manager: Arc<StateManager<DB>>,
config: &Config,
) -> anyhow::Result<()> {
match state_manager
.chain_store()
.settings()
.eth_mapping_up_to_date()?
{
Some(false) | None => {
let car_db_path = car_db_path(config)?;
let db: Arc<ManyCar<MemoryDB>> = Arc::default();
load_all_forest_cars(&db, &car_db_path)?;
let ts = db.heaviest_tipset()?;
populate_eth_mappings(&state_manager, &ts)?;
state_manager
.chain_store()
.settings()
.set_eth_mapping_up_to_date()
}
Some(true) => {
tracing::info!("Ethereum mapping up to date");
Ok(())
}
}
}