mirror of
https://github.com/openai/codex.git
synced 2026-10-19 16:17:49 +00:00
Move thread history projection into one blocking task (#49693)
## What changed Run rollout file opening, metadata lookup, seeking, reading, and record projection together in a single `tokio::task::spawn_blocking` call. Use synchronous file I/O in `read_projection_steps_sync` and carry the current tracing span into the blocking task. GitOrigin-RevId: a231a8b9a328f6d2e2095a29cde1718ccb6b3901
This commit is contained in:
@@ -1,3 +1,5 @@
|
||||
use std::io::Read;
|
||||
use std::io::Seek;
|
||||
use std::io::SeekFrom;
|
||||
use std::path::Path;
|
||||
|
||||
@@ -6,8 +8,6 @@ use codex_app_server_protocol::ThreadHistoryChangeSet;
|
||||
use codex_app_server_protocol::project_rollout_line;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_rollout::RolloutItem;
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tokio::io::AsyncSeekExt;
|
||||
use tracing::warn;
|
||||
|
||||
use super::LocalThreadStore;
|
||||
@@ -90,20 +90,39 @@ async fn read_projection_steps(
|
||||
subagent_history_start_ordinal: Option<u64>,
|
||||
) -> ThreadStoreResult<(Vec<RolloutProjectionStep>, u64)> {
|
||||
let path = rollout_path.to_path_buf();
|
||||
let file =
|
||||
tokio::task::spawn_blocking(move || codex_rollout::open_rollout_seekable_reader(&path))
|
||||
.await
|
||||
.map_err(|err| ThreadStoreError::Internal {
|
||||
message: format!("failed to join rollout projection read: {err}"),
|
||||
})?;
|
||||
let mut file = match file {
|
||||
Ok(file) => tokio::fs::File::from_std(file),
|
||||
let span = tracing::Span::current();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let _entered = span.enter();
|
||||
read_projection_steps_sync(
|
||||
&path,
|
||||
start_offset,
|
||||
expected_ordinal,
|
||||
thread_id,
|
||||
subagent_history_start_ordinal,
|
||||
)
|
||||
})
|
||||
.await
|
||||
.map_err(|err| ThreadStoreError::Internal {
|
||||
message: format!("failed to join rollout projection read: {err}"),
|
||||
})?
|
||||
}
|
||||
|
||||
/// Read and project the complete durable prefix in one blocking task.
|
||||
fn read_projection_steps_sync(
|
||||
rollout_path: &Path,
|
||||
start_offset: u64,
|
||||
expected_ordinal: u64,
|
||||
thread_id: ThreadId,
|
||||
subagent_history_start_ordinal: Option<u64>,
|
||||
) -> ThreadStoreResult<(Vec<RolloutProjectionStep>, u64)> {
|
||||
let mut file = match codex_rollout::open_rollout_seekable_reader(rollout_path) {
|
||||
Ok(file) => file,
|
||||
Err(err) if err.kind() == std::io::ErrorKind::NotFound && start_offset == 0 => {
|
||||
return Ok((Vec::new(), 0));
|
||||
}
|
||||
Err(err) => return Err(thread_store_io_error(err)),
|
||||
};
|
||||
let file_end_offset = file.metadata().await.map_err(thread_store_io_error)?.len();
|
||||
let file_end_offset = file.metadata().map_err(thread_store_io_error)?.len();
|
||||
let byte_count =
|
||||
file_end_offset
|
||||
.checked_sub(start_offset)
|
||||
@@ -115,10 +134,8 @@ async fn read_projection_steps(
|
||||
})?;
|
||||
let mut bytes = vec![0; byte_count];
|
||||
file.seek(SeekFrom::Start(start_offset))
|
||||
.await
|
||||
.map_err(thread_store_io_error)?;
|
||||
file.read_exact(bytes.as_mut_slice())
|
||||
.await
|
||||
.map_err(thread_store_io_error)?;
|
||||
// Only project the newline-terminated prefix; leave a trailing partial record for the next
|
||||
// pass.
|
||||
|
||||
Reference in New Issue
Block a user