mirror of
https://github.com/openai/codex.git
synced 2026-09-09 15:58:47 +00:00
Add bounded message history batches
This commit is contained in:
Charlie Marsh
330
codex-rs/message-history/src/batch.rs
Normal file
330
codex-rs/message-history/src/batch.rs
Normal file
@@ -0,0 +1,330 @@
|
||||
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.
|
||||
#[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.
|
||||
///
|
||||
/// A length alone cannot detect an in-place trim followed by an append, so cursors also retain the
|
||||
/// modification time. Filesystems without a modification time always 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`. It carries a byte position because the byte cap can make a batch contain fewer than
|
||||
/// 128 rows and because continuation lookups must not rescan already traversed prefixes.
|
||||
#[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,
|
||||
bytes: 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),
|
||||
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 when the file revision is unchanged.
|
||||
///
|
||||
/// Falling back to the forward scan preserves absolute row semantics after concurrent appends,
|
||||
/// trims, or rewrites, at the cost of rescanning that one request from the beginning.
|
||||
fn scan_batch(file: &mut File, cursor: HistoryBatchCursor) -> std::io::Result<HistoryBatch> {
|
||||
let metadata = file.metadata()?;
|
||||
let revision = HistoryFileRevision {
|
||||
len: metadata.len(),
|
||||
modified: metadata.modified().ok(),
|
||||
};
|
||||
if 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))?;
|
||||
scan_batch_forward(file, cursor.end_offset, revision)
|
||||
}
|
||||
|
||||
/// 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.
|
||||
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, file);
|
||||
|
||||
for offset in 0..=end_offset {
|
||||
let mut bytes = Vec::new();
|
||||
let read = reader.read_until(b'\n', &mut bytes)?;
|
||||
if read == 0 {
|
||||
break;
|
||||
}
|
||||
retain_row(&mut suffix, &mut suffix_bytes, offset, byte_position, bytes);
|
||||
byte_position += read as u64;
|
||||
}
|
||||
Ok(finish_batch(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 = Vec::new();
|
||||
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' && !reversed_row.is_empty() {
|
||||
reversed_row.reverse();
|
||||
let raw = RawHistoryBatchEntry {
|
||||
offset,
|
||||
byte_position: read_start + index as u64 + 1,
|
||||
bytes: std::mem::take(&mut reversed_row),
|
||||
};
|
||||
if !retain_newest_row(&mut entries, &mut entries_bytes, raw) {
|
||||
return Ok(finish_batch(entries, revision));
|
||||
}
|
||||
let Some(next_offset) = offset.checked_sub(1) else {
|
||||
return Ok(finish_batch(entries, revision));
|
||||
};
|
||||
offset = next_offset;
|
||||
reversed_row.push(b'\n');
|
||||
} else {
|
||||
reversed_row.push(byte);
|
||||
}
|
||||
}
|
||||
read_end = read_start;
|
||||
}
|
||||
|
||||
if !reversed_row.is_empty() {
|
||||
reversed_row.reverse();
|
||||
retain_newest_row(
|
||||
&mut entries,
|
||||
&mut entries_bytes,
|
||||
RawHistoryBatchEntry {
|
||||
offset,
|
||||
byte_position: 0,
|
||||
bytes: reversed_row,
|
||||
},
|
||||
);
|
||||
}
|
||||
Ok(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,
|
||||
offset: usize,
|
||||
byte_position: u64,
|
||||
bytes: Vec<u8>,
|
||||
) {
|
||||
let row_bytes = bytes.len();
|
||||
if row_bytes > MAX_BATCH_BYTES {
|
||||
suffix.clear();
|
||||
*suffix_bytes = row_bytes;
|
||||
suffix.push_back(RawHistoryBatchEntry {
|
||||
offset,
|
||||
byte_position,
|
||||
bytes,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
*suffix_bytes += row_bytes;
|
||||
suffix.push_back(RawHistoryBatchEntry {
|
||||
offset,
|
||||
byte_position,
|
||||
bytes,
|
||||
});
|
||||
while suffix.len() > MAX_BATCH_ROWS || *suffix_bytes > MAX_BATCH_BYTES {
|
||||
if let Some(removed) = suffix.pop_front() {
|
||||
*suffix_bytes -= removed.bytes.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.bytes.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) -> 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| HistoryBatchEntry {
|
||||
offset: raw.offset,
|
||||
entry: parse_entry(&raw.bytes),
|
||||
})
|
||||
.collect();
|
||||
HistoryBatch {
|
||||
entries,
|
||||
next_older_cursor,
|
||||
}
|
||||
}
|
||||
|
||||
fn 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()
|
||||
}
|
||||
227
codex-rs/message-history/src/batch_tests.rs
Normal file
227
codex-rs/message-history/src/batch_tests.rs
Normal file
@@ -0,0 +1,227 @@
|
||||
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_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);
|
||||
}
|
||||
@@ -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