16618685382

pending completion

Build # 16618685382

Build Type

Pull #419

github

Committed by

web-flow

Commit Message

Merge 3cdf83512 into 654b17f07

Pull Request Pull Request #419: feat(docs): Form Element Documentation

Run Details

10738 of 16725 relevant lines covered (64.2%)

2361.28 hits per line

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

0.0

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

//! Internal Mithril snapshot iterator functions.

use std::{
    fmt::Debug,
    path::{Path, PathBuf},
    sync::{Arc, Mutex},
};

use cardano_blockchain_types::{Fork, MultiEraBlock, Network, Point};
use logcall::logcall;
use tokio::task;
use tracing::{debug, error};
use tracing_log::log;

use crate::{
    error::{Error, Result},
    mithril_query::{make_mithril_iterator, ImmutableBlockIterator},
};

/// Search backwards by 60 slots (seconds) looking for a previous block.
/// This search window is doubled until the search succeeds.
const BACKWARD_SEARCH_SLOT_INTERVAL: u64 = 60;

/// Synchronous Inner Iterator state
struct MithrilSnapshotIteratorInner {
    /// The chain being iterated
    chain: Network,
    /// Where we really want to start iterating from
    start: Point,
    /// Previous iteration point.
    previous: Point,
    /// Inner iterator.
    inner: ImmutableBlockIterator,
}

impl Debug for MithrilSnapshotIteratorInner {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "MithrilSnapshotIteratorInner {{ chain: {:?}, start: {:?}, previous: {:?} }}",
            self.chain, self.start, self.previous
        )
    }
}

/// Wraps the iterator type returned by Pallas.
#[derive(Debug)]
pub(crate) struct MithrilSnapshotIterator {
    /// Mithril snapshot directory path
    path: PathBuf,
    /// Inner Mutable Synchronous Iterator State
    inner: Arc<Mutex<MithrilSnapshotIteratorInner>>,
}

/// Create a probe point used in iterations to find the start when its not exactly known.
pub(crate) fn probe_point(point: &Point, distance: u64) -> Point {
    /// Step back point to indicate origin.
    const STEP_BACK_ORIGIN: u64 = 0;
    // Now that we have the tip, step back about 4 block intervals from tip, and do a fuzzy
    // iteration to find the exact two blocks at the end of the immutable chain.
    // It is ok because slot implement saturating subtraction.
    #[allow(clippy::arithmetic_side_effects)]
    let step_back_search = point.slot_or_default() - distance.into();

    // We stepped back to the origin, so just return Origin
    if step_back_search == STEP_BACK_ORIGIN.into() {
        return Point::ORIGIN;
    }

    // Create a fuzzy search probe by making the hash zero length.
    Point::fuzzy(step_back_search)
}

impl MithrilSnapshotIterator {
    /// Returns `true` if the `MithrilSnapshotIterator` could read data without any issues
    /// (underlying mithril snapshot directory exists)
    pub(crate) fn is_valid(&self) -> bool {
        self.path.exists()
    }

    /// Try and probe to establish the iterator from the desired point.
    async fn try_fuzzy_iterator(
        chain: Network, path: &Path, from: &Point, search_interval: u64,
    ) -> Option<MithrilSnapshotIterator> {
        let point = probe_point(from, search_interval);
        let Ok(mut iterator) = make_mithril_iterator(path, &point, chain).await else {
            return None;
        };

        let mut previous = None;
        let mut this = None;

        loop {
            let next = iterator.next();

            match next {
                Some(Ok(raw_block)) => {
                    let Ok(block) = pallas::ledger::traverse::MultiEraBlock::decode(&raw_block)
                    else {
                        return None;
                    };

                    let point = Point::new(block.slot().into(), block.hash().into());
                    previous = this;
                    this = Some(point.clone());

                    debug!("Searching for {from}. {this:?} > {previous:?}");

                    // Stop as soon as we find the point, or exceed it.
                    if point >= *from {
                        break;
                    }
                },
                Some(Err(err)) => {
                    error!("Error while iterating fuzzy mithril data: {}", err);
                    return None;
                },
                None => break,
            };
        }

        debug!("Best Found for {from}. {this:?} > {previous:?}");

        // Fail if we didn't find the destination block, or its immediate predecessor.
        previous.as_ref()?;
        let this = this?;

        // Remake the iterator, based on the new known point.
        let Ok(iterator) = make_mithril_iterator(path, &this, chain).await else {
            return None;
        };

        Some(MithrilSnapshotIterator {
            path: path.to_path_buf(),
            inner: Arc::new(Mutex::new(MithrilSnapshotIteratorInner {
                chain,
                start: this,
                previous: previous?,
                inner: iterator,
            })),
        })
    }

