From 0a2b03898a335ade99e1172ebb46b682105bdd69 Mon Sep 17 00:00:00 2001 From: mack42 Date: Thu, 16 Jul 2026 17:48:26 -0400 Subject: [PATCH] Fix pub/sub message delivery to idle subscribers (v0.11.9) A subscriber did not receive published messages until it happened to send another byte to the server. The connection loop latches `in_pubsub` at the top of each iteration; when a SUBSCRIBE was processed inside the normal-mode branch, control fell through to a blocking socket read instead of the select! that also polls pubsub_rx, so a delivered message sat unread in the channel until the next loop iteration (triggered only by more client input). Re-enter the loop after processing commands if the connection has entered pub/sub mode, so the next iteration waits on the pub/sub channel. Also commit the integration test suite (previously untracked); the two pubsub tests failed against this bug and now pass. --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/server/connection.rs | 9 +++ tests/common/mod.rs | 97 +++++++++++++++++++++++++++++++ tests/integration_auth.rs | 39 +++++++++++++ tests/integration_collections.rs | 71 ++++++++++++++++++++++ tests/integration_pubsub.rs | 73 +++++++++++++++++++++++ tests/integration_strings.rs | 62 ++++++++++++++++++++ tests/integration_transactions.rs | 72 +++++++++++++++++++++++ 9 files changed, 425 insertions(+), 2 deletions(-) create mode 100644 tests/common/mod.rs create mode 100644 tests/integration_auth.rs create mode 100644 tests/integration_collections.rs create mode 100644 tests/integration_pubsub.rs create mode 100644 tests/integration_strings.rs create mode 100644 tests/integration_transactions.rs diff --git a/Cargo.lock b/Cargo.lock index f51b719..83eeefc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -791,7 +791,7 @@ dependencies = [ [[package]] name = "rcache" -version = "0.11.8" +version = "0.11.9" dependencies = [ "bytes", "crc16", diff --git a/Cargo.toml b/Cargo.toml index ad2c6b1..a4f5502 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rcache" -version = "0.11.8" +version = "0.11.9" edition = "2024" [dependencies] diff --git a/src/server/connection.rs b/src/server/connection.rs index 2683de8..ce0790a 100644 --- a/src/server/connection.rs +++ b/src/server/connection.rs @@ -191,6 +191,15 @@ impl Connection { self.stream.write_all(&data).await?; } + // A command just processed (e.g. SUBSCRIBE) may have put us into + // pub/sub mode. Re-enter the loop so the next iteration waits on + // the pub/sub channel via select!, rather than blocking on the + // socket read below — otherwise a published message sits unread + // in pubsub_rx until the client happens to send more data. + if !self.subscribed_channels.is_empty() || !self.subscribed_patterns.is_empty() { + continue; + } + // Read more data from the socket, bounded by an idle timeout. let n = match tokio::time::timeout( READ_IDLE_TIMEOUT, diff --git a/tests/common/mod.rs b/tests/common/mod.rs new file mode 100644 index 0000000..a8e18e7 --- /dev/null +++ b/tests/common/mod.rs @@ -0,0 +1,97 @@ +//! Shared test harness: spawn rcache on an ephemeral port and clean up. +//! +//! Each test gets its own server process and port so test isolation is +//! perfect and tests can run in parallel. + +use std::io::{BufRead, BufReader}; +use std::net::TcpListener; +use std::process::{Child, Command, Stdio}; +use std::sync::mpsc; +use std::time::{Duration, Instant}; + +pub struct Server { + child: Child, + pub port: u16, +} + +impl Server { + pub fn spawn() -> Self { + Self::spawn_with_args(&[]) + } + + pub fn spawn_with_args(extra: &[&str]) -> Self { + let port = pick_port(); + let bin = env!("CARGO_BIN_EXE_rcache"); + + let mut cmd = Command::new(bin); + cmd.arg("--port") + .arg(port.to_string()) + .arg("--bind") + .arg("127.0.0.1"); + for a in extra { + cmd.arg(a); + } + let mut child = cmd + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .expect("failed to spawn rcache binary"); + + // The child's stderr must be drained continuously, otherwise a full + // pipe will block the server. Spawn a background reader that signals + // when "Listening on" appears and then keeps the pipe drained for the + // lifetime of the child. + let stderr = child.stderr.take().expect("piped stderr"); + let (tx, rx) = mpsc::channel::<()>(); + std::thread::spawn(move || { + let reader = BufReader::new(stderr); + let mut signalled = false; + for line in reader.lines().map_while(Result::ok) { + if !signalled && line.contains("Listening on") { + let _ = tx.send(()); + signalled = true; + } + // keep consuming after signalling so the child does not block + } + }); + + // Wait for the readiness signal, with a connect-poll fallback. + let ready = rx.recv_timeout(Duration::from_secs(10)).is_ok(); + if !ready { + let until = Instant::now() + Duration::from_secs(5); + let mut connected = false; + while Instant::now() < until { + if std::net::TcpStream::connect(("127.0.0.1", port)).is_ok() { + connected = true; + break; + } + std::thread::sleep(Duration::from_millis(20)); + } + assert!(connected, "rcache did not start listening on port {port}"); + } + + Server { child, port } + } + + pub fn url(&self) -> String { + format!("redis://127.0.0.1:{}/", self.port) + } + + pub fn client(&self) -> redis::Client { + redis::Client::open(self.url()).expect("client open") + } +} + +impl Drop for Server { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +fn pick_port() -> u16 { + let listener = TcpListener::bind("127.0.0.1:0").expect("ephemeral bind"); + let port = listener.local_addr().expect("local_addr").port(); + drop(listener); + port +} diff --git a/tests/integration_auth.rs b/tests/integration_auth.rs new file mode 100644 index 0000000..762a887 --- /dev/null +++ b/tests/integration_auth.rs @@ -0,0 +1,39 @@ +mod common; + +use common::Server; +use redis::Commands; + +#[test] +fn no_auth_required_when_no_requirepass() { + let s = Server::spawn(); + let mut c = s.client().get_connection().unwrap(); + let _: () = c.set("k", "v").unwrap(); +} + +#[test] +fn auth_required_when_requirepass_set() { + let s = Server::spawn_with_args(&["--requirepass", "topsecret"]); + let mut c = s.client().get_connection().unwrap(); + + let res: Result = c.get("k"); + assert!(res.is_err(), "expected NOAUTH error"); + + let _: () = redis::cmd("AUTH") + .arg("topsecret") + .query(&mut c) + .expect("auth ok"); + + let v: Option = c.get("k").unwrap(); + assert!(v.is_none()); +} + +#[test] +fn auth_wrong_password_rejected() { + let s = Server::spawn_with_args(&["--requirepass", "topsecret"]); + let mut c = s.client().get_connection().unwrap(); + + let res: Result = redis::cmd("AUTH") + .arg("wrong") + .query(&mut c); + assert!(res.is_err(), "AUTH with wrong password must return WRONGPASS"); +} diff --git a/tests/integration_collections.rs b/tests/integration_collections.rs new file mode 100644 index 0000000..0b05ac6 --- /dev/null +++ b/tests/integration_collections.rs @@ -0,0 +1,71 @@ +mod common; + +use common::Server; +use redis::Commands; + +#[test] +fn list_lpush_rpush_lrange() { + let s = Server::spawn(); + let mut c = s.client().get_connection().unwrap(); + + let _: () = c.rpush("l", &["a", "b", "c"]).unwrap(); + let _: () = c.lpush("l", "z").unwrap(); + let items: Vec = c.lrange("l", 0, -1).unwrap(); + assert_eq!(items, vec!["z", "a", "b", "c"]); +} + +#[test] +fn hash_hset_hget_hgetall() { + let s = Server::spawn(); + let mut c = s.client().get_connection().unwrap(); + + let _: () = c.hset_multiple("h", &[("a", "1"), ("b", "2")]).unwrap(); + let v: String = c.hget("h", "a").unwrap(); + assert_eq!(v, "1"); + + let all: std::collections::HashMap = c.hgetall("h").unwrap(); + assert_eq!(all.get("a").map(String::as_str), Some("1")); + assert_eq!(all.get("b").map(String::as_str), Some("2")); +} + +#[test] +fn set_sadd_smembers_srem() { + let s = Server::spawn(); + let mut c = s.client().get_connection().unwrap(); + + let _: () = c.sadd("s", &["a", "b", "c"]).unwrap(); + let mut members: Vec = c.smembers("s").unwrap(); + members.sort(); + assert_eq!(members, vec!["a", "b", "c"]); + + let removed: i64 = c.srem("s", "b").unwrap(); + assert_eq!(removed, 1); +} + +#[test] +fn zset_zadd_zrange_zscore() { + let s = Server::spawn(); + let mut c = s.client().get_connection().unwrap(); + + let _: () = c.zadd_multiple("z", &[(1.0, "a"), (2.0, "b"), (3.0, "c")]).unwrap(); + let items: Vec = c.zrange("z", 0, -1).unwrap(); + assert_eq!(items, vec!["a", "b", "c"]); + + let score: f64 = c.zscore("z", "b").unwrap(); + assert_eq!(score, 2.0); +} + +#[test] +fn expire_and_ttl() { + let s = Server::spawn(); + let mut c = s.client().get_connection().unwrap(); + + let _: () = c.set("k", "v").unwrap(); + let _: bool = c.expire("k", 100).unwrap(); + let ttl: i64 = c.ttl("k").unwrap(); + assert!(ttl > 0 && ttl <= 100, "ttl in (0, 100]; got {ttl}"); + + let _: bool = c.persist("k").unwrap(); + let ttl: i64 = c.ttl("k").unwrap(); + assert_eq!(ttl, -1, "persist should clear ttl"); +} diff --git a/tests/integration_pubsub.rs b/tests/integration_pubsub.rs new file mode 100644 index 0000000..adbdc70 --- /dev/null +++ b/tests/integration_pubsub.rs @@ -0,0 +1,73 @@ +mod common; + +use common::Server; +use std::sync::mpsc; +use std::time::Duration; + +#[test] +fn subscribe_then_receive_published_message() { + let s = Server::spawn(); + let port = s.port; + + // Subscriber runs on its own thread because get_message() blocks. We hand + // the received payload back over a channel so we can apply a hard timeout. + let (tx, rx) = mpsc::channel::(); + std::thread::spawn(move || { + let client = redis::Client::open(format!("redis://127.0.0.1:{port}/")).unwrap(); + let mut conn = client.get_connection().unwrap(); + let mut pubsub = conn.as_pubsub(); + pubsub.subscribe("ch1").unwrap(); + let msg = pubsub.get_message().unwrap(); + let payload: String = msg.get_payload().unwrap(); + let _ = tx.send(payload); + }); + + // Give the subscriber time to register before publishing. + std::thread::sleep(Duration::from_millis(200)); + + let mut pub_conn = s.client().get_connection().unwrap(); + let n: i64 = redis::cmd("PUBLISH") + .arg("ch1") + .arg("hello") + .query(&mut pub_conn) + .unwrap(); + assert_eq!(n, 1, "exactly one subscriber should have received"); + + let received = rx + .recv_timeout(Duration::from_secs(5)) + .expect("subscriber did not deliver message within 5s"); + assert_eq!(received, "hello"); +} + +#[test] +fn pattern_subscribe_matches_glob() { + let s = Server::spawn(); + let port = s.port; + + let (tx, rx) = mpsc::channel::<(String, String)>(); + std::thread::spawn(move || { + let client = redis::Client::open(format!("redis://127.0.0.1:{port}/")).unwrap(); + let mut conn = client.get_connection().unwrap(); + let mut pubsub = conn.as_pubsub(); + pubsub.psubscribe("news.*").unwrap(); + let msg = pubsub.get_message().unwrap(); + let channel: String = msg.get_channel_name().to_string(); + let payload: String = msg.get_payload().unwrap(); + let _ = tx.send((channel, payload)); + }); + + std::thread::sleep(Duration::from_millis(200)); + + let mut pub_conn = s.client().get_connection().unwrap(); + let _: i64 = redis::cmd("PUBLISH") + .arg("news.sports") + .arg("goal") + .query(&mut pub_conn) + .unwrap(); + + let (ch, payload) = rx + .recv_timeout(Duration::from_secs(5)) + .expect("subscriber did not deliver pattern message within 5s"); + assert_eq!(ch, "news.sports"); + assert_eq!(payload, "goal"); +} diff --git a/tests/integration_strings.rs b/tests/integration_strings.rs new file mode 100644 index 0000000..dd990e5 --- /dev/null +++ b/tests/integration_strings.rs @@ -0,0 +1,62 @@ +mod common; + +use common::Server; +use redis::Commands; + +#[test] +fn set_get_del_basics() { + let s = Server::spawn(); + let mut c = s.client().get_connection().expect("connect"); + + let _: () = c.set("k", "v").unwrap(); + let v: String = c.get("k").unwrap(); + assert_eq!(v, "v"); + + let n: i64 = c.del("k").unwrap(); + assert_eq!(n, 1); + + let v: Option = c.get("k").unwrap(); + assert!(v.is_none()); +} + +#[test] +fn incr_decr_chain() { + let s = Server::spawn(); + let mut c = s.client().get_connection().unwrap(); + + let n: i64 = c.incr("counter", 1).unwrap(); + assert_eq!(n, 1); + let n: i64 = c.incr("counter", 5).unwrap(); + assert_eq!(n, 6); + let n: i64 = c.decr("counter", 2).unwrap(); + assert_eq!(n, 4); +} + +#[test] +fn mset_mget() { + let s = Server::spawn(); + let mut c = s.client().get_connection().unwrap(); + + let _: () = c.mset(&[("a", "1"), ("b", "2"), ("c", "3")]).unwrap(); + let vs: Vec> = c.mget(&["a", "b", "missing", "c"]).unwrap(); + assert_eq!(vs[0].as_deref(), Some("1")); + assert_eq!(vs[1].as_deref(), Some("2")); + assert!(vs[2].is_none()); + assert_eq!(vs[3].as_deref(), Some("3")); +} + +#[test] +fn append_strlen_getrange() { + let s = Server::spawn(); + let mut c = s.client().get_connection().unwrap(); + + let _: () = c.set("s", "hello").unwrap(); + let n: i64 = c.append("s", ", world").unwrap(); + assert_eq!(n, 12); + let v: String = c.get("s").unwrap(); + assert_eq!(v, "hello, world"); + let len: i64 = c.strlen("s").unwrap(); + assert_eq!(len, 12); + let sub: String = c.getrange("s", 7, 11).unwrap(); + assert_eq!(sub, "world"); +} diff --git a/tests/integration_transactions.rs b/tests/integration_transactions.rs new file mode 100644 index 0000000..89e7272 --- /dev/null +++ b/tests/integration_transactions.rs @@ -0,0 +1,72 @@ +mod common; + +use common::Server; +use redis::Commands; + +#[test] +fn multi_exec_atomic_commit() { + let s = Server::spawn(); + let mut c = s.client().get_connection().unwrap(); + + let (a, b): (i64, i64) = redis::pipe() + .atomic() + .incr("counter", 3) + .incr("counter", 4) + .query(&mut c) + .unwrap(); + assert_eq!((a, b), (3, 7)); + let final_val: i64 = c.get("counter").unwrap(); + assert_eq!(final_val, 7); +} + +#[test] +fn watch_aborts_on_mutation() { + let s = Server::spawn(); + let mut tx = s.client().get_connection().unwrap(); + let mut other = s.client().get_connection().unwrap(); + + let _: () = tx.set("balance", 100i64).unwrap(); + + // WATCH balance, then read it + let _: () = redis::cmd("WATCH").arg("balance").query(&mut tx).unwrap(); + let v: i64 = tx.get("balance").unwrap(); + assert_eq!(v, 100); + + // Another connection mutates the watched key + let _: () = other.set("balance", 200i64).unwrap(); + + // MULTI/EXEC must report abort (nil reply) + let res: redis::Value = redis::pipe() + .atomic() + .cmd("SET") + .arg("balance") + .arg(v + 10) + .query(&mut tx) + .unwrap(); + assert!(matches!(res, redis::Value::Nil), "expected nil from aborted EXEC, got {:?}", res); + + let final_val: i64 = other.get("balance").unwrap(); + assert_eq!(final_val, 200, "watched mutation should be preserved"); +} + +#[test] +fn watch_does_not_abort_without_mutation() { + let s = Server::spawn(); + let mut tx = s.client().get_connection().unwrap(); + + let _: () = tx.set("x", 1i64).unwrap(); + let _: () = redis::cmd("WATCH").arg("x").query(&mut tx).unwrap(); + let v: i64 = tx.get("x").unwrap(); + + let res: redis::Value = redis::pipe() + .atomic() + .cmd("SET") + .arg("x") + .arg(v + 1) + .query(&mut tx) + .unwrap(); + assert!(!matches!(res, redis::Value::Nil), "EXEC should succeed"); + + let final_val: i64 = tx.get("x").unwrap(); + assert_eq!(final_val, 2); +}