mirror of
https://github.com/openai/codex.git
synced 2026-10-21 17:48:44 +00:00
Bound managed app-server and updater stderr logging (#55856)
## Why Managed stderr logs previously retained a tail only before a new launch, allowing logs to grow during a long-running process. Direct stderr writes could also block lifecycle operations on disk I/O. ## What changed - Route managed app-server and updater logs through a shared nonblocking worker with a bounded queue. Drop records when the queue is full or a record exceeds 16 KiB, and avoid waiting for the worker at shutdown. - Prune live logs when they exceed 1 MiB, preserving the latest 256 KiB in `.log.previous` before truncating the current file. Keep inherited stderr handles usable after truncation. - Pause predecessor pruning during Windows updater handoff and resume it after a failed handoff only when ownership has been restored. - Keep ordinary stderr output when managed logging has not been initialized. ## Testing Add regression tests for stalled workers and queue overflow, tail retention, inherited append and nonappend handles, and pruning exclusion during updater handoff. GitOrigin-RevId: b6301962b1180fa6cdb231a56edce2429dc20b54
This commit is contained in:
Generated
+3
@@ -2077,6 +2077,7 @@ dependencies = [
|
||||
"clap",
|
||||
"codex-agent-extension",
|
||||
"codex-analytics",
|
||||
"codex-app-server-daemon",
|
||||
"codex-app-server-protocol",
|
||||
"codex-app-server-transport",
|
||||
"codex-arg0",
|
||||
@@ -2233,6 +2234,8 @@ dependencies = [
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tracing",
|
||||
"tracing-appender",
|
||||
"tracing-subscriber",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
|
||||
@@ -40,10 +40,12 @@ tokio = { workspace = true, features = [
|
||||
tempfile = { workspace = true }
|
||||
tokio-tungstenite = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
tracing-appender = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
codex-utils-cargo-bin = { workspace = true }
|
||||
pretty_assertions = { workspace = true }
|
||||
tracing-subscriber = { workspace = true }
|
||||
|
||||
[target.'cfg(windows)'.dependencies]
|
||||
windows-sys = { workspace = true, features = ["Win32_Foundation", "Win32_Storage_FileSystem", "Win32_System_IO", "Win32_System_JobObjects", "Win32_Security", "Win32_Security_Authorization", "Win32_System_Threading"] }
|
||||
windows-sys = { workspace = true, features = ["Win32_Foundation", "Win32_Storage_FileSystem", "Win32_System_Console", "Win32_System_IO", "Win32_System_JobObjects", "Win32_Security", "Win32_Security_Authorization", "Win32_System_Threading"] }
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
mod pid;
|
||||
mod stderr_log;
|
||||
pub(crate) mod stderr_log;
|
||||
#[cfg(windows)]
|
||||
pub(crate) mod windows;
|
||||
|
||||
|
||||
@@ -38,14 +38,33 @@ impl PidBackend {
|
||||
.context("failed to acknowledge updater startup")
|
||||
}
|
||||
|
||||
// Keep pruning excluded until successor startup and any ownership rollback finish.
|
||||
#[allow(clippy::await_holding_invalid_type)]
|
||||
pub(crate) async fn replace_current_updater(&self) -> Result<()> {
|
||||
// Retry busy pruning instead of waiting on the logging worker. Keep it paused
|
||||
// before open_stderr_log() truncates the shared log and throughout startup.
|
||||
let mut pruning_enabled = super::super::stderr_log::PRUNING_ENABLED
|
||||
.try_lock()
|
||||
.context("stderr pruning is busy; retry updater handoff")?;
|
||||
let record = PidRecord {
|
||||
pid: std::process::id(),
|
||||
process_start_time: read_process_start_time(std::process::id()).await?,
|
||||
process_identity: None,
|
||||
executable_identity: None,
|
||||
};
|
||||
self.start_inner(Some(record)).await?;
|
||||
*pruning_enabled = false;
|
||||
let result = self.start_inner(Some(record.clone())).await;
|
||||
// Resume only after ownership is restored. Failed cleanup may leave the
|
||||
// successor alive, and successful handoff leaves it in charge of retention.
|
||||
if result.is_err()
|
||||
&& matches!(
|
||||
self.read_pid_file_state().await,
|
||||
Ok(PidFileState::Running(previous)) if previous == record
|
||||
)
|
||||
{
|
||||
*pruning_enabled = true;
|
||||
}
|
||||
result?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -1,53 +1,191 @@
|
||||
//! Saves a bounded previous-launch stderr tail before a detached launch truncates its log.
|
||||
//! Preservation is best effort; missing or empty logs leave the last useful tail intact.
|
||||
//! Bounds managed stderr with the shared nonblocking logging worker and tail retention.
|
||||
//! The worker serializes writes with pruning and pauses pruning during Windows handoff.
|
||||
//! Native writers remain best effort.
|
||||
|
||||
use std::io;
|
||||
use std::io::Read;
|
||||
use std::io::Seek;
|
||||
use std::io::Write;
|
||||
use std::path::Path;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::OnceLock;
|
||||
use std::time::Instant;
|
||||
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tokio::io::AsyncSeekExt;
|
||||
use tokio::sync::Mutex;
|
||||
use tracing_appender::non_blocking::NonBlocking;
|
||||
use tracing_appender::non_blocking::NonBlockingBuilder;
|
||||
use tracing_appender::non_blocking::WorkerGuard;
|
||||
|
||||
const MAX_PREVIOUS_LOG_BYTES: u64 = 256 * 1024;
|
||||
const MAX_CURRENT_LOG_BYTES: u64 = 1024 * 1024;
|
||||
|
||||
const QUEUED_RECORDS: usize = 64;
|
||||
const MAX_RECORD_BYTES: usize = 16 * 1024;
|
||||
// Process lifetime: dropping WorkerGuard would wait for a potentially stalled disk.
|
||||
static LOGGER: OnceLock<(NonBlocking, WorkerGuard)> = OnceLock::new();
|
||||
// Handoff holds this lock through startup and leaves predecessor pruning disabled on success.
|
||||
// Both callers use try_lock so a stalled worker cannot make handoff wait on disk.
|
||||
pub(super) static PRUNING_ENABLED: Mutex<bool> = Mutex::const_new(true);
|
||||
|
||||
pub(crate) fn start(path: PathBuf) {
|
||||
LOGGER.get_or_init(|| {
|
||||
nonblocking(LogFile {
|
||||
path,
|
||||
file: None,
|
||||
can_prune: None,
|
||||
})
|
||||
});
|
||||
}
|
||||
|
||||
fn nonblocking(writer: impl Write + Send + 'static) -> (NonBlocking, WorkerGuard) {
|
||||
NonBlockingBuilder::default()
|
||||
.buffered_lines_limit(QUEUED_RECORDS)
|
||||
.lossy(/* is_lossy */ true)
|
||||
.finish(writer)
|
||||
}
|
||||
|
||||
/// Writes complete records through the bounded managed queue, or ordinary stderr otherwise.
|
||||
/// Each write must contain one complete record; oversized records are discarded before copying.
|
||||
pub fn stderr_writer() -> impl Write {
|
||||
StderrWriter
|
||||
}
|
||||
|
||||
struct StderrWriter;
|
||||
|
||||
impl Write for StderrWriter {
|
||||
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
|
||||
if let Some((writer, _guard)) = LOGGER.get() {
|
||||
if bytes.len() <= MAX_RECORD_BYTES {
|
||||
writer.clone().write_all(bytes)?;
|
||||
}
|
||||
Ok(bytes.len())
|
||||
} else {
|
||||
io::stderr().write(bytes)
|
||||
}
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
if LOGGER.get().is_none() {
|
||||
io::stderr().flush()?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
struct LogFile {
|
||||
path: PathBuf,
|
||||
file: Option<std::fs::File>,
|
||||
can_prune: Option<bool>,
|
||||
}
|
||||
|
||||
impl Write for LogFile {
|
||||
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
|
||||
// All filesystem work, including first open and legacy handle setup, stays on the worker.
|
||||
if *self
|
||||
.can_prune
|
||||
.get_or_insert_with(|| enable_append().is_ok())
|
||||
{
|
||||
let _ = prune(&self.path, &PRUNING_ENABLED);
|
||||
}
|
||||
let file = match self.file {
|
||||
Some(ref mut file) => file,
|
||||
None => self.file.insert(
|
||||
std::fs::OpenOptions::new()
|
||||
.create(true)
|
||||
.append(true)
|
||||
.open(&self.path)?,
|
||||
),
|
||||
};
|
||||
file.write_all(bytes)?;
|
||||
Ok(bytes.len())
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn enable_append() -> io::Result<()> {
|
||||
use std::os::fd::AsRawFd;
|
||||
|
||||
let fd = std::io::stderr().as_raw_fd();
|
||||
// SAFETY: F_GETFL reads flags from the borrowed descriptor and takes no pointers.
|
||||
let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) };
|
||||
if flags == -1 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
// SAFETY: F_SETFL changes status flags on the same descriptor without taking ownership.
|
||||
if unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_APPEND) } == -1 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(windows)]
|
||||
fn enable_append() -> io::Result<()> {
|
||||
use std::os::windows::io::AsRawHandle;
|
||||
use std::os::windows::io::FromRawHandle;
|
||||
use std::os::windows::io::IntoRawHandle;
|
||||
use std::os::windows::io::OwnedHandle;
|
||||
use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE;
|
||||
use windows_sys::Win32::Storage::FileSystem::FILE_GENERIC_WRITE;
|
||||
use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_DELETE;
|
||||
use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_READ;
|
||||
use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_WRITE;
|
||||
use windows_sys::Win32::Storage::FileSystem::FILE_WRITE_DATA;
|
||||
use windows_sys::Win32::Storage::FileSystem::ReOpenFile;
|
||||
use windows_sys::Win32::System::Console::STD_ERROR_HANDLE;
|
||||
use windows_sys::Win32::System::Console::SetStdHandle;
|
||||
|
||||
let stderr = std::io::stderr();
|
||||
// SAFETY: stderr is borrowed and remains open; reopening does not truncate it.
|
||||
let handle = unsafe {
|
||||
ReOpenFile(
|
||||
stderr.as_raw_handle(),
|
||||
// Append-only access makes writes follow EOF after truncation.
|
||||
FILE_GENERIC_WRITE & !FILE_WRITE_DATA,
|
||||
FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
|
||||
/*dwflagsandattributes*/ 0,
|
||||
)
|
||||
};
|
||||
if handle == INVALID_HANDLE_VALUE {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
// SAFETY: ReOpenFile returned a new, valid handle owned by this function.
|
||||
let handle = unsafe { OwnedHandle::from_raw_handle(handle) };
|
||||
// SAFETY: the append handle is valid and retained for the process lifetime on success.
|
||||
if unsafe { SetStdHandle(STD_ERROR_HANDLE, handle.as_raw_handle()) } == 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
// The process owns its new stderr. Keep the borrowed original valid for cached users.
|
||||
let _ = handle.into_raw_handle();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn prune(path: &Path, pruning_enabled: &Mutex<bool>) -> io::Result<()> {
|
||||
let Ok(enabled) = pruning_enabled.try_lock() else {
|
||||
return Ok(());
|
||||
};
|
||||
if !*enabled {
|
||||
return Ok(());
|
||||
}
|
||||
let metadata = std::fs::symlink_metadata(path)?;
|
||||
if !metadata.is_file() || metadata.len() <= MAX_CURRENT_LOG_BYTES {
|
||||
return Ok(());
|
||||
}
|
||||
preserve_tail(path)?;
|
||||
// Keep inherited handles connected, holding the pruning lock through truncation.
|
||||
std::fs::OpenOptions::new()
|
||||
.write(true)
|
||||
.open(path)?
|
||||
.set_len(0)
|
||||
}
|
||||
|
||||
pub(super) async fn preserve(path: &Path) {
|
||||
let started = Instant::now();
|
||||
let path = path.to_path_buf();
|
||||
let result = async {
|
||||
match tokio::fs::symlink_metadata(path).await {
|
||||
Ok(metadata) if !metadata.is_file() => return Ok(()),
|
||||
Ok(_) => {}
|
||||
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
|
||||
Err(error) => return Err(error.into()),
|
||||
}
|
||||
let mut file = match tokio::fs::File::open(path).await {
|
||||
Ok(file) => file,
|
||||
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
|
||||
Err(error) => return Err(error.into()),
|
||||
};
|
||||
let metadata = file.metadata().await?;
|
||||
let len = metadata.len();
|
||||
if !metadata.is_file() || len == 0 {
|
||||
return Ok(());
|
||||
}
|
||||
file.seek(io::SeekFrom::Start(
|
||||
len.saturating_sub(MAX_PREVIOUS_LOG_BYTES),
|
||||
))
|
||||
.await?;
|
||||
let mut bytes = Vec::new();
|
||||
file.take(MAX_PREVIOUS_LOG_BYTES)
|
||||
.read_to_end(&mut bytes)
|
||||
.await?;
|
||||
let previous = path.with_extension("log.previous");
|
||||
let temporary = path.with_extension("log.previous.tmp");
|
||||
let result = async {
|
||||
tokio::fs::write(&temporary, bytes).await?;
|
||||
tokio::fs::rename(&temporary, previous).await
|
||||
}
|
||||
.await;
|
||||
if result.is_err() {
|
||||
let _ = tokio::fs::remove_file(&temporary).await;
|
||||
}
|
||||
result?;
|
||||
tokio::task::spawn_blocking(move || preserve_tail(&path)).await??;
|
||||
anyhow::Ok(())
|
||||
}
|
||||
.await;
|
||||
@@ -56,6 +194,38 @@ pub(super) async fn preserve(path: &Path) {
|
||||
}
|
||||
}
|
||||
|
||||
fn preserve_tail(path: &Path) -> io::Result<()> {
|
||||
match std::fs::symlink_metadata(path) {
|
||||
Ok(metadata) if !metadata.is_file() => return Ok(()),
|
||||
Ok(_) => {}
|
||||
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
|
||||
Err(error) => return Err(error),
|
||||
}
|
||||
let mut file = match std::fs::File::open(path) {
|
||||
Ok(file) => file,
|
||||
Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
|
||||
Err(error) => return Err(error),
|
||||
};
|
||||
let metadata = file.metadata()?;
|
||||
let len = metadata.len();
|
||||
if !metadata.is_file() || len == 0 {
|
||||
return Ok(());
|
||||
}
|
||||
file.seek(io::SeekFrom::Start(
|
||||
len.saturating_sub(MAX_PREVIOUS_LOG_BYTES),
|
||||
))?;
|
||||
let mut bytes = Vec::new();
|
||||
file.take(MAX_PREVIOUS_LOG_BYTES).read_to_end(&mut bytes)?;
|
||||
let previous = path.with_extension("log.previous");
|
||||
let temporary = path.with_extension("log.previous.tmp");
|
||||
let result =
|
||||
std::fs::write(&temporary, bytes).and_then(|()| std::fs::rename(&temporary, previous));
|
||||
if result.is_err() {
|
||||
let _ = std::fs::remove_file(&temporary);
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "stderr_log_tests.rs"]
|
||||
mod tests;
|
||||
|
||||
@@ -1,7 +1,51 @@
|
||||
use super::MAX_CURRENT_LOG_BYTES;
|
||||
use super::MAX_PREVIOUS_LOG_BYTES;
|
||||
use super::preserve;
|
||||
use super::prune;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[cfg(any(unix, windows))]
|
||||
#[tokio::test]
|
||||
async fn pruning_resets_the_write_position_of_a_legacy_nonappend_writer() -> anyhow::Result<()> {
|
||||
use std::io::Write;
|
||||
|
||||
const CHILD_LOG: &str = "CODEX_TEST_LEGACY_STDERR_LOG";
|
||||
if let Some(path) = std::env::var_os(CHILD_LOG) {
|
||||
std::io::stderr().write_all(b"before pruning\n")?;
|
||||
let mut writer = super::LogFile {
|
||||
path: path.into(),
|
||||
file: None,
|
||||
can_prune: None,
|
||||
};
|
||||
writer.write_all(b"queued logging\n")?;
|
||||
std::io::stderr().write_all(b"still logging\n")?;
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let home = tempfile::tempdir()?;
|
||||
let path = home.path().join("daemon.stderr.log");
|
||||
let mut writer = std::fs::File::create(&path)?;
|
||||
writer.write_all(&vec![b'x'; MAX_CURRENT_LOG_BYTES as usize + 1])?;
|
||||
// Isolate the process-wide stderr replacement and inherit the legacy writer's offset.
|
||||
let status = tokio::process::Command::new(std::env::current_exe()?)
|
||||
.args([
|
||||
"--exact",
|
||||
"backend::stderr_log::tests::pruning_resets_the_write_position_of_a_legacy_nonappend_writer",
|
||||
"--nocapture",
|
||||
])
|
||||
.env(CHILD_LOG, &path)
|
||||
.stderr(writer)
|
||||
.kill_on_drop(true)
|
||||
.status()
|
||||
.await?;
|
||||
assert!(status.success(), "legacy stderr child failed: {status}");
|
||||
assert_eq!(
|
||||
tokio::fs::read(&path).await?,
|
||||
b"queued logging\nstill logging\n"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn preserves_bounded_tail_and_keeps_it_after_empty_launch() -> anyhow::Result<()> {
|
||||
let home = tempfile::tempdir()?;
|
||||
@@ -28,3 +72,146 @@ async fn preserves_bounded_tail_and_keeps_it_after_empty_launch() -> anyhow::Res
|
||||
assert_eq!(tokio::fs::read(&previous).await?, b"next launch");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[allow(clippy::await_holding_invalid_type)] // Exercise pruning exclusion during handoff I/O.
|
||||
async fn prunes_live_log_without_disconnecting_inherited_append_handles() -> anyhow::Result<()> {
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
let pruning_enabled = tokio::sync::Mutex::new(true);
|
||||
let home = tempfile::tempdir()?;
|
||||
let path = home.path().join("daemon.stderr.log");
|
||||
let mut writer = tokio::fs::OpenOptions::new()
|
||||
.create(true)
|
||||
.append(true)
|
||||
.open(&path)
|
||||
.await?;
|
||||
writer.write_all(b"startup\n").await?;
|
||||
writer.flush().await?;
|
||||
prune(&path, &pruning_enabled)?;
|
||||
assert_eq!(tokio::fs::read(&path).await?, b"startup\n");
|
||||
|
||||
let mut bytes = vec![b'x'; MAX_CURRENT_LOG_BYTES as usize];
|
||||
bytes.extend_from_slice(b"latest failure\n");
|
||||
writer.write_all(&bytes).await?;
|
||||
writer.flush().await?;
|
||||
|
||||
// A failed tail publication must leave the live log untouched.
|
||||
let previous = path.with_extension("log.previous");
|
||||
tokio::fs::create_dir(&previous).await?;
|
||||
assert!(prune(&path, &pruning_enabled).is_err());
|
||||
assert_eq!(
|
||||
tokio::fs::read(&path).await?,
|
||||
[b"startup\n".as_slice(), bytes.as_slice()].concat()
|
||||
);
|
||||
tokio::fs::remove_dir(&previous).await?;
|
||||
|
||||
// Handoff excludes pruning even while the shared log exceeds the limit.
|
||||
let mut handoff = pruning_enabled.try_lock()?;
|
||||
*handoff = false;
|
||||
prune(&path, &pruning_enabled)?;
|
||||
assert_eq!(
|
||||
tokio::fs::read(&path).await?,
|
||||
[b"startup\n".as_slice(), bytes.as_slice()].concat()
|
||||
);
|
||||
|
||||
// Failed handoff restores ownership, so retention resumes after unlocking.
|
||||
*handoff = true;
|
||||
drop(handoff);
|
||||
prune(&path, &pruning_enabled)?;
|
||||
assert_eq!(
|
||||
tokio::fs::read(previous).await?,
|
||||
bytes[bytes.len() - MAX_PREVIOUS_LOG_BYTES as usize..]
|
||||
);
|
||||
writer.write_all(b"still logging\n").await?;
|
||||
writer.flush().await?;
|
||||
assert_eq!(tokio::fs::read(&path).await?, b"still logging\n");
|
||||
|
||||
// Successful handoff keeps predecessor pruning disabled after unlocking.
|
||||
let mut handoff = pruning_enabled.try_lock()?;
|
||||
*handoff = false;
|
||||
writer.write_all(&bytes).await?;
|
||||
writer.write_all(b"successor startup\n").await?;
|
||||
writer.flush().await?;
|
||||
drop(handoff);
|
||||
prune(&path, &pruning_enabled)?;
|
||||
assert_eq!(
|
||||
tokio::fs::read(&path).await?,
|
||||
[
|
||||
b"still logging\n".as_slice(),
|
||||
bytes.as_slice(),
|
||||
b"successor startup\n",
|
||||
]
|
||||
.concat()
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn logging_does_not_wait_for_a_stalled_worker() -> anyhow::Result<()> {
|
||||
use std::io::Write;
|
||||
use std::sync::Barrier;
|
||||
use std::sync::mpsc;
|
||||
|
||||
const CHILD: &str = "CODEX_TEST_STALLED_LOG_WORKER";
|
||||
if std::env::var_os(CHILD).is_none() {
|
||||
let output = std::process::Command::new(std::env::current_exe()?)
|
||||
.args([
|
||||
"--exact",
|
||||
"backend::stderr_log::tests::logging_does_not_wait_for_a_stalled_worker",
|
||||
])
|
||||
.env(CHILD, "1")
|
||||
.output()?;
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"logging child failed: {}",
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
assert_eq!(output.stderr, Vec::<u8>::new());
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
struct StalledWriter(mpsc::Sender<()>);
|
||||
impl Write for StalledWriter {
|
||||
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
|
||||
self.0.send(()).unwrap();
|
||||
// No second participant: the worker stays stalled until process exit.
|
||||
Barrier::new(/* n */ 2).wait();
|
||||
Ok(bytes.len())
|
||||
}
|
||||
fn flush(&mut self) -> std::io::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
let (entered_tx, entered_rx) = mpsc::channel();
|
||||
assert!(
|
||||
super::LOGGER
|
||||
.set(super::nonblocking(StalledWriter(entered_tx)))
|
||||
.is_ok()
|
||||
);
|
||||
// The updater's diagnostic starts the stalled write.
|
||||
crate::diagnostics::event("stalled-worker", serde_json::json!({}));
|
||||
entered_rx.recv()?;
|
||||
// Exercise the same tracing formatter/writer combination used by app-server.
|
||||
let subscriber = tracing_subscriber::fmt()
|
||||
.with_writer(super::stderr_writer)
|
||||
.finish();
|
||||
tracing::subscriber::with_default(subscriber, || tracing::warn!("app-server event"));
|
||||
super::stderr_writer().write_all(&vec![b'x'; super::MAX_RECORD_BYTES + 1])?;
|
||||
for _ in 1..super::QUEUED_RECORDS {
|
||||
super::stderr_writer().write_all(&vec![b'x'; super::MAX_RECORD_BYTES])?;
|
||||
}
|
||||
super::stderr_writer().write_all(b"queue is full\n")?;
|
||||
super::stderr_writer().flush()?;
|
||||
assert_eq!(
|
||||
super::LOGGER
|
||||
.get()
|
||||
.unwrap()
|
||||
.0
|
||||
.error_counter()
|
||||
.dropped_lines(),
|
||||
1
|
||||
);
|
||||
// Returning also proves process shutdown does not wait for the stalled worker.
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -100,7 +100,7 @@ fn emit(payload: Payload<impl Serialize>) {
|
||||
process_pid: std::process::id(),
|
||||
payload,
|
||||
};
|
||||
let Ok(record) = serde_json::to_string(&record) else {
|
||||
let Ok(mut record) = serde_json::to_string(&record) else {
|
||||
return;
|
||||
};
|
||||
if tracing::dispatcher::has_been_set() {
|
||||
@@ -114,7 +114,8 @@ fn emit(payload: Payload<impl Serialize>) {
|
||||
// The detached updater has redirected stderr and no tracing subscriber.
|
||||
// A TUI can own a live terminal before installing its subscriber.
|
||||
// Never let a log write failure interrupt the lifecycle operation.
|
||||
let _ = writeln!(std::io::stderr().lock(), "{record}");
|
||||
record.push('\n');
|
||||
let _ = crate::stderr_writer().write_all(record.as_bytes());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -26,6 +26,7 @@ mod thread_recovery;
|
||||
mod update_loop;
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::io::Write;
|
||||
use std::path::Path;
|
||||
use std::path::PathBuf;
|
||||
use std::time::Duration;
|
||||
@@ -298,6 +299,20 @@ pub async fn run_pid_update_loop(
|
||||
update_loop::run(http_client_factory, restore_release).await
|
||||
}
|
||||
|
||||
pub use backend::stderr_log::stderr_writer;
|
||||
|
||||
/// Routes managed app-server logs through a bounded queue to a dedicated file worker.
|
||||
/// Full queues and oversized records are dropped; shutdown never waits for disk I/O.
|
||||
/// Native and other-process stderr writes remain best effort during pruning.
|
||||
pub fn maintain_stderr_log() -> Result<()> {
|
||||
backend::stderr_log::start(
|
||||
Daemon::from_environment()?
|
||||
.pid_file
|
||||
.with_extension("stderr.log"),
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn update(
|
||||
http_client_factory: codex_http_client::HttpClientFactory,
|
||||
) -> Result<UpdateOutput> {
|
||||
@@ -575,9 +590,9 @@ impl Daemon {
|
||||
#[cfg(windows)]
|
||||
backend::windows::ensure_detached_launch(managed_codex_bin)?;
|
||||
if let Err(err) = thread_recovery::discard_pending(self) {
|
||||
eprintln!(
|
||||
"warning: failed to clear stale daemon recovery before update: {err}"
|
||||
);
|
||||
let _ = stderr_writer().write_all(format!(
|
||||
"warning: failed to clear stale daemon recovery before update: {err}\n"
|
||||
).as_bytes());
|
||||
}
|
||||
diagnostics::event(
|
||||
"restart_requested",
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
//! Owns scheduled and manual installs, daemon restarts, and updater replacement.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::io::Write;
|
||||
use std::path::Path;
|
||||
#[cfg(unix)]
|
||||
use std::process::Command as StdCommand;
|
||||
@@ -79,13 +80,15 @@ pub(crate) async fn run(
|
||||
http_client_factory: HttpClientFactory,
|
||||
restore_release: Option<String>,
|
||||
) -> Result<()> {
|
||||
let daemon = Daemon::from_environment()?;
|
||||
crate::backend::stderr_log::start(daemon.update_pid_file.with_extension("stderr.log"));
|
||||
let http = RouteAwareClientPool::new_without_request_logging(
|
||||
http_client_factory,
|
||||
ClientRouteClass::Other,
|
||||
);
|
||||
run_with_http(
|
||||
&http,
|
||||
&Daemon::from_environment()?,
|
||||
&daemon,
|
||||
¤t_updater_identity().await?,
|
||||
restore_release,
|
||||
)
|
||||
@@ -228,7 +231,7 @@ async fn run_with_http(
|
||||
if listener.is_none() {
|
||||
return Err(err);
|
||||
}
|
||||
eprintln!("warning: failed to refresh managed updater: {err:#}");
|
||||
let _ = crate::stderr_writer().write_all(format!("warning: failed to refresh managed updater: {err:#}\n").as_bytes());
|
||||
next_check = Instant::now() + Duration::from_secs(30);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -35,6 +35,7 @@ axum = { workspace = true, default-features = false, features = [
|
||||
"ws",
|
||||
] }
|
||||
codex-analytics = { workspace = true }
|
||||
codex-app-server-daemon = { workspace = true }
|
||||
codex-agent-extension = { workspace = true }
|
||||
codex-arg0 = { workspace = true }
|
||||
codex-aws-auth = { workspace = true }
|
||||
|
||||
@@ -17,6 +17,7 @@ use std::collections::HashMap;
|
||||
use std::collections::HashSet;
|
||||
use std::io::ErrorKind;
|
||||
use std::io::Result as IoResult;
|
||||
use std::io::Write;
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
use std::sync::RwLock;
|
||||
@@ -733,12 +734,12 @@ pub async fn run_main_with_transport_options(
|
||||
let stderr_fmt: StderrLogLayer = match log_format_from_env() {
|
||||
LogFormat::Json => tracing_subscriber::fmt::layer()
|
||||
.json()
|
||||
.with_writer(std::io::stderr)
|
||||
.with_writer(codex_app_server_daemon::stderr_writer)
|
||||
.with_span_events(stderr_span_events())
|
||||
.with_filter(EnvFilter::from_default_env())
|
||||
.boxed(),
|
||||
LogFormat::Default => tracing_subscriber::fmt::layer()
|
||||
.with_writer(std::io::stderr)
|
||||
.with_writer(codex_app_server_daemon::stderr_writer)
|
||||
.with_span_events(stderr_span_events())
|
||||
.with_filter(EnvFilter::from_default_env())
|
||||
.boxed(),
|
||||
@@ -1521,10 +1522,8 @@ fn sqlite_recovery_notice(
|
||||
fn emit_state_db_backup_warning(message: &str) {
|
||||
warn!("{message}");
|
||||
if !tracing::dispatcher::has_been_set() {
|
||||
#[allow(clippy::print_stderr)]
|
||||
{
|
||||
eprintln!("{message}");
|
||||
}
|
||||
let _ =
|
||||
codex_app_server_daemon::stderr_writer().write_all(format!("{message}\n").as_bytes());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1285,6 +1285,14 @@ async fn cli_main(
|
||||
listen
|
||||
};
|
||||
let auth = auth.try_into_settings()?;
|
||||
if managed_daemon
|
||||
&& matches!(
|
||||
&transport,
|
||||
codex_app_server::AppServerTransport::UnixSocket { .. }
|
||||
)
|
||||
{
|
||||
let _ = codex_app_server_daemon::maintain_stderr_log();
|
||||
}
|
||||
let runtime_options = codex_app_server::AppServerRuntimeOptions {
|
||||
cloud_skills_snapshot,
|
||||
code_mode_host_transport: code_mode_host.into(),
|
||||
|
||||
Reference in New Issue
Block a user