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;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct HistoryBatchCursor {
end_offset: usize,
byte_anchor: Option<HistoryByteAnchor>,
}
impl HistoryBatchCursor {
pub fn new(end_offset: usize) -> Self {
Self {
end_offset,
byte_anchor: None,
}
}
pub fn end_offset(self) -> usize {
self.end_offset
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct HistoryByteAnchor {
position: u64,
revision: HistoryFileRevision,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct HistoryFileRevision {
len: u64,
modified: Option<SystemTime>,
}
#[derive(Clone, Debug, PartialEq)]
pub struct HistoryBatchEntry {
pub offset: usize,
pub entry: Option<HistoryEntry>,
}
#[derive(Clone, Debug, Default, PartialEq)]
pub struct HistoryBatch {
pub entries: Vec<HistoryBatchEntry>,
pub next_older_cursor: Option<HistoryBatchCursor>,
}
struct RawHistoryBatchEntry {
offset: usize,
byte_position: u64,
byte_len: usize,
bytes: Option<Vec<u8>>,
}
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",
))
}
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)
}
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)
}
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)
}
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;
}
}
}
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
}
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()
}