From bf95d715a478cd4a4e53cb5ea00dd2d749a666cc Mon Sep 17 00:00:00 2001 From: Disy Date: Tue, 20 Oct 2026 21:25:19 +0800 Subject: [PATCH] fix: make hosted downloads controllable and failures diagnosable --- .../helpers/hosted-download-controls.test.ts | 96 +++++ .../src/helpers/hosted-download-controls.ts | 45 +++ .../helpers/hosted-download-failures.test.ts | 21 ++ apps/app-frontend/src/helpers/hosted-packs.ts | 6 + apps/app-frontend/src/helpers/state.ts | 10 +- .../app-frontend/src/locales/en-US/index.json | 4 + .../app-frontend/src/locales/zh-CN/index.json | 4 + apps/app-frontend/src/pages/Downloads.vue | 84 ++++- apps/app/build.rs | 3 + apps/app/src/api/install.rs | 18 + packages/app-lib/src/api/pack/hosted.rs | 52 ++- .../app-lib/src/api/pack/hosted/control.rs | 329 ++++++++++++++++++ .../app-lib/src/api/pack/hosted/progress.rs | 3 + .../app-lib/src/api/pack/hosted/tagged.rs | 11 +- packages/app-lib/src/event/emit.rs | 81 ++++- packages/app-lib/src/event/mod.rs | 6 + packages/app-lib/src/util/fetch.rs | 147 +++++++- 17 files changed, 848 insertions(+), 72 deletions(-) create mode 100644 apps/app-frontend/src/helpers/hosted-download-controls.test.ts create mode 100644 apps/app-frontend/src/helpers/hosted-download-controls.ts create mode 100644 packages/app-lib/src/api/pack/hosted/control.rs diff --git a/apps/app-frontend/src/helpers/hosted-download-controls.test.ts b/apps/app-frontend/src/helpers/hosted-download-controls.test.ts new file mode 100644 index 0000000..9271252 --- /dev/null +++ b/apps/app-frontend/src/helpers/hosted-download-controls.test.ts @@ -0,0 +1,96 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import { + hostedDownloadActions, + hostedDownloadState, + runHostedDownloadAction, + runHostedDownloadControl, +} from './hosted-download-controls.ts' + +test('only a controllable parent hosted task offers pause, resume and stop', () => { + const bar = { + bar_type: { + type: 'hosted_pack_sync', + instance_id: 'one', + state: 'running' as const, + can_pause: true, + can_cancel: true, + }, + } + assert.deepEqual(hostedDownloadActions(bar), ['pause', 'cancel']) + assert.deepEqual(hostedDownloadActions({ bar_type: { ...bar.bar_type, state: 'paused' } }), [ + 'resume', + 'cancel', + ]) + for (const state of ['finalizing', 'canceling'] as const) { + assert.deepEqual(hostedDownloadActions({ bar_type: { ...bar.bar_type, state } }), []) + } + assert.deepEqual( + hostedDownloadActions({ bar_type: { ...bar.bar_type, type: 'hosted_mod_download' } }), + [], + ) + assert.deepEqual( + hostedDownloadActions({ bar_type: { type: 'hosted_pack_sync', instance_id: 'one' } }), + [], + ) +}) + +test('failed and canceled hosted tasks offer direct retry, preserving terminal state', () => { + for (const state of ['failed', 'canceled'] as const) { + const bar = { + bar_type: { type: 'hosted_pack_sync', instance_id: 'one', state, error: 'failure' }, + } + assert.deepEqual(hostedDownloadActions(bar), ['retry']) + assert.equal(hostedDownloadState(bar), state) + } + assert.equal(hostedDownloadState({ bar_type: { error: 'old backend failure' } }), 'failed') +}) + +test('resume refreshes the session before invoking controls; pause and stop need no session', async () => { + const calls: string[] = [] + const deps = { + prepareSession: async () => { + calls.push('session') + }, + invoke: async (command: string, args: { instanceId: string }) => { + calls.push(`${command}:${args.instanceId}`) + }, + } + await runHostedDownloadControl('resume', 'one', deps) + await runHostedDownloadControl('pause', 'one', deps) + await runHostedDownloadControl('cancel', 'one', deps) + assert.deepEqual(calls, [ + 'session', + 'plugin:install|hosted_resume:one', + 'plugin:install|hosted_pause:one', + 'plugin:install|hosted_cancel:one', + ]) +}) + +test('session refresh failures prevent resume', async () => { + await assert.rejects( + runHostedDownloadControl('resume', 'one', { + prepareSession: async () => { + throw new Error('signed out') + }, + invoke: async () => { + assert.fail('resume must not run') + }, + }), + /signed out/, + ) +}) + +test('retry immediately restarts the existing hosted instance through the manager', async () => { + const calls: string[] = [] + await runHostedDownloadAction('retry', 'existing-instance', { + retryHosted: async (instanceId) => { + calls.push(instanceId) + }, + control: async () => { + assert.fail('retry must use the full hosted transaction') + }, + }) + assert.deepEqual(calls, ['existing-instance']) +}) diff --git a/apps/app-frontend/src/helpers/hosted-download-controls.ts b/apps/app-frontend/src/helpers/hosted-download-controls.ts new file mode 100644 index 0000000..af80619 --- /dev/null +++ b/apps/app-frontend/src/helpers/hosted-download-controls.ts @@ -0,0 +1,45 @@ +import type { LoadingBar } from './state.ts' + +export type HostedDownloadControl = 'pause' | 'resume' | 'cancel' +export type HostedDownloadAction = HostedDownloadControl | 'retry' + +export function runHostedDownloadAction( + action: HostedDownloadAction, + instanceId: string, + operations: { + retryHosted: (instanceId: string) => Promise + control: (action: HostedDownloadControl, instanceId: string) => Promise + }, +) { + return action === 'retry' + ? operations.retryHosted(instanceId) + : operations.control(action, instanceId) +} + +export function hostedDownloadState(bar: LoadingBar) { + return bar.bar_type?.state ?? (bar.bar_type?.error ? 'failed' : 'running') +} + +export function hostedDownloadActions(bar: LoadingBar): HostedDownloadAction[] { + const type = bar.bar_type + if (type?.type !== 'hosted_pack_sync' || !type.instance_id) return [] + const state = hostedDownloadState(bar) + if (state === 'failed' || state === 'canceled') return ['retry'] + if (state !== 'running' && state !== 'paused') return [] + const actions: HostedDownloadAction[] = [] + if (type.can_pause) actions.push(state === 'paused' ? 'resume' : 'pause') + if (type.can_cancel) actions.push('cancel') + return actions +} + +export async function runHostedDownloadControl( + action: HostedDownloadControl, + instanceId: string, + dependencies: { + prepareSession: () => Promise + invoke: (command: string, args: { instanceId: string }) => Promise + }, +) { + if (action === 'resume') await dependencies.prepareSession() + await dependencies.invoke(`plugin:install|hosted_${action}`, { instanceId }) +} diff --git a/apps/app-frontend/src/helpers/hosted-download-failures.test.ts b/apps/app-frontend/src/helpers/hosted-download-failures.test.ts index 5c1b4fb..7055141 100644 --- a/apps/app-frontend/src/helpers/hosted-download-failures.test.ts +++ b/apps/app-frontend/src/helpers/hosted-download-failures.test.ts @@ -3,6 +3,27 @@ import test from 'node:test' import { createHostedDownloadFailures } from './hosted-download-failures.ts' +test('stopped hosted downloads retain their state and instance for direct retry', () => { + const state = createHostedDownloadFailures() + state.update({ + fraction: null, + loader_uuid: 'stopped', + event: { + type: 'hosted_pack_sync', + instance_id: 'existing-instance', + state: 'canceled', + error: 'Stopped by user', + can_pause: false, + can_cancel: false, + }, + }) + assert.equal(state.values()[0]?.bar_type?.state, 'canceled') + assert.equal(state.values()[0]?.bar_type?.instance_id, 'existing-instance') + state.begin('existing-instance', state.values()) + assert.equal(state.has('existing-instance'), false) + assert.equal(state.isRetired('stopped'), true) +}) + test('retry clears only its own failure before progress arrives and ignores late old events', () => { const state = createHostedDownloadFailures() const failed = (instanceId: string, loader: string, error: string) => ({ diff --git a/apps/app-frontend/src/helpers/hosted-packs.ts b/apps/app-frontend/src/helpers/hosted-packs.ts index 9904d00..5aa937a 100644 --- a/apps/app-frontend/src/helpers/hosted-packs.ts +++ b/apps/app-frontend/src/helpers/hosted-packs.ts @@ -8,6 +8,7 @@ import { skinSiteUser, waitForSkinSiteSession, } from '../composables/skin-site-session.ts' +import { type HostedDownloadControl, runHostedDownloadControl } from './hosted-download-controls.ts' import { authenticateInstancePlayer } from './instance-player.ts' export const HOSTED_ACCOUNT_REQUIRED_MESSAGE = '您需要登录正版或皮肤站账号后才能操作' @@ -146,6 +147,11 @@ export const hostedCreate = async (gameDirRoot?: string | null) => { } export const hostedBinding = (instanceId: string) => invokeHosted('plugin:install|hosted_binding', { instanceId }) +export const controlHostedDownload = (action: HostedDownloadControl, instanceId: string) => + runHostedDownloadControl(action, instanceId, { + prepareSession: prepareHostedSession, + invoke: invokeHosted, + }) export const hostedSync = async (instanceId: string) => { await requireHostedGameAccount() startHostedPackAttempt(instanceId) diff --git a/apps/app-frontend/src/helpers/state.ts b/apps/app-frontend/src/helpers/state.ts index 011fd72..ba14c3b 100644 --- a/apps/app-frontend/src/helpers/state.ts +++ b/apps/app-frontend/src/helpers/state.ts @@ -9,6 +9,9 @@ export interface LoadingBarType { batch_id?: string file_name?: string error?: string | null + state?: 'running' | 'paused' | 'canceling' | 'canceled' | 'failed' | 'finalizing' + can_pause?: boolean + can_cancel?: boolean type?: string version?: string instance_id?: string @@ -28,7 +31,12 @@ export interface LoadingBar { } export type OpeningCommandEvent = - 'RunMRPack' | 'InstallServer' | 'InstallVersion' | 'InstallMod' | 'InstallModpack' | string + | 'RunMRPack' + | 'InstallServer' + | 'InstallVersion' + | 'InstallMod' + | 'InstallModpack' + | string export interface OpeningCommand { event: OpeningCommandEvent diff --git a/apps/app-frontend/src/locales/en-US/index.json b/apps/app-frontend/src/locales/en-US/index.json index 13e7ee9..30685a3 100644 --- a/apps/app-frontend/src/locales/en-US/index.json +++ b/apps/app-frontend/src/locales/en-US/index.json @@ -1,4 +1,8 @@ { + "app.hosted-packs.progress.pause": { "message": "Pause" }, + "app.hosted-packs.progress.resume": { "message": "Resume" }, + "app.hosted-packs.progress.stop": { "message": "Stop" }, + "app.downloads.status.paused": { "message": "Paused" }, "app.skin-site.captcha.loading": { "message": "Loading captcha…" }, "app.skin-site.captcha.submit": { "message": "Submit" }, "app.skin-site.captcha.subtitle": { diff --git a/apps/app-frontend/src/locales/zh-CN/index.json b/apps/app-frontend/src/locales/zh-CN/index.json index 0ea5f84..525348e 100644 --- a/apps/app-frontend/src/locales/zh-CN/index.json +++ b/apps/app-frontend/src/locales/zh-CN/index.json @@ -1,4 +1,8 @@ { + "app.hosted-packs.progress.pause": { "message": "暂停" }, + "app.hosted-packs.progress.resume": { "message": "继续" }, + "app.hosted-packs.progress.stop": { "message": "停止" }, + "app.downloads.status.paused": { "message": "已暂停" }, "app.skin-site.captcha.loading": { "message": "正在加载验证码…" }, "app.skin-site.captcha.submit": { "message": "验证" }, "app.skin-site.captcha.subtitle": { diff --git a/apps/app-frontend/src/pages/Downloads.vue b/apps/app-frontend/src/pages/Downloads.vue index dd58120..d5541c9 100644 --- a/apps/app-frontend/src/pages/Downloads.vue +++ b/apps/app-frontend/src/pages/Downloads.vue @@ -84,7 +84,7 @@ - - {{ formatMessage(messages.hostedRetry) }} - + + + + + {{ formatMessage(messages.openInstance) }} + + focusedDownloadJobId(route.query.job)) const focusState = ref(createDownloadFocusState(focusedJobId.value)) const messages = defineMessages({ - hostedRetry: { - id: 'app.hosted-packs.progress.retry', - defaultMessage: 'Return to the instance to retry', - }, newDownload: { id: 'app.downloads.new-download', defaultMessage: 'New download' }, inProgress: { id: 'app.downloads.in-progress', defaultMessage: 'In progress' }, history: { id: 'app.downloads.history', defaultMessage: 'History' }, @@ -653,7 +675,15 @@ const messages = defineMessages({ unknownPhase: { id: 'app.downloads.phase.unknown', defaultMessage: 'Working' }, }) +const hostedActionMessages = defineMessages({ + pause: { id: 'app.hosted-packs.progress.pause', defaultMessage: 'Pause' }, + resume: { id: 'app.hosted-packs.progress.resume', defaultMessage: 'Resume' }, + cancel: { id: 'app.hosted-packs.progress.stop', defaultMessage: 'Stop' }, + retry: { id: 'app.downloads.retry', defaultMessage: 'Retry' }, +}) + const statusMessages = defineMessages({ + paused: { id: 'app.downloads.status.paused', defaultMessage: 'Paused' }, queued: { id: 'app.downloads.status.queued', defaultMessage: 'Queued' }, running: { id: 'app.downloads.status.running', defaultMessage: 'Running' }, canceling: { id: 'app.downloads.status.canceling', defaultMessage: 'Canceling' }, @@ -1262,6 +1292,7 @@ function toggleExpanded(jobId: string) { } async function withBusy(jobId: string, action: () => Promise) { + if (busy.value.has(jobId)) return busy.value = new Set([...busy.value, jobId]) try { await action() @@ -1278,6 +1309,21 @@ async function cancel(job: InstallJobSnapshot) { await withBusy(job.job_id, () => manager.cancel(job.job_id)) } +function hostedBusyKey(bar: LoadingBar, action: HostedDownloadAction) { + return `hosted-${action === 'retry' ? 'retry' : 'control'}:${bar.bar_type?.instance_id}` +} + +async function hostedAction(bar: LoadingBar, action: HostedDownloadAction) { + const instanceId = bar.bar_type?.instance_id + if (!instanceId || !hostedDownloadActions(bar).includes(action)) return + await withBusy(hostedBusyKey(bar, action), () => + runHostedDownloadAction(action, instanceId, { + retryHosted: manager.retryHosted, + control: controlHostedDownload, + }), + ) +} + async function retry(job: InstallJobSnapshot) { await withBusy(job.job_id, () => manager.retry(job.job_id)) } @@ -1293,12 +1339,14 @@ async function resolveMissing(job: InstallJobSnapshot) { (item) => item.status === 'skipped' && item.manual_url && item.project_id && item.version_id, ) - .map((item): CurseForgeManualDownloadItem => ({ - projectId: Number(item.project_id), - fileId: Number(item.version_id), - fileName: item.name, - websiteUrl: item.manual_url ?? undefined, - })) + .map( + (item): CurseForgeManualDownloadItem => ({ + projectId: Number(item.project_id), + fileId: Number(item.version_id), + fileName: item.name, + websiteUrl: item.manual_url ?? undefined, + }), + ) const instanceId = job.instance_id const hasGeneralMissingItems = job.items.some((item) => item.status === 'failed') if (instanceId && (job.provider === 'curse_forge' || fallbackCurseForgeItems.length > 0)) { diff --git a/apps/app/build.rs b/apps/app/build.rs index 5cef161..cb2cdd7 100644 --- a/apps/app/build.rs +++ b/apps/app/build.rs @@ -380,6 +380,9 @@ fn main() { "hosted_create", "hosted_binding", "hosted_sync", + "hosted_pause", + "hosted_resume", + "hosted_cancel", "hosted_instance_mode", "hosted_set_instance_mode", "install_pack_to_existing_instance", diff --git a/apps/app/src/api/install.rs b/apps/app/src/api/install.rs index d5d6b8d..1b5652c 100644 --- a/apps/app/src/api/install.rs +++ b/apps/app/src/api/install.rs @@ -20,6 +20,9 @@ pub fn init() -> tauri::plugin::TauriPlugin { hosted_create, hosted_binding, hosted_sync, + hosted_pause, + hosted_resume, + hosted_cancel, hosted_instance_mode, hosted_set_instance_mode, install_get_modpack_preview, @@ -105,6 +108,21 @@ pub async fn hosted_sync( Ok(theseus::pack::hosted::synchronize(&instance_id).await?) } +#[tauri::command] +pub async fn hosted_pause(instance_id: String) -> Result<()> { + Ok(theseus::pack::hosted::pause(&instance_id).await?) +} + +#[tauri::command] +pub async fn hosted_resume(instance_id: String) -> Result<()> { + Ok(theseus::pack::hosted::resume(&instance_id).await?) +} + +#[tauri::command] +pub async fn hosted_cancel(instance_id: String) -> Result<()> { + Ok(theseus::pack::hosted::cancel(&instance_id).await?) +} + #[derive(Deserialize)] #[serde(rename_all = "camelCase")] pub struct InstallCreateInstanceRequest { diff --git a/packages/app-lib/src/api/pack/hosted.rs b/packages/app-lib/src/api/pack/hosted.rs index 04f42e9..7c1e1f9 100644 --- a/packages/app-lib/src/api/pack/hosted.rs +++ b/packages/app-lib/src/api/pack/hosted.rs @@ -1,5 +1,7 @@ //! Administrator-approved skin-site packs. Only content-addressed changed files cross the network. +mod control; mod progress; +pub use control::{HostedTaskState, cancel, pause, resume}; mod tagged; mod transport; use crate::{ @@ -938,10 +940,20 @@ async fn synchronize_locked(instance_id: &str) -> crate::Result { let metadata = crate::instance::get(instance_id) .await? .ok_or_else(|| invalid("Unknown instance"))?; + let task = control::Registration::new(instance_id)?; let progress = PackProgress::new(instance_id, &metadata.instance.name).await?; - let result = - synchronize_with_progress(instance_id, metadata, &progress).await; + let result = task + .control + .run(|| { + synchronize_with_progress( + instance_id, + metadata.clone(), + &progress, + &task.control, + ) + }) + .await; if let Err(error) = &result { progress.fail(error); } @@ -952,6 +964,7 @@ async fn synchronize_with_progress( instance_id: &str, metadata: crate::state::InstanceMetadata, progress: &PackProgress, + control: &control::Control, ) -> crate::Result { if metadata.instance.linked_launcher.is_some() || metadata.instance.symlink_target.is_some() @@ -1275,7 +1288,7 @@ async fn synchronize_with_progress( &format!("文件 0/{total_files}"), true, ); - let results = stream::iter(pending_downloads.into_values().map( + downloaded += control::collect_downloads(stream::iter(pending_downloads.into_values().map( |(file, url, object)| { let current = ¤t; let completed = &completed; @@ -1342,34 +1355,14 @@ async fn synchronize_with_progress( request = request .with_header("Authorization", auth.clone()); } - let first_result = download_to_path( + download_to_path( request.clone(), &object, &state.fetch_semaphore, &state.pool, Some(&mut on_progress), ) - .await; - if let Err(error) = first_result { - tracing::warn!( - path = %file_path, - %error, - "Parallel StarLight file download failed; retrying after the batch load has eased" - ); - ensure_session(auth).await?; - tokio::time::sleep(std::time::Duration::from_millis( - 750, - )) - .await; - download_to_path( - request, - &object, - &state.fetch_semaphore, - &state.pool, - Some(&mut on_progress), - ) - .await?; - } + .await.map_err(|error| error.with_context(format!("下载整合包文件失败：{file_path}")))?; let completed_files = completed.fetch_add(1, Ordering::Relaxed) + 1; let completed_bytes = { @@ -1391,12 +1384,7 @@ async fn synchronize_with_progress( } }, )) - .buffer_unordered(concurrency) - .collect::>() - .await; - downloaded += results.into_iter().try_fold(0, |sum, result| { - Ok::(sum + result?) - })?; + .buffer_unordered(concurrency)).await?; } ensure_session(&auth).await?; downloaded += tagged::download( @@ -1446,6 +1434,8 @@ async fn synchronize_with_progress( previous, metadata, }; + // From this atomic boundary onward the rollback/runtime transaction must finish. + control.begin_finalizing()?; write_json(&root, JOURNAL, &journal)?; let result = async { progress.update(0, 0, "正在安装游戏组件", true); diff --git a/packages/app-lib/src/api/pack/hosted/control.rs b/packages/app-lib/src/api/pack/hosted/control.rs new file mode 100644 index 0000000..9768ba8 --- /dev/null +++ b/packages/app-lib/src/api/pack/hosted/control.rs @@ -0,0 +1,329 @@ +use futures::{Stream, StreamExt}; +use serde::{Deserialize, Serialize}; +use std::{ + future::Future, + sync::{Arc, LazyLock, Mutex}, +}; +use tokio::sync::watch; + +#[derive( + Clone, Copy, Debug, Default, Deserialize, Serialize, Hash, PartialEq, Eq, +)] +#[serde(rename_all = "snake_case")] +pub enum HostedTaskState { + #[default] + Running, + Paused, + Canceling, + Canceled, + Failed, + Finalizing, +} + +static CONTROLS: LazyLock>> = + LazyLock::new(dashmap::DashMap::new); + +pub(super) struct Control { + instance_id: String, + state: Mutex, + changes: watch::Sender, +} + +fn stopped() -> crate::Error { + super::invalid("已停止整合包下载；已下载的缓存会保留，可重试继续") +} + +impl Control { + fn new(instance_id: &str) -> Self { + Self { + instance_id: instance_id.to_owned(), + state: Mutex::new(HostedTaskState::Running), + changes: watch::channel(HostedTaskState::Running).0, + } + } + + fn state(&self) -> HostedTaskState { + *self.state.lock().unwrap_or_else(|error| error.into_inner()) + } + + fn transition( + &self, + allowed: &[HostedTaskState], + next: HostedTaskState, + ) -> crate::Result<()> { + let mut state = + self.state.lock().unwrap_or_else(|error| error.into_inner()); + if !allowed.contains(&state) { + return Err(super::invalid( + "任务状态已变化或正在应用安装文件，当前操作不可用", + )); + } + *state = next; + // Publish under the transition lock, so a late pause cannot overwrite finalization. + crate::event::emit::set_hosted_loading_state(&self.instance_id, next); + self.changes.send_replace(next); + Ok(()) + } + + fn pause(&self) -> crate::Result<()> { + self.transition(&[HostedTaskState::Running], HostedTaskState::Paused) + } + + fn resume(&self) -> crate::Result<()> { + self.transition(&[HostedTaskState::Paused], HostedTaskState::Running) + } + + fn cancel(&self) -> crate::Result<()> { + self.transition( + &[HostedTaskState::Running, HostedTaskState::Paused], + HostedTaskState::Canceling, + ) + } + + pub(super) fn begin_finalizing(&self) -> crate::Result<()> { + self.transition( + &[HostedTaskState::Running], + HostedTaskState::Finalizing, + ) + } + + /// The operation may be dropped only before begin_finalizing. Restarting + /// re-plans from verified cache objects and the downloader's partial files. + pub(super) async fn run( + &self, + mut operation: F, + ) -> crate::Result + where + F: FnMut() -> Fut, + Fut: Future>, + { + let mut changes = self.changes.subscribe(); + loop { + let state = *changes.borrow_and_update(); + match state { + HostedTaskState::Paused => { + changes.changed().await.map_err(|_| stopped())?; + continue; + } + HostedTaskState::Canceling => return Err(stopped()), + _ => {} + } + let work = operation(); + tokio::pin!(work); + loop { + tokio::select! { + biased; + changed = changes.changed() => { + changed.map_err(|_| stopped())?; + match self.state() { + HostedTaskState::Paused => break, + HostedTaskState::Canceling => return Err(stopped()), + _ => {}, + } + } + result = &mut work => { + match self.state() { + HostedTaskState::Paused => break, + HostedTaskState::Canceling => return Err(stopped()), + _ => return result, + } + } + } + } + } + } +} + +pub(super) struct Registration { + instance_id: String, + pub(super) control: Arc, +} + +impl Registration { + pub(super) fn new(instance_id: &str) -> crate::Result { + use dashmap::mapref::entry::Entry; + let control = Arc::new(Control::new(instance_id)); + match CONTROLS.entry(instance_id.to_owned()) { + Entry::Occupied(_) => { + return Err(super::invalid("此实例已有整合包下载任务")); + } + Entry::Vacant(entry) => { + entry.insert(control.clone()); + } + } + Ok(Self { + instance_id: instance_id.to_owned(), + control, + }) + } +} + +impl Drop for Registration { + fn drop(&mut self) { + CONTROLS.remove_if(&self.instance_id, |_, value| { + Arc::ptr_eq(value, &self.control) + }); + } +} + +fn lookup(instance_id: &str) -> crate::Result> { + CONTROLS + .get(instance_id) + .map(|entry| entry.value().clone()) + .ok_or_else(|| super::invalid("整合包下载任务已结束，请重试")) +} + +pub async fn pause(instance_id: &str) -> crate::Result<()> { + lookup(instance_id)?.pause() +} +pub async fn resume(instance_id: &str) -> crate::Result<()> { + lookup(instance_id)?.resume() +} +pub async fn cancel(instance_id: &str) -> crate::Result<()> { + lookup(instance_id)?.cancel() +} + +/// Stop the batch on its first error so stalled peers cannot hide failure. +pub(super) async fn collect_downloads(downloads: S) -> crate::Result +where + S: Stream>, +{ + tokio::pin!(downloads); + let mut total = 0; + while let Some(result) = downloads.next().await { + total += result?; + } + Ok(total) +} + +#[cfg(test)] +mod tests { + use super::*; + use futures::FutureExt; + use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }; + use tokio::sync::{Notify, Semaphore}; + + struct TransferGuard(Arc, Arc); + + impl Drop for TransferGuard { + fn drop(&mut self) { + self.0.fetch_add(1, Ordering::SeqCst); + self.1.notify_one(); + } + } + + #[tokio::test] + async fn pause_drops_live_transfers_and_resume_restarts_before_stop() { + let control = Arc::new(Control::new("pause-resume-test")); + let starts = Arc::new(Semaphore::new(0)); + let drops = Arc::new(AtomicUsize::new(0)); + let dropped = Arc::new(Notify::new()); + let task = { + let control = control.clone(); + let starts = starts.clone(); + let drops = drops.clone(); + let dropped = dropped.clone(); + tokio::spawn(async move { + control + .run(|| { + let guard = + TransferGuard(drops.clone(), dropped.clone()); + starts.add_permits(1); + async move { + let _guard = guard; + std::future::pending::>().await + } + }) + .await + }) + }; + starts.acquire().await.unwrap().forget(); + control.pause().unwrap(); + dropped.notified().await; + assert_eq!(drops.load(Ordering::SeqCst), 1); + assert_eq!(control.state(), HostedTaskState::Paused); + control.resume().unwrap(); + starts.acquire().await.unwrap().forget(); + control.cancel().unwrap(); + let result = task.await.unwrap(); + assert!(result.unwrap_err().to_string().contains("停止")); + assert_eq!(drops.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn stopping_an_already_paused_task_never_starts_a_transfer() { + let control = Arc::new(Control::new("paused-stop-test")); + control.pause().unwrap(); + let task = { + let control = control.clone(); + tokio::spawn(async move { + control + .run(|| async { + panic!("paused work was polled"); + Ok::<(), crate::Error>(()) + }) + .await + }) + }; + control.cancel().unwrap(); + assert!(task.await.unwrap().is_err()); + } + + #[test] + fn finalization_rejects_pause_and_stop() { + let control = Control::new("finalization-test"); + control.begin_finalizing().unwrap(); + assert_eq!(control.state(), HostedTaskState::Finalizing); + assert!(control.pause().is_err()); + assert!(control.cancel().is_err()); + } + + #[test] + fn pause_and_finalization_have_only_one_winner() { + for _ in 0..32 { + let control = Arc::new(Control::new("finalization-race-test")); + let barrier = Arc::new(std::sync::Barrier::new(2)); + let task = { + let control = control.clone(); + let barrier = barrier.clone(); + std::thread::spawn(move || { + barrier.wait(); + control.pause().is_ok() + }) + }; + barrier.wait(); + let finalized = control.begin_finalizing().is_ok(); + assert_ne!(finalized, task.join().unwrap()); + } + } + + #[tokio::test] + async fn first_failure_drops_stalled_peer_downloads() { + let drops = Arc::new(AtomicUsize::new(0)); + let guard = TransferGuard(drops.clone(), Arc::new(Notify::new())); + let downloads = futures::stream::iter(vec![ + async move { + let _guard = guard; + std::future::pending::>().await + } + .boxed(), + async { Err(super::super::invalid("failed file")) }.boxed(), + ]) + .buffer_unordered(2); + let result = collect_downloads(downloads).await; + assert!(result.unwrap_err().to_string().contains("failed file")); + assert_eq!(drops.load(Ordering::SeqCst), 1); + } + + #[test] + fn registration_is_unique_and_released_on_drop() { + let id = uuid::Uuid::new_v4().to_string(); + let registration = Registration::new(&id).unwrap(); + assert!(Registration::new(&id).is_err()); + drop(registration); + assert!(Registration::new(&id).is_ok()); + } +} diff --git a/packages/app-lib/src/api/pack/hosted/progress.rs b/packages/app-lib/src/api/pack/hosted/progress.rs index a412be1..254aa67 100644 --- a/packages/app-lib/src/api/pack/hosted/progress.rs +++ b/packages/app-lib/src/api/pack/hosted/progress.rs @@ -59,6 +59,9 @@ impl PackProgress { instance_id: instance_id.to_owned(), instance_name: name.to_owned(), error: None, + state: super::HostedTaskState::Running, + can_pause: true, + can_cancel: true, }, 1.0, "正在获取整合包信息", diff --git a/packages/app-lib/src/api/pack/hosted/tagged.rs b/packages/app-lib/src/api/pack/hosted/tagged.rs index 8f9b8ba..6545f8a 100644 --- a/packages/app-lib/src/api/pack/hosted/tagged.rs +++ b/packages/app-lib/src/api/pack/hosted/tagged.rs @@ -681,7 +681,7 @@ pub(super) async fn download( set_loading(&bar, 0, file.size, "等待下载")?; work.push((file, url, object, bar)); } - let results = + control::collect_downloads( stream::iter(work.into_iter().map(|(file, url, object, bar)| { let current = ¤t; let state = &state; @@ -740,10 +740,7 @@ pub(super) async fn download( result } })) - .buffer_unordered(concurrency) - .collect::>() - .await; - results - .into_iter() - .try_fold(0, |sum, result| Ok(sum + result?)) + .buffer_unordered(concurrency), + ) + .await } diff --git a/packages/app-lib/src/event/emit.rs b/packages/app-lib/src/event/emit.rs index e2f0c5d..251ea73 100644 --- a/packages/app-lib/src/event/emit.rs +++ b/packages/app-lib/src/event/emit.rs @@ -232,8 +232,24 @@ pub fn fail_hosted_loading(key: &LoadingBarId, error: &str) { && let Some(mut bar) = state.loading_bars.get_mut(&key.0) { match &mut bar.bar_type { - LoadingBarType::HostedPackSync { error: failure, .. } - | LoadingBarType::HostedModDownload { error: failure, .. } => { + LoadingBarType::HostedPackSync { + error: failure, + state, + can_pause, + can_cancel, + .. + } => { + use crate::api::pack::hosted::HostedTaskState; + *state = if *state == HostedTaskState::Canceling { + HostedTaskState::Canceled + } else { + HostedTaskState::Failed + }; + *can_pause = false; + *can_cancel = false; + *failure = Some(error.to_owned()); + } + LoadingBarType::HostedModDownload { error: failure, .. } => { *failure = Some(error.to_owned()) } _ => {} @@ -241,6 +257,67 @@ pub fn fail_hosted_loading(key: &LoadingBarId, error: &str) { } } +/// Update the authoritative snapshot and notify observers even while all +/// transfer futures are paused and no byte-progress event can be produced. +pub fn set_hosted_loading_state( + instance_id: &str, + next: crate::api::pack::hosted::HostedTaskState, +) { + use crate::api::pack::hosted::HostedTaskState; + let Ok(event_state) = crate::EventState::get() else { + return; + }; + for mut bar in event_state.loading_bars.iter_mut() { + let LoadingBarType::HostedPackSync { + instance_id: id, + state, + can_pause, + can_cancel, + .. + } = &mut bar.bar_type + else { + continue; + }; + if id != instance_id { + continue; + } + *state = next; + let controllable = + matches!(next, HostedTaskState::Running | HostedTaskState::Paused); + *can_pause = controllable; + *can_cancel = controllable; + match next { + HostedTaskState::Paused => { + bar.message = "已暂停，点击继续以恢复下载".into() + } + HostedTaskState::Canceling => { + bar.message = "正在停止整合包下载".into() + } + HostedTaskState::Running => { + bar.message = "正在继续整合包下载".into() + } + _ => {} + } + #[cfg(feature = "tauri")] + { + let _ = event_state.app.emit( + "loading", + LoadingPayload { + fraction: Some(if bar.total > 0.0 { + bar.current / bar.total + } else { + 0.0 + }), + message: bar.message.clone(), + event: bar.bar_type.clone(), + loader_uuid: bar.loading_bar_uuid, + total: Some(bar.total), + }, + ); + } + } +} + // emit_warning(message) pub async fn emit_warning(message: &str) -> crate::Result<()> { #[cfg(feature = "tauri")] diff --git a/packages/app-lib/src/event/mod.rs b/packages/app-lib/src/event/mod.rs index 52e5e1c..037e664 100644 --- a/packages/app-lib/src/event/mod.rs +++ b/packages/app-lib/src/event/mod.rs @@ -142,6 +142,12 @@ pub enum LoadingBarType { instance_id: String, instance_name: String, error: Option, + #[serde(default)] + state: crate::api::pack::hosted::HostedTaskState, + #[serde(default)] + can_pause: bool, + #[serde(default)] + can_cancel: bool, }, HostedModDownload { instance_id: String, diff --git a/packages/app-lib/src/util/fetch.rs b/packages/app-lib/src/util/fetch.rs index 72df1ba..3363433 100644 --- a/packages/app-lib/src/util/fetch.rs +++ b/packages/app-lib/src/util/fetch.rs @@ -689,6 +689,46 @@ fn bounded_diagnostic_text(value: impl AsRef, max_chars: usize) -> String { value.as_ref().chars().take(max_chars).collect() } +fn sanitize_download_error_text(value: &str) -> String { + static URLS: LazyLock = LazyLock::new(|| { + regex::Regex::new(r"(?i)\b(?:https?|socks4a?|socks5h?)://[^\s]+") + .expect("download diagnostic URL pattern should be valid") + }); + let sanitized = URLS.replace_all(value, |matched: ®ex::Captures<'_>| { + let url = &matched[0]; + // A malformed URL must not fall back to logging its userinfo. The + // normal helper also strips signed query strings and fragments. + if Url::parse(url).is_ok() { + sanitize_url_for_log(url) + } else { + "[redacted URL]".to_string() + } + }); + bounded_diagnostic_text(sanitized, 256) +} + +fn download_transport_error_detail(source: &reqwest::Error) -> String { + // reqwest's Display stops before the useful DNS/TLS/socket cause and can + // include a signed request URL. Walk only its underlying causes instead. + let mut details = Vec::new(); + let mut cause = source.source(); + while let Some(error) = cause { + let detail = sanitize_download_error_text(&error.to_string()); + if details.last() != Some(&detail) { + details.push(detail); + } + if details.len() == 8 { + break; + } + cause = error.source(); + } + if details.is_empty() { + sanitize_download_error_text(&source.to_string()) + } else { + bounded_diagnostic_text(details.join(": "), 256) + } +} + fn download_error_category(error: &crate::Error) -> &'static str { match error.raw.as_ref() { ErrorKind::FetchError(source) => { @@ -750,7 +790,16 @@ fn download_error_category(error: &crate::Error) -> &'static str { fn download_error_detail(error: &crate::Error) -> String { match error.raw.as_ref() { ErrorKind::FetchError(source) => source.status().map_or_else( - || format!("{} failure", download_error_category(error)), + || { + bounded_diagnostic_text( + format!( + "{} failure: {}", + download_error_category(error), + download_transport_error_detail(source) + ), + 256, + ) + }, |status| format!("HTTP {}", status.as_u16()), ), ErrorKind::LabrinthError(error) => error.status.map_or_else( @@ -929,8 +978,10 @@ fn add_starlight_direct_recovery_route(routes: &mut Vec) { .filter(|route| { route.proxy == ProxyPolicy::System && Url::parse(&route.url).ok().is_some_and(|url| { - url.host_str() == Some("skin.starlight.cool") - && url.path().starts_with("/starlight/mod/packs/files/") + matches!( + url.host_str(), + Some("skin.starlight.cool" | "download.starlight.cool") + ) && url.path().starts_with("/starlight/mod/packs/files/") }) }) .cloned() @@ -6308,7 +6359,9 @@ async fn download_to_path_inner( source = route.source.as_str(), attempt = attempts, max_attempts = file_attempt_budget, - error = %error, + proxy = ?route.proxy, + category = download_error_category(&error), + error = %download_error_detail(&error), "File download request failed; trying the next source or retry" ); last_error = Some(error); @@ -8857,18 +8910,86 @@ mod tests { #[test] fn starlight_hosted_files_have_a_direct_recovery_route() { - let mut routes = vec![official_route( + for url in [ "https://skin.starlight.cool/starlight/mod/packs/files/release/hash", - ResourceClass::Modpack, - )]; + "https://download.starlight.cool:20443/starlight/mod/packs/files/4/hash", + ] { + let mut routes = vec![official_route(url, ResourceClass::Modpack)]; + add_starlight_direct_recovery_route(&mut routes); + add_starlight_direct_recovery_route(&mut routes); - add_starlight_direct_recovery_route(&mut routes); + assert_eq!(routes.len(), 2, "{url}"); + assert_eq!(routes[0].proxy, ProxyPolicy::System); + assert_eq!(routes[1].proxy, ProxyPolicy::Direct); + assert_eq!(routes[0].url, routes[1].url); + assert!(routes[1].allow_sensitive_headers); + } + } - assert_eq!(routes.len(), 2); - assert_eq!(routes[0].proxy, ProxyPolicy::System); - assert_eq!(routes[1].proxy, ProxyPolicy::Direct); - assert_eq!(routes[0].url, routes[1].url); - assert!(routes[1].allow_sensitive_headers); + #[test] + fn starlight_direct_recovery_excludes_other_paths_and_lookalike_hosts() { + for url in [ + "https://download.starlight.cool:20443/starlight/mod/packs", + "https://download.starlight.cool.evil.example/starlight/mod/packs/files/4/hash", + "https://evil.example/starlight/mod/packs/files/4/hash", + "https://skin.starlight.cool/yggdrasil", + ] { + let mut routes = vec![official_route(url, ResourceClass::Modpack)]; + add_starlight_direct_recovery_route(&mut routes); + assert_eq!(routes.len(), 1, "{url}"); + } + } + + #[tokio::test] + async fn download_transport_diagnostic_preserves_causes_without_url_secrets() + { + let listener = + tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + drop(listener); + let url = format!( + "http://username:password@{address}/file?token=secret#fragment" + ); + let error: crate::Error = reqwest::Client::builder() + .no_proxy() + .build() + .unwrap() + .get(url) + .send() + .await + .unwrap_err() + .into(); + let detail = download_error_detail(&error); + assert_eq!(download_error_category(&error), "connect"); + assert!(detail.contains("connect"), "{detail}"); + assert!( + detail.contains("tcp") || detail.contains("os error"), + "{detail}" + ); + for secret in ["username", "password", "token", "secret", "fragment"] { + assert!(!detail.contains(secret), "{detail}"); + } + } + + #[test] + fn download_transport_diagnostic_scrubs_embedded_urls() { + let detail = sanitize_download_error_text( + "request https://username:password@example.com/file?token=secret#fragment failed: connecting to http://proxyuser:proxypass@proxy.example:8080/?key=proxysecret: connection refused", + ); + assert!(detail.contains("https://example.com/file"), "{detail}"); + assert!(detail.contains("http://proxy.example:8080/"), "{detail}"); + assert!(detail.contains("connection refused"), "{detail}"); + for secret in [ + "username", + "password", + "token", + "secret", + "fragment", + "proxyuser", + "proxypass", + ] { + assert!(!detail.contains(secret), "{detail}"); + } } #[test]