Add dedicated storage for paginated thread history (#32234)

## Why

Paginated thread history needs its own SQLite database to avoid adding lock
contention to the main state store.

## What changed

- Add the `thread_history_1.sqlite` path and migration scaffolding.
- Create tables and pagination indexes for projected turns and items, plus a
  per-thread projection checkpoint.
- Register the database with runtime diagnostics, Bazel inputs, and database
  telemetry.

GitOrigin-RevId: d194310835f1df2a2a29c7827d77ca37eabbd0a3
This commit is contained in:
Owen Lin authored and copyberry committed 2026-07-10 18:27:19 +00:00
1 parent 2da1b12822
commit 414217dc8a
7 files changed
+74 -7

No files matched your search

+1 -1
View File
@@ -142,7 +142,7 @@ async fn debug_clear_memories_resets_memories_db_without_state_db() -> Result<()
let codex_home = TempDir::new()?; let codex_home = TempDir::new()?;
let runtime = let runtime =
StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()).await?; StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()).await?;
drop(runtime); runtime.close().await;
let db_path = state_db_path(codex_home.path()); let db_path = state_db_path(codex_home.path());
let memories_db_path = memories_db_path(codex_home.path()); let memories_db_path = memories_db_path(codex_home.path());
+1
View File
@@ -7,6 +7,7 @@ codex_rust_crate(
"logs_migrations/**", "logs_migrations/**",
"memory_migrations/**", "memory_migrations/**",
"migrations/**", "migrations/**",
"thread_history_migrations/**",
]), ]),
crate_name = "codex_state", crate_name = "codex_state",
) )
+3
View File
@@ -85,6 +85,8 @@ pub use runtime::sqlite_error_detail_is_lock;
pub use runtime::sqlite_integrity_check; pub use runtime::sqlite_integrity_check;
pub use runtime::state_db_filename; pub use runtime::state_db_filename;
pub use runtime::state_db_path; pub use runtime::state_db_path;
pub use runtime::thread_history_db_filename;
pub use runtime::thread_history_db_path;
pub use telemetry::DbTelemetry; pub use telemetry::DbTelemetry;
pub use telemetry::DbTelemetryHandle; pub use telemetry::DbTelemetryHandle;
pub use telemetry::install_process_db_telemetry; pub use telemetry::install_process_db_telemetry;
@@ -98,6 +100,7 @@ pub const LOGS_DB_FILENAME: &str = "logs_2.sqlite";
pub const GOALS_DB_FILENAME: &str = "goals_1.sqlite"; pub const GOALS_DB_FILENAME: &str = "goals_1.sqlite";
pub const MEMORIES_DB_FILENAME: &str = "memories_1.sqlite"; pub const MEMORIES_DB_FILENAME: &str = "memories_1.sqlite";
pub const STATE_DB_FILENAME: &str = "state_5.sqlite"; pub const STATE_DB_FILENAME: &str = "state_5.sqlite";
pub const THREAD_HISTORY_DB_FILENAME: &str = "thread_history_1.sqlite";
/// Errors encountered during DB operations. Tags: [stage] /// Errors encountered during DB operations. Tags: [stage]
pub const DB_ERROR_METRIC: &str = "codex.db.error"; pub const DB_ERROR_METRIC: &str = "codex.db.error";
+7
View File
@@ -7,6 +7,7 @@ pub(crate) static STATE_MIGRATOR: Migrator = sqlx::migrate!("./migrations");
pub(crate) static LOGS_MIGRATOR: Migrator = sqlx::migrate!("./logs_migrations"); pub(crate) static LOGS_MIGRATOR: Migrator = sqlx::migrate!("./logs_migrations");
pub(crate) static GOALS_MIGRATOR: Migrator = sqlx::migrate!("./goals_migrations"); pub(crate) static GOALS_MIGRATOR: Migrator = sqlx::migrate!("./goals_migrations");
pub(crate) static MEMORIES_MIGRATOR: Migrator = sqlx::migrate!("./memory_migrations"); pub(crate) static MEMORIES_MIGRATOR: Migrator = sqlx::migrate!("./memory_migrations");
pub(crate) static THREAD_HISTORY_MIGRATOR: Migrator = sqlx::migrate!("./thread_history_migrations");
/// Allow an older Codex binary to open a database that has already been /// Allow an older Codex binary to open a database that has already been
/// migrated by a newer binary running in parallel. /// migrated by a newer binary running in parallel.
@@ -41,6 +42,12 @@ pub(crate) fn runtime_memories_migrator() -> Migrator {
runtime_migrator(&MEMORIES_MIGRATOR) runtime_migrator(&MEMORIES_MIGRATOR)
} }
// The paginated history projector will call this when it takes ownership of opening the database.
#[allow(dead_code)]
pub(crate) fn runtime_thread_history_migrator() -> Migrator {
runtime_migrator(&THREAD_HISTORY_MIGRATOR)
}
pub(crate) async fn repair_legacy_recency_migration_version( pub(crate) async fn repair_legacy_recency_migration_version(
pool: &SqlitePool, pool: &SqlitePool,
migrator: &Migrator, migrator: &Migrator,
+23 -6
View File
@@ -13,6 +13,7 @@ use crate::LogRow;
use crate::MEMORIES_DB_FILENAME; use crate::MEMORIES_DB_FILENAME;
use crate::STATE_DB_FILENAME; use crate::STATE_DB_FILENAME;
use crate::SortKey; use crate::SortKey;
use crate::THREAD_HISTORY_DB_FILENAME;
use crate::ThreadMetadata; use crate::ThreadMetadata;
use crate::ThreadMetadataBuilder; use crate::ThreadMetadataBuilder;
use crate::ThreadsPage; use crate::ThreadsPage;
@@ -144,7 +145,16 @@ const MEMORIES_DB: RuntimeDbSpec = RuntimeDbSpec {
migrate_phase: "migrate_memories", migrate_phase: "migrate_memories",
}; };
const RUNTIME_DBS: [RuntimeDbSpec; 4] = [STATE_DB, LOGS_DB, GOALS_DB, MEMORIES_DB]; const THREAD_HISTORY_DB: RuntimeDbSpec = RuntimeDbSpec {
label: "thread history DB",
filename: THREAD_HISTORY_DB_FILENAME,
kind: DbKind::ThreadHistory,
open_phase: "open_thread_history",
migrate_phase: "migrate_thread_history",
};
const RUNTIME_DBS: [RuntimeDbSpec; 5] =
[STATE_DB, LOGS_DB, GOALS_DB, MEMORIES_DB, THREAD_HISTORY_DB];
#[derive(Clone, Debug, Eq, PartialEq)] #[derive(Clone, Debug, Eq, PartialEq)]
pub struct RuntimeDbPath { pub struct RuntimeDbPath {
@@ -167,9 +177,9 @@ pub struct StateRuntime {
impl StateRuntime { impl StateRuntime {
/// Initialize the state runtime using the provided Codex home and default provider. /// Initialize the state runtime using the provided Codex home and default provider.
/// ///
/// This opens (and migrates) the SQLite databases under `codex_home`, /// This opens (and migrates) the SQLite databases under `codex_home`.
/// keeping logs in a dedicated file to reduce lock contention with the /// Logs and paginated thread history live in dedicated files to reduce
/// rest of the state store. /// lock contention with the rest of the state store.
pub async fn init(codex_home: PathBuf, default_provider: String) -> anyhow::Result<Arc<Self>> { pub async fn init(codex_home: PathBuf, default_provider: String) -> anyhow::Result<Arc<Self>> {
Self::init_inner( Self::init_inner(
codex_home, codex_home,
@@ -508,6 +518,14 @@ pub fn memories_db_path(codex_home: &Path) -> PathBuf {
MEMORIES_DB.path(codex_home) MEMORIES_DB.path(codex_home)
} }
pub fn thread_history_db_filename() -> String {
THREAD_HISTORY_DB.filename.to_string()
}
pub fn thread_history_db_path(codex_home: &Path) -> PathBuf {
THREAD_HISTORY_DB.path(codex_home)
}
pub fn runtime_db_paths(codex_home: &Path) -> Vec<RuntimeDbPath> { pub fn runtime_db_paths(codex_home: &Path) -> Vec<RuntimeDbPath> {
RUNTIME_DBS RUNTIME_DBS
.iter() .iter()
@@ -734,8 +752,7 @@ mod tests {
.collect::<BTreeSet<_>>(); .collect::<BTreeSet<_>>();
assert_eq!(phases, expected); assert_eq!(phases, expected);
runtime.pool.close().await; runtime.close().await;
runtime.logs_pool.close().await;
let _ = tokio::fs::remove_dir_all(codex_home).await; let _ = tokio::fs::remove_dir_all(codex_home).await;
} }
} }
+2
View File
@@ -41,6 +41,7 @@ pub(crate) enum DbKind {
Logs, Logs,
Goals, Goals,
Memories, Memories,
ThreadHistory,
} }
impl DbKind { impl DbKind {
@@ -50,6 +51,7 @@ impl DbKind {
Self::Logs => "logs", Self::Logs => "logs",
Self::Goals => "goals", Self::Goals => "goals",
Self::Memories => "memories", Self::Memories => "memories",
Self::ThreadHistory => "thread_history",
} }
} }
} }
@@ -0,0 +1,37 @@
CREATE TABLE thread_turns (
thread_id TEXT NOT NULL,
turn_id TEXT NOT NULL,
rollout_ordinal INTEGER NOT NULL,
status TEXT NOT NULL,
error_json TEXT,
started_at INTEGER,
completed_at INTEGER,
duration_ms INTEGER,
first_user_item_id TEXT,
final_agent_item_id TEXT,
PRIMARY KEY (thread_id, turn_id)
);
CREATE UNIQUE INDEX idx_thread_turns_page
ON thread_turns(thread_id, rollout_ordinal);
CREATE TABLE thread_items (
thread_id TEXT NOT NULL,
turn_id TEXT NOT NULL,
item_id TEXT NOT NULL,
rollout_ordinal INTEGER NOT NULL,
item_json TEXT NOT NULL,
PRIMARY KEY (thread_id, turn_id, item_id)
);
CREATE UNIQUE INDEX idx_thread_items_page
ON thread_items(thread_id, rollout_ordinal);
CREATE INDEX idx_thread_items_by_turn_page
ON thread_items(thread_id, turn_id, rollout_ordinal);
CREATE TABLE thread_history_projection_state (
thread_id TEXT PRIMARY KEY,
next_rollout_byte_offset INTEGER NOT NULL,
next_rollout_ordinal INTEGER NOT NULL
);