    /// Do a fuzzy search to establish the iterator.
    /// We use this when we don't know the previous point, and need to find it.
    #[allow(clippy::indexing_slicing)]
    #[logcall("debug")]
    async fn fuzzy_iterator(chain: Network, path: &Path, from: &Point) -> MithrilSnapshotIterator {
        let mut backwards_search = BACKWARD_SEARCH_SLOT_INTERVAL;
        loop {
            if let Some(iterator) =
                Self::try_fuzzy_iterator(chain, path, from, backwards_search).await
            {
                return iterator;
            }

            backwards_search = backwards_search.saturating_mul(2);
        }
    }

    /// Create a mithril iterator, optionally where we know the previous point.
    ///
    /// # Arguments
    ///
    /// `chain`: The blockchain network to iterate.
    /// `from`: The point to start iterating from.  If the `Point` does not contain a
    /// hash, the iteration start is fuzzy. `previous`: The previous point we are
    /// iterating, if known.    If the previous is NOT known, then the first block
    /// yielded by the iterator is discarded and becomes the known previous.
    #[allow(clippy::indexing_slicing)]
    #[logcall(ok = "debug", err = "error")]
    pub(crate) async fn new(
        chain: Network, path: &Path, from: &Point, previous_point: Option<Point>,
    ) -> Result<Self> {
        if from.is_fuzzy() || (!from.is_origin() && previous_point.is_none()) {
            return Ok(Self::fuzzy_iterator(chain, path, from).await);
        }

        let previous = if from.is_origin() {
            Point::ORIGIN
        } else {
            let Some(previous) = previous_point else {
                return Err(Error::Internal);
            };
            previous
        };

        debug!("Actual Mithril Iterator Start: {}", from);

        let iterator = make_mithril_iterator(path, from, chain).await?;

        Ok(MithrilSnapshotIterator {
            path: path.to_path_buf(),
            inner: Arc::new(Mutex::new(MithrilSnapshotIteratorInner {
                chain,
                start: from.clone(),
                previous,
                inner: iterator,
            })),
        })
    }

    /// Get the next block, in a way that is Async friendly.
    /// Returns the next block, or None if there are no more blocks.
    pub(crate) async fn next(&self) -> Option<MultiEraBlock> {
        let inner = self.inner.clone();

        let res = task::spawn_blocking(move || {
            #[allow(clippy::expect_used)]
            let mut inner_iterator = inner
                .lock()
                .expect("Safe here because the lock can't be poisoned");
            inner_iterator.next()
        })
        .await;

        res.unwrap_or_default()
    }
}

impl Iterator for MithrilSnapshotIteratorInner {
    type Item = MultiEraBlock;

    fn next(&mut self) -> Option<Self::Item> {
        for maybe_block in self.inner.by_ref() {
            if let Ok(block) = maybe_block {
                if !self.previous.is_unknown() {
                    // We can safely fully decode this block.
                    match MultiEraBlock::new(self.chain, block, &self.previous, Fork::IMMUTABLE) {
                        Ok(block_data) => {
                            // Update the previous point
                            // debug!("Pre Previous update 1 : {:?}", self.previous);
                            self.previous = block_data.point();
                            // debug!("Post Previous update 1 : {:?}", self.previous);

                            // Make sure we got to the start, otherwise this could be a block
                            // artifact from a discover previous point
                            // search.
                            if block_data < self.start {
                                continue;
                            }

                            return Some(block_data);
                        },
                        Err(error) => {
                            error!(previous=%self.previous, error=%error, "Error decoding a block from the snapshot");
                            break;
                        },
                    }
                }

                // We cannot fully decode this block because we don't know its previous point,
                // So this MUST be the first block in iteration, so use it as the previous.
                if let Ok(raw_decoded_block) =
                    pallas::ledger::traverse::MultiEraBlock::decode(&block)
                {
                    // debug!("Pre Previous update 2 : {:?}", self.previous);
                    self.previous = Point::new(
                        raw_decoded_block.slot().into(),
                        raw_decoded_block.hash().into(),
                    );
                    // debug!("Post Previous update 2 : {:?}", self.previous);
                    continue;
                }
                error!("Error decoding block to use for previous from the snapshot.");
                break;
            }

            error!("Error while fetching a block from the snapshot");
            break;
        }

        None
    }
}

