mirror of
https://github.com/openai/codex.git
synced 2026-08-23 13:09:46 +00:00
Add bounded batch lookups for message history (#33902)
## What changed - Add a public cursor-based API that reads history entries newest-first from an absolute offset. - Bound batches to 128 rows and 64 KiB while allowing a single oversized row so pagination always makes progress. - Reuse validated byte positions for unchanged, uncapped histories, and fall back to offset scans when files are rewritten or capped. - Preserve offsets for malformed rows and return them without parsed entries. ## Testing Add coverage for pagination, row and byte limits, oversized and malformed rows, file rewrites, capped-history trimming, appends, and log identity changes. GitOrigin-RevId: 57e52a4ac150cd15a5bdc3fef8a351a1f2eb0e3c
This commit is contained in:
Charlie Marsh
412
codex-rs/message-history/src/batch.rs
Normal file
412
codex-rs/message-history/src/batch.rs
Normal file
@@ -0,0 +1,412 @@
|
||||
use std::collections::VecDeque;
|
||||
use std::fs::File;
|
||||
use std::fs::OpenOptions;
|
||||
use std::io::BufRead;
|
||||
use std::io::BufReader;
|
||||
use std::io::Read;
|
||||
use std::io::Seek;
|
||||
use std::io::SeekFrom;
|
||||
use std::time::SystemTime;
|
||||
|
||||
use super::HISTORY_READ_BUFFER_SIZE;
|
||||
use super::HistoryConfig;
|
||||
use super::HistoryEntry;
|
||||
use super::MAX_RETRIES;
|
||||
use super::RETRY_SLEEP;
|
||||
use super::history_filepath;
|
||||
use super::log_identity;
|
||||
|
||||
const MAX_BATCH_ROWS: usize = 128;
|
||||
const MAX_BATCH_BYTES: usize = 64 * 1024;
|
||||
|
||||
/// Position of the newest record to include in a bounded history lookup.
|
||||
///
|
||||
/// The initial cursor identifies only an absolute row offset. Continuation cursors also retain a
|
||||
/// byte position so older batches can scan backward from the previous batch instead of rescanning
|
||||
/// the history prefix.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub struct HistoryBatchCursor {
|
||||
end_offset: usize,
|
||||
byte_anchor: Option<HistoryByteAnchor>,
|
||||
}
|
||||
|
||||
impl HistoryBatchCursor {
|
||||
/// Creates an initial cursor ending at the given absolute history offset.
|
||||
pub fn new(end_offset: usize) -> Self {
|
||||
Self {
|
||||
end_offset,
|
||||
byte_anchor: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the absolute history offset covered first by this cursor.
|
||||
pub fn end_offset(self) -> usize {
|
||||
self.end_offset
|
||||
}
|
||||
}
|
||||
|
||||
/// Validated row boundary used to continue scanning one unchanged file revision.
|
||||
///
|
||||
/// Byte positions and file lengths use `u64` to match filesystem and seek APIs, while history row
|
||||
/// offsets use `usize` to match collection indices.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
struct HistoryByteAnchor {
|
||||
position: u64,
|
||||
revision: HistoryFileRevision,
|
||||
}
|
||||
|
||||
/// File metadata that must remain unchanged before a byte position can be reused.
|
||||
///
|
||||
/// Byte positions are only reused for uncapped histories, which Codex writes append-only. Capped
|
||||
/// histories can be rewritten in place when they are trimmed, so their cursors always fall back to
|
||||
/// an offset scan. Filesystems without a modification time also fall back to an offset scan.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
struct HistoryFileRevision {
|
||||
len: u64,
|
||||
modified: Option<SystemTime>,
|
||||
}
|
||||
|
||||
/// One absolute history offset covered by a bounded lookup.
|
||||
///
|
||||
/// Malformed records retain their offset with `entry` set to `None`, allowing callers to continue
|
||||
/// searching older valid records without changing offset semantics.
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct HistoryBatchEntry {
|
||||
/// Zero-based position in the history file, counted from the oldest record.
|
||||
pub offset: usize,
|
||||
/// Parsed record, or `None` when the row at `offset` is malformed.
|
||||
pub entry: Option<HistoryEntry>,
|
||||
}
|
||||
|
||||
/// A bounded newest-first suffix ending at a requested absolute history offset.
|
||||
///
|
||||
/// `next_older_cursor` identifies the next position a caller should request after exhausting
|
||||
/// `entries`.
|
||||
#[derive(Clone, Debug, Default, PartialEq)]
|
||||
pub struct HistoryBatch {
|
||||
/// Covered records in newest-to-oldest order.
|
||||
pub entries: Vec<HistoryBatchEntry>,
|
||||
/// Next position to request after exhausting `entries`.
|
||||
pub next_older_cursor: Option<HistoryBatchCursor>,
|
||||
}
|
||||
|
||||
struct RawHistoryBatchEntry {
|
||||
offset: usize,
|
||||
byte_position: u64,
|
||||
byte_len: usize,
|
||||
/// Oversized rows remain unbuffered until the scan determines they belong in the result.
|
||||
bytes: Option<Vec<u8>>,
|
||||
}
|
||||
|
||||
/// Look up a bounded batch of history records ending at `cursor`.
|
||||
///
|
||||
/// The file is opened, identity-checked, and shared-locked once. Records are counted from the
|
||||
/// oldest offset on the initial lookup. Continuation lookups scan backward from the byte position
|
||||
/// returned with the previous batch. The result retains at most 128 rows and 64 KiB of raw JSONL,
|
||||
/// except that one oversized newest row is returned alone so callers always make progress.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns an I/O error when the history file cannot be opened, inspected, locked, or read.
|
||||
pub fn lookup_batch(
|
||||
log_id: u64,
|
||||
cursor: HistoryBatchCursor,
|
||||
config: &HistoryConfig,
|
||||
) -> std::io::Result<HistoryBatch> {
|
||||
let path = history_filepath(config);
|
||||
let mut file = OpenOptions::new().read(true).open(path)?;
|
||||
let current_log_id = log_identity(&file.metadata()?).unwrap_or(0);
|
||||
if log_id != 0 && current_log_id != log_id {
|
||||
return Ok(HistoryBatch::default());
|
||||
}
|
||||
|
||||
for _ in 0..MAX_RETRIES {
|
||||
match file.try_lock_shared() {
|
||||
Ok(()) => return scan_batch(&mut file, cursor, config),
|
||||
Err(std::fs::TryLockError::WouldBlock) => std::thread::sleep(RETRY_SLEEP),
|
||||
Err(error) => return Err(error.into()),
|
||||
}
|
||||
}
|
||||
|
||||
Err(std::io::Error::new(
|
||||
std::io::ErrorKind::WouldBlock,
|
||||
"could not acquire shared history lock after multiple attempts",
|
||||
))
|
||||
}
|
||||
|
||||
/// Selects the anchored backward scan only for an unchanged, uncapped history file.
|
||||
///
|
||||
/// Capped histories always use the forward scan because trimming rewrites them in place and a
|
||||
/// same-size rewrite may not be distinguishable from metadata on filesystems with coarse
|
||||
/// modification times. Falling back preserves absolute row semantics at the cost of rescanning
|
||||
/// that request from the beginning.
|
||||
fn scan_batch(
|
||||
file: &mut File,
|
||||
cursor: HistoryBatchCursor,
|
||||
config: &HistoryConfig,
|
||||
) -> std::io::Result<HistoryBatch> {
|
||||
let metadata = file.metadata()?;
|
||||
let revision = HistoryFileRevision {
|
||||
len: metadata.len(),
|
||||
modified: metadata.modified().ok(),
|
||||
};
|
||||
if config.max_bytes.is_none()
|
||||
&& let Some(anchor) = cursor.byte_anchor
|
||||
&& anchor.revision == revision
|
||||
{
|
||||
return scan_batch_backward(file, cursor.end_offset, anchor.position, revision);
|
||||
}
|
||||
|
||||
file.seek(SeekFrom::Start(0))?;
|
||||
let mut batch = scan_batch_forward(file, cursor.end_offset, revision)?;
|
||||
if config.max_bytes.is_some()
|
||||
&& let Some(next_older_cursor) = &mut batch.next_older_cursor
|
||||
{
|
||||
next_older_cursor.byte_anchor = None;
|
||||
}
|
||||
Ok(batch)
|
||||
}
|
||||
|
||||
/// Streams from byte zero through `end_offset`, retaining only the bounded newest suffix.
|
||||
///
|
||||
/// This path establishes byte positions for later continuation cursors and is also the safe
|
||||
/// fallback when an existing cursor belongs to an older file revision. Oversized rows are tracked
|
||||
/// by position and length, then materialized only if they remain in the returned suffix.
|
||||
fn scan_batch_forward(
|
||||
file: &mut File,
|
||||
end_offset: usize,
|
||||
revision: HistoryFileRevision,
|
||||
) -> std::io::Result<HistoryBatch> {
|
||||
let mut suffix = VecDeque::new();
|
||||
let mut suffix_bytes = 0usize;
|
||||
{
|
||||
let mut byte_position = 0u64;
|
||||
let mut reader = BufReader::with_capacity(HISTORY_READ_BUFFER_SIZE, &mut *file);
|
||||
|
||||
'rows: for offset in 0..=end_offset {
|
||||
let mut byte_len = 0usize;
|
||||
let mut bytes = Some(Vec::new());
|
||||
loop {
|
||||
let buffer = reader.fill_buf()?;
|
||||
if buffer.is_empty() {
|
||||
if byte_len == 0 {
|
||||
break 'rows;
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
let newline = buffer.iter().position(|byte| *byte == b'\n');
|
||||
let consumed = newline.map_or(buffer.len(), |index| index + 1);
|
||||
byte_len = byte_len.saturating_add(consumed);
|
||||
if let Some(buffered) = bytes.as_mut() {
|
||||
if byte_len <= MAX_BATCH_BYTES {
|
||||
buffered.extend_from_slice(&buffer[..consumed]);
|
||||
} else {
|
||||
bytes = None;
|
||||
}
|
||||
}
|
||||
reader.consume(consumed);
|
||||
if newline.is_some() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
retain_row(
|
||||
&mut suffix,
|
||||
&mut suffix_bytes,
|
||||
RawHistoryBatchEntry {
|
||||
offset,
|
||||
byte_position,
|
||||
byte_len,
|
||||
bytes,
|
||||
},
|
||||
);
|
||||
byte_position += byte_len as u64;
|
||||
}
|
||||
}
|
||||
|
||||
finish_materialized_batch(file, suffix.into_iter().rev().collect(), revision)
|
||||
}
|
||||
|
||||
/// Reads complete rows backward from a validated exclusive byte boundary.
|
||||
///
|
||||
/// `end_byte_position` must be the start of the row immediately newer than `end_offset`. Scanning
|
||||
/// in reverse lets each continuation touch only its own rows while preserving absolute offsets.
|
||||
fn scan_batch_backward(
|
||||
file: &mut File,
|
||||
end_offset: usize,
|
||||
end_byte_position: u64,
|
||||
revision: HistoryFileRevision,
|
||||
) -> std::io::Result<HistoryBatch> {
|
||||
let mut entries = Vec::new();
|
||||
let mut entries_bytes = 0usize;
|
||||
let mut reversed_row = Some(Vec::new());
|
||||
let mut row_byte_len = 0usize;
|
||||
let mut read_buffer = [0u8; HISTORY_READ_BUFFER_SIZE];
|
||||
let mut read_end = end_byte_position;
|
||||
let mut offset = end_offset;
|
||||
|
||||
while read_end > 0 {
|
||||
let read_start = read_end.saturating_sub(HISTORY_READ_BUFFER_SIZE as u64);
|
||||
let read_len = usize::try_from(read_end - read_start).unwrap_or(HISTORY_READ_BUFFER_SIZE);
|
||||
file.seek(SeekFrom::Start(read_start))?;
|
||||
file.read_exact(&mut read_buffer[..read_len])?;
|
||||
|
||||
for index in (0..read_len).rev() {
|
||||
let byte = read_buffer[index];
|
||||
if byte == b'\n' && row_byte_len > 0 {
|
||||
if let Some(bytes) = reversed_row.as_mut() {
|
||||
bytes.reverse();
|
||||
}
|
||||
let raw = RawHistoryBatchEntry {
|
||||
offset,
|
||||
byte_position: read_start + index as u64 + 1,
|
||||
byte_len: row_byte_len,
|
||||
bytes: reversed_row.take(),
|
||||
};
|
||||
if !retain_newest_row(&mut entries, &mut entries_bytes, raw) {
|
||||
return finish_materialized_batch(file, entries, revision);
|
||||
}
|
||||
let Some(next_offset) = offset.checked_sub(1) else {
|
||||
return finish_materialized_batch(file, entries, revision);
|
||||
};
|
||||
offset = next_offset;
|
||||
reversed_row = Some(vec![b'\n']);
|
||||
row_byte_len = 1;
|
||||
} else {
|
||||
row_byte_len = row_byte_len.saturating_add(1);
|
||||
if let Some(bytes) = reversed_row.as_mut() {
|
||||
if row_byte_len <= MAX_BATCH_BYTES {
|
||||
bytes.push(byte);
|
||||
} else {
|
||||
reversed_row = None;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
read_end = read_start;
|
||||
}
|
||||
|
||||
if row_byte_len > 0 {
|
||||
if let Some(bytes) = reversed_row.as_mut() {
|
||||
bytes.reverse();
|
||||
}
|
||||
retain_newest_row(
|
||||
&mut entries,
|
||||
&mut entries_bytes,
|
||||
RawHistoryBatchEntry {
|
||||
offset,
|
||||
byte_position: 0,
|
||||
byte_len: row_byte_len,
|
||||
bytes: reversed_row,
|
||||
},
|
||||
);
|
||||
}
|
||||
finish_materialized_batch(file, entries, revision)
|
||||
}
|
||||
|
||||
fn finish_materialized_batch(
|
||||
file: &mut File,
|
||||
mut entries: Vec<RawHistoryBatchEntry>,
|
||||
revision: HistoryFileRevision,
|
||||
) -> std::io::Result<HistoryBatch> {
|
||||
for entry in &mut entries {
|
||||
if entry.bytes.is_none() {
|
||||
file.seek(SeekFrom::Start(entry.byte_position))?;
|
||||
let mut bytes = vec![0; entry.byte_len];
|
||||
file.read_exact(&mut bytes)?;
|
||||
entry.bytes = Some(bytes);
|
||||
}
|
||||
}
|
||||
finish_batch(entries, revision)
|
||||
}
|
||||
|
||||
/// Retains the newest suffix seen by a forward scan under both row and byte caps.
|
||||
///
|
||||
/// A single oversized row replaces the suffix so the newest requested record is always returned
|
||||
/// and callers can continue to an older cursor.
|
||||
fn retain_row(
|
||||
suffix: &mut VecDeque<RawHistoryBatchEntry>,
|
||||
suffix_bytes: &mut usize,
|
||||
entry: RawHistoryBatchEntry,
|
||||
) {
|
||||
let row_bytes = entry.byte_len;
|
||||
if row_bytes > MAX_BATCH_BYTES {
|
||||
suffix.clear();
|
||||
*suffix_bytes = row_bytes;
|
||||
suffix.push_back(entry);
|
||||
return;
|
||||
}
|
||||
|
||||
*suffix_bytes += row_bytes;
|
||||
suffix.push_back(entry);
|
||||
while suffix.len() > MAX_BATCH_ROWS || *suffix_bytes > MAX_BATCH_BYTES {
|
||||
if let Some(removed) = suffix.pop_front() {
|
||||
*suffix_bytes -= removed.byte_len;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Appends one newest-to-oldest row and reports whether the backward scan should continue.
|
||||
///
|
||||
/// Returning `false` means the batch is complete. An oversized first row is retained alone;
|
||||
/// otherwise the row that would exceed a cap is left for the next batch.
|
||||
fn retain_newest_row(
|
||||
entries: &mut Vec<RawHistoryBatchEntry>,
|
||||
entries_bytes: &mut usize,
|
||||
entry: RawHistoryBatchEntry,
|
||||
) -> bool {
|
||||
let row_bytes = entry.byte_len;
|
||||
if entries.is_empty() && row_bytes > MAX_BATCH_BYTES {
|
||||
entries.push(entry);
|
||||
return false;
|
||||
}
|
||||
if entries.len() == MAX_BATCH_ROWS || entries_bytes.saturating_add(row_bytes) > MAX_BATCH_BYTES
|
||||
{
|
||||
return false;
|
||||
}
|
||||
*entries_bytes += row_bytes;
|
||||
entries.push(entry);
|
||||
true
|
||||
}
|
||||
|
||||
/// Parses newest-first rows and anchors the continuation at the oldest retained row's start.
|
||||
fn finish_batch(
|
||||
entries: Vec<RawHistoryBatchEntry>,
|
||||
revision: HistoryFileRevision,
|
||||
) -> std::io::Result<HistoryBatch> {
|
||||
let next_older_cursor = entries.last().and_then(|entry| {
|
||||
entry
|
||||
.offset
|
||||
.checked_sub(1)
|
||||
.map(|end_offset| HistoryBatchCursor {
|
||||
end_offset,
|
||||
byte_anchor: revision.modified.map(|_| HistoryByteAnchor {
|
||||
position: entry.byte_position,
|
||||
revision,
|
||||
}),
|
||||
})
|
||||
});
|
||||
let entries = entries
|
||||
.into_iter()
|
||||
.map(|raw| {
|
||||
let bytes = raw.bytes.ok_or_else(|| {
|
||||
std::io::Error::other("retained history row was not materialized")
|
||||
})?;
|
||||
Ok(HistoryBatchEntry {
|
||||
offset: raw.offset,
|
||||
entry: try_parse_entry(&bytes),
|
||||
})
|
||||
})
|
||||
.collect::<std::io::Result<Vec<_>>>()?;
|
||||
Ok(HistoryBatch {
|
||||
entries,
|
||||
next_older_cursor,
|
||||
})
|
||||
}
|
||||
|
||||
fn try_parse_entry(raw: &[u8]) -> Option<HistoryEntry> {
|
||||
let raw = raw.strip_suffix(b"\n").unwrap_or(raw);
|
||||
let raw = raw.strip_suffix(b"\r").unwrap_or(raw);
|
||||
serde_json::from_slice(raw).ok()
|
||||
}
|
||||
359
codex-rs/message-history/src/batch_tests.rs
Normal file
359
codex-rs/message-history/src/batch_tests.rs
Normal file
@@ -0,0 +1,359 @@
|
||||
use std::fs::File;
|
||||
use std::io::Write;
|
||||
|
||||
use codex_config::types::History;
|
||||
use pretty_assertions::assert_eq;
|
||||
use tempfile::TempDir;
|
||||
|
||||
use super::*;
|
||||
|
||||
fn entry(offset: usize, text: impl Into<String>) -> HistoryEntry {
|
||||
HistoryEntry {
|
||||
session_id: "session".to_string(),
|
||||
ts: offset as u64,
|
||||
text: text.into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn write_entries(home: &TempDir, entries: &[HistoryEntry]) -> HistoryConfig {
|
||||
let mut file = File::create(home.path().join(HISTORY_FILENAME)).expect("create history");
|
||||
for entry in entries {
|
||||
serde_json::to_writer(&mut file, entry).expect("serialize entry");
|
||||
writeln!(file).expect("write entry");
|
||||
}
|
||||
HistoryConfig::new(home.path(), &History::default())
|
||||
}
|
||||
|
||||
async fn batch_for(entries: &[HistoryEntry], end_offset: usize) -> (TempDir, HistoryBatch) {
|
||||
let home = TempDir::new().expect("temp dir");
|
||||
let config = write_entries(&home, entries);
|
||||
let (log_id, _) = history_metadata(&config).await;
|
||||
let batch = lookup_batch(log_id, HistoryBatchCursor::new(end_offset), &config)
|
||||
.expect("read history batch");
|
||||
(home, batch)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_batch_returns_bounded_newest_first_absolute_offsets() {
|
||||
let entries: Vec<_> = (0..400)
|
||||
.map(|offset| entry(offset, format!("row {offset}")))
|
||||
.collect();
|
||||
let (home, batch) = batch_for(&entries, /*end_offset*/ 399).await;
|
||||
|
||||
assert_eq!(batch.entries.len(), 128);
|
||||
assert_eq!(batch.entries.first().map(|entry| entry.offset), Some(399));
|
||||
assert_eq!(batch.entries.last().map(|entry| entry.offset), Some(272));
|
||||
let next_cursor = batch.next_older_cursor.expect("older cursor");
|
||||
assert_eq!(next_cursor.end_offset(), 271);
|
||||
assert_eq!(batch.entries[0].entry, Some(entries[399].clone()));
|
||||
|
||||
let config = HistoryConfig::new(home.path(), &History::default());
|
||||
let (log_id, _) = history_metadata(&config).await;
|
||||
let mut offsets: Vec<_> = batch.entries.iter().map(|entry| entry.offset).collect();
|
||||
let mut cursor = Some(next_cursor);
|
||||
while let Some(next_cursor) = cursor {
|
||||
let older = lookup_batch(log_id, next_cursor, &config).expect("read older history batch");
|
||||
offsets.extend(older.entries.iter().map(|entry| entry.offset));
|
||||
cursor = older.next_older_cursor;
|
||||
}
|
||||
assert_eq!(offsets, (0..400).rev().collect::<Vec<_>>());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_batch_invalidates_byte_cursor_after_in_place_rewrite() {
|
||||
let original: Vec<_> = (0..400)
|
||||
.map(|offset| entry(offset, format!("old row {offset}")))
|
||||
.collect();
|
||||
let (home, batch) = batch_for(&original, /*end_offset*/ 399).await;
|
||||
let cursor = batch.next_older_cursor.expect("older cursor");
|
||||
let config = HistoryConfig::new(home.path(), &History::default());
|
||||
let (log_id, _) = history_metadata(&config).await;
|
||||
let original_len = std::fs::metadata(home.path().join(HISTORY_FILENAME))
|
||||
.expect("history metadata")
|
||||
.len();
|
||||
|
||||
let replacement: Vec<_> = (0..500)
|
||||
.map(|offset| entry(offset, format!("replacement row {offset} with padding")))
|
||||
.collect();
|
||||
let replacement_config = write_entries(&home, &replacement);
|
||||
let replacement_len = std::fs::metadata(home.path().join(HISTORY_FILENAME))
|
||||
.expect("replacement history metadata")
|
||||
.len();
|
||||
assert!(replacement_len > original_len);
|
||||
let (replacement_log_id, _) = history_metadata(&replacement_config).await;
|
||||
assert_eq!(replacement_log_id, log_id);
|
||||
|
||||
let older =
|
||||
lookup_batch(log_id, cursor, &replacement_config).expect("read rewritten history batch");
|
||||
assert_eq!(older.entries.len(), 128);
|
||||
assert_eq!(older.entries[0].offset, 271);
|
||||
assert_eq!(older.entries[0].entry, Some(replacement[271].clone()));
|
||||
assert_eq!(older.entries[127].offset, 144);
|
||||
assert_eq!(older.entries[127].entry, Some(replacement[144].clone()));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_batch_rescans_capped_history_after_same_size_rewrite() {
|
||||
let home = TempDir::new().expect("temp dir");
|
||||
let original: Vec<_> = (0..400)
|
||||
.map(|offset| entry(offset, "a".repeat(20)))
|
||||
.collect();
|
||||
write_entries(&home, &original);
|
||||
let path = home.path().join(HISTORY_FILENAME);
|
||||
let original_metadata = std::fs::metadata(&path).expect("history metadata");
|
||||
let original_modified = original_metadata.modified().expect("history modified time");
|
||||
let history = History {
|
||||
max_bytes: Some(original_metadata.len() as usize),
|
||||
..History::default()
|
||||
};
|
||||
let config = HistoryConfig::new(home.path(), &history);
|
||||
let (log_id, _) = history_metadata(&config).await;
|
||||
let first = lookup_batch(log_id, HistoryBatchCursor::new(/*end_offset*/ 399), &config)
|
||||
.expect("read history batch");
|
||||
let cursor = first.next_older_cursor.expect("older cursor");
|
||||
|
||||
let mut replacement: Vec<_> = (0..400)
|
||||
.map(|offset| entry(offset, "b".repeat(20)))
|
||||
.collect();
|
||||
replacement[0].text.push_str(&"b".repeat(10));
|
||||
replacement[399].text.truncate(10);
|
||||
write_entries(&home, &replacement);
|
||||
let file = File::options()
|
||||
.write(true)
|
||||
.open(&path)
|
||||
.expect("open replacement history");
|
||||
file.set_times(std::fs::FileTimes::new().set_modified(original_modified))
|
||||
.expect("restore history modified time");
|
||||
let replacement_metadata = std::fs::metadata(&path).expect("replacement history metadata");
|
||||
assert_eq!(replacement_metadata.len(), original_metadata.len());
|
||||
assert_eq!(
|
||||
replacement_metadata.modified().ok(),
|
||||
Some(original_modified)
|
||||
);
|
||||
|
||||
let older = lookup_batch(log_id, cursor, &config).expect("read rewritten history batch");
|
||||
assert_eq!(
|
||||
older,
|
||||
HistoryBatch {
|
||||
entries: (144..=271)
|
||||
.rev()
|
||||
.map(|offset| HistoryBatchEntry {
|
||||
offset,
|
||||
entry: Some(replacement[offset].clone()),
|
||||
})
|
||||
.collect(),
|
||||
next_older_cursor: Some(HistoryBatchCursor::new(/*end_offset*/ 143)),
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_batch_stitches_chunks_and_keeps_malformed_offsets() {
|
||||
let home = TempDir::new().expect("temp dir");
|
||||
let first = entry(/*offset*/ 0, "a".repeat(HISTORY_READ_BUFFER_SIZE + 17));
|
||||
let third = entry(/*offset*/ 2, "third");
|
||||
let contents = format!(
|
||||
"{}\nnot-json\n{}\n",
|
||||
serde_json::to_string(&first).expect("serialize first"),
|
||||
serde_json::to_string(&third).expect("serialize third")
|
||||
);
|
||||
std::fs::write(home.path().join(HISTORY_FILENAME), contents).expect("write history");
|
||||
let config = HistoryConfig::new(home.path(), &History::default());
|
||||
let (log_id, _) = history_metadata(&config).await;
|
||||
|
||||
assert_eq!(
|
||||
lookup_batch(log_id, HistoryBatchCursor::new(/*end_offset*/ 2), &config,)
|
||||
.expect("read history batch"),
|
||||
HistoryBatch {
|
||||
entries: vec![
|
||||
HistoryBatchEntry {
|
||||
offset: 2,
|
||||
entry: Some(third),
|
||||
},
|
||||
HistoryBatchEntry {
|
||||
offset: 1,
|
||||
entry: None,
|
||||
},
|
||||
HistoryBatchEntry {
|
||||
offset: 0,
|
||||
entry: Some(first),
|
||||
},
|
||||
],
|
||||
next_older_cursor: None,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_batch_preserves_identity_append_trim_and_short_file_semantics() {
|
||||
let home = TempDir::new().expect("temp dir");
|
||||
let initial = vec![entry(/*offset*/ 0, "zero"), entry(/*offset*/ 1, "one")];
|
||||
let config = write_entries(&home, &initial);
|
||||
let (log_id, _) = history_metadata(&config).await;
|
||||
assert_eq!(
|
||||
lookup_batch(
|
||||
log_id.wrapping_add(1),
|
||||
HistoryBatchCursor::new(/*end_offset*/ 1),
|
||||
&config,
|
||||
)
|
||||
.expect("read history batch"),
|
||||
HistoryBatch::default()
|
||||
);
|
||||
|
||||
let mut file = std::fs::OpenOptions::new()
|
||||
.append(true)
|
||||
.open(home.path().join(HISTORY_FILENAME))
|
||||
.expect("open history");
|
||||
serde_json::to_writer(&mut file, &entry(/*offset*/ 2, "appended")).expect("serialize append");
|
||||
writeln!(file).expect("append entry");
|
||||
let batch = lookup_batch(log_id, HistoryBatchCursor::new(/*end_offset*/ 1), &config)
|
||||
.expect("read history batch");
|
||||
assert_eq!(
|
||||
batch.entries,
|
||||
vec![
|
||||
HistoryBatchEntry {
|
||||
offset: 1,
|
||||
entry: Some(initial[1].clone()),
|
||||
},
|
||||
HistoryBatchEntry {
|
||||
offset: 0,
|
||||
entry: Some(initial[0].clone()),
|
||||
},
|
||||
]
|
||||
);
|
||||
|
||||
let newest = "c".repeat(200);
|
||||
let history = History {
|
||||
max_bytes: Some(newest.len() + 80),
|
||||
..History::default()
|
||||
};
|
||||
let trimmed_config = HistoryConfig::new(home.path(), &history);
|
||||
append_entry(&newest, "session", &trimmed_config)
|
||||
.await
|
||||
.expect("append and trim");
|
||||
let trimmed = lookup_batch(
|
||||
log_id,
|
||||
HistoryBatchCursor::new(/*end_offset*/ 20),
|
||||
&trimmed_config,
|
||||
)
|
||||
.expect("read trimmed history batch");
|
||||
assert_eq!(trimmed.entries.len(), 1);
|
||||
assert_eq!(trimmed.entries[0].offset, 0);
|
||||
assert_eq!(
|
||||
trimmed.entries[0].entry.as_ref().map(|entry| &entry.text),
|
||||
Some(&newest)
|
||||
);
|
||||
assert_eq!(trimmed.next_older_cursor, None);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_batch_enforces_byte_cap_and_oversized_row_progress() {
|
||||
let entries: Vec<_> = (0..5)
|
||||
.map(|offset| {
|
||||
entry(
|
||||
offset,
|
||||
char::from(b'a' + offset as u8).to_string().repeat(20_000),
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
let (_home, batch) = batch_for(&entries, /*end_offset*/ 4).await;
|
||||
assert_eq!(batch.entries.len(), 3);
|
||||
assert_eq!(batch.entries.first().map(|entry| entry.offset), Some(4));
|
||||
assert_eq!(batch.entries.last().map(|entry| entry.offset), Some(2));
|
||||
assert_eq!(
|
||||
batch.next_older_cursor.expect("older cursor").end_offset(),
|
||||
1
|
||||
);
|
||||
|
||||
let entries = vec![
|
||||
entry(/*offset*/ 0, "small"),
|
||||
entry(/*offset*/ 1, "x".repeat(70_000)),
|
||||
];
|
||||
let (home, oversized) = batch_for(&entries, /*end_offset*/ 1).await;
|
||||
assert_eq!(oversized.entries.len(), 1);
|
||||
assert_eq!(oversized.entries[0].entry, Some(entries[1].clone()));
|
||||
let next_cursor = oversized.next_older_cursor.expect("older cursor");
|
||||
assert_eq!(next_cursor.end_offset(), 0);
|
||||
let config = HistoryConfig::new(home.path(), &History::default());
|
||||
let (log_id, _) = history_metadata(&config).await;
|
||||
let next = lookup_batch(log_id, next_cursor, &config).expect("read older history batch");
|
||||
assert_eq!(next.entries[0].entry, Some(entries[0].clone()));
|
||||
assert_eq!(next.next_older_cursor, None);
|
||||
|
||||
let entries = vec![
|
||||
entry(/*offset*/ 0, "x".repeat(70_000)),
|
||||
entry(/*offset*/ 1, "newest"),
|
||||
];
|
||||
let (home, newest) = batch_for(&entries, /*end_offset*/ 1).await;
|
||||
assert_eq!(
|
||||
newest.entries,
|
||||
vec![HistoryBatchEntry {
|
||||
offset: 1,
|
||||
entry: Some(entries[1].clone()),
|
||||
}]
|
||||
);
|
||||
let next_cursor = newest.next_older_cursor.expect("older cursor");
|
||||
let config = HistoryConfig::new(home.path(), &History::default());
|
||||
let (log_id, _) = history_metadata(&config).await;
|
||||
let oversized = lookup_batch(log_id, next_cursor, &config).expect("read oversized older row");
|
||||
assert_eq!(
|
||||
oversized,
|
||||
HistoryBatch {
|
||||
entries: vec![HistoryBatchEntry {
|
||||
offset: 0,
|
||||
entry: Some(entries[0].clone()),
|
||||
}],
|
||||
next_older_cursor: None,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_batch_defers_oversized_row_during_backward_scan() {
|
||||
let mut entries = vec![
|
||||
entry(/*offset*/ 0, "oldest"),
|
||||
entry(/*offset*/ 1, "x".repeat(70_000)),
|
||||
];
|
||||
entries.extend((2..8).map(|offset| entry(offset, "x".repeat(20_000))));
|
||||
let (home, newest) = batch_for(&entries, /*end_offset*/ 7).await;
|
||||
assert_eq!(
|
||||
newest
|
||||
.entries
|
||||
.iter()
|
||||
.map(|entry| entry.offset)
|
||||
.collect::<Vec<_>>(),
|
||||
vec![7, 6, 5]
|
||||
);
|
||||
|
||||
let config = HistoryConfig::new(home.path(), &History::default());
|
||||
let (log_id, _) = history_metadata(&config).await;
|
||||
let middle = lookup_batch(
|
||||
log_id,
|
||||
newest.next_older_cursor.expect("middle cursor"),
|
||||
&config,
|
||||
)
|
||||
.expect("read middle history batch");
|
||||
assert_eq!(
|
||||
middle
|
||||
.entries
|
||||
.iter()
|
||||
.map(|entry| entry.offset)
|
||||
.collect::<Vec<_>>(),
|
||||
vec![4, 3, 2]
|
||||
);
|
||||
|
||||
let oversized = lookup_batch(
|
||||
log_id,
|
||||
middle.next_older_cursor.expect("oversized cursor"),
|
||||
&config,
|
||||
)
|
||||
.expect("read oversized history row");
|
||||
assert_eq!(oversized.entries[0].entry, Some(entries[1].clone()));
|
||||
assert_eq!(
|
||||
oversized
|
||||
.next_older_cursor
|
||||
.expect("oldest cursor")
|
||||
.end_offset(),
|
||||
0
|
||||
);
|
||||
}
|
||||
@@ -37,6 +37,12 @@ use tokio::io::AsyncReadExt;
|
||||
use codex_config::types::History;
|
||||
use codex_config::types::HistoryPersistence;
|
||||
|
||||
mod batch;
|
||||
pub use batch::HistoryBatch;
|
||||
pub use batch::HistoryBatchCursor;
|
||||
pub use batch::HistoryBatchEntry;
|
||||
pub use batch::lookup_batch;
|
||||
|
||||
#[cfg(unix)]
|
||||
use std::os::unix::fs::OpenOptionsExt;
|
||||
#[cfg(unix)]
|
||||
@@ -433,5 +439,8 @@ fn log_identity(_metadata: &std::fs::Metadata) -> Option<u64> {
|
||||
None
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "batch_tests.rs"]
|
||||
mod batch_tests;
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
Reference in New Issue
Block a user