diff --git a/backend/src/admin/system/updater/mod.rs b/backend/src/admin/system/updater/mod.rs index ed0318d8f..614e39a66 100644 --- a/backend/src/admin/system/updater/mod.rs +++ b/backend/src/admin/system/updater/mod.rs @@ -110,6 +110,8 @@ struct SystemUpdateEvent { level: UpdateLogLevel, step: Option, message: String, + #[serde(skip_serializing_if = "is_false")] + terminal: bool, at: String, } @@ -157,18 +159,23 @@ pub(crate) async fn update_event_stream( let receiver = update_event_sender().subscribe(); let shutdown = shutdown_subscription(); let stream = futures::stream::unfold( - (receiver, shutdown), - |(mut receiver, mut shutdown)| async move { + (receiver, shutdown, false), + |(mut receiver, mut shutdown, close_after_send)| async move { + if close_after_send { + return None; + } + loop { tokio::select! { _ = shutdown.recv() => return None, received = receiver.recv() => { match received { Ok(message) => { + let close_after_send = message.terminal; let id = message.id.clone(); let data = serde_json::to_string(&message).unwrap_or_else(|_| "{}".to_string()); let event = Event::default().event("update").id(id).data(data); - return Some((Ok(event), (receiver, shutdown))); + return Some((Ok(event), (receiver, shutdown, close_after_send))); } Err(broadcast::error::RecvError::Lagged(_)) => continue, Err(broadcast::error::RecvError::Closed) => return None, @@ -192,7 +199,7 @@ pub(crate) async fn perform_update( .map_err(|_| AdminError::conflict("System update already running"))?; let config = SystemUpdateConfig::from_env(); if let Some(reason) = config.update_support_error() { - emit_update_event( + emit_terminal_update_event( UpdateLogLevel::Error, None, Some("preflight"), @@ -217,7 +224,7 @@ pub(crate) async fn perform_update( ) .await .map_err(|error| { - emit_update_event(UpdateLogLevel::Error, None, Some("release"), error.clone()); + emit_terminal_update_event(UpdateLogLevel::Error, None, Some("release"), error.clone()); AdminError::bad_gateway(error) })?; let info = update_info_from_release(&config, release.clone()); @@ -226,7 +233,7 @@ pub(crate) async fn perform_update( "远端最新版本已变更为 v{},请重新检查并确认", info.latest_version ); - emit_update_event( + emit_terminal_update_event( UpdateLogLevel::Warning, None, Some("release"), @@ -235,7 +242,7 @@ pub(crate) async fn perform_update( return Err(AdminError::conflict(message)); } if !info.has_update { - emit_update_event( + emit_terminal_update_event( UpdateLogLevel::Warning, None, Some("release"), @@ -261,14 +268,14 @@ pub(crate) async fn perform_update( ); let result = perform_release_update(&config, &release, &target_version, &operation_id).await; if let Err(error) = &result { - emit_update_event( + emit_terminal_update_event( UpdateLogLevel::Error, Some(&operation_id), Some("failed"), error.to_string(), ); } else { - emit_update_event( + emit_terminal_update_event( UpdateLogLevel::Success, Some(&operation_id), Some("done"), @@ -732,6 +739,25 @@ fn emit_update_event( operation_id: Option<&str>, step: Option<&str>, message: impl Into, +) { + emit_update_event_with_terminal(level, operation_id, step, message, false); +} + +fn emit_terminal_update_event( + level: UpdateLogLevel, + operation_id: Option<&str>, + step: Option<&str>, + message: impl Into, +) { + emit_update_event_with_terminal(level, operation_id, step, message, true); +} + +fn emit_update_event_with_terminal( + level: UpdateLogLevel, + operation_id: Option<&str>, + step: Option<&str>, + message: impl Into, + terminal: bool, ) { let now = Utc::now(); let event = SystemUpdateEvent { @@ -744,11 +770,16 @@ fn emit_update_event( level, step: step.map(ToString::to_string), message: message.into(), + terminal, at: now.to_rfc3339(), }; let _ = update_event_sender().send(event); } +fn is_false(value: &bool) -> bool { + !*value +} + fn format_bytes(bytes: u64) -> String { const KIB: u64 = 1024; const MIB: u64 = KIB * 1024; diff --git a/backend/tests/admin/system/mod.rs b/backend/tests/admin/system/mod.rs index 7c4bd01fc..29d528989 100644 --- a/backend/tests/admin/system/mod.rs +++ b/backend/tests/admin/system/mod.rs @@ -860,6 +860,70 @@ async fn update_events_should_close_on_shutdown() { assert!(body.is_empty()); } +#[tokio::test] +async fn update_events_should_close_after_terminal_update_log() { + let _guard = SYSTEM_ENV_LOCK.lock().await; + let repository = "zyycn/codex-proxy-rs-events-terminal-route"; + let github = MockServer::start().await; + let deploy = tempfile::tempdir().unwrap(); + let exe_path = deploy.path().join("codex-proxy-rs"); + let web_dist = deploy.path().join("web/dist"); + std::fs::create_dir_all(&web_dist).unwrap(); + std::fs::write(&exe_path, "old-binary").unwrap(); + std::fs::write(web_dist.join("index.html"), "old-web").unwrap(); + mount_latest_release_with_archive(&github, repository, "0.4.0", "new-binary", "new-web").await; + set_system_update_env(repository, &github.uri()); + set_system_update_paths(deploy.path(), &exe_path, &web_dist); + let (app, _dir) = admin_system_test_app("system-events-terminal.sqlite").await; + + let sse_response = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/system/update-events") + .header("cookie", "cpr_admin_session=session_1") + .header("x-request-id", "req_system_update_events_terminal") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let update_app = app.clone(); + let update_task = tokio::spawn(async move { + update_app + .oneshot( + Request::builder() + .method("POST") + .uri("/api/admin/system/update") + .header("cookie", "cpr_admin_session=session_1") + .header("x-request-id", "req_system_update_events_terminal_update") + .header(CONTENT_TYPE, "application/json") + .body(confirmed_update_body("0.4.0")) + .unwrap(), + ) + .await + .unwrap() + }); + + let body = tokio::time::timeout( + Duration::from_secs(2), + to_bytes(sse_response.into_body(), 128 * 1024), + ) + .await + .expect("update event stream should end after terminal update log") + .unwrap(); + let update_response = update_task.await.unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + + assert!( + update_response.status() == StatusCode::OK + && text.contains("\"step\":\"done\"") + && text.contains("\"terminal\":true") + && text.contains("更新文件已替换,等待服务重启生效") + ); +} + async fn get_update_detail(app: &axum::Router, refresh: bool, request_id: &str) -> Value { let response = app .clone() diff --git a/frontend/src/composables/useSystemUpdate.ts b/frontend/src/composables/useSystemUpdate.ts index 52a03ee94..02f6f8cd0 100644 --- a/frontend/src/composables/useSystemUpdate.ts +++ b/frontend/src/composables/useSystemUpdate.ts @@ -19,6 +19,7 @@ export interface SystemUpdateLog { level: SystemUpdateLogLevel step?: string message: string + terminal?: boolean at: string } @@ -84,8 +85,12 @@ function clearUpdateLogs() { updateStreamError.value = '' } -function connectUpdateEvents() { - if (updateEventSource.value) return +function connectUpdateEvents(options: { force?: boolean } = {}) { + if (options.force) { + disconnectUpdateEvents() + } else if (updateEventSource.value) { + return + } if (typeof EventSource === 'undefined') { updateStreamError.value = '当前浏览器不支持实时更新日志' @@ -97,6 +102,8 @@ function connectUpdateEvents() { updateEventSource.value = source source.onopen = () => { + if (updateEventSource.value !== source) return + updateStreaming.value = true updateStreamError.value = '' } @@ -106,6 +113,9 @@ function connectUpdateEvents() { const log = JSON.parse((event as MessageEvent).data) if (isUpdateLog(log)) { appendUpdateLog(log) + if (log.terminal) { + disconnectUpdateEvents(source) + } } } catch { updateStreamError.value = '更新日志解析失败' @@ -113,6 +123,8 @@ function connectUpdateEvents() { }) source.onerror = () => { + if (updateEventSource.value !== source) return + updateStreaming.value = false updateStreamError.value = '更新日志连接中断' if (source.readyState === EventSource.CLOSED) { @@ -121,8 +133,10 @@ function connectUpdateEvents() { } } -function disconnectUpdateEvents() { - updateEventSource.value?.close() +function disconnectUpdateEvents(source: EventSource | null = updateEventSource.value) { + source?.close() + if (source && updateEventSource.value !== source) return + updateEventSource.value = null updateStreaming.value = false } @@ -184,7 +198,7 @@ async function updateNow(targetVersion: string) { if (!canUpdate.value || !currentInfo || updating.value || !confirmedTargetVersion) return null clearUpdateLogs() - connectUpdateEvents() + connectUpdateEvents({ force: true }) updating.value = true updateError.value = '' updateSuccess.value = false diff --git a/frontend/src/layout/components/SystemUpdateModal.vue b/frontend/src/layout/components/SystemUpdateModal.vue index 30dc354f0..afe3ea041 100644 --- a/frontend/src/layout/components/SystemUpdateModal.vue +++ b/frontend/src/layout/components/SystemUpdateModal.vue @@ -45,7 +45,6 @@ const { updateNow, restartNow, clearRestartTimer, - connectUpdateEvents, disconnectUpdateEvents, } = useSystemUpdate() @@ -54,7 +53,6 @@ const updateConfirmOpen = shallowRef(false) const updateConfirmInfo = shallowRef(null) const updateConfirmPreviousTarget = shallowRef('') const preparingUpdate = shallowRef(false) -let updateLogPinTimer: number | undefined const statusView = computed(() => { if (updating.value) { @@ -137,7 +135,6 @@ const renderedReleaseNotes = computed(() => renderMarkdown(updateInfo.value?.not const showUpdateProgress = computed( () => hasUpdate.value || updating.value || restarting.value || updateLogRows.value.length > 0, ) -const shouldConnectUpdateEvents = computed(() => (open.value && hasUpdate.value) || updating.value) const updateConfirmRows = computed(() => [ { @@ -212,20 +209,6 @@ function pinUpdateLogsToBottom() { }) } -function startUpdateLogPinning() { - if (updateLogPinTimer !== undefined) return - - pinUpdateLogsToBottom() - updateLogPinTimer = window.setInterval(pinUpdateLogsToBottom, 250) -} - -function stopUpdateLogPinning() { - if (updateLogPinTimer === undefined) return - - window.clearInterval(updateLogPinTimer) - updateLogPinTimer = undefined -} - async function handleCheckUpdates(force = true) { try { const data = await checkUpdates(force) @@ -304,45 +287,17 @@ watch(open, (visible) => { } }) -watch( - shouldConnectUpdateEvents, - (shouldConnect) => { - if (shouldConnect) { - connectUpdateEvents() - } else { - disconnectUpdateEvents() - } - }, - { immediate: true }, -) - watch( () => updateLogs.value.at(-1)?.id, - () => { + (logId, previousLogId) => { + if (!logId || logId === previousLogId) return + pinUpdateLogsToBottom() }, -) - -watch( - () => updateLogRows.value.length, - () => { - pinUpdateLogsToBottom() - }, -) - -watch( - () => updating.value || restarting.value || updateStreaming.value, - (active) => { - if (active) { - startUpdateLogPinning() - return - } - stopUpdateLogPinning() - }, + { flush: 'post' }, ) onUnmounted(() => { - stopUpdateLogPinning() clearRestartTimer() disconnectUpdateEvents() })