16137717588

Committed 08 Jul 2025 08:12AM UTC coverage: 68.653%. First build

Build # 16137717588

Build Type

Pull #406

github

Committed by

web-flow

Commit Message

Merge 966560951 into 0d4c348fe

Pull Request Pull Request #406: feat(rust/signed-doc): Cleanup `DocType`, `DocumentRefs` related test cases

Run Details

185 of 189 new or added lines in 5 files covered. (97.88%)

12733 of 18547 relevant lines covered (68.65%)

2244.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/rust/cardano-chain-follower/src/chain_sync_config.rs

//! Cardano chain sync configuration.
//!
//! Independent of ANY followers, we allow a maximum of 3 Chains being updated, one for
//! each network. Chain Followers use the data supplied by the Chain-Sync.
//! This module configures the chain sync processes.

use std::sync::LazyLock;

use cardano_blockchain_types::Network;
use dashmap::DashMap;
use strum::IntoEnumIterator;
use tokio::{sync::Mutex, task::JoinHandle};
use tracing::{debug, error};

use crate::{
    chain_sync::chain_sync,
    error::{Error, Result},
    mithril_snapshot_config::MithrilSnapshotConfig,
    stats,
};

/// Default Follower block buffer size.
const DEFAULT_CHAIN_UPDATE_BUFFER_SIZE: usize = 32;

/// How many window (in slot) back from TIP is considered Immutable in the
/// absence of a mithril snapshot.
const DEFAULT_IMMUTABLE_SLOT_WINDOW: u64 = 12 * 60 * 60;

/// Type we use to manage the Sync Task handle map.
type SyncMap = DashMap<Network, Mutex<Option<JoinHandle<()>>>>;

/// Handle to the mithril sync thread. One for each Network ONLY.
static SYNC_JOIN_HANDLE_MAP: LazyLock<SyncMap> = LazyLock::new(|| {
    let map = DashMap::new();
    for network in Network::iter() {
        map.insert(network, Mutex::new(None));
    }
    map
});

/// A Follower Connection to the Cardano Network.
#[derive(Clone, Debug)]
pub struct ChainSyncConfig {
    /// Chain Network
    pub chain: Network,
    /// Relay Node Address
    pub(crate) relay_address: String,
    /// Block buffer size option.
    chain_update_buffer_size: usize,
    /// If we don't have immutable data, how far back from TIP is the data considered
    /// Immutable (in slots).
    immutable_slot_window: u64,
    /// Configuration of Mithril Snapshots.
    pub mithril_cfg: MithrilSnapshotConfig,
}

impl ChainSyncConfig {
    /// Sets the defaults for a given cardano network.
    /// Each network has a different set of defaults, so no single "default" can apply.
    /// This function is preferred to the `default()` standard function.
    #[must_use]
    pub fn default_for(chain: Network) -> Self {
        Self {
            chain,
            relay_address: chain.default_relay(),
            chain_update_buffer_size: DEFAULT_CHAIN_UPDATE_BUFFER_SIZE,
            immutable_slot_window: DEFAULT_IMMUTABLE_SLOT_WINDOW,
            mithril_cfg: MithrilSnapshotConfig::default_for(chain),
        }
    }

    /// Sets the relay address to use for Chain Sync.
    ///
    /// # Arguments
    ///
    /// * `relay`: Address to use for the blockchain relay node.
    #[must_use]
    pub fn relay(mut self, address: String) -> Self {
        self.relay_address = address;
        self
    }

    /// Sets the size of the chain updates buffer used by the Follower.
    ///
    /// # Arguments
    ///
    /// * `chain_update_buffer_size`: Size of the chain updates buffer.
    #[must_use]
    pub fn chain_update_buffer_size(mut self, block_buffer_size: usize) -> Self {
        self.chain_update_buffer_size = block_buffer_size;
        self
    }

    /// Sets the size of the Immutable window used when Mithril is not available.
    ///
    /// # Arguments
    ///
    /// * `window`: Size of the Immutable window.
    #[must_use]
    pub fn immutable_slot_window(mut self, window: u64) -> Self {
        self.immutable_slot_window = window;
        self
    }

