explorerd: introduce dedicated sync module for block synchronization

This update introduces a dedicated `sync` module, moving explorer synchronization-related logic from the RPC layer to the service layer. The RPC block code has been separated from the explorer's synchronization and reorganization logic. By creating a `sync` module, synchronization—one of the most complex and integral aspects of the explorer's functionality—has been isolated. This restructuring provides a cleaner separation of concerns and simplifies the management of synchronization complexity.

Updates include:
- Created `explorer/service/sync` module and moved the following functions from `rpc/blocks.rs`:
  - `sync_blocks`
  - `subscribe_blocks`
  - `reorg_blocks` (renamed from process_sync_blocks_reorg)
- Updated `mod.rs` to include the `sync` module definition
- Added `darkfid_client` to `ExplorerService` to fetch blocks from a Darkfi blockchain network during sync
- Updated the `contracts.rs` test module setup method to use the new `ExplorerService` constructor

bin/explorer/explorerd/src/main.rs

@@ -33,8 +33,8 @@ use darkfi::{
 
 use crate::{
     config::ExplorerNetworkConfig,
-    rpc::{blocks::subscribe_blocks, DarkfidRpcClient},
-    service::ExplorerService,
+    rpc::DarkfidRpcClient,
+    service::{sync::subscribe_sync_blocks, ExplorerService},
 };
 
 /// Configuration management across multiple networks (localnet, testnet, mainnet)
@@ -123,7 +123,7 @@ impl Explorerd {
         let darkfid_client = Arc::new(DarkfidRpcClient::new());
 
         // Create explorer service
-        let service = ExplorerService::new(db_path)?;
+        let service = ExplorerService::new(db_path, darkfid_client.clone())?;
 
         // Initialize the explorer service
         service.init().await?;
@@ -189,7 +189,7 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
 
         // Sync blocks
         info!(target: "explorerd", "Syncing blocks from darkfid...");
-        if let Err(e) = explorer.sync_blocks(args.reset).await {
+        if let Err(e) = explorer.service.sync_blocks(args.reset).await {
             let error_message = format!("Error syncing blocks: {:?}", e);
             error!(target: "explorerd", "{error_message}");
             return Err(Error::DatabaseError(error_message));
@@ -197,7 +197,7 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'static>>) -> Result<()> {
 
         // Subscribe blocks
         info!(target: "explorerd", "Subscribing to new blocks...");
-        match subscribe_blocks(explorer.clone(), config.endpoint.clone(), ex.clone()).await {
+        match subscribe_sync_blocks(explorer.clone(), config.endpoint.clone(), ex.clone()).await {
             Ok((sub_task, lst_task)) => {
                 subscriber_task = Some(sub_task);
                 listener_task = Some(lst_task);

bin/explorer/explorerd/src/rpc/blocks.rs

@@ -16,32 +16,25 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{sync::Arc, time::Instant};
-
-use log::{debug, error, info, warn};
+use log::error;
 use tinyjson::JsonValue;
-use url::Url;
 
 use darkfi::{
     blockchain::BlockInfo,
-    rpc::{
-        client::RpcClient,
-        jsonrpc::{
-            ErrorCode::{InternalError, InvalidParams, ParseError},
-            JsonError, JsonRequest, JsonResponse, JsonResult,
-        },
+    rpc::jsonrpc::{
+        ErrorCode::{InternalError, InvalidParams, ParseError},
+        JsonError, JsonResponse, JsonResult,
     },
-    system::{Publisher, StoppableTask, StoppableTaskPtr},
-    util::{encoding::base64, time::fmt_duration},
-    Error, Result,
+    util::encoding::base64,
+    Result,
 };
 use darkfi_serial::deserialize_async;
 
-use crate::{error::handle_database_error, rpc::DarkfidRpcClient, Explorerd};
+use crate::{rpc::DarkfidRpcClient, Explorerd};
 
 impl DarkfidRpcClient {
     /// Retrieves a block from at a given height returning the corresponding [`BlockInfo`].
-    async fn get_block_by_height(&self, height: u32) -> Result<BlockInfo> {
+    pub async fn get_block_by_height(&self, height: u32) -> Result<BlockInfo> {
         let params = self
             .request(
                 "blockchain.get_block",
@@ -55,7 +48,7 @@ impl DarkfidRpcClient {
     }
 
     /// Retrieves the last confirmed block returning the block height and its header hash.
-    async fn get_last_confirmed_block(&self) -> Result<(u32, String)> {
+    pub async fn get_last_confirmed_block(&self) -> Result<(u32, String)> {
         let rep =
             self.request("blockchain.last_confirmed_block", &JsonValue::Array(vec![])).await?;
         let params = rep.get::<Vec<JsonValue>>().unwrap();
@@ -67,207 +60,6 @@ impl DarkfidRpcClient {
 }
 
 impl Explorerd {
-    /// 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::process_sync_blocks_reorg`] function, whose
-    /// documentation provides more details on the reorg process during block syncing.
-    pub async fn sync_blocks(&self, reset: bool) -> Result<()> {
-        // Grab last synced block height from the explorer's database.
-        let last_synced_block = self.service.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.service.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.process_sync_blocks_reorg(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.service.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.service.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 process_sync_blocks_reorg(
-        &self,
-        last_synced_height: u32,
-        last_darkfid_height: u32,
-    ) -> Result<u32> {
-        // 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.service.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.service.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().to_string()
-                                );
-                            }
-                        }
-                    }
-                    // 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.service.reset_explorer_state(0)?;
-        }
-
-        // Return the last height we reorged to
-        Ok(cur_height)
-    }
-
     // RPCAPI:
     // Queries the database to retrieve last N blocks.
     // Returns an array of readable blocks upon success.
@@ -402,172 +194,3 @@ impl Explorerd {
         }
     }
 }
-
-/// 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_blocks(
-    explorer: Arc<Explorerd>,
-    endpoint: Url,
-    ex: Arc<smol::Executor<'static>>,
-) -> 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::<Vec<JsonValue>>().unwrap();
-                        if params.is_empty() {
-                            return Err(Error::UnexpectedJsonRpc(
-                                "Notification parameters are empty".to_string(),
-                            ))
-                        }
-
-                        for param in params {
-                            let param = param.get::<String>().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().to_string());
-                            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))
-}

bin/explorer/explorerd/src/service/contracts.rs

@@ -316,7 +316,7 @@ pub fn untar_source(tar_bytes: &[u8]) -> Result<Vec<ContractSourceFile>> {
 /// the `ExplorerService` when handling contract-related operations.
 #[cfg(test)]
 mod tests {
-    use std::{fs::File, io::Read, path::Path};
+    use std::{fs::File, io::Read, path::Path, sync::Arc};
 
     use tar::Archive;
     use tempdir::TempDir;
@@ -325,7 +325,7 @@ mod tests {
     use darkfi_sdk::crypto::MONEY_CONTRACT_ID;
 
     use super::*;
-    use crate::test_utils::init_logger;
+    use crate::{rpc::DarkfidRpcClient, test_utils::init_logger};
 
     /// Tests the adding of [`ContractMetaData`] to the store by adding
     /// metadata, and verifying the inserted data matches the expected results.
@@ -469,7 +469,10 @@ mod tests {
         let db_path = temp_dir.path().join("sled_db");
 
         // Initialize the explorer service
-        ExplorerService::new(db_path.to_string_lossy().into_owned())
+        ExplorerService::new(
+            db_path.to_string_lossy().into_owned(),
+            Arc::new(DarkfidRpcClient::new()),
+        )
     }
 
     /// This Auxiliary function verifies that the loaded native contract source paths match the expected results

bin/explorer/explorerd/src/service/mod.rs

@@ -16,11 +16,13 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+use std::sync::Arc;
+
 use log::debug;
 
 use darkfi::Result;
 
-use crate::store::ExplorerDb;
+use crate::{rpc::DarkfidRpcClient, store::ExplorerDb};
 
 /// Handles core block-related functionality
 pub mod blocks;
@@ -34,6 +36,9 @@ pub mod statistics;
 /// Manages transaction data processing
 pub mod transactions;
 
+/// Manages synchronization with darkfid
+pub mod sync;
+
 /// Represents the service layer for the Explorer application, bridging the RPC layer and the database.
 /// It encapsulates explorer business logic and provides a unified interface for core functionalities,
 /// providing a clear separation of concerns between RPC handling and data management layers.
@@ -48,15 +53,17 @@ pub mod transactions;
 pub struct ExplorerService {
     /// Explorer database instance
     pub db: ExplorerDb,
+    /// JSON-RPC client used to execute requests to Darkfi blockchain nodes
+    pub darkfid_client: Arc<DarkfidRpcClient>,
 }
 
 impl ExplorerService {
     /// Creates a new `ExplorerService` instance.
-    pub fn new(db_path: String) -> Result<Self> {
+    pub fn new(db_path: String, darkfid_client: Arc<DarkfidRpcClient>) -> Result<Self> {
         // Initialize explorer database
         let db = ExplorerDb::new(db_path)?;
 
-        Ok(Self { db })
+        Ok(Self { db, darkfid_client })
     }
 
     /// Initializes the explorer service by deploying native contracts and loading native contract

bin/explorer/explorerd/src/service/sync.rs

@@ -0,0 +1,427 @@
+/* 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 <https://www.gnu.org/licenses/>.
+ */
+
+//! # 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<u32> {
+        // 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().to_string()
+                                );
+                            }
+                        }
+                    }
+                    // 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<Explorerd>,
+    endpoint: Url,
+    ex: Arc<smol::Executor<'static>>,
+) -> 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::<Vec<JsonValue>>().unwrap();
+                        if params.is_empty() {
+                            return Err(Error::UnexpectedJsonRpc(
+                                "Notification parameters are empty".to_string(),
+                            ))
+                        }
+
+                        for param in params {
+                            let param = param.get::<String>().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().to_string());
+                            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))
+}