simplexmq/plans/20260328_01_server_batched_sub_processing.md

Server: batched SUB command processing

Implementation plan for Part 1 of RFC 2026-03-28-subscription-performance.

Current state

When a batch of ~135 SUB commands arrives, the server already batches:

• Queue record lookups (getQueueRecs in receive, Server.hs:1151)
• Command verification (verifyLoadedQueue, Server.hs:1152)

But command processing is per-command (foldrM process in client, Server.hs:1372-1375). Each SUB calls subscribeQueueAndDeliver which calls tryPeekMsg - one DB query per queue. For Postgres, that's ~135 individual SELECT ... FROM messages WHERE recipient_id = ? ORDER BY message_id ASC LIMIT 1 queries per batch.

Goal

Replace ~135 individual message peek queries with 1 batched query per batch. No protocol changes.

Implementation

Step 1: Add tryPeekMsgs to MsgStoreClass

File: src/Simplex/Messaging/Server/MsgStore/Types.hs

Add to MsgStoreClass:

tryPeekMsgs :: s -> [StoreQueue s] -> ExceptT ErrorType IO (Map RecipientId Message)

Returns a map from recipient ID to earliest pending message for each queue that has one. Queues with no messages are absent from the map.

Step 2: Parameterize deliver to accept pre-fetched message

File: src/Simplex/Messaging/Server.hs

Currently deliver (inside subscribeQueueAndDeliver, line 1641) calls tryPeekMsg ms q. Add a parameter for an optional pre-fetched message:

deliver :: Maybe Message -> (Bool, Maybe Sub) -> M s ResponseAndMessage
deliver prefetchedMsg (hasSub, sub_) = do
  stats <- asks serverStats
  fmap (either ((,Nothing) . err) id) $ liftIO $ runExceptT $ do
    msg_ <- maybe (tryPeekMsg ms q) (pure . Just) prefetchedMsg
    ...

When Nothing is passed, falls back to individual tryPeekMsg (existing behavior). When Just msg is passed, uses it directly (batched path).

Step 3: Pre-fetch messages before the processing loop

File: src/Simplex/Messaging/Server.hs

Currently (lines 1372-1375):

forever $
  atomically (readTBQueue rcvQ)
    >>= foldrM process ([], [])
    >>= \(rs_, msgs) -> ...

Add a pre-fetch step before the existing loop:

forever $ do
  batch <- atomically (readTBQueue rcvQ)
  msgMap <- prefetchMsgs batch
  foldrM (process msgMap) ([], []) batch
    >>= \(rs_, msgs) -> ...

prefetchMsgs scans the batch, collects queues from SUB commands that have a verified queue (q_ = Just (q, _)), calls tryPeekMsgs once, returns the map. For batches with no SUBs it returns an empty map (no DB call).

process passes the looked-up message (or Nothing) through to processCommand and down to deliver.

The foldrM process loop, processCommand, subscribeQueueAndDeliver, and all other command handlers stay structurally the same. Only deliver gains one parameter, and the client loop gains one pre-fetch call.

Step 4: Review

Review the typeclass signature and server usage. Confirm the interface has the right shape before implementing store backends.

Step 5: Implement for each store backend

Postgres

File: src/Simplex/Messaging/Server/MsgStore/Postgres.hs

Single query using DISTINCT ON:

SELECT DISTINCT ON (recipient_id)
  recipient_id, msg_id, msg_ts, msg_quota, msg_ntf_flag, msg_body
FROM messages
WHERE recipient_id IN ?
ORDER BY recipient_id, message_id ASC

Build Map RecipientId Message from results.

STM

File: src/Simplex/Messaging/Server/MsgStore/STM.hs

Loop over queues, call tryPeekMsg for each, collect into map.

Journal

File: src/Simplex/Messaging/Server/MsgStore/Journal.hs

Loop over queues, call tryPeekMsg for each, collect into map.

Step 6: Handle edge cases

1. Mixed batches: prefetchMsgs collects only SUB queues. Non-SUB commands get Nothing for the pre-fetched message and process unchanged.

2. Already-subscribed queues: Include in pre-fetch - deliver is called for re-SUBs too (delivers pending message).

3. Service subscriptions: The pre-fetch doesn't care about service state. sharedSubscribeQueue handles service association in STM; message peek is the same.

4. Error queues: Verification errors from receive are Left values in the batch. prefetchMsgs only looks at Right values with SUB commands.

5. Empty pre-fetch: If batch has no SUBs (e.g., all ACKs), prefetchMsgs returns empty map, no DB call made.

Step 7: Batch other commands (future, not in scope)

The same pattern (pre-fetch before loop, parameterize handler) can extend to:

• ACK with tryDelPeekMsg - batch delete+peek
• GET with tryPeekMsg - same map lookup

Lower priority since these don't have the N-at-once pattern of subscriptions.

File changes summary

File	Change
src/Simplex/Messaging/Server/MsgStore/Types.hs	Add tryPeekMsgs to typeclass
src/Simplex/Messaging/Server/MsgStore/Postgres.hs	Implement tryPeekMsgs with batch SQL
src/Simplex/Messaging/Server/MsgStore/STM.hs	Implement tryPeekMsgs as loop
src/Simplex/Messaging/Server/MsgStore/Journal.hs	Implement tryPeekMsgs as loop
src/Simplex/Messaging/Server.hs	Add prefetchMsgs, parameterize deliver

Testing

1. Existing server tests must pass unchanged (correctness preserved).
2. Add a test that subscribes a batch of queues (some with pending messages, some without) and verifies all get correct SOK + MSG responses.
3. Prometheus metrics: existing qSub stat should still increment correctly.

Performance expectation

For 300K queues across ~2200 batches:

• Before: ~300K individual DB queries
• After: ~2200 batched DB queries (one per batch of ~135)
• ~136x reduction in DB round-trips