Restart Action Cable when Redis reconnect_attempt limit is reached

actioncable/CHANGELOG.md

@@ -1,3 +1,7 @@
+*   Restart Action Cable when Redis reconnect_attempt limit is reached.
+
+    *Heinrich Lee Yu, Elijah Buck*
+
 *   Record ping on every Action Cable message.
 
     Previously only `ping` and `welcome` message types were keeping the connection active.

actioncable/lib/action_cable/subscription_adapter/redis.rb

@@ -169,6 +169,10 @@ def ensure_listener_running
                   if retry_connecting?
                     when_connected { resubscribe }
                     retry
+                  else
+                    @adapter.server.logger.info("Redis reconnect_attempts limit exceeded; restarting.")
+                    @thread = nil
+                    @adapter.server.restart
                   end
                 end
               end

actioncable/test/subscription_adapter/redis_test.rb

@@ -150,3 +150,44 @@ def connection_id
     end
   end
 end
+
+class RedisAdapterTest::ConnectionError < RedisAdapterTest
+  require "redis"
+  class FailingRedis < ::Redis
+    cattr_accessor :should_raise
+    def subscribe(*channels, &block)
+      if FailingRedis.should_raise
+        FailingRedis.should_raise = false
+        raise RedisClient::ConnectionError.new
+      end
+      super
+    end
+  end
+
+  def test_connection_restart_on_connection_error
+    ActionCable::SubscriptionAdapter::Redis.redis_connector = ->(config) do
+      FailingRedis.new(config.except(:adapter, :channel_prefix))
+    end
+    server = ActionCable::Server::Base.new
+    @adapter = server.config.pubsub_adapter.new(server)
+    subscribe_as_queue("channel", @adapter) do |queue|
+      thread_id = @adapter.send(:listener).instance_variable_get("@thread").object_id
+
+      # the first reconnect will fail, triggering the restart
+      FailingRedis.should_raise = true
+      drop_pubsub_connections
+
+      # wait for server to restart
+      10.times do
+        new_thread_id = @adapter.send(:listener).instance_variable_get("@thread")&.object_id
+        break if new_thread_id && new_thread_id != thread_id
+        sleep 0.1
+      end
+
+      subscribe_as_queue("newchannel", @adapter) do |queue|
+        @tx_adapter.broadcast("newchannel", "hello world")
+        assert_equal "hello world", queue.pop
+      end
+    end
+  end
+end