Merge pull request rails#46562 from palkan/feat/action-cable-redis-resubscribe

Restore Action Cable Redis pub/sub listener on connection failure

Author: byroot

actioncable/CHANGELOG.md

@@ -1,3 +1,7 @@
+*   Redis pub/sub adapter now automatically reconnects when Redis connection is lost.
+
+    *Vladimir Dementyev*
+
 *   The `connected()` callback can now take a `{reconnected}` parameter to differentiate
     connections from reconnections.
 

actioncable/lib/action_cable/subscription_adapter/redis.rb

@@ -44,7 +44,7 @@ def redis_connection_for_subscriptions
 
       private
         def listener
-          @listener || @server.mutex.synchronize { @listener ||= Listener.new(self, @server.event_loop) }
+          @listener || @server.mutex.synchronize { @listener ||= Listener.new(self, config_options, @server.event_loop) }
         end
 
         def redis_connection_for_broadcasts
@@ -54,11 +54,15 @@ def redis_connection_for_broadcasts
         end
 
         def redis_connection
-          self.class.redis_connector.call(@server.config.cable.symbolize_keys.merge(id: identifier))
+          self.class.redis_connector.call(config_options)
+        end
+
+        def config_options
+          @config_options ||= @server.config.cable.symbolize_keys.merge(id: identifier)
         end
 
         class Listener < SubscriberMap
-          def initialize(adapter, event_loop)
+          def initialize(adapter, config_options, event_loop)
             super()
 
             @adapter = adapter
@@ -67,6 +71,11 @@ def initialize(adapter, event_loop)
             @subscribe_callbacks = Hash.new { |h, k| h[k] = [] }
             @subscription_lock = Mutex.new
 
+            @reconnect_attempt = 0
+            # Use the same config as used by Redis conn
+            @reconnect_attempts = config_options.fetch(:reconnect_attempts, 1)
+            @reconnect_attempts = Array.new(@reconnect_attempts, 0) if @reconnect_attempts.is_a?(Integer)
+
             @subscribed_client = nil
 
             @when_connected = []
@@ -82,6 +91,7 @@ def listen(conn)
                 on.subscribe do |chan, count|
                   @subscription_lock.synchronize do
                     if count == 1
+                      @reconnect_attempt = 0
                       @subscribed_client = original_client
 
                       until @when_connected.empty?
@@ -148,8 +158,16 @@ def ensure_listener_running
               @thread ||= Thread.new do
                 Thread.current.abort_on_exception = true
 
-                conn = @adapter.redis_connection_for_subscriptions
-                listen conn
+                begin
+                  conn = @adapter.redis_connection_for_subscriptions
+                  listen conn
+                rescue ConnectionError
+                  reset
+                  if retry_connecting?
+                    when_connected { resubscribe }
+                    retry
+                  end
+                end
               end
             end
 
@@ -161,7 +179,36 @@ def when_connected(&block)
               end
             end
 
+            def retry_connecting?
+              @reconnect_attempt += 1
+
+              return false if @reconnect_attempt > @reconnect_attempts.size
+
+              sleep_t = @reconnect_attempts[@reconnect_attempt - 1]
+
+              sleep(sleep_t) if sleep_t > 0
+
+              true
+            end
+
+            def resubscribe
+              channels = @sync.synchronize do
+                @subscribers.keys
+              end
+              @subscribed_client.subscribe(*channels) unless channels.empty?
+            end
+
+            def reset
+              @subscription_lock.synchronize do
+                @subscribed_client = nil
+                @subscribe_callbacks.clear
+                @when_connected.clear
+              end
+            end
+
             if ::Redis::VERSION < "5"
+              ConnectionError = ::Redis::ConnectionError
+
               class SubscribedClient
                 def initialize(raw_client)
                   @raw_client = raw_client
@@ -194,6 +241,8 @@ def extract_subscribed_client(conn)
                 SubscribedClient.new(conn._client)
               end
             else
+              ConnectionError = RedisClient::ConnectionError
+
               def extract_subscribed_client(conn)
                 conn
               end

actioncable/test/subscription_adapter/redis_test.rb

@@ -15,6 +15,55 @@ def cable_config
       end
     end
   end
+
+  def test_reconnections
+    subscribe_as_queue("channel") do |queue|
+      subscribe_as_queue("other channel") do |queue_2|
+        @tx_adapter.broadcast("channel", "hello world")
+
+        assert_equal "hello world", queue.pop
+
+        drop_pubsub_connections
+        wait_pubsub_connection(redis_conn, "channel")
+
+        @tx_adapter.broadcast("channel", "hallo welt")
+
+        assert_equal "hallo welt", queue.pop
+
+        drop_pubsub_connections
+        wait_pubsub_connection(redis_conn, "channel")
+        wait_pubsub_connection(redis_conn, "other channel")
+
+        @tx_adapter.broadcast("channel", "hola mundo")
+        @tx_adapter.broadcast("other channel", "other message")
+
+        assert_equal "hola mundo", queue.pop
+        assert_equal "other message", queue_2.pop
+      end
+    end
+  end
+
+  private
+    def redis_conn
+      @redis_conn ||= ::Redis.new(cable_config.except(:adapter))
+    end
+
+    def drop_pubsub_connections
+      # Emulate connection failure by dropping all connections
+      redis_conn.client("kill", "type", "pubsub")
+    end
+
+    def wait_pubsub_connection(redis_conn, channel, timeout: 2)
+      wait = timeout
+      loop do
+        break if redis_conn.pubsub("numsub", channel).last > 0
+
+        sleep 0.1
+        wait -= 0.1
+
+        raise "Timed out to subscribe to #{channel}" if wait <= 0
+      end
+    end
 end
 
 class RedisAdapterTest::AlternateConfiguration < RedisAdapterTest