agent: process SMP messages concurrently between different connections

[epoberezkin]

No description provided.

[simplex-chat-agent]

What this branch does

Removes the two single-thread bottlenecks in message reception by replacing internal TBQueues with callbacks.

Layer 1 (SMP client → agent). AgentClient.msgQ and SMPClientAgent.msgQ are gone. getProtocolClient now takes Maybe (ServerTransmissionBatch … -> IO ()) instead of Maybe (TBQueue …). Each protocol client still creates an internal per-client msgQ (created iff a callback is supplied) and a dedicated readMsgs thread (added to the client's raceAny_) that drains it and invokes the callback. Because processMsgs (receive loop) and writeSMPMessage (inline MSG in a SUB response) both write to that single per-client queue read by a single thread, per-client ordering is preserved without the MVar lock the plan proposed. Across clients the callbacks run on independent threads → cross-connection parallelism. The agent wires processServerMsg c = subscriber c (was a dedicated subscriber thread; now a per-batch callback wrapping agentOperationBracket AORcvNetwork … processSMPTransmissions).

Layer 2 (agent → app). AgentClient.subQ is replaced by processEvent :: Bool -> ATransmission -> IO () (the Bool selects blocking vs non-blocking, replacing nonBlockingWriteTBQueue). New helpers notifyEvent/nonBlockingNotifyEvent wrap it; all ~30 writeTBQueue subQ / nonBlockingWriteTBQueue subQ sites across Agent.hs, Agent/Client.hs, NtfSubSupervisor.hs, FileTransfer/Agent.hs are converted. The pendingMsgs/isFullTBQueue subQ buffering in runCommandProcessing and processSMPTransmissions is deleted (runProcessCmd/runProcessSMP wrappers removed). endAgentOperation and friends now return Bool so the SUSPENDED event is emitted via notifyEvent outside STM (new endAgentOp helper) rather than by writing to subQ inside the transaction.

getSMPAgentClient(_) now takes the processEvent callback and no longer starts threads; thread startup is split into the new startSMPAgentClient. The Ntf server's ntfSubscriber receiveSMP loop becomes the receiveSMPMessage callback threaded through newNtfServerEnv/newNtfSubscriber/newSMPClientAgent; several Ntf functions (pushNotification, getOrCreatePushWorker, runPushWorker) are de-ReaderT'd to plain IO with explicit store/stats args, plus a new incNtfStatT_. Tests gain a wrapper AgentClient record (real client + test subQ + skippedQ) with DOWN/UP reorder tolerance, since cross-connection event ordering is now nondeterministic.

The refactor is coherent and internally consistent; no leftover references to the removed subQ/msgQ fields, and the callback plumbing type-checks by construction. Comments below are minor.

Points to confirm

E.uninterruptibleMask_ scope in processSMPTransmissions (Agent.hs:3081). Each transmission is now processed under uninterruptibleMask_, which spans DB access, ratchet decryption, withConnLock acquisition (blocking STM), and the synchronous app callback (notifyEvent, which may block under backpressure). uninterruptibleMask_ (vs mask_) makes even those blocking points non-interruptible, so a readMsgs thread stuck on a contended conn lock or a slow callback will not respond to disposeAgentClient/reconnect cancellation. This looks deliberate (protecting ratchet-advance + ACK from mid-processing cancellation now that processing runs in the cancellable SMP-client thread rather than the old dedicated subscriber thread), but it's a meaningful shutdown/liveness tradeoff and worth an explicit confirmation/comment. If the intent is only to defer async delivery at safe points, mask_ would be the lighter tool.

Minor

• SMPClientAgentConfig.msgQSize (default 2048) is now dead — newSMPClientAgent no longer reads it, and the per-client msgQ is sized from smpCfg's qSize. For the Ntf subscriber this also changes the effective receive buffer from 2048 to the SMP client config qSize. Consider removing the field or documenting the new sizing.
• getAgentQueuesInfo is retained but msgQInfo/subQInfo are now hardcoded TBQueueInfo {qLength = 0, qFull = False}. The per-client sndQ/rcvQ info is still real, but the two top-level fields are now permanently zero (misleading for the control/debug consumer). The plan called for removing them entirely.
• Two whitespace nits introduced by the diff (inline suggestions): trailing whitespace at Notifications/Server.hs:658, double space in Client/Agent.hs:256.
• newSMPProxyAgent passes a no-op (\_ _ -> pure ()) callback — proxy (SSender) never receives server messages, so silently dropping is fine, but worth a one-line comment.

[simplex-chat-agent]

Double space after = introduced in this hunk.

Suggested change

	connectClient :: SMPClientAgent p -> SMPServer -> SMPClientVar -> IO (Either SMPClientError SMPClient)
	connectClient ca@SMPClientAgent {agentCfg, dbService, smpClients, smpSessions, msgQ, randomDrg, startedAt} srv v = case dbService of
	connectClient ca@SMPClientAgent {agentCfg, dbService, smpClients, smpSessions, processMsg, randomDrg, startedAt} srv v = case dbService of
	Just dbs -> runExceptT $ do
	connectClient :: SMPClientAgent p -> SMPServer -> SMPClientVar -> IO (Either SMPClientError SMPClient)
	connectClient ca@SMPClientAgent {agentCfg, dbService, smpClients, smpSessions, processMsg, randomDrg, startedAt} srv v = case dbService of
	Just dbs -> runExceptT $ do

[simplex-chat-agent]

Trailing whitespace introduced in this hunk.

Suggested change

	q <- newTBQueueIO pushQSize
	tId <- mkWeakThreadId =<< forkIO (runPushWorker s srvHost_ isOwn q)
	atomically $ putTMVar (sessionVar v) PushWorker {workerQ = q, workerThreadId = tId}
	q <- newTBQueueIO pushQSize
	tId <- mkWeakThreadId =<< forkIO (runPushWorker s srvHost_ isOwn q)
	atomically $ putTMVar (sessionVar v) PushWorker {workerQ = q, workerThreadId = tId}

[simplex-chat-agent]

Summary

This branch removes the two single-thread bottlenecks in message reception by replacing shared queues with callbacks:

• Agent msgQ (shared TBQueue, single subscriber thread) is gone. getSMPAgentClient_ now takes a processEvent callback, and newAgentClient stores a processServerMsg :: AgentClient -> ServerTransmissionBatch … -> IO () that runs subscriber/processSMPTransmissions directly on each SMP client's own thread.
• Agent subQ (shared TBQueue, single consumer) is replaced by processEvent :: Bool -> ATransmission -> IO () (the Bool selects blocking vs non-blocking via notifyEvent / nonBlockingNotifyEvent).
• Agent startup is split: getSMPAgentClient_ builds the client, new startSMPAgentClient starts the background threads.
• The Ntf server's ntfSubscriber loses its receiveSMP loop; the processing logic becomes receiveSMPMessage, threaded through newNtfServerEnv/newNtfSubscriber/newSMPClientAgent as a callback. This forced pushNotification/getOrCreatePushWorker/runPushWorker from M to IO with explicit store/stats args — large but mechanical.

Notably, the implementation does not match the MVar-lock design in plans/…: Client.hs keeps a per-client internal msgQ and adds a dedicated readMsgs reader thread. processMsgs (correlation thread) and writeSMPMessage (inline MSG on SUB response) both write to this per-client queue, and the single readMsgs thread drains it into the callback. That gives per-client serialization and cross-client parallelism without an explicit lock — cleaner than the plan. This is correct and worth the deviation.

I read all 17 changed files plus Client.hs, Agent.hs, Agent/Client.hs, Client/Agent.hs, the Ntf server/env, and the test harness in full. The refactor is coherent and the endAgentOperation/suspendOperation change (returning Bool so SUSPENDED is emitted outside STM, since notification is now an IO callback) is handled consistently at every call site. Only one trivial nit is inline (double blank line). The items below are for discussion, not blockers.

Discussion points

1. Callback runs under uninterruptibleMask_ while holding the connection lock. processSMPTransmissions now wraps each transmission in E.uninterruptibleMask_, and message events reach the app callback via notify_ → notifyEvent inside withConnLock c connId "processSMP". So the application's event handler executes both under an uninterruptible mask and while holding the agent's per-connection lock. If that handler blocks indefinitely (or blocks on something that needs the same connId lock), the SMP client thread becomes unkillable and teardown hangs. The forkIOWithUnmask ($ work) change in runWorkerAsync is precisely to stop workers forked during this window from inheriting the mask (else killThread hangs) — evidence the masking interaction is delicate. Worth confirming the chat callback can never block uninterruptibly, and documenting this contract.

2. nonBlocking* no longer guarantees non-blocking. The old nonBlockingWriteTBQueue forked a writer when subQ was full, guaranteeing the caller (often holding a lock, e.g. notifySub', newProtocolClient CONNECT) never blocked. nonBlockingNotifyEvent c t = processEvent c False t just passes False and calls the callback synchronously; the non-blocking guarantee is now delegated to the callback. Fine if the app honors it, but the contract moved out of this repo — worth noting.

3. Dead / misleading leftovers. SMPClientAgentConfig.msgQSize (Client/Agent.hs:110, default 2048) is no longer read by newSMPClientAgent — dead field. getAgentQueuesInfo (Agent/Client.hs:2860) now fabricates msgQInfo/subQInfo as {qLength = 0, qFull = False} since those queues are gone; the reported values are meaningless. The plan called for removing AgentQueuesInfo/getAgentQueuesInfo entirely — either drop them or drop the two synthetic fields so monitoring isn't misleading. Neither line is inside the diff, so no inline suggestion.

Already flagged by the bot (not re-raised)

• Double space after = in connectClient (Client/Agent.hs).
• Trailing whitespace after forkIO (runPushWorker …) (Notifications/Server.hs).

[simplex-chat-agent]

The rewrite of subscriber left a double blank line before cleanupManager. Collapse to a single blank line to match the rest of the module.

Suggested change

	cleanupManager :: AgentClient -> AM' ()
	cleanupManager :: AgentClient -> AM' ()