MQTT rx_msg and pending_publishes buffers grow unbounded

[robjarawan]

what

The MQTT transport layer (moth/mqtt.py) has no size limit on its internal message buffers. If the consumer stalls or falls behind, these buffers grow without bound, eventually exhausting memory.

There are two separate buffer systems affected:

1. rx_msg rotating lists (receive path)

5 rotating lists (rx_msg[0] through rx_msg[4]) are used as a double-buffering scheme to reduce lock contention between the MQTT callback thread and the application thread. The on_message callback appends to whichever list rx_msg_iFromBroker points at (line 556). The application drains from rx_msg_iToApp in newMessages(). If the application is slower than the broker, the lists grow without limit.

2. pending_publishes deque (publish path)

pending_publishes is a collections.deque() (line 492) that tracks message IDs waiting for publish acknowledgement. On disconnect, any remaining entries are logged as lost (line 209-213). But during normal operation, if the broker is slow to ack, this deque grows unbounded.

why this matters

These are 24/7 daemon processes. Even a brief consumer stall (slow disk, NFS hang, large batch processing) can cause a spike that does not get cleaned up until the messages are eventually consumed. Under sustained load imbalance, this is a memory leak.

possible approaches

Would like the team's input on which direction makes sense:

Option A -- maxlen cap with drop-oldest. Set a maximum size on each rx_msg list (e.g. via a config option like rxBufferMax). When the buffer is full, drop the oldest messages. Simple to implement, but dropping messages silently is risky for data integrity -- the sender thinks they were delivered.

Option B -- maxlen cap with NACK/backpressure. When the buffer hits a cap, stop calling loop() or use MQTT flow control to apply backpressure to the broker. Safest for data integrity but more complex to implement and may need different handling for MQTTv5 vs v3.1.1.

Option C -- bounded deque with warning. Use collections.deque(maxlen=N) so old entries are silently dropped, but log a warning when the buffer is consistently full. A compromise between simplicity and visibility. Still loses messages though.

Option D -- monitor and alert only. Do not cap the buffers, but add metrics tracking (buffer high-water mark, current size) reported during housekeeping. Operators can then tune prefetch, batch, or instance counts to keep up. No data loss risk, but does not actually prevent OOM.

For pending_publishes, Option A or C is likely fine since the messages were already sent -- we are just waiting for acks. For rx_msg, data integrity matters more so Option B or D might be more appropriate.

Thoughts?

[reidsunderland]

I'm not familiar with the MQTT code.

Publishing

In general, I don't think we want to do any kind of limiting or message dropping. If a publish is not acknowledged, we can't be confident that it was actually published and we don't want ambiguity. If a publish is not acked by the MQTT broker then imo we should assume the publish failed and try again later.

In the AMQP implementation, publishing is a blocking operation and we publish one message at a time. If it works, we can move on and publish the next message, if it fails, the message is placed in the post-retry diskqueue to be retried later.

Taking a quick look at the Paho docs, I see there's a wait_for_publish method that would enable us to do a blocking publish. Maybe this would be a safer option?

Subscribing

"if the application is slower than the broker" -- in AMQP, this is controlled by the prefetch setting. AMQP brokers will deliver <= prefetch messages to the client, then stop delivering until those messages are acknowledged. The idea is that sr3 should not receive or accept more messages than it can handle, the queue should build up on the broker instead, potentially allowing another node or instance to receive those messages.

The MQTT code has max_inflight_messages, which seems like the same idea. I think this is intended to prevent those lists from growing without limit, but I don't know if it actually works (the comments and default value of 20000 make me doubtful 😄 )

[petersilva]

@reidsunderland talked about transmission, I feel the same way about reception. I think this a wrong track. Normally, one would just put the data into messages immediately. The reason for the double buffering in the first place is that if we did not do that, there would be message loss. This is a contrast in the history of MQTT vs. AMQP applications... AMQP, coming out of sending financial data around always treated every message as precious and message loss was a serious problem.

In contrast, MQTT came out of the world of sensors. If you lose one message, you'll just get the next sensor reading a few seconds later. In classic MQTT message flow is critical and message loss is not a big deal. That philosophy is why MQTT 3.x is not the greatest fit for Sarracenia. In this application, losing messages is a big problem, since we don't know if a replacement will ever be produced, so it results in meaningful data loss.

MQTTv5 switches to TCP for message transport, and the recovery mechanisms are more oriented towards treating individual messages as valuable. The problem the AI is pointing out here is a theoretical one of memory exhaustion, and it proposes to address it by throwing out messages. That's just not appropriate for this application. We'd actually rather exhaust memory and crash in order to avoid losing messages for as long as possible.

[petersilva]

Option A -- nope ... dropping messages inappropriate.

Option B -- already implemented by max_inflight_messages.

Option C -- nope message loss is bad.

Option D -- We have monitoring of RSS... and memory size... we do have some monitoring that will see if this problem gets out of hand. We even have restart logic that addresses the memory leak in the worst case.