mirror of
https://github.com/openai/codex.git
synced 2026-11-05 20:10:33 +00:00
Keep exec-server shutdown separate from probe endpoints
This commit is contained in:
@@ -24,13 +24,6 @@ The CLI entrypoint supports:
|
||||
- `ws://IP:PORT` (default)
|
||||
- `--remote URL --executor-id ID [--name NAME]`
|
||||
|
||||
When running with `--listen ws://IP:PORT`, the same listener also serves basic
|
||||
HTTP probes:
|
||||
|
||||
- `GET /readyz` returns `200 OK` while the listener is accepting new websocket
|
||||
connections, then `503 Service Unavailable` once graceful shutdown begins.
|
||||
- `GET /healthz` returns `200 OK` while the listener is still able to serve HTTP.
|
||||
|
||||
Remote mode registers the local exec-server with the executor registry,
|
||||
then reconnects to the service-provided rendezvous websocket as the executor.
|
||||
It requires a bearer token in `CODEX_EXEC_SERVER_REMOTE_BEARER_TOKEN`.
|
||||
|
||||
@@ -1,3 +1,10 @@
|
||||
#[cfg(windows)]
|
||||
use std::process::Stdio;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Duration;
|
||||
|
||||
use axum::extract::ws::Message as AxumWebSocketMessage;
|
||||
use axum::extract::ws::WebSocket as AxumWebSocket;
|
||||
use codex_app_server_protocol::JSONRPCMessage;
|
||||
@@ -5,12 +12,6 @@ use futures::Sink;
|
||||
use futures::SinkExt;
|
||||
use futures::Stream;
|
||||
use futures::StreamExt;
|
||||
#[cfg(windows)]
|
||||
use std::process::Stdio;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Duration;
|
||||
use tokio::io::AsyncRead;
|
||||
use tokio::io::AsyncWrite;
|
||||
use tokio::process::Child;
|
||||
|
||||
@@ -141,7 +141,6 @@ async fn run_websocket_listener(
|
||||
let router = Router::new()
|
||||
.route("/", any(websocket_upgrade_handler))
|
||||
.route("/readyz", get(readiness_handler))
|
||||
.route("/healthz", get(health_check_handler))
|
||||
.with_state(ExecServerWebSocketState {
|
||||
processor: processor.clone(),
|
||||
shutdown_token: shutdown_token.clone(),
|
||||
@@ -173,15 +172,7 @@ struct ExecServerWebSocketState {
|
||||
shutdown_token: CancellationToken,
|
||||
}
|
||||
|
||||
async fn readiness_handler(State(state): State<ExecServerWebSocketState>) -> StatusCode {
|
||||
if state.shutdown_token.is_cancelled() {
|
||||
StatusCode::SERVICE_UNAVAILABLE
|
||||
} else {
|
||||
StatusCode::OK
|
||||
}
|
||||
}
|
||||
|
||||
async fn health_check_handler() -> StatusCode {
|
||||
async fn readiness_handler() -> StatusCode {
|
||||
StatusCode::OK
|
||||
}
|
||||
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
use std::net::SocketAddr;
|
||||
use std::time::Duration;
|
||||
|
||||
use axum::extract::State;
|
||||
use axum::http::StatusCode;
|
||||
use codex_app_server_protocol::JSONRPCMessage;
|
||||
use codex_app_server_protocol::JSONRPCNotification;
|
||||
use codex_app_server_protocol::JSONRPCRequest;
|
||||
@@ -17,17 +15,13 @@ use tokio::time::timeout;
|
||||
|
||||
use super::DEFAULT_LISTEN_URL;
|
||||
use super::ExecServerListenTransport;
|
||||
use super::ExecServerWebSocketState;
|
||||
use super::health_check_handler;
|
||||
use super::parse_listen_url;
|
||||
use super::readiness_handler;
|
||||
use super::run_stdio_connection_with_io;
|
||||
use crate::ExecServerRuntimePaths;
|
||||
use crate::protocol::INITIALIZE_METHOD;
|
||||
use crate::protocol::INITIALIZED_METHOD;
|
||||
use crate::protocol::InitializeParams;
|
||||
use crate::protocol::InitializeResponse;
|
||||
use crate::server::processor::ConnectionProcessor;
|
||||
|
||||
#[test]
|
||||
fn parse_listen_url_accepts_default_websocket_url() {
|
||||
@@ -131,31 +125,6 @@ fn parse_listen_url_accepts_websocket_url() {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn readiness_handler_reports_shutdown_state() {
|
||||
let shutdown_token = tokio_util::sync::CancellationToken::new();
|
||||
let state = ExecServerWebSocketState {
|
||||
processor: ConnectionProcessor::new(test_runtime_paths()),
|
||||
shutdown_token: shutdown_token.clone(),
|
||||
};
|
||||
|
||||
assert_eq!(
|
||||
readiness_handler(State(state.clone())).await,
|
||||
StatusCode::OK
|
||||
);
|
||||
|
||||
shutdown_token.cancel();
|
||||
assert_eq!(
|
||||
readiness_handler(State(state)).await,
|
||||
StatusCode::SERVICE_UNAVAILABLE
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn health_check_handler_stays_ok_during_shutdown() {
|
||||
assert_eq!(health_check_handler().await, StatusCode::OK);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_listen_url_rejects_invalid_websocket_url() {
|
||||
let err = parse_listen_url("ws://localhost:1234")
|
||||
|
||||
@@ -6,21 +6,15 @@ use common::exec_server::exec_server;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn exec_server_serves_kubernetes_style_probe_routes() -> anyhow::Result<()> {
|
||||
async fn exec_server_serves_readyz_alongside_websocket_endpoint() -> anyhow::Result<()> {
|
||||
let mut server = exec_server().await?;
|
||||
let http_base_url = server
|
||||
.websocket_url()
|
||||
.strip_prefix("ws://")
|
||||
.expect("websocket URL should use ws://");
|
||||
|
||||
let readyz = reqwest::get(format!("http://{http_base_url}/readyz")).await?;
|
||||
assert_eq!(readyz.status(), reqwest::StatusCode::OK);
|
||||
|
||||
let healthz = reqwest::get(format!("http://{http_base_url}/healthz")).await?;
|
||||
assert_eq!(healthz.status(), reqwest::StatusCode::OK);
|
||||
|
||||
let legacy_health = reqwest::get(format!("http://{http_base_url}/health")).await?;
|
||||
assert_eq!(legacy_health.status(), reqwest::StatusCode::NOT_FOUND);
|
||||
let response = reqwest::get(format!("http://{http_base_url}/readyz")).await?;
|
||||
assert_eq!(response.status(), reqwest::StatusCode::OK);
|
||||
|
||||
server.shutdown().await?;
|
||||
Ok(())
|
||||
|
||||
Reference in New Issue
Block a user