diff --git a/src-tauri/src/config/profiles.rs b/src-tauri/src/config/profiles.rs index c6e84c90d..30f1c9c1b 100644 --- a/src-tauri/src/config/profiles.rs +++ b/src-tauri/src/config/profiles.rs @@ -3,6 +3,7 @@ use super::{ prfitem::{PrfItem, PrfSelected, normalize_profile_home_url}, }; use crate::core::notify::{Refresh, announce}; +use crate::utils::retry::{RetryError, RetryPolicy, retry, retry_with_state}; use crate::{ core::{handle, tray::Tray}, utils::{ @@ -17,6 +18,7 @@ use parking_lot::Mutex; use serde::{Deserialize, Serialize}; use serde_yaml_ng::Mapping; use smartstring::alias::String; +use std::num::NonZeroUsize; use std::{ collections::{HashMap, HashSet}, path::{Component, Path}, @@ -601,19 +603,21 @@ fn is_activation_current(generation: u64) -> bool { } async fn fetch_proxies_with_timeout() -> Result { - tokio::time::timeout(MIHOMO_OPERATION_TIMEOUT, async { - loop { - match handle::Handle::mihomo().get_proxies().await { - Ok(proxies) => return proxies, - Err(err) => { - logging!(debug, Type::Config, "mihomo proxies are not ready yet: {err:#}"); - tokio::time::sleep(Duration::from_millis(500)).await; - } - } - } - }) + let interval = Duration::from_millis(500); + let attempts = + NonZeroUsize::MIN.saturating_add((MIHOMO_OPERATION_TIMEOUT.as_millis() / interval.as_millis()) as usize + 1); + tokio::time::timeout( + MIHOMO_OPERATION_TIMEOUT, + retry(RetryPolicy::fixed(attempts, interval), |_| async { + handle::Handle::mihomo().get_proxies().await.map_err(|error| { + logging!(debug, Type::Config, "mihomo proxies are not ready yet: {error:#}"); + RetryError::Retry(error) + }) + }), + ) .await - .context("timed out while waiting for mihomo proxies") + .context("timed out while waiting for mihomo proxies")? + .map_err(Into::into) } async fn select_node_with_timeout(group_name: &String, node: &String) -> Result<()> { @@ -829,48 +833,58 @@ fn unsettled_selections(selected: &[PrfSelected], proxies: &Proxies) -> Vec, generation: u64) { let deadline = Instant::now() + SELECTED_NODES_SETTLE_DEADLINE; - loop { - tokio::time::sleep(SELECTED_NODES_SETTLE_INTERVAL).await; - if !is_activation_current(generation) { - return; - } - let Ok(snapshot) = fetch_proxies_with_timeout().await else { - // Unreachable core: the deadline still applies, so this cannot spin forever. - if Instant::now() >= deadline { - return; - } - continue; - }; - if !is_activation_current(generation) { - return; - } - - let pending = unsettled_selections(selected, &snapshot); - if pending.is_empty() { - return; - } - if Instant::now() >= deadline { - logging!( - warn, - Type::Config, - "gave up putting back {} selected node(s) the core never loaded: {}", - pending.len(), - pending.iter().map(String::as_str).collect::>().join(", ") - ); - return; - } - - let plan = reconcile_selected_nodes(selected, None, &snapshot); - if apply_activations(&plan.activations, completed, generation) - .await - .is_none() - { - return; - } - if is_activation_current(generation) { - announce(Refresh::Clash); - } - } + let attempts = NonZeroUsize::MIN.saturating_add( + (SELECTED_NODES_SETTLE_DEADLINE.as_millis() / SELECTED_NODES_SETTLE_INTERVAL.as_millis()) as usize, + ); + tokio::time::sleep(SELECTED_NODES_SETTLE_INTERVAL).await; + let _ = retry_with_state( + RetryPolicy::fixed(attempts, SELECTED_NODES_SETTLE_INTERVAL), + (selected, completed, generation, deadline), + |state, _| { + Box::pin(async move { + let (selected, completed, generation, deadline) = state; + if !is_activation_current(*generation) { + return Ok(()); + } + let Ok(snapshot) = fetch_proxies_with_timeout().await else { + return if Instant::now() >= *deadline { + Ok(()) + } else { + Err(RetryError::Retry(())) + }; + }; + if !is_activation_current(*generation) { + return Ok(()); + } + let pending = unsettled_selections(selected, &snapshot); + if pending.is_empty() { + return Ok(()); + } + if Instant::now() >= *deadline { + logging!( + warn, + Type::Config, + "gave up putting back {} selected node(s) the core never loaded: {}", + pending.len(), + pending.iter().map(String::as_str).collect::>().join(", ") + ); + return Ok(()); + } + let plan = reconcile_selected_nodes(selected, None, &snapshot); + if apply_activations(&plan.activations, completed, *generation) + .await + .is_none() + { + return Ok(()); + } + if is_activation_current(*generation) { + announce(Refresh::Clash); + } + Err(RetryError::Retry(())) + }) + }, + ) + .await; } /// Releases the first-pass waiter even when restoration exits early. diff --git a/src-tauri/src/core/manager/mod.rs b/src-tauri/src/core/manager/mod.rs index b20338412..26164816e 100644 --- a/src-tauri/src/core/manager/mod.rs +++ b/src-tauri/src/core/manager/mod.rs @@ -285,48 +285,56 @@ impl CoreManager { anyhow::bail!("core startup blocked after mixed proxy port fallback failure: {reason}"); } - let mut retries = 0; - loop { - match self.start_core().await { - Ok(()) => { - crate::config::Config::notify_startup_mixed_port_fallback(); - return Ok(!matches!(*self.get_running_mode(), RunningMode::NotRunning)); - } - Err(start_error) if retries < MAX_PORT_FALLBACK_RETRIES => { - if !matches!(*self.get_running_mode(), RunningMode::NotRunning) { + use crate::utils::retry::{RetryError, RetryPolicy, retry}; + retry( + RetryPolicy::fixed( + std::num::NonZeroUsize::MIN.saturating_add(MAX_PORT_FALLBACK_RETRIES), + std::time::Duration::ZERO, + ), + |retries| async move { + match self.start_core().await { + Ok(()) => { crate::config::Config::notify_startup_mixed_port_fallback(); - return Err(start_error); + Ok(!matches!(*self.get_running_mode(), RunningMode::NotRunning)) } - match crate::config::Config::resolve_startup_mixed_port().await { - Ok(true) => { - retries += 1; - tracing::Span::current().record("retries", retries); - logging!( - warn, - Type::Core, - "Retrying core startup after mixed proxy port fallback ({}/{}): {start_error:#}", - retries, - MAX_PORT_FALLBACK_RETRIES - ); - } - Ok(false) => { + Err(start_error) if retries < MAX_PORT_FALLBACK_RETRIES => { + if !matches!(*self.get_running_mode(), RunningMode::NotRunning) { crate::config::Config::notify_startup_mixed_port_fallback(); - return Err(start_error); + return Err(RetryError::Stop(start_error)); } - Err(fallback_error) => { - crate::config::Config::block_startup_core(&fallback_error); - return Err(start_error.context(format!( - "the mixed proxy port fallback did not rescue core startup: {fallback_error:#}" - ))); + match crate::config::Config::resolve_startup_mixed_port().await { + Ok(true) => { + let retries = retries + 1; + tracing::Span::current().record("retries", retries); + logging!( + warn, + Type::Core, + "Retrying core startup after mixed proxy port fallback ({}/{}): {start_error:#}", + retries, + MAX_PORT_FALLBACK_RETRIES + ); + Err(RetryError::Retry(start_error)) + } + Ok(false) => { + crate::config::Config::notify_startup_mixed_port_fallback(); + Err(RetryError::Stop(start_error)) + } + Err(fallback_error) => { + crate::config::Config::block_startup_core(&fallback_error); + Err(RetryError::Stop(start_error.context(format!( + "the mixed proxy port fallback did not rescue core startup: {fallback_error:#}" + )))) + } } } + Err(error) => { + crate::config::Config::notify_startup_mixed_port_fallback(); + Err(RetryError::Stop(error)) + } } - Err(error) => { - crate::config::Config::notify_startup_mixed_port_fallback(); - return Err(error); - } - } - } + }, + ) + .await } } diff --git a/src-tauri/src/core/manager/state.rs b/src-tauri/src/core/manager/state.rs index 346485ac7..165f04b6e 100644 --- a/src-tauri/src/core/manager/state.rs +++ b/src-tauri/src/core/manager/state.rs @@ -1,6 +1,7 @@ #[cfg(test)] use super::claim_core_readiness_generation; use super::{CoreManager, PROFILE_SELECTIONS_PENDING_COMMIT, RunningMode}; +use crate::utils::retry::{RetryError, RetryPolicy, retry}; use crate::{ AsyncHandler, config::Config, @@ -11,6 +12,7 @@ use crate::{ use anyhow::{Context as _, Result}; use clash_verge_logging::Type; use log::Level; +use std::num::NonZeroUsize; use std::path::Path; use tauri_plugin_mihomo::MihomoExt as _; use tauri_plugin_shell::ShellExt as _; @@ -41,14 +43,14 @@ async fn retry_service_start( mut start: Start, ) -> Result<()> where - Start: FnMut() -> StartFuture, - StartFuture: std::future::Future>, + Start: FnMut() -> StartFuture + Send, + StartFuture: std::future::Future> + Send, { - let mut last_error = None; - for attempt in 0..attempts { - match start().await { - Ok(()) => return Ok(()), - Err(error) => { + let attempts = NonZeroUsize::new(attempts).ok_or_else(|| anyhow::anyhow!("service start failed"))?; + retry(RetryPolicy::fixed(attempts, retry_delay), |attempt| { + let future = start(); + async move { + future.await.map_err(|error| { logging!( warn, Type::Core, @@ -60,16 +62,14 @@ where .downcast_ref::() .is_some_and(|refusal| service::StageRequest::is_about_the_bundle(refusal.code)) { - return Err(error); + RetryError::Stop(error) + } else { + RetryError::Retry(error) } - last_error = Some(error); - if attempt + 1 < attempts { - tokio::time::sleep(retry_delay).await; - } - } + }) } - } - Err(last_error.unwrap_or_else(|| anyhow::anyhow!("service start failed"))) + }) + .await } impl CoreManager { @@ -95,14 +95,16 @@ async fn poll_sidecar_readiness( mut probe: F, ) -> Result<()> where - F: FnMut() -> Fut, - Fut: std::future::Future>, + F: FnMut() -> Fut + Send, + Fut: std::future::Future> + Send, { - let mut last_error = None; - for attempt in 0..max_attempts { - match probe().await { - Ok(()) => return Ok(()), - Err(error) => { + let attempts = NonZeroUsize::new(max_attempts) + .context("sidecar readiness was configured with no attempts") + .context("Mihomo API did not become ready")?; + retry(RetryPolicy::fixed(attempts, retry_delay), |attempt| { + let future = probe(); + async move { + future.await.map_err(|error| { logging!( debug, Type::Core, @@ -110,16 +112,12 @@ where attempt + 1, max_attempts ); - last_error = Some(error); - } + RetryError::Retry(error) + }) } - if attempt + 1 < max_attempts { - tokio::time::sleep(retry_delay).await; - } - } - Err(last_error - .unwrap_or_else(|| anyhow::anyhow!("sidecar readiness was configured with no attempts")) - .context("Mihomo API did not become ready")) + }) + .await + .context("Mihomo API did not become ready") } fn should_clear_terminated_sidecar(running_mode: &RunningMode, current_pid: Option, terminated_pid: u32) -> bool { diff --git a/src-tauri/src/core/proxy_control.rs b/src-tauri/src/core/proxy_control.rs index 0ec2949c8..e9f482247 100644 --- a/src-tauri/src/core/proxy_control.rs +++ b/src-tauri/src/core/proxy_control.rs @@ -522,25 +522,34 @@ pub async fn clear() -> Result<()> { /// A failed system call is usually a transient RPC hiccup, so give it a few tries. async fn clear_with_retry() -> Result<()> { - for attempt in 1..CLEAR_ATTEMPTS { - match clear_inner().await { - Err(error) + use crate::utils::retry::{RetryError, RetryPolicy, retry}; + retry( + RetryPolicy::fixed( + std::num::NonZeroUsize::MIN.saturating_add((CLEAR_ATTEMPTS - 1) as usize), + CLEAR_RETRY_DELAY, + ), + |attempt| async move { + clear_inner().await.map_err(|error| { if matches!( SysproxyFailure::from_chain(&error), Some(SysproxyFailure::SystemCallFailed) - ) => - { - logging!( - warn, - Type::Core, - "clearing the system proxy failed (attempt {attempt}/{CLEAR_ATTEMPTS}); retrying: {error:#}" - ); - tokio::time::sleep(CLEAR_RETRY_DELAY).await; - } - other => return other, - } - } - clear_inner().await + ) { + if attempt + 1 < CLEAR_ATTEMPTS as usize { + logging!( + warn, + Type::Core, + "clearing the system proxy failed (attempt {}/{CLEAR_ATTEMPTS}); retrying: {error:#}", + attempt + 1 + ); + } + RetryError::Retry(error) + } else { + RetryError::Stop(error) + } + }) + }, + ) + .await } #[tracing::instrument(skip_all, level = "info", fields(route = tracing::field::Empty))] diff --git a/src-tauri/src/core/runstate/mod.rs b/src-tauri/src/core/runstate/mod.rs index e06979ed5..e906256e5 100644 --- a/src-tauri/src/core/runstate/mod.rs +++ b/src-tauri/src/core/runstate/mod.rs @@ -130,20 +130,21 @@ impl RunStateStore { /// Retries transport failures but stops at the first readable reply, including a rejection. pub async fn await_ready(&self, attempts: usize, interval: Duration) -> Result { - let mut last_error = None; - for attempt in 0..attempts { + use crate::utils::retry::{RetryError, RetryPolicy, retry}; + let attempts = std::num::NonZeroUsize::new(attempts).ok_or_else(|| { + ReadyWaitError::Unreachable(anyhow::anyhow!( + "service readiness wait was configured with no attempts" + )) + })?; + retry(RetryPolicy::fixed(attempts, interval), |_| async { match self.env.probe_service_version().await { - Ok(reply) => return self.record_reply(&reply).map_err(ReadyWaitError::Rejected), - Err(error) => last_error = Some(error), + Ok(reply) => self + .record_reply(&reply) + .map_err(|error| RetryError::Stop(ReadyWaitError::Rejected(error))), + Err(error) => Err(RetryError::Retry(ReadyWaitError::Unreachable(error))), } - if attempt + 1 < attempts { - tokio::time::sleep(interval).await; - } - } - - Err(ReadyWaitError::Unreachable(last_error.unwrap_or_else(|| { - anyhow::anyhow!("service readiness wait was configured with no attempts") - }))) + }) + .await } fn record_reply(&self, reply: &ServiceVersionReply) -> Result { diff --git a/src-tauri/src/core/service.rs b/src-tauri/src/core/service.rs index 3f0a8d25a..498d087eb 100644 --- a/src-tauri/src/core/service.rs +++ b/src-tauri/src/core/service.rs @@ -315,35 +315,40 @@ static SERVICE_CORE_STAGING_GENERATION: AtomicU64 = AtomicU64::new(0); #[cfg(any(all(target_os = "macos", feature = "verge-dev"), test))] fn create_service_core_staging_file(directory: &Path, core_name: &std::ffi::OsStr) -> Result<(PathBuf, std::fs::File)> { - for _ in 0..32 { - let generation = SERVICE_CORE_STAGING_GENERATION.fetch_add(1, Ordering::Relaxed); - let temporary_name = format!( - ".{}.{}.{generation}.tmp", - core_name.to_string_lossy(), - std::process::id() - ); - let temporary_path = directory.join(temporary_name); - match std::fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(&temporary_path) - { - Ok(file) => return Ok((temporary_path, file)), - Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {} - Err(error) => { - return Err(error).with_context(|| { + use crate::utils::retry::{RetryError, RetryPolicy, retry_sync}; + retry_sync( + RetryPolicy::fixed( + std::num::NonZeroUsize::MIN.saturating_add(31), + std::time::Duration::ZERO, + ), + |_| { + let generation = SERVICE_CORE_STAGING_GENERATION.fetch_add(1, Ordering::Relaxed); + let temporary_name = format!( + ".{}.{}.{generation}.tmp", + core_name.to_string_lossy(), + std::process::id() + ); + let temporary_path = directory.join(temporary_name); + match std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&temporary_path) + { + Ok(file) => Ok((temporary_path, file)), + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + Err(RetryError::Retry(anyhow::anyhow!( + "failed to create a unique temporary development Service core in {}", + directory.display() + ))) + } + Err(error) => Err(RetryError::Stop(anyhow::Error::from(error).context({ format!( "failed to create temporary development Service core {}", temporary_path.display() ) - }); + }))), } - } - } - - bail!( - "failed to create a unique temporary development Service core in {}", - directory.display() + }, ) } @@ -1014,38 +1019,47 @@ pub(crate) fn request_runtime_provider_sync(delay: Duration) { } AsyncHandler::spawn(move || async move { tokio::time::sleep(delay).await; - for attempt in 1..=RUNTIME_PROVIDER_SYNC_ATTEMPTS { - let outcome = { - let _serial = PROVIDER_SYNC_SERIAL.lock().await; - if attempt == 1 { - PROVIDER_SYNC_QUEUED.store(false, Ordering::Release); + use crate::utils::retry::{RetryError, RetryPolicy, retry}; + let _ = retry( + RetryPolicy::fixed( + std::num::NonZeroUsize::MIN.saturating_add((RUNTIME_PROVIDER_SYNC_ATTEMPTS - 1) as usize), + constants::timing::RUNTIME_PROVIDER_SYNC_RETRY_DELAY, + ), + |index| async move { + let attempt = index as u32 + 1; + let outcome = { + let _serial = PROVIDER_SYNC_SERIAL.lock().await; + if attempt == 1 { + PROVIDER_SYNC_QUEUED.store(false, Ordering::Release); + } + sync_runtime_providers_by_service().await + }; + match outcome { + Ok(ProviderSync { pending: 0, .. }) => Ok(()), + Ok(ProviderSync { pending, .. }) if attempt < RUNTIME_PROVIDER_SYNC_ATTEMPTS => { + logging!( + info, + Type::Service, + "{pending} provider caches are not ready yet; retrying" + ); + Err(RetryError::Retry(())) + } + Ok(ProviderSync { pending, .. }) => { + logging!(warn, Type::Service, "{pending} provider caches were not synced"); + Ok(()) + } + Err(error) => { + logging!( + warn, + Type::Service, + "failed to sync provider caches from the service: {error:#}" + ); + Ok(()) + } } - sync_runtime_providers_by_service().await - }; - match outcome { - Ok(ProviderSync { pending: 0, .. }) => return, - Ok(ProviderSync { pending, .. }) if attempt < RUNTIME_PROVIDER_SYNC_ATTEMPTS => { - logging!( - info, - Type::Service, - "{pending} provider caches are not ready yet; retrying" - ); - } - Ok(ProviderSync { pending, .. }) => { - logging!(warn, Type::Service, "{pending} provider caches were not synced"); - return; - } - Err(error) => { - logging!( - warn, - Type::Service, - "failed to sync provider caches from the service: {error:#}" - ); - return; - } - } - tokio::time::sleep(constants::timing::RUNTIME_PROVIDER_SYNC_RETRY_DELAY).await; - } + }, + ) + .await; }); } @@ -1353,21 +1367,29 @@ async fn create_sync_temp(target: &Path) -> Result<(PathBuf, tokio::fs::File)> { .file_name() .map(|name| name.to_string_lossy()) .unwrap_or_default(); - for _ in 0..8 { - let sequence = SYNC_TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed); - let temp = target.with_file_name(format!(".{name}.sync-{}-{sequence}.tmp", std::process::id())); - match tokio::fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(&temp) - .await - { - Ok(file) => return Ok((temp, file)), - Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {} - Err(error) => return Err(error).with_context(|| format!("failed to create {}", temp.display())), - } - } - bail!("no free temporary name beside {}", target.display()) + use crate::utils::retry::{RetryError, RetryPolicy, retry}; + retry( + RetryPolicy::fixed(std::num::NonZeroUsize::MIN.saturating_add(7), std::time::Duration::ZERO), + |_| async { + let sequence = SYNC_TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed); + let temp = target.with_file_name(format!(".{name}.sync-{}-{sequence}.tmp", std::process::id())); + match tokio::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&temp) + .await + { + Ok(file) => Ok((temp, file)), + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Err(RetryError::Retry( + anyhow::anyhow!("no free temporary name beside {}", target.display()), + )), + Err(error) => Err(RetryError::Stop( + anyhow::anyhow!(error).context(format!("failed to create {}", temp.display())), + )), + } + }, + ) + .await } async fn same_contents(temp: &Path, target: &Path) -> bool { @@ -1734,22 +1756,28 @@ async fn recover_after_owner_loss_while_locked(reason: OwnerRecoveryReason) { } async fn clear_proxy_after_owner_loss() { - let mut last_error = None; - for attempt in 1..=3 { - match proxy_control::clear().await { - Ok(()) => return, - Err(error) => { + use crate::utils::retry::{RetryError, RetryPolicy, retry}; + let result = retry( + RetryPolicy::fixed( + std::num::NonZeroUsize::MIN.saturating_add(2), + Duration::from_millis(100), + ), + |index| async move { + let attempt = index + 1; + proxy_control::clear().await.map_err(|error| { logging!( warn, Type::Service, "proxy clear attempt {attempt}/3 after owner loss failed: {error:#}" ); - last_error = Some(error); - tokio::time::sleep(Duration::from_millis(100)).await; - } - } - } - if let Some(error) = last_error { + RetryError::Retry(error) + }) + }, + ) + .await; + if let Err(error) = result { + // The existing recovery path also waits after its final failed clear. + tokio::time::sleep(Duration::from_millis(100)).await; logging!( error, Type::Service, diff --git a/src-tauri/src/core/updater.rs b/src-tauri/src/core/updater.rs index f90b72418..c748c44d0 100644 --- a/src-tauri/src/core/updater.rs +++ b/src-tauri/src/core/updater.rs @@ -388,12 +388,14 @@ impl SilentUpdater { // The new webview may not accept evaluation immediately. std::thread::spawn(move || { - for i in 0..10 { - std::thread::sleep(std::time::Duration::from_millis(100 * (i + 1))); - if window.eval(&js).is_ok() { - return; - } - } + use crate::utils::retry::{RetryError, RetryPolicy, retry_sync}; + let _ = retry_sync( + RetryPolicy::fixed(std::num::NonZeroUsize::MIN.saturating_add(9), std::time::Duration::ZERO), + |index| { + std::thread::sleep(std::time::Duration::from_millis(100 * (index as u64 + 1))); + window.eval(&js).map_err(RetryError::Retry) + }, + ); }); logging!(info, Type::System, "Update splash window shown"); diff --git a/src-tauri/src/feat/core_upgrade.rs b/src-tauri/src/feat/core_upgrade.rs index b139c0a22..805dbdb59 100644 --- a/src-tauri/src/feat/core_upgrade.rs +++ b/src-tauri/src/feat/core_upgrade.rs @@ -359,24 +359,31 @@ fn stage_core(target: &Path, package: &[u8], version: &str) -> Result Result<(PathBuf, File)> { - for _ in 0..32 { - let generation = STAGING_GENERATION.fetch_add(1, Ordering::Relaxed); - let path = directory.join(format!( - ".{}.{}.{generation}.tmp", - core_name.to_string_lossy(), - std::process::id() - )); + use crate::utils::retry::{RetryError, RetryPolicy, retry_sync}; + retry_sync( + RetryPolicy::fixed( + std::num::NonZeroUsize::MIN.saturating_add(31), + std::time::Duration::ZERO, + ), + |_| { + let generation = STAGING_GENERATION.fetch_add(1, Ordering::Relaxed); + let path = directory.join(format!( + ".{}.{}.{generation}.tmp", + core_name.to_string_lossy(), + std::process::id() + )); - match File::options().write(true).create_new(true).open(&path) { - Ok(file) => return Ok((path, file)), - Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {} - Err(error) => { - return Err(error).with_context(|| format!("failed to create {}", path.display())); + match File::options().write(true).create_new(true).open(&path) { + Ok(file) => Ok((path, file)), + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Err(RetryError::Retry( + anyhow::anyhow!("failed to create a unique staging file in {}", directory.display()), + )), + Err(error) => Err(RetryError::Stop( + anyhow::Error::from(error).context(format!("failed to create {}", path.display())), + )), } - } - } - - bail!("failed to create a unique staging file in {}", directory.display()) + }, + ) } #[cfg(windows)] diff --git a/src-tauri/src/utils/macos_launch_guard.rs b/src-tauri/src/utils/macos_launch_guard.rs index 1fa5869df..863136aa8 100644 --- a/src-tauri/src/utils/macos_launch_guard.rs +++ b/src-tauri/src/utils/macos_launch_guard.rs @@ -109,44 +109,61 @@ fn current_user_home() -> anyhow::Result { } .max(1024); - loop { - if buffer_size > MAX_BUFFER_SIZE { - anyhow::bail!("effective-user home lookup exceeded the maximum buffer size"); - } + use crate::utils::retry::{RetryError, RetryPolicy, retry_sync}; + retry_sync( + RetryPolicy::fixed( + std::num::NonZeroUsize::MIN.saturating_add(11), + std::time::Duration::ZERO, + ), + |_| { + if buffer_size > MAX_BUFFER_SIZE { + return Err(RetryError::Stop(anyhow::anyhow!( + "effective-user home lookup exceeded the maximum buffer size" + ))); + } - let mut passwd = std::mem::MaybeUninit::::zeroed(); - let mut result = std::ptr::null_mut(); - let mut buffer = vec![0_u8; buffer_size]; - let code = unsafe { - libc::getpwuid_r( - uid, - passwd.as_mut_ptr(), - buffer.as_mut_ptr().cast(), - buffer.len(), - &mut result, - ) - }; - if code == libc::ERANGE { - buffer_size = buffer_size.saturating_mul(2); - continue; - } - if code != 0 { - return Err(std::io::Error::from_raw_os_error(code).into()); - } - if result.is_null() { - anyhow::bail!("effective user {uid} has no password database entry"); - } + let mut passwd = std::mem::MaybeUninit::::zeroed(); + let mut result = std::ptr::null_mut(); + let mut buffer = vec![0_u8; buffer_size]; + let code = unsafe { + libc::getpwuid_r( + uid, + passwd.as_mut_ptr(), + buffer.as_mut_ptr().cast(), + buffer.len(), + &mut result, + ) + }; + if code == libc::ERANGE { + buffer_size = buffer_size.saturating_mul(2); + return Err(RetryError::Retry(anyhow::anyhow!( + "effective-user home lookup exceeded the maximum buffer size" + ))); + } + if code != 0 { + return Err(RetryError::Stop(std::io::Error::from_raw_os_error(code).into())); + } + if result.is_null() { + return Err(RetryError::Stop(anyhow::anyhow!( + "effective user {uid} has no password database entry" + ))); + } - let passwd = unsafe { passwd.assume_init() }; - if passwd.pw_dir.is_null() { - anyhow::bail!("effective user {uid} has no home directory"); - } - let home = unsafe { CStr::from_ptr(passwd.pw_dir) }; - if home.to_bytes().is_empty() { - anyhow::bail!("effective user {uid} has an empty home directory"); - } - return Ok(PathBuf::from(std::ffi::OsStr::from_bytes(home.to_bytes()))); - } + let passwd = unsafe { passwd.assume_init() }; + if passwd.pw_dir.is_null() { + return Err(RetryError::Stop(anyhow::anyhow!( + "effective user {uid} has no home directory" + ))); + } + let home = unsafe { CStr::from_ptr(passwd.pw_dir) }; + if home.to_bytes().is_empty() { + return Err(RetryError::Stop(anyhow::anyhow!( + "effective user {uid} has an empty home directory" + ))); + } + Ok(PathBuf::from(std::ffi::OsStr::from_bytes(home.to_bytes()))) + }, + ) } pub const fn move_decision( diff --git a/src-tauri/src/utils/retry.rs b/src-tauri/src/utils/retry.rs index 502cb70f0..b34373d86 100644 --- a/src-tauri/src/utils/retry.rs +++ b/src-tauri/src/utils/retry.rs @@ -19,6 +19,14 @@ pub struct RetryPolicy { } impl RetryPolicy { + pub const fn fixed(attempts: NonZeroUsize, delay: Duration) -> Self { + Self { + attempts, + backoff: Backoff::Fixed(delay), + jitter: false, + } + } + fn delays(self) -> ExponentialBackoff { let (base, cap, factor) = match self.backoff { Backoff::Fixed(delay) => (delay, delay, 1.0), @@ -38,7 +46,6 @@ impl RetryPolicy { pub enum RetryError { Retry(E), - #[allow(dead_code)] Stop(E), } @@ -89,6 +96,49 @@ pub async fn try_strategies< .await } +pub async fn retry_with_state( + policy: RetryPolicy, + mut state: S, + mut operation: impl for<'a> FnMut(&'a mut S, usize) -> futures::future::BoxFuture<'a, Result>> + Send, +) -> Result { + let mut result = operation(&mut state, 0).await; + for (index, delay) in policy.delays().enumerate() { + match result { + Ok(_) | Err(RetryError::Stop(_)) => break, + Err(RetryError::Retry(_)) => { + if !delay.is_zero() { + tokio::time::sleep(delay).await; + } + } + } + result = operation(&mut state, index + 1).await; + } + result.map_err(|error| match error { + RetryError::Retry(error) | RetryError::Stop(error) => error, + }) +} + +pub fn retry_sync( + policy: RetryPolicy, + mut operation: impl FnMut(usize) -> Result>, +) -> Result { + let mut result = operation(0); + for (index, delay) in policy.delays().enumerate() { + match result { + Ok(_) | Err(RetryError::Stop(_)) => break, + Err(RetryError::Retry(_)) => { + if !delay.is_zero() { + std::thread::sleep(delay); + } + } + } + result = operation(index + 1); + } + result.map_err(|error| match error { + RetryError::Retry(error) | RetryError::Stop(error) => error, + }) +} + #[cfg(test)] mod tests { use super::*; @@ -141,6 +191,34 @@ mod tests { assert_eq!(visited, ["direct", "clash", "system"]); } + #[tokio::test] + async fn stateful_and_sync_retries_preserve_progress_and_stop() { + let policy = RetryPolicy::fixed(NonZeroUsize::MIN.saturating_add(2), Duration::ZERO); + let result = retry_with_state(policy, Vec::new(), |visited, attempt| { + Box::pin(async move { + visited.push(attempt); + if visited.len() == 3 { + Ok(visited.clone()) + } else { + Err(RetryError::Retry(())) + } + }) + }) + .await; + assert_eq!(result, Ok(vec![0, 1, 2])); + let mut visited = Vec::new(); + let result = retry_sync(policy, |attempt| { + visited.push(attempt); + if attempt == 1 { + Err::<(), _>(RetryError::Stop("terminal")) + } else { + Err(RetryError::Retry("retry")) + } + }); + assert_eq!(result, Err("terminal")); + assert_eq!(visited, [0, 1]); + } + #[test] fn backoff_keeps_the_attempt_budget_and_exponential_cap() { let policy = RetryPolicy { diff --git a/src-tauri/src/utils/server.rs b/src-tauri/src/utils/server.rs index 40d87b769..5175a02c5 100644 --- a/src-tauri/src/utils/server.rs +++ b/src-tauri/src/utils/server.rs @@ -54,22 +54,34 @@ pub async fn check_singleton() -> Result { .context("failed to initialize singleton lock")?; if !try_lock_instance(&lock).context("failed to acquire singleton lock")? { let deadline = std::time::Instant::now() + Duration::from_secs(20); - loop { - if std::time::Instant::now() >= deadline { - bail!("another app instance is starting"); - } - // A restarting instance releases this lock only after spawning this - // process, and its embedded server is already down, so waiting on - // the notify probe alone can never promote this instance. - if try_lock_instance(&lock).context("failed to acquire singleton lock")? { - break; - } - if let Ok(record) = read_instance_record(&record_path) - && notify_existing_instance(&record).await - { - return Ok(SingletonDisposition::Secondary); - } - tokio::time::sleep(Duration::from_millis(50)).await; + use crate::utils::retry::{RetryError, RetryPolicy, retry}; + let secondary = retry( + RetryPolicy::fixed( + std::num::NonZeroUsize::MIN.saturating_add(400), + Duration::from_millis(50), + ), + |_| async { + if std::time::Instant::now() >= deadline { + return Err(RetryError::Stop(anyhow::anyhow!("another app instance is starting"))); + } + // A restarting primary releases the lock after spawning us, with its server down. + if try_lock_instance(&lock) + .context("failed to acquire singleton lock") + .map_err(RetryError::Stop)? + { + return Ok(false); + } + if let Ok(record) = read_instance_record(&record_path) + && notify_existing_instance(&record).await + { + return Ok(true); + } + Err(RetryError::Retry(anyhow::anyhow!("another app instance is starting"))) + }, + ) + .await?; + if secondary { + return Ok(SingletonDisposition::Secondary); } }