fix(pubsub): Skip stale pubsub messages

[dranikpg]

Fixes #5139

Dragonlfy had the following issue:

• After any kind of subscribe is issued, the client is in pubsub mode. No commannds, except pubsub ones, can be executed
• To exit the pubsub mode, the client has to unsubscribe from all channels. In that case the server replies with 0 as the number of pending subscriptions
• Many clients, like io-redis, rely on this condition to exit pubsub mode
• The reply is a regular resp3 push message over the same connection. The client can be stuck in pubsub for some time if it has a large socket buffer, large pile of previous messages buffered somewhere, etc... It only matters it's delivered eventually and delivered last
• Dragonfly checks subscriber validity only at the point of fetching them from the channel store. After that, it waits on backpressure and dispatches to the clients native threads. During this time, they might have already unsubscribed. So for Dragonfly, reading a successful reply from "unsubscribe" doesn't guarantee no more messages will be delivered.
• This PR fixes the "unsubscribe all" case. In general, we have the same problem when unsubscribing only from a specific channel.

[kostasrim]

We can apply the similar filter at the point where we publish right ?

[dranikpg]

Not sure I understand 🙂 At the time we publish we just fetch all subscribers and they're all valid at that instant. The problem is that in-between determining candidates for publishing and actually sending the message on a client's thread the client performs operations (like unsubbing)

[kostasrim]

Yes I see. What I had in mind is to detect the presence of pending unsubscribe messages in the dispatch queue at the point we call SendPubMessageAsync. Thinking about it now it's unnecessarily complicated as the connection might be subscribed to multiple channels or the dispatch q might contain sequences of subscribe/unsubscribe messages making it difficult to detect if what we about to push to the dispatch queue is considered stale