{"record":{"id":"59aa3418e32dd765","repo":"nats-io/nats-server","slug":"qos1-in-flight-window-is-full-d-messages-and-je","errorCode":null,"errorMessage":"QoS1 in-flight window is full (%d messages) and JetStream has not acknowledged the oldest message within %v; failing the connection, the client will re-send unacknowledged PUBLISH packets on reconnect","messagePattern":"QoS1 in-flight window is full \\((.+?) messages\\) and JetStream has not acknowledged the oldest message within (.+?); failing the connection, the client will re-send unacknowledged PUBLISH packets on reconnect","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"server/mqtt.go","lineNumber":4727,"sourceCode":"\t\treturn errMQTTAckPipelineStopped\n\tdefault:\n\t}\n\n\tenqueued := false\n\tselect {\n\tcase pipe.q <- ack:\n\t\tenqueued = true\n\tdefault:\n\t\t// Window full: wait for an available slot in the pipeline.\n\t\tt := time.NewTimer(jsa.timeout)\n\t\tdefer t.Stop()\n\t\tselect {\n\t\tcase pipe.q <- ack:\n\t\t\tenqueued = true\n\t\tcase <-pipe.quitCh:\n\t\tcase <-t.C:\n\t\t\tjsa.replies.Delete(ack.reply)\n\t\t\treturn fmt.Errorf(\"QoS1 in-flight window is full (%d messages) and JetStream has not acknowledged \"+\n\t\t\t\t\"the oldest message within %v; failing the connection, \"+\n\t\t\t\t\"the client will re-send unacknowledged PUBLISH packets on reconnect\",\n\t\t\t\tmqttMaxAcksInFlight, jsa.timeout)\n\t\t}\n\t}\n\tif !enqueued {\n\t\t// Stopped while waiting for a slot, never admitted.\n\t\tjsa.replies.Delete(ack.reply)\n\t\treturn errMQTTAckPipelineStopped\n\t}\n\n\t// A stop may have raced the enqueue: the close-time drain could run\n\t// before the entry landed and miss it. Drain again; both drains only\n\t// dequeue and delete, so overlapping is harmless.\n\tselect {\n\tcase <-pipe.quitCh:\n\t\tpipe.shutdown()\n\t\treturn errMQTTAckPipelineStopped","sourceCodeStart":4709,"sourceCodeEnd":4745,"githubUrl":"https://github.com/nats-io/nats-server/blob/3a66a489d262bf89b71a71c955c94920394532f3/server/mqtt.go#L4709-L4745","documentation":"The MQTT JetStream bridge bounds the number of un-acked QoS1 messages in flight (mqttMaxAcksInFlight). When the window is full, new acks must be enqueued via a bounded select; if the pipe stays full past jsa.timeout, the server fails the connection with this error so the client resends unacknowledged PUBLISH packets on reconnect.","triggerScenarios":"An MQTT client publishes QoS1 messages faster than JetStream acks them, filling mqttMaxAcksInFlight slots; the pending ack cannot be enqueued within jsa.timeout because the pipe is full and not draining.","commonSituations":"JetStream backing store is slow or quorum-degraded, stream limits/maximums block commits, a single aggressive MQTT publisher flooding QoS1 messages, or network partition slowing JS acks.","solutions":["Throttle the MQTT publisher's QoS1 rate (client-side in-flight window, e.g. max_inflight in the client) so it doesn't outpace JS acks","Fix JetStream performance/health: check stream info, raft/quorum state, disk latency, and increase resources","Increase jsa.timeout (server mqtt js ack timeout config) if latency is legitimately high but bounded","Verify stream limits (max msgs, max bytes, retention) are not causing ack stalls"],"exampleFix":"// before (client)\nopts.MaxReconnect = 5\n// after (client-side QoS1 throttling)\nopts := mqtt.NewClientOptions().SetWriteTimeout(10*time.Second)\nmqttClient.SetMaxInflight(64) // stay below server mqttMaxAcksInFlight","handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"if err := publish(qos1payload); err != nil && strings.Contains(err.Error(), \"in-flight window is full\") {\n  time.Sleep(backoff)\n  resumeFromLastUnacked()\n}","preventionTips":["Throttle client QoS1 publish rate below JetStream's sustained ack throughput","Keep client in-flight window below the server's mqttMaxAcksInFlight","Alert on JetStream ack latency and stream limit near-capacity","Fix storage/quorum problems promptly — backpressure here indicates JS slowness"],"tags":["mqtt","jetstream","backpressure","inflight-window","qos1"],"backgroundTag":"jetstream-ack-timeout","analyzedSha":"3a66a489d262bf89b71a71c955c94920394532f3","analyzedAt":"2026-09-02T04:41:54.247Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T15:18:49.778Z"}