mirror of
https://github.com/openai/codex.git
synced 2026-10-18 17:18:52 +00:00
Avoid SQLite stalls from connection setup and stderr span logging (#49032)
## Why Setting `auto_vacuum` on every connection can wait for a writer lock even when the mode is unchanged. Logging every span entry and exit can also block SQLx workers on undrained stderr while they hold a write transaction. ## What changed - Set incremental vacuum only for empty databases or databases using full auto-vacuum, before enabling WAL. Preserve existing nonempty databases with auto-vacuum disabled. - Emit span creation and closure events in app-server text and JSON stderr logs, preserving busy/idle timings and explicit events while omitting span entry and exit records. ## Testing Add regression tests for opening and expanding writable pools while another connection holds a writer lock, and for transaction completion with undrained stderr in both log formats. Verify vacuum modes and retained log diagnostics. GitOrigin-RevId: 40d4ef32c7efc49f4240b987277ba45551259335
This commit is contained in:
@@ -78,6 +78,7 @@ use tracing::info;
|
||||
use tracing::warn;
|
||||
use tracing_subscriber::EnvFilter;
|
||||
use tracing_subscriber::Layer;
|
||||
use tracing_subscriber::fmt::format::FmtSpan;
|
||||
use tracing_subscriber::layer::SubscriberExt;
|
||||
use tracing_subscriber::registry::Registry;
|
||||
use tracing_subscriber::util::SubscriberInitExt;
|
||||
@@ -164,6 +165,10 @@ enum LogFormat {
|
||||
|
||||
type StderrLogLayer = Box<dyn Layer<Registry> + Send + Sync + 'static>;
|
||||
|
||||
fn stderr_span_events() -> FmtSpan {
|
||||
FmtSpan::NEW | FmtSpan::CLOSE
|
||||
}
|
||||
|
||||
/// Control-plane messages from the processor/transport side to the outbound router task.
|
||||
///
|
||||
/// `run_main_with_transport_options` uses two loops/tasks:
|
||||
@@ -703,16 +708,19 @@ pub async fn run_main_with_transport_options(
|
||||
// Install a simple subscriber so `tracing` output is visible. Users can
|
||||
// control the log level with `RUST_LOG` and switch to JSON logs with
|
||||
// `LOG_FORMAT=json`.
|
||||
// SQLx enters the caller's span for each command. Skip enter/exit records
|
||||
// that can block its worker on stderr while holding a write transaction.
|
||||
// Preserve span boundaries, busy/idle timings, and explicit events.
|
||||
let stderr_fmt: StderrLogLayer = match log_format_from_env() {
|
||||
LogFormat::Json => tracing_subscriber::fmt::layer()
|
||||
.json()
|
||||
.with_writer(std::io::stderr)
|
||||
.with_span_events(tracing_subscriber::fmt::format::FmtSpan::FULL)
|
||||
.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_span_events(tracing_subscriber::fmt::format::FmtSpan::FULL)
|
||||
.with_span_events(stderr_span_events())
|
||||
.with_filter(EnvFilter::from_default_env())
|
||||
.boxed(),
|
||||
};
|
||||
@@ -1524,6 +1532,10 @@ fn analytics_rpc_transport(transport: &AppServerTransport) -> AppServerRpcTransp
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "stderr_logging_tests.rs"]
|
||||
mod stderr_logging_tests;
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::LogFormat;
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
//! Regressions for SQLite progress and stderr span diagnostics.
|
||||
|
||||
use crate::LogFormat;
|
||||
use crate::StderrLogLayer;
|
||||
use crate::stderr_span_events;
|
||||
use anyhow::Context;
|
||||
use anyhow::Result;
|
||||
use codex_state::SqliteConfig;
|
||||
use codex_utils_absolute_path::test_support::PathExt;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::io::Read;
|
||||
use std::sync::Mutex;
|
||||
use std::time::Duration;
|
||||
use tempfile::TempDir;
|
||||
use test_case::test_case;
|
||||
use tokio::sync::oneshot;
|
||||
use tokio::time::timeout;
|
||||
use tracing_subscriber::EnvFilter;
|
||||
use tracing_subscriber::Layer;
|
||||
use tracing_subscriber::layer::SubscriberExt;
|
||||
|
||||
#[test_case(LogFormat::Json; "json")]
|
||||
#[test_case(LogFormat::Default; "text")]
|
||||
#[tokio::test]
|
||||
async fn sqlite_transaction_completes_with_undrained_stderr(format: LogFormat) -> Result<()> {
|
||||
let dir = TempDir::new()?;
|
||||
let config = SqliteConfig::new_for_testing(dir.path().abs());
|
||||
let path = dir.path().join("logging.sqlite");
|
||||
let writer_pool = config.open_read_write_pool(&path).await?;
|
||||
sqlx::query("CREATE TABLE writes (id INTEGER PRIMARY KEY, value INTEGER NOT NULL)")
|
||||
.execute(&writer_pool)
|
||||
.await?;
|
||||
// Open both pools before tracing so this isolates logging during a transaction.
|
||||
let contender_pool = config.open_read_write_pool(&path).await?;
|
||||
let (mut reader, writer) = std::io::pipe()?;
|
||||
let fmt = tracing_subscriber::fmt::layer()
|
||||
.with_writer(Mutex::new(writer))
|
||||
.with_span_events(stderr_span_events());
|
||||
let filter = EnvFilter::new("off,codex_app_server::stderr_logging_tests=info");
|
||||
let fmt: StderrLogLayer = match format {
|
||||
LogFormat::Json => fmt.json().with_filter(filter).boxed(),
|
||||
LogFormat::Default => fmt.with_filter(filter).boxed(),
|
||||
};
|
||||
let subscriber = tracing_subscriber::registry().with(fmt);
|
||||
let (locked_tx, locked_rx) = oneshot::channel();
|
||||
let (finished_tx, finished_rx) = oneshot::channel();
|
||||
|
||||
// A synchronous stderr write can block a runtime thread as well as SQLx's
|
||||
// worker. Keep the timeout on a separate thread so regressions cannot hang it.
|
||||
let worker = std::thread::spawn(move || {
|
||||
let result = tracing::subscriber::with_default(subscriber, || -> Result<()> {
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()?;
|
||||
let span = tracing::info_span!("sqlite_transaction");
|
||||
span.in_scope(|| {
|
||||
runtime.block_on(async {
|
||||
let mut transaction = writer_pool.begin().await?;
|
||||
sqlx::query("INSERT INTO writes VALUES (1, 0)")
|
||||
.execute(&mut *transaction)
|
||||
.await?;
|
||||
let _ = locked_tx.send(());
|
||||
// Enough commands to fill an unread OS pipe if every entry
|
||||
// and exit of the caller's span produces a log record.
|
||||
for _ in 0..4096 {
|
||||
sqlx::query("UPDATE writes SET value = value + 1 WHERE id = 1")
|
||||
.execute(&mut *transaction)
|
||||
.await?;
|
||||
}
|
||||
transaction.commit().await?;
|
||||
tracing::info!("transaction committed");
|
||||
tracing::warn!("explicit warning");
|
||||
writer_pool.close().await;
|
||||
Ok(())
|
||||
})
|
||||
})
|
||||
});
|
||||
let _ = finished_tx.send(result);
|
||||
});
|
||||
|
||||
let progress = timeout(Duration::from_secs(/*secs*/ 15), async {
|
||||
locked_rx.await?;
|
||||
let write = async {
|
||||
sqlx::query("INSERT INTO writes VALUES (2, 0)")
|
||||
.execute(&contender_pool)
|
||||
.await
|
||||
.context("competing writer could not acquire the SQLite lock")
|
||||
};
|
||||
tokio::try_join!(write, async { finished_rx.await? })?;
|
||||
Ok::<_, anyhow::Error>(())
|
||||
})
|
||||
.await;
|
||||
|
||||
// Always drain before reporting a failure, releasing a worker blocked in
|
||||
// tracing if enter/exit logging is accidentally restored.
|
||||
let logs = tokio::task::spawn_blocking(move || {
|
||||
let mut logs = String::new();
|
||||
reader.read_to_string(&mut logs)?;
|
||||
Ok::<_, std::io::Error>(logs)
|
||||
})
|
||||
.await??;
|
||||
worker.join().expect("SQLite worker panicked");
|
||||
progress.context("SQLite transaction stalled with undrained stderr")??;
|
||||
let rows: Vec<(i64, i64)> = sqlx::query_as("SELECT id, value FROM writes ORDER BY id")
|
||||
.fetch_all(&contender_pool)
|
||||
.await?;
|
||||
contender_pool.close().await;
|
||||
assert_eq!(rows, vec![(1, 4096), (2, 0)]);
|
||||
|
||||
let lines: Vec<_> = logs.lines().collect();
|
||||
assert_eq!(lines.len(), 4, "{logs}");
|
||||
for (line, message) in
|
||||
lines
|
||||
.iter()
|
||||
.zip(["new", "transaction committed", "explicit warning", "close"])
|
||||
{
|
||||
assert!(line.contains(message), "{logs}");
|
||||
}
|
||||
assert!(lines[2].contains("WARN"), "{logs}");
|
||||
for field in ["time.busy", "time.idle"] {
|
||||
assert!(lines[3].contains(field), "{logs}");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -1013,3 +1013,75 @@ async fn repair_recency_migration_succeeds_while_another_connection_holds_writer
|
||||
pool.close().await;
|
||||
repair_result.expect("current migration history should not need the writer slot");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn writable_pool_reads_do_not_wait_for_an_existing_writer() {
|
||||
let sqlite_home = crate::runtime::test_support::unique_temp_dir();
|
||||
tokio::fs::create_dir_all(&sqlite_home).await.unwrap();
|
||||
let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| {
|
||||
let _ = std::fs::remove_dir_all(sqlite_home);
|
||||
});
|
||||
let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
|
||||
|
||||
for (index, (setup, expected_mode)) in [
|
||||
("PRAGMA auto_vacuum = INCREMENTAL", 2_i64),
|
||||
("PRAGMA auto_vacuum = FULL", 2),
|
||||
("PRAGMA auto_vacuum = NONE; VACUUM", 0),
|
||||
]
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
{
|
||||
let path = sqlite_home.join(format!("pool-{index}.sqlite"));
|
||||
let pool = sqlite.open_read_write_pool(&path).await.unwrap();
|
||||
assert_eq!(
|
||||
sqlx::query_scalar::<_, i64>("PRAGMA auto_vacuum")
|
||||
.fetch_one(&pool)
|
||||
.await
|
||||
.unwrap(),
|
||||
2,
|
||||
"new databases must support incremental vacuum"
|
||||
);
|
||||
sqlx::query(setup).execute(&pool).await.unwrap();
|
||||
sqlx::query("CREATE TABLE sample (value INTEGER); INSERT INTO sample VALUES (7)")
|
||||
.execute(&pool)
|
||||
.await
|
||||
.unwrap();
|
||||
pool.close().await;
|
||||
|
||||
// Reopening converts FULL, but preserves legacy NONE without VACUUM.
|
||||
let pool = sqlite.open_read_write_pool(&path).await.unwrap();
|
||||
let mut connection = pool.acquire().await.unwrap();
|
||||
let writer = connection.begin_with("BEGIN IMMEDIATE").await.unwrap();
|
||||
let read_result = tokio::time::timeout(std::time::Duration::from_secs(2), async {
|
||||
let readers = sqlite.open_read_write_pool(&path).await.unwrap();
|
||||
let mode = sqlx::query_scalar::<_, i64>("PRAGMA auto_vacuum")
|
||||
.fetch_one(&readers)
|
||||
.await
|
||||
.unwrap();
|
||||
let mut held_connections = Vec::new();
|
||||
let mut values = Vec::new();
|
||||
// Keep every connection checked out to exercise lazy pool expansion too.
|
||||
for _ in 0..5 {
|
||||
let mut reader = readers.acquire().await.unwrap();
|
||||
values.push(
|
||||
sqlx::query_scalar::<_, i64>("SELECT value FROM sample")
|
||||
.fetch_one(&mut *reader)
|
||||
.await
|
||||
.unwrap(),
|
||||
);
|
||||
held_connections.push(reader);
|
||||
}
|
||||
drop(held_connections);
|
||||
readers.close().await;
|
||||
(mode, values)
|
||||
})
|
||||
.await;
|
||||
writer.rollback().await.unwrap();
|
||||
drop(connection);
|
||||
pool.close().await;
|
||||
assert_eq!(
|
||||
read_result.expect("opening and expanding a read pool must not wait for a writer"),
|
||||
(expected_mode, vec![7; 5])
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,9 +16,7 @@ use sqlx::ConnectOptions;
|
||||
use sqlx::Error;
|
||||
use sqlx::SqlitePool;
|
||||
use sqlx::migrate::Migrator;
|
||||
use sqlx::sqlite::SqliteAutoVacuum;
|
||||
use sqlx::sqlite::SqliteConnectOptions;
|
||||
use sqlx::sqlite::SqliteJournalMode;
|
||||
use sqlx::sqlite::SqlitePoolOptions;
|
||||
use sqlx::sqlite::SqliteSynchronous;
|
||||
use std::path::Path;
|
||||
@@ -299,13 +297,39 @@ impl SqliteConfig {
|
||||
let options = SqliteConnectOptions::new()
|
||||
.filename(path)
|
||||
.create_if_missing(true)
|
||||
.journal_mode(SqliteJournalMode::Wal)
|
||||
.synchronous(SqliteSynchronous::Normal)
|
||||
.auto_vacuum(SqliteAutoVacuum::Incremental)
|
||||
.busy_timeout(Duration::from_secs(5))
|
||||
.log_statements(LevelFilter::Off);
|
||||
SqlitePoolOptions::new()
|
||||
.max_connections(5)
|
||||
.after_connect(|connection, _metadata| {
|
||||
Box::pin(async move {
|
||||
let mode: i64 = sqlx::query_scalar("PRAGMA auto_vacuum")
|
||||
.fetch_one(&mut *connection)
|
||||
.await?;
|
||||
// The setter takes the writer lock even when the mode is unchanged.
|
||||
// Initialize it before WAL creates the first database page. Existing
|
||||
// NONE databases require VACUUM to convert; leave those alone.
|
||||
let empty = if mode == 0 {
|
||||
sqlx::query_scalar::<_, bool>(
|
||||
"SELECT NOT EXISTS (SELECT 1 FROM sqlite_schema)",
|
||||
)
|
||||
.fetch_one(&mut *connection)
|
||||
.await?
|
||||
} else {
|
||||
false
|
||||
};
|
||||
if mode == 1 || empty {
|
||||
sqlx::query("PRAGMA auto_vacuum = INCREMENTAL")
|
||||
.execute(&mut *connection)
|
||||
.await?;
|
||||
}
|
||||
sqlx::query("PRAGMA journal_mode = WAL")
|
||||
.execute(connection)
|
||||
.await?;
|
||||
Ok(())
|
||||
})
|
||||
})
|
||||
.connect_with(options)
|
||||
.await
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user