fix: unblock system update restart

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-07-04 15:11:00 +08:00
co-authored by Codex
parent 3d1e636c5c
commit 9e5ee747bf
6 changed files with 67 additions and 23 deletions
+22 -13
View File
@@ -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<Sse<impl Stream<Item = Result<Event, Infallible>>>, 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()))
}
+6 -1
View File
@@ -9,7 +9,7 @@ use std::{
};
use tokio::signal;
use tokio::sync::broadcast;
use tokio::sync::broadcast::{self, Receiver};
static SHUTDOWN_REQUESTS: OnceLock<broadcast::Sender<()>> = 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);
+38 -1
View File
@@ -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()
-1
View File
@@ -165,7 +165,6 @@ Release 产物:
```text
codex-proxy-rs_<version>_linux_amd64.tar.gz
codex-proxy-rs_<version>_linux_arm64.tar.gz
codex-proxy-rs_<version>_darwin_amd64.tar.gz
codex-proxy-rs_<version>_darwin_arm64.tar.gz
codex-proxy-rs_<version>_windows_amd64.zip
checksums.txt
@@ -261,6 +261,7 @@ async function restartNow() {
restartCountdown.value = restartCountdownSeconds
updateError.value = ''
clearRestartTimer()
disconnectUpdateEvents()
try {
await restartSystem()
-7
View File
@@ -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