New subscribers receiving old messages

[tbuehlmann]

There seems to be a bug involving the Listener's last_id. It is used for memoizing the Message that was sent last so it knows what to send next. The thing is, while there's no subscriber connected and new Messages are broadcasted, the last_id number doesn't change, so the next connecting subscriber will receive all the "pending" Messages from the past.

Here's an example with a background job that broadcasts a Message every 5 seconds:

00:00 background job broadcasts message #1
00:05 background job broadcasts message #2
00:06 client connects (last_id is initialized to 2) and receives all messages with id > 2 (which is none)
00:10 background job broadcasts message #3 (client receives it, last_id is set to 3)
00:15 background job broadcasts message #4 (client receives it, last_id is set to 4)
00:16 client disconnects (last_id stays at 4, this is important)
00:20 background job broadcasts message #5
00:25 background job broadcasts message #6
00:30 background job broadcasts message #7
00:35 background job broadcasts message #8
00:40 background job broadcasts message #9
00:41 client connects (last_id is still 4) and receives all messages with id > 4 (which is 5 messages from the past)

Here's a test for this behaviour:

test "broadcasting_history" do
  @tx_adapter.broadcast("channel", "1")
  @tx_adapter.broadcast("channel", "2")

  # first client, doesn't see old messages
  subscribe_as_queue("channel") do |queue|
    assert_empty queue

    @tx_adapter.broadcast("channel", "3")
    @tx_adapter.broadcast("channel", "4")

    assert_equal "3", queue.pop
    assert_equal "4", queue.pop
  end

  @tx_adapter.broadcast("channel", "5")
  @tx_adapter.broadcast("channel", "6")
  @tx_adapter.broadcast("channel", "7")
  @tx_adapter.broadcast("channel", "8")
  @tx_adapter.broadcast("channel", "9")

  # second client, sees old messages
  subscribe_as_queue("channel") do |queue|
    assert_empty queue # raises as queue has 5 items
  end
end

I thought about instant-trimming after broadcasting, but that's not really an option as there could be multiple Listeners (like in two different Puma instances) and one Listener deleting a Message before the other Listener could send it is bad.

So we might need a last_id per channel and reset it when the last subscriber for that channel disconnects (or something else entirely I didn't think of). If there was a last_id per channel (like a hash), the SolidCable::Message.broadcastable scope would need to change as well.

[jmarsh24]

I have also faced the same issue in my applications! ✋🏼

[npezza93]

Hey there! Thanks for the detailed bug report and test case! I think #66 should fix the issue. Do you want to give it a spin?

[tbuehlmann]

This helps for when a single channel is involved but doesn't cut it when more channels are used yet. This is my testcase with an added second channel:

  test "broadcasting_history" do
    @tx_adapter.broadcast("channel", "1")
    @tx_adapter.broadcast("channel", "2")

    # first client, doesn't see old messages
    subscribe_as_queue("channel") do |queue|
      assert_empty queue

      @tx_adapter.broadcast("channel", "3")
      @tx_adapter.broadcast("channel", "4")

      # ----- this is new -----
      @tx_adapter.broadcast("other", "1")
      @tx_adapter.broadcast("other", "2")

      # another client that connects to a different channel
      subscribe_as_queue("other") do |other_queue|
        assert_empty other_queue # raises as queue has 2 items
      end
      # ----- this is new -----

      assert_equal "3", queue.pop
      assert_equal "4", queue.pop
    end

    @tx_adapter.broadcast("channel", "5")
    @tx_adapter.broadcast("channel", "6")
    @tx_adapter.broadcast("channel", "7")
    @tx_adapter.broadcast("channel", "8")
    @tx_adapter.broadcast("channel", "9")

    # second client, sees old messages
    subscribe_as_queue("channel") do |queue|
      assert_empty queue # raises as queue has 5 items
    end
  end

I don't see a clever way of handling this with just a single last_id, I think we need a last_id per channel and reset it when a channel is removed (after the last channel subscriber unsubscribes/disconnects).

I also think there's now a race condition as different threads are calling add_channel/reset_last_id! and self.last_id = message.id. last_id could be set to nil by add_channel and set to an id by self.last_id = message.id inside broadcast_messages immediately after, even if last_id should stay nil. This happened to me when I tried to fix it before.

One solution that worked for me is this PoC: tbuehlmann@f8df35f

@channels becomes a Hash and memoizes the last id like this: {"channel" => 2, "other_channel" => 4}

It could be its own @last_ids Hash while @channels stays being a Set, but well, implementation detail.

It's important to note that Message.broadcastable changed and now uses the last id per channel to figure out the right Messages to deliver. And that we need a Mutex or else the mentioned race condition will occur at some time. Enforceable by having this for example:

::SolidCable::Message.broadcastable(channels).
  each do |message|
  broadcast(message.channel, message.payload)
  sleep 1
  channels[message.channel] = message.id
end

Edit: I haven't tested how performance would suffer from such a Mutex and calculating the last id everytime a new channel is added, btw.

[npezza93]

What's your polling interval set to?

[tbuehlmann]

I didn't change the default, but can you elaborate on why that'd be relevant?