use std::fs::{File, OpenOptions};
use std::io;
use std::path::Path;
use crate::error::{Error, ErrorContext, Result};
use crate::format::scan::FileScan;
use crate::limits::Limits;
use crate::lock::acquire_writer_lock;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum RecoveryAction {
None,
TruncateIncompleteTail,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct RecoveryPlan {
action: RecoveryAction,
file_size: u64,
last_good_offset: u64,
bytes_to_remove: u64,
}
impl RecoveryPlan {
pub fn action(&self) -> RecoveryAction {
self.action
}
pub fn file_size(&self) -> u64 {
self.file_size
}
pub fn last_good_offset(&self) -> u64 {
self.last_good_offset
}
pub fn bytes_to_remove(&self) -> u64 {
self.bytes_to_remove
}
pub fn requires_repair(&self) -> bool {
self.action == RecoveryAction::TruncateIncompleteTail
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct RecoverySummary {
original_file_size: u64,
repaired_file_size: u64,
bytes_removed: u64,
}
impl RecoverySummary {
pub fn original_file_size(&self) -> u64 {
self.original_file_size
}
pub fn repaired_file_size(&self) -> u64 {
self.repaired_file_size
}
pub fn bytes_removed(&self) -> u64 {
self.bytes_removed
}
}
pub fn inspect_recovery<P: AsRef<Path>>(path: P) -> Result<RecoveryPlan> {
inspect_recovery_with_limits(path, Limits::default())
}
pub fn inspect_recovery_with_limits<P: AsRef<Path>>(
path: P,
limits: Limits,
) -> Result<RecoveryPlan> {
let scan = FileScan::open(path.as_ref(), limits)?;
let (_file, plan) = discover(scan)?;
Ok(plan)
}
pub fn repair_incomplete_tail<P: AsRef<Path>>(path: P) -> Result<RecoverySummary> {
repair_incomplete_tail_with_limits(path, Limits::default())
}
pub fn repair_incomplete_tail_with_limits<P: AsRef<Path>>(
path: P,
limits: Limits,
) -> Result<RecoverySummary> {
let file = OpenOptions::new()
.read(true)
.write(true)
.open(path.as_ref())
.map_err(|error| Error::io(error, None).with_context(ErrorContext::File))?;
acquire_writer_lock(&file)?;
repair_locked_file(file, limits, &FileRepairOperations)
}
trait RepairOperations {
fn physical_length(&self, file: &File) -> io::Result<u64>;
fn set_len(&self, file: &File, length: u64) -> io::Result<()>;
fn sync(&self, file: &File) -> io::Result<()>;
}
struct FileRepairOperations;
impl RepairOperations for FileRepairOperations {
fn physical_length(&self, file: &File) -> io::Result<u64> {
file.metadata().map(|metadata| metadata.len())
}
fn set_len(&self, file: &File, length: u64) -> io::Result<()> {
file.set_len(length)
}
fn sync(&self, file: &File) -> io::Result<()> {
file.sync_all()
}
}
fn discover(mut scan: FileScan) -> Result<(File, RecoveryPlan)> {
let walk = scan.walk_data_frames(|_frame, _block| Ok(()))?;
let plan = RecoveryPlan::from_walk(
scan.file_size(),
walk.last_good_offset,
walk.incomplete_tail,
)?;
Ok((scan.into_file(), plan))
}
fn repair_locked_file(
file: File,
limits: Limits,
operations: &dyn RepairOperations,
) -> Result<RecoverySummary> {
let scan = FileScan::from_file(file, limits)?;
let (file, plan) = discover(scan)?;
let current_size = operations
.physical_length(&file)
.map_err(|error| Error::io(error, None).with_context(ErrorContext::File))?;
if current_size != plan.file_size {
return Err(Error::io(
io::Error::other("the file changed while recovery was being inspected"),
Some(current_size),
)
.with_context(ErrorContext::File));
}
if !plan.requires_repair() {
return Err(Error::invalid_argument(
"the file is complete; there is no incomplete tail to repair",
)
.with_context(ErrorContext::File));
}
operations
.set_len(&file, plan.last_good_offset)
.map_err(|error| {
Error::io(error, Some(plan.last_good_offset))
.with_message_prefix(BEFORE_TRUNCATION)
.with_context(ErrorContext::File)
})?;
finish_repair(file, limits, &plan, operations)
.map_err(|error| error.with_message_prefix(AFTER_TRUNCATION))
}
const BEFORE_TRUNCATION: &str = "the incomplete tail was not removed and the file is unchanged";
const AFTER_TRUNCATION: &str =
"the incomplete tail was already removed, so the file may already be shorter than it was";
fn finish_repair(
file: File,
limits: Limits,
plan: &RecoveryPlan,
operations: &dyn RepairOperations,
) -> Result<RecoverySummary> {
operations.sync(&file).map_err(|error| {
Error::io(error, Some(plan.last_good_offset))
.with_message_prefix("the repaired file could not be synchronized")
.with_context(ErrorContext::File)
})?;
let post_scan = FileScan::from_file(file, limits)?;
let (file, post_plan) = discover(post_scan)?;
if post_plan.action != RecoveryAction::None
|| post_plan.file_size != plan.last_good_offset
|| post_plan.last_good_offset != plan.last_good_offset
{
return Err(Error::corruption(
"post-repair validation did not produce the expected complete file",
Some(plan.last_good_offset),
)
.with_context(ErrorContext::File));
}
drop(file);
let bytes_removed = plan
.file_size
.checked_sub(post_plan.file_size)
.ok_or_else(|| {
Error::corruption(
"repaired file grew beyond its original size",
Some(post_plan.file_size),
)
.with_context(ErrorContext::File)
})?;
Ok(RecoverySummary {
original_file_size: plan.file_size,
repaired_file_size: post_plan.file_size,
bytes_removed,
})
}
impl RecoveryPlan {
fn from_walk(file_size: u64, last_good_offset: u64, incomplete_tail: bool) -> Result<Self> {
let bytes_to_remove = if incomplete_tail {
if last_good_offset >= file_size {
return Err(Error::corruption(
"an incomplete tail does not extend beyond the last complete frame",
Some(last_good_offset),
)
.with_context(ErrorContext::File));
}
file_size.checked_sub(last_good_offset).ok_or_else(|| {
Error::corruption(
"the last complete frame is beyond the captured file extent",
Some(last_good_offset),
)
.with_context(ErrorContext::File)
})?
} else {
if last_good_offset != file_size {
return Err(Error::corruption(
"a complete recovery walk did not reach the file extent",
Some(last_good_offset),
)
.with_context(ErrorContext::File));
}
0
};
let action = if incomplete_tail {
RecoveryAction::TruncateIncompleteTail
} else {
RecoveryAction::None
};
Ok(Self {
action,
file_size,
last_good_offset,
bytes_to_remove,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
#[derive(Default)]
struct FailingOperations {
length: Option<u64>,
fail_length: bool,
fail_set_len: bool,
fail_sync: bool,
length_delta: i64,
}
impl RepairOperations for FailingOperations {
fn physical_length(&self, file: &File) -> io::Result<u64> {
if self.fail_length {
return Err(io::Error::other("injected metadata failure"));
}
match self.length {
Some(length) => Ok(length),
None => file.metadata().map(|metadata| metadata.len()),
}
}
fn set_len(&self, file: &File, length: u64) -> io::Result<()> {
if self.fail_set_len {
return Err(io::Error::other("injected set_len failure"));
}
file.set_len(length.wrapping_add(self.length_delta as u64))
}
fn sync(&self, file: &File) -> io::Result<()> {
if self.fail_sync {
return Err(io::Error::other("injected sync failure"));
}
file.sync_all()
}
}
#[test]
fn plan_requires_removal_only_for_an_incomplete_data_tail() {
let complete = RecoveryPlan::from_walk(10, 10, false).unwrap();
assert_eq!(complete.action(), RecoveryAction::None);
assert_eq!(complete.bytes_to_remove(), 0);
let incomplete = RecoveryPlan::from_walk(14, 10, true).unwrap();
assert_eq!(incomplete.action(), RecoveryAction::TruncateIncompleteTail);
assert!(incomplete.requires_repair());
assert_eq!(incomplete.bytes_to_remove(), 4);
}
#[test]
fn injected_set_len_failure_preserves_the_file() {
let fixture = Fixture::incomplete();
let error = fixture.repair(FailingOperations {
fail_set_len: true,
..FailingOperations::default()
});
assert_eq!(error.kind(), crate::ErrorKind::Io);
assert!(error.message().starts_with(BEFORE_TRUNCATION), "{error}");
assert!(!error.message().contains(AFTER_TRUNCATION), "{error}");
assert_eq!(fixture.length(), fixture.original_size);
assert_eq!(fixture.bytes(), fixture.original_bytes);
}
#[test]
fn injected_sync_failure_does_not_claim_durability() {
let fixture = Fixture::incomplete();
let error = fixture.repair(FailingOperations {
fail_sync: true,
..FailingOperations::default()
});
assert_eq!(error.kind(), crate::ErrorKind::Io);
assert!(error.message().starts_with(AFTER_TRUNCATION), "{error}");
assert!(!error.message().contains(BEFORE_TRUNCATION), "{error}");
assert!(fixture.length() < fixture.original_size);
}
#[test]
fn a_length_that_moved_under_the_plan_refuses_before_truncating() {
for reported in [0, 1, u64::MAX] {
let fixture = Fixture::incomplete();
let error = fixture.repair(FailingOperations {
length: Some(reported),
..FailingOperations::default()
});
assert_eq!(error.kind(), crate::ErrorKind::Io);
assert!(error.message().contains("changed"), "{error}");
assert_eq!(fixture.bytes(), fixture.original_bytes);
}
}
#[test]
fn an_unreadable_length_refuses_before_truncating() {
let fixture = Fixture::incomplete();
let error = fixture.repair(FailingOperations {
fail_length: true,
..FailingOperations::default()
});
assert_eq!(error.kind(), crate::ErrorKind::Io);
assert_eq!(fixture.bytes(), fixture.original_bytes);
}
#[test]
fn a_wrong_truncation_target_fails_post_repair_validation() {
for delta in [-64_i64, -8, -1, 1, 8, 64] {
let fixture = Fixture::incomplete();
let error = fixture.repair(FailingOperations {
length_delta: delta,
..FailingOperations::default()
});
assert_eq!(error.kind(), crate::ErrorKind::Corruption, "delta {delta}");
assert!(
error.message().starts_with(AFTER_TRUNCATION),
"delta {delta}: {error}"
);
assert!(!error.message().contains(BEFORE_TRUNCATION), "{error}");
}
}
struct Fixture {
path: PathBuf,
original_bytes: Vec<u8>,
original_size: u64,
}
impl Fixture {
fn incomplete() -> Self {
static NEXT_FIXTURE: AtomicU64 = AtomicU64::new(0);
let mut bytes = include_bytes!("../spec/v0.2/fixtures/minimal/minimal.acta").to_vec();
bytes.push(0);
let path = std::env::temp_dir().join(format!(
"acta-recovery-unit-{}-{}.acta",
std::process::id(),
NEXT_FIXTURE.fetch_add(1, Ordering::Relaxed)
));
let _ = std::fs::remove_file(&path);
std::fs::write(&path, &bytes).unwrap();
Self {
path,
original_size: bytes.len() as u64,
original_bytes: bytes,
}
}
fn repair(&self, operations: FailingOperations) -> Error {
let file = OpenOptions::new()
.read(true)
.write(true)
.open(&self.path)
.unwrap();
acquire_writer_lock(&file).unwrap();
match repair_locked_file(file, Limits::default(), &operations) {
Ok(summary) => panic!("repair unexpectedly succeeded: {summary:?}"),
Err(error) => error,
}
}
fn length(&self) -> u64 {
std::fs::metadata(&self.path).unwrap().len()
}
fn bytes(&self) -> Vec<u8> {
std::fs::read(&self.path).unwrap()
}
}
impl Drop for Fixture {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.path);
}
}
}