messaging services

plans/20260328_01_server_batched_sub_processing.md

@@ -0,0 +1,152 @@
+# Server: batched SUB command processing
+
+Implementation plan for Part 1 of [RFC 2026-03-28-subscription-performance](../rfcs/2026-03-28-subscription-performance.md).
+
+## 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`:
+
+```haskell
+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:
+
+```haskell
+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):
+
+```haskell
+forever $
+  atomically (readTBQueue rcvQ)
+    >>= foldrM process ([], [])
+    >>= \(rs_, msgs) -> ...
+```
+
+Add a pre-fetch step before the existing loop:
+
+```haskell
+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`:
+
+```sql
+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

rfcs/2024-09-01-smp-message-storage.md → plans/done/2024-09-01-smp-message-storage.md

@@ -1,16 +1,17 @@
-# SMP server message storage
+
+# SMP router message storage
 
 ## Problem
 
-Currently SMP servers store all queues in server memory. As the traffic grows, so does the number of undelivered messages. What is worse, Haskell is not avoiding heap fragmentation when messages are allocated and then de-allocated - undelivered messages use ByteString and GC cannot move them around, as they use pinned memory.
+Currently SMP routers store all queues in router memory. As the traffic grows, so does the number of undelivered messages. What is worse, Haskell is not avoiding heap fragmentation when messages are allocated and then de-allocated - undelivered messages use ByteString and GC cannot move them around, as they use pinned memory.
 
 ## Possible solutions
 
 ### Solution 1: solve only GC fragmentation problem
 
 Move from ByteString to some other primitive to store messages in memory long term, e.g. ShortByteString, or manage allocation/de-allocation of stored messages manually in some other way.
 
-Pros: the simplest solution that avoids substantial re-engineering of the server.
+Pros: the simplest solution that avoids substantial re-engineering of the router.
 
 Cons:
 - not a long term solution, as memory growth still has limits.
@@ -22,12 +23,12 @@ Use files or RocksDB to store messages.
 
 Pros:
 - much lower memory usage.
-- no message loss in case of abnormal server termination (important until clients have delivery redundancy).
+- no message loss in case of abnormal router termination (important until clients have delivery redundancy).
 - this is a long term solution, and at some point it might need to be done anyway.
 
 Cons:
 - substantial re-engineering costs and risks.
-- metadata privacy. Currently we only save undelivered messages when server is restarted, with this approach all messages will be stored for some time. this argument is limited, as hosting providers of VMs can make memory snapshots too, on the other hand they are harder to analyze than files. On another hand, with this approach messages will be stored for a shorter time.
+- metadata privacy. Currently we only save undelivered messages when router is restarted, with this approach all messages will be stored for some time. this argument is limited, as hosting providers of VMs can make memory snapshots too, on the other hand they are harder to analyze than files. On another hand, with this approach messages will be stored for a shorter time.
 
 #### RocksDB and other key-value stores
 
@@ -67,7 +68,7 @@ queueLogLine =
     %s"write_msg=" digits
 ```
 
-When queue is first requested by the server:
+When queue is first requested by the router:
 
 ```c
 if queue folder exists:
@@ -87,7 +88,7 @@ nextReadMsg = read_msg
 open write_file in AppendMode
 ```
 
-When message is added to the queue (assumes that queue state is loaded to server memory, if not the previous section will be done first):
+When message is added to the queue (assumes that queue state is loaded to router memory, if not the previous section will be done first):
 
 ```c
 if write_msg > max_queue_messages:
@@ -128,7 +129,7 @@ else
     nextReadByte = current position in file
 ```
 
