From 9e5ee747bf255a418111cf0aa2698e449698efba Mon Sep 17 00:00:00 2001 From: zyycn Date: Sat, 4 Jul 2026 15:11:00 +0800 Subject: [PATCH] fix: unblock system update restart Co-authored-by: Codex --- backend/src/admin/system/updater/mod.rs | 35 +++++++++++------- backend/src/runtime/shutdown.rs | 7 +++- backend/tests/admin/system/mod.rs | 39 ++++++++++++++++++++- docs/architecture.md | 1 - frontend/src/composables/useSystemUpdate.ts | 1 + release/platforms.yaml | 7 ---- 6 files changed, 67 insertions(+), 23 deletions(-) diff --git a/backend/src/admin/system/updater/mod.rs b/backend/src/admin/system/updater/mod.rs index ad448e217..ed0318d8f 100644 --- a/backend/src/admin/system/updater/mod.rs +++ b/backend/src/admin/system/updater/mod.rs @@ -25,7 +25,7 @@ use tokio::sync::{broadcast, Mutex}; use crate::{ admin::auth::session::AdminAuth, admin::response::{AdminEnvelope, AdminError, AdminResponse}, - runtime::shutdown::{request_process_restart, request_shutdown}, + runtime::shutdown::{request_process_restart, request_shutdown, shutdown_subscription}, }; use super::state::{ @@ -155,20 +155,29 @@ pub(crate) async fn update_event_stream( _auth: AdminAuth, ) -> Result>>, AdminError> { let receiver = update_event_sender().subscribe(); - let stream = futures::stream::unfold(receiver, |mut receiver| async move { - loop { - match receiver.recv().await { - Ok(message) => { - 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)); + let shutdown = shutdown_subscription(); + let stream = futures::stream::unfold( + (receiver, shutdown), + |(mut receiver, mut shutdown)| async move { + loop { + tokio::select! { + _ = shutdown.recv() => return None, + received = receiver.recv() => { + match received { + Ok(message) => { + 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))); + } + Err(broadcast::error::RecvError::Lagged(_)) => continue, + Err(broadcast::error::RecvError::Closed) => return None, + } + } } - Err(broadcast::error::RecvError::Lagged(_)) => continue, - Err(broadcast::error::RecvError::Closed) => return None, } - } - }); + }, + ); Ok(Sse::new(stream).keep_alive(KeepAlive::default())) } diff --git a/backend/src/runtime/shutdown.rs b/backend/src/runtime/shutdown.rs index 91f5ada1d..0e86d3fb3 100644 --- a/backend/src/runtime/shutdown.rs +++ b/backend/src/runtime/shutdown.rs @@ -9,7 +9,7 @@ use std::{ }; use tokio::signal; -use tokio::sync::broadcast; +use tokio::sync::broadcast::{self, Receiver}; static SHUTDOWN_REQUESTS: OnceLock> = OnceLock::new(); static RESTART_REQUESTED: AtomicBool = AtomicBool::new(false); @@ -47,6 +47,11 @@ pub fn request_shutdown() { let _ = shutdown_sender().send(()); } +/// 订阅进程关闭请求。 +pub fn shutdown_subscription() -> Receiver<()> { + shutdown_sender().subscribe() +} + /// 请求进程完成优雅关闭后以新二进制替换当前进程。 pub fn request_process_restart(executable_path: PathBuf) { let _ = RESTART_EXECUTABLE_PATH.set(executable_path); diff --git a/backend/tests/admin/system/mod.rs b/backend/tests/admin/system/mod.rs index dd77353cc..f3b0b42d3 100644 --- a/backend/tests/admin/system/mod.rs +++ b/backend/tests/admin/system/mod.rs @@ -6,7 +6,7 @@ use std::{ }; use axum::{ - body::Body, + body::{to_bytes, Body}, http::{header::CONTENT_TYPE, Request, StatusCode}, }; use codex_proxy_rs::{ @@ -821,6 +821,43 @@ async fn update_events_should_open_authenticated_sse_stream() { assert!(response.status() == StatusCode::OK && content_type.starts_with("text/event-stream")); } +#[tokio::test] +async fn update_events_should_close_on_shutdown() { + let _guard = SYSTEM_ENV_LOCK.lock().await; + let deploy = tempfile::tempdir().unwrap(); + set_system_update_env( + "zyycn/codex-proxy-rs-events-shutdown-route", + "http://127.0.0.1:9", + ); + set_system_update_paths( + deploy.path(), + &deploy.path().join("codex-proxy-rs"), + &deploy.path().join("web/dist"), + ); + let (app, _dir) = admin_system_test_app("system-events-shutdown.sqlite").await; + + let response = app + .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_shutdown") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + codex_proxy_rs::runtime::shutdown::request_shutdown(); + + let body = tokio::time::timeout(Duration::from_secs(2), to_bytes(response.into_body(), 1024)) + .await + .expect("update event stream should end after shutdown") + .unwrap(); + + assert!(body.is_empty()); +} + async fn get_update_detail(app: &axum::Router, refresh: bool, request_id: &str) -> Value { let response = app .clone() diff --git a/docs/architecture.md b/docs/architecture.md index e19b626dc..2f4f5cf7a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -165,7 +165,6 @@ Release 产物： ```text codex-proxy-rs__linux_amd64.tar.gz codex-proxy-rs__linux_arm64.tar.gz -codex-proxy-rs__darwin_amd64.tar.gz codex-proxy-rs__darwin_arm64.tar.gz codex-proxy-rs__windows_amd64.zip checksums.txt diff --git a/frontend/src/composables/useSystemUpdate.ts b/frontend/src/composables/useSystemUpdate.ts index 5e42b3715..52a03ee94 100644 --- a/frontend/src/composables/useSystemUpdate.ts +++ b/frontend/src/composables/useSystemUpdate.ts @@ -261,6 +261,7 @@ async function restartNow() { restartCountdown.value = restartCountdownSeconds updateError.value = '' clearRestartTimer() + disconnectUpdateEvents() try { await restartSystem() diff --git a/release/platforms.yaml b/release/platforms.yaml index 4544b137b..b0f273d0f 100644 --- a/release/platforms.yaml +++ b/release/platforms.yaml @@ -17,13 +17,6 @@ platforms: docker: true dockerPlatform: linux/arm64 - - os: darwin - arch: amd64 - runner: macos-15-intel - target: x86_64-apple-darwin - binary: codex-proxy-rs - archive: tar.gz - - os: darwin arch: arm64 runner: macos-15