diff --git a/crates/epix-runtime/src/lib.rs b/crates/epix-runtime/src/lib.rs index 7ea25ca..ba5b4f3 100644 --- a/crates/epix-runtime/src/lib.rs +++ b/crates/epix-runtime/src/lib.rs @@ -480,13 +480,28 @@ async fn announce_loop( // from previous runs, plus the runtime-contributed list (Syncronite's // live bootstrap) - re-read every pass, like EpixNet's loadTrackersFile // in its announce loop. - let all = state.all_trackers(&trackers).await; + let all = std::sync::Arc::new(state.all_trackers(&trackers).await); if all.is_empty() { return; } + // Bounded-concurrent per-xite announces. Strictly serial passes made + // a cold start unusable: each xite's announce waits out its slowest + // tracker (75s ceiling), so a boot pass over dozens of xites with the + // usual dead trackers ground on for tens of minutes - during which + // the loop could not react to Tor coming up either. Four at a time + // keeps the dial pressure modest while cutting the pass to a quarter. + let gate = std::sync::Arc::new(tokio::sync::Semaphore::new(4)); + let mut passes = tokio::task::JoinSet::new(); for address in state.xite_addresses().await { - state.announce_to_trackers(&address, &all).await; + let state = state.clone(); + let all = all.clone(); + let gate = gate.clone(); + passes.spawn(async move { + let Ok(_permit) = gate.acquire_owned().await else { return }; + state.announce_to_trackers(&address, &all).await; + }); } + while passes.join_next().await.is_some() {} // AnnounceBitTorrent: also announce to any configured HTTP(S) BT // trackers and fold their peers in. if let Some(bt) = state.config_get("bt_trackers").await.and_then(|v| v.as_array().cloned()) { diff --git a/crates/epix-ui/src/command.rs b/crates/epix-ui/src/command.rs index eb0d4c4..7a569dc 100644 --- a/crates/epix-ui/src/command.rs +++ b/crates/epix-ui/src/command.rs @@ -1090,14 +1090,27 @@ impl WsCommand for ServerInfo { } /// `announcerStats` - per-tracker announce status for the dashboard. +/// `{planned: true}` also includes overlay-gated trackers (waiting on +/// Tor/I2P), so the health drawer can list everything the node will try. +/// The default answer stays gated-free: dashboards that predate the flag +/// count every returned entry in their health ratio. struct AnnouncerStats; #[async_trait] impl WsCommand for AnnouncerStats { fn name(&self) -> &'static str { "announcerStats" } - async fn handle(&self, s: &WsSession, _p: &Value) -> Result { - Ok(s.state.announcer_stats().await) + async fn handle(&self, s: &WsSession, p: &Value) -> Result { + let planned = p + .get("planned") + .or_else(|| p.as_array().and_then(|a| a.first())) + .and_then(Value::as_bool) + .unwrap_or(false); + if planned { + Ok(s.state.announcer_stats_planned().await) + } else { + Ok(s.state.announcer_stats().await) + } } } diff --git a/crates/epix-ui/src/state.rs b/crates/epix-ui/src/state.rs index f2cd0d1..c662209 100644 --- a/crates/epix-ui/src/state.rs +++ b/crates/epix-ui/src/state.rs @@ -8132,6 +8132,7 @@ impl AppState { // many timeouts and the dashboard's per-tracker stats would trickle in. let mut set = tokio::task::JoinSet::new(); let mut skipped = 0; + let mut seeded = false; for tracker in trackers.iter().cloned() { if tracker_gated(&tracker, tor_on, &tor_st, i2p_on) { self.mark_tracker_gated(&tracker).await; @@ -8141,6 +8142,7 @@ impl AppState { skipped += 1; continue; } + seeded |= self.mark_tracker_announcing(&tracker).await; let sender = sender.clone(); let key = key.clone(); let advert = advert.clone(); @@ -8160,6 +8162,12 @@ impl AppState { (tracker, result, started.elapsed()) }); } + // New trackers entered the stats as "announcing": push right away so + // the dashboard lists what this pass is trying instead of waiting up + // to 75s for the slowest announce to resolve. + if seeded { + self.push_announcer_info(&key).await; + } let all = self.absorb_announce_results(set, &key).await; self.add_peers(address, all.clone()).await; let skip_note = if skipped > 0 { format!(" ({skipped} backed off)") } else { String::new() }; @@ -8288,6 +8296,32 @@ impl AppState { backoff.retain(|_, &mut (_, tried)| now - tried < STALE_SECS); } + /// Seed a stats entry for a tracker this pass is about to try, so the + /// dashboard can list it before its announce resolves. Only a tracker + /// with no entry yet gets the "announcing" status: one with a verdict + /// keeps showing that verdict while the retry is in flight. Returns + /// whether a new entry was created. + async fn mark_tracker_announcing(&self, tracker: &epix_xite::Tracker) -> bool { + let key = tracker_stat_key(tracker); + let now = now_secs(); + let mut stats = self.tracker_stats.write().await; + if let Some(entry) = stats.get_mut(&key) { + // A gated entry whose overlay came back up is being tried now. + let gated = entry.get("status").and_then(|s| s.as_str()) == Some("gated"); + if gated { + if let Some(obj) = entry.as_object_mut() { + obj.insert("status".into(), json!("announcing")); + } + } + return gated; + } + stats.insert( + key, + json!({ "status": "announcing", "num_request": 0, "num_success": 0, "num_error": 0, "num_added": 0, "time_request": 0, "time_success": 0, "time_first_request": now }), + ); + true + } + /// Mark a tracker skipped because its overlay is down (Tor off for an /// onion announcer, no I2P transport for an i2p one) so the dashboard can /// exclude it from the working-tracker denominator instead of counting it @@ -8339,6 +8373,23 @@ impl AppState { ) } + /// [`Self::announcer_stats`] including overlay-gated entries, so the + /// dashboard's health drawer can list every tracker the node plans to + /// try once its overlay comes up. Served only to a caller that asks + /// (`announcerStats {planned: true}`): the shipped dashboard counts + /// every returned entry in its health ratio, so gated rows must never + /// reach it unrequested. + pub async fn announcer_stats_planned(&self) -> Value { + Value::Object( + self.tracker_stats + .read() + .await + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect(), + ) + } + /// Mark a peer connected/disconnected for a xite. pub async fn set_peer_connected(&self, address: &str, addr: &PeerAddr, connected: bool) { if let Some(x) = self.xites.write().await.get_mut(address) { @@ -8645,6 +8696,10 @@ impl AppState { // The dashboard shows the Tor state live (serverChanged). if changed { self.push_server_info().await; + // A Tor state move changes which trackers are dialable. Wake the + // announce loop so onion trackers gated during boot get tried + // seconds after Tor comes up, not at the next periodic pass. + self.trackers_changed.notify_one(); } } @@ -16650,6 +16705,9 @@ impl AppState { }; if changed { self.push_server_info().await; + // Same rule as set_tor_status: an I2P reachability move regates + // i2p trackers, so the announce loop should run early. + self.trackers_changed.notify_one(); } }