{"record":{"id":"2bf5946b2dc6530b","repo":"nats-io/nats-server","slug":"invalid-mqtt-session","errorCode":null,"errorMessage":"invalid MQTT session","messagePattern":"invalid MQTT session","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"server/mqtt.go","lineNumber":244,"sourceCode":"\terrMQTTTopicFilterCannotBeEmpty   = errors.New(\"topic filter cannot be empty\")\n\terrMQTTMalformedVarInt            = errors.New(\"malformed variable int\")\n\terrMQTTSecondConnectPacket        = errors.New(\"received a second CONNECT packet\")\n\terrMQTTServerNameMustBeSet        = errors.New(\"mqtt requires server name to be explicitly set\")\n\terrMQTTUserMixWithUsersNKeys      = errors.New(\"mqtt authentication username not compatible with presence of users/nkeys\")\n\terrMQTTTokenMixWIthUsersNKeys     = errors.New(\"mqtt authentication token not compatible with presence of users/nkeys\")\n\terrMQTTAckWaitMustBePositive      = errors.New(\"ack wait must be a positive value\")\n\terrMQTTJSAPITimeoutMustBePositive = errors.New(\"JS API timeout must be a positive value\")\n\terrMQTTStandaloneNeedsJetStream   = errors.New(\"mqtt requires JetStream to be enabled if running in standalone mode\")\n\terrMQTTConnFlagReserved           = errors.New(\"connect flags reserved bit not set to 0\")\n\terrMQTTWillAndRetainFlag          = errors.New(\"if Will flag is set to 0, Will Retain flag must be 0 too\")\n\terrMQTTPasswordFlagAndNoUser      = errors.New(\"password flag set but username flag is not\")\n\terrMQTTCIDEmptyNeedsCleanFlag     = errors.New(\"when client ID is empty, clean session flag must be set to 1\")\n\terrMQTTEmptyWillTopic             = errors.New(\"empty Will topic not allowed\")\n\terrMQTTEmptyUsername              = errors.New(\"empty user name not allowed\")\n\terrMQTTTopicIsEmpty               = errors.New(\"topic cannot be empty\")\n\terrMQTTPacketIdentifierIsZero     = errors.New(\"packet identifier cannot be 0\")\n\terrMQTTUnsupportedCharacters      = errors.New(\"character not supported for MQTT topics\")\n\terrMQTTInvalidSession             = errors.New(\"invalid MQTT session\")\n\terrMQTTInvalidRetainFlags         = errors.New(\"invalid retained message flags\")\n\terrMQTTInvalidRetainedMessage     = errors.New(\"invalid retained message\")\n\terrMQTTSessionCollision           = errors.New(\"stored session does not match client ID\")\n\terrMQTTInvalidPublishLength       = errors.New(\"invalid publish message, variable header exceeds remaining length\")\n\terrMQTTAckPipelineStopped         = errors.New(\"QoS1 PUBACK pipeline has shut down while admitting a message, \" +\n\t\t\"abandoning the wait for its JetStream ack; failing the connection, \" +\n\t\t\"the client will re-send unacknowledged PUBLISH packets on reconnect\")\n)\n\ntype srvMQTT struct {\n\tlistener     net.Listener\n\tlistenerErr  error\n\tauthOverride bool\n\tsessmgr      mqttSessionManager\n}\n\ntype mqttSessionManager struct {\n\tmu       sync.RWMutex","sourceCodeStart":226,"sourceCodeEnd":262,"githubUrl":"https://github.com/nats-io/nats-server/blob/3a66a489d262bf89b71a71c955c94920394532f3/server/mqtt.go#L226-L262","documentation":"errMQTTInvalidSession is returned when an MQTT connection's session state is missing or does not match the connection performing the operation. At server/mqtt.go:5185 it fires if c.mqtt.sess is nil; at server/mqtt.go:5193 it fires if the locked session's owning connection sess.c is not the current connection. It guards QoS ack-processing paths (e.g. PUBREC handling) that require a valid, bound session.","triggerScenarios":"Processing a QoS ack (PUBREC/PUBACK path) after the connection's session has been torn down (sess == nil), or when the session object has been rebound to a different connection (sess.c != c), e.g. during session takeover by a new client with the same client ID.","commonSituations":"Client reconnects with the same client ID while old-connection acks are still in flight (session takeover); server shutting down or closing the session concurrently with ack processing; protocol races where the session is cleared before a queued ack is handled.","solutions":["Use a unique client ID per connection (or rely on the server's takeover semantics) so an old connection does not process acks against a session it no longer owns.","Ensure the client completes its QoS handshake on one connection before reconnecting; resend unacknowledged PUBLISH packets on the new connection instead.","If you hit this in server code, check that the session is locked/valid (sess != nil && sess.c == c) before ack processing, and bail out cleanly as the code already does."],"exampleFix":"// client: reuse of same client ID from two processes\nclientId := \"device-1\" // both instances use it\n// after: unique per instance\nclientId := fmt.Sprintf(\"device-1-%s\", instanceUUID)","handlingStrategy":"try-catch","validationCode":"func sessionValid(c *clientConn) bool {\n    if c.mqtt == nil || c.mqtt.sess == nil {\n        return false\n    }\n    return c.mqtt.sess.c == c\n}","typeGuard":"func (c *clientConn) hasValidSession() bool {\n    return c.mqtt != nil && c.mqtt.sess != nil && c.mqtt.sess.c == c\n}","tryCatchPattern":"if err := processAck(packet); err != nil {\n    if strings.Contains(err.Error(), \"invalid MQTT session\") {\n        // session taken over or closed; drop in-flight acks and reconnect\n        log.Info(\"session no longer valid; resending unacked publishes after reconnect\")\n        reconnectAndResend()\n        return nil\n    }\n    return err\n}","preventionTips":["Use unique client IDs per connection/process to avoid session takeover races","Drain pending QoS acks before closing or reconnecting a client","On reconnect, resend unacknowledged QoS>0 PUBLISH packets rather than assuming old session state"],"tags":["mqtt","session","connection-state"],"backgroundTag":"invalid-session-state","analyzedSha":"3a66a489d262bf89b71a71c955c94920394532f3","analyzedAt":"2026-09-02T04:41:54.247Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}