Fix pub/sub message delivery to idle subscribers (v0.11.9)

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rcache"
-version = "0.11.8"
+version = "0.11.9"
 edition = "2024"
 
 [dependencies]

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,

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
+}

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<String, _> = 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<String> = 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<String, _> = redis::cmd("AUTH")
+        .arg("wrong")
+        .query(&mut c);
+    assert!(res.is_err(), "AUTH with wrong password must return WRONGPASS");
+}

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<String> = 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<String, String> = 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<String> = 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<String> = 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");
+}

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::<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.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");
+}

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<String> = 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<Option<String>> = 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");
+}