all chains use the same config loader

Author: pragmaxim

chain/Cargo.toml

@@ -22,7 +22,9 @@ async-trait = "0.1.81"
 redb = "3.0.2"
 serde = { version = "1.0.219", features = ["derive"] }
 tokio = { version = "1.45.1", features = ["full", "tracing"] }
+tokio-util = "0.7.16"
 tower-http = { version = "0.6.6", features = ["cors"] }
 tokio-stream = { version = "0.1.17", features = ["sync"] }
 thiserror = "2.0.14"
-sysinfo = "0.37.0"
+sysinfo = "0.37.0"
+dotenv = "0.15.0"

chain/src/api.rs

@@ -8,6 +8,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 use redb::{Durability, StorageError};
 use tokio::task::JoinError;
+use tokio_util::sync::CancellationToken;
 use redbit::storage::table_writer_api::TaskResult;
 
 #[derive(Debug, thiserror::Error)]
@@ -84,5 +85,10 @@ pub trait BlockProvider<FB: SizeLike, TB: BlockLike>: Send + Sync {
     fn block_processor(&self) -> Arc<dyn Fn(&FB) -> Result<TB, ChainError> + Send + Sync>;
     fn get_processed_block(&self, hash: <TB::Header as BlockHeaderLike>::Hash) -> Result<Option<TB>, ChainError>;
     async fn get_chain_tip(&self) -> Result<TB::Header, ChainError>;
-    fn stream(&self, remote_chain_tip_header: TB::Header, last_persisted_header: Option<TB::Header>, durability: Durability) -> Pin<Box<dyn Stream<Item = FB> + Send + 'static>>;
+    fn stream(
+        &self,
+        remote_chain_tip_header: TB::Header,
+        last_persisted_header: Option<TB::Header>,
+        durability: Durability
+    ) -> (Pin<Box<dyn Stream<Item = FB> + Send + 'static>>, CancellationToken);
 }

chain/src/chain_config.rs

@@ -0,0 +1,36 @@
+use config::{Config, ConfigError, Environment, File};
+use dotenv::dotenv;
+use serde::de::DeserializeOwned;
+use std::fmt::Debug;
+use std::sync::Once;
+use redbit::info;
+
+static DOTENV_ONCE: Once = Once::new();
+
+fn ensure_dotenv_loaded() {
+    DOTENV_ONCE.call_once(|| {
+        match dotenv() {
+            Ok(_) => info!("Config loaded including .env file."),
+            Err(_) => info!("Config loaded without .env file."),
+        }
+    });
+}
+
+pub fn load_config<T>(path: &str, prefix: &str) -> Result<T, ConfigError>
+where
+    T: DeserializeOwned + Debug,
+{
+    ensure_dotenv_loaded();
+
+    let builder = Config::builder()
+        .add_source(File::with_name(path).required(true))
+        .add_source(
+            Environment::with_prefix(prefix)
+                .try_parsing(true)
+                .separator("__"),
+        );
+
+    let cfg = builder.build()?.try_deserialize::<T>()?;
+    info!("{:#?}", cfg);
+    Ok(cfg)
+}

chain/src/launcher.rs

@@ -1,7 +1,7 @@
 use crate::api::{BlockChainLike, BlockLike, BlockProvider, SizeLike};
 use crate::scheduler::Scheduler;
 use crate::settings::{AppConfig, DbCacheSize, HttpSettings, IndexerSettings};
-use crate::{combine, ChainError};
+use crate::{chain_config, combine, ChainError};
 use futures::future::ready;
 use redbit::storage::init::{Storage, StorageOwner};
 use redbit::{error, info, serve, AppError, OpenApiRouter, RequestState, WriteTxContext};
@@ -124,7 +124,7 @@ pub async fn launch<FB: SizeLike + 'static, TB: BlockLike + 'static, CTX: WriteT
 where
     F: FnOnce(Arc<Storage>) -> Arc<dyn BlockChainLike<TB, CTX>>,
 {
-    let config = AppConfig::new("config/settings").expect("Failed to load app config");
+    let config: AppConfig = chain_config::load_config("config/settings", "REDBIT").expect("Failed to load Redbit settings");
     maybe_console_init();
     let (created, storage_owner, storage_view) = build_storage(&config).await?;
     let chain: Arc<dyn BlockChainLike<TB, CTX>> = build_chain(Arc::clone(&storage_view));

chain/src/lib.rs

@@ -8,5 +8,6 @@ pub mod launcher;
 pub mod task;
 pub mod batcher;
 pub mod stats;
+pub mod chain_config;
 
 pub use api::{BlockHeaderLike, SizeLike, BlockLike, BlockChainLike, ChainError};

chain/src/settings.rs

@@ -1,11 +1,9 @@
-use config::{Config, ConfigError, Environment, File};
 use serde::Deserialize;
 use std::fmt::Debug;
 use std::net::SocketAddr;
 use std::str::FromStr;
 use std::time::Duration;
 use sysinfo::System;
-use redbit::info;
 
 #[derive(Clone, Copy)]
 pub struct Parallelism(pub usize);
@@ -145,15 +143,3 @@ pub struct HttpSettings {
     pub enable: bool,
     pub bind_address: SocketAddr,
 }
-
-impl AppConfig {
-    pub fn new(path: &str) -> Result<Self, ConfigError> {
-        let builder =
-            Config::builder()
-                .add_source(File::with_name(path).required(true))
-                .add_source(Environment::with_prefix("REDBIT").try_parsing(true).separator("__"));
-        let config = builder.build()?.try_deserialize::<AppConfig>()?;
-        info!("{:#?}", config);
-        Ok(config)
-    }
-}

chain/src/syncer.rs

@@ -66,14 +66,15 @@ impl<FB: SizeLike + 'static, TB: BlockLike + 'static, CTX: WriteTxContext + 'sta
             let block_provider = Arc::clone(&self.block_provider);
             let shutdown = shutdown.clone();
             task::spawn_named("fetch", async move {
-                let mut s = block_provider.stream(node_chain_tip_header.clone(), last_persisted_header, default_durability);
+                let (mut s, cancel_token) = block_provider.stream(node_chain_tip_header.clone(), last_persisted_header, default_durability);
                 let mut buf: Vec<FB> = Vec::new();
                 let mut buf_bytes: usize = 0;
                 loop {
                     tokio::select! {
                         biased;
                         _ = combine::await_shutdown(shutdown.clone()) => {
                             info!("fetch: shutdown");
+                            cancel_token.cancel();
                             break;
                         }
                         item = s.next() => {
@@ -152,7 +153,7 @@ impl<FB: SizeLike + 'static, TB: BlockLike + 'static, CTX: WriteTxContext + 'sta
                                                         }
                                                     }
                                                     Err(e) => {
-                                                        error!("processing failure: {e}");
+                                                        warn!("Processing warning: {e}");
                                                     }
                                                 }
                                             }

chains/btc/Cargo.toml

@@ -36,6 +36,7 @@ redbit = { path = "../../redbit" }
 chain = { path = "../../chain" }
 tokio = { version = "1.45.1", features = ["full", "tracing"] }
 tokio-stream = { version = "0.1.17", features = ["sync"] }
+tokio-util = "0.7.16"
 config = "0.15.11"
 bitcoin = "0.32.0"
 bitcoin_hashes = "0.16.0"
@@ -46,7 +47,6 @@ bech32 = "0.11.0"
 bs58 = { version = "0.5", features = ["check"] }
 axum = {  version = "0.8.4", features = ["default", "macros"] }
 thiserror = "2.0.12"
-dotenv = "0.15.0"
 chrono = "0.4.38"
 anyhow = "1.0.80"
 bincode = "2.0.1"

chains/btc/src/block_provider.rs

@@ -1,9 +1,9 @@
 use crate::codec::{TAG_NON_ADDR, TAG_OP_RETURN, TAG_SEGWIT};
-use crate::config::BitcoinConfig;
 use crate::model_v1::{Address, Block, BlockHash, BlockPointer, Header, Height, InputRef, MerkleRoot, ScriptHash, Timestamp, Transaction, TransactionPointer, TxHash, Utxo, Weight};
 use crate::rest_client::{BtcCBOR, BtcClient};
-use crate::ExplorerError;
+use crate::{BitcoinConfig, ExplorerError};
 use async_trait::async_trait;
+use tokio_util::sync::CancellationToken;
 use chain::api::{BlockProvider, ChainError};
 use chain::monitor::BoxWeight;
 use chain::settings::Parallelism;
@@ -13,6 +13,7 @@ use redbit::info;
 use redbit::*;
 use serde_json;
 use std::{fs, pin::Pin, sync::Arc};
+use chain::chain_config;
 use redbit::redb::Durability;
 
 pub struct BtcBlockProvider {
@@ -22,7 +23,7 @@ pub struct BtcBlockProvider {
 
 impl BtcBlockProvider {
     pub fn new() -> Result<Arc<Self>, ExplorerError> {
-        let config = BitcoinConfig::new("config/btc").expect("Failed to load Bitcoin configuration");
+        let config: BitcoinConfig = chain_config::load_config("config/btc", "BITCOIN").expect("Failed to load Bitcoin configuration");
         let client = Arc::new(BtcClient::new(&config)?);
         let fetching_par: Parallelism = config.fetching_parallelism.clone();
         Ok(Arc::new(BtcBlockProvider { client, fetching_par }))
@@ -193,7 +194,7 @@ impl BlockProvider<BtcCBOR, Block> for BtcBlockProvider {
         remote_chain_tip_header: Header,
         last_persisted_header: Option<Header>,
         durability: Durability
-    ) -> Pin<Box<dyn Stream<Item = BtcCBOR> + Send + 'static>> {
+    ) -> (Pin<Box<dyn Stream<Item = BtcCBOR> + Send + 'static>>, CancellationToken) {
         let height_to_index_from = last_persisted_header.map_or(1, |h| h.height.0 + 1);
         let heights = height_to_index_from..=remote_chain_tip_header.height.0;
         let client = Arc::clone(&self.client);
@@ -204,11 +205,12 @@ impl BlockProvider<BtcCBOR, Block> for BtcBlockProvider {
                     client.get_block_by_height(Height(height)).await.expect("Failed to fetch block by height")
                 }
             });
-        match durability {
+        let stream = match durability {
             Durability::None => s.buffer_unordered(self.fetching_par.0).boxed(),
             Durability::Immediate => s.buffered(self.fetching_par.0).boxed(),
             _ => unreachable!("Only None and Immediate durability modes are supported"),
-        }
+        };
+        (stream, CancellationToken::new())
     }
 }