1#![allow(clippy::await_holding_lock)]
19
20use snarkos_utilities::{SignalHandler, Stoppable};
21
22use snarkvm::{
23 prelude::{Deserialize, DeserializeOwned, Ledger, Network, Serialize, block::Block, store::ConsensusStorage},
24 utilities::{flatten_error, unchecked_deserialize},
25};
26
27use anyhow::{Context, Result, anyhow, bail};
28use colored::Colorize;
29#[cfg(feature = "locktick")]
30use locktick::{parking_lot::Mutex, tokio::Mutex as TMutex};
31#[cfg(not(feature = "locktick"))]
32use parking_lot::Mutex;
33use reqwest::Client;
34use std::{
35 cmp,
36 sync::{
37 Arc,
38 atomic::{AtomicBool, AtomicU32, Ordering},
39 },
40 time::{Duration, Instant},
41};
42#[cfg(not(feature = "locktick"))]
43use tokio::sync::Mutex as TMutex;
44use tokio::task::JoinHandle;
45
46const BLOCKS_PER_FILE: u32 = 50;
48const CONCURRENT_REQUESTS: u32 = 16;
50const MAXIMUM_PENDING_BLOCKS: u32 = BLOCKS_PER_FILE * CONCURRENT_REQUESTS * 2;
52const MAXIMUM_REQUEST_ATTEMPTS: u8 = 10;
54
55pub const CDN_BASE_URL: &str = "https://cdn.provable.com/v0/blocks";
57
58#[cfg(feature = "metrics")]
60fn update_block_metrics(height: u32) {
61 crate::metrics::gauge(crate::metrics::bft::HEIGHT, height as f64);
63}
64
65pub type SyncResult = Result<u32, (u32, anyhow::Error)>;
66
67pub struct CdnBlockSync {
72 base_url: http::Uri,
73 task: Mutex<Option<JoinHandle<SyncResult>>>,
75 done: AtomicBool,
77}
78
79impl CdnBlockSync {
80 pub fn new<N: Network, C: ConsensusStorage<N>>(
82 base_url: http::Uri,
83 ledger: Ledger<N, C>,
84 stoppable: Arc<SignalHandler>,
85 ) -> Self {
86 let task = {
87 let base_url = base_url.clone();
88 tokio::spawn(async move { Self::worker(base_url, ledger, stoppable).await })
89 };
90
91 debug!("Started sync from CDN at {base_url}");
92 Self { done: AtomicBool::new(false), base_url, task: Mutex::new(Some(task)) }
93 }
94
95 pub fn is_done(&self) -> bool {
99 self.done.load(Ordering::SeqCst)
100 }
101
102 pub async fn wait(&self) -> Result<SyncResult> {
107 let Some(hdl) = self.task.lock().take() else {
108 bail!("CDN task was already awaited");
109 };
110
111 let result = hdl.await.map_err(|err| anyhow!("Failed to wait for CDN task: {err}"));
112 self.done.store(true, Ordering::SeqCst);
113 result
114 }
115
116 async fn worker<N: Network, C: ConsensusStorage<N>>(
117 base_url: http::Uri,
118 ledger: Ledger<N, C>,
119 stoppable: Arc<dyn Stoppable>,
120 ) -> SyncResult {
121 let start_height = ledger.latest_height() + 1;
123 let ledger_clone = ledger.clone();
125 let result = load_blocks(&base_url, start_height, None, stoppable, move |block: Block<N>| {
126 ledger_clone
127 .advance_to_next_block(&block)
128 .with_context(|| format!("Failed to advance to block {} at height {}", block.hash(), block.height()))
129 })
130 .await;
131
132 match result {
135 Ok(completed_height) => Ok(completed_height),
136 Err((completed_height, error)) => {
137 warn!("{}", flatten_error(error.context("Failed to sync block(s) from the CDN")));
138
139 if completed_height != start_height {
141 debug!("Synced the ledger up to block {completed_height}");
142
143 let node_height = *ledger.vm().block_store().heights().max().unwrap_or_default();
145 if node_height != completed_height {
147 return Err((
148 completed_height,
149 anyhow!("The ledger height does not match the last sync height"),
150 ));
151 }
152
153 if let Err(err) = ledger.get_block(node_height) {
155 return Err((completed_height, err));
156 }
157 }
158
159 Ok(completed_height)
160 }
161 }
162 }
163
164 pub async fn get_cdn_height(&self) -> anyhow::Result<u32> {
165 let client = Client::builder().use_rustls_tls().build()?;
166 cdn_height::<BLOCKS_PER_FILE>(&client, &self.base_url).await
167 }
168}
169
170pub async fn load_blocks<N: Network>(
175 base_url: &http::Uri,
176 start_height: u32,
177 end_height: Option<u32>,
178 stoppable: Arc<dyn Stoppable>,
179 process: impl FnMut(Block<N>) -> Result<()> + Clone + Send + Sync + 'static,
180) -> Result<u32, (u32, anyhow::Error)> {
181 let client = match Client::builder().use_rustls_tls().build() {
183 Ok(client) => client,
184 Err(error) => {
185 return Err((start_height.saturating_sub(1), anyhow!("Failed to create a CDN request client - {error}")));
186 }
187 };
188
189 let cdn_height = match cdn_height::<BLOCKS_PER_FILE>(&client, base_url).await {
191 Ok(cdn_height) => cdn_height,
192 Err(error) => return Err((start_height, error)),
193 };
194 if cdn_height < start_height {
196 return Err((
197 start_height,
198 anyhow!("The given start height ({start_height}) must be less than the CDN height ({cdn_height})"),
199 ));
200 }
201
202 let end_height = cmp::min(end_height.unwrap_or(cdn_height), cdn_height);
205 if end_height < start_height {
207 return Err((
208 start_height,
209 anyhow!("The given end height ({end_height}) must not be less than the start height ({start_height})"),
210 ));
211 }
212
213 let cdn_start = start_height - (start_height % BLOCKS_PER_FILE);
215 let cdn_end = end_height;
217 if cdn_start >= cdn_end {
219 return Ok(cdn_end);
220 }
221
222 let pending_blocks: Arc<TMutex<Vec<Block<N>>>> = Default::default();
224
225 let timer = Instant::now();
227
228 let pending_blocks_clone = pending_blocks.clone();
230 let base_url = base_url.to_owned();
231
232 {
233 let stoppable = stoppable.clone();
234 tokio::spawn(async move {
235 download_block_bundles(client, &base_url, cdn_start, cdn_end, pending_blocks_clone, stoppable).await;
236 });
237 }
238
239 let threadpool = Arc::new(rayon::ThreadPoolBuilder::new().build().unwrap());
241
242 let mut current_height = start_height.saturating_sub(1);
244 while current_height < end_height - 1 {
245 if stoppable.is_stopped() {
247 info!("Stopping block sync at {} - shutting down", current_height);
248 return Ok(current_height);
250 }
251
252 let mut candidate_blocks = pending_blocks.lock().await;
253
254 let Some(next_height) = candidate_blocks.first().map(|b| b.height()) else {
256 debug!("No pending blocks yet");
257 drop(candidate_blocks);
258 tokio::time::sleep(Duration::from_secs(3)).await;
259 continue;
260 };
261
262 if next_height > current_height + 1 {
264 debug!("Waiting for the first relevant blocks ({} pending)", candidate_blocks.len());
266 drop(candidate_blocks);
267 tokio::time::sleep(Duration::from_secs(1)).await;
268 continue;
269 }
270
271 let retained_blocks = candidate_blocks.split_off(BLOCKS_PER_FILE as usize);
273 let next_blocks = std::mem::replace(&mut *candidate_blocks, retained_blocks);
274 drop(candidate_blocks);
275
276 let mut process_clone = process.clone();
278 let stoppable_clone = stoppable.clone();
279 let threadpool_clone = threadpool.clone();
280 current_height = tokio::task::spawn_blocking(move || {
281 threadpool_clone.install(|| {
282 for block in next_blocks.into_iter().filter(|b| (start_height..end_height).contains(&b.height())) {
283 if stoppable_clone.is_stopped() {
285 info!("Stopping block sync at {} - the node is shutting down", current_height);
286 break;
288 }
289
290 let block_height = block.height();
292
293 process_clone(block)?;
295
296 current_height = block_height;
298
299 #[cfg(feature = "metrics")]
301 update_block_metrics(current_height);
302
303 log_progress::<BLOCKS_PER_FILE>(timer, current_height, cdn_start, cdn_end, "block");
305 }
306
307 Ok(current_height)
308 })
309 })
310 .await
311 .map_err(|e| (current_height, e.into()))?
312 .map_err(|e| (current_height, e))?;
313 }
314
315 Ok(current_height)
316}
317
318async fn download_block_bundles<N: Network>(
319 client: Client,
320 base_url: &http::Uri,
321 cdn_start: u32,
322 cdn_end: u32,
323 pending_blocks: Arc<TMutex<Vec<Block<N>>>>,
324 stoppable: Arc<dyn Stoppable>,
325) {
326 let active_requests: Arc<AtomicU32> = Default::default();
328
329 let mut start = cdn_start;
330 while start < cdn_end - 1 {
331 if stoppable.is_stopped() {
333 break;
334 }
335
336 let num_pending_blocks = pending_blocks.lock().await.len();
338 if num_pending_blocks >= MAXIMUM_PENDING_BLOCKS as usize {
339 debug!("Maximum number of pending blocks reached ({num_pending_blocks}), waiting...");
340 tokio::time::sleep(Duration::from_secs(5)).await;
341 continue;
342 }
343
344 let active_request_count = active_requests.load(Ordering::Relaxed);
347 let num_requests =
348 cmp::min(CONCURRENT_REQUESTS, (MAXIMUM_PENDING_BLOCKS - num_pending_blocks as u32) / BLOCKS_PER_FILE)
349 .saturating_sub(active_request_count);
350
351 for i in 0..num_requests {
353 let start = start + i * BLOCKS_PER_FILE;
354 let end = start + BLOCKS_PER_FILE;
355
356 if end > cdn_end + BLOCKS_PER_FILE {
358 debug!("Finishing network requests to the CDN...");
359 break;
360 }
361
362 let client_clone = client.clone();
363 let base_url_clone = base_url.clone();
364 let pending_blocks_clone = pending_blocks.clone();
365 let active_requests_clone = active_requests.clone();
366 let stoppable_clone = stoppable.clone();
367 tokio::spawn(async move {
368 active_requests_clone.fetch_add(1, Ordering::Relaxed);
370
371 let ctx = format!("blocks {start} to {end}");
372 debug!("Requesting {ctx} (of {cdn_end})");
373
374 let blocks_url = format!("{base_url_clone}/{start}.{end}.blocks");
376 let ctx = format!("blocks {start} to {end}");
377 let mut attempts = 0;
379 let request_time = Instant::now();
380
381 loop {
382 match cdn_get(client_clone.clone(), &blocks_url, &ctx).await {
384 Ok::<Vec<Block<N>>, _>(blocks) => {
385 let mut pending_blocks = pending_blocks_clone.lock().await;
387 for block in blocks {
388 match pending_blocks.binary_search_by_key(&block.height(), |b| b.height()) {
389 Ok(_idx) => warn!("Found a duplicate pending block at height {}", block.height()),
390 Err(idx) => pending_blocks.insert(idx, block),
391 }
392 }
393 debug!("Received {ctx} {}", format!("(in {:.2?})", request_time.elapsed()).dimmed());
394 break;
395 }
396 Err(error) => {
397 attempts += 1;
400 if attempts > MAXIMUM_REQUEST_ATTEMPTS {
401 warn!("Maximum number of requests to {blocks_url} reached - shutting down...");
402 stoppable_clone.stop();
403 break;
404 }
405 tokio::time::sleep(Duration::from_secs(attempts as u64 * 10)).await;
406 warn!("{error} - retrying ({attempts} attempt(s) so far)");
407 }
408 }
409 }
410
411 active_requests_clone.fetch_sub(1, Ordering::Relaxed);
413 });
414 }
415
416 start += BLOCKS_PER_FILE * num_requests;
418
419 tokio::time::sleep(Duration::from_secs(1)).await;
421 }
422
423 debug!("Finished network requests to the CDN");
424}
425
426async fn cdn_height<const BLOCKS_PER_FILE: u32>(client: &Client, base_url: &http::Uri) -> Result<u32> {
431 #[derive(Deserialize, Serialize, Debug)]
433 struct LatestState {
434 exclusive_height: u32,
435 inclusive_height: u32,
436 hash: String,
437 }
438 let latest_json_url = format!("{base_url}/latest.json");
440 let response = match client.get(latest_json_url).send().await {
442 Ok(response) => response,
443 Err(error) => bail!("Failed to fetch the CDN height - {error}"),
444 };
445 let bytes = match response.bytes().await {
447 Ok(bytes) => bytes,
448 Err(error) => bail!("Failed to parse the CDN height response - {error}"),
449 };
450 let latest_state_string = match bincode::deserialize::<String>(&bytes) {
452 Ok(string) => string,
453 Err(error) => {
454 let bytes_as_string = String::from_utf8_lossy(&bytes);
455 bail!("Failed to deserialize the CDN height response - {error} - {bytes_as_string}")
456 }
457 };
458 let tip = match serde_json::from_str::<LatestState>(&latest_state_string) {
460 Ok(latest) => latest.exclusive_height,
461 Err(error) => bail!("Failed to extract the CDN height response - {error}"),
462 };
463 let tip = tip.saturating_sub(10);
465 Ok(tip - (tip % BLOCKS_PER_FILE) + BLOCKS_PER_FILE)
467}
468
469async fn cdn_get<T: 'static + DeserializeOwned + Send>(client: Client, url: &str, ctx: &str) -> Result<T> {
471 let response = match client.get(url).send().await {
473 Ok(response) => response,
474 Err(error) => bail!("Failed to fetch {ctx} - {error}"),
475 };
476 let bytes = match response.bytes().await {
478 Ok(bytes) => bytes,
479 Err(error) => bail!("Failed to parse {ctx} - {error}"),
480 };
481
482 match tokio::task::spawn_blocking(move || (unchecked_deserialize::<T>(&bytes), bytes)).await {
484 Ok((Ok(objects), _)) => Ok(objects),
485 Ok((Err(error), response_bytes)) => {
486 let bytes_as_string = String::from_utf8_lossy(&response_bytes);
487 bail!("Failed to deserialize {ctx} - {error} - {bytes_as_string}")
488 }
489 Err(error) => {
490 bail!("Failed to join task for {ctx} - {error}")
491 }
492 }
493}
494
495fn to_human_readable_duration(duration: Duration) -> String {
502 const SECS_PER_MIN: u64 = 60;
505 const MINS_PER_HOUR: u64 = 60;
506 const SECS_PER_HOUR: u64 = SECS_PER_MIN * MINS_PER_HOUR;
507 const HOURS_PER_DAY: u64 = 24;
508 const SECS_PER_DAY: u64 = SECS_PER_HOUR * HOURS_PER_DAY;
509
510 let duration = duration.as_secs();
511
512 if duration < 1 {
513 "less than one second".to_string()
514 } else if duration < SECS_PER_MIN {
515 format!("{duration} seconds")
516 } else if duration < SECS_PER_HOUR {
517 format!("{} minutes", duration / SECS_PER_MIN)
518 } else if duration < SECS_PER_DAY {
519 let mins = duration / SECS_PER_MIN;
520 format!("{hours} hours and {remainder} minutes", hours = mins / 60, remainder = mins % 60)
521 } else {
522 let days = duration / SECS_PER_DAY;
523 let hours = (duration % SECS_PER_DAY) / SECS_PER_HOUR;
524 format!("{days} days and {hours} hours")
525 }
526}
527
528fn log_progress<const OBJECTS_PER_FILE: u32>(
530 timer: Instant,
531 current_index: u32,
532 cdn_start: u32,
533 mut cdn_end: u32,
534 object_name: &str,
535) {
536 debug_assert!(cdn_start <= cdn_end);
537 debug_assert!(current_index <= cdn_end);
538 debug_assert!(cdn_end >= 1);
539
540 cdn_end -= 1;
542
543 let sync_percentage =
545 (current_index.saturating_sub(cdn_start) * 100).checked_div(cdn_end.saturating_sub(cdn_start)).unwrap_or(100);
546
547 let num_files_done = 1 + (current_index - cdn_start) / OBJECTS_PER_FILE;
549 let num_files_remaining = 1 + (cdn_end.saturating_sub(current_index)) / OBJECTS_PER_FILE;
551 let millis_per_file = timer.elapsed().as_millis() / num_files_done as u128;
553 let slowdown = 100 * num_files_remaining as u128;
555 let time_remaining = {
557 let remaining = num_files_remaining as u128 * millis_per_file + slowdown;
558 to_human_readable_duration(Duration::from_secs((remaining / 1000) as u64))
559 };
560 let estimate = format!("(started at height {cdn_start}, est. {time_remaining} remaining)");
562 info!(
564 "Reached {object_name} {current_index} of {cdn_end} - Sync is {sync_percentage}% complete {}",
565 estimate.dimmed()
566 );
567}
568
569#[cfg(test)]
570mod tests {
571 use super::{BLOCKS_PER_FILE, CDN_BASE_URL, cdn_height, load_blocks, log_progress};
572
573 use snarkos_utilities::SimpleStoppable;
574
575 use snarkvm::prelude::{MainnetV0, block::Block};
576
577 use http::Uri;
578 use parking_lot::RwLock;
579 use std::{sync::Arc, time::Instant};
580
581 type CurrentNetwork = MainnetV0;
582
583 fn check_load_blocks(start: u32, end: Option<u32>, expected: usize) {
584 let blocks = Arc::new(RwLock::new(Vec::new()));
585 let blocks_clone = blocks.clone();
586 let process = move |block: Block<CurrentNetwork>| {
587 blocks_clone.write().push(block);
588 Ok(())
589 };
590
591 let testnet_cdn_url = Uri::try_from(format!("{CDN_BASE_URL}/mainnet")).unwrap();
592
593 let rt = tokio::runtime::Runtime::new().unwrap();
594 rt.block_on(async {
595 let completed_height =
596 load_blocks(&testnet_cdn_url, start, end, SimpleStoppable::new(), process).await.unwrap();
597 assert_eq!(blocks.read().len(), expected);
598 if expected > 0 {
599 assert_eq!(blocks.read().last().unwrap().height(), completed_height);
600 }
601 for (i, block) in blocks.read().iter().enumerate() {
603 assert_eq!(block.height(), start + i as u32);
604 }
605 });
606 }
607
608 #[test]
609 fn test_load_blocks_0_to_50() {
610 let start_height = 0;
611 let end_height = Some(50);
612 check_load_blocks(start_height, end_height, 50);
613 }
614
615 #[test]
616 fn test_load_blocks_50_to_100() {
617 let start_height = 50;
618 let end_height = Some(100);
619 check_load_blocks(start_height, end_height, 50);
620 }
621
622 #[test]
623 fn test_load_blocks_0_to_123() {
624 let start_height = 0;
625 let end_height = Some(123);
626 check_load_blocks(start_height, end_height, 123);
627 }
628
629 #[test]
630 fn test_load_blocks_46_to_234() {
631 let start_height = 46;
632 let end_height = Some(234);
633 check_load_blocks(start_height, end_height, 188);
634 }
635
636 #[test]
637 fn test_cdn_height() {
638 let rt = tokio::runtime::Runtime::new().unwrap();
639 let client = reqwest::Client::builder().use_rustls_tls().build().unwrap();
640 let testnet_cdn_url = Uri::try_from(format!("{CDN_BASE_URL}/mainnet")).unwrap();
641 rt.block_on(async {
642 let height = cdn_height::<BLOCKS_PER_FILE>(&client, &testnet_cdn_url).await.unwrap();
643 assert!(height > 0);
644 });
645 }
646
647 #[test]
648 fn test_log_progress() {
649 let timer = Instant::now();
651 let cdn_start = 0;
652 let cdn_end = 100;
653 let object_name = "blocks";
654 log_progress::<10>(timer, 0, cdn_start, cdn_end, object_name);
655 log_progress::<10>(timer, 10, cdn_start, cdn_end, object_name);
656 log_progress::<10>(timer, 20, cdn_start, cdn_end, object_name);
657 log_progress::<10>(timer, 30, cdn_start, cdn_end, object_name);
658 log_progress::<10>(timer, 40, cdn_start, cdn_end, object_name);
659 log_progress::<10>(timer, 50, cdn_start, cdn_end, object_name);
660 log_progress::<10>(timer, 60, cdn_start, cdn_end, object_name);
661 log_progress::<10>(timer, 70, cdn_start, cdn_end, object_name);
662 log_progress::<10>(timer, 80, cdn_start, cdn_end, object_name);
663 log_progress::<10>(timer, 90, cdn_start, cdn_end, object_name);
664 log_progress::<10>(timer, 100, cdn_start, cdn_end, object_name);
665 }
666}