Auto-reconnect subscribe connection on backend failure. Refs cable-cr/cable#105

README.md

@@ -127,6 +127,25 @@ Options;
 5. Commit it
 6. GO TO 1
 
+### With Docker
+
+If you'd rather not install Crystal locally, a `compose.yaml` is provided that
+brings up Redis and a Crystal container with the source bind-mounted:
+
+```sh
+docker compose run --rm app shards install
+docker compose run --rm app crystal spec
+docker compose run --rm app crystal tool format spec/ src/
+docker compose run --rm app ./bin/ameba
+docker compose down -v   # tear down + drop volumes when done
+```
+
+The Crystal version defaults to `latest`; override it to match the CI floor:
+
+```sh
+CRYSTAL_VERSION=1.10.0 docker compose run --rm app crystal spec
+```
+
 ## Contributing
 
 1. Fork it (<https://github.com/cable-cr/cable-redis/fork>)

compose.yaml

@@ -0,0 +1,35 @@
+# Development environment for cable-redis.
+#
+# Usage:
+#   docker compose run --rm app shards install
+#   docker compose run --rm app crystal spec
+#   docker compose run --rm app sh                # interactive shell
+#   docker compose exec redis redis-cli           # poke at Redis
+#   docker compose down -v                        # tear down + drop volumes
+#
+# Override the Crystal version (matches the CI matrix: 1.10.0, latest):
+#   CRYSTAL_VERSION=1.10.0 docker compose run --rm app crystal spec
+services:
+  redis:
+    image: redis:7-alpine
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 2s
+      timeout: 3s
+      retries: 10
+
+  app:
+    image: crystallang/crystal:${CRYSTAL_VERSION:-latest}-alpine
+    working_dir: /app
+    volumes:
+      - .:/app:z
+      - crystal-cache:/root/.cache/crystal
+    environment:
+      CABLE_BACKEND_URL: redis://redis:6379
+    depends_on:
+      redis:
+        condition: service_healthy
+    command: ["sh"]
+
+volumes:
+  crystal-cache:

spec/cable-redis_spec.cr

@@ -15,6 +15,87 @@ describe Cable::RedisBackend do
       socket.messages.should contain({"identifier" => {channel: "ChatChannel", room: "1"}.to_json, "message" => JSON.parse(%({"foo": "bar"}))}.to_json)
     end
   end
+
+  # Reproduces the failure mode from cable-cr/cable#105 against a real Redis:
+  # kill the subscribe-side TCP, wait for the reconnect loop to swap in a
+  # fresh connection and replay subscriptions, then verify message dispatch
+  # resumes. This test is timing-sensitive by nature — if it flakes, the
+  # `sleep` after the kill needs to grow to comfortably exceed
+  # Cable::RedisBackend::SUBSCRIBE_RECONNECT_BACKOFF + replay time.
+  describe "subscribe-connection recovery (cable-cr/cable#105)" do
+    it "resumes message dispatch after the pubsub TCP is killed" do
+      connect do |connection, socket|
+        connection.receive({"command" => "subscribe", "identifier" => {channel: "ChatChannel", room: "1"}.to_json}.to_json)
+        sleep 0.1
+
+        # Baseline: messages flow before the kill.
+        Cable.server.publish(channel: "chat_1", message: %({"foo": "before-kill"}))
+        sleep 0.2
+        socket.messages.any?(&.includes?("before-kill")).should be_true
+
+        # Kill all pubsub TCP connections from the Redis side. This is the same
+        # signal as `redis-cli CLIENT KILL TYPE pubsub` and matches the
+        # production reproduction in the issue.
+        backend = Cable.server.backend.as(Cable::RedisBackend)
+        killed = backend.publish_connection.run({"CLIENT", "KILL", "TYPE", "pubsub"})
+        killed.to_s.to_i.should be > 0
+
+        # Give the reconnect loop time to: notice the death, wait out the
+        # backoff, build a fresh connection, and replay subscriptions onto it.
+        sleep(Cable::RedisBackend::SUBSCRIBE_RECONNECT_BACKOFF + 1.5.seconds)
+
+        # If recovery worked, a fresh publish flows through the new connection.
+        Cable.server.publish(channel: "chat_1", message: %({"foo": "after-kill"}))
+        sleep 0.5
+
+        socket.messages.any?(&.includes?("after-kill")).should be_true
+      end
+    end
+
+    # Exercises the harder failure mode: not just a clean socket drop, but the
+    # entire backend being unreachable across multiple reconnect cycles. The
+    # original PR only wrapped the pubsub block in a rescue; the
+    # `Redis::Connection.new(...)` call on the reopen path was unprotected, so
+    # the first DNS/connect failure during an outage crashed the subscribe
+    # fiber and message dispatch never recovered, even after Redis came back.
+    it "survives reopen failures and recovers when the backend comes back" do
+      connect do |connection, socket|
+        connection.receive({"command" => "subscribe", "identifier" => {channel: "ChatChannel", room: "1"}.to_json}.to_json)
+        sleep 100.milliseconds
+
+        Cable.server.publish(channel: "chat_1", message: %({"foo": "before-outage"}))
+        sleep 200.milliseconds
+        socket.messages.any?(&.includes?("before-outage")).should be_true
+
+        real_url = Cable.settings.url
+
+        # Point the reconnect target at a port nothing's listening on, then
+        # kill the live pubsub socket. The next reopen will hit
+        # Socket::ConnectError — the regression this spec guards.
+        Cable.settings.url = "redis://127.0.0.1:1"
+
+        backend = Cable.server.backend.as(Cable::RedisBackend)
+        killed = backend.publish_connection.run({"CLIENT", "KILL", "TYPE", "pubsub"})
+        killed.to_s.to_i.should be > 0
+
+        # Sit in the failing-reopen state for a few backoff cycles. Before the
+        # fix, the fiber would die on the first iteration here.
+        sleep(Cable::RedisBackend::SUBSCRIBE_RECONNECT_BACKOFF * 3 + 500.milliseconds)
+
+        # Bring the backend back by restoring the URL. The next iteration's
+        # reopen should succeed and replay_tracked_subscriptions should
+        # re-register chat_1.
+        Cable.settings.url = real_url
+        sleep(Cable::RedisBackend::SUBSCRIBE_RECONNECT_BACKOFF + 1.5.seconds)
+
+        Cable.server.publish(channel: "chat_1", message: %({"foo": "after-recovery"}))
+        sleep 500.milliseconds
+        socket.messages.any?(&.includes?("after-recovery")).should be_true
+      ensure
+        Cable.settings.url = real_url if real_url
+      end
+    end
+  end
 end
 
 private class ChatChannel < Cable::Channel

src/cable-redis.cr

@@ -1,16 +1,33 @@
 require "redis"
+require "socket"
 
 module Cable
   class RedisBackend < Cable::BackendCore
     VERSION = "0.1.0"
 
+    # How long to wait before reconnecting the subscribe connection after
+    # the pubsub block loop crashes (see cable-cr/cable#105).
+    SUBSCRIBE_RECONNECT_BACKOFF = 1.second
+
     register "redis"  # redis://
     register "rediss" # rediss://
 
     # connection management
     getter redis_subscribe : Redis::Connection = Redis::Connection.new(URI.parse(Cable.settings.url))
     getter redis_publish : Redis::Client = Redis::Client.new(URI.parse(Cable.settings.url))
 
+    # Tracks every stream identifier we have an active SUBSCRIBE on, so that
+    # if the underlying TCP dies we can replay the subscriptions onto a fresh
+    # connection (see cable-cr/cable#105). The internal control channel passed
+    # to open_subscribe_connection is excluded — it's re-subscribed implicitly
+    # when the reconnect loop re-enters the pubsub block.
+    @subscribed_channels = Set(String).new
+    @subscribed_channels_mutex = Mutex.new
+
+    # Set when the backend is being torn down. The reconnect loop checks this
+    # so a clean shutdown does not get interpreted as a transient failure.
+    @shutting_down : Bool = false
+
     # connection management
     def subscribe_connection : Redis::Connection
       redis_subscribe
@@ -21,54 +38,96 @@ module Cable
     end
 
     def close_subscribe_connection
-      return if redis_subscribe.nil?
-
+      @shutting_down = true
       redis_subscribe.unsubscribe
       redis_subscribe.close
+    rescue IO::Error
+      # connection was already torn down
     end
 
     def close_publish_connection
-      return if redis_publish.nil?
-
       redis_publish.close
+    rescue IO::Error
+      # connection was already torn down
     end
 
     # internal pub/sub
+    #
+    # The pubsub block loop is wrapped in a reconnect cycle so that a transient
+    # backend death (Redis restart, network blip, idle reap) does not permanently
+    # kill message dispatch — see cable-cr/cable#105.
     def open_subscribe_connection(channel)
-      return if redis_subscribe.nil?
-
-      redis_subscribe.subscribe(channel) do |subscription|
-        subscription.on_message do |sub_channel, message|
-          if sub_channel == Cable::INTERNAL[:channel] && message == "ping"
-            Cable::Logger.debug { "Cable::Server#subscribe -> PONG" }
-          elsif sub_channel == Cable::INTERNAL[:channel] && message == "debug"
-            Cable.server.debug
-          else
-            Cable.server.fiber_channel.send({sub_channel, message})
-            Cable::Logger.debug { "Cable::Server#subscribe channel:#{sub_channel} message:#{message}" }
+      loop do
+        begin
+          # Replay any tracked subscriptions onto the (possibly fresh) connection
+          # once the pubsub block has actually entered subscribe mode. The spawned
+          # fiber runs while the main fiber is blocked waiting for messages.
+          spawn(name: "Cable::RedisBackend - replay subscriptions") { replay_tracked_subscriptions }
+
+          redis_subscribe.subscribe(channel) do |subscription|
+            subscription.on_message do |sub_channel, message|
+              if sub_channel == Cable::INTERNAL[:channel] && message == "ping"
+                Cable::Logger.debug { "Cable::Server#subscribe -> PONG" }
+              elsif sub_channel == Cable::INTERNAL[:channel] && message == "debug"
+                Cable.server.debug
+              else
+                Cable.server.fiber_channel.send({sub_channel, message})
+                Cable::Logger.debug { "Cable::Server#subscribe channel:#{sub_channel} message:#{message}" }
+              end
+            end
           end
+          # Falling through here means the subscribe block returned without
+          # raising. jgaskins/redis exits its read loop cleanly when `read?`
+          # returns nil — which is what `CLIENT KILL TYPE pubsub` and other
+          # server-side disconnects look like — so we must treat a clean
+          # return as a reconnect signal, not as success.
+        rescue e : IO::Error | Socket::Error
+          Cable::Logger.error(exception: e) { "Cable::RedisBackend subscribe loop crashed" }
+          Cable.settings.on_error.call(e, "Cable::RedisBackend#open_subscribe_connection (reconnecting)", nil)
+        end
+
+        break if @shutting_down
+        Cable::Logger.warn { "Cable::RedisBackend subscribe disconnected; reconnecting in #{SUBSCRIBE_RECONNECT_BACKOFF.total_seconds}s" }
+        sleep SUBSCRIBE_RECONNECT_BACKOFF
+        break if @shutting_down
+
+        begin
+          @redis_subscribe = Redis::Connection.new(URI.parse(Cable.settings.url))
+        rescue e : IO::Error | Socket::Error
+          # Backend is still unreachable (DNS unresolvable, TCP refused,
+          # network down, etc.). Leave the dead @redis_subscribe in place;
+          # the next loop iteration's `.subscribe` call will raise IO::Error
+          # immediately, get caught by the outer rescue, and we'll back off
+          # and retry the open here. on_error is intentionally NOT invoked
+          # for reopen failures — during a multi-second outage this loop
+          # ticks once per second and would flood error trackers.
+          Cable::Logger.warn(exception: e) { "Cable::RedisBackend reopen failed; will retry after backoff" }
         end
       end
     end
 
     # external pub/sub
     def publish_message(stream_identifier : String, message : String)
-      return if redis_subscribe.nil?
-
       redis_publish.publish(stream_identifier, message)
     end
 
     # channel management
     def subscribe(stream_identifier : String)
-      return if redis_subscribe.nil?
-
+      @subscribed_channels_mutex.synchronize { @subscribed_channels << stream_identifier }
       redis_subscribe.subscribe(stream_identifier)
+    rescue e : IO::Error
+      # The connection is dead — the recovery loop in open_subscribe_connection
+      # will replay this subscription when it reconnects.
+      Cable::Logger.error(exception: e) { "Cable::RedisBackend subscribe(#{stream_identifier}) failed; will be replayed after reconnect" }
     end
 
     def unsubscribe(stream_identifier : String)
-      return if redis_subscribe.nil?
-
+      @subscribed_channels_mutex.synchronize { @subscribed_channels.delete(stream_identifier) }
       redis_subscribe.unsubscribe(stream_identifier)
+    rescue e : IO::Error
+      # Connection died before we could send UNSUBSCRIBE — the channel has been
+      # removed from the replay set so the next reconnect will not resubscribe.
+      Cable::Logger.error(exception: e) { "Cable::RedisBackend unsubscribe(#{stream_identifier}) failed" }
     end
 
     # ping/pong
@@ -87,5 +146,17 @@ module Cable
       result = redis_publish.run({"ping"})
       Cable::Logger.debug { "Cable::BackendPinger.ping_publish_connection -> #{result}" }
     end
+
+    private def replay_tracked_subscriptions
+      channels_to_replay = @subscribed_channels_mutex.synchronize { @subscribed_channels.dup }
+      channels_to_replay.each do |id|
+        begin
+          redis_subscribe.subscribe(id)
+        rescue e : IO::Error
+          # The next reconnect iteration will retry from the tracked set.
+          Cable::Logger.error(exception: e) { "Cable::RedisBackend channel replay failed for #{id}" }
+        end
+      end
+    end
   end
 end