@@ -702,10 +702,6 @@ dedupMessageIds = \messageIds ->
702702 else {n: acc.n + 1, items: union acc.items [mid]}) {n: 0, items: []} capped
703703 in result.items
704704
705−-- Remove the N oldest acked entries (by ackedAt timestamp)
706−evictOldestAckedIds = \items n ->
707− if n <= 0 then items
708− else diff items (takeRelation n (sortBy (\a -> a.ackedAt) items))
709705
710706-- Constant-time string comparison to prevent timing side-channels.
711707-- Always iterates maxTokenLen characters regardless of input length;
@@ -742,9 +738,9 @@ checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntr
742738 else do r <- entries; yield (if matchKey r then updateCount r else r)
743739 in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs; yield r) else newEntries}
744740
745−-- Remove the single item with the smallest key value (LRU-style eviction)
746−evictOneOldest = \items getKey ->
747− let result = fold (\acc x ->
741+-- Extract the item with the smallest key and return both the item and remaining list
742+extractOldest = \items getKey ->
743+ fold (\acc x ->
748744 case acc.candidate of
749745 Nothing {} -> {candidate: Just {value: x}, items: acc.items}
750746 Just {value: c} ->
@@ -752,9 +748,23 @@ evictOneOldest = \items getKey ->
752748 then {candidate: Just {value: x}, items: union acc.items [c]}
753749 else {candidate: acc.candidate, items: union acc.items [x]}
754750 ) {candidate: Nothing {}, items: []} items
755− in case result.candidate of
751+
752+-- Remove the single item with the smallest key value (LRU-style eviction)
753+evictOneOldest = \items getKey ->
754+ let r = extractOldest items getKey in
755+ case r.candidate of
756756 Nothing {} -> items
757− Just {} -> result.items
757+ Just {} -> r.items
758+
759+-- Remove the N items with the smallest key value. O(n*k) — efficient when k is small.
760+evictSmallest = \items n getKey ->
761+ if n <= 0 then {kept: items, evicted: []}
762+ else let r = extractOldest items getKey in
763+ case r.candidate of
764+ Nothing {} -> {kept: items, evicted: []}
765+ Just {value: c} ->
766+ let rest = evictSmallest r.items (n - 1) getKey in
767+ {kept: rest.kept, evicted: union [c] rest.evicted}
758768
759769-- Enforce per-key capacity: if a key has >= maxPerKey items, evict the oldest for that key.
760770-- Also prunes expired items.
@@ -974,7 +984,10 @@ sendBatchedGossip = \peerServers allEvents hops ->
974984
975985broadcastOfflineGossip = \t peerServers expiredLocal ->
976986 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do
977− let gossipEvents = mergeByKey (\a b -> a.pubkey == b.pubkey) (\x -> x.timestamp) [] (offlineGossipEvents expiredLocal)
987+ let cappedExpired = takeRelation maxPresenceEntries expiredLocal
988+ when (count expiredLocal > maxPresenceEntries)
989+ (println ("gossip: capped offline broadcast to " ++ show maxPresenceEntries ++ " entries (had " ++ show (count expiredLocal) ++ ")"))
990+ let gossipEvents = mergeByKey (\a b -> a.pubkey == b.pubkey) (\x -> x.timestamp) [] (offlineGossipEvents cappedExpired)
978991 sendBatchedGossip peerServers gossipEvents gossipOriginHops)
979992
980993
@@ -1124,10 +1137,11 @@ retryForwardsForServer = \entries remaining ->
11241137 Err e -> do
11251138 println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
11261139 recordFailure f.toServer
1140+ -- Stop retrying: server is unreachable, remaining forwards will fail too
11271141 Ok _ -> do
11281142 removeForward f.messageId f.toServer
11291143 clearFailure f.toServer
1130− retryForwardsForServer rest (remaining - 1)
1144+ retryForwardsForServer rest (remaining - 1)
11311145
11321146-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
11331147retryPendingForwards : [ForwardEntry] -> IO {network} {}
@@ -1210,9 +1224,9 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
12101224 if excess <= 0
12111225 then do set *ackedIds = combined; yield []
12121226 else do
1213− let evicted = takeRelation excess (sortBy (\a -> a.ackedAt) combined)
1214− set *ackedIds = diff combined evicted
1215− yield evicted
1227+ let eviction = evictSmallest combined excess (\a -> a.ackedAt)
1228+ set *ackedIds = eviction.kept
1229+ yield eviction.evicted
12161230 when (count overflow > 0) (do
12171231 dropped <- atomic do
12181232 ackOverflow <- *ackOverflow
@@ -1221,14 +1235,14 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
12211235 if overflowExcess <= 0
12221236 then do set *ackOverflow = updated; yield 0
12231237 else do
1224− let oldest = takeRelation overflowExcess (sortBy (\a -> a.ackedAt) updated)
1225− set *ackOverflow = diff updated oldest
1238+ let overflowEviction = evictSmallest updated overflowExcess (\a -> a.ackedAt)
1239+ set *ackOverflow = overflowEviction.kept
12261240 deliveredIds <- *deliveredIds
1227− let newDelivered = do a <- oldest; yield {messageId: a.messageId, toKey: a.toKey, deliveredAt: a.ackedAt}
1241+ let newDelivered = do a <- overflowEviction.evicted; yield {messageId: a.messageId, toKey: a.toKey, deliveredAt: a.ackedAt}
12281242 let combinedDelivered = union deliveredIds newDelivered
12291243 let deliveredExcess = count combinedDelivered - maxDeliveredIds
12301244 set *deliveredIds = if deliveredExcess <= 0 then combinedDelivered
1231− else diff combinedDelivered (takeRelation deliveredExcess (sortBy (\d -> d.deliveredAt) combinedDelivered))
1245+ else (evictSmallest combinedDelivered deliveredExcess (\d -> d.deliveredAt)).kept
12321246 yield overflowExcess
12331247 when (dropped > 0)
12341248 (println ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries moved to delivered set for " ++ server)))
@@ -1363,9 +1377,9 @@ ackPolledMessages = \pubkey cursor t -> do
13631377 ackedIds <- *ackedIds
13641378 let combinedAckedIds = union ackedIds ackedFederated
13651379 let excessCount = count combinedAckedIds - maxAckedIds
1366− let evictedEntries = if excessCount <= 0 then [] else takeRelation excessCount (sortBy (\a -> a.ackedAt) combinedAckedIds)
1367− set *ackedIds = diff combinedAckedIds evictedEntries
1368− yield {pendingAcks: ackedFederated, evictedEntries: evictedEntries}
1380+ let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
1381+ set *ackedIds = eviction.kept
1382+ yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
13691383 else yield {pendingAcks: [], evictedEntries: []}
13701384
13711385pollNewEvents = \pubkey clampedCursor t -> do
@@ -1653,27 +1667,29 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
16531667 else yield {valid: True {}}
16541668 if validation.valid
16551669 then do
1656− -- Reject if server started draining after challenge was issued
1657− isDraining <- *draining
1658− if isDraining then yield (respond {token: "", expiresAt: 0})
1659− else do
16601670 token <- genToken
1661− peerServers <- atomic do
1662− sessions <- *sessions
1663− presence <- *presence
1664− let afterRevoke = if revokeOthers then do s <- sessions; where s.pubkey != pubkey; yield s else sessions
1665− let capped = capPerKeyAndGlobal afterRevoke (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
1666− set *sessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
1667− let presenceWithout = presenceWithoutLocalKey presence pubkey
1668− let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
1669− let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
1670− set *presence = updatedPresence
1671− yield (if federationEnabled
1672− then collectPeerServers updatedPresence (\p -> p.server)
1673− else [])
1674− fork (when federationEnabled
1675− (sendBatchedGossip peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops))
1676− yield (respond {token: token, expiresAt: expiry})
1671+ authResult <- atomic do
1672+ isDraining <- *draining
1673+ if isDraining then yield (Nothing {})
1674+ else do
1675+ sessions <- *sessions
1676+ presence <- *presence
1677+ let afterRevoke = if revokeOthers then do s <- sessions; where s.pubkey != pubkey; yield s else sessions
1678+ let capped = capPerKeyAndGlobal afterRevoke (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
1679+ set *sessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
1680+ let presenceWithout = presenceWithoutLocalKey presence pubkey
1681+ let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
1682+ let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
1683+ set *presence = updatedPresence
1684+ yield (Just {value: if federationEnabled
1685+ then collectPeerServers updatedPresence (\p -> p.server)
1686+ else []})
1687+ case authResult of
1688+ Nothing {} -> yield (respond {token: "", expiresAt: 0})
1689+ Just {value: peerServers} -> do
1690+ fork (when federationEnabled
1691+ (sendBatchedGossip peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops))
1692+ yield (respond {token: token, expiresAt: expiry})
16771693 else yield (respond {token: "", expiresAt: 0})
16781694
16791695-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor
@@ -1700,16 +1716,15 @@ handlePoll = \cursor authorization respond -> do
17001716-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
17011717handleSendMessage = \to encryptedBlob timestamp messageId authorization respond -> do
17021718 t <- now
1703− draining <- *draining
1704− if draining then yield (respond {status: MsDraining {}, messageId: ""})
1705− else do
17061719 globalRateLimited <- checkGlobalRate t
17071720 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: ""})
17081721 else do
17091722 effectiveMsgId <- if messageId != "" then yield messageId else genToken
17101723 let clampedTs = clampTimestamp timestamp t
17111724 result <- atomic do
1712− withSessionAuth authorization t
1725+ draining <- *draining
1726+ if draining then yield {send: SendFailed {status: MsDraining {}}, senderRateLimited: False {}, globalCapReached: False {}}
1727+ else withSessionAuth authorization t
17131728 (yield {send: SendFailed {status: MsUnauthorized {}}, senderRateLimited: False {}, globalCapReached: False {}})
17141729 (\senderKey -> do
17151730 -- Reject self-sends (sender == recipient wastes a mailbox slot)
@@ -1801,14 +1816,13 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
18011816-- Returns local server if key is online locally, otherwise the remote with latest expiry.
18021817handleLookupPresence = \pubkey authorization respond -> do
18031818 t <- now
1804− draining <- *draining
1805− if draining then yield (respond {server: "", authorized: False {}})
1806− else do
18071819 globalRateLimited <- checkGlobalRate t
18081820 if globalRateLimited then yield (respond {server: "", authorized: False {}})
18091821 else do
18101822 result <- atomic do
1811− withSessionAuth authorization t
1823+ draining <- *draining
1824+ if draining then yield {server: "", authorized: False {}}
1825+ else withSessionAuth authorization t
18121826 (yield {server: "", authorized: False {}})
18131827 (\senderKey -> do
18141828 lookupRates <- *lookupRates
@@ -1879,13 +1893,13 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
18791893 ackedIds <- *ackedIds
18801894 deliveredIds <- *deliveredIds
18811895 if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds
1882− then yield {forward: ForwardDuplicate {}, globalCapReached: False {}}
1896+ then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}}
18831897 else do
18841898 totalMsgCount <- *totalMessageCount
18851899 if totalMsgCount >= maxMessages
1886− then yield {forward: ForwardRejected {}, globalCapReached: True {}}
1900+ then yield {forward: ForwardRejected {}, globalCapReached: True {}, recvRateLimited: False {}}
18871901 else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1888− then yield {forward: ForwardRejected {}, globalCapReached: False {}}
1902+ then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
18891903 else do
18901904 recvRates <- *recvRates
18911905 let recvResult = checkRateWindow recvRates (\r -> r.pubkey == toKey)
@@ -1893,7 +1907,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
18931907 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
18941908 maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
18951909 if recvResult.limited
1896− then yield {forward: ForwardRejected {}, globalCapReached: False {}}
1910+ then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: True {}}
18971911 else do
18981912 set *recvRates = recvResult.entries
18991913 seqCounter <- *seqCounter
@@ -1905,11 +1919,13 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
19051919 set *messages = updatedMessages
19061920 set *totalMessageCount = totalMsgCount + 1
19071921 signalMessageForKey toKey seqTs
1908− yield {forward: ForwardStored {}, globalCapReached: False {}}
1922+ yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
19091923 case result.forward of
19101924 ForwardRejected {} ->
19111925 if result.globalCapReached
19121926 then println ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
1927+ else if result.recvRateLimited
1928+ then println ("forward: recv rate limited (" ++ show maxRecvRatePerKey ++ "/s per key), rejecting for " ++ toKey)
19131929 else println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
19141930 _ -> yield {}
19151931 yield (case result.forward of
@@ -1939,7 +1955,8 @@ handleFederationAck = \messageIds originServer fromServer authorization respond
19391955 let pastGrace = do f <- removable; where any messages (\m -> m.messageId == f.messageId && m.toKey == f.toKey && t - m.receivedAt >= ackGracePeriod); yield f
19401956 let updatedMessages = do m <- messages; where none pastGrace (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m
19411957 set *messages = updatedMessages
1942− set *totalMessageCount = count updatedMessages
1958+ totalMsgCount <- *totalMessageCount
1959+ set *totalMessageCount = totalMsgCount - count pastGrace
19431960 yield (respond {ok: True {}, error: ""}))
19441961
19451962-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
@@ -1962,7 +1979,9 @@ handleDrain = \authorization respond -> do
19621979 ackedIds <- *ackedIds
19631980 let pendingForwardCount = count forwards
19641981 let pendingAckCount = count ackedIds
1965− -- Block on gossip (don't fork) so drain completes before process exit
1982+ -- Block on gossip (don't fork) so drain completes before process exit.
1983+ -- NOTE: If peers are unreachable, this blocks until fetch returns (no timeout in runtime).
1984+ -- Consider killing the process with a deadline after calling /drain.
19661985 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
19671986 let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
19681987 sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
@@ -2167,8 +2186,11 @@ main = do
21672186 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
21682187 && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0
21692188 && maxPresenceTtl > 0 && maxForwardsEntries > 0 && maxAckedIds > 0
2189+ let peerSecretsOk = none federationPeerSecrets (\p -> isBadServerName (normalizeServerName p.server) || p.secret == "")
21702190 if not configOk
21712191 then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
2192+ else if not peerSecretsOk
2193+ then println "FATAL: invalid federationPeerSecrets — server names must pass validation and secrets must be non-empty"
21722194 else do
21732195 t <- now
21742196 atomic do set *startedAt = t