#![cfg(feature = "persy")]
use beam::actor::Actor;
use beam::types::Value;
use std::env;
use std::time::Duration;
fn unique_persy_path(test_name: &str) -> String {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
env::temp_dir()
.join(format!(
"beam-persy-{}-{}-{}.persy",
test_name,
std::process::id(),
nanos
))
.to_str()
.expect("temp path must be utf-8")
.to_string()
}
fn node_with_persy(path: &str) -> beam::Node {
use beam::Config;
use beam::adapters::PersyStorage;
let storage = PersyStorage::new_with_path(path);
beam::Node::new_with_config(
Config::default(),
vec![Box::new(storage) as Box<dyn Actor>],
vec![],
)
}
#[tokio::test]
async fn e2e_persy_put_get_roundtrip() {
let path = unique_persy_path("roundtrip");
let mut node = node_with_persy(&path);
node.get("k")
.put("v".into())
.await
.expect("persy put should ack after fsync");
let got = node.get("k").once(Some(Duration::from_secs(2))).await;
assert_eq!(got, Some(Value::Text("v".to_string())));
let _ = std::fs::remove_file(&path);
}
#[tokio::test]
async fn e2e_persy_sequential_puts_serialize_correctly() {
let path = unique_persy_path("sequential");
let mut node = node_with_persy(&path);
for i in 0..25 {
let key = format!("conc_{}", i);
let val = format!("val_{}", i);
node.get(&key)
.put(val.clone().into())
.await
.expect("put should ack");
let got = node.get(&key).once(Some(Duration::from_secs(2))).await;
assert_eq!(
got,
Some(Value::Text(val.clone())),
"put {} → get observed {:?}, expected {:?}",
i,
got,
val
);
}
let _ = std::fs::remove_file(&path);
}
#[tokio::test]
async fn e2e_persy_nested_children_roundtrip() {
let path = unique_persy_path("nested");
let mut node = node_with_persy(&path);
for (child, val) in [
("child_a", "a_value"),
("child_b", "b_value"),
("child_c", "c_value"),
] {
node.get("parent")
.get(child)
.put(val.into())
.await
.expect("child put");
}
for (child, expected) in [
("child_a", "a_value"),
("child_b", "b_value"),
("child_c", "c_value"),
] {
let got = node
.get("parent")
.get(child)
.once(Some(Duration::from_secs(2)))
.await;
assert_eq!(got, Some(Value::Text(expected.to_string())));
}
let _ = std::fs::remove_file(&path);
}
#[tokio::test]
async fn e2e_persy_lww_prefers_newer_value() {
let path = unique_persy_path("lww");
let mut node = node_with_persy(&path);
node.get("lww_key").put("older".into()).await.unwrap();
tokio::time::sleep(Duration::from_millis(10)).await;
node.get("lww_key").put("newer".into()).await.unwrap();
let got = node.get("lww_key").once(Some(Duration::from_secs(2))).await;
assert_eq!(got, Some(Value::Text("newer".to_string())));
let _ = std::fs::remove_file(&path);
}
#[tokio::test]
async fn e2e_persy_isolated_persistence_two_files() {
let path_a = unique_persy_path("iso_a");
let path_b = unique_persy_path("iso_b");
{
let mut node = node_with_persy(&path_a);
node.get("alpha").put("1".into()).await.unwrap();
let got = node.get("alpha").once(Some(Duration::from_secs(2))).await;
assert_eq!(got, Some(Value::Text("1".to_string())));
}
{
let mut node = node_with_persy(&path_b);
node.get("beta").put("2".into()).await.unwrap();
let got = node.get("beta").once(Some(Duration::from_secs(2))).await;
assert_eq!(got, Some(Value::Text("2".to_string())));
}
let _ = std::fs::remove_file(&path_a);
let _ = std::fs::remove_file(&path_b);
}