use std::fs::{File, OpenOptions};
use std::io;
use std::os::unix::fs::FileExt;
use std::path::Path;
use crate::Lsn;
use crate::crc::crc32c;
use crate::error::{Result, WalError};
use crate::record::{self, Decoded, RECORD_HEADER_SIZE};
pub(crate) const HEADER_SIZE: u64 = 64;
const MAGIC: [u8; 8] = *b"WAL\0SEG1";
const FORMAT_VERSION: u16 = 1;
const VERSION_OFF: usize = 8;
const FLAGS_OFF: usize = 10;
const BASE_LSN_OFF: usize = 12;
const CREATED_OFF: usize = 20;
const HEADER_CRC_OFF: usize = 60;
#[derive(Debug, Clone, Copy)]
pub(crate) struct SegmentHeader {
pub(crate) base_lsn: Lsn,
}
pub(crate) fn encode_header(base_lsn: Lsn, created_unix_nanos: u64) -> [u8; HEADER_SIZE as usize] {
let mut h = [0u8; HEADER_SIZE as usize];
h[0..8].copy_from_slice(&MAGIC);
h[VERSION_OFF..VERSION_OFF + 2].copy_from_slice(&FORMAT_VERSION.to_le_bytes());
let _ = FLAGS_OFF;
h[BASE_LSN_OFF..BASE_LSN_OFF + 8].copy_from_slice(&base_lsn.0.to_le_bytes());
h[CREATED_OFF..CREATED_OFF + 8].copy_from_slice(&created_unix_nanos.to_le_bytes());
let crc = crc32c(&h[0..HEADER_CRC_OFF]);
h[HEADER_CRC_OFF..HEADER_CRC_OFF + 4].copy_from_slice(&crc.to_le_bytes());
h
}
pub(crate) fn decode_header(bytes: &[u8]) -> Result<SegmentHeader> {
if bytes.len() < HEADER_SIZE as usize {
return Err(WalError::BadSegmentHeader);
}
if bytes[0..8] != MAGIC {
return Err(WalError::BadSegmentHeader);
}
let stored_crc = u32::from_le_bytes(
bytes[HEADER_CRC_OFF..HEADER_CRC_OFF + 4]
.try_into()
.unwrap(),
);
if crc32c(&bytes[0..HEADER_CRC_OFF]) != stored_crc {
return Err(WalError::BadSegmentHeader);
}
let version = u16::from_le_bytes(bytes[VERSION_OFF..VERSION_OFF + 2].try_into().unwrap());
if version != FORMAT_VERSION {
return Err(WalError::BadSegmentHeader);
}
let base_lsn = Lsn(u64::from_le_bytes(
bytes[BASE_LSN_OFF..BASE_LSN_OFF + 8].try_into().unwrap(),
));
if base_lsn.is_none() {
return Err(WalError::BadSegmentHeader);
}
Ok(SegmentHeader { base_lsn })
}
pub(crate) fn filename_for(base_lsn: Lsn) -> String {
format!("{:020}.wal", base_lsn.0)
}
pub(crate) fn parse_base_lsn(name: &str) -> Option<u64> {
let digits = name.strip_suffix(".wal")?;
if digits.len() != 20 || !digits.bytes().all(|b| b.is_ascii_digit()) {
return None;
}
digits.parse::<u64>().ok()
}
#[cfg(target_os = "macos")]
pub(crate) fn sync_data_fully(file: &File) -> io::Result<()> {
rustix::fs::fcntl_fullfsync(file).map_err(io::Error::from)
}
#[cfg(not(target_os = "macos"))]
pub(crate) fn sync_data_fully(file: &File) -> io::Result<()> {
file.sync_data()
}
pub(crate) fn create(dir: &Path, base_lsn: Lsn, segment_size: u64) -> Result<File> {
let path = dir.join(filename_for(base_lsn));
let file = OpenOptions::new()
.read(true)
.write(true)
.create_new(true)
.open(&path)?;
match rustix::fs::fallocate(&file, rustix::fs::FallocateFlags::empty(), 0, segment_size) {
Ok(()) => {}
Err(rustix::io::Errno::OPNOTSUPP | rustix::io::Errno::NOSYS) => {
write_zeros(&file, 0, segment_size)?;
}
Err(e) => return Err(io::Error::from(e).into()),
}
let header = encode_header(base_lsn, created_unix_nanos());
file.write_all_at(&header, 0)?;
sync_data_fully(&file)?;
Ok(file)
}
fn created_unix_nanos() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0)
}
#[derive(Debug)]
pub(crate) enum ScanOutcome {
Record {
lsn: Lsn,
payload_len: usize,
framed_len: usize,
},
CleanEnd,
Invalid,
}
fn read_full_at(file: &File, buf: &mut [u8], offset: u64) -> io::Result<bool> {
let mut filled = 0;
while filled < buf.len() {
match file.read_at(&mut buf[filled..], offset + filled as u64) {
Ok(0) => return Ok(false),
Ok(n) => filled += n,
Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue,
Err(e) => return Err(e),
}
}
Ok(true)
}
pub(crate) fn read_record_at(
file: &File,
offset: u64,
segment_size: u64,
max_record_size: u32,
buf: &mut Vec<u8>,
) -> Result<ScanOutcome> {
debug_assert!(
u64::from(max_record_size) + 91 <= segment_size,
"scanner assumes an open()-validated config (§5.3)"
);
let remaining = segment_size.saturating_sub(offset);
if remaining < RECORD_HEADER_SIZE as u64 {
return Ok(ScanOutcome::CleanEnd);
}
buf.resize(RECORD_HEADER_SIZE, 0);
if !read_full_at(file, &mut buf[..RECORD_HEADER_SIZE], offset)? {
return Ok(ScanOutcome::Invalid);
}
if let Decoded::Sentinel = record::decode(&buf[..RECORD_HEADER_SIZE], max_record_size) {
return Ok(ScanOutcome::CleanEnd);
}
let length = u32::from_le_bytes(buf[4..8].try_into().unwrap());
if length > max_record_size {
return Ok(ScanOutcome::Invalid);
}
let framed_u64 = record::framed_size_u64(u64::from(length));
if framed_u64 > remaining {
return Ok(ScanOutcome::Invalid);
}
let framed = framed_u64 as usize;
buf.resize(framed, 0);
if !read_full_at(
file,
&mut buf[RECORD_HEADER_SIZE..framed],
offset + RECORD_HEADER_SIZE as u64,
)? {
return Ok(ScanOutcome::Invalid);
}
match record::decode(&buf[..framed], max_record_size) {
Decoded::Record {
lsn,
payload,
framed_len,
} => {
let payload_len = payload.len();
debug_assert_eq!(framed_len, framed);
Ok(ScanOutcome::Record {
lsn,
payload_len,
framed_len,
})
}
_ => Ok(ScanOutcome::Invalid),
}
}
pub(crate) fn zero_to_eof(file: &File, from: u64, segment_size: u64) -> Result<()> {
write_zeros(file, from, segment_size)?;
if sync_data_fully(file).is_err() {
return Err(WalError::FsyncFailed);
}
Ok(())
}
fn write_zeros(file: &File, from: u64, to: u64) -> io::Result<()> {
const CHUNK: usize = 64 * 1024;
let zeros = [0u8; CHUNK];
let mut off = from;
while off < to {
let n = std::cmp::min(CHUNK as u64, to - off) as usize;
file.write_all_at(&zeros[..n], off)?;
off += n as u64;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn header_round_trips() {
let h = encode_header(Lsn(100_001), 1234);
let decoded = decode_header(&h).expect("valid header");
assert_eq!(decoded.base_lsn, Lsn(100_001));
}
#[test]
fn corrupted_magic_rejected() {
let mut h = encode_header(Lsn(1), 0);
h[0] ^= 0xFF;
assert!(matches!(decode_header(&h), Err(WalError::BadSegmentHeader)));
}
#[test]
fn corrupted_crc_rejected() {
let mut h = encode_header(Lsn(7), 0);
h[BASE_LSN_OFF] ^= 0x01;
assert!(matches!(decode_header(&h), Err(WalError::BadSegmentHeader)));
}
#[test]
fn base_lsn_zero_rejected() {
let h = encode_header(Lsn(0), 0);
assert!(matches!(decode_header(&h), Err(WalError::BadSegmentHeader)));
}
#[test]
fn wrong_version_rejected() {
let mut h = encode_header(Lsn(1), 0);
h[VERSION_OFF] = 2;
let crc = crc32c(&h[0..HEADER_CRC_OFF]);
h[HEADER_CRC_OFF..HEADER_CRC_OFF + 4].copy_from_slice(&crc.to_le_bytes());
assert!(matches!(decode_header(&h), Err(WalError::BadSegmentHeader)));
}
#[test]
fn short_header_rejected() {
assert!(matches!(
decode_header(&[0u8; 10]),
Err(WalError::BadSegmentHeader)
));
}
#[test]
fn filename_round_trips_including_edges() {
for base in [1u64, 42, 100_001, u64::MAX] {
let name = filename_for(Lsn(base));
assert_eq!(name.len(), 24); assert_eq!(parse_base_lsn(&name), Some(base));
}
}
#[test]
fn parse_rejects_non_segment_names() {
assert_eq!(parse_base_lsn("LOCK"), None);
assert_eq!(parse_base_lsn("0000000000000000001.wal"), None); assert_eq!(parse_base_lsn("000000000000000000001.wal"), None); assert_eq!(parse_base_lsn("0000000000000000000a.wal"), None); assert_eq!(parse_base_lsn("00000000000000000001.dat"), None); }
}