use super::document::TextDocument;
use super::error::TextStoreError;
use super::store::{meta_at, meta_key};
use crate::errors::StorageError;
use crate::store::StorageResult;
use crate::store::config::FileWritePolicy;
use crate::store::facts::{Facts, StoreFile as StoreFileFact};
use crate::store::screening::Noticed;
use crate::store::traits::StoreLayout;
use amethystate_core::path::StorePath;
use error_stack::ResultExt;
use parking_lot::{Mutex, RwLock};
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tempfile::NamedTempFile;
use tracing::{error, warn};
#[derive(Clone)]
pub struct StoreFile<D> {
pub path: PathBuf,
pub backup_path: PathBuf,
pub doc: Arc<RwLock<D>>,
pub write_policy: FileWritePolicy,
flush: Arc<Mutex<()>>,
wrote: Arc<Mutex<Option<u128>>>,
}
impl<D: TextDocument> StoreFile<D> {
pub fn new(path: PathBuf, initial_doc: D, write_policy: FileWritePolicy) -> Self {
let backup_path = StoreLayout::rewrite_copy_of(&path);
Self {
path,
backup_path,
doc: Arc::new(RwLock::new(initial_doc)),
write_policy,
flush: Arc::new(Mutex::new(())),
wrote: Arc::new(Mutex::new(None)),
}
}
pub fn create_backup(&self) -> StorageResult<()> {
if self.path.exists() {
std::fs::copy(&self.path, &self.backup_path)
.map_err(TextStoreError::from)
.change_context(StorageError::Open)
.attach_store_file(&self.path)
.attach_with(|| format!("backup: {}", self.backup_path.display()))?;
}
Ok(())
}
pub fn read_or_recover(&self) -> StorageResult<D> {
self.read_or_recover_unless(|_| None)
}
pub fn read_or_recover_unless(
&self,
suspect: impl Fn(&D) -> Option<&'static str>,
) -> StorageResult<D> {
let read = match self.lost_its_file() {
true => Err(error_stack::Report::new(StorageError::Open)
.attach(StoreFileFact(self.path.clone()))
.attach("the file is gone and the copy kept for it is not")),
false => self.load_or_empty().and_then(|doc| match suspect(&doc) {
Some(why) => Err(error_stack::Report::new(StorageError::Open)
.attach(StoreFileFact(self.path.clone()))
.attach(why)),
None => Ok(doc),
}),
};
match read {
Ok(doc) => Ok(doc),
Err(unreadable) => match self.recover_from_backup() {
Some(doc) => {
warn!(
path = %self.path.display(),
backup = %self.backup_path.display(),
"the file could not be read and was restored from the backup a \
previous open left behind"
);
Ok(doc)
}
None => Err(unreadable),
},
}
}
fn lost_its_file(&self) -> bool {
!self.path.exists() && self.backup_path.exists()
}
fn recover_from_backup(&self) -> Option<D> {
if !self.backup_path.exists() {
return None;
}
let content = std::fs::read_to_string(&self.backup_path).ok()?;
let doc = D::parse(&content).ok()?;
self.forget_what_the_file_holds();
std::fs::copy(&self.backup_path, &self.path).ok()?;
Some(doc)
}
pub fn load_or_empty(&self) -> StorageResult<D> {
if self.path.exists() {
let content = std::fs::read_to_string(&self.path)
.map_err(TextStoreError::from)
.change_context(StorageError::Open)
.attach_store_file(&self.path)?;
D::parse(&content).attach_store_file(&self.path)
} else {
Ok(D::empty())
}
}
pub fn persist(&self) -> StorageResult<()> {
self.persist_while(None).map(|_| ())
}
pub(crate) fn persist_while(&self, left: Option<Standing>) -> StorageResult<Wrote> {
let _flushing = self.flush.lock();
*self.wrote.lock() = None;
let content = self.doc.read().serialize().attach_store_file(&self.path)?;
let still = || match left {
Some(left) => standing_of(&self.path) == Some(left),
None => true,
};
let replaced = persist_atomic(&self.path, &content, self.write_policy, &still)
.map_err(TextStoreError::from)
.change_context(StorageError::Flush)
.attach_store_file(&self.path)?;
if replaced {
*self.wrote.lock() = Some(hash_of(&content));
}
Ok(match replaced {
true => Wrote::Replaced(what_we_left(&self.path, &content)),
false => Wrote::FileMoved,
})
}
pub(crate) fn persist_document(&self, doc: &D) -> StorageResult<()> {
let _flushing = self.flush.lock();
*self.wrote.lock() = None;
let content = doc.serialize().attach_store_file(&self.path)?;
persist_atomic(&self.path, &content, self.write_policy, &|| true)
.map_err(TextStoreError::from)
.change_context(StorageError::Flush)
.attach_store_file(&self.path)?;
*self.wrote.lock() = Some(hash_of(&content));
Ok(())
}
pub(crate) fn wrote_exactly(&self, content: &str) -> bool {
matches!(*self.wrote.lock(), Some(held) if held == hash_of(content))
}
pub(crate) fn agrees_with(&self, content: &str) {
*self.wrote.lock() = Some(hash_of(content));
}
pub(crate) fn forget_what_the_file_holds(&self) {
*self.wrote.lock() = None;
}
pub fn restore_from_backup(&self, fallback_to_initial: &D) -> bool {
*self.doc.write() = fallback_to_initial.clone();
self.forget_what_the_file_holds();
if self.backup_path.exists() {
if let Err(io) = std::fs::copy(&self.backup_path, &self.path) {
error!(
file = %self.path.display(),
backup = %self.backup_path.display(),
error = %io,
"the copy taken at the start of this open could not be put back, so the \
file holds what the open that failed had written"
);
return false;
}
self.remove_backup("after putting it back");
} else if self.path.exists()
&& let Err(io) = std::fs::remove_file(&self.path)
{
error!(
file = %self.path.display(),
error = %io,
"this open created the file and could not take it away again, so a store \
that was never opened is left on disk"
);
return false;
}
true
}
pub fn clean_backup(&self) {
if self.backup_path.exists() {
self.remove_backup("after the open went through");
}
}
fn remove_backup(&self, when: &'static str) {
if let Err(io) = std::fs::remove_file(&self.backup_path) {
warn!(
file = %self.path.display(),
backup = %self.backup_path.display(),
error = %io,
"the copy beside the store could not be removed {when}, and the next open \
reads one as an unfinished previous run"
);
}
}
}
#[derive(Clone)]
pub struct StoreFiles<D: TextDocument> {
pub data: StoreFile<D>,
pub meta: StoreFile<D>,
}
fn held_key() -> StorePath {
meta_at(&meta_key("held", &StorePath::root()))
}
fn migrating_key() -> StorePath {
meta_at(&meta_key("migrating", &StorePath::root()))
}
#[derive(serde::Serialize, serde::Deserialize)]
struct Migrating {
meta: String,
data: String,
was: String,
}
pub(super) fn has_no_keys<D: TextDocument>(doc: &D) -> bool {
doc.scan_keys(&StorePath::root())
.map(|children| children.is_empty())
.unwrap_or(false)
}
impl<D: TextDocument> StoreFiles<D> {
pub fn load(&self) -> StorageResult<(D, D)> {
let meta = self
.meta
.read_or_recover()
.and_then(|found| self.settle_a_write_under_way(found));
let held = matches!(
meta.as_ref()
.ok()
.and_then(|held: &D| held.get(&held_key()))
.map(D::deserialize_node::<bool>),
Some(Ok(true))
);
let data = self
.data
.read_or_recover_unless(|doc: &D| match held && has_no_keys(doc) {
true => Some("the last save left keys here and the file now holds none"),
false => None,
})
.attach("role: the store's data")?;
let meta = meta.attach("role: the store's schema bookkeeping")?;
Ok((data, meta))
}
pub(crate) fn persist_while(&self, left: Option<Standing>) -> StorageResult<Wrote> {
self.remember_what_the_data_holds()?;
self.meta
.persist()
.attach("role: the store's schema bookkeeping")?;
self.data
.persist_while(left)
.attach("role: the store's data")
}
fn remember_what_the_data_holds(&self) -> StorageResult<()> {
let holds = !has_no_keys(&*self.data.doc.read());
let key = held_key();
{
let mut guard = self.meta.doc.write();
let said = matches!(
guard.get(&key).map(D::deserialize_node::<bool>),
Some(Ok(true))
);
if said == holds {
return Ok(());
}
let node = D::serialize_node(&holds, &Noticed::unlimited())
.change_context(StorageError::Meta)
.attach_key(&key)?;
guard
.set(&key, node)
.change_context(StorageError::Meta)
.attach_key(&key)?;
}
Ok(())
}
fn settle_a_write_under_way(&self, found: D) -> StorageResult<D> {
let record: Migrating = match found.get(&migrating_key()) {
None => return Ok(found),
Some(node) => D::deserialize_node(node)
.change_context(StorageError::Meta)
.attach_key(&migrating_key())?,
};
let now = standing_of(&self.data.path)
.map(|(_, _, held)| format!("{held:032x}"))
.unwrap_or_default();
let settled = if now == record.data {
warn!(
path = %self.data.path.display(),
"an open stopped after writing a migrated store's data and before recording \
it, and the record it left finishes the metadata"
);
D::parse(&record.meta).attach_store_file(&self.meta.path)?
} else if now == record.was {
warn!(
path = %self.data.path.display(),
"an open stopped before writing a migrated store's data, so the migration runs \
again"
);
let mut dropped = found.clone();
dropped
.delete(&migrating_key())
.change_context(StorageError::Meta)
.attach_key(&migrating_key())?;
dropped
} else {
return Err(error_stack::Report::new(StorageError::Open)
.attach(StoreFileFact(self.data.path.clone()))
.attach(
"an open stopped while writing a migration here, and the data file is \
neither what it found nor what it was writing, so which data the metadata \
describes cannot be told",
));
};
self.meta
.persist_document(&settled)
.attach("role: the store's schema bookkeeping")?;
self.data.clean_backup();
self.meta.clean_backup();
Ok(settled)
}
fn write_both(&self, found_data: &D, found_meta: &D) -> StorageResult<()> {
let finished = self
.meta
.doc
.read()
.serialize()
.attach_store_file(&self.meta.path)?;
let writing = self
.data
.doc
.read()
.serialize()
.attach_store_file(&self.data.path)?;
let record = Migrating {
meta: finished,
data: format!("{:032x}", hash_of(&writing)),
was: standing_of(&self.data.path)
.map(|(_, _, held)| format!("{held:032x}"))
.unwrap_or_default(),
};
let node = D::serialize_node(&record, &Noticed::unlimited())
.change_context(StorageError::Meta)
.attach_key(&migrating_key())?;
let mut under_way = found_meta.clone();
under_way
.set(&migrating_key(), node)
.change_context(StorageError::Meta)
.attach_key(&migrating_key())?;
self.meta
.create_backup()
.attach("role: the store's schema bookkeeping")?;
self.data.create_backup().attach("role: the store's data")?;
let begun = self
.meta
.persist_document(&under_way)
.attach("role: the store's schema bookkeeping")
.inspect(|()| stop_the_open_after("meta"))
.and_then(|()| {
self.data
.persist()
.attach("role: the store's data")
.inspect(|()| stop_the_open_after("data"))
});
if let Err(why) = begun {
let meta_back = self.meta.restore_from_backup(found_meta);
let data_back = self.data.restore_from_backup(found_data);
return Err(why.attach(match meta_back && data_back {
true => "the files this open wrote were put back from their copies",
false => {
"the files this open wrote could not all be put back from their copies, \
and hold what it wrote - the log names which"
}
}));
}
self.data.clean_backup();
self.meta.clean_backup();
self.meta
.persist()
.attach("role: the store's schema bookkeeping")
.attach(
"the data landed and the metadata still records the write under way, which \
the next open finishes",
)?;
sweep_temporaries(&self.data.path);
sweep_temporaries(&self.meta.path);
Ok(())
}
pub fn write_what_the_open_changed(&self, found_data: &D, found_meta: &D) -> StorageResult<()> {
self.remember_what_the_data_holds()?;
let meta = changed(&self.meta, found_meta)?;
let data = changed(&self.data, found_data)?;
if meta && data {
return self.write_both(found_data, found_meta);
}
if meta {
self.meta
.create_backup()
.attach("role: the store's schema bookkeeping")?;
}
if data {
self.data.create_backup().attach("role: the store's data")?;
}
let written = match meta {
true => self
.meta
.persist()
.attach("role: the store's schema bookkeeping")
.inspect(|()| stop_the_open_after("meta")),
false => Ok(()),
}
.and_then(|()| match data {
true => self
.data
.persist()
.attach("role: the store's data")
.inspect(|()| stop_the_open_after("data")),
false => Ok(()),
});
if let Err(why) = written {
let meta_back = !meta || self.meta.restore_from_backup(found_meta);
let data_back = !data || self.data.restore_from_backup(found_data);
return Err(why.attach(match meta_back && data_back {
true => "the files this open wrote were put back from their copies",
false => {
"the files this open wrote could not all be put back from their copies, \
and hold what it wrote - the log names which"
}
}));
}
if meta {
self.meta.clean_backup();
}
if data {
self.data.clean_backup();
}
sweep_temporaries(&self.data.path);
sweep_temporaries(&self.meta.path);
Ok(())
}
}
#[cfg(feature = "test-utils")]
fn stop_the_open_after(written: &str) {
if std::env::var("AMETHYSTATE_STOP_THE_OPEN_AFTER").as_deref() == Ok(written) {
std::process::abort();
}
}
#[cfg(not(feature = "test-utils"))]
fn stop_the_open_after(_written: &str) {}
fn changed<D: TextDocument>(file: &StoreFile<D>, found: &D) -> StorageResult<bool> {
if !file.path.exists() {
return Ok(true);
}
let now = file.doc.read().serialize().attach_store_file(&file.path)?;
let then = found.serialize().attach_store_file(&file.path)?;
Ok(now != then)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Wrote {
Replaced(Option<Standing>),
FileMoved,
}
pub(super) fn hash_of(content: &str) -> u128 {
xxhash_rust::xxh3::xxh3_128(content.as_bytes())
}
pub(crate) type Standing = (u64, std::time::SystemTime, u128);
pub(super) fn standing_of(file: &Path) -> Option<Standing> {
let held = std::fs::metadata(file).ok()?;
let modified = held.modified().ok()?;
let content = std::fs::read(file).ok()?;
Some((held.len(), modified, xxhash_rust::xxh3::xxh3_128(&content)))
}
fn what_we_left(file: &Path, content: &str) -> Option<Standing> {
standing_of(file).filter(|(_, _, held)| *held == hash_of(content))
}
fn persist_atomic(
path: &Path,
content: &str,
policy: FileWritePolicy,
still: &dyn Fn() -> bool,
) -> io::Result<bool> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let mut written = None;
for attempt in 0..policy.write.attempts.max(1) {
match write_temp(path, content) {
Ok(tmp) => {
written = Some(tmp);
break;
}
Err(e) if attempt + 1 >= policy.write.attempts => return Err(e),
Err(_) => std::thread::sleep(policy.write.pause),
}
}
let mut tmp = written.expect("the loop above returns rather than falling through");
if !still() {
return Ok(false);
}
for attempt in 0..policy.replace.attempts.max(1) {
match tmp.persist(path) {
Ok(_) => return Ok(true),
Err(e) if attempt + 1 >= policy.replace.attempts => return Err(e.error),
Err(e) => {
tmp = e.file;
std::thread::sleep(policy.replace.pause);
}
}
}
unreachable!("the loop above returns on its last attempt")
}
fn write_temp(target: &Path, content: &str) -> io::Result<NamedTempFile> {
let dir = target.parent().unwrap_or(Path::new("."));
let mut tmp = tempfile::Builder::new()
.prefix(&format!("{}{}", temporaries_of(target), a_mark_of_ours()))
.suffix(TEMPORARY)
.rand_bytes(0)
.tempfile_in(dir)?;
tmp.write_all(content.as_bytes())?;
tmp.as_file().sync_all()?;
Ok(tmp)
}
const TEMPORARY: &str = ".tmp";
const WRITTEN_BY_US: u64 = 0x616d_6574_6879_7374;
fn a_mark_of_ours() -> String {
let nonce = uuid::Uuid::new_v4().as_u128() as u32;
format!("{nonce:08x}{:08x}", vouched_for(nonce))
}
fn vouched_for(nonce: u32) -> u32 {
let mut bytes = [0u8; 12];
bytes[..4].copy_from_slice(&nonce.to_le_bytes());
bytes[4..].copy_from_slice(&WRITTEN_BY_US.to_le_bytes());
xxhash_rust::xxh3::xxh3_64(&bytes) as u32
}
fn is_a_mark_of_ours(mark: &str) -> bool {
if mark.len() != 16 {
return false;
}
let (nonce, claimed) = mark.split_at(8);
let Ok(nonce) = u32::from_str_radix(nonce, 16) else {
return false;
};
let Ok(claimed) = u32::from_str_radix(claimed, 16) else {
return false;
};
vouched_for(nonce) == claimed
}
fn temporaries_of(target: &Path) -> String {
let named = target
.file_name()
.map(|name| name.to_string_lossy().into_owned())
.unwrap_or_default();
format!("{named}.")
}
fn sweep_temporaries(target: &Path) {
let Some(dir) = target.parent() else {
return;
};
let Ok(entries) = std::fs::read_dir(dir) else {
return;
};
let prefix = temporaries_of(target);
for entry in entries.flatten() {
let name = entry.file_name();
let name = name.to_string_lossy();
let Some(rest) = name.strip_prefix(&prefix) else {
continue;
};
let Some(mark) = rest.strip_suffix(TEMPORARY) else {
continue;
};
if is_a_mark_of_ours(mark) && abandoned(&entry) {
let _ = std::fs::remove_file(entry.path());
}
}
}
const ABANDONED_AFTER: std::time::Duration = std::time::Duration::from_secs(600);
fn abandoned(entry: &std::fs::DirEntry) -> bool {
entry
.metadata()
.and_then(|held| held.modified())
.ok()
.and_then(|modified| modified.elapsed().ok())
.is_some_and(|age| age >= ABANDONED_AFTER)
}
#[cfg(all(test, feature = "json"))]
mod tests {
use super::super::json::json_doc::JsonDocument;
use super::*;
use amethystate_core::test_utils::TempPath;
fn holding(at: &Path, what: &str) -> StoreFile<JsonDocument> {
let doc = JsonDocument::parse(what).unwrap();
StoreFile::new(at.to_path_buf(), doc, FileWritePolicy::default())
}
#[test]
fn a_mark_this_library_wrote_reads_back_as_its_own() {
for _ in 0..64 {
assert!(is_a_mark_of_ours(&a_mark_of_ours()));
}
}
#[test]
fn a_name_of_the_same_shape_is_not_a_mark() {
assert!(
!is_a_mark_of_ours("0123456789abcdef"),
"sixteen hex characters are a shape, and a sweep that takes a shape takes \
whatever else happens to have it"
);
assert!(!is_a_mark_of_ours("draft"), "a name somebody typed");
assert!(!is_a_mark_of_ours(""), "nothing at all");
assert!(
!is_a_mark_of_ours("00000000ffffffff"),
"a nonce with the wrong answer beside it"
);
}
#[test]
fn a_mark_with_one_character_changed_stops_being_one() {
let mark = a_mark_of_ours();
let mut bent: Vec<char> = mark.chars().collect();
bent[3] = if bent[3] == 'a' { 'b' } else { 'a' };
assert!(!is_a_mark_of_ours(&bent.into_iter().collect::<String>()));
}
#[test]
fn a_sweep_takes_this_library_s_leftovers_and_leaves_everything_else() {
let at = TempPath::new("sweeping");
let dir = at
.path()
.parent()
.expect("a temporary has a directory")
.to_path_buf();
let data = dir.join("settings.json");
let ours = dir.join(format!("settings.json.{}{TEMPORARY}", a_mark_of_ours()));
let shaped_the_same = dir.join(format!("settings.json.0123456789abcdef{TEMPORARY}"));
let somebody_elses = dir.join(format!("settings.json.draft{TEMPORARY}"));
let another_store = dir.join(format!("other.json.{}{TEMPORARY}", a_mark_of_ours()));
for file in [
&data,
&ours,
&shaped_the_same,
&somebody_elses,
&another_store,
] {
std::fs::write(file, "x").unwrap();
dated_back(file, std::time::Duration::from_secs(3600));
}
sweep_temporaries(&data);
assert!(
!ours.exists(),
"the leftover a killed write of ours left is still there"
);
assert!(data.exists(), "the sweep took the store's own file");
assert!(
shaped_the_same.exists(),
"a name that merely takes the shape of ours was taken away"
);
assert!(somebody_elses.exists());
assert!(
another_store.exists(),
"the sweep reached past its own file into another store's leftovers"
);
for file in [&data, &shaped_the_same, &somebody_elses, &another_store] {
let _ = std::fs::remove_file(file);
}
}
fn dated_back(file: &Path, by: std::time::Duration) {
std::fs::File::options()
.write(true)
.open(file)
.and_then(|held| held.set_modified(std::time::SystemTime::now() - by))
.unwrap();
}
#[test]
fn a_temporary_too_young_to_be_abandoned_is_left_to_its_writer() {
let at = TempPath::new("sweeping_young");
let dir = at
.path()
.parent()
.expect("a temporary has a directory")
.to_path_buf();
let data = dir.join("settings.json");
let in_flight = dir.join(format!("settings.json.{}{TEMPORARY}", a_mark_of_ours()));
std::fs::write(&in_flight, "x").unwrap();
sweep_temporaries(&data);
assert!(
in_flight.exists(),
"another store's replacement still under way was taken from under it"
);
let _ = std::fs::remove_file(&in_flight);
}
#[test]
fn a_file_that_moved_since_it_was_read_is_left_alone() {
let at = TempPath::new("persist_while");
let file = holding(at.path(), r#"{"ours":1}"#);
file.persist().unwrap();
let read_at = standing_of(at.path());
std::fs::write(at.path(), r#"{"theirs":2}"#).unwrap();
assert_eq!(
file.persist_while(read_at).unwrap(),
Wrote::FileMoved,
"the file was written by somebody else after it was read"
);
assert_eq!(
std::fs::read_to_string(at.path()).unwrap(),
r#"{"theirs":2}"#,
"the replacement went ahead over a file this save had never seen"
);
assert!(
matches!(
file.persist_while(standing_of(at.path())).unwrap(),
Wrote::Replaced(_)
),
"the same save against the file as it stands now must land"
);
assert!(
std::fs::read_to_string(at.path()).unwrap().contains("ours"),
"the replacement that was allowed did not happen"
);
}
#[test]
fn a_file_rewritten_to_the_same_length_and_time_is_left_alone_too() {
let at = TempPath::new("persist_same_standing");
let file = holding(at.path(), r#"{"ours":1}"#);
file.persist().unwrap();
let read_at = standing_of(at.path());
let modified = std::fs::metadata(at.path()).unwrap().modified().unwrap();
let theirs = std::fs::read_to_string(at.path())
.unwrap()
.replace("ours", "them");
std::fs::write(at.path(), &theirs).unwrap();
std::fs::File::options()
.write(true)
.open(at.path())
.and_then(|held| held.set_modified(modified))
.unwrap();
assert_eq!(
file.persist_while(read_at).unwrap(),
Wrote::FileMoved,
"somebody else wrote bytes of the same length and the clock says nothing moved"
);
assert_eq!(std::fs::read_to_string(at.path()).unwrap(), theirs);
}
#[test]
fn a_restore_that_puts_the_copy_back_says_it_did() {
let at = TempPath::new("restore_lands");
let file = holding(at.path(), r#"{"ours":1}"#);
std::fs::write(at.path(), r#"{"half":2}"#).unwrap();
std::fs::write(&file.backup_path, r#"{"ours":1}"#).unwrap();
assert!(file.restore_from_backup(&JsonDocument::parse(r#"{"ours":1}"#).unwrap()));
assert_eq!(std::fs::read_to_string(at.path()).unwrap(), r#"{"ours":1}"#);
}
#[test]
fn a_restore_that_cannot_put_the_copy_back_says_it_did_not() {
let at = TempPath::new("restore_blocked");
let file = holding(at.path(), r#"{"ours":1}"#);
std::fs::write(&file.backup_path, r#"{"ours":1}"#).unwrap();
std::fs::create_dir(at.path()).unwrap();
assert!(!file.restore_from_backup(&JsonDocument::parse(r#"{"ours":1}"#).unwrap()));
}
}