/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2026 Dyne.org foundation
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*/
use std::{
collections::{HashMap, HashSet},
sync::Arc,
};
use smol::lock::Mutex;
use tracing::{debug, error, info};
use darkfi::{
net::settings::Settings,
rpc::{
jsonrpc::JsonSubscriber,
server::{listen_and_serve, RequestHandler},
settings::RpcSettings,
},
system::{ExecutorPtr, StoppableTask, StoppableTaskPtr},
validator::{Validator, ValidatorConfig, ValidatorPtr},
Error, Result,
};
use darkfi_sdk::crypto::keypair::Network;
#[cfg(test)]
mod tests;
mod error;
use error::{server_error, RpcError};
/// JSON-RPC requests handler and methods
mod rpc;
use rpc::{management::ManagementRpcHandler, DefaultRpcHandler};
/// Validator async tasks
pub mod task;
use task::{consensus::ConsensusInitTaskConfig, consensus_init_task, garbage_collect_task};
/// P2P net protocols
mod proto;
use proto::{DarkfidP2pHandler, DarkfidP2pHandlerPtr};
/// Miners registry
mod registry;
use registry::{DarkfiMinersRegistry, DarkfiMinersRegistryPtr};
/// Atomic pointer to the DarkFi node
pub type DarkfiNodePtr = Arc;
/// Structure representing a DarkFi node
pub struct DarkfiNode {
/// Validator(node) pointer
validator: ValidatorPtr,
/// P2P network protocols handler
p2p_handler: DarkfidP2pHandlerPtr,
/// Node miners registry pointer
registry: DarkfiMinersRegistryPtr,
/// A map of various subscribers exporting live info from the blockchain
subscribers: HashMap<&'static str, JsonSubscriber>,
/// Main JSON-RPC connection tracker
rpc_connections: Mutex>,
/// Management JSON-RPC connection tracker
management_rpc_connections: Mutex>,
}
impl DarkfiNode {
pub async fn new(
validator: ValidatorPtr,
p2p_handler: DarkfidP2pHandlerPtr,
registry: DarkfiMinersRegistryPtr,
subscribers: HashMap<&'static str, JsonSubscriber>,
) -> Result {
Ok(Arc::new(Self {
validator,
p2p_handler,
registry,
subscribers,
rpc_connections: Mutex::new(HashSet::new()),
management_rpc_connections: Mutex::new(HashSet::new()),
}))
}
}
/// Atomic pointer to the DarkFi daemon
pub type DarkfidPtr = Arc;
/// Structure representing a DarkFi daemon
pub struct Darkfid {
/// Darkfi node instance
node: DarkfiNodePtr,
/// `dnet` background task
dnet_task: StoppableTaskPtr,
/// Main JSON-RPC background task
rpc_task: StoppableTaskPtr,
/// Management JSON-RPC background task
management_rpc_task: StoppableTaskPtr,
/// Consensus protocol background task
consensus_task: StoppableTaskPtr,
/// Node garbage collection background task
gc_task: StoppableTaskPtr,
}
impl Darkfid {
/// Initialize a DarkFi daemon.
///
/// Generates a new `DarkfiNode` for provided configuration,
/// along with all the corresponding background tasks.
pub async fn init(
network: Network,
sled_db: &sled_overlay::sled::Db,
config: &ValidatorConfig,
net_settings: &Settings,
ex: &ExecutorPtr,
) -> Result {
info!(target: "darkfid::Darkfid::init", "Initializing a Darkfi daemon...");
// Initialize validator
let validator = Validator::new(sled_db, config).await?;
// Initialize P2P network
let p2p_handler = DarkfidP2pHandler::init(net_settings, ex).await?;
// Initialize the miners registry
let registry = DarkfiMinersRegistry::init(network, &validator).await?;
// Here we initialize various subscribers that can export live blockchain/consensus data.
let mut subscribers = HashMap::new();
subscribers.insert("blocks", JsonSubscriber::new("blockchain.subscribe_blocks"));
subscribers.insert("txs", JsonSubscriber::new("blockchain.subscribe_txs"));
subscribers.insert("proposals", JsonSubscriber::new("blockchain.subscribe_proposals"));
subscribers.insert("dnet", JsonSubscriber::new("dnet.subscribe_events"));
// Initialize node
let node = DarkfiNode::new(validator, p2p_handler, registry, subscribers).await?;
// Generate the background tasks
let dnet_task = StoppableTask::new();
let rpc_task = StoppableTask::new();
let management_rpc_task = StoppableTask::new();
let consensus_task = StoppableTask::new();
let gc_task = StoppableTask::new();
info!(target: "darkfid::Darkfid::init", "Darkfi daemon initialized successfully!");
Ok(Arc::new(Self {
node,
dnet_task,
rpc_task,
management_rpc_task,
consensus_task,
gc_task,
}))
}
/// Start the DarkFi daemon in the given executor, using the
/// provided JSON-RPC settings and consensus initialization
/// configuration.
pub async fn start(
&self,
executor: &ExecutorPtr,
rpc_settings: &RpcSettings,
management_rpc_settings: &RpcSettings,
stratum_rpc_settings: &Option,
mm_rpc_settings: &Option,
config: &ConsensusInitTaskConfig,
) -> Result<()> {
info!(target: "darkfid::Darkfid::start", "Starting Darkfi daemon...");
// Start the `dnet` task
info!(target: "darkfid::Darkfid::start", "Starting dnet subs task");
let dnet_sub_ = self.node.subscribers.get("dnet").unwrap().clone();
let p2p_ = self.node.p2p_handler.p2p.clone();
self.dnet_task.clone().start(
async move {
let dnet_sub = p2p_.dnet_subscribe().await;
loop {
let event = dnet_sub.receive().await;
debug!(target: "darkfid::Darkfid::dnet_task", "Got dnet event: {event:?}");
dnet_sub_.notify(vec![event.into()].into()).await;
}
},
|res| async {
match res {
Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
Err(e) => error!(target: "darkfid::Darkfid::start", "Failed starting dnet subs task: {e}"),
}
},
Error::DetachedTaskStopped,
executor.clone(),
);
// Start the main JSON-RPC task
info!(target: "darkfid::Darkfid::start", "Starting main JSON-RPC server");
let node_ = self.node.clone();
self.rpc_task.clone().start(
listen_and_serve::(rpc_settings.clone(), self.node.clone(), None, executor.clone()),
|res| async move {
match res {
Ok(()) | Err(Error::RpcServerStopped) => >::stop_connections(&node_).await,
Err(e) => error!(target: "darkfid::Darkfid::start", "Failed starting main JSON-RPC server: {e}"),
}
},
Error::RpcServerStopped,
executor.clone(),
);
// Start the management JSON-RPC task
info!(target: "darkfid::Darkfid::start", "Starting management JSON-RPC server");
let node_ = self.node.clone();
self.management_rpc_task.clone().start(
listen_and_serve::(management_rpc_settings.clone(), self.node.clone(), None, executor.clone()),
|res| async move {
match res {
Ok(()) | Err(Error::RpcServerStopped) => >::stop_connections(&node_).await,
Err(e) => error!(target: "darkfid::Darkfid::start", "Failed starting management JSON-RPC server: {e}"),
}
},
Error::RpcServerStopped,
executor.clone(),
);
// Start the miners registry
info!(target: "darkfid::Darkfid::start", "Starting miners registry");
self.node.registry.start(executor, &self.node, stratum_rpc_settings, mm_rpc_settings)?;
// Start the P2P network
info!(target: "darkfid::Darkfid::start", "Starting P2P network");
self.node.p2p_handler.start(executor, &self.node).await?;
// Generate the signal queue smol channel
let (sender, receiver) = smol::channel::unbounded::<()>();
// Start the consensus protocol
info!(target: "darkfid::Darkfid::start", "Starting consensus protocol task");
self.consensus_task.clone().start(
consensus_init_task(
self.node.clone(),
config.clone(),
sender,
),
|res| async move {
match res {
Ok(()) | Err(Error::ConsensusTaskStopped) | Err(Error::MinerTaskStopped) => { /* Do nothing */ }
Err(e) => error!(target: "darkfid::Darkfid::start", "Failed starting consensus initialization task: {e}"),
}
},
Error::ConsensusTaskStopped,
executor.clone(),
);
// Start the garbage collection task
info!(target: "darkfid::Darkfid::start", "Starting garbage collection task");
self.gc_task.clone().start(
garbage_collect_task(receiver, self.node.clone()),
|res| async {
match res {
Ok(()) | Err(Error::GarbageCollectionTaskStopped) => { /* Do nothing */ }
Err(e) => {
error!(target: "darkfid", "Failed starting garbage collection task: {e}")
}
}
},
Error::GarbageCollectionTaskStopped,
executor.clone(),
);
info!(target: "darkfid::Darkfid::start", "Darkfi daemon started successfully!");
Ok(())
}
/// Stop the DarkFi daemon.
pub async fn stop(&self) -> Result<()> {
info!(target: "darkfid::Darkfid::stop", "Terminating Darkfi daemon...");
// Stop the `dnet` node
info!(target: "darkfid::Darkfid::stop", "Stopping dnet subs task...");
self.dnet_task.stop().await;
// Stop the main JSON-RPC task
info!(target: "darkfid::Darkfid::stop", "Stopping main JSON-RPC server...");
self.rpc_task.stop().await;
// Stop the management JSON-RPC task
info!(target: "darkfid::Darkfid::stop", "Stopping management JSON-RPC server...");
self.management_rpc_task.stop().await;
// Stop the miners registry
info!(target: "darkfid::Darkfid::stop", "Stopping miners registry...");
self.node.registry.stop().await;
// Stop the P2P network
info!(target: "darkfid::Darkfid::stop", "Stopping P2P network protocols handler...");
self.node.p2p_handler.stop().await;
// Stop the garbage collection task
info!(target: "darkfid::Darkfid::stop", "Stopping garbage collection task...");
self.gc_task.stop().await;
// Stop the consensus task
info!(target: "darkfid::Darkfid::stop", "Stopping consensus task...");
self.consensus_task.stop().await;
// Flush sled database data
info!(target: "darkfid::Darkfid::stop", "Flushing sled database...");
let flushed_bytes =
self.node.validator.read().await.blockchain.sled_db.flush_async().await?;
info!(target: "darkfid::Darkfid::stop", "Flushed {flushed_bytes} bytes");
info!(target: "darkfid::Darkfid::stop", "Darkfi daemon terminated successfully!");
Ok(())
}
}