|
1 | 1 | use anyhow::{bail, Result}; |
| 2 | +use arc_swap::ArcSwapOption; |
2 | 3 | use axum::extract::State; |
3 | 4 | use axum::http::{header, HeaderMap, StatusCode, Uri}; |
4 | 5 | use axum::response::{IntoResponse, Response}; |
5 | 6 | use axum::{routing::get, Router}; |
6 | 7 | use std::net::SocketAddr; |
7 | 8 | use std::path::{Path, PathBuf}; |
8 | 9 | use std::sync::Arc; |
| 10 | +use std::time::Duration; |
9 | 11 | use tracing::info; |
10 | 12 |
|
11 | 13 | use ckbadger_store::{ |
@@ -33,13 +35,83 @@ pub struct ApiServiceConfig { |
33 | 35 | pub dob_decode_dir: PathBuf, |
34 | 36 | /// Directory where API writes cycles calculation request files for the indexer worker. |
35 | 37 | pub cycles_request_dir: Option<std::path::PathBuf>, |
36 | | - /// Path to the network-crawler store primary. The API opens a read-only |
37 | | - /// secondary only when this primary already exists (opt-in crawler). |
| 38 | + /// Path to the network-crawler store primary. The API attaches a read-only |
| 39 | + /// secondary immediately or when this opt-in primary later appears. |
38 | 40 | pub network_data_path: String, |
39 | 41 | /// Whether the network crawler is enabled in config (surfaced to the UI). |
40 | 42 | pub crawler_enabled: bool, |
41 | 43 | } |
42 | 44 |
|
| 45 | +const NETWORK_STORE_RETRY_INTERVAL: Duration = Duration::from_secs(1); |
| 46 | + |
| 47 | +fn open_network_secondary_if_present( |
| 48 | + primary_path: &Path, |
| 49 | + runtime_config: StoreRuntimeConfig, |
| 50 | +) -> Result<Option<Arc<CkbadgerStore>>> { |
| 51 | + if !primary_path.join("CURRENT").exists() { |
| 52 | + return Ok(None); |
| 53 | + } |
| 54 | + |
| 55 | + let secondary_path = secondary_store_path(primary_path, SecondaryStoreOwner::Api); |
| 56 | + CkbadgerStore::open_network_secondary_with_runtime( |
| 57 | + primary_path, |
| 58 | + secondary_path.as_path(), |
| 59 | + runtime_config, |
| 60 | + ) |
| 61 | + .map(Arc::new) |
| 62 | + .map(Some) |
| 63 | +} |
| 64 | + |
| 65 | +async fn wait_for_network_store_secondary( |
| 66 | + slot: Arc<ArcSwapOption<CkbadgerStore>>, |
| 67 | + primary_path: PathBuf, |
| 68 | + runtime_config: StoreRuntimeConfig, |
| 69 | + retry_interval: Duration, |
| 70 | +) { |
| 71 | + let mut last_error = None; |
| 72 | + loop { |
| 73 | + if slot.load().is_some() { |
| 74 | + return; |
| 75 | + } |
| 76 | + |
| 77 | + let open_primary_path = primary_path.clone(); |
| 78 | + let open_result = tokio::task::spawn_blocking(move || { |
| 79 | + open_network_secondary_if_present(&open_primary_path, runtime_config) |
| 80 | + }) |
| 81 | + .await; |
| 82 | + |
| 83 | + match open_result { |
| 84 | + Ok(Ok(Some(store))) => { |
| 85 | + slot.store(Some(store)); |
| 86 | + info!( |
| 87 | + "Network store secondary attached after crawler startup: {}", |
| 88 | + primary_path.display() |
| 89 | + ); |
| 90 | + return; |
| 91 | + } |
| 92 | + Ok(Ok(None)) => {} |
| 93 | + Ok(Err(error)) => { |
| 94 | + let message = error.to_string(); |
| 95 | + if last_error.as_deref() != Some(message.as_str()) { |
| 96 | + tracing::warn!( |
| 97 | + "network store present but failed to open secondary; will retry: {message}" |
| 98 | + ); |
| 99 | + last_error = Some(message); |
| 100 | + } |
| 101 | + } |
| 102 | + Err(error) => { |
| 103 | + let message = format!("network store secondary open task failed: {error}"); |
| 104 | + if last_error.as_deref() != Some(message.as_str()) { |
| 105 | + tracing::warn!("{message}; will retry"); |
| 106 | + last_error = Some(message); |
| 107 | + } |
| 108 | + } |
| 109 | + } |
| 110 | + |
| 111 | + tokio::time::sleep(retry_interval).await; |
| 112 | + } |
| 113 | +} |
| 114 | + |
43 | 115 | /// One backend network the frontend proxy can route to. |
44 | 116 | #[derive(Clone, Debug)] |
45 | 117 | pub struct FrontendNetwork { |
@@ -116,34 +188,38 @@ pub async fn run_api(config: ApiServiceConfig) -> Result<()> { |
116 | 188 | config.store_runtime_config, |
117 | 189 | )?); |
118 | 190 |
|
119 | | - // The network-crawler store is opt-in: open a read-only secondary only when |
120 | | - // the crawler has already produced a primary (CURRENT marker present). A |
121 | | - // missing primary or an open failure is a normal `None`, never a startup |
122 | | - // error — the API stays read-only and never writes this store. |
123 | | - let network_store = { |
124 | | - let primary = Path::new(&config.network_data_path); |
125 | | - if primary.join("CURRENT").exists() { |
126 | | - let sec = secondary_store_path(&config.network_data_path, SecondaryStoreOwner::Api); |
127 | | - info!( |
128 | | - "Opening ckbadger network store (secondary) at: {} -> {}", |
129 | | - config.network_data_path, |
130 | | - sec.display() |
131 | | - ); |
132 | | - match CkbadgerStore::open_network_secondary_with_runtime( |
133 | | - primary, |
134 | | - sec.as_path(), |
135 | | - config.store_runtime_config, |
136 | | - ) { |
137 | | - Ok(s) => Some(Arc::new(s)), |
138 | | - Err(e) => { |
139 | | - tracing::warn!("network store present but failed to open secondary: {e}"); |
140 | | - None |
141 | | - } |
| 191 | + // The API remains read-only, but unlike the chain stores the opt-in network |
| 192 | + // primary may appear after the API has already started. Try once before |
| 193 | + // constructing the router, then keep retrying in the background until the |
| 194 | + // crawler creates/upgrades the primary and the secondary can be attached. |
| 195 | + let network_primary_path = PathBuf::from(&config.network_data_path); |
| 196 | + let initial_network_store = |
| 197 | + match open_network_secondary_if_present(&network_primary_path, config.store_runtime_config) |
| 198 | + { |
| 199 | + Ok(store) => store, |
| 200 | + Err(error) => { |
| 201 | + tracing::warn!( |
| 202 | + "network store present but failed to open secondary; will retry: {error}" |
| 203 | + ); |
| 204 | + None |
142 | 205 | } |
143 | | - } else { |
144 | | - None |
145 | | - } |
146 | | - }; |
| 206 | + }; |
| 207 | + if initial_network_store.is_some() { |
| 208 | + info!( |
| 209 | + "Opened ckbadger network store (secondary) at: {}", |
| 210 | + config.network_data_path |
| 211 | + ); |
| 212 | + } |
| 213 | + let should_wait_for_network_store = initial_network_store.is_none(); |
| 214 | + let network_store = Arc::new(ArcSwapOption::from(initial_network_store)); |
| 215 | + if should_wait_for_network_store { |
| 216 | + tokio::spawn(wait_for_network_store_secondary( |
| 217 | + network_store.clone(), |
| 218 | + network_primary_path, |
| 219 | + config.store_runtime_config, |
| 220 | + NETWORK_STORE_RETRY_INTERVAL, |
| 221 | + )); |
| 222 | + } |
147 | 223 |
|
148 | 224 | let app_config = AppConfig { |
149 | 225 | store, |
@@ -668,6 +744,51 @@ mod tests { |
668 | 744 | assert!(!config.crawler_enabled); |
669 | 745 | } |
670 | 746 |
|
| 747 | + #[tokio::test] |
| 748 | + async fn network_secondary_attaches_when_primary_appears_after_api_start() { |
| 749 | + use ckbadger_store::LatestStatus; |
| 750 | + |
| 751 | + let dir = tempfile::tempdir().expect("network store tempdir"); |
| 752 | + let primary_path = dir.path().join("network"); |
| 753 | + let slot = Arc::new(ArcSwapOption::from(None)); |
| 754 | + let waiter = tokio::spawn(wait_for_network_store_secondary( |
| 755 | + slot.clone(), |
| 756 | + primary_path.clone(), |
| 757 | + StoreRuntimeConfig::default(), |
| 758 | + Duration::from_millis(10), |
| 759 | + )); |
| 760 | + |
| 761 | + tokio::time::sleep(Duration::from_millis(25)).await; |
| 762 | + assert!(slot.load_full().is_none()); |
| 763 | + |
| 764 | + let primary = |
| 765 | + CkbadgerStore::open_network_with_runtime(&primary_path, StoreRuntimeConfig::default()) |
| 766 | + .expect("open network primary"); |
| 767 | + primary |
| 768 | + .put_network_status(&LatestStatus { |
| 769 | + round_id: 9, |
| 770 | + started: 100, |
| 771 | + finished: 200, |
| 772 | + ..Default::default() |
| 773 | + }) |
| 774 | + .expect("seed network status"); |
| 775 | + |
| 776 | + tokio::time::timeout(Duration::from_secs(5), waiter) |
| 777 | + .await |
| 778 | + .expect("secondary attachment timed out") |
| 779 | + .expect("secondary attachment task panicked"); |
| 780 | + |
| 781 | + let secondary = slot.load_full().expect("network secondary attached"); |
| 782 | + assert_eq!( |
| 783 | + secondary |
| 784 | + .get_network_status() |
| 785 | + .expect("read network status") |
| 786 | + .expect("network status exists") |
| 787 | + .round_id, |
| 788 | + 9 |
| 789 | + ); |
| 790 | + } |
| 791 | + |
671 | 792 | #[test] |
672 | 793 | fn test_frontend_service_config_fields() { |
673 | 794 | let config = FrontendServiceConfig { |
|
0 commit comments