diff --git a/backend/crates/gateway-api/src/admin/system.rs b/backend/crates/gateway-api/src/admin/system.rs index d67ee9f7c..8016c2fdb 100644 --- a/backend/crates/gateway-api/src/admin/system.rs +++ b/backend/crates/gateway-api/src/admin/system.rs @@ -206,6 +206,7 @@ struct SystemUpdateEventView { step: Option, message: String, terminal: bool, + progress_percent: Option, at: String, } @@ -408,6 +409,7 @@ impl From for SystemUpdateEventView { step: event.step, message: event.message, terminal: event.terminal, + progress_percent: event.progress_percent, at: event.occurred_at.to_rfc3339(), } } @@ -421,6 +423,7 @@ impl SystemUpdateEventView { "step": self.step, "message": self.message, "terminal": self.terminal, + "progressPercent": self.progress_percent, "at": self.at, }) } diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index e6d66fcd2..7a2807651 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -84,6 +84,10 @@ pub(super) struct AdminTestFixture { impl AdminTestFixture { pub async fn new() -> Self { + Self::with_system(Arc::new(UnusedSystem)).await + } + + pub async fn with_system(system: Arc) -> Self { let api_key = Arc::new(Mutex::new(None)); let auth = Arc::new(MemoryAuthStore::new(api_key.clone())); let settings = Arc::new(MemorySettingsStore::new(api_key)); @@ -113,7 +117,7 @@ impl AdminTestFixture { providers, Arc::new(NoopSnapshot), Arc::new(NoopProbe), - Arc::new(UnusedSystem), + system, ) .await .expect("initialize test admin services"); diff --git a/backend/crates/gateway-api/tests/admin/system.rs b/backend/crates/gateway-api/tests/admin/system.rs index e971379e6..3ff254d48 100644 --- a/backend/crates/gateway-api/tests/admin/system.rs +++ b/backend/crates/gateway-api/tests/admin/system.rs @@ -1,4 +1,24 @@ -use gateway_api::admin::system::{UpdateDetailQuery, UpdateRequest}; +use std::sync::Arc; + +use async_trait::async_trait; +use axum::{ + Router, + body::{Body, to_bytes}, + http::{Request, header}, +}; +use chrono::Utc; +use futures::stream; +use gateway_admin::{ + model::system::{ + SystemOperationAccepted, SystemUpdateDetail, SystemUpdateEvent, SystemUpdateEventLevel, + SystemUpdateStatus, SystemVersion, + }, + ports::system::{SystemOperationError, SystemOperations, SystemUpdateEventStream}, +}; +use gateway_api::admin::system::{self, UpdateDetailQuery, UpdateRequest}; +use tower::ServiceExt as _; + +use super::{AdminTestFixture, AdminTestState, unavailable_system}; #[test] fn update_detail_query_should_default_to_no_refresh_and_reject_unknown_fields() { @@ -16,3 +36,80 @@ fn update_request_should_preserve_target_version_for_domain_validation() { serde_json::from_str::(r#"{"targetVersion":"v0.2.0","extra":1}"#).is_err() ); } + +#[tokio::test] +async fn update_event_stream_should_preserve_download_progress_percent() { + let fixture = AdminTestFixture::with_system(Arc::new(ProgressSystem)).await; + fixture.auth.insert_session("valid-session"); + let response = app(fixture.state()) + .oneshot( + Request::builder() + .uri("/api/admin/system/update/events") + .header(header::COOKIE, "cpr_admin_session=valid-session") + .header("x-request-id", "req_system_update_events") + .body(Body::empty()) + .expect("update event request"), + ) + .await + .expect("update event response"); + let body = to_bytes(response.into_body(), 1024 * 1024) + .await + .expect("update event body"); + let body = String::from_utf8(body.to_vec()).expect("UTF-8 event stream"); + + assert!( + body.contains(r#""progressPercent":40"#), + "unexpected update event stream: {body}" + ); +} + +fn app(state: AdminTestState) -> Router { + system::router::().with_state(state) +} + +struct ProgressSystem; + +#[async_trait] +impl SystemOperations for ProgressSystem { + async fn version(&self) -> Result { + Err(unavailable_system()) + } + + async fn update_detail(&self, _: bool) -> Result { + Err(unavailable_system()) + } + + fn update_events(&self) -> SystemUpdateEventStream { + Box::pin(stream::once(async { + SystemUpdateEvent { + id: "update-progress-1".to_owned(), + operation_id: Some("update-1".to_owned()), + level: SystemUpdateEventLevel::Info, + step: Some("download".to_owned()), + message: "已下载 4.0 MiB / 10.0 MiB (40%)".to_owned(), + terminal: true, + progress_percent: Some(40), + occurred_at: Utc::now(), + } + })) + } + + async fn perform_update( + &self, + _: Option, + ) -> Result { + Err(unavailable_system()) + } + + async fn update_status(&self) -> Result { + Err(unavailable_system()) + } + + async fn rollback(&self) -> Result { + Err(unavailable_system()) + } + + async fn restart(&self) -> Result { + Err(unavailable_system()) + } +} diff --git a/backend/crates/gateway-host/src/system_update/download.rs b/backend/crates/gateway-host/src/system_update/download.rs index 32b512728..4e7a031ea 100644 --- a/backend/crates/gateway-host/src/system_update/download.rs +++ b/backend/crates/gateway-host/src/system_update/download.rs @@ -49,7 +49,7 @@ pub(crate) async fn download_file( .map_err(|error| super::internal(format!("failed to create download file: {error}")))?; let mut stream = response.bytes_stream(); let mut downloaded = 0_u64; - let mut last_percent = 0_u8; + let mut next_progress = 10_u8; while let Some(chunk) = stream.next().await { let chunk = chunk.map_err(|error| upstream(format!("download stream failed: {error}")))?; downloaded = downloaded.saturating_add(u64::try_from(chunk.len()).unwrap_or(u64::MAX)); @@ -59,9 +59,17 @@ pub(crate) async fn download_file( file.write_all(&chunk) .map_err(|error| super::internal(format!("failed to write download: {error}")))?; let percent = download_percent(downloaded, expected_size); - if percent > last_percent { - last_percent = percent; - events.emit_progress(operation_id, "downloading release archive", percent); + if percent >= next_progress || downloaded == expected_size { + events.emit_progress( + operation_id, + &format!( + "已下载 {} / {} ({percent}%)", + format_bytes(downloaded), + format_bytes(expected_size) + ), + percent, + ); + next_progress = percent.saturating_add(10); } } file.sync_all() @@ -155,3 +163,15 @@ fn download_percent(downloaded: u64, total: u64) -> u8 { ) .unwrap_or(100) } + +pub(crate) fn format_bytes(bytes: u64) -> String { + const KIB: u64 = 1024; + const MIB: u64 = KIB * 1024; + if bytes >= MIB { + return format!("{:.1} MiB", bytes as f64 / MIB as f64); + } + if bytes >= KIB { + return format!("{:.1} KiB", bytes as f64 / KIB as f64); + } + format!("{bytes} B") +} diff --git a/backend/crates/gateway-host/src/system_update/mod.rs b/backend/crates/gateway-host/src/system_update/mod.rs index 7d89d2d85..aa456ebb0 100644 --- a/backend/crates/gateway-host/src/system_update/mod.rs +++ b/backend/crates/gateway-host/src/system_update/mod.rs @@ -232,8 +232,15 @@ impl ProcessSystemOperations { .as_deref() .ok_or_else(|| conflict("update repository is not configured"))?; self.events - .info(None, Some("release"), "fetching latest release"); - let release = fetch_latest(&self.config.github_api_base, repository).await?; + .info(None, Some("release"), "正在获取最新 Release 信息"); + let release = match fetch_latest(&self.config.github_api_base, repository).await { + Ok(release) => release, + Err(error) => { + self.events + .error_terminal(None, Some("release"), error.to_string()); + return Err(error); + } + }; let detail = detail_from_release(&self.config, &release); if detail.latest_version != target { self.events.warning_terminal( @@ -258,12 +265,17 @@ impl ProcessSystemOperations { Some(&target), &self.config.version, )?; + self.events.info( + Some(&operation_id), + Some("prepare"), + format!("准备更新到 v{target}"), + ); let result = self.install_release(&release, &target, &operation_id).await; match &result { Ok(()) => self.events.success_terminal( Some(&operation_id), Some("done"), - "release files replaced", + "更新文件已替换，等待服务重启生效", ), Err(error) => { self.events @@ -294,7 +306,23 @@ impl ProcessSystemOperations { version: &str, operation_id: &str, ) -> Result<(), OperationError> { + self.events.info( + Some(operation_id), + Some("asset"), + "正在选择匹配当前平台的更新包", + ); let archive = select_archive(release, version)?; + self.events.info( + Some(operation_id), + Some("asset"), + format!( + "已选择更新包 {} ({})", + archive.name, + download::format_bytes(archive.size) + ), + ); + self.events + .info(Some(operation_id), Some("verify"), "正在校验更新资源"); if archive.size == 0 || archive.size > MAX_DOWNLOAD_SIZE { return Err(invalid("release archive size is invalid")); } @@ -306,12 +334,16 @@ impl ProcessSystemOperations { if checksum.size == 0 || checksum.size > MAX_CHECKSUM_SIZE { return Err(invalid("release checksum size is invalid")); } + self.events + .info(Some(operation_id), Some("prepare"), "正在创建临时更新目录"); fs::create_dir_all(&self.config.update_temp_dir) .map_err(|error| internal(format!("failed to prepare update temp dir: {error}")))?; let temp_root = fs::canonicalize(&self.config.update_temp_dir) .map_err(|error| internal(format!("failed to resolve update temp dir: {error}")))?; let temp = UpdateTempDir::create(&temp_root)?; let archive_path = temp.path().join(&archive.name); + self.events + .info(Some(operation_id), Some("download"), "开始下载更新包"); download_file( &archive.browser_download_url, &archive_path, @@ -321,6 +353,10 @@ impl ProcessSystemOperations { &self.events, ) .await?; + self.events + .success(Some(operation_id), Some("download"), "更新包下载完成"); + self.events + .info(Some(operation_id), Some("checksum"), "正在校验 checksum"); verify_checksum( &archive_path, &archive.name, @@ -329,12 +365,23 @@ impl ProcessSystemOperations { &self.config.github_api_base, ) .await?; + self.events + .success(Some(operation_id), Some("checksum"), "checksum 校验通过"); + self.events + .info(Some(operation_id), Some("extract"), "正在解压更新包"); let extracted = extract_release(&archive_path, temp.path())?; + self.events + .success(Some(operation_id), Some("extract"), "更新包解压完成"); + self.events + .info(Some(operation_id), Some("replace"), "正在替换应用文件"); replace_release_files( &self.config.executable_path()?, &self.config.web_dist_dir, extracted, - ) + )?; + self.events + .success(Some(operation_id), Some("replace"), "应用文件替换完成"); + Ok(()) } } @@ -515,6 +562,17 @@ impl UpdateEvents { ); } + fn success(&self, operation_id: Option<&str>, step: Option<&str>, message: impl Into) { + self.emit( + SystemUpdateEventLevel::Success, + operation_id, + step, + message, + None, + false, + ); + } + fn warning_terminal( &self, operation_id: Option<&str>, diff --git a/backend/crates/gateway-host/tests/system_update/mod.rs b/backend/crates/gateway-host/tests/system_update/mod.rs index a589ad48b..0540c6833 100644 --- a/backend/crates/gateway-host/tests/system_update/mod.rs +++ b/backend/crates/gateway-host/tests/system_update/mod.rs @@ -204,6 +204,68 @@ async fn update_events_should_open_authenticated_sse_stream() { assert!(stream.next().await.is_some()); } +#[tokio::test] +async fn update_events_should_preserve_complete_release_stage_sequence() { + let server = MockServer::start().await; + let fixture = Fixture::new(); + fixture + .mount_release( + &server, + TARGET_VERSION, + ArchiveKind::Safe, + ChecksumKind::Valid, + ) + .await; + let service = fixture.service(&server); + let events = service.update_events(); + service + .perform_update(Some(TARGET_VERSION.to_owned())) + .await + .expect("update"); + let steps = events + .map(|event| event.step.expect("release update step")) + .collect::>() + .await; + + assert_eq!( + steps, + [ + "release", "prepare", "asset", "asset", "verify", "prepare", "download", "download", + "download", "checksum", "checksum", "extract", "extract", "replace", "replace", "done", + ] + ); +} + +#[tokio::test] +async fn update_events_should_report_download_bytes_and_percent() { + let server = MockServer::start().await; + let fixture = Fixture::new(); + fixture + .mount_release( + &server, + TARGET_VERSION, + ArchiveKind::Safe, + ChecksumKind::Valid, + ) + .await; + let service = fixture.service(&server); + let events = service.update_events(); + service + .perform_update(Some(TARGET_VERSION.to_owned())) + .await + .expect("update"); + let progress = events + .filter_map( + |event| async move { event.progress_percent.map(|value| (value, event.message)) }, + ) + .collect::>() + .await; + + assert!(progress.iter().any(|(percent, message)| { + *percent == 100 && message.starts_with("已下载 ") && message.ends_with(" (100%)") + })); +} + #[tokio::test] async fn update_should_fail_when_release_checksum_is_missing() { let server = MockServer::start().await; diff --git a/frontend/src/stores/modules/system-update.ts b/frontend/src/stores/modules/system-update.ts index fd954b6cc..3ce0f1d76 100644 --- a/frontend/src/stores/modules/system-update.ts +++ b/frontend/src/stores/modules/system-update.ts @@ -1,4 +1,4 @@ -import { useEventSource } from '@vueuse/core' +import { until, useEventSource } from '@vueuse/core' import { delay } from 'es-toolkit' import { defineStore } from 'pinia' @@ -14,6 +14,7 @@ import { ApiError } from '@/api/request' import { errorMessage } from '@/utils/async' const maxUpdateLogs = 200 +const updateEventReadyTimeoutMs = 3_000 const restartReadyTimeoutMs = 60_000 const restartProbeTimeoutMs = 2_000 const restartReadyPollIntervalMs = 500 @@ -106,19 +107,30 @@ export const useSystemUpdateStore = defineStore('system-update', () => { } }) - function connectUpdateEvents(force = false) { + async function connectUpdateEvents(force = false) { if (force) { disconnectUpdateEvents() } else if (updateEventSource.value) { - return + return updateEventStatus.value === 'OPEN' } updateStreamError.value = '' openUpdateEventSource() if (!updateEventSource.value) { updateStreamError.value = '当前浏览器不支持实时更新日志' + return false } + + if (updateEventStatus.value !== 'OPEN') { + await until(updateEventStatus).toBe('OPEN', { + timeout: updateEventReadyTimeoutMs, + }) + } + const connected = updateEventStatus.value === 'OPEN' + if (!connected) + updateStreamError.value = '更新日志连接超时' + return connected } function disconnectUpdateEvents() { @@ -186,7 +198,7 @@ export const useSystemUpdateStore = defineStore('system-update', () => { return null clearUpdateLogs() - connectUpdateEvents(true) + await connectUpdateEvents(true) updating.value = true updateError.value = '' updateSuccess.value = false