use crate::iso_time::{format_iso_milliseconds, now_iso};
use crate::subvolume_error::MigrateError;
use crate::subvolume_file_write_steps::{sync_directory_to_disk, write_all_then_sync};
use serde_json::{Map, Value};
use std::collections::BTreeMap;
use std::os::unix::fs::OpenOptionsExt;
use std::path::{Path, PathBuf};
const SCHEMA: u64 = 1;
const PRIVATE_FILE_MODE: u32 = 0o600;
const TEMPORARY_SUFFIX: &str = ".tmp";
fn is_leftover_name(candidate: &[u8], journal_name: &[u8]) -> bool {
let Some(after_name) = candidate.strip_prefix(journal_name) else { return false };
let Some(rest) = after_name.strip_prefix(TEMPORARY_SUFFIX.as_bytes()) else { return false };
if rest.is_empty() {
return true;
}
let Some(numbers) = rest.strip_prefix(b".") else { return false };
let mut parts = numbers.split(is_dot);
let process: Option<&[u8]> = parts.next();
let clock: Option<&[u8]> = parts.next();
return is_number(process) && is_number(clock) && parts.next().is_none();
}
fn is_dot(byte: &u8) -> bool {
return *byte == b'.';
}
fn is_number(part: Option<&[u8]>) -> bool {
return match part {
Some(digits) => !digits.is_empty() && digits.iter().all(u8::is_ascii_digit),
None => false,
};
}
fn broken_journal(path: &Path, why: &str) -> MigrateError {
return MigrateError::Journal(format!("the journal {} {why}; read it, and delete it only if no migration is in progress", path.display()));
}
fn write_and_rename(sibling: &Path, target: &Path, bytes: &[u8]) -> std::io::Result<()> {
write_exclusively(sibling, bytes)?;
if let Err(error) = std::fs::rename(sibling, target) {
let _removed: std::io::Result<()> = std::fs::remove_file(sibling);
return Err(error);
}
return match target.parent() {
Some(directory) => sync_directory_to_disk(directory),
None => Ok(()),
};
}
fn write_exclusively(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
let mut file: std::fs::File = std::fs::OpenOptions::new().write(true).create_new(true).mode(PRIVATE_FILE_MODE).open(path)?;
let written: std::io::Result<()> = write_all_then_sync(&mut file, bytes);
if let Err(error) = written {
let _removed: std::io::Result<()> = std::fs::remove_file(path);
return Err(error);
}
return Ok(());
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Journal {
path: PathBuf,
started: String,
states: BTreeMap<String, String>,
times: BTreeMap<String, i64>,
persisted: bool,
}
impl Journal {
pub fn load(path: &Path) -> Result<Journal, MigrateError> {
let text: String = match std::fs::read_to_string(path) {
Ok(found) => found,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
return Ok(Journal { path: path.to_path_buf(), started: now_iso(), states: BTreeMap::new(), times: BTreeMap::new(), persisted: false });
}
Err(error) => return Err(MigrateError::io("read the journal", path, &error)),
};
let value: Value = match serde_json::from_str(text.as_str()) {
Ok(parsed) => parsed,
Err(error) => return Err(broken_journal(path, format!("is not valid JSON ({error})").as_str())),
};
if value.get("schema").and_then(Value::as_u64) != Some(SCHEMA) {
return Err(broken_journal(path, "has an unknown schema version"));
}
let Some(started_text) = value.get("started").and_then(Value::as_str) else {
return Err(broken_journal(path, "has no start time"));
};
let started: String = started_text.to_string();
let Some(steps) = value.get("steps").and_then(Value::as_object) else {
return Err(broken_journal(path, "has no steps"));
};
let mut states: BTreeMap<String, String> = BTreeMap::new();
let mut times: BTreeMap<String, i64> = BTreeMap::new();
for (name, record) in steps {
let Some(state) = record.get("state").and_then(Value::as_str) else {
return Err(broken_journal(path, "has a step without a state"));
};
states.insert(name.clone(), state.to_string());
if let Some(milliseconds) = record.get("epoch_ms").and_then(Value::as_i64) {
times.insert(name.clone(), milliseconds);
}
}
return Ok(Journal { path: path.to_path_buf(), started, states, times, persisted: true });
}
pub fn started(&self) -> &str {
return self.started.as_str();
}
pub fn path(&self) -> &Path {
return self.path.as_path();
}
pub fn is_persisted(&self) -> bool {
return self.persisted;
}
pub fn seconds_of(&self, step: &str) -> Option<i64> {
let milliseconds: &i64 = self.times.get(step)?;
return Some(milliseconds.div_euclid(1000));
}
pub fn state(&self, step: &str) -> Option<&str> {
return self.states.get(step).map(String::as_str);
}
pub fn is_done(&self, step: &str) -> bool {
return self.state(step) == Some("done");
}
pub fn has_begun(&self, step: &str) -> bool {
return matches!(self.state(step), Some("started" | "done"));
}
pub fn mark(&mut self, step: &str, state: &str) -> Result<(), MigrateError> {
self.states.insert(step.to_string(), state.to_string());
let now: i64 = match std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) {
Ok(elapsed) => i64::try_from(elapsed.as_millis()).unwrap_or(i64::MAX),
Err(_) => 0,
};
self.times.insert(step.to_string(), now);
let mut steps: Map<String, Value> = Map::new();
for (name, recorded) in &self.states {
let milliseconds: i64 = self.times.get(name).copied().unwrap_or(now);
let mut record: Map<String, Value> = Map::new();
record.insert(String::from("state"), Value::String(recorded.clone()));
record.insert(String::from("at"), Value::String(format_iso_milliseconds(milliseconds)));
record.insert(String::from("epoch_ms"), Value::from(milliseconds));
steps.insert(name.clone(), Value::Object(record));
}
let mut document: Map<String, Value> = Map::new();
document.insert(String::from("schema"), Value::from(SCHEMA));
document.insert(String::from("started"), Value::String(self.started.clone()));
document.insert(String::from("steps"), Value::Object(steps));
let text: String = serde_json::to_string_pretty(&Value::Object(document)).unwrap_or_default();
let sibling_path: PathBuf = self.temporary_path();
let written: std::io::Result<()> = write_and_rename(sibling_path.as_path(), self.path.as_path(), text.as_bytes());
MigrateError::from_io_result(written, "write the journal", self.path.as_path())?;
self.persisted = true;
self.remove_leftovers();
return Ok(());
}
fn temporary_path(&self) -> PathBuf {
let nanos: u128 = match std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) {
Ok(elapsed) => elapsed.as_nanos(),
Err(_) => 0,
};
let mut name: std::ffi::OsString = self.path.as_os_str().to_os_string();
name.push(format!("{TEMPORARY_SUFFIX}.{}.{nanos}", std::process::id()));
return PathBuf::from(name);
}
fn remove_leftovers(&self) {
let (Some(directory), Some(journal_name)) = (self.path.parent(), self.path.file_name()) else { return };
let Ok(listing) = std::fs::read_dir(directory) else { return };
for item in listing.flatten() {
let name: std::ffi::OsString = item.file_name();
if !is_leftover_name(name.as_encoded_bytes(), journal_name.as_encoded_bytes()) {
continue;
}
let is_plain_file: bool = match item.file_type() {
Ok(kind) => kind.is_file(),
Err(_) => false,
};
if is_plain_file {
let _removed: std::io::Result<()> = std::fs::remove_file(item.path());
}
}
}
pub fn delete(&mut self) -> Result<(), MigrateError> {
match std::fs::remove_file(&self.path) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(MigrateError::io("delete the journal", self.path.as_path(), &error)),
}
self.states.clear();
self.times.clear();
self.persisted = false;
return Ok(());
}
}
#[cfg(test)]
#[path = "subvolume_journal_tests.rs"]
mod tests;