基于AnyCable实现类WhatsApp消息时重连后广播失败排查

Hey there, let's break down why your forwarding_queues method runs but doesn't deliver the broadcast when users reconnect quickly, and how to fix it. This is a common gotcha with Action Cable's connection lifecycle.

Why This Happens #

The core issue is delayed cleanup of old connections:

• Action Cable relies on heartbeats to detect disconnected clients, which has a small built-in delay. When a user reconnects fast, the server might still have the old connection marked as active in its internal subscription registry.
• When you call broadcast_to current_user, Action Cable sends the message to all active subscriptions for that user—including the stale old connection that's already disconnected from the client. The new connection might not be fully registered yet when the broadcast fires, so it misses the message.
• Even if you see "THE USER WAS DISCONNECTED" in logs, the server's subscription mapping might not have updated yet to remove the old connection.

Fixes to Try #

Here are three actionable solutions, ordered by scalability and reliability:

1. Use Unique Connection IDs for Streams (Recommended)

This ensures broadcasts only go to the current active connection, regardless of stale connections hanging around.

First, add a unique connection ID to your base connection class:

module ApplicationCable
  class Connection < ActionCable::Connection::Base
    identified_by :current_user, :connection_id

    def connect
      self.current_user = find_verified_user # Keep your existing auth logic here
      self.connection_id = SecureRandom.uuid # Generate a unique ID per connection
    end
  end
end

Then modify your UserChannel to use this unique ID for streaming:

class UserChannel < ApplicationCable::Channel
  after_subscribe :forwarding_queues

  def subscribed
    # Stream to a unique channel tied to the user + their current connection
    stream_from "user_#{current_user.id}_#{connection_id}"
  end

  def forwarding_queues
    # Broadcast only to this specific connection's stream
    ActionCable.server.broadcast(
      "user_#{current_user.id}_#{connection_id}", 
      { message: 'You are back!' }
    )
  end
end

This way, even if old connections linger, your broadcast will only target the new, active connection's stream.

2. Clean Up Old Subscriptions on Reconnect

For smaller apps, you can explicitly unsubscribe a user's old connections when they subscribe again:

class UserChannel < ApplicationCable::Channel
  after_subscribe :forwarding_queues

  def subscribed
    # Iterate through all active connections and unsubscribe old ones for this user
    ActionCable.server.connections.each do |conn|
      next unless conn.current_user == current_user && conn != self.connection

      conn.subscriptions.each do |sub|
        conn.unsubscribe(sub) if sub.channel == self.class.name
      end
    end

    stream_for current_user
  end

  def forwarding_queues
    UserChannel.broadcast_to current_user, { message: 'You are back!' }
  end
end

Note: This can impact performance with many active users, since it iterates all connections. Use this only if you have a small user base.

3. Enhance Disconnect Cleanup

Add a disconnect callback to your channel to immediately stop streams when a connection drops:

class UserChannel < ApplicationCable::Channel
  after_subscribe :forwarding_queues

  def subscribed
    stream_for current_user
  end

  def disconnect
    stop_all_streams
    super
  end

  def forwarding_queues
    UserChannel.broadcast_to current_user, { message: 'You are back!' }
  end
end

This helps with clean disconnections, but won't fix issues where the client drops offline abruptly (like a network crash)—since the server still waits for a heartbeat timeout to trigger disconnect.

Bonus: Persist Undelivered Messages #

For a WhatsApp-like experience, you should store undelivered messages in your database instead of relying on in-memory queues. This ensures messages aren't lost if the server restarts, and lets you reliably fetch them on reconnection:

def forwarding_queues
  # Fetch all undelivered messages for the user
  undelivered_messages = current_user.messages.where(delivered: false)

  undelivered_messages.each do |message|
    UserChannel.broadcast_to(current_user, { message: message.content })
    message.update(delivered: true)
  end

  # Send welcome back message
  UserChannel.broadcast_to(current_user, { message: 'You are back!' })
end

This adds durability to your message system, and works even if connection cleanup is delayed.

内容的提问来源于stack exchange，提问作者Darlyn