Persistent DEVICE client

This page explains how TBMQ stores and delivers messages for persistent DEVICE clients, covering the original PostgreSQL-based design, its limitations, and the Redis-based architecture introduced in v2.0.

In TBMQ 1.x, PostgreSQL handled message persistence and retrieval for persistent DEVICE clients. While it performed well initially, PostgreSQL could only scale vertically. As the number of persistent MQTT sessions grew, its architecture became a bottleneck. Redis was chosen as the replacement due to its horizontal scalability, native clustering support, and widespread adoption.

Redis is not on the publisher’s critical path. A publish is acknowledged once Kafka has accepted it; the DEVICE copy is written to Redis afterwards, by a separate consumer:

1. The message is appended to tbmq.msg.all and the publisher is acknowledged.
2. The message dispatcher matches subscribers. For every persistent DEVICE subscriber it produces a copy to tbmq.msg.persisted (12 partitions by default), keyed by client ID — so all messages for one client stay in one partition, in order, and are always handled by the same consumer.
3. A tbmq.msg.persisted consumer (3 consumers and 3 threads by default) groups the polled batch per client ID and makes one atomic Redis call per client, which stores the whole group and assigns its MQTT packet IDs.
4. That call returns the packet ID in use before the batch, so the broker can re-derive the same IDs the script assigned and hand each message to the node where the client is connected — in process if that is this node, otherwise over tbmq.msg.downlink.persisted.$SERVICE_ID.
5. If the client is offline, step 4 is skipped entirely: the copy simply stays in Redis until the client reconnects.

Two practical consequences follow. Redis is sized by the number of pending messages rather than by throughput, and a brief Redis problem does not lose an accepted message, because Kafka still holds it and the consumer retries the write.

That retry budget is finite, though. Each pack waits TB_DEVICE_PERSISTED_MSG_PACK_PROCESSING_TIMEOUT (20 s) for its Redis calls, and TB_DEVICE_PERSISTED_MSG_ACK_STRATEGY_TYPE defaults to RETRY_ALL with TB_DEVICE_PERSISTED_MSG_ACK_STRATEGY_RETRIES = 3 and a 1 s pause between attempts. Once those are exhausted the Kafka offset is committed anyway: the copies that never reached Redis are counted as dropped in the broker’s statistics and the stream moves on, rather than stalling behind an unreachable Redis. Set the retry count to 0 for unlimited retries if you would rather have the persisted-message stream block until Redis recovers.

Two rules decide whether a copy is stored at all, and both must pass: the publisher must have used QoS 1 or 2, and the subscription must have been made with QoS 1 or 2. A QoS 0 publish is never persisted for anyone, and a QoS 0 subscription of a persistent client receives messages only while it is connected.

In TBMQ 1.x, persistent DEVICE clients used PostgreSQL for message persistence. Two tables were central to this design.

device_session_ctx — maintained session state per persistent MQTT client:

                      Table "public.device_session_ctx"
       Column       |          Type          | Nullable
--------------------+------------------------+---------
 client_id          | character varying(255) | not null
 last_updated_time  | bigint                 | not null
 last_serial_number | bigint                 |
 last_packet_id     | integer                |
Indexes:
    "device_session_ctx_pkey" PRIMARY KEY, btree (client_id)

• last_packet_id — packet ID of the last MQTT message received.
• last_serial_number — ever-increasing counter; prevents ordering issues when the MQTT packet ID wraps around at 65535.

device_publish_msg — stored messages pending delivery to persistent subscribers:

                         Table "public.device_publish_msg"
          Column          |          Type          | Nullable
--------------------------+------------------------+---------
 client_id                | character varying(255) | not null
 serial_number            | bigint                 | not null
 topic                    | character varying      | not null
 time                     | bigint                 | not null
 packet_id                | integer                |
 packet_type              | character varying(255) |
 qos                      | integer                | not null
 payload                  | bytea                  | not null
 user_properties          | character varying      |
 retain                   | boolean                |
 msg_expiry_interval      | integer                |
 payload_format_indicator | integer                |
 content_type             | character varying(255) |
 response_topic           | character varying(255) |
 correlation_data         | bytea                  |