    /// Sets the Mithril snapshot Config the `ChainSync` will use.
    ///
    /// # Arguments
    ///
    /// * `cfg`: Mithril snapshot configuration.
    #[must_use]
    pub fn mithril_cfg(mut self, cfg: MithrilSnapshotConfig) -> Self {
        self.mithril_cfg = cfg;
        self
    }

    /// Runs Chain Synchronization.
    ///
    /// Must be done BEFORE the chain can be followed.
    ///
    /// # Returns
    ///
    /// `Result<()>`: On success.
    ///
    /// # Errors
    ///
    /// `Error`: On error.
    pub async fn run(self) -> Result<()> {
        debug!(
            chain = self.chain.to_string(),
            "Chain Synchronization Starting"
        );

        stats::sync_started(self.chain);

        // Start the Chain Sync - IFF its not already running.
        let lock_entry = match SYNC_JOIN_HANDLE_MAP.get(&self.chain) {
            None => {
                error!("Join Map improperly initialized: Missing {}!!", self.chain);
                return Err(Error::Internal); // Should not get here.
            },
            Some(entry) => entry,
        };
        let mut locked_handle = lock_entry.value().lock().await;

        if (*locked_handle).is_some() {
            debug!("Chain Sync Already Running for {}", self.chain);
            return Err(Error::ChainSyncAlreadyRunning(self.chain));
        }

        // Start the Mithril Snapshot Follower
        let rx = self.mithril_cfg.run().await?;

        let config = self.clone();
        // Start Chain Sync
        *locked_handle = Some(tokio::spawn(async move {
            stats::start_thread(config.chain, stats::thread::name::CHAIN_SYNC, true);
            chain_sync(config.clone(), rx).await;
            stats::stop_thread(config.chain, stats::thread::name::CHAIN_SYNC);
        }));

        // sync_map.insert(chain, handle);
        debug!("Chain Sync for {} : Started", self.chain);

        Ok(())
    }
}

