gitea-admin/clash-verge-rev
1
0
Fork 0
mirror of https://github.com/clash-verge-rev/clash-verge-rev.git synced 2026-10-30 10:42:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(utils): migrate remaining bounded retry loops

Browse Source
This commit is contained in:
Tunglies
2026-10-01 14:34:16 +08:00
committed by GitHub
parent 9d1b400299
commit 5375ce360f
11 changed files with 478 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/config/profiles.rs
+68 -54
View File
@@ -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<Proxies> {
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<Stri
/// Retries selections while provider-backed groups finish loading.
async fn settle_pending_selections(selected: &[PrfSelected], completed: &mut HashMap<String, String>, 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::<Vec<_>>().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::<Vec<_>>().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.
src-tauri/src/core/manager/mod.rs
+43 -35
View File
@@ -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
}
}
src-tauri/src/core/manager/state.rs
+29 -31
View File
@@ -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<Start, StartFuture>(
mut start: Start,
) -> Result<()>
where
Start: FnMut() -> StartFuture,
StartFuture: std::future::Future<Output = Result<()>>,
Start: FnMut() -> StartFuture + Send,
StartFuture: std::future::Future<Output = Result<()>> + 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::<service::ServiceStartRefusal>()
.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<F, Fut>(
mut probe: F,
) -> Result<()>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<()>>,
F: FnMut() -> Fut + Send,
Fut: std::future::Future<Output = Result<()>> + 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<u32>, terminated_pid: u32) -> bool {
src-tauri/src/core/proxy_control.rs
+25 -16
View File
@@ -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))]
src-tauri/src/core/runstate/mod.rs
+13 -12
View File
@@ -130,20 +130,21 @@ impl<E: RunStateEnv> RunStateStore<E> {
/// Retries transport failures but stops at the first readable reply, including a rejection.
pub async fn await_ready(&self, attempts: usize, interval: Duration) -> Result<RunState, ReadyWaitError> {
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<RunState> {
src-tauri/src/core/service.rs
+109 -81
View File
@@ -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,
src-tauri/src/core/updater.rs
+8 -6
View File
@@ -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");
src-tauri/src/feat/core_upgrade.rs
+23 -16
View File
@@ -359,24 +359,31 @@ fn stage_core(target: &Path, package: &[u8], version: &str) -> Result<StagedCore
}
fn create_staging_file(directory: &Path, core_name: &OsStr) -> 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)]
src-tauri/src/utils/macos_launch_guard.rs
+53 -36
View File
@@ -109,44 +109,61 @@ fn current_user_home() -> anyhow::Result<PathBuf> {
}
.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::<libc::passwd>::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::<libc::passwd>::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(
src-tauri/src/utils/retry.rs
+79 -1
View File
@@ -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<E> {
Retry(E),
#[allow(dead_code)]
Stop(E),
}
@@ -89,6 +96,49 @@ pub async fn try_strategies<
.await
}
pub async fn retry_with_state<S: Send, T: Send, E: Send>(
policy: RetryPolicy,
mut state: S,
mut operation: impl for<'a> FnMut(&'a mut S, usize) -> futures::future::BoxFuture<'a, Result<T, RetryError<E>>> + Send,
) -> Result<T, E> {
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<T, E>(
policy: RetryPolicy,
mut operation: impl FnMut(usize) -> Result<T, RetryError<E>>,
) -> Result<T, E> {
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 {
src-tauri/src/utils/server.rs
+28 -16
View File
@@ -54,22 +54,34 @@ pub async fn check_singleton() -> Result<SingletonDisposition> {
.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);
}
}