[FIXED] MQTT: rapid load-balanced (re-)CONNECT to cluster causes races #4734
Conversation
|
Actually, still failed the new test when I run it with |
|
While testing with --count=X I saw that persistent sessions were hitting a similar race condition when the session persist notification was not processed in-time; it was causing invalid seq errors. The processing of persist notification was in a separate go-routine fed from ipQueue so it could load session messages for inspection. However, the only data we need from the loaded message was the client ID (hash) which we can add to the JS ACK subject and use directly... This is what the last commit does. This PR shortens the notification processing time by cutting out a goroutine/queue, but more importantly a loadMsg call from this processing. |
| as.processJSAPIReplies, &sid, &subs); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| // We will listen for replies to session persist requests so that we can | ||
| // detect the use of a session with the same client ID anywhere in the cluster. | ||
| if err := as.createSubscription(mqttJSARepliesPrefix+"*."+mqttJSASessPersist+".*", | ||
| // `$MQTT.JSA.{js-id}.SP.{client-id-hash}.{uuid}` | ||
| if err := as.createSubscription(mqttJSARepliesPrefix+"*."+mqttJSASessPersist+".*.*", |
There was a problem hiding this comment.
So does that mean that server with this fix will not be able to co-operate with other servers (say current v2.10.3)? That is, a current server that would persist a session with a reply subject on $MQTT.JSA.<serverId>.SP.<nuid> would not be received by a server with this fix. Maybe it's ok, but I am just raising this to make sure that you thought about it.
There was a problem hiding this comment.
Right, thanks for raising it. I did consider the compatibility issue and kinda punted on it, because of the "edge condition" nature of the use-case. I considered serializing the clientID into the uuid token using a different separator, but that felt too hacky for a permanent solution to a temporary edge-case. Nothing else that I could think of would make this PR backwards-operable, i.e. broadcasting to the <2.10.(N) servers in a way that they'd understand. I could easily add another listening subscription to pick up their messages, but that'd be 1-way only.
All in all, 1/5 leave as is and require that all servers in an MQTT cluster are upgraded/downgraded at approximately the same time. (Note for others, this is not affecting the session store itself, just the ACK change notifications.)
There was a problem hiding this comment.
Having said that, I think using say, $MQTT.JSA.{js-id}.SP.{client-id-hash}_{uuid} would work just fine.
There was a problem hiding this comment.
@kozlovic you agree with ^^? (leaving as is?)
There was a problem hiding this comment.
The advantage is that you could revert some of the create subscription to keep the same number of tokens. But you would need to do more processing to extract the client ID from the last token. Up to you.
levb
changed the title
Inline persistent sess notification processing PR feedback: nit _EMPTY_ PR feedback: more robust error handling, _EMPTY_ PR feedback: error handling
levb
force-pushed
the
lev-mqtt-delsess-error
branch
from
b7ff15f
to
6626091
Compare
The tests explain the condition. TL;DR: a rapid, load-balanced sequence of connects/disconnects from the same client to a cluster was causing failures.
This PR:
@derekcollison please note a highly variable test run times O(1ms) - O(100ms), including targeting the same server. Not sure what's going on there. (
go test -v ./server --run TestMQTTClusterConnectDisconnectprints them out):Same test failing in
main: