use std::{
fs::File,
io::{self, Read},
path::Path,
};
pub(super) const MAX_FILE_BYTES: u64 = 4 * 1024 * 1024;
pub(super) const SMALL_FILE_READ_BYTES: u64 = 128 * 1024;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum ReadKind {
Empty,
Small,
Buffered,
Prefix,
}
#[derive(Debug)]
pub(super) enum ReadOutcome {
Bytes {
bytes: Vec<u8>,
bytes_scanned: u64,
truncated: bool,
#[cfg_attr(not(test), allow(dead_code))]
kind: ReadKind,
},
Skipped,
}
pub(super) fn read_file_bytes(path: &Path) -> io::Result<ReadOutcome> {
let metadata = match path.symlink_metadata() {
Ok(metadata) => metadata,
Err(error)
if matches!(
error.kind(),
io::ErrorKind::NotFound | io::ErrorKind::PermissionDenied
) =>
{
return Ok(ReadOutcome::Skipped);
}
Err(error) => return Err(error),
};
if !metadata.file_type().is_file() {
return Ok(ReadOutcome::Skipped);
}
let size = metadata.len();
if size == 0 {
return Ok(ReadOutcome::Bytes {
bytes: Vec::new(),
bytes_scanned: 0,
truncated: false,
kind: ReadKind::Empty,
});
}
let mut file = match File::open(path) {
Ok(file) => file,
Err(error)
if matches!(
error.kind(),
io::ErrorKind::NotFound | io::ErrorKind::PermissionDenied
) =>
{
return Ok(ReadOutcome::Skipped);
}
Err(error) => return Err(error),
};
if size <= MAX_FILE_BYTES {
let (bytes, grew) = read_bounded(&mut file, MAX_FILE_BYTES, size, true)?;
let kind = if bytes.len() as u64 <= SMALL_FILE_READ_BYTES {
ReadKind::Small
} else {
ReadKind::Buffered
};
return Ok(ReadOutcome::Bytes {
bytes_scanned: bytes.len() as u64,
bytes,
truncated: grew,
kind,
});
}
let (bytes, _) = read_bounded(&mut file, MAX_FILE_BYTES, size, false)?;
Ok(ReadOutcome::Bytes {
bytes_scanned: bytes.len() as u64,
bytes,
truncated: true,
kind: ReadKind::Prefix,
})
}
fn read_bounded<R: Read>(
reader: &mut R,
limit: u64,
capacity_hint: u64,
detect_growth: bool,
) -> io::Result<(Vec<u8>, bool)> {
let read_limit = limit.saturating_add(u64::from(detect_growth));
let mut bytes = Vec::with_capacity(capacity_hint.min(read_limit) as usize);
reader.take(read_limit).read_to_end(&mut bytes)?;
let grew = bytes.len() as u64 > limit;
bytes.truncate(limit as usize);
Ok((bytes, grew))
}
#[cfg(test)]
mod tests {
use super::*;
use std::fs::{self, OpenOptions};
use std::io::{Seek, SeekFrom, Write};
use std::sync::mpsc;
struct GatedReader<R> {
reader: R,
started: Option<mpsc::Sender<()>>,
continue_read: Option<mpsc::Receiver<()>>,
}
impl<R: Read> Read for GatedReader<R> {
fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
if let Some(started) = self.started.take() {
started.send(()).unwrap();
}
if let Some(continue_read) = self.continue_read.take() {
continue_read.recv().unwrap();
}
self.reader.read(buffer)
}
}
#[test]
fn empty_file_uses_empty_branch() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("empty.txt");
File::create(&path).unwrap();
let ReadOutcome::Bytes {
bytes,
bytes_scanned,
kind,
truncated,
} = read_file_bytes(&path).unwrap()
else {
panic!("expected bytes");
};
assert!(bytes.is_empty());
assert_eq!(bytes_scanned, 0);
assert_eq!(kind, ReadKind::Empty);
assert!(!truncated);
}
#[test]
fn small_file_uses_vec_branch() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("small.txt");
fs::write(&path, "needle").unwrap();
let ReadOutcome::Bytes {
bytes,
kind,
truncated,
..
} = read_file_bytes(&path).unwrap()
else {
panic!("expected bytes");
};
assert_eq!(bytes, b"needle");
assert_eq!(kind, ReadKind::Small);
assert!(!truncated);
}
#[test]
fn large_file_uses_buffered_policy_branch() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("large.txt");
fs::write(&path, vec![b'a'; SMALL_FILE_READ_BYTES as usize + 1]).unwrap();
let ReadOutcome::Bytes {
bytes,
kind,
truncated,
..
} = read_file_bytes(&path).unwrap()
else {
panic!("expected bytes");
};
assert_eq!(bytes.len(), SMALL_FILE_READ_BYTES as usize + 1);
assert_eq!(kind, ReadKind::Buffered);
assert!(!truncated);
}
#[test]
fn oversized_file_reads_prefix_only() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("huge.txt");
let mut file = OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.open(&path)
.unwrap();
file.seek(SeekFrom::Start(MAX_FILE_BYTES + 9)).unwrap();
file.write_all(b"x").unwrap();
let ReadOutcome::Bytes {
bytes,
bytes_scanned,
kind,
truncated,
} = read_file_bytes(&path).unwrap()
else {
panic!("expected bytes");
};
assert_eq!(bytes.len(), MAX_FILE_BYTES as usize);
assert_eq!(bytes_scanned, bytes.len() as u64);
assert_eq!(kind, ReadKind::Prefix);
assert!(truncated);
}
#[test]
fn stale_capacity_hint_reads_growth_below_cap() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("stale-hint.txt");
fs::write(&path, vec![b'a'; (SMALL_FILE_READ_BYTES + 1) as usize]).unwrap();
let mut file = File::open(&path).unwrap();
let (bytes, grew) = read_bounded(&mut file, MAX_FILE_BYTES, 5, true).unwrap();
assert_eq!(bytes.len(), (SMALL_FILE_READ_BYTES + 1) as usize);
assert!(!grew);
}
#[test]
fn stale_capacity_hint_probes_beyond_cap() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("stale-hint-probe.txt");
fs::write(&path, vec![b'a'; (MAX_FILE_BYTES + 1) as usize]).unwrap();
let mut file = File::open(&path).unwrap();
let (bytes, grew) = read_bounded(&mut file, MAX_FILE_BYTES, 5, true).unwrap();
assert_eq!(bytes.len(), MAX_FILE_BYTES as usize);
assert!(grew);
assert!(bytes.len() <= MAX_FILE_BYTES as usize);
}
#[test]
fn bounded_read_overlaps_mutation_deterministically() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("concurrent.txt");
fs::write(&path, b"initial").unwrap();
let mut file = File::open(&path).unwrap();
let (started_tx, started_rx) = mpsc::channel();
let (continue_tx, continue_rx) = mpsc::channel();
let (mutation_done_tx, mutation_done_rx) = mpsc::channel();
let writer_path = path.clone();
let writer = std::thread::spawn(move || {
started_rx.recv().unwrap();
OpenOptions::new()
.write(true)
.open(&writer_path)
.unwrap()
.set_len(MAX_FILE_BYTES + 1)
.unwrap();
mutation_done_tx.send(()).unwrap();
continue_tx.send(()).unwrap();
});
let mut reader = GatedReader {
reader: &mut file,
started: Some(started_tx),
continue_read: Some(continue_rx),
};
let (bytes, grew) = read_bounded(&mut reader, MAX_FILE_BYTES, 7, true).unwrap();
mutation_done_rx.recv().unwrap();
writer.join().unwrap();
assert_eq!(bytes.len(), MAX_FILE_BYTES as usize);
assert!(grew);
assert!(bytes.len() <= MAX_FILE_BYTES as usize);
}
#[test]
fn bounded_read_handles_truncation_after_read_starts() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("concurrent-truncate.txt");
fs::write(&path, vec![b'a'; (SMALL_FILE_READ_BYTES + 1) as usize]).unwrap();
let mut file = File::open(&path).unwrap();
let (started_tx, started_rx) = mpsc::channel();
let (continue_tx, continue_rx) = mpsc::channel();
let (mutation_done_tx, mutation_done_rx) = mpsc::channel();
let writer_path = path.clone();
let writer = std::thread::spawn(move || {
started_rx.recv().unwrap();
OpenOptions::new()
.write(true)
.open(&writer_path)
.unwrap()
.set_len(0)
.unwrap();
mutation_done_tx.send(()).unwrap();
continue_tx.send(()).unwrap();
});
let mut reader = GatedReader {
reader: &mut file,
started: Some(started_tx),
continue_read: Some(continue_rx),
};
let (bytes, grew) =
read_bounded(&mut reader, MAX_FILE_BYTES, SMALL_FILE_READ_BYTES + 1, true).unwrap();
mutation_done_rx.recv().unwrap();
writer.join().unwrap();
assert!(bytes.is_empty());
assert!(!grew);
assert!(bytes.len() <= MAX_FILE_BYTES as usize);
}
#[test]
fn opened_file_read_handles_truncation_without_exceeding_cap() {
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("truncated.txt");
let initial_size = SMALL_FILE_READ_BYTES + 1;
fs::write(&path, vec![b'a'; initial_size as usize]).unwrap();
let mut file = File::open(&path).unwrap();
OpenOptions::new()
.write(true)
.open(&path)
.unwrap()
.set_len(0)
.unwrap();
let (bytes, grew) = read_bounded(&mut file, MAX_FILE_BYTES, initial_size, true).unwrap();
assert_eq!(bytes.len(), 0);
assert!(!grew);
}
#[cfg(unix)]
#[test]
fn opened_file_read_is_stable_after_concurrent_path_replacement() {
use std::fs::rename;
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("replaced.txt");
let initial_size = SMALL_FILE_READ_BYTES + 1;
fs::write(&path, vec![b'a'; initial_size as usize]).unwrap();
let file = File::open(&path).unwrap();
let replacement = temp.path().join("replacement.txt");
fs::write(&replacement, b"replacement").unwrap();
let (started_tx, started_rx) = mpsc::channel();
let (continue_tx, continue_rx) = mpsc::channel();
let (mutation_done_tx, mutation_done_rx) = mpsc::channel();
let writer_path = path.clone();
let writer = std::thread::spawn(move || {
started_rx.recv().unwrap();
rename(&replacement, &writer_path).unwrap();
mutation_done_tx.send(()).unwrap();
});
let reader = std::thread::spawn(move || {
let mut reader = GatedReader {
reader: file,
started: Some(started_tx),
continue_read: Some(continue_rx),
};
read_bounded(&mut reader, MAX_FILE_BYTES, initial_size, true)
});
mutation_done_rx.recv().unwrap();
continue_tx.send(()).unwrap();
writer.join().unwrap();
let (bytes, grew) = reader.join().unwrap().unwrap();
assert!(bytes.len() <= MAX_FILE_BYTES as usize);
assert_eq!(bytes.len(), initial_size as usize);
assert_eq!(bytes[0], b'a');
assert!(!grew);
}
#[cfg(unix)]
#[test]
fn fifo_is_skipped_before_open() {
use std::process::Command;
let temp = tempfile::TempDir::new().unwrap();
let path = temp.path().join("pipe");
let status = Command::new("mkfifo").arg(&path).status().unwrap();
assert!(status.success());
assert!(matches!(
read_file_bytes(&path).unwrap(),
ReadOutcome::Skipped
));
}
}