Menu
akurai-tasks
publicLatest change 91b48cca5fcd809e56adaa690dbdb30b657281cc - fix: recover stale database lock on restart by Ólafur Búi Ólafsson
use super::*;
use sha2::{Digest, Sha256};
use std::fs::{self, File, OpenOptions};
use std::io::{Read, Write};
pub struct ExclusiveLock {
path: PathBuf,
_file: File,
}
impl Drop for ExclusiveLock {
fn drop(&mut self) {
let _ = fs::remove_file(&self.path);
}
}
impl Store {
pub fn path(&self) -> &Path {
self.path.as_path()
}
pub fn acquire_exclusive_lock(database: &Path) -> Result<ExclusiveLock> {
let path = database.with_extension("lock");
for attempt in 0..2 {
match OpenOptions::new().write(true).create_new(true).open(&path) {
Ok(mut file) => {
writeln!(file, "{}", std::process::id()).map_err(storage)?;
file.sync_all().map_err(storage)?;
return Ok(ExclusiveLock { path, _file: file });
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists && attempt == 0 => {
let pid = fs::read_to_string(&path)
.ok()
.and_then(|value| value.trim().parse::<u32>().ok());
if pid.is_some_and(|value| Path::new(&format!("/proc/{value}")).exists()) {
return Err(Error::Conflict(format!(
"database is already owned by process {}",
pid.unwrap_or_default()
)));
}
fs::remove_file(&path).map_err(storage)?;
}
Err(error) => {
return Err(Error::Conflict(format!(
"database is already owned or lock cannot be created: {error}"
)));
}
}
}
Err(Error::Conflict("database lock recovery failed".into()))
}
pub fn doctor(&self) -> Result<Value> {
let mut db = self
.db
.lock()
.map_err(|_| Error::Storage("lock poisoned".into()))?;
let entries = db.range(b"", &[u8::MAX]).map_err(storage)?;
let mut records = 0i64;
let mut items = 0i64;
let mut events = 0i64;
let mut last_event = 0i64;
for (key, bytes) in entries {
let value = decode(&bytes)?;
records += 1;
if key.starts_with(b"item/") {
items += 1;
let project = field(&value, "project")
.ok_or_else(|| Error::Storage("item missing project".into()))?;
if get_value(&mut db, &format!("project/{project}"))?.is_none() {
return Err(Error::Storage(format!(
"item references missing project {project}"
)));
}
}
if key.starts_with(b"event/") {
events += 1;
let sequence = value
.get("sequence")
.and_then(Value::as_i64)
.ok_or_else(|| Error::Storage("event missing sequence".into()))?;
if sequence <= last_event {
return Err(Error::Storage(
"event sequence is not strictly increasing".into(),
));
}
last_event = sequence;
}
}
Ok(object(vec![
("ok", Value::Bool(true)),
("records", Value::Int(records)),
("items", Value::Int(items)),
("events", Value::Int(events)),
("lastEvent", Value::Int(last_event)),
]))
}
pub fn backup(&self, output: &Path) -> Result<Value> {
if let Some(parent) = output.parent() {
fs::create_dir_all(parent).map_err(storage)?;
}
let mut db = self
.db
.lock()
.map_err(|_| Error::Storage("lock poisoned".into()))?;
db.commit().map_err(storage)?;
fs::copy(self.path.as_path(), output).map_err(storage)?;
File::open(output)
.and_then(|file| file.sync_all())
.map_err(storage)?;
let (bytes, sha256) = file_digest(output)?;
let manifest = object(vec![
("version", Value::Int(1)),
("database", string(&self.path.display().to_string())),
("backup", string(&output.display().to_string())),
("bytes", Value::Int(bytes as i64)),
("sha256", string(&sha256)),
("createdAt", Value::Int(now())),
]);
let manifest_path = output.with_extension(format!(
"{}manifest.json",
output
.extension()
.and_then(|v| v.to_str())
.map(|v| format!("{v}."))
.unwrap_or_default()
));
fs::write(&manifest_path, manifest.to_json()).map_err(storage)?;
File::open(&manifest_path)
.and_then(|file| file.sync_all())
.map_err(storage)?;
Ok(manifest)
}
pub fn restore(database: &Path, backup: &Path) -> Result<Value> {
let _lock = Self::acquire_exclusive_lock(database)?;
let parent = database.parent().unwrap_or_else(|| Path::new("."));
fs::create_dir_all(parent).map_err(storage)?;
let staged = database.with_extension(format!("restore-{}", std::process::id()));
fs::copy(backup, &staged).map_err(storage)?;
File::open(&staged)
.and_then(|file| file.sync_all())
.map_err(storage)?;
{
let mut db = BTree::open(&staged).map_err(storage)?;
for (_, bytes) in db.range(b"", &[u8::MAX]).map_err(storage)? {
decode(&bytes)?;
}
}
let replaced = database.with_extension(format!("replaced-{}", now()));
if database.exists() {
fs::rename(database, &replaced).map_err(storage)?;
}
if let Err(error) = fs::rename(&staged, database) {
if replaced.exists() {
let _ = fs::rename(&replaced, database);
}
return Err(storage(error));
}
File::open(parent)
.and_then(|file| file.sync_all())
.map_err(storage)?;
let (bytes, sha256) = file_digest(database)?;
Ok(object(vec![
("ok", Value::Bool(true)),
("database", string(&database.display().to_string())),
(
"replaced",
if replaced.exists() {
string(&replaced.display().to_string())
} else {
Value::Null
},
),
("bytes", Value::Int(bytes as i64)),
("sha256", string(&sha256)),
]))
}
}
fn file_digest(path: &Path) -> Result<(u64, String)> {
let mut file = File::open(path).map_err(storage)?;
let mut hash = Sha256::new();
let mut buffer = [0u8; 64 * 1024];
let mut bytes = 0u64;
loop {
let read = file.read(&mut buffer).map_err(storage)?;
if read == 0 {
break;
}
bytes += read as u64;
hash.update(&buffer[..read]);
}
let mut digest = String::with_capacity(64);
for byte in hash.finalize() {
digest.push_str(&format!("{byte:02x}"));
}
Ok((bytes, digest))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn backup_doctor_and_restore_preserve_records() {
let base = std::env::temp_dir().join(format!(
"akurai-tasks-maint-{}-{}",
std::process::id(),
now()
));
fs::create_dir_all(&base).unwrap();
let database = base.join("tasks.db");
let backup = base.join("backup.db");
let restored = base.join("restored.db");
let store = Store::open(&database).unwrap();
store
.create_project("OPS", "Operations", vec![], "admin", "project")
.unwrap();
assert_eq!(store.doctor().unwrap().get("ok"), Some(&Value::Bool(true)));
let manifest = store.backup(&backup).unwrap();
assert!(
manifest
.get("sha256")
.and_then(Value::as_str)
.unwrap()
.len()
== 64
);
drop(store);
Store::restore(&restored, &backup).unwrap();
let restored_store = Store::open(&restored).unwrap();
assert_eq!(
field(&restored_store.get_project("OPS").unwrap(), "name"),
Some("Operations")
);
drop(restored_store);
let _ = fs::remove_dir_all(base);
}
}