crates/graphar-falkordb/src/falkordb_container.rs
//! Shared FalkorDB **autostart** helper — the single implementation behind both
//! the `--ignored` integration tests and `knut-popviz`'s "just-work" startup.
//!
//! Per constellation **LAW 7**, containers start from Rust via `testcontainers`
//! driving the Podman socket — never a shell script. This module:
//!
//! 1. [`probe_reachable`] — a cheap, non-hanging TCP probe of a `redis://` URL.
//! 2. [`falkordb_image`] — resolve the image from `KNUT_FALKORDB_IMAGE`
//! (default `localhost/knut-falkordb-valkey9:latest` — valkey-9, never stock redis).
//! 3. [`start_falkordb`] — bring up a FalkorDB container, map `6379`, wait until
//! the module answers, and hand back a [`FalkorContainer`] guard that stops
//! it on drop plus the mapped `redis://` URL.
//! 4. [`ensure_falkordb`] — the front door: probe the URL; if reachable, use it
//! as-is; otherwise start a container and return its mapped URL + guard.
//!
//! ## Prerequisites for the live path (`ensure`/`start`)
//!
//! - Podman socket: `systemctl --user start podman.socket`
//! - `DOCKER_HOST=unix:///run/user/$UID/podman/podman.sock` (auto-detected when
//! unset — see [`ensure_docker_host`]).
//!
//! `testcontainers`/Podman are only pulled when the `autostart` feature is on,
//! so the lean / wasm build of `graphar-falkordb` stays free of them.
use std::time::Duration;
use testcontainers::{
ContainerAsync, GenericImage, ImageExt, core::ContainerPort, runners::AsyncRunner,
};
use crate::FalkorDbLoader;
/// How long the reachability probe waits before declaring a port unreachable.
/// Short by design: an unreachable / refused port returns immediately, and a
/// firewalled (silently dropped) one must not hang the UI launch.
const PROBE_TIMEOUT: Duration = Duration::from_millis(500);
/// Outcome of probing a `redis://` URL.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Reach {
/// A TCP handshake to host:port completed — something is listening.
Reachable,
/// Connection refused, timed out, or the URL could not be parsed.
Unreachable,
}
impl Reach {
pub fn is_reachable(self) -> bool {
matches!(self, Reach::Reachable)
}
}
/// Parse host + port out of a `redis://host:port[/db]` URL, defaulting the port
/// to Redis's 6379 when omitted. Returns `None` if no host can be extracted.
fn host_port(url: &str) -> Option<(String, u16)> {
// Strip scheme (`redis://`, `rediss://`, or none) and any trailing path / query.
let rest = url.split("://").last().unwrap_or(url);
let authority = rest.split(['/', '?']).next().unwrap_or(rest);
// Drop any `user:pass@` credentials.
let hostport = authority.rsplit('@').next().unwrap_or(authority);
if hostport.is_empty() {
return None;
}
match hostport.rsplit_once(':') {
Some((host, port)) => {
let host = if host.is_empty() { "127.0.0.1" } else { host };
let port = port.parse().unwrap_or(6379);
Some((host.to_string(), port))
}
None => Some((hostport.to_string(), 6379)),
}
}
/// Cheap, **non-hanging** reachability probe: try a single TCP connect to the
/// URL's host:port with a short timeout. A listening port → [`Reach::Reachable`];
/// a closed/refused/filtered port or an unparseable URL → [`Reach::Unreachable`].
///
/// This is the gate `ensure_falkordb` uses to decide whether to autostart, and
/// it never blocks longer than [`PROBE_TIMEOUT`].
pub fn probe_reachable(url: &str) -> Reach {
let Some((host, port)) = host_port(url) else {
return Reach::Unreachable;
};
// Resolve and try each candidate address; the first successful connect wins.
let addrs = match std::net::ToSocketAddrs::to_socket_addrs(&(host.as_str(), port)) {
Ok(addrs) => addrs,
Err(_) => return Reach::Unreachable,
};
for addr in addrs {
if std::net::TcpStream::connect_timeout(&addr, PROBE_TIMEOUT).is_ok() {
return Reach::Reachable;
}
}
Reach::Unreachable
}
/// Point `DOCKER_HOST` at the rootless Podman socket when it is unset — the same
/// auto-detection the integration tests use, lifted here so callers share it.
pub 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}")) };
}
}
}
/// Resolve the FalkorDB image `(name, tag)` from `KNUT_FALKORDB_IMAGE`.
///
/// Defaults to our **FalkorDB-on-Valkey-9** build `localhost/knut-falkordb-valkey9:latest`
/// — the FalkorDB GraphBLAS module compiled against **Valkey 9** (built from
/// `containers/falkordb-valkey9/Containerfile`), **never** the stock
/// `docker.io/falkordb/falkordb` image (which serves the module on **Redis**). The owner's
/// law: the graph module must run on Valkey 9+, never redis. Any `KNUT_FALKORDB_IMAGE`
/// override MUST likewise be a valkey-9 FalkorDB image.
pub fn falkordb_image() -> (String, String) {
let image = std::env::var("KNUT_FALKORDB_IMAGE")
.unwrap_or_else(|_| "localhost/knut-falkordb-valkey9:latest".into());
match image.rsplit_once(':') {
// Guard against a `:` inside a host:port registry prefix with no tag.
Some((name, tag)) if !tag.contains('/') => (name.to_string(), tag.to_string()),
_ => (image, "latest".into()),
}
}
/// A running FalkorDB container plus its mapped `redis://` URL. Dropping it stops
/// the container (testcontainers' default `RemoveOnDrop`).
pub struct FalkorContainer {
/// Kept alive so the container lives as long as this guard.
_container: ContainerAsync<GenericImage>,
/// `redis://127.0.0.1:<mapped-port>` — connect here.
pub url: String,
}
impl FalkorContainer {
/// The mapped `redis://` URL to connect to.
pub fn url(&self) -> &str {
&self.url
}
}
/// Start a FalkorDB container (image from [`falkordb_image`]), map `6379`, and
/// wait until the FalkorDB module answers — then return the guard + mapped URL.
///
/// The TCP port opening does not mean the module is loaded, so after the port is
/// mapped we retry [`FalkorDbLoader::connect`] (a `redis://` handshake) until it
/// succeeds or the attempt budget is exhausted.
pub async fn start_falkordb() -> crate::Result<FalkorContainer> {
ensure_docker_host();
let (name, tag) = falkordb_image();
let container = GenericImage::new(name, tag)
.with_exposed_port(ContainerPort::Tcp(6379))
.with_startup_timeout(Duration::from_secs(120))
.start()
.await
.map_err(testcontainers_err)?;
let port = container
.get_host_port_ipv4(6379)
.await
.map_err(testcontainers_err)?;
let url = format!("redis://127.0.0.1:{port}");
// The module needs a moment after the TCP port opens — retry the handshake.
let mut last_err = None;
let mut ready = false;
for _ in 0..30 {
match FalkorDbLoader::connect(&url, "knut_autostart_probe").await {
Ok(_) => {
ready = true;
break;
}
Err(e) => {
last_err = Some(e);
tokio::time::sleep(Duration::from_secs(1)).await;
}
}
}
if !ready && let Some(e) = last_err {
return Err(e);
}
Ok(FalkorContainer {
_container: container,
url,
})
}
/// The front door: ensure a FalkorDB is reachable at `url`.
///
/// - If [`probe_reachable`] says the URL is up, return `(url, None)` — the
/// caller's own DB is used untouched.
/// - Otherwise [`start_falkordb`] a container and return its mapped URL plus the
/// guard; keep the guard alive for as long as you need the DB (drop = stop).
pub async fn ensure_falkordb(url: &str) -> crate::Result<(String, Option<FalkorContainer>)> {
if probe_reachable(url).is_reachable() {
return Ok((url.to_string(), None));
}
let container = start_falkordb().await?;
let mapped = container.url.clone();
Ok((mapped, Some(container)))
}
/// Wrap a `testcontainers` error (a boxed `dyn Error`) into a `FalkorError` so
/// callers get one error type. We reuse the `Redis` variant's textual surface by
/// constructing a redis IO error carrying the message.
fn testcontainers_err(e: testcontainers::TestcontainersError) -> crate::FalkorError {
crate::FalkorError::Redis(redis::RedisError::from(std::io::Error::other(format!(
"container start failed: {e}"
))))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn host_port_parses_variants() {
assert_eq!(
host_port("redis://127.0.0.1:6379"),
Some(("127.0.0.1".into(), 6379))
);
assert_eq!(
host_port("redis://localhost:1234/0"),
Some(("localhost".into(), 1234))
);
// Default port when omitted.
assert_eq!(host_port("redis://myhost"), Some(("myhost".into(), 6379)));
// Credentials are stripped.
assert_eq!(
host_port("redis://user:pass@db.example:6380"),
Some(("db.example".into(), 6380))
);
}
/// The probe must return *promptly* and report `Unreachable` for a port that
/// nothing is listening on — never hang. Heavy live-container paths are
/// `#[ignore]`d below; this one is cheap and always runs.
#[test]
fn probe_closed_port_is_unreachable_and_does_not_hang() {
// Bind an ephemeral port, learn its number, then drop the listener so the
// port is (almost certainly) closed — connect should be refused fast.
let port = {
let l = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
l.local_addr().unwrap().port()
};
let url = format!("redis://127.0.0.1:{port}");
let start = std::time::Instant::now();
assert_eq!(probe_reachable(&url), Reach::Unreachable);
assert!(
start.elapsed() < Duration::from_secs(2),
"probe of a closed port must not hang"
);
}
/// A port with a live listener must probe as `Reachable`.
#[test]
fn probe_listening_port_is_reachable() {
let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
let port = listener.local_addr().unwrap().port();
// Keep `listener` in scope so the port stays open for the probe.
let url = format!("redis://127.0.0.1:{port}");
assert_eq!(probe_reachable(&url), Reach::Reachable);
drop(listener);
}
/// Unparseable / empty URLs are `Unreachable`, not a panic.
#[test]
fn garbage_url_is_unreachable() {
assert_eq!(probe_reachable(""), Reach::Unreachable);
assert_eq!(probe_reachable("redis://"), Reach::Unreachable);
}
/// Image resolution honours the env var and defaults to the **valkey-9** build,
/// never stock redis. Serialized implicitly by running single-threaded is not
/// guaranteed, so this only asserts the default when the var is absent.
///
/// **RED-when-broken (the owner's law):** the FalkorDB module must run on Valkey 9+,
/// never on stock `docker.io/falkordb/falkordb` (module-on-Redis). If someone reverts
/// the default to the stock image, this turns red.
#[test]
fn default_image_is_valkey9_never_redis_when_env_absent() {
// Only assert the default shape when the operator hasn't set the var.
if std::env::var("KNUT_FALKORDB_IMAGE").is_err() {
let (name, tag) = falkordb_image();
assert_eq!(name, "localhost/knut-falkordb-valkey9");
assert_eq!(tag, "latest");
// NEVER the stock upstream image (which serves the module on Redis).
assert!(
!name.contains("falkordb/falkordb") && !name.contains("docker.io/falkordb"),
"the default FalkorDB image must be our valkey-9 build, never stock redis: {name}:{tag}",
);
}
}
/// Full autostart needs a live Podman + image — heavy, so `#[ignore]`d like
/// the existing container tests. Run with:
/// `cargo test -p graphar-falkordb --features autostart -- --ignored`.
#[tokio::test]
#[ignore = "needs podman + a FalkorDB image; run with --ignored"]
async fn ensure_starts_container_when_unreachable() {
// A definitely-closed port forces the autostart branch.
let port = {
let l = std::net::TcpListener::bind(("127.0.0.1", 0)).unwrap();
l.local_addr().unwrap().port()
};
let (url, guard) = ensure_falkordb(&format!("redis://127.0.0.1:{port}"))
.await
.expect("autostart FalkorDB");
assert!(guard.is_some(), "a container guard is returned");
assert!(url.starts_with("redis://127.0.0.1:"));
// The mapped URL must now be reachable.
assert_eq!(probe_reachable(&url), Reach::Reachable);
}
}