@@ -452,10 +452,11 @@ clearFailure = \server -> atomic (do
452452 *forwardFailures = filter (\f -> f.server != server) forwardFailures)
453453
454454-- ============================================
455−-- Common utility queries
455+-- Common helpers
456456-- ============================================
457−-- Pure helpers shared by send/poll/federation paths: monotonic sequence
458−-- generation and presence lookup.
457+-- Helpers shared by send/poll/federation paths: monotonic sequence
458+-- generation, presence queries, and atomic mailbox/presence writes
459+-- (callers supply the surrounding `atomic` block).
459460
460461-- Strictly-monotonic next sequence timestamp: max of wall clock and prior+1.
461462-- Guarantees /poll cursors order events even when many arrive in the same ms.
@@ -481,6 +482,28 @@ collectPeers = \items getServer -> filter isRemote (map getServer items)
481482knownPeerServersFromPresence : [PresenceEntry] -> [Text]
482483knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.server)
483484
485+-- Append one ciphertext to the local mailbox under a fresh monotonic
486+-- sequence and bump pollWakeup so blocked /poll holders unblock. Caller
487+-- wraps in `atomic`.
488+appendMessage : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWakeup, rw *seqCounter} {}
489+appendMessage = \toKey encryptedBlob t -> do
490+ msgs <- *messages
491+ seq <- *seqCounter
492+ let seqTs = nextSeqTs t seq
493+ *seqCounter = seqTs
494+ *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs}]
495+ *pollWakeup = seqTs
496+
497+-- Replace any local presence rows for `pubkey` with one fresh row at
498+-- `expiry`, and return the post-update peer broadcast set. Caller wraps in
499+-- `atomic`.
500+refreshLocalPresence : [PresenceEntry] -> PubkeyHex -> Timestamp -> IO {rw *presence} [Text]
501+refreshLocalPresence = \pres pubkey expiry -> do
502+ let kept = filter (\p -> not (p.pubkey == pubkey && isLocalServer p.server)) pres
503+ let newPres = union kept [{pubkey, server: serverHost, expiresAt: expiry}]
504+ *presence = newPres
505+ yield (knownPeerServersFromPresence newPres)
506+
484507-- ============================================
485508-- Pruning
486509-- ============================================
@@ -638,10 +661,8 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
638661 let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp}
639662 *sessions = capList (union pruned [newSession]) maxSessions
640663 pres <- *presence
641− let prunedPres = filter (\p -> not (p.pubkey == pubkey && isLocalServer p.server)) pres
642− let newPres = union prunedPres [{pubkey, server: serverHost, expiresAt: expiry}]
643− *presence = newPres
644− yield (Just {value: knownPeerServersFromPresence newPres}))
664+ peers <- refreshLocalPresence pres pubkey expiry
665+ yield (Just {value: peers}))
645666 case result of
646667 Nothing {} -> yield (Err {error: {status: 401, message: "invalid"}})
647668 Just {value: peers} -> do
@@ -672,10 +693,8 @@ handlePoll = \cursor authorization clientIp -> do
672693 let stale = not any (\p -> mine p && p.expiresAt - t >= sessionExpiry / 2) pres
673694 refreshPeers <- if stale
674695 then do
675− let kept = filter (\p -> not mine p) pres
676− let newPres = union kept [{pubkey, server: serverHost, expiresAt: t + sessionExpiry}]
677− *presence = newPres
678− yield (Just {value: knownPeerServersFromPresence newPres})
696+ peers <- refreshLocalPresence pres pubkey (t + sessionExpiry)
697+ yield (Just {value: peers})
679698 else yield (Nothing {})
680699 yield (Just {value: {pubkey, clampedCursor, startWakeup, refreshPeers}}))
681700 case setup of
@@ -769,12 +788,7 @@ handleSendMessages = \messages authorization clientIp -> do
769788-- via `resolveSendTarget`.
770789commitSendMessage : PubkeyHex -> BlobHex -> [Text] -> Int<Ms> -> IO {rw *forwards, rw *messages, rw *pollWakeup, rw *seqCounter} {}
771790commitSendMessage = \to encryptedBlob remoteServers t -> do
772− msgs <- *messages
773− seq <- *seqCounter
774− let seqTs = nextSeqTs t seq
775− *seqCounter = seqTs
776− *messages = union msgs [{toKey: to, encryptedBlob, receivedAt: seqTs}]
777− *pollWakeup = seqTs
791+ appendMessage to encryptedBlob t
778792 when federationEnabled (do
779793 fwds <- *forwards
780794 let newFwds = map (\s -> {toServer: s, toKey: to, encryptedBlob, retries: 0}) (filter (\s -> not any (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s) fwds) remoteServers)
@@ -860,15 +874,10 @@ handleRecvGossip = \events fromServer -> do
860874storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWakeup, r *presence, rw *seqCounter} (Result HttpError {})
861875storeFederated = \toKey encryptedBlob t -> do
862876 pres <- *presence
863− msgs <- *messages
864877 if not (hasLocalPresence pres toKey t)
865878 then yield (Err {error: {status: 404, message: "no_presence"}})
866879 else do
867− seq <- *seqCounter
868− let seqTs = nextSeqTs t seq
869− *seqCounter = seqTs
870− *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs}]
871− *pollWakeup = seqTs
880+ appendMessage toKey encryptedBlob t
872881 yield (Ok {value: {}})
873882
874883-- POST /federation/forward: receive a single federated message destined for