From d583e73c4d1204f1e9f654ef87065e9f0dd07ac7 Mon Sep 17 00:00:00 2001 From: jif Date: Mon, 21 Sep 2026 15:38:25 +0000 Subject: [PATCH] Fetch local message board thread summaries in one query (#47055) ## Why Listing threads issued two additional SQL queries per root to fetch its reply count and latest reply. ## What changed Fetch roots, reply counts, and latest replies in a single query. Materialize the selected page before looking up reply summaries, preserving creation and activity ordering, pagination, and preview limits. ## Testing Strengthen pagination assertions to check complete thread summaries for threads without replies, including Unicode preview truncation, reply counts, and last activity timestamps. GitOrigin-RevId: cbbed36418a349a18daf0771f4ed27aa47de0bc1 --- .../agent-message-board/src/local/queries.rs | 64 +++++++++---------- .../agent-message-board/tests/local_board.rs | 36 ++++++++--- 2 files changed, 56 insertions(+), 44 deletions(-) diff --git a/codex-rs/ext/agent-message-board/src/local/queries.rs b/codex-rs/ext/agent-message-board/src/local/queries.rs index 06dc04e5ce..44f084247e 100644 --- a/codex-rs/ext/agent-message-board/src/local/queries.rs +++ b/codex-rs/ext/agent-message-board/src/local/queries.rs @@ -131,12 +131,11 @@ impl AgentMessageBoard for LocalAgentMessageBoard { } let window = Window::new(&query.page)?; let order = direction(query.direction); - let mut sql = QueryBuilder::new("SELECT p.payload FROM posts p WHERE p.board="); - sql.push_bind(self.identity.to_string()) - .push(" AND p.channel=") - .push_bind(query.channel_name) - .push(" AND p.id=p.root") - .push(" ORDER BY "); + // Select the page before looking up reply summaries, especially when + // activity sorting examines more roots than the page will return. + let mut sql = QueryBuilder::new( + "WITH page AS MATERIALIZED (SELECT p.board,p.id,p.payload,p.seq,", + ); match query.sort { ThreadSort::Created => { sql.push("p.timestamp"); @@ -145,42 +144,39 @@ impl AgentMessageBoard for LocalAgentMessageBoard { sql.push("(SELECT MAX(r.timestamp) FROM posts r WHERE r.board=p.board AND r.root=p.id)"); } } - sql.push(" ") + sql.push(" AS sort_timestamp FROM posts p WHERE p.board=") + .push_bind(self.identity.to_string()) + .push(" AND p.channel=") + .push_bind(query.channel_name) + .push(" AND p.id=p.root ORDER BY sort_timestamp ") .push(order) .push(",p.seq ") .push(order) .push(" LIMIT ") .push_bind((window.limit + 1) as i64) .push(" OFFSET ") - .push_bind(window.offset()); - let roots = decode_posts( - sql.build_query_scalar::() - .fetch_all(&mut *tx) - .await - .map_err(storage_error)?, - )?; + .push_bind(window.offset()) + .push( + ") SELECT p.payload, + (SELECT COUNT(*) FROM posts r WHERE r.board=p.board AND r.root=p.id AND r.id<>r.root), + (SELECT r.payload FROM posts r WHERE r.board=p.board AND r.root=p.id AND r.id<>r.root + ORDER BY r.timestamp DESC,r.seq DESC LIMIT 1) + FROM page p ORDER BY p.sort_timestamp ", + ) + .push(order) + .push(",p.seq ") + .push(order); + let rows = sql + .build_query_as::<(String, i64, Option)>() + .fetch_all(&mut *tx) + .await + .map_err(storage_error)?; let chars = (query.max_chars_per_post.get() as usize) - .min(MAX_READ_CHARS / (2 * roots.len().min(window.limit).max(1))); - let mut threads = Vec::with_capacity(roots.len()); - for root in roots { + .min(MAX_READ_CHARS / (2 * rows.len().min(window.limit).max(1))); + let mut threads = Vec::with_capacity(rows.len()); + for (root, count, last) in rows { + let root: StoredPost = serde_json::from_str(&root).map_err(storage_error)?; let id = root.metadata.message_id; - let count: i64 = sqlx::query_scalar( - "SELECT COUNT(*) FROM posts WHERE board=? AND root=? AND id<>root", - ) - .bind(self.identity.to_string()) - .bind(id.to_string()) - .fetch_one(&mut *tx) - .await - .map_err(storage_error)?; - let last: Option = sqlx::query_scalar( - "SELECT payload FROM posts WHERE board=? AND root=? AND id<>root - ORDER BY timestamp DESC,seq DESC LIMIT 1", - ) - .bind(self.identity.to_string()) - .bind(id.to_string()) - .fetch_optional(&mut *tx) - .await - .map_err(storage_error)?; let last: Option = last .map(|value| serde_json::from_str(&value)) .transpose() diff --git a/codex-rs/ext/agent-message-board/tests/local_board.rs b/codex-rs/ext/agent-message-board/tests/local_board.rs index d1c6e74eb6..a4aba522d7 100644 --- a/codex-rs/ext/agent-message-board/tests/local_board.rs +++ b/codex-rs/ext/agent-message-board/tests/local_board.rs @@ -436,11 +436,19 @@ async fn queries_page_discussions_and_search_unicode() { }; let page = board.list_threads(root, query.clone()).await.unwrap(); assert_eq!( - page.results - .iter() - .map(|thread| thread.thread_id) - .collect::>(), - vec![first.message_id] + page.results, + vec![ThreadSummary { + thread_id: first.message_id, + root_post: PostPreview { + metadata: first.clone(), + text_preview: "É".into(), + n_chars: 6, + truncated: true, + }, + reply_count: 0, + last_activity_at: first.created_at, + latest_reply: None, + }] ); let next = board .list_threads( @@ -456,11 +464,19 @@ async fn queries_page_discussions_and_search_unicode() { .await .unwrap(); assert_eq!( - next.results - .iter() - .map(|thread| thread.thread_id) - .collect::>(), - vec![second.message_id] + next.results, + vec![ThreadSummary { + thread_id: second.message_id, + root_post: PostPreview { + metadata: second.clone(), + text_preview: "s".into(), + n_chars: 6, + truncated: true, + }, + reply_count: 0, + last_activity_at: second.created_at, + latest_reply: None, + }] ); assert_eq!(next.next_cursor, None); let reply = board