fix(redis): expect a single ack count for targeted broadcast with ack

crates/socketioxide-redis/src/lib.rs

@@ -506,7 +506,13 @@ impl<E: SocketEmitter, R: Driver> CoreAdapter<E> for CustomRedisAdapter<E, R> {
         let req = RequestOut::new(self.uid, RequestTypeOut::BroadcastWithAck(&packet), &opts);
         let req_id = req.id;
 
-        let remote_serv_cnt = self.server_count().await?.saturating_sub(1);
+        // When the request targets a specific server, only this server will answer,
+        // otherwise all the other servers will.
+        let remote_serv_cnt = if opts.server_id.is_none() {
+            self.server_count().await?.saturating_sub(1)
+        } else {
+            1
+        };
 
         let (tx, rx) = mpsc::channel(self.config.ack_response_buffer + remote_serv_cnt as usize);
         self.responses.lock().unwrap().insert(req_id, tx);

crates/socketioxide-redis/tests/sockets.rs

@@ -168,3 +168,66 @@ pub async fn remote_socket_emit_with_ack() {
     assert_eq!(timeout_rcv!(&mut rx1), r#"421["test","hello"]"#);
     assert_eq!(timeout_rcv!(&mut rx2), r#"421["test","hello"]"#);
 }
+
+#[tokio::test]
+pub async fn remote_socket_emit_with_ack_targeted_terminates() {
+    // A targeted ack request is only sent to the server owning the socket, so
+    // the stream must terminate as soon as this server answered. It only
+    // manifests with 3+ servers, as with 2 servers the targeted count is
+    // equal to `server_count - 1`.
+    let [io1, io2, io3] = fixture::spawn_servers::<3>();
+
+    io1.ns("/", async || ()).await.unwrap();
+    io2.ns("/", async || ()).await.unwrap();
+    io3.ns("/", async || ()).await.unwrap();
+
+    let (_, mut rx1) = io1.new_dummy_sock("/", ()).await;
+    let (stx2, mut rx2) = io2.new_dummy_sock("/", ()).await;
+    let (stx3, mut rx3) = io3.new_dummy_sock("/", ()).await;
+
+    timeout_rcv!(&mut rx1); // connect packet
+    timeout_rcv!(&mut rx2); // connect packet
+    timeout_rcv!(&mut rx3); // connect packet
+
+    use futures_util::StreamExt;
+
+    let local_id = io1.config().server_id;
+    let io2_id = io2.config().server_id;
+    let mut tested = 0;
+    let sockets = io1.fetch_sockets().await.unwrap();
+    for socket in sockets {
+        // Only the remote sockets take the targeted network path.
+        if socket.data().server_id == local_id {
+            continue;
+        }
+        let (stx, rx) = if socket.data().server_id == io2_id {
+            (&stx2, &mut rx2)
+        } else {
+            (&stx3, &mut rx3)
+        };
+
+        let stream = socket
+            .emit_with_ack::<_, [String; 1]>("test", "hello")
+            .await
+            .unwrap();
+
+        // Make the remote client answer the ack.
+        let packet = timeout_rcv!(rx, 1000); // "42<ack_id>[\"test\",\"hello\"]"
+        let ack_id = &packet[2..packet.find('[').unwrap()];
+        stx.send(engineioxide::Packet::Message(
+            format!("3{ack_id}[\"oof\"]").into(),
+        ))
+        .await
+        .unwrap();
+
+        futures_util::pin_mut!(stream);
+        let ack = stream.next().await;
+        assert!(matches!(ack, Some((_, Ok(_)))), "expected the ack");
+        assert!(
+            futures_core::FusedStream::is_terminated(&stream),
+            "the ack stream should terminate right after the targeted ack"
+        );
+        tested += 1;
+    }
+    assert!(tested > 0, "expected remote sockets to be tested");
+}