crates/graphar-falkordb/tests/robot_response_timeout.rs
//! **STINKY-19 R22 — a wedged FalkorDB link must FAIL, not sit there.**
//!
//! `knut-popviz`'s `falkor_parity` was recorded on 2026-08-01 running **24
//! minutes without completing**: the client process at 0:00 CPU, all 66 threads
//! parked, and the server holding a partially received 163 KB `graph.QUERY`
//! (`qbuf=163227 qbuf-free=603 idle=182`) whose remaining bytes never arrived.
//! It was killed by hand. Nothing failed, nothing logged, nothing timed out —
//! and because the whole tier is `#[ignore]`d, nobody noticed for three weeks.
//!
//! The reason it could hang forever is one line of defaulting:
//! `Client::get_multiplexed_async_connection()` leaves `response_timeout` at
//! `None`, and `redis`'s `Pipeline::send_recv` then does a bare
//! `receiver.await` on the reply oneshot — no deadline, no cancellation, no
//! error path. A link that stops moving is indistinguishable from a slow query,
//! forever.
//!
//! [`FalkorDbLoader::with_response_timeout`] is the opt-in ceiling. This file is
//! its guard, and it is built to be RED in both directions:
//!
//! * **[`a_silent_server_is_refused_instead_of_awaited_forever`]** — a real TCP
//! server that completes redis's handshake and then never answers the
//! `GRAPH.QUERY` at all. With a 400 ms ceiling the call must come back as an
//! error well inside the test's own 8 s outer bound. If
//! `with_response_timeout` ever stops reaching redis's deadline, this reports
//! a NAMED failure instead of hanging the suite — a guard for a hang must
//! never fail by hanging.
//! * **[`without_a_ceiling_the_same_socket_hangs_forever`]** — the control that
//! makes the first test mean something. The *identical* socket, with no
//! ceiling set, must still be pending after 2 s. Without this leg a
//! `with_response_timeout` that did nothing at all (or a `run_query` that
//! errored instantly for some unrelated reason — a refused connect, a parse
//! failure) would satisfy the first test perfectly. The pair can only pass if
//! the ceiling is the thing that changed the outcome.
//!
//! Deliberately NOT container-gated and NOT `#[ignore]`d: it binds a loopback
//! port and needs no FalkorDB, so it runs in every plain `cargo test`. The
//! defect it guards was invisible precisely because its neighbours were opt-in.
use std::time::{Duration, Instant};
use graphar_falkordb::FalkorDbLoader;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
/// How long the "no ceiling" control waits before calling the call hung. Must be
/// comfortably longer than the ceiling the other test sets, or the control would
/// pass for the wrong reason.
const CONTROL_WAIT: Duration = Duration::from_secs(2);
/// The ceiling under test — short enough to keep the suite quick, long enough
/// that it cannot be mistaken for an instant connection failure.
const CEILING: Duration = Duration::from_millis(400);
/// A socket that speaks just enough RESP to be connected to, and then goes deaf.
///
/// `redis-rs` sends `CLIENT SETINFO lib-name` / `lib-ver` while establishing a
/// multiplexed connection (that is where the `lib-name=redis-rs lib-ver=0.27.6`
/// in the hung server's client list came from), and it awaits those replies — so
/// a server that is silent from the very first byte would wedge `connect`
/// itself and this file would be testing the wrong function. Instead: answer
/// `+OK` to every command that arrives before the first `GRAPH.QUERY`, then
/// answer nothing ever again while holding the connection open.
///
/// Holding it open is the point. A server that *closed* the socket would give
/// `run_query` a clean I/O error, and the test would pass without the ceiling
/// ever being consulted — that is the shape of an accidental green, so it is
/// avoided deliberately.
async fn silent_after_handshake() -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind loopback");
let addr = listener.local_addr().expect("local addr");
tokio::spawn(async move {
// One client is all this test makes; a loop would outlive the test.
let Ok((mut sock, _)) = listener.accept().await else {
return;
};
let mut buf = vec![0u8; 64 * 1024];
let mut gone_deaf = false;
loop {
let Ok(n) = sock.read(&mut buf).await else { return };
if n == 0 {
return; // client hung up — nothing left to be silent at
}
if gone_deaf {
continue; // drain and ignore: never reply again
}
let chunk = &buf[..n];
if String::from_utf8_lossy(chunk).to_ascii_uppercase().contains("GRAPH.QUERY") {
gone_deaf = true;
continue;
}
// Every RESP command is an array, so the number of `*` frame headers
// in the chunk is the number of commands redis-rs just wrote — it
// may batch the two SETINFOs into one write, and replying once to a
// two-command write would hang the handshake instead of the query.
let commands = chunk.iter().filter(|b| **b == b'*').count().max(1);
for _ in 0..commands {
if sock.write_all(b"+OK\r\n").await.is_err() {
return;
}
}
}
});
format!("redis://{addr}")
}
/// **The fix: a link that stops moving comes back as an error.**
#[tokio::test]
async fn a_silent_server_is_refused_instead_of_awaited_forever() {
let url = silent_after_handshake().await;
let mut loader = FalkorDbLoader::connect(&url, "r22")
.await
.expect("the fake server completes the handshake, so connect must succeed")
.with_response_timeout(CEILING);
let started = Instant::now();
// The outer bound is this test's own safety net: a guard against a hang that
// fails BY hanging teaches nobody anything and wedges the suite the same way
// falkor_parity wedged a box for 24 minutes.
let outcome = tokio::time::timeout(
Duration::from_secs(8),
loader.run_query("MATCH (n) RETURN count(n)"),
)
.await
.unwrap_or_else(|_| {
panic!(
"STINKY-19 R22: run_query never returned in 8 s against a server that accepts \
and never answers, with a {CEILING:?} response ceiling set. \
`with_response_timeout` is not reaching redis's `send_recv` deadline — which \
is exactly the wedge that ran falkor_parity for 24 minutes with nothing to \
show for it."
)
});
let err = outcome.expect_err(
"a server that never sends a single byte of reply cannot produce a successful \
GRAPH.QUERY — an Ok here means the reply was fabricated, not received",
);
let elapsed = started.elapsed();
assert!(
elapsed < CONTROL_WAIT,
"the ceiling is {CEILING:?} but the call took {elapsed:?}, which is past the \
{CONTROL_WAIT:?} the no-ceiling control waits. If those two overlap, this test and \
its control can both pass without the ceiling doing anything."
);
let text = err.to_string().to_ascii_lowercase();
assert!(
text.contains("timed out") || text.contains("timeout"),
"the refusal must SAY it was a timeout, so an operator reading a CI log learns the \
link wedged rather than that the query was bad. Got: {err}"
);
}
/// **The control — without the ceiling the same socket hangs, exactly as R22 recorded.**
///
/// This is the leg that makes the test above non-vacuous. It asserts the DEFECT,
/// on purpose: same fake server, same query, no ceiling, and after
/// [`CONTROL_WAIT`] the future must still be pending. If redis ever grows a
/// default deadline, or the fake server starts answering, or `connect` begins
/// failing for an unrelated reason, this goes red and tells us the sibling test
/// above has stopped proving anything.
#[tokio::test]
async fn without_a_ceiling_the_same_socket_hangs_forever() {
let url = silent_after_handshake().await;
let mut loader = FalkorDbLoader::connect(&url, "r22")
.await
.expect("the fake server completes the handshake, so connect must succeed");
let outcome = tokio::time::timeout(
CONTROL_WAIT,
loader.run_query("MATCH (n) RETURN count(n)"),
)
.await;
assert!(
outcome.is_err(),
"with no response ceiling, `get_multiplexed_async_connection`'s `response_timeout` \
is None and `Pipeline::send_recv` awaits its reply oneshot with no deadline — so \
this call must STILL be pending after {CONTROL_WAIT:?}. It returned instead, which \
means the sibling test's error is not being produced by the ceiling and that test \
is now proving nothing. Outcome: {:?}",
outcome.map(|r| r.map(|_| "Ok(reply)").unwrap_or("Err(..)")),
);
}