crates/knut-bifrost/tests/lifecycle_falkordb.rs
//! The create→detect→ingest→teardown lifecycle against a **real FalkorDB**.
//!
//! Same shape as `tests/lifecycle.rs`, but the recording sink is replaced by a
//! live `FalkorDbSink` backed by a FalkorDB container (testcontainers handles
//! setup + teardown). Each round stages a parquet drop and then asserts the
//! actual node count in FalkorDB via `GRAPH.QUERY`.
//!
//! ## Prerequisites
//! - Podman socket: `systemctl --user start podman.socket`
//! - FalkorDB image (override with `KNUT_FALKORDB_IMAGE`):
//! `podman build -t knut-falkordb-valkey9 -f containers/falkordb-valkey9/Containerfile .`
//! (or set `KNUT_FALKORDB_IMAGE=localhost/knut-falkordb-valkey9:latest`)
//!
//! ## Run
//! ```bash
//! cargo test -p knut-bifrost --test lifecycle_falkordb -- --ignored --nocapture
//! ```
use std::sync::Arc;
use std::time::Duration;
use arrow_array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use knut_bifrost::{
Listener, MappingSpec, SyncState, TableState, TransformRegistry, sink::FalkorDbSink,
source::FileSource,
};
use testcontainers::{GenericImage, ImageExt, core::ContainerPort, runners::AsyncRunner};
const MAPPING: &str = r#"
version: 1
tables:
- table: people
mode: incremental
pipeline:
- vertex:
label: Person
id_column: id
properties: [name]
"#;
fn people_batch(ids: &[i64]) -> RecordBatch {
let names: Vec<String> = ids.iter().map(|i| format!("person_{i}")).collect();
RecordBatch::try_new(
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
])),
vec![
Arc::new(Int64Array::from(ids.to_vec())),
Arc::new(StringArray::from(names)),
],
)
.unwrap()
}
fn drop_chunk(root: &std::path::Path, name: &str, ids: &[i64]) {
let dir = root.join("people");
std::fs::create_dir_all(&dir).unwrap();
graphar::io::write_chunk(
dir.join(name),
std::slice::from_ref(&people_batch(ids)),
&graphar::FileType::Parquet,
)
.unwrap();
}
fn ensure_docker_host() {
if std::env::var("DOCKER_HOST").is_err() {
let uid = std::process::Command::new("id")
.arg("-u")
.output()
.ok()
.and_then(|o| String::from_utf8(o.stdout).ok())
.and_then(|s| s.trim().parse::<u32>().ok())
.unwrap_or(1000);
let sock = format!("/run/user/{uid}/podman/podman.sock");
if std::path::Path::new(&sock).exists() {
unsafe { std::env::set_var("DOCKER_HOST", format!("unix://{sock}")) };
}
}
}
async fn count(conn: &mut redis::aio::MultiplexedConnection, graph: &str, q: &str) -> i64 {
fn walk(v: &redis::Value) -> Option<i64> {
match v {
redis::Value::Int(i) => Some(*i),
redis::Value::Array(items) | redis::Value::Set(items) => items.iter().find_map(walk),
redis::Value::BulkString(b) => String::from_utf8_lossy(b).parse().ok(),
_ => None,
}
}
let v: redis::Value = redis::cmd("GRAPH.QUERY")
.arg(graph)
.arg(q)
.query_async(conn)
.await
.unwrap();
walk(&v).unwrap_or(-1)
}
#[tokio::test]
#[ignore = "needs podman + a FalkorDB image; run with --ignored"]
async fn lifecycle_ingests_into_real_falkordb() {
ensure_docker_host();
// ── setup: FalkorDB container ───────────────────────────────────────────
let image = std::env::var("KNUT_FALKORDB_IMAGE")
.unwrap_or_else(|_| "localhost/knut-falkordb-valkey9:latest".into());
let (name, tag) = image.rsplit_once(':').unwrap_or((image.as_str(), "latest"));
let falkor = GenericImage::new(name, tag)
.with_exposed_port(ContainerPort::Tcp(6379))
.with_startup_timeout(Duration::from_secs(120))
.start()
.await
.expect("start falkordb (build the image first, or set KNUT_FALKORDB_IMAGE)");
let port = falkor.get_host_port_ipv4(6379).await.unwrap();
let url = format!("redis://127.0.0.1:{port}");
let graph = "lifecycle";
let client = redis::Client::open(url.clone()).unwrap();
let mut conn = client.get_multiplexed_async_connection().await.unwrap();
// ── a listener over a watched dir → real FalkorDB sink ──────────────────
let drops = tempfile::tempdir().unwrap();
let state = tempfile::tempdir().unwrap();
let mut listener = Listener::new(
MappingSpec::from_yaml(MAPPING).unwrap(),
TransformRegistry::new(),
FileSource::new(drops.path()),
FalkorDbSink::connect(&url, graph).await.unwrap(),
SyncState::open(state.path().join("state.json")).unwrap(),
);
// ── round 1: create a file → detect → ingest ────────────────────────────
drop_chunk(drops.path(), "chunk0.parquet", &[1, 2, 3]);
let round = listener.sync_once().await.unwrap();
assert_eq!(round.failed, vec![], "round 1 failures");
assert_eq!(round.synced.len(), 1);
assert_eq!(round.synced[0].1, 3, "3 rows shipped");
assert_eq!(
count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
3,
"3 Person nodes in FalkorDB",
);
// ── idle round: nothing new → no change ─────────────────────────────────
let idle = listener.sync_once().await.unwrap();
assert!(idle.synced.is_empty(), "no new files → idle");
assert_eq!(
count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
3
);
// ── round 2: a second drop → incremental delta ──────────────────────────
drop_chunk(drops.path(), "chunk1.parquet", &[4, 5]);
let round = listener.sync_once().await.unwrap();
assert_eq!(round.failed, vec![], "round 2 failures");
assert_eq!(round.synced[0].1, 2, "only the new file's 2 rows");
assert_eq!(
count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
5,
"5 Person nodes after the incremental drop",
);
// ── re-ingesting the same id MERGEs (idempotent), not duplicates ────────
drop_chunk(drops.path(), "chunk2.parquet", &[5, 6]);
listener.sync_once().await.unwrap();
assert_eq!(
count(&mut conn, graph, "MATCH (n:Person) RETURN count(n)").await,
6,
"id 5 re-merged (not duplicated); only 6 added",
);
// teardown: container + tempdirs drop here.
}
#[tokio::test]
#[ignore = "needs podman + a Valkey/FalkorDB image; run with --ignored"]
async fn valkey_state_ledger_is_shared_across_instances() {
ensure_docker_host();
// The FalkorDB image is a Valkey 9 server — use it as the ledger store.
let image = std::env::var("KNUT_FALKORDB_IMAGE")
.unwrap_or_else(|_| "localhost/knut-falkordb-valkey9:latest".into());
let (name, tag) = image.rsplit_once(':').unwrap_or((image.as_str(), "latest"));
let valkey = GenericImage::new(name, tag)
.with_exposed_port(ContainerPort::Tcp(6379))
.with_startup_timeout(Duration::from_secs(120))
.start()
.await
.expect("start valkey/falkordb");
let port = valkey.get_host_port_ipv4(6379).await.unwrap();
let url = format!("redis://127.0.0.1:{port}");
let key = "bifrost:test:state";
// Instance 1 commits some progress.
let mut s1 = SyncState::open_valkey(&url, key).await.unwrap();
assert_eq!(
s1.get("lake.users"),
TableState::default(),
"fresh ledger is empty"
);
s1.advance(
"lake.users",
TableState {
snapshot_id: Some(7),
done_files: vec!["chunk0.parquet".into(), "chunk1.parquet".into()],
..Default::default()
},
)
.await
.unwrap();
// A *separate* instance opening the same hash sees the committed state —
// the shared ledger survives multi-instance (unlike the file backend).
let s2 = SyncState::open_valkey(&url, key).await.unwrap();
let st = s2.get("lake.users");
assert_eq!(st.snapshot_id, Some(7));
assert_eq!(st.done_files, vec!["chunk0.parquet", "chunk1.parquet"]);
assert!(st.synced_at.is_some(), "advance stamped synced_at");
}