1	//! Internal Mithril snapshot iterator functions.
2
3	use std::{
4	fmt::Debug,
5	path::{Path, PathBuf},
6	sync::{Arc, Mutex},
7	};
8
9	use cardano_blockchain_types::{Fork, MultiEraBlock, Network, Point};
10	use logcall::logcall;
11	use tokio::task;
12	use tracing::{debug, error};
13	use tracing_log::log;
14
15	use crate::{
16	error::{Error, Result},
17	mithril_query::{make_mithril_iterator, ImmutableBlockIterator},
18	};
19
20	/// Search backwards by 60 slots (seconds) looking for a previous block.
21	/// This search window is doubled until the search succeeds.
22	const BACKWARD_SEARCH_SLOT_INTERVAL: u64 = 60;
23
24	/// Synchronous Inner Iterator state
25	struct MithrilSnapshotIteratorInner {
26	/// The chain being iterated
27	chain: Network,
28	/// Where we really want to start iterating from
29	start: Point,
30	/// Previous iteration point.
31	previous: Point,
32	/// Inner iterator.
33	inner: ImmutableBlockIterator,
34	}
35
36	impl Debug for MithrilSnapshotIteratorInner {
37	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {	×
38	write!(	×
39	f,	×
40	"MithrilSnapshotIteratorInner {{ chain: {:?}, start: {:?}, previous: {:?} }}",	×
41	self.chain, self.start, self.previous	×
42	)	×
43	}	×
44	}
45
46	/// Wraps the iterator type returned by Pallas.
47	#[derive(Debug)]
48	pub(crate) struct MithrilSnapshotIterator {
49	/// Mithril snapshot directory path
50	path: PathBuf,
51	/// Inner Mutable Synchronous Iterator State
52	inner: Arc<Mutex<MithrilSnapshotIteratorInner>>,
53	}
54
55	/// Create a probe point used in iterations to find the start when its not exactly known.
56	pub(crate) fn probe_point(point: &Point, distance: u64) -> Point {	×
57	/// Step back point to indicate origin.
58	const STEP_BACK_ORIGIN: u64 = 0;
59	// Now that we have the tip, step back about 4 block intervals from tip, and do a fuzzy
60	// iteration to find the exact two blocks at the end of the immutable chain.
61	// It is ok because slot implement saturating subtraction.
62	#[allow(clippy::arithmetic_side_effects)]
63	let step_back_search = point.slot_or_default() - distance.into();	×
64		×
65	// We stepped back to the origin, so just return Origin	×
66	if step_back_search == STEP_BACK_ORIGIN.into() {	×
67	return Point::ORIGIN;	×
68	}	×
69		×
70	// Create a fuzzy search probe by making the hash zero length.	×
71	Point::fuzzy(step_back_search)	×
72	}	×
73
74	impl MithrilSnapshotIterator {
75	/// Returns `true` if the `MithrilSnapshotIterator` could read data without any issues
76	/// (underlying mithril snapshot directory exists)
77	pub(crate) fn is_valid(&self) -> bool {	×
78	self.path.exists()	×
79	}	×
80
81	/// Try and probe to establish the iterator from the desired point.
82	async fn try_fuzzy_iterator(	×
83	chain: Network, path: &Path, from: &Point, search_interval: u64,	×
84	) -> Option<MithrilSnapshotIterator> {	×
85	let point = probe_point(from, search_interval);	×
86	let Ok(mut iterator) = make_mithril_iterator(path, &point, chain).await else {	×
87	return None;	×
88	};
89
90	let mut previous = None;	×
91	let mut this = None;	×
92
93	loop {
94	let next = iterator.next();	×
95
96	match next {	×
97	Some(Ok(raw_block)) => {	×
98	let Ok(block) = pallas::ledger::traverse::MultiEraBlock::decode(&raw_block)	×
99	else {
100	return None;	×
101	};
102
103	let point = Point::new(block.slot().into(), block.hash().into());	×
104	previous = this;	×
105	this = Some(point.clone());	×
106		×
107	debug!("Searching for {from}. {this:?} > {previous:?}");	×
108
109	// Stop as soon as we find the point, or exceed it.
110	if point >= *from {	×
111	break;	×
112	}	×
113	},
114	Some(Err(err)) => {	×
115	error!("Error while iterating fuzzy mithril data: {}", err);	×
116	return None;	×
117	},
118	None => break,	×
119	};
120	}
121
122	debug!("Best Found for {from}. {this:?} > {previous:?}");	×
123
124	// Fail if we didn't find the destination block, or its immediate predecessor.
125	previous.as_ref()?;	×
126	let this = this?;	×
127
128	// Remake the iterator, based on the new known point.
129	let Ok(iterator) = make_mithril_iterator(path, &this, chain).await else {	×
130	return None;	×
131	};
132
133	Some(MithrilSnapshotIterator {
134	path: path.to_path_buf(),	×
135	inner: Arc::new(Mutex::new(MithrilSnapshotIteratorInner {	×
136	chain,	×
137	start: this,	×
138	previous: previous?,	×
139	inner: iterator,	×
140	})),
141	})
142	}	×
143
144	/// Do a fuzzy search to establish the iterator.
145	/// We use this when we don't know the previous point, and need to find it.
146	#[allow(clippy::indexing_slicing)]
147	#[logcall("debug")]	×
148	async fn fuzzy_iterator(chain: Network, path: &Path, from: &Point) -> MithrilSnapshotIterator {	×
149	let mut backwards_search = BACKWARD_SEARCH_SLOT_INTERVAL;	×
150	loop {
151	if let Some(iterator) =	×
152	Self::try_fuzzy_iterator(chain, path, from, backwards_search).await	×
153	{
154	return iterator;	×
155	}	×
156		×
157	backwards_search = backwards_search.saturating_mul(2);	×
158	}
159	}	×
160
161	/// Create a mithril iterator, optionally where we know the previous point.
162	///
163	/// # Arguments
164	///
165	/// `chain`: The blockchain network to iterate.
166	/// `from`: The point to start iterating from. If the `Point` does not contain a
167	/// hash, the iteration start is fuzzy. `previous`: The previous point we are
168	/// iterating, if known. If the previous is NOT known, then the first block
169	/// yielded by the iterator is discarded and becomes the known previous.
170	#[allow(clippy::indexing_slicing)]
171	#[logcall(ok = "debug", err = "error")]	×
172	pub(crate) async fn new(
173	chain: Network, path: &Path, from: &Point, previous_point: Option<Point>,
174	) -> Result<Self> {	×
175	if from.is_fuzzy() \|\| (!from.is_origin() && previous_point.is_none()) {	×
176	return Ok(Self::fuzzy_iterator(chain, path, from).await);	×
177	}	×
178
179	let previous = if from.is_origin() {	×
180	Point::ORIGIN	×
181	} else {
182	let Some(previous) = previous_point else {	×
183	return Err(Error::Internal);	×
184	};
185	previous	×
186	};
187
188	debug!("Actual Mithril Iterator Start: {}", from);	×
189
190	let iterator = make_mithril_iterator(path, from, chain).await?;	×
191
192	Ok(MithrilSnapshotIterator {	×
193	path: path.to_path_buf(),	×
194	inner: Arc::new(Mutex::new(MithrilSnapshotIteratorInner {	×
195	chain,	×
196	start: from.clone(),	×
197	previous,	×
198	inner: iterator,	×
199	})),	×
200	})	×
201	}	×
202
203	/// Get the next block, in a way that is Async friendly.
204	/// Returns the next block, or None if there are no more blocks.
205	pub(crate) async fn next(&self) -> Option<MultiEraBlock> {	×
206	let inner = self.inner.clone();	×
207
208	let res = task::spawn_blocking(move \|\| {	×
209	#[allow(clippy::expect_used)]	×
210	let mut inner_iterator = inner	×
211	.lock()	×
212	.expect("Safe here because the lock can't be poisoned");	×
213	inner_iterator.next()	×
214	})	×
215	.await;	×
216
217	res.unwrap_or_default()	×
218	}	×
219	}
220
221	impl Iterator for MithrilSnapshotIteratorInner {
222	type Item = MultiEraBlock;
223
224	fn next(&mut self) -> Option<Self::Item> {	×
225	for maybe_block in self.inner.by_ref() {	×
226	if let Ok(block) = maybe_block {	×
227	if !self.previous.is_unknown() {	×
228	// We can safely fully decode this block.
229	match MultiEraBlock::new(self.chain, block, &self.previous, Fork::IMMUTABLE) {	×
230	Ok(block_data) => {	×
231	// Update the previous point	×
232	// debug!("Pre Previous update 1 : {:?}", self.previous);	×
233	self.previous = block_data.point();	×
234	// debug!("Post Previous update 1 : {:?}", self.previous);	×
235		×
236	// Make sure we got to the start, otherwise this could be a block	×
237	// artifact from a discover previous point	×
238	// search.	×
239	if block_data < self.start {	×
240	continue;	×
241	}	×
242		×
243	return Some(block_data);	×
244	},
245	Err(error) => {	×
246	error!(previous=%self.previous, error=%error, "Error decoding a block from the snapshot");	×
247	break;	×
248	},
249	}
250	}	×
251
252	// We cannot fully decode this block because we don't know its previous point,
253	// So this MUST be the first block in iteration, so use it as the previous.
254	if let Ok(raw_decoded_block) =	×
255	pallas::ledger::traverse::MultiEraBlock::decode(&block)	×
256	{
257	// debug!("Pre Previous update 2 : {:?}", self.previous);
258	self.previous = Point::new(	×
259	raw_decoded_block.slot().into(),	×
260	raw_decoded_block.hash().into(),	×
261	);	×
262	// debug!("Post Previous update 2 : {:?}", self.previous);	×
263	continue;	×
264	}	×
265	error!("Error decoding block to use for previous from the snapshot.");	×
266	break;	×
267	}	×
268		×
269	error!("Error while fetching a block from the snapshot");	×
270	break;	×
271	}
272
273	None	×
274	}	×
275	}

input-output-hk / catalyst-libs / 16618685382

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