use std::path::Path;
use crate::error::{Result, WalError};
use crate::recovery::recover;
use super::atomic_io::fsync_directory;
use super::meta::SegmentMeta;
pub(crate) fn discard_unused_segment(wal_dir: &Path, path: &Path, roll_err: WalError) -> WalError {
let removed = match std::fs::remove_file(path) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e),
};
let cleanup = removed
.map_err(WalError::Io)
.and_then(|()| fsync_directory(wal_dir));
match cleanup {
Ok(()) => roll_err,
Err(cleanup_err) => WalError::RollCleanupFailed {
path: path.display().to_string(),
roll: Box::new(roll_err),
cleanup: Box::new(cleanup_err),
},
}
}
pub fn check_resume_above_previous(segments: &[SegmentMeta], resumed_next_lsn: u64) -> Result<()> {
let Some((last, earlier)) = segments.split_last() else {
return Ok(());
};
for previous in earlier.iter().rev() {
let info = recover(&previous.path)?;
if info.record_count == 0 {
continue;
}
if info.last_lsn >= resumed_next_lsn {
return Err(WalError::SegmentOverlapsPrevious {
path: last.path.display().to_string(),
first_lsn: last.first_lsn,
previous_path: previous.path.display().to_string(),
previous_last_lsn: info.last_lsn,
});
}
return Ok(());
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::record::RecordType;
use crate::segment::{discover_segments, segment_path};
use crate::writer::{WalWriter, WalWriterConfig};
fn config() -> WalWriterConfig {
WalWriterConfig {
use_direct_io: false,
..Default::default()
}
}
fn write_segment(dir: &Path, first_lsn: u64, records: u64) {
let path = segment_path(dir, first_lsn);
let mut writer = WalWriter::open_with_start_lsn(&path, config(), first_lsn).unwrap();
for i in 0..records {
writer
.append(RecordType::Put as u32, 1, 0, 0, &i.to_le_bytes())
.unwrap();
}
writer.sync().unwrap();
}
#[test]
fn an_empty_segment_inside_the_previous_segments_lsns_is_refused() {
let dir = tempfile::tempdir().unwrap();
write_segment(dir.path(), 1, 5);
write_segment(dir.path(), 3, 0);
let segments = discover_segments(dir.path()).unwrap();
let err = check_resume_above_previous(&segments, 3).unwrap_err();
assert!(
matches!(
err,
WalError::SegmentOverlapsPrevious {
first_lsn: 3,
previous_last_lsn: 5,
..
}
),
"{err}"
);
}
#[test]
fn an_empty_segment_after_the_previous_one_resumes() {
let dir = tempfile::tempdir().unwrap();
write_segment(dir.path(), 1, 5);
write_segment(dir.path(), 6, 0);
let segments = discover_segments(dir.path()).unwrap();
check_resume_above_previous(&segments, 6).unwrap();
}
#[test]
fn discard_removes_the_file_and_keeps_the_roll_error() {
let dir = tempfile::tempdir().unwrap();
let path = segment_path(dir.path(), 9);
std::fs::write(&path, b"").unwrap();
let err = discard_unused_segment(dir.path(), &path, WalError::Sealed);
assert!(matches!(err, WalError::Sealed), "{err}");
assert!(!path.exists());
}
#[test]
fn discard_of_a_missing_file_keeps_the_roll_error() {
let dir = tempfile::tempdir().unwrap();
let path = segment_path(dir.path(), 9);
let err = discard_unused_segment(dir.path(), &path, WalError::Sealed);
assert!(matches!(err, WalError::Sealed), "{err}");
}
#[test]
fn a_failed_removal_names_both_failures() {
let dir = tempfile::tempdir().unwrap();
let path = segment_path(dir.path(), 9);
std::fs::create_dir(&path).unwrap();
std::fs::write(path.join("x"), b"x").unwrap();
let err = discard_unused_segment(dir.path(), &path, WalError::Sealed);
assert!(matches!(err, WalError::RollCleanupFailed { .. }), "{err}");
let msg = err.to_string();
assert!(msg.contains("sealed") || msg.contains("Sealed"), "{msg}");
assert!(msg.contains(&path.display().to_string()), "{msg}");
}
}