mirror of
https://github.com/openai/codex.git
synced 2026-10-21 04:45:00 +00:00
Support graceful daemon shutdown on Windows (#42364)
## Why Detached Windows processes cannot receive console signals, so app-server daemons need another way to enter the existing graceful and forced shutdown flow. ## What changed - Watch the path in `CODEX_DAEMON_SHUTDOWN_FILE` for a request addressed to the current process, and consume only a matching PID. - Race daemon shutdown requests with Ctrl-C on Windows while leaving unmanaged launches on the existing console-signal path. - Exercise the SIGTERM-equivalent graceful and forced WebSocket shutdown cases on Windows. ## Testing Added unit coverage ensuring inherited control paths and malformed requests cannot consume another process's shutdown request. GitOrigin-RevId: 426a3c12edc344802395d4d26288f2bb015c484f
This commit is contained in:
@@ -0,0 +1,54 @@
|
||||
//! Detached Windows processes cannot receive console signals. The daemon passes
|
||||
//! a request path inside its user-only state directory. Only the addressed PID
|
||||
//! consumes a request, triggering the SIGTERM drain without changing the wire API.
|
||||
|
||||
use std::io;
|
||||
use std::io::Read;
|
||||
use std::path::Path;
|
||||
#[cfg(windows)]
|
||||
use std::path::PathBuf;
|
||||
#[cfg(windows)]
|
||||
use std::time::Duration;
|
||||
|
||||
#[cfg(windows)]
|
||||
pub const DAEMON_SHUTDOWN_FILE_ENV: &str = "CODEX_DAEMON_SHUTDOWN_FILE";
|
||||
|
||||
/// Waits for and consumes one daemon shutdown request. Without a managed launch
|
||||
/// environment this stays pending, leaving normal console shutdown unchanged.
|
||||
#[cfg(windows)]
|
||||
pub async fn daemon_shutdown_signal() -> io::Result<()> {
|
||||
let Some(path) = std::env::var_os(DAEMON_SHUTDOWN_FILE_ENV).map(PathBuf::from) else {
|
||||
return std::future::pending().await;
|
||||
};
|
||||
loop {
|
||||
if take_shutdown_request(&path, std::process::id())? {
|
||||
return Ok(());
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
}
|
||||
|
||||
fn take_shutdown_request(path: &Path, pid: u32) -> io::Result<bool> {
|
||||
let file = match std::fs::File::open(path) {
|
||||
Ok(file) => file,
|
||||
Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(false),
|
||||
Err(err) => return Err(err),
|
||||
};
|
||||
// Descendant app-servers may inherit the control path. Only the intended
|
||||
// process can consume a request. Bound reads to a u32 PID plus one extra byte.
|
||||
let mut contents = Vec::new();
|
||||
file.take(/*limit*/ 11).read_to_end(&mut contents)?;
|
||||
if contents != pid.to_string().as_bytes() {
|
||||
return Ok(false);
|
||||
}
|
||||
// Synchronous consumption cannot lose a request to select cancellation.
|
||||
match std::fs::remove_file(path) {
|
||||
Ok(()) => Ok(true),
|
||||
Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(false),
|
||||
Err(err) => Err(err),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "daemon_shutdown_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,22 @@
|
||||
use super::take_shutdown_request;
|
||||
|
||||
#[test]
|
||||
fn inherited_control_path_cannot_consume_another_process_request() {
|
||||
let directory = tempfile::tempdir().expect("directory");
|
||||
let request = directory.path().join("shutdown");
|
||||
std::fs::write(&request, "1234").expect("request");
|
||||
assert!(!take_shutdown_request(&request, /*pid*/ 5678).expect("descendant probe"));
|
||||
assert!(take_shutdown_request(&request, /*pid*/ 1234).expect("parent probe"));
|
||||
assert!(!take_shutdown_request(&request, /*pid*/ 1234).expect("already consumed"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn incomplete_or_invalid_requests_are_not_consumed() {
|
||||
let directory = tempfile::tempdir().expect("directory");
|
||||
let request = directory.path().join("shutdown");
|
||||
for contents in [b"".as_slice(), b"123", b"1234junk", &[0xff; 20]] {
|
||||
std::fs::write(&request, contents).expect("request");
|
||||
assert!(!take_shutdown_request(&request, /*pid*/ 1234).expect("probe"));
|
||||
assert!(request.exists());
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,9 @@
|
||||
#[cfg(any(windows, test))]
|
||||
mod daemon_shutdown;
|
||||
#[cfg(windows)]
|
||||
pub use daemon_shutdown::DAEMON_SHUTDOWN_FILE_ENV;
|
||||
#[cfg(windows)]
|
||||
pub use daemon_shutdown::daemon_shutdown_signal;
|
||||
mod outgoing_message;
|
||||
mod transport;
|
||||
|
||||
|
||||
@@ -217,11 +217,28 @@ async fn shutdown_signal() -> IoResult<ShutdownSignal> {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(not(unix))]
|
||||
#[cfg(windows)]
|
||||
{
|
||||
tokio::signal::ctrl_c()
|
||||
.await
|
||||
.map(|_| ShutdownSignal::Forceable)
|
||||
let console_signal = async {
|
||||
if tokio::signal::ctrl_c().await.is_err() {
|
||||
// A detached daemon has no console; keep its local control path active.
|
||||
std::future::pending::<()>().await;
|
||||
}
|
||||
};
|
||||
let daemon_signal = async {
|
||||
let result = codex_app_server_transport::daemon_shutdown_signal().await;
|
||||
// The processor retries listener errors; the updater instead needs
|
||||
// immediate errors from its nonblocking shutdown probe.
|
||||
if result.is_err() {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
|
||||
}
|
||||
result
|
||||
};
|
||||
tokio::select! {
|
||||
_ = console_signal => Ok(ShutdownSignal::Forceable),
|
||||
result = daemon_signal =>
|
||||
result.map(|_| ShutdownSignal::Forceable),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -508,6 +508,12 @@ pub(super) async fn spawn_websocket_server_with_args(
|
||||
.stderr(Stdio::piped())
|
||||
.env("CODEX_HOME", codex_home)
|
||||
.env("RUST_LOG", "warn");
|
||||
#[cfg(windows)]
|
||||
cmd.env(
|
||||
codex_app_server_transport::DAEMON_SHUTDOWN_FILE_ENV,
|
||||
codex_home.join("daemon.shutdown"),
|
||||
);
|
||||
|
||||
let mut process = cmd
|
||||
.kill_on_drop(true)
|
||||
.spawn()
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
//! Signal-driven shutdown tests; Windows uses the detached daemon control request
|
||||
//! for SIGTERM-equivalent graceful and forced shutdown coverage.
|
||||
|
||||
use super::connection_handling_websocket::DEFAULT_READ_TIMEOUT;
|
||||
use super::connection_handling_websocket::WsClient;
|
||||
use super::connection_handling_websocket::connect_websocket;
|
||||
@@ -19,6 +22,8 @@ use codex_app_server_protocol::UserInput as V2UserInput;
|
||||
use core_test_support::responses;
|
||||
use futures::SinkExt;
|
||||
use futures::StreamExt;
|
||||
use pretty_assertions::assert_eq;
|
||||
#[cfg(unix)]
|
||||
use std::process::Command as StdCommand;
|
||||
use tempfile::TempDir;
|
||||
use tokio::process::Child;
|
||||
@@ -31,6 +36,7 @@ use wiremock::Mock;
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::matchers::path_regex;
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn websocket_transport_ctrl_c_waits_for_running_turn_before_exit() -> Result<()> {
|
||||
let GracefulCtrlCFixture {
|
||||
@@ -56,6 +62,7 @@ async fn websocket_transport_ctrl_c_waits_for_running_turn_before_exit() -> Resu
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn websocket_transport_second_ctrl_c_forces_exit_while_turn_running() -> Result<()> {
|
||||
let GracefulCtrlCFixture {
|
||||
@@ -91,7 +98,7 @@ async fn websocket_transport_sigterm_waits_for_running_turn_before_exit() -> Res
|
||||
mut ws,
|
||||
} = start_ctrl_c_restart_fixture(Duration::from_secs(3)).await?;
|
||||
|
||||
send_sigterm(&process)?;
|
||||
send_sigterm(&process, _codex_home.path())?;
|
||||
assert_process_does_not_exit_within(&mut process, Duration::from_millis(300)).await?;
|
||||
|
||||
let status = wait_for_process_exit_within(
|
||||
@@ -116,10 +123,10 @@ async fn websocket_transport_second_sigterm_forces_exit_while_turn_running() ->
|
||||
mut ws,
|
||||
} = start_ctrl_c_restart_fixture(Duration::from_secs(3)).await?;
|
||||
|
||||
send_sigterm(&process)?;
|
||||
send_sigterm(&process, _codex_home.path())?;
|
||||
assert_process_does_not_exit_within(&mut process, Duration::from_millis(300)).await?;
|
||||
|
||||
send_sigterm(&process)?;
|
||||
send_sigterm(&process, _codex_home.path())?;
|
||||
let status = wait_for_process_exit_within(
|
||||
&mut process,
|
||||
Duration::from_secs(2),
|
||||
@@ -133,6 +140,7 @@ async fn websocket_transport_second_sigterm_forces_exit_while_turn_running() ->
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn websocket_transport_repeated_sighup_keeps_waiting_for_running_turn() -> Result<()> {
|
||||
let GracefulCtrlCFixture {
|
||||
@@ -257,18 +265,22 @@ async fn wait_for_responses_post(server: &wiremock::MockServer, wait_for: Durati
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn send_sigint(process: &Child) -> Result<()> {
|
||||
send_signal(process, "-INT")
|
||||
}
|
||||
|
||||
fn send_sigterm(process: &Child) -> Result<()> {
|
||||
#[cfg(unix)]
|
||||
fn send_sigterm(process: &Child, _home: &std::path::Path) -> Result<()> {
|
||||
send_signal(process, "-TERM")
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn send_sighup(process: &Child) -> Result<()> {
|
||||
send_signal(process, "-HUP")
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn send_signal(process: &Child, signal: &str) -> Result<()> {
|
||||
let pid = process
|
||||
.id()
|
||||
@@ -325,3 +337,12 @@ async fn expect_websocket_disconnect(stream: &mut WsClient) -> Result<()> {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(windows)]
|
||||
fn send_sigterm(process: &Child, home: &std::path::Path) -> Result<()> {
|
||||
std::fs::write(
|
||||
home.join("daemon.shutdown"),
|
||||
process.id().context("server pid")?.to_string(),
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -16,7 +16,7 @@ mod compaction;
|
||||
mod config_requirements_in_app_browser;
|
||||
mod config_rpc;
|
||||
mod connection_handling_websocket;
|
||||
#[cfg(unix)]
|
||||
#[cfg(any(unix, windows))]
|
||||
mod connection_handling_websocket_unix;
|
||||
#[cfg(unix)]
|
||||
mod curated_mcp_sync;
|
||||
|
||||
Reference in New Issue
Block a user