/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2025 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 .
*/
//! # Sync Module
//!
//! The `sync` module is responsible for synchronizing the explorer's database with the Darkfi
//! blockchain network. It ensures consistency between the explorer and the blockchain by
//! fetching missing blocks, handling reorganizations (reorgs), and subscribing to live updates
//! through Darkfi's JSON-RPC service.
//!
//! ## Responsibilities
//!
//! - **Block Synchronization**: Handles fetching and storing blocks from a Darkfi
//! blockchain node during startup or when syncing, ensuring the explorer stays synchronized
//! with the latest confirmed blocks.
//! - **Real-Time Updates**: Subscribes to Darkfi's JSON-RPC notification service,
//! allowing the explorer to process and sync new blocks as they are confirmed.
//! - **Reorg Handling**: Detects and resolves blockchain reorganizations by identifying
//! the last common block (in case of divergence) and re-aligning the explorer's state with the
//! latest blockchain state. Reorgs are an importnt part of synchronization because they prevent
//! syncing invalid or outdated states, ensuring the explorer maintains an accurate view of a
//! Darkfi blockchain network.
use std::{sync::Arc, time::Instant};
use log::{debug, error, info, warn};
use tinyjson::JsonValue;
use url::Url;
use darkfi::{
blockchain::BlockInfo,
rpc::{
client::RpcClient,
jsonrpc::{JsonRequest, JsonResult},
},
system::{Publisher, StoppableTask, StoppableTaskPtr},
util::{encoding::base64, time::fmt_duration},
Error,
};
use darkfi_serial::deserialize_async;
use crate::{error::handle_database_error, service::ExplorerService, Explorerd};
impl ExplorerService {
/// Synchronizes blocks between the explorer and a Darkfi blockchain node, ensuring
/// the database remains consistent by syncing any missing or outdated blocks.
///
/// If provided `reset` is true, the explorer's blockchain-related and metric sled trees are purged
/// and syncing starts from the genesis block. The function also handles reorgs by re-aligning the
/// explorer state to the correct height when blocks are outdated. Returns a result indicating
/// success or failure.
///
/// Reorg handling is delegated to the [`Self::reorg_blocks`] function, whose
/// documentation provides more details on the reorg process during block syncing.
pub async fn sync_blocks(&self, reset: bool) -> darkfi::Result<()> {
// Grab last synced block height from the explorer's database.
let last_synced_block = self.last_block().map_err(|e| {
handle_database_error(
"rpc_blocks::sync_blocks",
"[sync_blocks] Retrieving last synced block failed",
e,
)
})?;
// Grab the last confirmed block height and hash from the darkfi node
let (last_darkfid_height, last_darkfid_hash) =
self.darkfid_client.get_last_confirmed_block().await?;
// Initialize the current height to sync from, starting from genesis block if last sync block does not exist
let (last_synced_height, last_synced_hash) = last_synced_block
.map_or((0, "".to_string()), |(height, header_hash)| (height, header_hash));
// Declare a mutable variable to track the current sync height while processing blocks
let mut current_height = last_synced_height;
info!(target: "explorerd::rpc_blocks::sync_blocks", "Syncing from block number: {current_height}");
info!(target: "explorerd::rpc_blocks::sync_blocks", "Last confirmed darkfid block: {last_darkfid_height} - {last_darkfid_hash}");
// A reorg is detected if the hash of the last synced block differs from the hash of the last confirmed block,
// unless the reset flag is set or the current height is 0
let reorg_detected = last_synced_hash != last_darkfid_hash && !reset && current_height != 0;
// If the reset flag is set, reset the explorer state and start syncing from the genesis block height.
// Otherwise, handle reorgs if detected, or proceed to the next block if not at the genesis height.
if reset {
self.reset_explorer_state(0)?;
current_height = 0;
info!(target: "explorerd::rpc_blocks::sync_blocks", "Reset explorer database based on set reset parameter");
} else if reorg_detected {
// Record the start time to measure the duration of potential reorg
let start_reorg_time = Instant::now();
// Process reorg
current_height = self.reorg_blocks(last_synced_height, last_darkfid_height).await?;
// Log only if a reorg occurred (i.e., the explorer wasn't merely catching up to Darkfi node blocks)
if current_height != last_synced_height {
info!(target: "explorerd::rpc_blocks::sync_blocks", "Completed reorg to height: {current_height} [{}]", fmt_duration(start_reorg_time.elapsed()));
}
// Prepare to sync the next block after reorg if not from genesis height
if current_height != 0 {
current_height += 1;
}
} else if current_height != 0 {
// Resume syncing from the block after the last synced height
current_height += 1;
}
// Record the sync start time to measure the total block sync duration
let sync_start_time = Instant::now();
// Track the number of blocks synced for reporting
let mut blocks_synced = 0;
// Sync blocks until the explorer is up to date with the last confirmed block
while current_height <= last_darkfid_height {
// Record the start time to measure the duration it took to sync the block
let block_sync_start = Instant::now();
// Retrieve the block from darkfi node by height
let block = match self.darkfid_client.get_block_by_height(current_height).await {
Ok(r) => r,
Err(e) => {
return Err(handle_database_error(
"rpc_blocks::sync_blocks",
"[sync_blocks] RPC client request failed",
e,
))
}
};
// Store the retrieved block in the explorer's database
if let Err(e) = self.put_block(&block).await {
return Err(handle_database_error(
"rpc_blocks::sync_blocks",
"[sync_blocks] Put block failed",
e,
));
};
debug!(
target: "explorerd::rpc_blocks::sync_blocks",
"Synced block {current_height} [{}]",
fmt_duration(block_sync_start.elapsed())
);
// Increment the current height to sync the next block
current_height += 1;
// Increment the count of successfully synced blocks
blocks_synced += 1;
}
info!(
target: "explorerd::rpc_blocks::sync_blocks",
"Synced {} blocks: explorer blocks total {} [{}]",
blocks_synced,
self.db.blockchain.blocks.len(),
fmt_duration(sync_start_time.elapsed()),
);
Ok(())
}
/// Handles blockchain reorganizations (reorgs) during the explorer node's startup synchronization
/// with Darkfi nodes, ensuring the explorer provides a consistent and accurate view of the blockchain.
///
/// A reorg occurs when the blocks stored by the blockchain nodes diverge from those stored by the explorer.
/// This function resolves inconsistencies by identifying the point of divergence, searching backward through
/// block heights, and comparing block hashes between the explorer database and the blockchain node. Once a
/// common block height is found, the explorer is re-aligned to that height.
///
/// If no common block can be found, the explorer resets to the "genesis height," removing all blocks,
/// transactions, and metrics from its database to resynchronize with the canonical chain from the nodes.
///
/// Returns the last height at which the explorer's state was successfully re-aligned with the blockchain.
async fn reorg_blocks(
&self,
last_synced_height: u32,
last_darkfid_height: u32,
) -> darkfi::Result {
// Log reorg detection in the case that explorer height is greater or equal to height of darkfi node
if last_synced_height >= last_darkfid_height {
info!(target: "explorerd::rpc_blocks::process_sync_blocks_reorg",
"Reorg detected with heights: explorer.{last_synced_height} >= darkfid.{last_darkfid_height}");
}
// Declare a mutable variable to track the current height while searching for a common block
let mut cur_height = last_synced_height;
// Search for an explorer block that matches a darkfi node block
while cur_height > 0 {
let synced_block = self.get_block_by_height(cur_height)?;
debug!(target: "explorerd::rpc_blocks::process_sync_blocks_reorg", "Searching for common block: {}", cur_height);
// Check if we found a synced block for current height being searched
if let Some(synced_block) = synced_block {
// Fetch the block from darkfi node to check for a match
match self.darkfid_client.get_block_by_height(cur_height).await {
Ok(darkfid_block) => {
// If hashes match, we've found the point of divergence
if synced_block.header_hash == darkfid_block.hash().to_string() {
// If hashes match but the cur_height differs from the last synced height, reset the explorer state
if cur_height != last_synced_height {
self.reset_explorer_state(cur_height)?;
debug!(target: "explorerd::rpc_blocks::process_sync_blocks_reorg", "Completed reorg to height: {cur_height}");
}
break;
} else {
// Log reorg detection with height and header hash mismatch details
if cur_height == last_synced_height {
info!(
target: "explorerd::rpc_blocks::process_sync_blocks_reorg",
"Reorg detected at height {}: explorer.{} != darkfid.{}",
cur_height,
synced_block.header_hash,
darkfid_block.hash()
);
}
}
}
// Continue searching for blocks that do not exist on darkfi nodes
Err(Error::JsonRpcError((-32121, _))) => (),
Err(e) => {
return Err(handle_database_error(
"rpc_blocks::process_sync_blocks_reorg",
"[process_sync_blocks_reorg] RPC client request failed",
e,
))
}
}
}
// Move to previous block to search for a match
cur_height = cur_height.saturating_sub(1);
}
// Check if genesis block reorg is needed
if cur_height == 0 {
self.reset_explorer_state(0)?;
}
// Return the last height we reorged to
Ok(cur_height)
}
}
/// Subscribes to darkfid's JSON-RPC notification endpoint that serves
/// new confirmed blocks. Upon receiving them, store them to the database.
pub async fn subscribe_sync_blocks(
explorer: Arc,
endpoint: Url,
ex: Arc>,
) -> darkfi::Result<(StoppableTaskPtr, StoppableTaskPtr)> {
// Grab last confirmed block
let (last_darkfid_height, last_darkfid_hash) =
explorer.darkfid_client.get_last_confirmed_block().await?;
// Grab last synced block
let (mut height, hash) = match explorer.service.last_block() {
Ok(Some((height, hash))) => (height, hash),
Ok(None) => (0, "".to_string()),
Err(e) => {
return Err(Error::DatabaseError(format!(
"[subscribe_blocks] Retrieving last synced block failed: {e:?}"
)))
}
};
// Evaluates whether there is a mismatch between the last confirmed block and the last synced block
let blocks_mismatch = (last_darkfid_height != height || last_darkfid_hash != hash)
&& last_darkfid_height != 0
&& height != 0;
// Check if there is a mismatch, throwing an error to prevent operating in a potentially inconsistent state
if blocks_mismatch {
warn!(target: "explorerd::rpc_blocks::subscribe_blocks",
"Warning: Last synced block is not the last confirmed block: \
last_darkfid_height={last_darkfid_height}, last_synced_height={height}, last_darkfid_hash={last_darkfid_hash}, last_synced_hash={hash}");
warn!(target: "explorerd::rpc_blocks::subscribe_blocks", "You should first fully sync the blockchain, and then subscribe");
return Err(Error::DatabaseError(
"[subscribe_blocks] Blockchain not fully synced".to_string(),
));
}
info!(target: "explorerd::rpc_blocks::subscribe_blocks", "Subscribing to receive notifications of incoming blocks");
let publisher = Publisher::new();
let subscription = publisher.clone().subscribe().await;
let _ex = ex.clone();
let subscriber_task = StoppableTask::new();
subscriber_task.clone().start(
// Weird hack to prevent lifetimes hell
async move {
let ex = _ex.clone();
let rpc_client = RpcClient::new(endpoint, ex).await?;
let req = JsonRequest::new("blockchain.subscribe_blocks", JsonValue::Array(vec![]));
rpc_client.subscribe(req, publisher).await
},
|res| async move {
match res {
Ok(()) => { /* Do nothing */ }
Err(e) => error!(target: "explorerd::rpc_blocks::subscribe_blocks", "[subscribe_blocks] JSON-RPC server error: {e:?}"),
}
},
Error::RpcServerStopped,
ex.clone(),
);
info!(target: "explorerd::rpc_blocks::subscribe_blocks", "Detached subscription to background");
let listener_task = StoppableTask::new();
listener_task.clone().start(
// Weird hack to prevent lifetimes hell
async move {
loop {
match subscription.receive().await {
JsonResult::Notification(n) => {
debug!(target: "explorerd::rpc_blocks::subscribe_blocks", "Got Block notification from darkfid subscription");
if n.method != "blockchain.subscribe_blocks" {
return Err(Error::UnexpectedJsonRpc(format!(
"Got foreign notification from darkfid: {}",
n.method
)))
}
// Verify parameters
if !n.params.is_array() {
return Err(Error::UnexpectedJsonRpc(
"Received notification params are not an array".to_string(),
))
}
let params = n.params.get::>().unwrap();
if params.is_empty() {
return Err(Error::UnexpectedJsonRpc(
"Notification parameters are empty".to_string(),
))
}
for param in params {
let param = param.get::().unwrap();
let bytes = base64::decode(param).unwrap();
let darkfid_block: BlockInfo = match deserialize_async(&bytes).await {
Ok(b) => b,
Err(e) => {
return Err(Error::UnexpectedJsonRpc(format!(
"[subscribe_blocks] Deserializing block failed: {e:?}"
)))
},
};
info!(target: "explorerd::rpc_blocks::subscribe_blocks", "========================================================================================");
info!(target: "explorerd::rpc_blocks::subscribe_blocks", "| Block Notification: {} |", darkfid_block.hash());
info!(target: "explorerd::rpc_blocks::subscribe_blocks", "========================================================================================");
// Store darkfi node block height for later use
let darkfid_block_height = darkfid_block.header.height;
// Check if we need to perform a reorg due to mismatch in block heights
if darkfid_block_height <= height {
info!(target: "explorerd::rpc_blocks::subscribe_blocks",
"Reorg detected with heights: darkfid.{darkfid_block_height} <= explorer.{height}");
// Calculate the reset height
let reset_height = darkfid_block_height.saturating_sub(1);
// Record the start time to measure the duration of the reorg
let start_reorg_time = Instant::now();
// Execute the reorg by resetting the explorer state to reset height
explorer.service.reset_explorer_state(reset_height)?;
info!(target: "explorerd::rpc_blocks::subscribe_blocks", "Completed reorg to height: {reset_height} [{}]", fmt_duration(start_reorg_time.elapsed()));
}
// Record the start time to measure the duration to store the block
let start_reorg_time = Instant::now();
if let Err(e) = explorer.service.put_block(&darkfid_block).await {
return Err(Error::DatabaseError(format!(
"[subscribe_blocks] Put block failed: {e:?}"
)))
}
info!(target: "explorerd::rpc_blocks::subscribe_blocks", "Stored new block at height: {} [{}]", darkfid_block.header.height, fmt_duration(start_reorg_time.elapsed()));
// Process the next block
height = darkfid_block.header.height;
}
}
JsonResult::Error(e) => {
// Some error happened in the transmission
return Err(Error::UnexpectedJsonRpc(format!("Got error from JSON-RPC: {e:?}")))
}
x => {
// And this is weird
return Err(Error::UnexpectedJsonRpc(format!(
"Got unexpected data from JSON-RPC: {x:?}"
)))
}
}
};
},
|res| async move {
match res {
Ok(()) => { /* Do nothing */ }
Err(e) => error!(target: "explorerd::rpc_blocks::subscribe_blocks", "[subscribe_blocks] JSON-RPC server error: {e:?}"),
}
},
Error::RpcServerStopped,
ex,
);
Ok((subscriber_task, listener_task))
}