diff --git a/CHANGELOG.md b/CHANGELOG.md
index 310f5e3..fcbbb66 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,15 @@ Versions up to and including `0.19.3` are documented in the
## [Unreleased]
+### Changed
+
+- **`ssubscribe` accepts shard channels of several slots in a cluster.** The
+ server requires one slot per `SSUBSCRIBE`, and the client refused a
+ cross-slot one with `MismatchedKeySlots`. A shard subscription is per
+ channel, so nothing is lost by splitting it: one `SSUBSCRIBE` is now sent per
+ slot, to the master owning it, and the caller is answered once every master
+ has confirmed.
+
### Fixed
- **A sharded `PubSubStream` spanning several slots is released in a
diff --git a/src/commands/pub_sub_commands.rs b/src/commands/pub_sub_commands.rs
index 0d2b928..b772bec 100644
--- a/src/commands/pub_sub_commands.rs
+++ b/src/commands/pub_sub_commands.rs
@@ -186,10 +186,15 @@ pub trait PubSubCommands<'a>: Sized {
prepare_command(self, cmd("SPUBLISH").key(shardchannel).arg(message))
}
- /// Subscribes the client to the specified channels.
+ /// Subscribes the client to the specified shard channels.
+ ///
+ /// On a cluster, the shard channels may hash to different slots: one
+ /// `SSUBSCRIBE` is sent per slot, to the master owning it, and the call
+ /// returns once every one of them is confirmed. The same goes for the
+ /// `SUNSUBSCRIBE` the stream sends when it is closed or dropped.
///
/// # See Also
- /// [](https://redis.io/commands/subscribe/)
+ /// [](https://redis.io/commands/ssubscribe/)
#[allow(async_fn_in_trait)]
async fn ssubscribe(self, shardchannels: impl Serialize) -> Result;
diff --git a/src/network/cluster_connection.rs b/src/network/cluster_connection.rs
index 7d4284e..82c16a0 100644
--- a/src/network/cluster_connection.rs
+++ b/src/network/cluster_connection.rs
@@ -151,18 +151,21 @@ fn is_broadcast_pub_sub_command(command: &Command) -> bool {
)
}
-/// Whether the command is a `SUNSUBSCRIBE` naming shard channels of more than
-/// one slot.
+/// Whether the command is a `SSUBSCRIBE` or a `SUNSUBSCRIBE` naming shard
+/// channels of more than one slot.
///
-/// A single `SSUBSCRIBE` must name one slot, but a `PubSubStream` may add shard
-/// channels of other slots afterwards, and its `close` and `drop` release them
-/// all in one `SUNSUBSCRIBE`. Routed as a single-slot command, that cleanup was
-/// refused before reaching any node: every subscription stayed on its master
-/// and in the client's registry. Cut per slot like a multi-shard command, each
-/// master hears about its own shard channels — the acknowledgements are matched
-/// per channel, so the caller is answered once all of them are in.
-fn is_shard_unsubscribe_spanning_slots(command: &Command) -> bool {
- if command.name() != b"SUNSUBSCRIBE" {
+/// The server requires a single slot per command, but a shard subscription is
+/// per channel: nothing is atomic about subscribing to two of them at once.
+/// Cut per slot like a multi-shard command, each master hears about its own
+/// shard channels, and the acknowledgements are matched per channel by the
+/// router, so the caller is answered once all of them are in.
+///
+/// For `SUNSUBSCRIBE` this is not a convenience: a `PubSubStream` collects
+/// shard channels one `ssubscribe` at a time, and its `close` and `drop`
+/// release them all in one command. Routed as a single-slot command, that
+/// cleanup was refused before reaching any node.
+fn is_shard_pub_sub_spanning_slots(command: &Command) -> bool {
+ if !matches!(command.name(), b"SSUBSCRIBE" | b"SUNSUBSCRIBE") {
return false;
}
@@ -205,7 +208,7 @@ fn route(command: &Command) -> Result {
});
}
- if is_shard_unsubscribe_spanning_slots(command) {
+ if is_shard_pub_sub_spanning_slots(command) {
return Ok(Route::MultiShard);
}
diff --git a/src/tests/pub_sub_commands.rs b/src/tests/pub_sub_commands.rs
index 34ce55f..9e146d6 100644
--- a/src/tests/pub_sub_commands.rs
+++ b/src/tests/pub_sub_commands.rs
@@ -1597,39 +1597,69 @@ async fn closing_a_shard_stream_spanning_slots_succeeds_in_cluster() -> Result<(
Ok(())
}
-/// A SSUBSCRIBE refused before being sent -- its shard channels span two
-/// slots -- must leave nothing behind: the client's next subscriptions are
-/// confirmed and receive their messages.
+/// A single `ssubscribe` may name shard channels of several slots: the cluster
+/// connection sends one `SSUBSCRIBE` per slot, to the master owning it. The
+/// server refuses a cross-slot one with `CROSSSLOT`, and the client used to
+/// refuse it itself with `MismatchedKeySlots`.
+///
+/// This is the reproduction of the original report, run to the end: subscribe,
+/// receive from both masters, drop, subscribe again.
#[tokio::test]
#[serial]
-async fn a_refused_cross_slot_ssubscribe_leaves_the_client_usable() -> Result<()> {
+async fn a_cross_slot_ssubscribe_is_split_per_slot_in_cluster() -> Result<()> {
log_try_init();
+ // `{a}` and `{b}` hash to slots served by two different masters of the
+ // three-node test cluster.
+ const CHANNELS: [&str; 2] = ["cross_slot_ssubscribe{a}", "cross_slot_ssubscribe{b}"];
+
let client = get_cluster_test_client().await?;
let publisher = get_cluster_test_client().await?;
- let refused = client
- .ssubscribe(["refused_ssubscribe{a}", "refused_ssubscribe{b}"])
- .await;
- assert!(matches!(
- refused.map(|_| ()).unwrap_err().kind(),
- ErrorKind::Client(ClientError::MismatchedKeySlots)
- ));
+ // A shard subscription lives on the master owning its slot only, so each
+ // master is asked about its own.
+ let shard_results: Vec = client.cluster_shards().await?;
+ let mut master_clients = Vec::new();
+ for shard in &shard_results {
+ let master_node = shard.nodes.iter().find(|n| n.role == "master").unwrap();
+ master_clients.push(
+ Client::connect((master_node.ip.clone(), master_node.port.unwrap()).into_config()?)
+ .await?,
+ );
+ }
- let mut stream = timeout(
- Duration::from_secs(5),
- TimeoutKind::Command,
- client.ssubscribe("refused_ssubscribe{a}"),
- )
- .await??;
- publisher
- .spublish("refused_ssubscribe{a}", "payload")
+ for round in 0..2 {
+ let mut stream = timeout(
+ Duration::from_secs(5),
+ TimeoutKind::Command,
+ client.ssubscribe(CHANNELS),
+ )
+ .await??;
+
+ for channel in CHANNELS {
+ publisher.spublish(channel, round.to_string()).await?;
+ let message = timeout(Duration::from_secs(5), TimeoutKind::Command, stream.next())
+ .await?
+ .unwrap()?;
+ assert_eq!(channel.as_bytes(), message.channel());
+ }
+
+ drop(stream);
+
+ // The client clears its registry only when each master confirms the
+ // unsubscription, which `drop` does not wait for.
+ wait_for(Duration::from_secs(5), || async {
+ let mut subscribed = HashSet::::new();
+ for master_client in &master_clients {
+ let channels: HashSet = master_client
+ .pub_sub_shardchannels("cross_slot_ssubscribe*")
+ .await?;
+ subscribed.extend(channels);
+ }
+ Ok(subscribed.is_empty())
+ })
.await?;
- let message = timeout(Duration::from_secs(5), TimeoutKind::Command, stream.next())
- .await?
- .unwrap()?;
- assert_eq!(b"payload", message.payload());
+ }
- stream.close().await?;
Ok(())
}