Indexes:
    "device_publish_msg_pkey" PRIMARY KEY, btree (client_id, serial_number)
    "idx_device_publish_msg_packet_id" btree (client_id, packet_id)

Key columns:

• time — system timestamp when the message was stored; used for periodic cleanup of expired messages.
• msg_expiry_interval — expiration time (seconds) for MQTT 5.0 messages with an expiry property. If absent, the message remains valid until removed by time or size-based cleanup.

When messages arrive for a client from the shared Kafka topic, the broker queries device_session_ctx to fetch the latest last_packet_id and last_serial_number. These are incremented and assigned to each message before insertion into device_publish_msg.

Limitation: based on the TimescaleDB blog, vanilla PostgreSQL handles up to ~300k inserts per second under ideal conditions, but this depends on hardware, schema, and workload. Vertical scaling can extend this ceiling, but a per-table hard limit is eventually reached. For an MQTT broker expecting millions of persistent sessions, this was a fundamental constraint.

Redis was chosen as the replacement due to in-memory operation (low latency), native cluster support, and horizontal scalability. The migration also enabled replacing periodic time-based cleanup with per-message TTL.

Using multiple Redis data structures per client means using multiple keys per persistent session. Redis Cluster distributes keys across hash slots for horizontal scaling, but multi-key operations must target the same slot — otherwise a cross-slot error is raised.

TBMQ uses the persistent MQTT client ID as a hash tag in key names — the client ID is enclosed in {}. Redis hashes all keys sharing the same {} content to the same slot, guaranteeing that all keys for one client reside together and multi-key operations proceed without errors.

The keys are laid out this way regardless of topology, so the same code path works for all three supported connection types — REDIS_CONNECTION_TYPE accepts standalone (the default), cluster, and sentinel. Every key can also carry a common prefix (CACHE_PREFIX, empty by default), which is applied before the hash tag, so it never breaks slot co-location.

In high-throughput environments, multiple messages can arrive simultaneously for the same client. Without atomic operations, sequential updates to multiple data structures risk race conditions and partial updates, leading to message loss or incorrect ordering.

Lua scripts execute as a single isolated unit — no other commands can run concurrently during execution. TBMQ uses a separate Lua script per operation, ensuring all keys accessed within a script reside in the same hash slot and all updates are atomic:

All five are loaded with SCRIPT LOAD at startup and afterwards invoked by SHA (EVALSHA), so the script body is not shipped with every call. If Redis has forgotten the script — after a restart, a failover, or a SCRIPT FLUSH — the NOSCRIPT error is caught, the script is reloaded once (concurrent callers share that single reload), and the call is retried; a plain EVAL with the full body is the last-resort fallback.

TBMQ uses two Redis data structures per client, across three keys:

Sorted sets (ZSETs) — maintain message order using the score as a continuously increasing counter (equivalent to serial_number in PostgreSQL). The sorted set stores references to message payload keys, not the payloads themselves, to avoid memory overhead and enable per-message TTL.

Strings — store the full message payload with a TTL (EX), providing O(1) write, read, and delete complexity without affecting the sorted set.

Example — adding messages to the sorted set:

# Message with MQTT packet ID 65534
ZADD {client_id}_messages 65534 {client_id}_messages_65534

# Message with packet ID 65535 (maximum)
ZADD {client_id}_messages 65535 {client_id}_messages_65535

# Packet ID wraps around to 1 — score continues growing to 65536
ZADD {client_id}_messages 65536 {client_id}_messages_1

The score grows monotonically even when the MQTT packet ID wraps at 65535, preserving correct message ordering.