-When message delivery is acknowledged, the read queue needs to be advanced, and possibly switched to read from the current write_queue:
+When message delivery is acknowledged, the read queue needs to be advanced, and possibly switched to read from the current write queue:
 
 ```c
 if nextReadByte == read_byte:
@@ -162,9 +163,9 @@ Most Linux systems use EXT4 filesystem where the file lookup time scales linearl
 
 So storing all queue folders in one folder won't scale.
 
-To solve this problem we could use recipient queue ID in base64url format not as a folder name, but as a folder path, splitting it to path fragments of some length. The number of fragments can be configurable and migration to a different fragment size can be supported as the number of queues on a given server grows.
+To solve this problem we could use recipient queue ID in base64url format not as a folder name, but as a folder path, splitting it to path fragments of some length. The number of fragments can be configurable and migration to a different fragment size can be supported as the number of queues on a given router grows.
 
-Currently, queue ID is 24 bytes random number, thus allowing 2^192 possible queue IDs. If we assume that a server must hold 1b queues, it means that we have ~2^162 possible addresses for each existing queue. 24 bytes in base64 is 32 characters that can be split into say 8 fragments with 4 characters each, so that queue folder path for queue with ID `abcdefghijklmnopqrstuvwxyz012345` would be:
+Currently, queue ID is 24 bytes random number, thus allowing 2^192 possible queue IDs. If we assume that a router must hold 1b queues, it means that we have ~2^162 possible addresses for each existing queue. 24 bytes in base64 is 32 characters that can be split into say 8 fragments with 4 characters each, so that queue folder path for queue with ID `abcdefghijklmnopqrstuvwxyz012345` would be:
 
 `/var/opt/simplex/messages/abcd/efgh/ijkl/mnop/qrst/uvwx/yz01/2345`
 
@@ -174,6 +175,6 @@ So we could use an unequal split of path, two letters each and the last being lo
 
 `/var/opt/simplex/messages/ab/cd/ef/ghijklmnopqrstuvwxyz012345`
 
-The first three levels in this case can have 4096 subfolders each, and it gives 68b possible subfolders (64^2^3), so the last level will be sparse in case of 1b queues on the server. So we could make it 4 levels with 2 letters to never think about it, accounting for a large variance of the random numbers distribution:
+The first three levels in this case can have 4096 subfolders each, and it gives 68b possible subfolders (64^2^3), so the last level will be sparse in case of 1b queues on the router. So we could make it 4 levels with 2 letters to never think about it, accounting for a large variance of the random numbers distribution:
 
 `/var/opt/simplex/messages/ab/cd/ef/gh/ijklmnopqrstuvwxyz012345`

rfcs/2024-09-15-shared-port.md → plans/done/2024-09-15-shared-port.md

@@ -1,30 +1,31 @@
+
 # Sharing protocol ports with HTTPS
 
-Some networks block all ports other than web ports, including port 5223 used for SMP protocol by default. Running SMP servers on a common web port 443 would allow them to work on more networks. The servers would need to provide an HTTPS page for browsers (and probes).
+Some networks block all ports other than web ports, including port 5223 used for SMP protocol by default. Running SMP routers on a common web port 443 would allow them to work on more networks. The routers would need to provide an HTTPS page for browsers (and probes).
 
 ## Problem
 
 Browsers and tools rely on system CA bundles instead of certificate pinning.
 The crypto parameters used by HTTPS are different from what the protocols use.
 Public certificate providers like LetsEncrypt can only sign specific types of keys and Ed25519 isn't one of them.
 
-This means a server should distinguish browser and protocol clients and adjust its behavior to match.
+This means a router should distinguish browser and protocol clients and adjust its behavior to match.
 
 ## Solution
 
 `tls` package has a server hook that allows producing a different set of `TLS.Credentials` according to a client-provided "Server Name Indication" extension.
 
 Since LE certificates are only handed out to domain names, TLS client will be sending the SNI.
 However client transports are constructed over connected sockets and the SNI wouldn't be present unless explicitly requested.
-When a client sends SNI, then it's a browser and a web credentials should be used.
+When a client sends SNI, then it's a browser and web credentials should be used.
 Otherwise it's a protocol client to be offered the self-signed ca, cert and key.
 
 When a transport colocated with a HTTPS, its ALPN list should be extended with `h2 http/1.1`.
 The browsers will send it, and it should be checked before running transport client.
-If HTTP ALPN is detected, then the client connection is served with HTTP `Application` instead (the same "server information" page).
+If HTTP ALPN is detected, then the client connection is served with HTTP `Application` instead (the same "router information" page).
 
-If some client connects to server IP, doesn't send SNI and doesn't send ALPN, it will look like a pre-handshake client.
-In that case a server will send its handshake first.
+If some client connects to router IP, doesn't send SNI and doesn't send ALPN, it will look like a pre-handshake client.
+In that case a router will send its handshake first.
 This can be mitigated by delaying its handshake and letting the probe to issue its HTTP request.
 
 ## Implementation plan
@@ -43,7 +44,7 @@ runServer (tcpPort, ATransport t) = do
       else runClient serverSignKey t h `runReaderT` env -- performs serverHandshake etc as usual
 ```
 
