diff options
Diffstat (limited to 'src/controller.rs')
| -rw-r--r-- | src/controller.rs | 97 |
1 files changed, 81 insertions, 16 deletions
diff --git a/src/controller.rs b/src/controller.rs index 1384ad8..4686118 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -47,6 +47,7 @@ pub struct Controller { logs: std::collections::VecDeque<String>, next_check: Option<tokio::time::Instant>, measurements: Vec<crate::diagnostics::Observation>, + cooldowns: std::collections::HashMap<String, Option<tokio::time::Instant>>, } impl Controller { @@ -73,6 +74,7 @@ impl Controller { logs: std::collections::VecDeque::new(), next_check: Some(tokio::time::Instant::now()), measurements: vec![], + cooldowns: std::collections::HashMap::new(), }) } @@ -145,7 +147,7 @@ impl Controller { name: entry.name, host: entry.connection.as_ref().map(|c| c.host.clone()), port: entry.connection.as_ref().map(|c| c.port), - status: entry.status, + status: entry.status.as_str().into(), explanation: entry.explanation, }); } @@ -154,9 +156,13 @@ impl Controller { } pub async fn restore(&mut self) -> Result<()> { - if self.storage.state()?.0 { - self.runtime.arm().await?; - self.reconcile().await?; + let (enabled, selected, _) = self.storage.state()?; + if enabled { + self.runtime.block().await?; + if selected.is_some() { + self.runtime.arm().await?; + self.reconcile().await?; + } } else { self.runtime.disable().await?; } @@ -211,6 +217,16 @@ impl Controller { } Operation::DeleteSubscription { name } => { let id = self.subscription(&name)?.id.clone(); + let (enabled, selected, _) = self.storage.state()?; + if enabled + && self + .storage + .entries(&id)? + .iter() + .any(|entry| Some(&entry.id) == selected.as_ref()) + { + self.runtime.block().await?; + } let mut definitions = self.storage.definitions.clone(); definitions.subscriptions.retain(|s| s.id != id); self.storage.save_definitions(&definitions)?; @@ -241,10 +257,16 @@ impl Controller { } Operation::Select { id } => { self.usable_entry(&id)?; + if self.storage.state()?.0 { + self.runtime.block().await?; + } self.storage.select(Some(&id))?; self.reconcile().await?; } Operation::Unset => { + if self.storage.state()?.0 { + self.runtime.block().await?; + } self.storage.select(None)?; self.reconcile().await?; } @@ -302,21 +324,48 @@ impl Controller { .ok_or_else(|| anyhow!("Select a configuration or supply its ID before testing"))?; let entry = self.usable_entry(id)?; let settings = self.settings(); - let probe = crate::xray::Probe::start( - entry.connection.as_ref().unwrap(), - if speed { - settings.maximum_duration_seconds - } else { - 5 - }, + let url = if speed { + &settings.speed_url + } else { + &settings.ping_url + }; + let provider = url::Url::parse(url)?.origin().ascii_serialization(); + if let Some(deadline) = self.cooldowns.get(&provider) { + ensure!( + deadline.is_some_and(|deadline| tokio::time::Instant::now() >= deadline), + "The diagnostic provider requested a pause; retry after its Retry-After delay" + ); + } + let probe = self + .runtime + .probe( + entry.connection.as_ref().unwrap(), + if speed { + settings.maximum_duration_seconds + } else { + 5 + }, + ) + .await?; + let result = crate::diagnostics::measure( + &settings, + &probe.proxy, + &probe.certificate, + id.into(), + generation, + speed, ) - .await?; - let result = - crate::diagnostics::measure(&settings, &probe.proxy, id.into(), generation, speed) - .await; + .await; probe.stop().await; match result { Ok(observation) => { + if let Some(wait) = observation.retry_after_seconds { + self.cooldowns.insert( + provider, + tokio::time::Instant::now() + .checked_add(std::time::Duration::from_secs(wait)), + ); + } if !speed && Some(id) == selected.as_deref() { let wait = observation.retry_after_seconds.unwrap_or(30).max(30); self.next_check = tokio::time::Instant::now() @@ -421,6 +470,19 @@ impl Controller { } } } + let (enabled, selected, _) = self.storage.state()?; + if enabled { + if let Some(id) = selected { + if let Some(previous) = old.iter().find(|entry| entry.id == id) { + let replacement = entries.iter().find(|entry| entry.id == id); + if replacement.and_then(|entry| entry.connection.as_ref()) + != previous.connection.as_ref() + { + self.runtime.block().await?; + } + } + } + } let affected: std::collections::HashSet<_> = old.iter().map(|e| e.id.as_str()).collect(); self.measurements .retain(|o| !affected.contains(o.configuration_id.as_str())); @@ -443,7 +505,7 @@ impl Controller { .find_entry(id)? .ok_or_else(|| anyhow!("Configuration not found; run configuration list"))?; ensure!( - entry.connection.is_some(), + entry.status == EntryStatus::Supported && entry.connection.is_some(), "Configuration is invalid or unsupported; inspect its explanation" ); Ok(entry) @@ -452,6 +514,9 @@ impl Controller { async fn reconcile(&mut self) -> Result<()> { let (enabled, selected, generation) = self.storage.state()?; if enabled { + if selected.is_none() { + return self.runtime.block().await; + } let entry = selected.map(|id| self.usable_entry(&id)).transpose()?; if let Err(error) = self .runtime |