Batch metadata reads when resolving thread names (#49305)

## Why

Thread name resolution performs a separate SQLite metadata lookup for each thread. Batch these reads to reduce database queries for large result sets.

## What changed

Add `StateRuntime::get_threads` and use it in `resolve_thread_names`. Fetch metadata in batches of at most 999 IDs to stay within SQLite's bind limit, skipping missing or invalid records independently so valid metadata remains available.

## Testing

Add regression coverage for reads spanning multiple batches, invalid and missing records, duplicate IDs, and empty input.

GitOrigin-RevId: a403416f1da5adea56705c70839850760d3eb550
This commit is contained in:
jif
2026-09-29 16:36:07 +00:00
committed by copyberry
parent a6f09397aa
commit c2837d8ece
4 changed files with 108 additions and 5 deletions
+1
View File
@@ -54,6 +54,7 @@ mod rollout_migration;
#[cfg(test)]
pub(crate) mod test_support;
mod thread_attachments;
mod thread_metadata;
mod thread_section_order;
mod thread_sections;
mod threads;
@@ -0,0 +1,50 @@
//! Batched metadata reads for best-effort thread enrichment. Missing or invalid rows are skipped
//! independently so one unreadable thread does not hide metadata for the rest of a page.
use std::collections::HashMap;
use codex_protocol::ThreadId;
use sqlx::QueryBuilder;
use sqlx::Sqlite;
use super::StateRuntime;
use super::threads::push_thread_select_columns;
use crate::ThreadMetadata;
use crate::model::ThreadRow;
const THREAD_METADATA_BATCH_SIZE: usize = 999;
impl StateRuntime {
/// Fetch metadata for multiple threads, omitting missing or invalid records.
pub async fn get_threads(
&self,
thread_ids: &[ThreadId],
) -> anyhow::Result<HashMap<ThreadId, ThreadMetadata>> {
let mut threads = HashMap::with_capacity(thread_ids.len());
// Keep each query within SQLite's bind limit, including for large search result sets.
for thread_ids in thread_ids.chunks(THREAD_METADATA_BATCH_SIZE) {
let mut builder = QueryBuilder::<Sqlite>::new("");
push_thread_select_columns(&mut builder);
builder.push(" FROM threads WHERE threads.id IN (");
let mut separated = builder.separated(", ");
for thread_id in thread_ids {
separated.push_bind(thread_id.to_string());
}
separated.push_unseparated(")");
let rows = builder.build().fetch_all(self.pool.as_ref()).await?;
for row in rows {
if let Ok(metadata) =
ThreadRow::try_from_row(&row).and_then(ThreadMetadata::try_from)
{
threads.insert(metadata.id, metadata);
}
}
}
Ok(threads)
}
}
#[cfg(test)]
#[path = "thread_metadata_tests.rs"]
mod tests;
@@ -0,0 +1,48 @@
//! Regression coverage for batched metadata reads and independent row decoding failures.
use std::collections::HashMap;
use codex_protocol::ThreadId;
use codex_utils_absolute_path::test_support::PathExt;
use pretty_assertions::assert_eq;
use super::StateRuntime;
use crate::SqliteConfig;
use crate::runtime::test_support::test_thread_metadata;
use crate::runtime::test_support::unique_temp_dir;
#[tokio::test]
async fn get_threads_preserves_valid_metadata_across_batches() -> anyhow::Result<()> {
let home = unique_temp_dir();
let runtime = StateRuntime::init(
SqliteConfig::new_for_testing(home.as_path().abs()),
"test-provider".to_string(),
)
.await?;
let mut ids = (0..1_001).map(|_| ThreadId::new()).collect::<Vec<_>>();
let mut expected = HashMap::new();
for id in [ids[0], ids[1_000]] {
let mut metadata = test_thread_metadata(&home, id, home.clone());
metadata.title = format!("Title for {id}");
metadata.name = Some(format!("Name for {id}"));
runtime.upsert_thread(&metadata).await?;
expected.insert(id, runtime.get_thread(id).await?.expect("persisted thread"));
}
// An unreadable record must not discard the other names in the same page.
let invalid = test_thread_metadata(&home, ids[1], home.clone());
runtime.upsert_thread(&invalid).await?;
sqlx::query("UPDATE threads SET created_at_ms = ? WHERE id = ?")
.bind(i64::MAX)
.bind(invalid.id.to_string())
.execute(runtime.pool.as_ref())
.await?;
assert!(runtime.get_thread(invalid.id).await.is_err());
ids.push(ids[0]);
assert_eq!(runtime.get_threads(&ids).await?, expected);
assert_eq!(runtime.get_threads(&[]).await?, HashMap::new());
runtime.close().await;
std::fs::remove_dir_all(home)?;
Ok(())
}
+9 -5
View File
@@ -252,17 +252,21 @@ pub(super) async fn resolve_thread_names(
.await
.unwrap_or_default();
if let Some(state_db_ctx) = store.state_db().await {
let thread_ids = thread_history_modes.keys().copied().collect::<Vec<_>>();
let metadata_by_id = state_db_ctx
.get_threads(&thread_ids)
.await
.unwrap_or_default();
for (&thread_id, &history_mode) in thread_history_modes {
let Ok(Some(metadata)) = state_db_ctx.get_thread(thread_id).await else {
let Some(metadata) = metadata_by_id.get(&thread_id) else {
continue;
};
let name = match history_mode {
ThreadHistoryMode::Legacy => distinct_thread_metadata_title(&metadata),
ThreadHistoryMode::Paginated => sqlite_thread_name(&metadata),
ThreadHistoryMode::Legacy => distinct_thread_metadata_title(metadata),
ThreadHistoryMode::Paginated => sqlite_thread_name(metadata),
};
if let Some(name) = name {
if history_mode == ThreadHistoryMode::Legacy
&& has_guardian_default_title(&metadata)
if history_mode == ThreadHistoryMode::Legacy && has_guardian_default_title(metadata)
{
names.entry(thread_id).or_insert(name);
} else {