Packet IDs are assigned inside the script, not by the caller. For each message in the batch the script increments the stored last_packet_id (wrapping from 65535 back to 1), writes that ID into the message, and names the payload key after it. The score is a separate counter, seeded from the highest score currently in the sorted set — or from last_packet_id when the set is empty — and incremented once per message. Two messages therefore never collide on a score even though their packet IDs eventually repeat. The script returns the packet ID that was in use before the batch, which is how the broker re-derives the same IDs for the copies it forwards to the node holding the connection.

Example — storing a message payload with TTL:

SET {client_id}_messages_1 "{
  \"packetType\":\"PUBLISH\",
  \"payload\":\"eyJkYXRhIjoidGJtcWlzYXdlc29tZSJ9\",
  \"time\":1736333110026,
  \"clientId\":\"client\",
  \"retained\":false,
  \"packetId\":1,
  \"topicName\":\"europe/ua/kyiv/client/0\",
  \"qos\":1
}" EX 600

The EX value is per message: it is the MQTT 5.0 Message Expiry Interval the publisher sent, and when the message carries none, the broker-wide default MQTT_PERSISTENT_SESSION_DEVICE_PERSISTED_MESSAGES_TTL (604800 s — one week). This replaces the periodic cleanup job the PostgreSQL design needed: an expired message disappears on its own, and only its now-dangling reference in the sorted set is left to prune (see Message retrieval and cleanup on reconnect).

Message payloads can also be retrieved or removed individually with O(1) complexity, without affecting the sorted set:

GET {client_id}_messages_1

DEL {client_id}_messages_1

Last packet ID — a separate string key stores the last MQTT packet ID assigned:

GET {client_id}_last_packet_id
"1"

This is needed on reconnect to determine the correct packet ID for the next message. The sorted set cannot be relied on for this because it may be empty or fully removed.

The maximum number of stored messages per persistent DEVICE client is controlled by:

# Maximum number of PUBLISH messages stored for each persisted DEVICE client
limit: "${MQTT_PERSISTENT_SESSION_DEVICE_PERSISTED_MESSAGES_LIMIT:10000}"

This limit controls memory allocation per persistent client. For example, a client might connect, register a persistent session, and then disconnect immediately. Without a limit, messages would accumulate indefinitely while waiting for a potential reconnection, causing unbounded memory usage.

The maximum configurable value is 65535 (MQTT protocol limit) — the broker validates this on startup and refuses to start with a higher value. When new messages arrive and the limit is reached, the oldest messages are trimmed from the sorted set and their payload string keys are deleted. This runs at the end of the same script that stores the batch, so a client never observes more than limit pending messages:

-- Get the number of elements to be removed
local numElementsToRemove = redis.call('ZCARD', messagesKey) - maxMessagesSize
-- Check if trimming is needed
if numElementsToRemove > 0 then
    -- Get the elements to be removed (oldest ones)
    local trimmedElements = redis.call('ZRANGE', messagesKey, 0, numElementsToRemove - 1)
    -- Iterate over the elements and remove them
    for _, key in ipairs(trimmedElements) do
        -- Remove the message from the string data structure
        redis.call('DEL', key)
        -- Remove the message reference from the sorted set
        redis.call('ZREM', messagesKey, key)
    end
end

When a DEVICE client reconnects, undelivered messages are retrieved and references to expired messages are cleaned up at the same time:

local messagesKey = KEYS[1]
local maxMessagesSize = tonumber(ARGV[1])
local elements = redis.call('ZRANGE', messagesKey, 0, -1)
local messages = {}
for _, key in ipairs(elements) do
    if redis.call('EXISTS', key) == 1 then
        local msgJson = redis.call('GET', key)
        table.insert(messages, msgJson)
    else
        -- Expired — remove the stale reference from the sorted set
        redis.call('ZREM', messagesKey, key)
    end
end
return messages

ZRANGE messagesKey 0 -1 walks the set by ascending score, so messages come back oldest first and are re-delivered in their original order. Each returned message carries the packetType it was stored with, and that field decides what the client actually receives (see below).

Nothing is removed at delivery time — a message leaves Redis only when the client has confirmed it, which is what makes redelivery after a dropped connection possible:

Rewriting the stored copy is what makes a QoS 2 handshake resumable. If the connection drops between PUBREC and PUBCOMP, the pending entry is no longer a PUBLISH: on reconnect the client receives a PUBREL for that packet ID rather than the payload again, so the exchange continues where it stopped instead of restarting — and the payload is not transmitted twice.

When a session ends for good — a clean start connect, an expired session, or an admin clearing it — a single script deletes every payload key, the sorted set, and the packet-ID counter, leaving no orphaned keys behind. See Clean persistent sessions for the operations that trigger this.

Delivery is driven by the client’s DEVICE actor, which owns the session’s in-flight state:

• DUP flag — the actor tracks the packet IDs it has already sent. A message re-sent with a packet ID that is still in flight goes out with DUP = 1.
• Expiry re-check — expiry is evaluated again just before delivery, and the remaining interval is written into the outgoing message’s properties, so a client that reconnects after an hour sees an accurate remaining TTL rather than the publisher’s original value.
• Backpressure — when Netty reports the channel non-writable (the client is reading too slowly), the actor stops delivering instead of buffering unboundedly. Once the channel is writable again it re-reads the pending set from Redis, with an exponential backoff while acknowledgements are still outstanding. Nothing is lost, because nothing was removed. See Backpressure.
• Actor lifetime — after a disconnect the actor is kept alive for ACTORS_SYSTEM_PERSISTED_DEVICE_WAIT_BEFORE_ACTOR_STOP_MINUTES (5 minutes by default), so a quick reconnect reuses the warm in-memory state instead of rebuilding it.

Delivery is flushed per message by default for persistent DEVICE clients, which favors latency; buffering is available for throughput-oriented deployments. Both are covered in Message delivery strategies. Note that APPLICATION clients default the other way.

DEVICE clients do not use Kafka consumer groups for shared subscriptions the way APPLICATION clients do. TBMQ picks exactly one target per message, at dispatch time:

• A member is connected — a round-robin strategy selects one connected member (skipping a local member whose channel is currently non-writable), and the message follows the ordinary DEVICE path, stored under that member’s client ID.
• No member is connected, at least one has a persistent session — the message is stored under the subscription’s own key (topic filter + share name) instead of any client ID. The group thus has a single shared backlog rather than one copy per member.
• No member is connected and none is persistent — the message is dropped for that group.

When a member later connects, it drains the group’s backlog only if no other DEVICE member of that subscription is already connected — otherwise new messages are being routed to that member instead. While draining, packet IDs are re-mapped into the reconnecting client’s own sequence (the mapping back to the stored IDs is kept in memory so acknowledgements still remove the right entries), the QoS is downgraded to the subscription’s QoS, and the subscription identifier is attached.

A shared subscription established with QoS 0 is skipped when draining: its messages are delivered at QoS 0, are never acknowledged, and are left for another member or for the TTL to remove.

To validate Redis scalability, TBMQ was tested with a P2P MQTT pattern — one publisher per subscriber, each pair creating its own persistent session. This stresses exactly the per-session write path Redis needs to handle.

A pre-migration prototype test established a 30k msg/sec ceiling with PostgreSQL. After migrating to Redis with the existing Jedis client (synchronous), throughput reached only 40k msg/sec — a modest improvement because Jedis processes each Redis command sequentially, blocking until completion before issuing the next.

Migrating to Lettuce, an asynchronous client built on Netty, raised throughput to 60k msg/sec by enabling parallel, non-blocking Redis operations.

Jedis: ~66k commands/sec per node at 40k msg/sec — Lua scripts trigger multiple Redis operations per message Lettuce: ~100k commands/sec per node at 60k msg/sec — increase aligns with the jump from 40k to 60k msg/sec

For the full P2P performance test methodology and results, see the 1M msg/sec P2P performance test.

For the opposite trade-off — a dedicated Kafka topic and consumer per client, sized for high-rate backend consumers — see persistent APPLICATION clients.