-The web app and server live outside, so `runHttp` has to be provided by the `runSMPServer` caller.
+The web app and router live outside, so `runHttp` has to be provided by the `runSMPServer` caller.
 Additonally, Warp is using its `InternalInfo` object that's scoped to `withII` bracket.
 
 ```haskell
@@ -65,11 +66,9 @@ The implementation relies on a few modification to upstream code:
 - `warp`: Only the re-export of `serveConnection` is needed.
   Unfortunately the most recent `warp` version can't be used right away due to dependency cascade around `http-5` and `auto-update-2`.
   So a fork containing the backported re-export has to be used until the dependencies are refreshed.
-
-
 ### TLS.ServerParams
 
-When a server has port sharing enabled, a new set of TLS params is loaded and combined with transport params:
+When a router has port sharing enabled, a new set of TLS params is loaded and combined with transport params:
 
 ```haskell
 newEnv config = do
@@ -129,7 +128,7 @@ key: /etc/opt/simplex/web.key
 # key: /etc/letsencrypt/live/smp.hostname.tld/privkey.pem
 ```
 
-When `TRANSPORT.port` matches `WEB.https` the transport server becomes shared.
+When `TRANSPORT.port` matches `WEB.https` the transport router becomes shared.
 
 Perhaps a more desirable option would be explicit configuration resulting in additional transported to run:
 
@@ -148,16 +147,16 @@ key: /etc/opt/simplex/web.key
 
 ## Caveats
 
-Serving static files and the protocols togother may pose a problem for those who currently use dedicated web servers as they should switch to embedded http handlers.
+Serving static files and the protocols together may pose a problem for those who currently use dedicated web servers as they should switch to embedded http handlers.
 
 As before, using embedded HTTP server is increasing attack surface.
 
-Users who want to run everything on a single host will have to add and extra IP address and bind servers to specific IPs instead of 0.0.0.0.
-An amalgamated server binary can be provided that would contain both SMP and XFTP servers, where transport will dispatch connections by handshake ALPN.
+Users who want to run everything on a single host will have to add an extra IP address and bind routers to specific IPs instead of 0.0.0.0.
+An amalgamated router binary can be provided that would contain both SMP and XFTP routers, where transport will dispatch connections by handshake ALPN.
 
 ## Alternative: Use transports routable with reverse-proxies
 
 An "industrial" reverse proxy may do the ALPN routing, serving HTTP by itself and delegating `smp` and `xftp` to protocol servers.
 Same with the `websockets`.
 
-Since this in effect does TLS termination, the protocol servers will have to rely on credentials from protocol handshakes.
+Since this in effect does TLS termination, the protocol routers will have to rely on credentials from protocol handshakes.

rfcs/2024-11-25-journal-expiration.md → plans/done/2024-11-25-journal-expiration.md

@@ -1,8 +1,9 @@
+
 # Expiring messages in journal storage
 
 ## Problem
 
-The journal storage servers recently migrated to do not delete delivered or expired messages, they only update pointers to journal file lines. The messages are actually deleted when the whole journal file is deleted (when fully deleted or fully expired).
+The journal storage routers recently migrated to do not delete delivered or expired messages, they only update pointers to journal file lines. The messages are actually deleted when the whole journal file is deleted (when fully deleted or fully expired).
 
 The problem is that in case the queue stops receiving the new messages then writing of messages won't switch to the new journal file, and the current journal file containing delivered or expired messages would never be deleted.
 

rfcs/2026-02-17-fix-subq-deadlock.md → plans/done/2026-02-17-fix-subq-deadlock.md

@@ -1,3 +1,4 @@
+
 # Fix subQ deadlock: blocking writeTBQueue inside connLock
 
 ## Problem