agent/codex
1
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Bump SQLx to pick up newer bundled SQLite (#24728)

Browse Source 
## Why

Codex stores thread, log, goal, and memory state in bundled SQLite
databases through SQLx. We have a suspected SQLite WAL-reset corruption
issue under heavy concurrent writer load, especially when multiple
subagents are active. The existing `sqlx 0.8.6` dependency kept us on an
older `libsqlite3-sys` / bundled SQLite, so this PR moves the SQLx stack
far enough forward to pick up the newer bundled SQLite library.

## What changed

- Bump the workspace `sqlx` dependency to `0.9.0`.
- Use the SQLx 0.9 feature names explicitly: `runtime-tokio`,
`tls-rustls`, and `sqlite-bundled`.
- Update `Cargo.lock` so `sqlx-sqlite` resolves through `libsqlite3-sys
0.37.0`.
- Refresh `MODULE.bazel.lock` for the dependency changes.
- Adapt `codex-state` to SQLx 0.9:
- build dynamic state queries with `QueryBuilder<Sqlite>` instead of
passing dynamic `String`s to `sqlx::query`;
- remove the old `QueryBuilder` lifetime parameter from helper
signatures;
- preserve SQLx's new `Migrator` fields when constructing runtime
migrators.

## Verification

- `just test -p codex-state`
- `just bazel-lock-check`
- `cargo check -p codex-state --tests`
This commit is contained in:
jif-oai
2026-05-27 18:44:07 +02:00
committed by GitHub
Unverified
parent aa184548b1
commit 379511dcea
7 changed files with 591 additions and 342 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MODULE.bazel.lock
Generated
+30 -25
View File
File diff suppressed because one or more lines are too long
codex-rs/Cargo.lock
Generated
+438 -235
View File
File diff suppressed because it is too large Load Diff
codex-rs/Cargo.toml
+4 -3
View File
@@ -366,13 +366,14 @@ sha2 = "0.10"
shlex = "1.3.0"
similar = "2.7.0"
socket2 = "0.6.1"
sqlx = { version = "0.8.6", default-features = false, features = [
sqlx = { version = "0.9.0", default-features = false, features = [
"chrono",
"json",
"macros",
"migrate",
"runtime-tokio-rustls",
"sqlite",
"runtime-tokio",
"tls-rustls",
"sqlite-bundled",
"time",
"uuid",
] }
codex-rs/state/src/migrations.rs
+2
View File
@@ -19,6 +19,8 @@ fn runtime_migrator(base: &'static Migrator) -> Migrator {
ignore_missing: true,
locking: base.locking,
no_tx: base.no_tx,
table_name: base.table_name.clone(),
create_schemas: base.create_schemas.clone(),
}
}
codex-rs/state/src/runtime/goals.rs
+55 -32
View File
@@ -399,44 +399,78 @@ WHERE thread_id = ?
}
let now_ms = datetime_to_epoch_millis(Utc::now());
let active_or_stopped_status_filter =
"status IN ('active', 'paused', 'blocked', 'usage_limited', 'budget_limited')";
let status_filter = match mode {
GoalAccountingMode::ActiveStatusOnly => "status = 'active'",
GoalAccountingMode::ActiveOnly => "status IN ('active', 'budget_limited')",
GoalAccountingMode::ActiveOrComplete => {
"status IN ('active', 'budget_limited', 'complete')"
}
GoalAccountingMode::ActiveOrStopped => {
"status IN ('active', 'paused', 'blocked', 'usage_limited', 'budget_limited')"
}
GoalAccountingMode::ActiveOrStopped => active_or_stopped_status_filter,
};
let budget_limit_status_filter = match mode {
GoalAccountingMode::ActiveStatusOnly
| GoalAccountingMode::ActiveOnly
| GoalAccountingMode::ActiveOrComplete => "status = 'active'",
GoalAccountingMode::ActiveOrStopped => {
"status IN ('active', 'paused', 'blocked', 'usage_limited', 'budget_limited')"
}
GoalAccountingMode::ActiveOrStopped => active_or_stopped_status_filter,
};
let goal_id_filter = if expected_goal_id.is_some() {
"goal_id = ?"
} else {
"1 = 1"
};
let query = format!(
let mut builder = QueryBuilder::<Sqlite>::new(
r#"
UPDATE thread_goals
SET
time_used_seconds = time_used_seconds + ?,
tokens_used = tokens_used + ?,
time_used_seconds = time_used_seconds +
"#,
);
builder.push_bind(time_delta_seconds);
builder.push(
r#",
tokens_used = tokens_used +
"#,
);
builder.push_bind(token_delta);
builder.push(
r#",
status = CASE
WHEN {budget_limit_status_filter} AND token_budget IS NOT NULL AND tokens_used + ? >= token_budget
THEN ?
WHEN
"#,
);
builder.push(budget_limit_status_filter);
builder.push(
r#"
AND token_budget IS NOT NULL
AND tokens_used +
"#,
);
builder.push_bind(token_delta);
builder.push(
r#"
>= token_budget
THEN
"#,
);
builder.push_bind(crate::ThreadGoalStatus::BudgetLimited.as_str());
builder.push(
r#"
ELSE status
END,
updated_at_ms = ?
WHERE thread_id = ?
AND {status_filter}
AND {goal_id_filter}
updated_at_ms =
"#,
);
builder.push_bind(now_ms);
builder.push(
r#"
WHERE thread_id =
"#,
);
builder.push_bind(thread_id.to_string());
builder.push(" AND ");
builder.push(status_filter);
if let Some(expected_goal_id) = expected_goal_id {
builder.push(" AND goal_id = ").push_bind(expected_goal_id);
}
builder.push(
r#"
RETURNING
thread_id,
goal_id,
@@ -450,18 +484,7 @@ RETURNING
"#,
);
let mut query = sqlx::query(&query)
.bind(time_delta_seconds)
.bind(token_delta)
.bind(token_delta)
.bind(crate::ThreadGoalStatus::BudgetLimited.as_str())
.bind(now_ms)
.bind(thread_id.to_string());
if let Some(expected_goal_id) = expected_goal_id {
query = query.bind(expected_goal_id);
}
let row = query.fetch_optional(self.pool.as_ref()).await?;
let row = builder.build().fetch_optional(self.pool.as_ref()).await?;
let Some(row) = row else {
return Ok(GoalAccountingOutcome::Unchanged(
codex-rs/state/src/runtime/logs.rs
+25 -18
View File
@@ -343,11 +343,23 @@ WHERE id IN (
let max_bytes = usize::try_from(LOG_PARTITION_SIZE_LIMIT_BYTES).unwrap_or(usize::MAX);
// Bound the fetched rows in SQL first so over-retained partitions do not have to load
// every row into memory, then apply the exact whole-line byte cap after formatting.
let requested_threads = vec!["(?)"; thread_ids.len()].join(", ");
let query = format!(
let mut builder = QueryBuilder::<Sqlite>::new(
r#"
WITH requested_threads(thread_id) AS (
VALUES {requested_threads}
VALUES
"#,
);
{
let mut separated = builder.separated(", ");
for thread_id in thread_ids {
separated
.push("(")
.push_bind_unseparated(*thread_id)
.push_unseparated(")");
}
}
builder.push(
r#"
),
latest_processes AS (
SELECT (
@@ -388,16 +400,13 @@ bounded_feedback_logs AS (
)
SELECT ts, ts_nanos, level, feedback_log_body
FROM bounded_feedback_logs
WHERE cumulative_estimated_bytes <= ?
ORDER BY ts DESC, ts_nanos DESC, id DESC
"#
WHERE cumulative_estimated_bytes <=
"#,
);
let mut sql = sqlx::query_as::<_, FeedbackLogRow>(query.as_str());
for thread_id in thread_ids {
sql = sql.bind(thread_id);
}
let rows = sql
.bind(LOG_PARTITION_SIZE_LIMIT_BYTES)
builder.push_bind(LOG_PARTITION_SIZE_LIMIT_BYTES);
builder.push(" ORDER BY ts DESC, ts_nanos DESC, id DESC");
let rows = builder
.build_query_as::<FeedbackLogRow>()
.fetch_all(self.logs_pool.as_ref())
.await?;
@@ -463,7 +472,7 @@ fn format_feedback_log_line(
line
}
fn push_log_filters<'a>(builder: &mut QueryBuilder<'a, Sqlite>, query: &'a LogQuery) {
fn push_log_filters(builder: &mut QueryBuilder<Sqlite>, query: &LogQuery) {
if !query.levels_upper.is_empty() {
builder.push(" AND UPPER(level) IN (");
{
@@ -511,11 +520,7 @@ fn push_log_filters<'a>(builder: &mut QueryBuilder<'a, Sqlite>, query: &'a LogQu
}
}
fn push_like_filters<'a>(
builder: &mut QueryBuilder<'a, Sqlite>,
column: &str,
filters: &'a [String],
) {
fn push_like_filters(builder: &mut QueryBuilder<Sqlite>, column: &str, filters: &[String]) {
if filters.is_empty() {
return;
}
@@ -613,6 +618,8 @@ mod tests {
ignore_missing: false,
locking: true,
no_tx: false,
table_name: LOGS_MIGRATOR.table_name.clone(),
create_schemas: LOGS_MIGRATOR.create_schemas.clone(),
};
let pool = SqlitePool::connect_with(
SqliteConnectOptions::new()
codex-rs/state/src/runtime/threads.rs
+37 -29
View File
@@ -252,20 +252,16 @@ LIMIT 2
parent_thread_id: ThreadId,
status: Option<crate::DirectionalThreadSpawnEdgeStatus>,
) -> anyhow::Result<Vec<ThreadId>> {
let mut query = String::from(
"SELECT child_thread_id FROM thread_spawn_edges WHERE parent_thread_id = ?",
let mut builder = QueryBuilder::<Sqlite>::new(
"SELECT child_thread_id FROM thread_spawn_edges WHERE parent_thread_id = ",
);
if status.is_some() {
query.push_str(" AND status = ?");
}
query.push_str(" ORDER BY child_thread_id");
let mut sql = sqlx::query(query.as_str()).bind(parent_thread_id.to_string());
builder.push_bind(parent_thread_id.to_string());
if let Some(status) = status {
sql = sql.bind(status.to_string());
builder.push(" AND status = ").push_bind(status.to_string());
}
builder.push(" ORDER BY child_thread_id");
let rows = sql.fetch_all(self.pool.as_ref()).await?;
let rows = builder.build().fetch_all(self.pool.as_ref()).await?;
rows.into_iter()
.map(|row| {
ThreadId::try_from(row.try_get::<String, _>("child_thread_id")?).map_err(Into::into)
@@ -278,36 +274,48 @@ LIMIT 2
root_thread_id: ThreadId,
status: Option<crate::DirectionalThreadSpawnEdgeStatus>,
) -> anyhow::Result<Vec<ThreadId>> {
let status_filter = if status.is_some() {
" AND status = ?"
} else {
""
};
let query = format!(
let mut builder = QueryBuilder::<Sqlite>::new(
r#"
WITH RECURSIVE subtree(child_thread_id, depth) AS (
SELECT child_thread_id, 1
FROM thread_spawn_edges
WHERE parent_thread_id = ?{status_filter}
WHERE parent_thread_id =
"#,
);
builder.push_bind(root_thread_id.to_string());
if let Some(status) = status {
let status = status.to_string();
builder.push(" AND status = ").push_bind(status.clone());
builder.push(
r#"
UNION ALL
SELECT edge.child_thread_id, subtree.depth + 1
FROM thread_spawn_edges AS edge
JOIN subtree ON edge.parent_thread_id = subtree.child_thread_id
WHERE 1 = 1{status_filter}
WHERE status =
"#,
);
builder.push_bind(status);
} else {
builder.push(
r#"
UNION ALL
SELECT edge.child_thread_id, subtree.depth + 1
FROM thread_spawn_edges AS edge
JOIN subtree ON edge.parent_thread_id = subtree.child_thread_id
"#,
);
}
builder.push(
r#"
)
SELECT child_thread_id
FROM subtree
ORDER BY depth ASC, child_thread_id ASC
"#
"#,
);
let mut sql = sqlx::query(query.as_str()).bind(root_thread_id.to_string());
if let Some(status) = status {
let status = status.to_string();
sql = sql.bind(status.clone()).bind(status);
}
let rows = sql.fetch_all(self.pool.as_ref()).await?;
let rows = builder.build().fetch_all(self.pool.as_ref()).await?;
rows.into_iter()
.map(|row| {
ThreadId::try_from(row.try_get::<String, _>("child_thread_id")?).map_err(Into::into)
@@ -999,7 +1007,7 @@ fn one_thread_id_from_rows(
}
}
pub(super) fn push_thread_select_columns(builder: &mut QueryBuilder<'_, Sqlite>) {
pub(super) fn push_thread_select_columns(builder: &mut QueryBuilder<Sqlite>) {
builder.push(
r#"
SELECT
@@ -1076,7 +1084,7 @@ pub struct ThreadFilterOptions<'a> {
}
pub(super) fn push_thread_filters<'a>(
builder: &mut QueryBuilder<'a, Sqlite>,
builder: &mut QueryBuilder<Sqlite>,
options: ThreadFilterOptions<'a>,
) {
let ThreadFilterOptions {
@@ -1156,7 +1164,7 @@ pub(super) fn push_thread_filters<'a>(
}
pub(super) fn push_thread_order_and_limit(
builder: &mut QueryBuilder<'_, Sqlite>,
builder: &mut QueryBuilder<Sqlite>,
sort_key: SortKey,
sort_direction: SortDirection,
limit: usize,