main.rs (24971B)
1 use std::{ 2 collections::BTreeMap, 3 net::SocketAddr, 4 path::{Path, PathBuf}, 5 sync::Arc, 6 time::{Duration, Instant}, 7 }; 8 9 use anyhow::{Context, Result, bail}; 10 use iuna::{ 11 adapters::{ 12 chain_store::SqliteChainStore, config_store, http, p2p, stratum, 13 ui_data_store::SqliteUiDataStore, wallet_store, 14 }, 15 app::{ 16 NodeCore, PeerBook, SharedNode, SharedPeerBook, StratumStatus, debug_logging_enabled, 17 now_ms, set_debug_logging, 18 }, 19 domain::{ 20 Amount, ChainSnapshot, GenesisBurn, Ledger, MAX_VDF_ROUNDS, MICRO_IUNA, 21 VDF_TARGET_BLOCK_MS, run_vdf, 22 }, 23 }; 24 use tokio::sync::Mutex; 25 26 mod cli; 27 use cli::{ 28 ChainMode, CliOptions, apply_cli_p2p_config_overrides, configured_p2p_announce_addr, 29 configured_p2p_bind_addr, initial_burn_fee, initial_burn_per_block, validate_wallet_for_mode, 30 }; 31 #[cfg(test)] 32 use cli::{default_data_dir, help_text}; 33 34 const GENESIS_BOOTSTRAP_BURN_AMOUNT: Amount = MICRO_IUNA; 35 const GENESIS_INITIAL_BURN_PER_BLOCK: Amount = config_store::DEFAULT_BURN_AMOUNT; 36 const GENESIS_INITIAL_BURN_FEE: Amount = config_store::DEFAULT_BURN_FEE; 37 const VDF_MEASUREMENT_INITIAL_ROUNDS: u64 = 1_000; 38 const VDF_MEASUREMENT_MAX_ROUNDS: u64 = 10_000_000; 39 const VDF_MEASUREMENT_MIN_ELAPSED: Duration = Duration::from_millis(150); 40 41 #[tokio::main] 42 async fn main() -> Result<()> { 43 let Some(opts) = CliOptions::parse()? else { 44 return Ok(()); 45 }; 46 set_debug_logging(opts.debug); 47 let debug_logging = opts.debug; 48 let wallet_path = opts.wallet_path(); 49 let config_path = opts.config_path(); 50 let wallet_file_exists = wallet_path.exists(); 51 validate_wallet_for_mode(&opts, &wallet_path, wallet_file_exists)?; 52 let chain_db_path = opts.chain_db_path(); 53 let ui_data_db_path = ui_data_db_path(&chain_db_path); 54 let chain_store = SqliteChainStore::open(&chain_db_path)?; 55 let ui_data_store = SqliteUiDataStore::open(&ui_data_db_path)?; 56 let persisted_chain_exists = chain_store.load()?.is_some(); 57 if opts.chain_mode == ChainMode::Genesis && persisted_chain_exists { 58 bail!( 59 "--genesis refuses to run because chain database already contains a blockchain at {}; start without --genesis to resume it", 60 chain_store.path().display() 61 ); 62 } 63 let mut ui_config = config_store::load_or_create(&config_path)?; 64 let ui_config_dirty = apply_cli_p2p_config_overrides(&opts, &mut ui_config); 65 let p2p_announce_addr = configured_p2p_announce_addr(&opts, &ui_config)?; 66 let configured_p2p_addr = configured_p2p_bind_addr(&opts, &ui_config); 67 let p2p_accept_inbound = ui_config.p2p_accept_inbound; 68 let advertised_p2p_addr = p2p_announce_addr.unwrap_or(configured_p2p_addr); 69 let wallet_load = load_startup_wallet(&wallet_path)?; 70 let wallet_address = wallet_load.address().to_string(); 71 if opts.chain_mode == ChainMode::Genesis { 72 ui_config.setup_complete = false; 73 ui_config.mining_enabled = true; 74 ui_config.pow_mining_enabled = false; 75 ui_config.burn_per_block = GENESIS_INITIAL_BURN_PER_BLOCK; 76 ui_config.burn_fee = GENESIS_INITIAL_BURN_FEE; 77 config_store::save(&config_path, &ui_config)?; 78 } else if ui_config_dirty { 79 config_store::save(&config_path, &ui_config)?; 80 } 81 let ledger = 82 initialize_ledger(&opts, &wallet_address, &chain_store, advertised_p2p_addr).await?; 83 let has_chain = opts.has_chain() || persisted_chain_exists; 84 let initial_burn_per_block = initial_burn_per_block(&opts, &ui_config); 85 let initial_burn_fee = initial_burn_fee(&opts, &ui_config); 86 87 let mut node_core = match wallet_load { 88 StartupWallet::Unlocked { 89 wallet, 90 owned_blinded_transactions, 91 } => { 92 let mut node = NodeCore::from_ledger_with_burn_fee_and_enabled( 93 wallet, 94 ledger, 95 ui_config.mining_enabled, 96 initial_burn_per_block, 97 initial_burn_fee, 98 ); 99 node.restore_owned_blinded_transactions(owned_blinded_transactions)?; 100 node 101 } 102 StartupWallet::Locked { address } => NodeCore::from_locked_wallet_address( 103 address, 104 ledger, 105 ui_config.mining_enabled, 106 initial_burn_per_block, 107 initial_burn_fee, 108 ), 109 }; 110 node_core.set_pow_mining_workers(ui_config.pow_mining_workers); 111 node_core.set_pow_mining_enabled(ui_config.pow_mining_enabled); 112 node_core.set_recovery_vdf_top_rank_percent(ui_config.recovery_vdf_top_rank_percent); 113 let node: SharedNode = Arc::new(Mutex::new(node_core)); 114 let ui_config = Arc::new(Mutex::new(ui_config)); 115 let mut peers = ui_config.lock().await.peers.clone(); 116 peers.extend(opts.peers); 117 let peers: SharedPeerBook = Arc::new(Mutex::new(PeerBook::from_addresses(peers))); 118 if has_chain { 119 let initial_snapshot = { node.lock().await.chain_snapshot() }; 120 let keep_metrics = ui_config.lock().await.keep_track_of_metrics; 121 persist_chain_snapshot(&chain_store, initial_snapshot.clone()).await?; 122 warm_ui_data_store(&ui_data_store, initial_snapshot, keep_metrics).await?; 123 } else { 124 clear_ui_data_store(&ui_data_store).await?; 125 } 126 127 println!("iuna wallet: {}", node.lock().await.wallet_address()); 128 if node.lock().await.wallet_is_locked() { 129 println!("wallet locked: unlock it in the management UI"); 130 } 131 println!("wallet file: {}", wallet_path.display()); 132 println!("config file: {}", config_path.display()); 133 println!("chain database: {}", chain_store.path().display()); 134 println!("UI data database: {}", ui_data_store.path().display()); 135 println!("management UI: http://{}", opts.http_addr); 136 if p2p_accept_inbound { 137 println!("p2p listener: {}", configured_p2p_addr); 138 } else { 139 println!("p2p listener: disabled (outbound-only)"); 140 } 141 if p2p_accept_inbound { 142 if let Some(addr) = p2p_announce_addr { 143 println!("p2p announce address: {addr}"); 144 } 145 } 146 println!( 147 "automatic finalization: VDF-driven, burning {} IUNA per block with {} IUNA per byte fee rate", 148 format_iuna(initial_burn_per_block), 149 format_iuna(initial_burn_fee) 150 ); 151 152 let gossip = p2p::GossipNetwork::start( 153 Arc::clone(&node), 154 Arc::clone(&peers), 155 configured_p2p_addr, 156 p2p_announce_addr, 157 p2p_accept_inbound, 158 ) 159 .await?; 160 let mut stratum_status = StratumStatus { 161 enabled: false, 162 listen_addr: None, 163 }; 164 if let Some(stratum_addr) = opts.stratum_addr { 165 let stratum = 166 stratum::StratumServer::start(Arc::clone(&node), gossip.clone(), stratum_addr).await?; 167 println!("stratum listener: {}", stratum.listen_addr()); 168 stratum_status = StratumStatus { 169 enabled: true, 170 listen_addr: Some(stratum.listen_addr().to_string()), 171 }; 172 } 173 174 let persistence_node = Arc::clone(&node); 175 let persistence_store = chain_store.clone(); 176 let persistence_ui_data_store = ui_data_store.clone(); 177 let persistence_config = Arc::clone(&ui_config); 178 let persistence_initial_tip = { 179 let node = node.lock().await; 180 if node.has_real_chain() { 181 Some(node.chain_tip_hash()) 182 } else { 183 None 184 } 185 }; 186 tokio::spawn(async move { 187 run_chain_persistence( 188 persistence_node, 189 persistence_store, 190 persistence_ui_data_store, 191 persistence_config, 192 persistence_initial_tip, 193 ) 194 .await; 195 }); 196 197 let finalizer_node = Arc::clone(&node); 198 let finalizer_gossip = gossip.clone(); 199 tokio::spawn(async move { 200 run_automatic_finalizer(finalizer_node, finalizer_gossip, debug_logging).await; 201 }); 202 203 let pow_miner_node = Arc::clone(&node); 204 let pow_miner_gossip = gossip.clone(); 205 tokio::spawn(async move { 206 run_automatic_pow_miner(pow_miner_node, pow_miner_gossip, debug_logging).await; 207 }); 208 209 let sync_node = Arc::clone(&node); 210 let sync_gossip = gossip.clone(); 211 tokio::spawn(async move { 212 run_peer_sync(sync_node, sync_gossip, debug_logging).await; 213 }); 214 215 if !has_chain { 216 println!("setup mode: waiting to join or create a chain"); 217 } 218 219 http::serve( 220 node, 221 peers, 222 gossip, 223 ui_config, 224 http::ServeOptions { 225 config_path, 226 chain_store, 227 ui_data_store, 228 wallet_path, 229 stratum: stratum_status, 230 addr: opts.http_addr, 231 }, 232 ) 233 .await 234 } 235 236 enum StartupWallet { 237 Unlocked { 238 wallet: iuna::domain::Wallet, 239 owned_blinded_transactions: Vec<iuna::domain::OwnedBlindedTransaction>, 240 }, 241 Locked { 242 address: String, 243 }, 244 } 245 246 impl StartupWallet { 247 fn address(&self) -> &str { 248 match self { 249 Self::Unlocked { wallet, .. } => wallet.address(), 250 Self::Locked { address } => address, 251 } 252 } 253 } 254 255 fn load_startup_wallet(wallet_path: &Path) -> Result<StartupWallet> { 256 match wallet_store::load_or_create(wallet_path) { 257 Ok(wallet) => { 258 let owned_blinded_transactions = 259 wallet_store::load_owned_blinded_transactions(wallet_path, None)?; 260 Ok(StartupWallet::Unlocked { 261 wallet, 262 owned_blinded_transactions, 263 }) 264 } 265 Err(error) => { 266 let Some(metadata) = wallet_store::metadata(wallet_path)? else { 267 return Err(error); 268 }; 269 if metadata.encrypted { 270 Ok(StartupWallet::Locked { 271 address: metadata.address, 272 }) 273 } else { 274 Err(error) 275 } 276 } 277 } 278 } 279 280 fn format_iuna(amount: Amount) -> String { 281 let whole = amount / MICRO_IUNA; 282 let fractional = amount % MICRO_IUNA; 283 if fractional == 0 { 284 whole.to_string() 285 } else { 286 let mut fractional = format!("{fractional:06}"); 287 while fractional.ends_with('0') { 288 fractional.pop(); 289 } 290 format!("{whole}.{fractional}") 291 } 292 } 293 294 fn ui_data_db_path(chain_db_path: &Path) -> PathBuf { 295 chain_db_path.with_file_name("ui_data.sqlite3") 296 } 297 298 async fn initialize_ledger( 299 opts: &CliOptions, 300 wallet_address: &str, 301 chain_store: &SqliteChainStore, 302 advertised_p2p_addr: SocketAddr, 303 ) -> Result<Ledger> { 304 if let Some(snapshot) = chain_store.load()? { 305 if opts.chain_mode == ChainMode::Genesis { 306 bail!( 307 "--genesis refuses to run because chain database already contains a blockchain at {}; start without --genesis to resume it", 308 chain_store.path().display() 309 ); 310 } 311 let height = snapshot_height(&snapshot); 312 let ledger = Ledger::from_persisted_snapshot(snapshot).with_context(|| { 313 format!( 314 "failed to load chain database {}", 315 chain_store.path().display() 316 ) 317 })?; 318 println!( 319 "resumed chain from {} at height {height}", 320 chain_store.path().display() 321 ); 322 Ok(ledger) 323 } else { 324 match opts.chain_mode { 325 ChainMode::Setup => Ok(setup_ledger()), 326 ChainMode::Genesis => start_genesis_ledger(wallet_address), 327 ChainMode::Join => join_chain_ledger(&opts.join_peers, advertised_p2p_addr).await, 328 } 329 } 330 } 331 332 fn snapshot_height(snapshot: &ChainSnapshot) -> u64 { 333 snapshot 334 .blocks 335 .last() 336 .map(|block| block.height) 337 .unwrap_or(0) 338 } 339 340 fn setup_ledger() -> Ledger { 341 Ledger::new(BTreeMap::new(), 1) 342 } 343 344 fn start_genesis_ledger(wallet_address: &str) -> Result<Ledger> { 345 let vdf_rounds = measure_initial_vdf_rounds(); 346 let mut genesis = BTreeMap::new(); 347 genesis.insert(wallet_address.to_string(), GENESIS_BOOTSTRAP_BURN_AMOUNT); 348 Ledger::new_with_genesis_burns( 349 genesis, 350 vec![GenesisBurn::new( 351 wallet_address, 352 GENESIS_BOOTSTRAP_BURN_AMOUNT, 353 )], 354 vdf_rounds, 355 ) 356 } 357 358 fn measure_initial_vdf_rounds() -> u64 { 359 let seed = "iuna-vdf-calibration"; 360 let (measured_rounds, elapsed) = measure_vdf_rounds( 361 seed, 362 VDF_MEASUREMENT_INITIAL_ROUNDS, 363 VDF_MEASUREMENT_MIN_ELAPSED, 364 VDF_MEASUREMENT_MAX_ROUNDS, 365 ); 366 let rounds = extrapolate_vdf_rounds( 367 measured_rounds, 368 elapsed, 369 Duration::from_millis(VDF_TARGET_BLOCK_MS), 370 ); 371 println!( 372 "measured {measured_rounds} VDF rounds in {:.3}ms; initial VDF rounds: {rounds}", 373 elapsed.as_secs_f64() * 1000.0 374 ); 375 rounds 376 } 377 378 fn measure_vdf_rounds( 379 seed: &str, 380 initial_rounds: u64, 381 min_elapsed: Duration, 382 max_rounds_per_attempt: u64, 383 ) -> (u64, Duration) { 384 let mut rounds = initial_rounds.max(1).min(max_rounds_per_attempt.max(1)); 385 let mut measured_rounds = 0_u64; 386 let mut measured_elapsed = Duration::ZERO; 387 388 loop { 389 let started = Instant::now(); 390 let _ = run_vdf(seed, rounds); 391 measured_elapsed += started.elapsed(); 392 measured_rounds = measured_rounds.saturating_add(rounds); 393 394 if measured_elapsed >= min_elapsed || rounds >= max_rounds_per_attempt { 395 return (measured_rounds, measured_elapsed); 396 } 397 rounds = rounds.saturating_mul(2).min(max_rounds_per_attempt); 398 } 399 } 400 401 fn extrapolate_vdf_rounds(measured_rounds: u64, elapsed: Duration, target: Duration) -> u64 { 402 let elapsed_ns = elapsed.as_nanos().max(1); 403 let target_ns = target.as_nanos().max(1); 404 let rounds = u128::from(measured_rounds) 405 .saturating_mul(target_ns) 406 .saturating_div(elapsed_ns) 407 .max(1); 408 rounds.min(u128::from(MAX_VDF_ROUNDS)) as u64 409 } 410 411 async fn join_chain_ledger(join_peers: &[String], advertised_addr: SocketAddr) -> Result<Ledger> { 412 let mut errors = Vec::new(); 413 for peer in join_peers { 414 match p2p::fetch_snapshot_with_announcement(peer, Some(advertised_addr)).await { 415 Ok(snapshot) => { 416 let height = snapshot 417 .blocks 418 .last() 419 .map(|block| block.height) 420 .unwrap_or(0); 421 println!("joined chain from {peer} at height {height}"); 422 return Ledger::from_snapshot(snapshot); 423 } 424 Err(error) => { 425 errors.push(format!("{peer}: {error:#}")); 426 } 427 } 428 } 429 430 bail!( 431 "could not join any requested peer; refusing to start a separate chain: {}", 432 errors.join("; ") 433 ) 434 } 435 436 async fn run_automatic_finalizer(node: SharedNode, gossip: p2p::GossipNetwork, debug: bool) { 437 let mut last_logged_skip: Option<(u64, String)> = None; 438 loop { 439 if !node.lock().await.has_real_chain() { 440 tokio::time::sleep(std::time::Duration::from_secs(1)).await; 441 continue; 442 } 443 let (height, plan, outbox) = { 444 let mut node = node.lock().await; 445 let height = node.chain_height(); 446 let plan = node.prepare_automatic_finalization(now_ms()); 447 let outbox = node.drain_outbox(); 448 (height, plan, outbox) 449 }; 450 451 if let Err(error) = gossip.broadcast(outbox).await { 452 if debug { 453 eprintln!("p2p broadcast failed after automatic burn: {error:#}"); 454 } 455 } 456 457 let Some(work) = plan.work else { 458 if let Some(reason) = &plan.skipped_reason { 459 let skip = (height, reason.clone()); 460 if debug && last_logged_skip.as_ref() != Some(&skip) { 461 println!("auto-finalization skipped at height {height}: {reason}"); 462 last_logged_skip = Some(skip); 463 } 464 } 465 tokio::time::sleep(std::time::Duration::from_secs(1)).await; 466 continue; 467 }; 468 469 last_logged_skip = None; 470 if debug { 471 println!( 472 "leader selected locally for candidate block {}; running VDF for {} rounds", 473 work.height(), 474 work.vdf_rounds() 475 ); 476 } 477 478 let seed = work.vdf_seed().to_string(); 479 let rounds = work.vdf_rounds(); 480 let publish_at_ms = work.timestamp_ms(); 481 let vdf_output = match tokio::task::spawn_blocking(move || run_vdf(&seed, rounds)).await { 482 Ok(output) => output, 483 Err(error) => { 484 if debug { 485 eprintln!("VDF worker failed: {error:#}"); 486 } 487 continue; 488 } 489 }; 490 491 let completed_at_ms = now_ms(); 492 let publish_timestamp_ms = completed_at_ms.max(publish_at_ms); 493 if completed_at_ms < publish_at_ms { 494 let wait_ms = publish_at_ms - completed_at_ms; 495 if debug { 496 println!( 497 "VDF completed early for candidate block {}; waiting {:.3}s for rank time slot", 498 work.height(), 499 wait_ms as f64 / 1000.0 500 ); 501 } 502 tokio::time::sleep(std::time::Duration::from_millis(wait_ms)).await; 503 } 504 505 let (finalized, outbox) = { 506 let mut node = node.lock().await; 507 let finalized = node.complete_prepared_block_at(work, vdf_output, publish_timestamp_ms); 508 let outbox = node.drain_outbox(); 509 (finalized, outbox) 510 }; 511 512 match finalized { 513 Ok(block) if debug => { 514 println!("auto-finalized block {} ({})", block.height, block.hash); 515 } 516 Ok(_) => {} 517 Err(error) if debug => println!("auto-finalization skipped after VDF: {error:#}"), 518 Err(_) => {} 519 } 520 521 if let Err(error) = gossip.broadcast(outbox).await { 522 if debug { 523 eprintln!("p2p broadcast failed after automatic block: {error:#}"); 524 } 525 } 526 527 tokio::task::yield_now().await; 528 } 529 } 530 531 async fn run_automatic_pow_miner(node: SharedNode, gossip: p2p::GossipNetwork, debug: bool) { 532 loop { 533 tokio::time::sleep(std::time::Duration::from_secs(1)).await; 534 let (height, job) = { 535 let mut node = node.lock().await; 536 if !node.pow_mining_enabled() { 537 continue; 538 } 539 if !node.has_real_chain() { 540 continue; 541 } 542 let height = node.chain_height(); 543 let job = match node.prepare_automatic_pow_mining_job() { 544 Ok(job) => job, 545 Err(error) => { 546 node.record_automatic_pow_mining_error(format!( 547 "automatic PoW mining failed: {error:#}" 548 )); 549 None 550 } 551 }; 552 (height, job) 553 }; 554 let Some(job) = job else { 555 continue; 556 }; 557 558 let search = tokio::task::spawn_blocking(move || job.search()).await; 559 let (pow_mined, outbox) = { 560 let mut node = node.lock().await; 561 let pow_mined = match search { 562 Ok(Ok((job, outcome))) => { 563 match node.finish_automatic_pow_mining_job(job, outcome) { 564 Ok(tx) => tx, 565 Err(error) => { 566 node.record_automatic_pow_mining_error(format!( 567 "automatic PoW mining failed: {error:#}" 568 )); 569 None 570 } 571 } 572 } 573 Ok(Err(error)) => { 574 node.record_automatic_pow_mining_error(format!( 575 "automatic PoW mining failed: {error:#}" 576 )); 577 None 578 } 579 Err(error) => { 580 node.record_automatic_pow_mining_error(format!( 581 "automatic PoW mining task failed: {error:#}" 582 )); 583 None 584 } 585 }; 586 let outbox = node.drain_outbox(); 587 (pow_mined, outbox) 588 }; 589 590 if let Err(error) = gossip.broadcast(outbox).await { 591 if debug { 592 eprintln!("p2p broadcast failed after automatic PoW mining: {error:#}"); 593 } 594 } 595 596 if debug { 597 if let Some(tx) = &pow_mined { 598 println!( 599 "auto-pow queued mine action for height {} ({})", 600 height, 601 tx.signature() 602 ); 603 } 604 } 605 } 606 } 607 608 async fn run_peer_sync(node: SharedNode, gossip: p2p::GossipNetwork, debug: bool) { 609 loop { 610 tokio::time::sleep(std::time::Duration::from_secs(5)).await; 611 let envelopes = { 612 let mut node = node.lock().await; 613 let mut envelopes = vec![node.peer_status()]; 614 envelopes.extend(node.drain_outbox()); 615 envelopes.extend(node.mempool_gossip()); 616 envelopes 617 }; 618 let mut envelopes = envelopes; 619 envelopes.push(gossip.peer_exchange().await); 620 if let Err(error) = gossip.broadcast(envelopes).await { 621 if debug { 622 eprintln!("p2p sync gossip failed: {error:#}"); 623 } 624 } 625 } 626 } 627 628 async fn run_chain_persistence( 629 node: SharedNode, 630 store: SqliteChainStore, 631 ui_data_store: SqliteUiDataStore, 632 ui_config: Arc<Mutex<config_store::UiConfig>>, 633 initial_saved_tip: Option<String>, 634 ) { 635 run_chain_persistence_with_interval( 636 node, 637 store, 638 ui_data_store, 639 ui_config, 640 Duration::from_secs(2), 641 initial_saved_tip, 642 ) 643 .await; 644 } 645 646 async fn run_chain_persistence_with_interval( 647 node: SharedNode, 648 store: SqliteChainStore, 649 ui_data_store: SqliteUiDataStore, 650 ui_config: Arc<Mutex<config_store::UiConfig>>, 651 interval: Duration, 652 initial_saved_tip: Option<String>, 653 ) { 654 let mut last_saved_tip = initial_saved_tip; 655 loop { 656 tokio::time::sleep(interval).await; 657 let snapshot = { 658 let node = node.lock().await; 659 if !node.has_real_chain() { 660 continue; 661 } 662 node.chain_snapshot() 663 }; 664 let Some(tip_hash) = snapshot.blocks.last().map(|block| block.hash.clone()) else { 665 continue; 666 }; 667 if last_saved_tip.as_deref() == Some(tip_hash.as_str()) { 668 continue; 669 } 670 671 let keep_metrics = ui_config.lock().await.keep_track_of_metrics; 672 match persist_chain_and_project_ui_data(&store, &ui_data_store, snapshot, keep_metrics) 673 .await 674 { 675 Ok(()) => last_saved_tip = Some(tip_hash), 676 Err(error) if debug_logging_enabled() => { 677 eprintln!("chain persistence failed: {error:#}") 678 } 679 Err(_) => {} 680 } 681 } 682 } 683 684 async fn persist_chain_and_project_ui_data( 685 store: &SqliteChainStore, 686 ui_data_store: &SqliteUiDataStore, 687 snapshot: ChainSnapshot, 688 keep_metrics: bool, 689 ) -> Result<()> { 690 persist_chain_snapshot(store, snapshot.clone()).await?; 691 project_ui_data_store(ui_data_store, snapshot, keep_metrics).await 692 } 693 694 async fn persist_chain_snapshot(store: &SqliteChainStore, snapshot: ChainSnapshot) -> Result<()> { 695 let store = store.clone(); 696 tokio::task::spawn_blocking(move || store.save(&snapshot)) 697 .await 698 .context("chain persistence worker failed")??; 699 Ok(()) 700 } 701 702 async fn warm_ui_data_store( 703 store: &SqliteUiDataStore, 704 snapshot: ChainSnapshot, 705 keep_metrics: bool, 706 ) -> Result<()> { 707 println!("warming UI data database..."); 708 let started = Instant::now(); 709 project_ui_data_store(store, snapshot, keep_metrics).await?; 710 println!( 711 "UI data database ready in {:.2}s", 712 started.elapsed().as_secs_f64() 713 ); 714 Ok(()) 715 } 716 717 async fn project_ui_data_store( 718 store: &SqliteUiDataStore, 719 snapshot: ChainSnapshot, 720 keep_metrics: bool, 721 ) -> Result<()> { 722 let store = store.clone(); 723 tokio::task::spawn_blocking(move || store.project_snapshot(&snapshot, keep_metrics)) 724 .await 725 .context("UI data projection worker failed")??; 726 Ok(()) 727 } 728 729 async fn clear_ui_data_store(store: &SqliteUiDataStore) -> Result<()> { 730 println!("clearing UI data database..."); 731 let started = Instant::now(); 732 let store = store.clone(); 733 tokio::task::spawn_blocking(move || store.clear_all()) 734 .await 735 .context("UI data cleanup worker failed")??; 736 println!( 737 "UI data database ready in {:.2}s", 738 started.elapsed().as_secs_f64() 739 ); 740 Ok(()) 741 } 742 743 #[cfg(test)] 744 #[path = "main_tests.rs"] 745 mod tests;