fix: guarantee QoS 1 buffered messages deliver before live messages on reconnect - #4

Summary

• Fixes a race condition where live QoS 1 messages could arrive at a reconnecting client before buffered (offline) messages stored in persistence
• Adds aedes-persistence-redis v11.2.2 as a dependency, which now supports Redis Cluster via opts.cluster (ioredis)

Root Cause

The connectActions pipeline runs in this order:

authenticate → fetchSubs → restoreSubs → registerClient → doConnack → emptyQueue

restoreSubs registers mqemitter listeners — live messages start flowing immediately. emptyQueue drains the persistence store only after that, so buffered messages arrived after live ones.

Fix

Three changes to lib/client.js and lib/handlers/connect.js:

1. client._drainingQueue flag (constructor) — false by default
2. deliverQoS gate — when _drainingQueue is true, incoming live QoS packets are pushed to client._liveQueue instead of written immediately
3. emptyQueue — before/after drain:
   • Sets _drainingQueue = true before streaming from persistence
   • afterDrain callback clears the flag and flushes _liveQueue in order
4. client.close() — drains _liveQueue callbacks so mqemitter never stalls if the client closes mid-drain

Message flow after fix:

restoreSubs → live message arrives → deliverQoS → _drainingQueue=true → pushed to _liveQueue (held)
emptyQueue drains persistence → writes buffered packets to client  ← buffered first ✓
afterDrain → _drainingQueue=false → flushes _liveQueue             ← live after ✓

aedes-persistence-redis

Previous comment in the server blocked this with "does not support connecting to the clustered redis". v11.2.2 resolves this — it accepts opts.cluster and internally uses ioredis Cluster mode.

Test plan

• Subscriber connects with cleanSession=false, QoS 1 subscription
• Subscriber disconnects; server publishes N messages to its topic
• Subscriber reconnects — verify all N buffered messages arrive before any live messages
• Subscriber disconnects mid-drain — verify no mqemitter stall / process hang
• clean=true session — verify unaffected (_drainingQueue is set and cleared with empty stream)

🤖 Generated with Claude Code