use std::error::Error;
use std::sync::Arc;
use std::time::Duration;
use beamr::module::ModuleRegistry;
use beamr::scheduler::{Scheduler, SchedulerConfig};
use super::journal;
use crate::shard::router::{ShardMode, ShardRouter, shard_dir};
use crate::store::{DiskStore, NodeStore, StoreError};
use crate::tree::{LeafNode, Node};
type TestResult = Result<(), Box<dyn Error>>;
const TIMEOUT: Duration = Duration::from_secs(5);
fn scheduler() -> Result<Arc<Scheduler>, Box<dyn Error>> {
Scheduler::new(SchedulerConfig::default(), Arc::new(ModuleRegistry::new()))
.map(Arc::new)
.map_err(|message| -> Box<dyn Error> { message.into() })
}
fn leaf(key: &[u8], value: &[u8]) -> Result<Node, Box<dyn Error>> {
Ok(Node::Leaf(LeafNode::new(vec![(
key.to_vec(),
value.to_vec(),
)])?))
}
#[test]
fn wall_l1_failpoint_refuses_materialisation() -> TestResult {
let _guard = journal::test_guard();
let dir = tempfile::tempdir()?;
let router = ShardRouter::new(
scheduler()?,
dir.path(),
4,
crate::tree::TreePolicy::V1_DEFAULT,
ShardMode::Create,
crate::db::root_advance::RootAdvanceSeam::new(),
)
.ok_or("non-zero shard_count")?;
journal::arm_failpoint();
let outcome = router.handle_for_shard(0);
assert!(
outcome.is_err(),
"an injected data_dir fence failure must refuse materialisation"
);
assert!(
router.materialised_shard_ids().is_empty(),
"a refused materialisation must leave no usable shard (no durable claim)"
);
router.shutdown_all(TIMEOUT);
Ok(())
}
#[test]
fn wall_l1_lossy_cut_loses_shard_entry() -> TestResult {
let _guard = journal::test_guard();
let dir = tempfile::tempdir()?;
let shard0 = shard_dir(dir.path(), 0);
let router = ShardRouter::new(
scheduler()?,
dir.path(),
4,
crate::tree::TreePolicy::V1_DEFAULT,
ShardMode::Create,
crate::db::root_advance::RootAdvanceSeam::new(),
)
.ok_or("non-zero shard_count")?;
router.handle_for_shard(0).map_err(|error| error.message)?;
assert_eq!(
journal::pending_len(),
0,
"a real data_dir fence must leave nothing pending"
);
router.shutdown_all(TIMEOUT);
journal::cut()?;
assert!(shard0.is_dir(), "a fenced shard entry survives the CUT");
let dir = tempfile::tempdir()?;
let shard0 = shard_dir(dir.path(), 0);
let router = ShardRouter::new(
scheduler()?,
dir.path(),
4,
crate::tree::TreePolicy::V1_DEFAULT,
ShardMode::Create,
crate::db::root_advance::RootAdvanceSeam::new(),
)
.ok_or("non-zero shard_count")?;
journal::set_lossy(true);
router.handle_for_shard(0).map_err(|error| error.message)?;
journal::set_lossy(false);
assert!(
journal::pending_len() >= 1,
"an unfenced shard entry must remain pending in the journal"
);
router.shutdown_all(TIMEOUT);
journal::cut()?;
assert!(
!shard0.exists(),
"the CUT must rewind the unfenced shard-0 entry (silent empty shard)"
);
Ok(())
}
#[test]
fn wall_l1_open_of_never_materialised_shard_passes_the_fence() -> TestResult {
let _guard = journal::test_guard();
let dir = tempfile::tempdir()?;
let router = ShardRouter::new(
scheduler()?,
dir.path(),
4,
crate::tree::TreePolicy::V1_DEFAULT,
ShardMode::Open,
crate::db::root_advance::RootAdvanceSeam::new(),
)
.ok_or("non-zero shard_count")?;
assert!(!shard_dir(dir.path(), 0).exists());
router
.handle_for_shard(0)
.map_err(|error| -> Box<dyn Error> { error.message.into() })?;
assert!(
shard_dir(dir.path(), 0).is_dir(),
"opening a never-materialised shard must create and fence its entry"
);
assert_eq!(router.materialised_shard_ids(), vec![0]);
router.shutdown_all(TIMEOUT);
Ok(())
}
#[test]
fn wall_l2_failpoint_refuses_store_construction() -> TestResult {
let _guard = journal::test_guard();
let shard = tempfile::tempdir()?;
let store_dir = shard.path().join("store");
journal::arm_failpoint();
let outcome = DiskStore::with_cache_capacity(&store_dir, 16);
assert!(
matches!(outcome, Err(StoreError::Io(_))),
"an injected store-root fence failure must refuse with StoreError::Io, got {outcome:?}"
);
Ok(())
}
#[test]
fn wall_l2_lossy_cut_loses_store_root_entry() -> TestResult {
let _guard = journal::test_guard();
let shard = tempfile::tempdir()?;
let store_dir = shard.path().join("store");
let _store = DiskStore::with_cache_capacity(&store_dir, 16)?;
assert_eq!(
journal::pending_len(),
0,
"a real store-root fence leaves nothing pending"
);
journal::cut()?;
assert!(store_dir.is_dir(), "a fenced store root survives the CUT");
let shard = tempfile::tempdir()?;
let store_dir = shard.path().join("store");
journal::set_lossy(true);
let _store = DiskStore::with_cache_capacity(&store_dir, 16)?;
journal::set_lossy(false);
assert!(
journal::pending_len() >= 1,
"an unfenced store root must remain pending"
);
journal::cut()?;
assert!(
!store_dir.exists(),
"the CUT must rewind the unfenced store-root entry"
);
Ok(())
}
#[test]
fn wall_l3_failpoint_refuses_and_retains() -> TestResult {
let _guard = journal::test_guard();
let shard = tempfile::tempdir()?;
let store_dir = shard.path().join("store");
let mut store = DiskStore::with_cache_capacity(&store_dir, 64)?;
journal::reset();
store.put(&leaf(b"k", b"v")?)?;
assert_eq!(
store.pending_entry_fences().len(),
1,
"a new prefix subdir must owe a store-root entry fence"
);
journal::arm_failpoint();
let outcome = store.sync_dirty_dirs();
assert!(
matches!(outcome, Err(StoreError::Io(_))),
"an injected barrier fence failure must refuse, got {outcome:?}"
);
assert_eq!(
store.pending_entry_fences().len(),
1,
"a failed barrier must RETAIN the un-fenced store-root obligation (R4c)"
);
store.sync_dirty_dirs()?;
assert!(
store.pending_entry_fences().is_empty(),
"the retry must clear it"
);
Ok(())
}
#[test]
fn wall_l3_lossy_cut_loses_prefix_subdir_entry() -> TestResult {
let _guard = journal::test_guard();
let shard = tempfile::tempdir()?;
let store_dir = shard.path().join("store");
let mut store = DiskStore::with_cache_capacity(&store_dir, 64)?;
journal::reset();
let hash = store.put(&leaf(b"live", b"value")?)?;
let node_path = store_dir_node_path(&store_dir, &hash);
store.sync_dirty_dirs()?;
assert_eq!(
journal::pending_len(),
0,
"a real barrier leaves nothing pending"
);
journal::cut()?;
assert!(
node_path.exists(),
"a fenced prefix subdir survives the CUT"
);
let shard = tempfile::tempdir()?;
let store_dir = shard.path().join("store");
let mut store = DiskStore::with_cache_capacity(&store_dir, 64)?;
journal::reset();
let hash = store.put(&leaf(b"doomed", b"value")?)?;
let node_path = store_dir_node_path(&store_dir, &hash);
let prefix_dir = node_path
.parent()
.ok_or("node has a prefix dir")?
.to_path_buf();
journal::set_lossy(true);
store.sync_dirty_dirs()?;
journal::set_lossy(false);
assert!(
journal::pending_len() >= 1,
"an unfenced prefix subdir must remain pending"
);
journal::cut()?;
assert!(
!prefix_dir.exists(),
"the CUT must rewind the unfenced prefix subdir and the node beneath it"
);
Ok(())
}
fn store_dir_node_path(
store_dir: &std::path::Path,
hash: &crate::tree::Hash,
) -> std::path::PathBuf {
let hex = hash.to_string();
let (prefix, file_name) = hex.split_at(2);
store_dir.join(prefix).join(file_name)
}