From f15325f5ac774b2e722910c5d54293fae80036bf Mon Sep 17 00:00:00 2001 From: Alex Le Date: Tue, 15 Sep 2026 09:02:08 -0700 Subject: [PATCH 1/4] feat(pubsub): sharded Pub/Sub for cluster mode Implement the sharded (cluster-mode) Pub/Sub verbs from Part 4 of the Pub/Sub plan (issue #135): - publish(message, channel, sharded: true) -> SPUBLISH; stays batchable inside pipelined/multi - ssubscribe / sunsubscribe (blocking) and ssubscribe_lazy / sunsubscribe_lazy (non-blocking) - connect-time pubsub: { subscriptions: { sharded: [...] } } The four sharded subscribe verbs and the sharded connection config are cluster-only and raise ArgumentError in standalone. publish(sharded: true) is left un-guarded so it stays batchable; glide-core rejects it in standalone. Adds the Valkey#cluster_mode? predicate, read as a method by the command mixin the same way validate_resp3! reads protocol. Signed-off-by: Alex Le --- CHANGELOG.md | 1 + lib/valkey.rb | 26 +++- lib/valkey/commands/pubsub_commands.rb | 55 +++++-- test/integration/valkey/pubsub_test.rb | 191 ++++++++++++++++++++++++- test/unit/glide/pubsub_test.rb | 190 +++++++++++++++++++++++- test/unit/pipeline_test.rb | 20 +++ 6 files changed, 459 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0cf6df6f..e3c8a681 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,7 @@ * `ft_search` is unsupported inside `pipelined`/`multi` (a queued command yields a `Valkey::Future`, not a reply) and with the `flatten_map: true` compatibility option; both raise `ArgumentError`. Use `call` in those cases. * Ruby: PubSub: added support for `subscribe(*channels, timeout_ms:)`, `unsubscribe(*channels, timeout_ms:)`, `publish(message, channel)`, `get_pubsub_message`, `try_get_pubsub_message`, and the connection-time `pubsub: { subscriptions: { exact: [...] } }` option. Requires `protocol: :resp3` ([#308](https://github.com/valkey-io/valkey-glide-ruby/pull/308)) * Ruby: PubSub: added support for `psubscribe(*patterns, timeout_ms:)`, `punsubscribe(*patterns, timeout_ms:)`, `subscribe_lazy(*channels)`, `unsubscribe_lazy(*channels)`, `psubscribe_lazy(*patterns)`, `punsubscribe_lazy(*patterns)`, and the `pubsub: { callback:, context: }` connection option for callback mode delivery ([#316](https://github.com/valkey-io/valkey-glide-ruby/pull/316)) +* Ruby: PubSub: added sharded (cluster-mode) support: `ssubscribe(*channels, timeout_ms:)`, `sunsubscribe(*channels, timeout_ms:)`, `ssubscribe_lazy(*channels)`, `sunsubscribe_lazy(*channels)`, and `publish(message, channel, sharded: true)`, plus the connection-time `pubsub: { subscriptions: { sharded: [...] } }` option. The four sharded subscribe verbs and sharded config require `cluster_mode: true` (raising `ArgumentError` otherwise) and Valkey 7.0.0+; `publish(sharded: true)` stays batchable inside `pipelined`/`multi`. Also adds the `Valkey#cluster_mode?` predicate ([#TBD](https://github.com/valkey-io/valkey-glide-ruby/pulls)) * Ruby: fixed cd workflow to correctly build the ffi with **glibc 2.17** ([#223](https://github.com/valkey-io/valkey-glide-ruby/issues/223)) * Ruby: scripting commands now dispatch real `EVAL` / `EVALSHA` / `SCRIPT LOAD` to the server instead of a client-side script container ([#213](https://github.com/valkey-io/valkey-glide-ruby/issues/213)). Three behavior changes: * `eval` / `evalsha` (and the `_ro` variants) now accept the standard integer key-count form used by `valkey-cli` and the Valkey docs — `eval(script, 1, "mykey", "myarg")`. It previously made the count `KEYS[1]`, shifted the real key into `ARGV[1]`, and dropped the remaining arguments without raising. diff --git a/lib/valkey.rb b/lib/valkey.rb index 61e60db4..19d94988 100644 --- a/lib/valkey.rb +++ b/lib/valkey.rb @@ -325,7 +325,8 @@ def initialize(options = {}) } end - pubsub_config = parse_pubsub_configs(options[:pubsub], protocol: options[:protocol]) + pubsub_config = parse_pubsub_configs(options[:pubsub], protocol: options[:protocol], + cluster_mode: options[:cluster_mode] ? true : false) json_options.merge!(pubsub_config) @pubsub_receiver = Valkey::Glide::PubSubReceiver.make(pubsub_configs: options[:pubsub]) @@ -373,6 +374,18 @@ def initialize(options = {}) @queued_commands = [] end + # Whether this client is connected in cluster mode. + # + # Read as a method (not an ivar) so command mixins can gate cluster-only + # verbs the same way {Commands::PubSubCommands} reads {#protocol}. On a + # {Pipeline}, which mixes in the same commands but has no connection, this + # is never reached: those verbs are rejected earlier by the batch layer. + # + # @return [Boolean] + def cluster_mode? + @cluster_mode + end + # Closes the client and frees the native connection. def close return unless @close_lock&.try_lock @@ -876,19 +889,24 @@ def convert_response(res, &block) # callback: ->(message, context) { ... }, # callback handler # context: my_app_state # callback context # } - def parse_pubsub_configs(pubsub_configs, protocol: nil) + def parse_pubsub_configs(pubsub_configs, protocol: nil, cluster_mode: false) subscriptions = (pubsub_configs || {})[:subscriptions] || {} return {} if subscriptions.empty? - validate_pubsub_subscriptions!(subscriptions, protocol: protocol) + validate_pubsub_subscriptions!(subscriptions, protocol: protocol, cluster_mode: cluster_mode) { "pubsub_subscriptions" => pubsub_subscriptions_to_ffi(subscriptions) } end - def validate_pubsub_subscriptions!(subscriptions, protocol:) + def validate_pubsub_subscriptions!(subscriptions, protocol:, cluster_mode: false) unknown_modes = subscriptions.keys - SUBSCRIPTION_MODES.keys raise ArgumentError, unknown_pubsub_mode_message(unknown_modes) if unknown_modes.any? raise Resp3RequiredError, protocol unless RESP3_VALUES.include?(protocol) + + return if cluster_mode + return if Array(subscriptions[:sharded]).empty? + + raise ArgumentError, "Sharded Pub/Sub subscriptions are only available in cluster mode (cluster_mode: true)" end def pubsub_subscriptions_to_ffi(subscriptions) diff --git a/lib/valkey/commands/pubsub_commands.rb b/lib/valkey/commands/pubsub_commands.rb index b9f35436..26ce1ed5 100644 --- a/lib/valkey/commands/pubsub_commands.rb +++ b/lib/valkey/commands/pubsub_commands.rb @@ -160,12 +160,18 @@ def punsubscribe(*patterns, timeout_ms: 0) # @param [Integer, Float] timeout_ms maximum time in milliseconds to wait for the server to # confirm; `0` blocks indefinitely # @return [void] returns once the server has confirmed the subscription - # @raise [ArgumentError] if timeout_ms is negative + # @raise [ArgumentError] if the client is not in cluster mode, or if timeout_ms is negative + # @raise [Valkey::Resp3RequiredError] GLIDE Pub/Sub requires RESP3 # @raise [Valkey::TimeoutError] if the timeout expires before the server confirms - # @raise [NotImplementedError] this method is not implemented yet # # @see https://valkey.io/commands/ssubscribe/ - def ssubscribe(*channels, timeout_ms: 0) = raise(NotImplementedError, "#{__method__} is not implemented yet") + def ssubscribe(*channels, timeout_ms: 0) + validate_resp3! + validate_cluster_mode!(__method__) + raise ArgumentError, "No channels provided for subscription" if channels.empty? + + send_command(RequestType::SSUBSCRIBE_BLOCKING, channels.map(&:to_s) + [parse_timeout(timeout_ms)]) + end # Unsubscribe from sharded channels, waiting for the server to confirm the change. # @@ -181,12 +187,17 @@ def ssubscribe(*channels, timeout_ms: 0) = raise(NotImplementedError, "#{__metho # @param [Integer, Float] timeout_ms maximum time in milliseconds to wait for the server to # confirm; `0` blocks indefinitely # @return [void] returns once the server has confirmed the change - # @raise [ArgumentError] if timeout_ms is negative + # @raise [ArgumentError] if the client is not in cluster mode, or if timeout_ms is negative + # @raise [Valkey::Resp3RequiredError] GLIDE Pub/Sub requires RESP3 # @raise [Valkey::TimeoutError] if the timeout expires before the server confirms - # @raise [NotImplementedError] this method is not implemented yet # # @see https://valkey.io/commands/sunsubscribe/ - def sunsubscribe(*channels, timeout_ms: 0) = raise(NotImplementedError, "#{__method__} is not implemented yet") + def sunsubscribe(*channels, timeout_ms: 0) + validate_resp3! + validate_cluster_mode!(__method__) + + send_command(RequestType::SUNSUBSCRIBE_BLOCKING, channels.map(&:to_s) + [parse_timeout(timeout_ms)]) + end # Subscribe to exact channels without waiting for the server to confirm. # @@ -272,10 +283,14 @@ def punsubscribe_lazy(*patterns) # # @param [Array] channels the sharded channels to subscribe to; an empty list is rejected # @return [void] returns as soon as the desired subscription state is updated - # @raise [NotImplementedError] this method is not implemented yet + # @raise [ArgumentError] if the client is not in cluster mode, or if the channel list is empty + # @raise [Valkey::Resp3RequiredError] GLIDE Pub/Sub requires RESP3 # # @see https://valkey.io/commands/ssubscribe/ - def ssubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet") + def ssubscribe_lazy(*channels) + validate_cluster_mode!(__method__) + send_lazy_subscription(RequestType::SSUBSCRIBE, channels, reject_empty: true) + end # Unsubscribe from sharded channels without waiting for the server to confirm. # @@ -289,10 +304,14 @@ def ssubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is no # @param [Array] channels the sharded channels to unsubscribe from; an empty list unsubscribes # from all sharded channels # @return [void] returns as soon as the desired subscription state is updated - # @raise [NotImplementedError] this method is not implemented yet + # @raise [ArgumentError] if the client is not in cluster mode + # @raise [Valkey::Resp3RequiredError] GLIDE Pub/Sub requires RESP3 # # @see https://valkey.io/commands/sunsubscribe/ - def sunsubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet") + def sunsubscribe_lazy(*channels) + validate_cluster_mode!(__method__) + send_lazy_subscription(RequestType::SUNSUBSCRIBE, channels) + end # Publish a message on a Pub/Sub channel. # @@ -314,14 +333,12 @@ def sunsubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is # @return [Integer] the number of subscriptions that received the message: in cluster mode the # subscriptions on the node the request was routed to, in standalone the subscriptions on the primary # node, which excludes subscriptions configured on replicas - # @raise [NotImplementedError] sharded publish is not implemented yet # # @see https://valkey.io/commands/publish/ # @see https://valkey.io/commands/spublish/ def publish(message, channel, sharded: false) - raise NotImplementedError, "Sharded publish is not implemented yet" if sharded - - send_command(RequestType::PUBLISH, [channel.to_s, message.to_s]) + request_type = sharded ? RequestType::SPUBLISH : RequestType::PUBLISH + send_command(request_type, [channel.to_s, message.to_s]) end # Get this connection's subscription state: what the client asked for and what the server confirmed. @@ -474,6 +491,16 @@ def validate_resp3! raise Resp3RequiredError, protocol unless RESP3_VALUES.include?(protocol) end + # Sharded Pub/Sub is a cluster-only feature (SSUBSCRIBE et al. exist only + # in cluster mode, since Valkey 7.0.0). Read `cluster_mode?` as a method, + # mirroring how `validate_resp3!` reads `protocol`, so the check stays + # decoupled from the client's ivars. + def validate_cluster_mode!(command) + return if cluster_mode? + + raise ArgumentError, "#{command} is only available in cluster mode (cluster_mode: true)" + end + def send_lazy_subscription(request_type, channels, reject_empty: false, noun: "channels") validate_resp3! raise ArgumentError, "No #{noun} provided for subscription" if reject_empty && channels.empty? diff --git a/test/integration/valkey/pubsub_test.rb b/test/integration/valkey/pubsub_test.rb index 71682877..111460c6 100644 --- a/test/integration/valkey/pubsub_test.rb +++ b/test/integration/valkey/pubsub_test.rb @@ -461,6 +461,181 @@ def test_callback_exception_is_contained subscriber&.close end + # --- Sharded Pub/Sub (cluster mode) -------------------------------------- + + def test_sharded_message_round_trip + skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + + channel = unique_channel + + with_client do |subscriber| + subscriber.ssubscribe(channel) + r.publish("sharded-msg", channel, sharded: true) + received = wait_for_message(subscriber) + + assert_equal "sharded-msg", received.message + assert_equal channel, received.channel + assert_nil received.pattern + end + end + + def test_spublish_returns_the_receiver_count + skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + + channel = unique_channel + + with_client do |subscriber| + assert_equal 0, r.publish("nobody", channel, sharded: true) + + subscriber.ssubscribe(channel) + assert_equal 1, r.publish("counted", channel, sharded: true) + end + end + + def test_sunsubscribe_stops_sharded_delivery + skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + + channel = unique_channel + + with_client do |subscriber| + subscriber.ssubscribe(channel) + r.publish("before", channel, sharded: true) + assert_equal "before", wait_for_message(subscriber).message + + subscriber.sunsubscribe(channel) + + assert_equal 0, r.publish("after", channel, sharded: true) + assert_nil wait_for_message(subscriber, timeout: UNSUB_WAIT_TIME) + end + end + + def test_sunsubscribe_all_sharded_channels + skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + + # Same hash tag keeps both channels in one slot, so a single ssubscribe + # call is routed to one node and both land in the actual subscriptions. + channels = Array.new(2) { |i| unique_channel("{shardtag}-#{i}") } + + with_client do |subscriber| + subscriber.ssubscribe(*channels) + + subscriber.sunsubscribe + + channels.each { |channel| assert_equal 0, r.publish("orphan", channel, sharded: true) } + assert_nil wait_for_message(subscriber, timeout: UNSUB_WAIT_TIME) + end + end + + def test_ssubscribe_lazy_eventually_delivers + skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + + channel = unique_channel + + with_client do |subscriber| + subscriber.ssubscribe_lazy(channel) + + received = publish_until_received_sharded("lazy-sharded-msg", channel, subscriber) + + assert_equal "lazy-sharded-msg", received.message + assert_equal channel, received.channel + assert_nil received.pattern + end + end + + def test_sunsubscribe_lazy_eventually_stops_delivery + skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + + channel = unique_channel + + with_client do |subscriber| + subscriber.ssubscribe_lazy(channel) + publish_until_received_sharded("before", channel, subscriber) + + subscriber.sunsubscribe_lazy(channel) + + assert_delivery_ceased(channel, subscriber, sharded: true) + end + end + + def test_connection_time_sharded_subscription + skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + + channel = unique_channel + + with_client(pubsub: { subscriptions: { sharded: [channel] } }) do |client| + received = publish_until_received_sharded("connect-time-sharded", channel, client) + + assert_equal "connect-time-sharded", received.message + assert_equal channel, received.channel + assert_nil received.pattern + end + end + + def test_sharded_publish_reaches_a_subscriber_in_a_different_slot + skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + + # Distinct hash tags force the two channels onto different slots (and thus + # likely different owning nodes). A same-slot pair would pass even if the + # SPUBLISH were misrouted, so the different-slot case is what proves the + # publish is routed by the channel it names, not the subscriber's node. + subscribed_channel = unique_channel("{slot-a}") + other_slot_channel = unique_channel("{slot-b}") + + with_client do |subscriber| + subscriber.ssubscribe(subscribed_channel) + + # A message on a channel in a different slot must not arrive here. + r.publish("wrong-slot", other_slot_channel, sharded: true) + assert_nil wait_for_message(subscriber, timeout: UNSUB_WAIT_TIME) + + # A message on the subscribed channel must arrive, proving the SPUBLISH + # was routed to the node owning that channel's slot. + r.publish("right-slot", subscribed_channel, sharded: true) + assert_equal "right-slot", wait_for_message(subscriber).message + end + end + + def test_sharded_publish_is_batchable_in_a_pipeline + skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + + channel = unique_channel + + with_client do |subscriber| + subscriber.ssubscribe(channel) + + counts = r.pipelined { |p| p.publish("piped", channel, sharded: true) } + + assert_equal [1], counts + assert_equal "piped", wait_for_message(subscriber).message + end + end + + def test_sharded_verbs_require_cluster_mode_in_standalone + skip "covers the standalone rejection path" if cluster_mode? + + with_client do |client| + { + ssubscribe: -> { client.ssubscribe("shard1") }, + sunsubscribe: -> { client.sunsubscribe }, + ssubscribe_lazy: -> { client.ssubscribe_lazy("shard1") }, + sunsubscribe_lazy: -> { client.sunsubscribe_lazy } + }.each do |name, call| + error = assert_raises(ArgumentError, "#{name} must require cluster mode") { call.call } + assert_match(/cluster mode/, error.message) + end + end + end + + def test_sharded_connection_config_requires_cluster_mode_in_standalone + skip "covers the standalone rejection path" if cluster_mode? + + error = assert_raises(ArgumentError) do + _new_client(protocol: :resp3, pubsub: { subscriptions: { sharded: [unique_channel] } }) + end + + assert_match(/cluster mode/, error.message) + end + private def with_client(options = {}) @@ -490,6 +665,18 @@ def publish_until_received(message, channel, subscriber, timeout: MESSAGE_WAIT_S end end + def publish_until_received_sharded(message, channel, subscriber, timeout: MESSAGE_WAIT_SECONDS) + deadline = monotonic_now + timeout + + loop do + r.publish(message, channel, sharded: true) + received = wait_for_message(subscriber, timeout: POLL_INTERVAL_SECONDS) + return received if received + + flunk("no sharded message on #{channel} within #{timeout}s") if monotonic_now >= deadline + end + end + def wait_until(timeout: MESSAGE_WAIT_SECONDS) deadline = monotonic_now + timeout @@ -523,7 +710,7 @@ def collect_callback_message(queue, payload, channel, timeout: MESSAGE_WAIT_SECO end end - def assert_delivery_ceased(channel, subscriber) + def assert_delivery_ceased(channel, subscriber, sharded: false) quiet_windows = 2 deadline = monotonic_now + MESSAGE_WAIT_SECONDS quiet_count = 0 @@ -531,7 +718,7 @@ def assert_delivery_ceased(channel, subscriber) until quiet_count >= quiet_windows flunk("delivery did not cease on #{channel} within #{MESSAGE_WAIT_SECONDS}s") if monotonic_now >= deadline - r.publish("probe-ceased-#{SecureRandom.hex(4)}", channel) + r.publish("probe-ceased-#{SecureRandom.hex(4)}", channel, sharded: sharded) message = wait_for_message(subscriber, timeout: UNSUB_WAIT_TIME) if message diff --git a/test/unit/glide/pubsub_test.rb b/test/unit/glide/pubsub_test.rb index 2c0561d6..c5e53e2b 100644 --- a/test/unit/glide/pubsub_test.rb +++ b/test/unit/glide/pubsub_test.rb @@ -16,10 +16,11 @@ class RecordingClient < Valkey attr_reader :sent_commands - def initialize(response: nil, protocol: :resp3, callback: nil) # rubocop:disable Lint/MissingSuper + def initialize(response: nil, protocol: :resp3, callback: nil, cluster_mode: false) # rubocop:disable Lint/MissingSuper @sent_commands = [] @response = response @protocol = protocol + @cluster_mode = cluster_mode @pubsub_receiver = Valkey::Glide::PubSubReceiver.new(callback: callback) @close_lock = Mutex.new @pid = Process.pid @@ -120,13 +121,29 @@ def test_pubsub_parse_config_ok subscriptions: { exact: ["news", :symbols], pattern: ["news.*"], sharded: ["news.shard"] } } - parsed = parse_pubsub_configs.call(pubsub_config, protocol: :resp3) + parsed = parse_pubsub_configs.call(pubsub_config, protocol: :resp3, cluster_mode: true) expected = { "pubsub_subscriptions" => { "0" => %w[news symbols], "1" => ["news.*"], "2" => ["news.shard"] } } assert_equal expected, parsed end + def test_pubsub_parse_config_without_sharded_does_not_require_cluster_mode + pubsub_config = { subscriptions: { exact: ["news"], pattern: ["news.*"] } } + + parsed = parse_pubsub_configs.call(pubsub_config, protocol: :resp3) + + assert_equal({ "pubsub_subscriptions" => { "0" => ["news"], "1" => ["news.*"] } }, parsed) + end + + def test_pubsub_sharded_config_requires_cluster_mode + error = assert_raises(ArgumentError) do + parse_pubsub_configs.call({ subscriptions: { sharded: ["shard-chan"] } }, protocol: :resp3) + end + + assert_match(/cluster mode/, error.message) + end + def test_pubsub_parse_config_nil config = parse_pubsub_configs.call(nil) expected = parse_pubsub_configs.call({}) @@ -402,11 +419,168 @@ def test_publish_works_without_resp3 assert_equal %w[news hello], client.last_command.args end + # --- Sharded Pub/Sub (cluster mode) -------------------------------------- + + def test_publish_defaults_to_unsharded + client = RecordingClient.new(response: 0) + + client.publish("hello", "news") + + assert_equal Valkey::RequestType::PUBLISH, client.last_command.request_type + assert_equal %w[news hello], client.last_command.args + end + + def test_publish_sharded_uses_spublish_with_channel_first + client = RecordingClient.new(response: 0, cluster_mode: true) + + client.publish("hello", "shard-chan", sharded: true) + + assert_equal Valkey::RequestType::SPUBLISH, client.last_command.request_type + # Signature is (message, channel); wire order is . + assert_equal %w[shard-chan hello], client.last_command.args + end + + def test_publish_sharded_works_without_resp3 + client = RecordingClient.new(response: 0, protocol: nil, cluster_mode: true) + + client.publish("hello", "shard-chan", sharded: true) + + assert_equal Valkey::RequestType::SPUBLISH, client.last_command.request_type + assert_equal %w[shard-chan hello], client.last_command.args + end + + def test_ssubscribe_dispatches_sblocking_with_timeout + client = RecordingClient.new(cluster_mode: true) + + client.ssubscribe("shard1", "shard2", timeout_ms: 2000) + + assert_equal Valkey::RequestType::SSUBSCRIBE_BLOCKING, client.last_command.request_type + assert_equal %w[shard1 shard2 2000], client.last_command.args + end + + def test_ssubscribe_defaults_to_indefinite_timeout + client = RecordingClient.new(cluster_mode: true) + + client.ssubscribe("shard1") + + assert_equal %w[shard1 0], client.last_command.args + end + + def test_ssubscribe_coerces_arguments_and_truncates_timeout + client = RecordingClient.new(cluster_mode: true) + + client.ssubscribe(:shard1, 42, timeout_ms: 1500.6) + + assert_equal %w[shard1 42 1500], client.last_command.args + end + + def test_ssubscribe_rounds_sub_millisecond_timeout_up + client = RecordingClient.new(cluster_mode: true) + + client.ssubscribe("shard1", timeout_ms: 0.4) + + assert_equal %w[shard1 1], client.last_command.args + end + + def test_ssubscribe_rejects_a_negative_timeout + client = RecordingClient.new(cluster_mode: true) + + assert_raises(ArgumentError) { client.ssubscribe("shard1", timeout_ms: -1) } + assert_empty client.sent_commands + end + + def test_ssubscribe_without_channels_raises + client = RecordingClient.new(cluster_mode: true) + + error = assert_raises(ArgumentError) { client.ssubscribe } + + assert_equal "No channels provided for subscription", error.message + assert_empty client.sent_commands + end + + def test_sunsubscribe_dispatches_sblocking_with_timeout + client = RecordingClient.new(cluster_mode: true) + + client.sunsubscribe("shard1", timeout_ms: 3000) + + assert_equal Valkey::RequestType::SUNSUBSCRIBE_BLOCKING, client.last_command.request_type + assert_equal %w[shard1 3000], client.last_command.args + end + + def test_sunsubscribe_without_channels_targets_all + client = RecordingClient.new(cluster_mode: true) + + client.sunsubscribe + + assert_equal Valkey::RequestType::SUNSUBSCRIBE_BLOCKING, client.last_command.request_type + assert_equal %w[0], client.last_command.args + end + + def test_lazy_sharded_verbs_dispatch_without_a_timeout_argument + client = RecordingClient.new(cluster_mode: true) + + client.ssubscribe_lazy("shard1") + client.sunsubscribe_lazy("shard1") + + expected = [ + [Valkey::RequestType::SSUBSCRIBE, %w[shard1]], + [Valkey::RequestType::SUNSUBSCRIBE, %w[shard1]] + ] + + assert_equal(expected, client.sent_commands.map { |command| [command.request_type, command.args] }) + end + + def test_sunsubscribe_lazy_without_channels_targets_all + client = RecordingClient.new(cluster_mode: true) + + client.sunsubscribe_lazy + + assert_equal Valkey::RequestType::SUNSUBSCRIBE, client.last_command.request_type + assert_empty client.last_command.args + end + + def test_ssubscribe_lazy_without_channels_raises + client = RecordingClient.new(cluster_mode: true) + + error = assert_raises(ArgumentError) { client.ssubscribe_lazy } + + assert_equal "No channels provided for subscription", error.message + assert_empty client.sent_commands + end + + def test_sharded_verbs_require_cluster_mode + client = RecordingClient.new # standalone + + { + ssubscribe: -> { client.ssubscribe("shard1") }, + sunsubscribe: -> { client.sunsubscribe }, + ssubscribe_lazy: -> { client.ssubscribe_lazy("shard1") }, + sunsubscribe_lazy: -> { client.sunsubscribe_lazy } + }.each do |name, call| + error = assert_raises(ArgumentError, "#{name} must require cluster mode") { call.call } + assert_match(/cluster mode/, error.message) + end + + assert_empty client.sent_commands + end + + def test_sharded_publish_is_allowed_in_standalone + # publish stays batchable and un-guarded; the core decides. It must not + # raise the client-side cluster-mode ArgumentError. + client = RecordingClient.new(response: 0) # standalone + + client.publish("hello", "shard-chan", sharded: true) + + assert_equal Valkey::RequestType::SPUBLISH, client.last_command.request_type + end + # --- RESP3 requirement --------------------------------------------------- def test_subscription_methods_reject_a_non_resp3_protocol [nil, :resp2, "resp2", 2].each do |protocol| - client = RecordingClient.new(protocol: protocol) + # Cluster mode so the sharded verbs clear their cluster-only guard and + # reach the RESP3 check; the non-sharded verbs are unaffected by it. + client = RecordingClient.new(protocol: protocol, cluster_mode: true) guarded_calls(client).each do |name, call| # Timeout so a missing guard fails the assertion instead of blocking in @@ -421,7 +595,7 @@ def test_subscription_methods_reject_a_non_resp3_protocol def test_subscription_methods_accept_every_resp3_spelling [:resp3, "resp3", 3].each do |protocol| - client = RecordingClient.new(protocol: protocol) + client = RecordingClient.new(protocol: protocol, cluster_mode: true) client.subscribe("news") client.unsubscribe @@ -431,6 +605,10 @@ def test_subscription_methods_accept_every_resp3_spelling client.unsubscribe_lazy client.psubscribe_lazy("news.*") client.punsubscribe_lazy + client.ssubscribe("shard1") + client.sunsubscribe + client.ssubscribe_lazy("shard1") + client.sunsubscribe_lazy assert_nil client.try_get_pubsub_message, "protocol #{protocol.inspect} must be accepted" assert_equal "hello", queued_message(client).message @@ -469,6 +647,10 @@ def guarded_calls(client) unsubscribe_lazy: -> { client.unsubscribe_lazy }, psubscribe_lazy: -> { client.psubscribe_lazy("news.*") }, punsubscribe_lazy: -> { client.punsubscribe_lazy }, + ssubscribe: -> { client.ssubscribe("shard1") }, + sunsubscribe: -> { client.sunsubscribe }, + ssubscribe_lazy: -> { client.ssubscribe_lazy("shard1") }, + sunsubscribe_lazy: -> { client.sunsubscribe_lazy }, get_pubsub_message: -> { client.get_pubsub_message }, try_get_pubsub_message: -> { client.try_get_pubsub_message } } diff --git a/test/unit/pipeline_test.rb b/test/unit/pipeline_test.rb index 74ea3494..1e643bc6 100644 --- a/test/unit/pipeline_test.rb +++ b/test/unit/pipeline_test.rb @@ -51,4 +51,24 @@ def test_pubsub_commands_raise_argument_error assert_equal "#{name} is not supported inside pipelined/multi", error.message end end + + def test_sharded_subscribe_verbs_are_unsupported_in_a_pipeline + %i[ssubscribe sunsubscribe ssubscribe_lazy sunsubscribe_lazy].each do |name| + assert_includes Valkey::Pipeline::PUBSUB_UNSUPPORTED, name + end + end + + def test_publish_is_batchable_including_sharded + pipeline = Valkey::Pipeline.new + + plain = pipeline.publish("hello", "news") + sharded = pipeline.publish("hello", "shard-chan", sharded: true) + + assert_instance_of Valkey::Future, plain + assert_instance_of Valkey::Future, sharded + assert_equal [ + [Valkey::RequestType::PUBLISH, %w[news hello], nil], + [Valkey::RequestType::SPUBLISH, %w[shard-chan hello], nil] + ], pipeline.commands + end end From 267792a04659d66e1382fddc8398049010d6bfc0 Mon Sep 17 00:00:00 2001 From: Alex Le Date: Tue, 15 Sep 2026 09:02:57 -0700 Subject: [PATCH 2/4] docs(pubsub): link CHANGELOG entry to PR #317 Signed-off-by: Alex Le --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e3c8a681..ccc29f3c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,7 +30,7 @@ * `ft_search` is unsupported inside `pipelined`/`multi` (a queued command yields a `Valkey::Future`, not a reply) and with the `flatten_map: true` compatibility option; both raise `ArgumentError`. Use `call` in those cases. * Ruby: PubSub: added support for `subscribe(*channels, timeout_ms:)`, `unsubscribe(*channels, timeout_ms:)`, `publish(message, channel)`, `get_pubsub_message`, `try_get_pubsub_message`, and the connection-time `pubsub: { subscriptions: { exact: [...] } }` option. Requires `protocol: :resp3` ([#308](https://github.com/valkey-io/valkey-glide-ruby/pull/308)) * Ruby: PubSub: added support for `psubscribe(*patterns, timeout_ms:)`, `punsubscribe(*patterns, timeout_ms:)`, `subscribe_lazy(*channels)`, `unsubscribe_lazy(*channels)`, `psubscribe_lazy(*patterns)`, `punsubscribe_lazy(*patterns)`, and the `pubsub: { callback:, context: }` connection option for callback mode delivery ([#316](https://github.com/valkey-io/valkey-glide-ruby/pull/316)) -* Ruby: PubSub: added sharded (cluster-mode) support: `ssubscribe(*channels, timeout_ms:)`, `sunsubscribe(*channels, timeout_ms:)`, `ssubscribe_lazy(*channels)`, `sunsubscribe_lazy(*channels)`, and `publish(message, channel, sharded: true)`, plus the connection-time `pubsub: { subscriptions: { sharded: [...] } }` option. The four sharded subscribe verbs and sharded config require `cluster_mode: true` (raising `ArgumentError` otherwise) and Valkey 7.0.0+; `publish(sharded: true)` stays batchable inside `pipelined`/`multi`. Also adds the `Valkey#cluster_mode?` predicate ([#TBD](https://github.com/valkey-io/valkey-glide-ruby/pulls)) +* Ruby: PubSub: added sharded (cluster-mode) support: `ssubscribe(*channels, timeout_ms:)`, `sunsubscribe(*channels, timeout_ms:)`, `ssubscribe_lazy(*channels)`, `sunsubscribe_lazy(*channels)`, and `publish(message, channel, sharded: true)`, plus the connection-time `pubsub: { subscriptions: { sharded: [...] } }` option. The four sharded subscribe verbs and sharded config require `cluster_mode: true` (raising `ArgumentError` otherwise) and Valkey 7.0.0+; `publish(sharded: true)` stays batchable inside `pipelined`/`multi`. Also adds the `Valkey#cluster_mode?` predicate ([#317](https://github.com/valkey-io/valkey-glide-ruby/pull/317)) * Ruby: fixed cd workflow to correctly build the ffi with **glibc 2.17** ([#223](https://github.com/valkey-io/valkey-glide-ruby/issues/223)) * Ruby: scripting commands now dispatch real `EVAL` / `EVALSHA` / `SCRIPT LOAD` to the server instead of a client-side script container ([#213](https://github.com/valkey-io/valkey-glide-ruby/issues/213)). Three behavior changes: * `eval` / `evalsha` (and the `_ro` variants) now accept the standard integer key-count form used by `valkey-cli` and the Valkey docs — `eval(script, 1, "mykey", "myarg")`. It previously made the count `KEYS[1]`, shifted the real key into `ARGV[1]`, and dropped the remaining arguments without raising. From bbaeb7b6b9a253a49aac5469250cc4975828e182 Mon Sep 17 00:00:00 2001 From: Alex Le Date: Thu, 17 Sep 2026 12:19:41 -0700 Subject: [PATCH 3/4] test(pubsub): gate sharded Pub/Sub tests on server 7.0 The sharded verbs were gated on cluster mode only, so on CI's redis 6.2 cluster leg they ran against a server with no SSUBSCRIBE. The blocking ssubscribe never returned and glide-core retried the permanent error without backoff, flooding the log until the job hit its timeout. Add the missing omit_version("7.0") gate alongside the cluster-mode check, in one shared private helper. Signed-off-by: Alex Le --- test/integration/valkey/pubsub_test.rb | 24 +++++++++++++++--------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/test/integration/valkey/pubsub_test.rb b/test/integration/valkey/pubsub_test.rb index 111460c6..6b0402ba 100644 --- a/test/integration/valkey/pubsub_test.rb +++ b/test/integration/valkey/pubsub_test.rb @@ -464,7 +464,7 @@ def test_callback_exception_is_contained # --- Sharded Pub/Sub (cluster mode) -------------------------------------- def test_sharded_message_round_trip - skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + skip_unless_sharded_pubsub channel = unique_channel @@ -480,7 +480,7 @@ def test_sharded_message_round_trip end def test_spublish_returns_the_receiver_count - skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + skip_unless_sharded_pubsub channel = unique_channel @@ -493,7 +493,7 @@ def test_spublish_returns_the_receiver_count end def test_sunsubscribe_stops_sharded_delivery - skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + skip_unless_sharded_pubsub channel = unique_channel @@ -510,7 +510,7 @@ def test_sunsubscribe_stops_sharded_delivery end def test_sunsubscribe_all_sharded_channels - skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + skip_unless_sharded_pubsub # Same hash tag keeps both channels in one slot, so a single ssubscribe # call is routed to one node and both land in the actual subscriptions. @@ -527,7 +527,7 @@ def test_sunsubscribe_all_sharded_channels end def test_ssubscribe_lazy_eventually_delivers - skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + skip_unless_sharded_pubsub channel = unique_channel @@ -543,7 +543,7 @@ def test_ssubscribe_lazy_eventually_delivers end def test_sunsubscribe_lazy_eventually_stops_delivery - skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + skip_unless_sharded_pubsub channel = unique_channel @@ -558,7 +558,7 @@ def test_sunsubscribe_lazy_eventually_stops_delivery end def test_connection_time_sharded_subscription - skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + skip_unless_sharded_pubsub channel = unique_channel @@ -572,7 +572,7 @@ def test_connection_time_sharded_subscription end def test_sharded_publish_reaches_a_subscriber_in_a_different_slot - skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + skip_unless_sharded_pubsub # Distinct hash tags force the two channels onto different slots (and thus # likely different owning nodes). A same-slot pair would pass even if the @@ -596,7 +596,7 @@ def test_sharded_publish_reaches_a_subscriber_in_a_different_slot end def test_sharded_publish_is_batchable_in_a_pipeline - skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + skip_unless_sharded_pubsub channel = unique_channel @@ -638,6 +638,12 @@ def test_sharded_connection_config_requires_cluster_mode_in_standalone private + def skip_unless_sharded_pubsub + skip "sharded Pub/Sub is cluster-only" unless cluster_mode? + + omit_version("7.0") + end + def with_client(options = {}) subscriber = _new_client(options.merge(protocol: :resp3)) yield subscriber From 8a1dd04fa500d59485be1fae39651f0a0c239834 Mon Sep 17 00:00:00 2001 From: Alex Le Date: Thu, 17 Sep 2026 14:27:21 -0700 Subject: [PATCH 4/4] some refactoring Signed-off-by: Alex Le --- CHANGELOG.md | 2 +- lib/valkey.rb | 12 +++--------- lib/valkey/commands/pubsub_commands.rb | 6 +----- 3 files changed, 5 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ccc29f3c..5be95f3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,7 +30,7 @@ * `ft_search` is unsupported inside `pipelined`/`multi` (a queued command yields a `Valkey::Future`, not a reply) and with the `flatten_map: true` compatibility option; both raise `ArgumentError`. Use `call` in those cases. * Ruby: PubSub: added support for `subscribe(*channels, timeout_ms:)`, `unsubscribe(*channels, timeout_ms:)`, `publish(message, channel)`, `get_pubsub_message`, `try_get_pubsub_message`, and the connection-time `pubsub: { subscriptions: { exact: [...] } }` option. Requires `protocol: :resp3` ([#308](https://github.com/valkey-io/valkey-glide-ruby/pull/308)) * Ruby: PubSub: added support for `psubscribe(*patterns, timeout_ms:)`, `punsubscribe(*patterns, timeout_ms:)`, `subscribe_lazy(*channels)`, `unsubscribe_lazy(*channels)`, `psubscribe_lazy(*patterns)`, `punsubscribe_lazy(*patterns)`, and the `pubsub: { callback:, context: }` connection option for callback mode delivery ([#316](https://github.com/valkey-io/valkey-glide-ruby/pull/316)) -* Ruby: PubSub: added sharded (cluster-mode) support: `ssubscribe(*channels, timeout_ms:)`, `sunsubscribe(*channels, timeout_ms:)`, `ssubscribe_lazy(*channels)`, `sunsubscribe_lazy(*channels)`, and `publish(message, channel, sharded: true)`, plus the connection-time `pubsub: { subscriptions: { sharded: [...] } }` option. The four sharded subscribe verbs and sharded config require `cluster_mode: true` (raising `ArgumentError` otherwise) and Valkey 7.0.0+; `publish(sharded: true)` stays batchable inside `pipelined`/`multi`. Also adds the `Valkey#cluster_mode?` predicate ([#317](https://github.com/valkey-io/valkey-glide-ruby/pull/317)) +* Ruby: PubSub: added support for Sharded pubsub commands([#317](https://github.com/valkey-io/valkey-glide-ruby/pull/317)) * Ruby: fixed cd workflow to correctly build the ffi with **glibc 2.17** ([#223](https://github.com/valkey-io/valkey-glide-ruby/issues/223)) * Ruby: scripting commands now dispatch real `EVAL` / `EVALSHA` / `SCRIPT LOAD` to the server instead of a client-side script container ([#213](https://github.com/valkey-io/valkey-glide-ruby/issues/213)). Three behavior changes: * `eval` / `evalsha` (and the `_ro` variants) now accept the standard integer key-count form used by `valkey-cli` and the Valkey docs — `eval(script, 1, "mykey", "myarg")`. It previously made the count `KEYS[1]`, shifted the real key into `ARGV[1]`, and dropped the remaining arguments without raising. diff --git a/lib/valkey.rb b/lib/valkey.rb index 19d94988..3aeb9d92 100644 --- a/lib/valkey.rb +++ b/lib/valkey.rb @@ -374,12 +374,7 @@ def initialize(options = {}) @queued_commands = [] end - # Whether this client is connected in cluster mode. - # - # Read as a method (not an ivar) so command mixins can gate cluster-only - # verbs the same way {Commands::PubSubCommands} reads {#protocol}. On a - # {Pipeline}, which mixes in the same commands but has no connection, this - # is never reached: those verbs are rejected earlier by the batch layer. + # True if client is in cluster mode. # # @return [Boolean] def cluster_mode? @@ -903,10 +898,9 @@ def validate_pubsub_subscriptions!(subscriptions, protocol:, cluster_mode: false raise ArgumentError, unknown_pubsub_mode_message(unknown_modes) if unknown_modes.any? raise Resp3RequiredError, protocol unless RESP3_VALUES.include?(protocol) - return if cluster_mode - return if Array(subscriptions[:sharded]).empty? + return unless Array(subscriptions[:sharded]).any? && !cluster_mode - raise ArgumentError, "Sharded Pub/Sub subscriptions are only available in cluster mode (cluster_mode: true)" + raise ArgumentError, "Sharded Pub/Sub subscriptions are only available in cluster mode." end def pubsub_subscriptions_to_ffi(subscriptions) diff --git a/lib/valkey/commands/pubsub_commands.rb b/lib/valkey/commands/pubsub_commands.rb index 26ce1ed5..f16f6cb9 100644 --- a/lib/valkey/commands/pubsub_commands.rb +++ b/lib/valkey/commands/pubsub_commands.rb @@ -491,14 +491,10 @@ def validate_resp3! raise Resp3RequiredError, protocol unless RESP3_VALUES.include?(protocol) end - # Sharded Pub/Sub is a cluster-only feature (SSUBSCRIBE et al. exist only - # in cluster mode, since Valkey 7.0.0). Read `cluster_mode?` as a method, - # mirroring how `validate_resp3!` reads `protocol`, so the check stays - # decoupled from the client's ivars. def validate_cluster_mode!(command) return if cluster_mode? - raise ArgumentError, "#{command} is only available in cluster mode (cluster_mode: true)" + raise ArgumentError, "#{command} is only available in cluster mode." end def send_lazy_subscription(request_type, channels, reject_empty: false, noun: "channels")