1	//! Cardano chain sync configuration.
2	//!
3	//! Independent of ANY followers, we allow a maximum of 3 Chains being updated, one for
4	//! each network. Chain Followers use the data supplied by the Chain-Sync.
5	//! This module configures the chain sync processes.
6
7	use std::sync::LazyLock;
8
9	use cardano_blockchain_types::Network;
10	use dashmap::DashMap;
11	use strum::IntoEnumIterator;
12	use tokio::{sync::Mutex, task::JoinHandle};
13	use tracing::{debug, error};
14
15	use crate::{
16	chain_sync::chain_sync,
17	error::{Error, Result},
18	mithril_snapshot_config::MithrilSnapshotConfig,
19	stats,
20	};
21
22	/// Default Follower block buffer size.
23	const DEFAULT_CHAIN_UPDATE_BUFFER_SIZE: usize = 32;
24
25	/// How many window (in slot) back from TIP is considered Immutable in the
26	/// absence of a mithril snapshot.
27	const DEFAULT_IMMUTABLE_SLOT_WINDOW: u64 = 12 * 60 * 60;
28
29	/// Type we use to manage the Sync Task handle map.
30	type SyncMap = DashMap<Network, Mutex<Option<JoinHandle<()>>>>;
31
32	/// Handle to the mithril sync thread. One for each Network ONLY.
33	static SYNC_JOIN_HANDLE_MAP: LazyLock<SyncMap> = LazyLock::new(\|\| {	×
34	let map = DashMap::new();	×
35	for network in Network::iter() {	×
36	map.insert(network, Mutex::new(None));	×
37	}	×
38	map	×
39	});	×
40
41	/// A Follower Connection to the Cardano Network.
42	#[derive(Clone, Debug)]
43	pub struct ChainSyncConfig {
44	/// Chain Network
45	pub chain: Network,
46	/// Relay Node Address
47	pub(crate) relay_address: String,
48	/// Block buffer size option.
49	chain_update_buffer_size: usize,
50	/// If we don't have immutable data, how far back from TIP is the data considered
51	/// Immutable (in slots).
52	immutable_slot_window: u64,
53	/// Configuration of Mithril Snapshots.
54	pub mithril_cfg: MithrilSnapshotConfig,
55	}
56
57	impl ChainSyncConfig {
58	/// Sets the defaults for a given cardano network.
59	/// Each network has a different set of defaults, so no single "default" can apply.
60	/// This function is preferred to the `default()` standard function.
61	#[must_use]
62	pub fn default_for(chain: Network) -> Self {	×
63	Self {	×
64	chain,	×
65	relay_address: chain.default_relay(),	×
66	chain_update_buffer_size: DEFAULT_CHAIN_UPDATE_BUFFER_SIZE,	×
67	immutable_slot_window: DEFAULT_IMMUTABLE_SLOT_WINDOW,	×
68	mithril_cfg: MithrilSnapshotConfig::default_for(chain),	×
69	}	×
70	}	×
71
72	/// Sets the relay address to use for Chain Sync.
73	///
74	/// # Arguments
75	///
76	/// * `relay`: Address to use for the blockchain relay node.
77	#[must_use]
78	pub fn relay(mut self, address: String) -> Self {	×
79	self.relay_address = address;	×
80	self	×
81	}	×
82
83	/// Sets the size of the chain updates buffer used by the Follower.
84	///
85	/// # Arguments
86	///
87	/// * `chain_update_buffer_size`: Size of the chain updates buffer.
88	#[must_use]
89	pub fn chain_update_buffer_size(mut self, block_buffer_size: usize) -> Self {	×
90	self.chain_update_buffer_size = block_buffer_size;	×
91	self	×
92	}	×
93
94	/// Sets the size of the Immutable window used when Mithril is not available.
95	///
96	/// # Arguments
97	///
98	/// * `window`: Size of the Immutable window.
99	#[must_use]
100	pub fn immutable_slot_window(mut self, window: u64) -> Self {	×
101	self.immutable_slot_window = window;	×
102	self	×
103	}	×
104
105	/// Sets the Mithril snapshot Config the `ChainSync` will use.
106	///
107	/// # Arguments
108	///
109	/// * `cfg`: Mithril snapshot configuration.
110	#[must_use]
111	pub fn mithril_cfg(mut self, cfg: MithrilSnapshotConfig) -> Self {	×
112	self.mithril_cfg = cfg;	×
113	self	×
114	}	×
115
116	/// Runs Chain Synchronization.
117	///
118	/// Must be done BEFORE the chain can be followed.
119	///
120	/// # Returns
121	///
122	/// `Result<()>`: On success.
123	///
124	/// # Errors
125	///
126	/// `Error`: On error.
127	pub async fn run(self) -> Result<()> {	×
128	debug!(	×
129	chain = self.chain.to_string(),	×
130	"Chain Synchronization Starting"	×
131	);
132
133	stats::sync_started(self.chain);	×
134
135	// Start the Chain Sync - IFF its not already running.
136	let lock_entry = match SYNC_JOIN_HANDLE_MAP.get(&self.chain) {	×
137	None => {
138	error!("Join Map improperly initialized: Missing {}!!", self.chain);	×
139	return Err(Error::Internal); // Should not get here.	×
140	},
141	Some(entry) => entry,	×
142	};
143	let mut locked_handle = lock_entry.value().lock().await;	×
144
145	if (*locked_handle).is_some() {	×
146	debug!("Chain Sync Already Running for {}", self.chain);	×
147	return Err(Error::ChainSyncAlreadyRunning(self.chain));	×
148	}	×
149
150	// Start the Mithril Snapshot Follower
151	let rx = self.mithril_cfg.run().await?;	×
152
153	let config = self.clone();	×
154	// Start Chain Sync	×
155	*locked_handle = Some(tokio::spawn(async move {	×
156	stats::start_thread(config.chain, stats::thread::name::CHAIN_SYNC, true);	×
157	chain_sync(config.clone(), rx).await;	×
158	stats::stop_thread(config.chain, stats::thread::name::CHAIN_SYNC);	×
159	}));	×
160		×
161	// sync_map.insert(chain, handle);	×
162	debug!("Chain Sync for {} : Started", self.chain);	×
163
164	Ok(())	×
165	}	×
166	}

input-output-hk / catalyst-libs / 16137717588

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous