use std::fs;
use std::os::unix::fs::MetadataExt;
use std::path::Path;
use std::sync::Arc;
use haematite::{Database, EventStore};
use tempfile::TempDir;
use super::{DurabilityError, DurableStore, HaematiteStore, ScanCursor, ScanPage, StoredEntry};
type FixtureResult<T> = Result<T, Box<dyn std::error::Error>>;
#[derive(Debug, Default)]
struct CopyCounts {
files: u64,
bytes: u64,
directories: u64,
}
fn copy_directory(source: &Path, destination: &Path) -> FixtureResult<CopyCounts> {
let mut counts = CopyCounts::default();
for entry in fs::read_dir(source)? {
let entry = entry?;
let source = entry.path();
let destination = destination.join(entry.file_name());
let kind = entry.file_type()?;
if kind.is_dir() {
fs::create_dir(&destination)?;
fs::set_permissions(&destination, entry.metadata()?.permissions())?;
let nested = copy_directory(&source, &destination)?;
counts.files = counts
.files
.checked_add(nested.files)
.ok_or("fixture file count overflow")?;
counts.bytes = counts
.bytes
.checked_add(nested.bytes)
.ok_or("fixture byte count overflow")?;
counts.directories = counts
.directories
.checked_add(nested.directories)
.and_then(|count| count.checked_add(1))
.ok_or("fixture directory count overflow")?;
} else if kind.is_file() {
let bytes = fs::copy(&source, &destination)?;
counts.files = counts
.files
.checked_add(1)
.ok_or("fixture file count overflow")?;
counts.bytes = counts
.bytes
.checked_add(bytes)
.ok_or("fixture byte count overflow")?;
let before = fs::metadata(&source)?;
let after = fs::metadata(&destination)?;
if before.dev() == after.dev() && before.ino() == after.ino() {
return Err("durability fixture copy shares its source inode".into());
}
if fs::read(&source)? != fs::read(&destination)? {
return Err("durability fixture copy differs from its source bytes".into());
}
} else {
return Err("durability fixture contains a link or non-regular file".into());
}
}
Ok(counts)
}
pub(super) fn copied(name: &str) -> FixtureResult<CopiedStore> {
let root = std::env::var_os("LIMINAL_DURABILITY_FIXTURES")
.ok_or("durability fixture setup did not provide LIMINAL_DURABILITY_FIXTURES")?;
let source = Path::new(&root).join(name);
if !source.is_dir() {
return Err(format!("durability fixture image {name} is missing").into());
}
let directory = DirectoryGuard {
directory: Some(
tempfile::Builder::new()
.prefix("durability-copy-")
.tempdir_in(&root)?,
),
};
let counts = copy_directory(&source, directory.path()?)?;
println!("fixture copy {name}: {counts:?}");
let database = Database::open(
directory.path()?,
haematite::RuntimeOptions {
long_wait_interval: std::time::Duration::from_secs(5),
reclaim_start: haematite::ReclaimStart::AtOpen,
},
)?;
Ok(CopiedStore {
store: HaematiteStore::new(Arc::new(EventStore::new(database))),
directory,
})
}
pub(super) fn settled_claims() -> FixtureResult<usize> {
let root = std::env::var_os("LIMINAL_DURABILITY_FIXTURES")
.ok_or("durability fixture setup did not provide LIMINAL_DURABILITY_FIXTURES")?;
let manifest: serde_json::Value =
serde_json::from_slice(&fs::read(Path::new(&root).join("manifest.json"))?)?;
let count = manifest["settled_claims"]
.as_u64()
.ok_or("durability fixture manifest has no settled claim count")?;
Ok(usize::try_from(count)?)
}
#[derive(Debug)]
struct DirectoryGuard {
directory: Option<TempDir>,
}
impl DirectoryGuard {
fn path(&self) -> Result<&Path, DurabilityError> {
self.directory
.as_ref()
.map(TempDir::path)
.ok_or(DurabilityError::EphemeralStoreDetached)
}
}
impl Drop for DirectoryGuard {
fn drop(&mut self) {
if let Some(directory) = self.directory.take() {
if std::thread::panicking() {
let retained = directory.keep();
eprintln!("durability fixture unwind retained {}", retained.display());
} else {
let path = directory.path().to_path_buf();
if let Err(error) = directory.close() {
eprintln!(
"durability fixture removal failed at {}: {error}",
path.display()
);
}
}
}
}
}
#[derive(Debug)]
pub(super) struct CopiedStore {
store: HaematiteStore,
directory: DirectoryGuard,
}
impl CopiedStore {
fn store(&self) -> Result<&HaematiteStore, DurabilityError> {
self.directory.path()?;
Ok(&self.store)
}
}
#[async_trait::async_trait]
impl DurableStore for CopiedStore {
async fn apply_routed_batch(
&self,
route_key: &str,
items: Vec<crate::durability::RoutedBatchItem>,
) -> Result<(), crate::durability::DurabilityError> {
self.store()?.apply_routed_batch(route_key, items).await
}
async fn get_routed(
&self,
route_key: &str,
key: &[u8],
) -> Result<Option<Vec<u8>>, crate::durability::DurabilityError> {
self.store()?.get_routed(route_key, key).await
}
async fn append(
&self,
stream_key: &str,
payload: Vec<u8>,
expected_seq: u64,
) -> Result<u64, DurabilityError> {
self.store()?
.append(stream_key, payload, expected_seq)
.await
}
async fn append_all(
&self,
stream_key: &str,
payloads: Vec<Vec<u8>>,
expected_seq: u64,
) -> Result<u64, DurabilityError> {
self.store()?
.append_all(stream_key, payloads, expected_seq)
.await
}
async fn read_from(
&self,
stream_key: &str,
offset: u64,
limit: usize,
) -> Result<Vec<StoredEntry>, DurabilityError> {
self.store()?.read_from(stream_key, offset, limit).await
}
async fn cas(&self, key: &str, old_value: u64, new_value: u64) -> Result<(), DurabilityError> {
self.store()?.cas(key, old_value, new_value).await
}
async fn read_stream_next_seq(&self, stream_key: &str) -> Result<Option<u64>, DurabilityError> {
self.store()?.read_stream_next_seq(stream_key).await
}
async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
self.store()?.read_value(key).await
}
async fn scan_page(
&self,
prefix: &str,
cursor: Option<ScanCursor>,
limit: usize,
) -> Result<ScanPage, DurabilityError> {
self.store()?.scan_page(prefix, cursor, limit).await
}
async fn flush(&self) -> Result<(), DurabilityError> {
self.store()?.flush().await
}
async fn truncate_before(
&self,
stream_key: &str,
before_seq: u64,
) -> Result<u64, DurabilityError> {
self.store()?.truncate_before(stream_key, before_seq).await
}
}