@@ -682,10 +682,10 @@ nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
682682-- Deduplicate message IDs, preserving order, capped at maxMessageIds
683683dedupMessageIds : [MessageIdHex] -> [MessageIdHex]
684684dedupMessageIds = \messageIds ->
685+ let capped = takeRelation maxMessageIds messageIds in
685686 let result = fold (\acc mid ->
686− if acc.n >= maxMessageIds then acc
687− else if any acc.items (\a -> a == mid) then acc
688− else {n: acc.n + 1, items: union acc.items [mid]}) {n: 0, items: []} messageIds
687+ if any acc.items (\a -> a == mid) then acc
688+ else {n: acc.n + 1, items: union acc.items [mid]}) {n: 0, items: []} capped
689689 in result.items
690690
691691-- Remove the N oldest acked entries (by ackedAt timestamp)
@@ -707,12 +707,8 @@ safeCompareLoop = \s1 s2 n result ->
707707
708708safeCompare : Text -> Text -> Bool
709709safeCompare = \a b ->
710− let lenA = count (chars a) in
711− let lenB = count (chars b) in
712− safeCompareLoop a b (max maxTokenLen (max lenA lenB)) 0
713− + (if lenA != lenB then 1 else 0)
714− + (if a == "" then 1 else 0)
715− + (if b == "" then 1 else 0) == 0
710+ if a == "" || b == "" then False {}
711+ else safeCompareLoop a b maxTokenLen 0 == 0
716712
717713-- Generic sliding-window rate limiter. Returns whether the request is rate-limited
718714-- and the updated entries list. Callers provide match/create/update functions for their entry type.
@@ -825,6 +821,14 @@ countMessagesFor : [Message] -> PubkeyHex -> Timestamp -> Int
825821countMessagesFor = \messages toKey t ->
826822 count (do m <- messages; where m.toKey == toKey && m.receivedAt >= t - messageMaxAge; yield m)
827823
824+-- Count total messages and messages for a specific key in a single pass (avoids two O(n) scans)
825+messageCounts : [Message] -> PubkeyHex -> Timestamp -> {total: Int, forKey: Int}
826+messageCounts = \messages toKey t ->
827+ fold (\acc m ->
828+ {total: acc.total + 1,
829+ forKey: if m.toKey == toKey && m.receivedAt >= t - messageMaxAge then acc.forKey + 1 else acc.forKey}
830+ ) {total: 0, forKey: 0} messages
831+
828832hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool
829833hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)
830834
@@ -867,54 +871,69 @@ sendToEachN = \servers action remaining ->
867871-- Remove all expired entries from every collection. Returns expired local presence
868872-- entries (for broadcasting offline gossip) and the set of peer servers known before pruning.
869873pruneExpired = \t -> do
870− challenges <- *challenges
871− set *challenges = do x <- challenges; where x.expiresAt > t; yield x
872− sessions <- *sessions
873− set *sessions = do x <- sessions; where x.expiresAt > t; yield x
874− presenceBefore <- *presence
875− let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)
876− let presenceAfter = do x <- presenceBefore; where x.expiresAt > t; yield x
877− set *presence = presenceAfter
878− let messageCutoff = t - unackedMessageMaxAge
879− let retainCutoff = t - messageMaxAge
880− messages <- *messages
881− set *messages = do m <- messages; where m.receivedAt > messageCutoff; yield m
882− forwards <- *forwards
883− set *forwards = do f <- forwards; where f.createdAt > retainCutoff; yield f
884− ackedIds <- *ackedIds
885− set *ackedIds = do a <- ackedIds; where a.ackedAt > retainCutoff; yield a
886− sendRates <- *sendRates
887− set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r
888− forwardFailures <- *forwardFailures
889− set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f
890− challengeRates <- *challengeRates
891− set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r
892− federationRates <- *federationRates
893− set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs; yield r
894− activePolls <- *activePolls
895− set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p
896− ipRates <- *ipRates
897− set *ipRates = do r <- ipRates; where t - r.windowStart < ipRateWindowMs; yield r
898− lookupRates <- *lookupRates
899− set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs; yield r
900− recvRates <- *recvRates
901− set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r
902− recentGossip <- *recentGossip
903− set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
904− ackOverflow <- *ackOverflow
905− set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a
906− deliveredIds <- *deliveredIds
907− set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > retainCutoff; yield d
908− yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
874+ -- Prune presence (must be atomic for before/after comparison)
875+ presenceResult <- atomic do
876+ presenceBefore <- *presence
877+ let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)
878+ let presenceAfter = do x <- presenceBefore; where x.expiresAt > t; yield x
879+ set *presence = presenceAfter
880+ yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
881+ -- Prune auth state
882+ atomic do
883+ challenges <- *challenges
884+ set *challenges = do x <- challenges; where x.expiresAt > t; yield x
885+ sessions <- *sessions
886+ set *sessions = do x <- sessions; where x.expiresAt > t; yield x
887+ -- Prune messages (largest collection, isolated for minimal lock time)
888+ atomic do
889+ let messageCutoff = t - unackedMessageMaxAge
890+ messages <- *messages
891+ set *messages = do m <- messages; where m.receivedAt > messageCutoff; yield m
892+ -- Prune federation state
893+ atomic do
894+ let retainCutoff = t - messageMaxAge
895+ forwards <- *forwards
896+ set *forwards = do f <- forwards; where f.createdAt > retainCutoff; yield f
897+ ackedIds <- *ackedIds
898+ set *ackedIds = do a <- ackedIds; where a.ackedAt > retainCutoff; yield a
899+ forwardFailures <- *forwardFailures
900+ set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f
901+ ackOverflow <- *ackOverflow
902+ set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a
903+ deliveredIds <- *deliveredIds
904+ set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > retainCutoff; yield d
905+ -- Prune rate limiters and ephemeral state
906+ atomic do
907+ sendRates <- *sendRates
908+ set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r
909+ challengeRates <- *challengeRates
910+ set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r
911+ federationRates <- *federationRates
912+ set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs; yield r
913+ activePolls <- *activePolls
914+ set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p
915+ ipRates <- *ipRates
916+ set *ipRates = do r <- ipRates; where t - r.windowStart < ipRateWindowMs; yield r
917+ lookupRates <- *lookupRates
918+ set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs; yield r
919+ recvRates <- *recvRates
920+ set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r
921+ recentGossip <- *recentGossip
922+ set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
923+ yield presenceResult
909924
910925-- Opportunistic pruning: runs at most once per second, triggered by request handlers
911926maybePrune = \t -> do
912− lastPruned <- *lastPrunedAt
913− if t - lastPruned < 1000
927+ shouldRun <- atomic do
928+ lastPruned <- *lastPrunedAt
929+ if t - lastPruned < 1000
930+ then yield (False {})
931+ else do
932+ set *lastPrunedAt = t
933+ yield (True {})
934+ if not shouldRun
914935 then yield {expiredLocal: [], peerServers: []}
915− else do
916− set *lastPrunedAt = t
917− pruneExpired t
936+ else pruneExpired t
918937
919938splitAtN = \items n ->
920939 fold (\acc e ->
@@ -936,7 +955,7 @@ sendBatchedGossipToServer = \server allEvents hops ->
936955 authorization: federationAuth,
937956 events: batch, hopsRemaining: hops, fromServer: serverHostN})
938957 case result of
939− Err _ -> println ("federation: gossip to " ++ server ++ " failed")
958+ Err e -> println ("federation: gossip to " ++ server ++ " failed: " ++ show e)
940959 Ok _ -> yield {})
941960
942961sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}
@@ -969,7 +988,7 @@ mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
969988 withinCapacity: capResult.withinCapacity}
970989
971990maybePruneAndBroadcast = \t -> do
972− pruneResult <- atomic do maybePrune t
991+ pruneResult <- maybePrune t
973992 forkBroadcastOffline t pruneResult
974993
975994checkFederationRate = \server t -> atomic do
@@ -1042,7 +1061,9 @@ withFederationAuth = \authorization t respond fromServer precondition action ->
10421061 println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
10431062 maybePruneAndBroadcast t
10441063 yield (respond {ok: False {}, error: "rate_limited"})
1045− else action
1064+ else do
1065+ clearFailure fromServer
1066+ action
10461067
10471068withSessionAuth = \authorization t onUnauth onAuth -> do
10481069 sessions <- *sessions
@@ -1098,8 +1119,8 @@ retryForwardsForServer = \entries remaining ->
10981119 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
10991120 fromServer: serverHostN, authorization: federationAuth})
11001121 case result of
1101− Err _ -> do
1102− println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId)
1122+ Err e -> do
1123+ println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
11031124 recordFailure f.toServer
11041125 Ok _ -> do
11051126 removeForward f.messageId f.toServer
@@ -1147,8 +1168,8 @@ sendBatchedAcksToServer = \serverAcks server ->
11471168 forEachBatch serverAcks maxMessageIds (\batch -> do
11481169 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
11491170 case result of
1150− Err _ -> do
1151− println ("federation: ack to " ++ server ++ " failed")
1171+ Err e -> do
1172+ println ("federation: ack to " ++ server ++ " failed: " ++ show e)
11521173 recordFailure server
11531174 Ok _ -> do
11541175 clearFailure server
@@ -1177,7 +1198,8 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
11771198 sendToEach evictedOriginServers (\server -> do
11781199 fetchResult <- fetchFederationAck server (do a <- acksForServer evictedToRetry server; yield a.messageId)
11791200 case fetchResult of
1180− Err _ -> do
1201+ Err e -> do
1202+ println ("federation: evicted ack to " ++ server ++ " failed: " ++ show e)
11811203 overflow <- atomic do
11821204 ackedIds <- *ackedIds
11831205 let reinsert = acksForServer evictedToRetry server
@@ -1220,8 +1242,8 @@ retryOverflowAcks = \overflow -> do
12201242 forEachBatch (acksForServer overflow server) maxMessageIds (\batch -> do
12211243 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
12221244 case result of
1223− Err _ -> do
1224− println ("federation: overflow ack to " ++ server ++ " failed")
1245+ Err e -> do
1246+ println ("federation: overflow ack to " ++ server ++ " failed: " ++ show e)
12251247 recordFailure server
12261248 Ok _ -> do
12271249 clearFailure server
@@ -1233,9 +1255,9 @@ retryOverflowAcks = \overflow -> do
12331255-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
12341256-- nearing expiry. Called by both the background task and the /maintenance/prune endpoint.
12351257pruneAndRetry = \t -> do
1258+ atomic do set *lastPrunedAt = t
1259+ pr <- pruneExpired t
12361260 result <- atomic do
1237− set *lastPrunedAt = t
1238− pr <- pruneExpired t
12391261 forwards <- *forwards
12401262 ackedIds2 <- *ackedIds
12411263 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a
@@ -1247,8 +1269,8 @@ pruneAndRetry = \t -> do
12471269 let localActive = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []
12481270 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []
12491271 when shouldRefresh (set *lastPresenceRefresh = t)
1250− yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks, pendingOverflow: pendingOverflow, localActive: localActive, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
1251− broadcastOfflineGossip t result.peerServers result.expiredLocal
1272+ yield {pendingForwards: forwards, pendingAcks: pendingAcks, pendingOverflow: pendingOverflow, localActive: localActive, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
1273+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal
12521274 when federationEnabled (do
12531275 retryPendingForwards result.pendingForwards
12541276 sendAcksAndClean result.pendingAcks
@@ -1474,6 +1496,10 @@ forwardCandidates = \targets messages forwards t ->
14741496challengeMatch = \challenges pubkey challenge ip t ->
14751497 any challenges (\c -> c.pubkey == pubkey && safeCompare c.challenge challenge && c.expiresAt > t && (ip == "" || c.ip == "" || c.ip == ip))
14761498
1499+-- Check if a challenge exists for this key, ignoring IP (for unconditional consumption)
1500+challengeExistsForKey = \challenges pubkey challenge t ->
1501+ any challenges (\c -> c.pubkey == pubkey && safeCompare c.challenge challenge && c.expiresAt > t)
1502+
14771503challengesExcluding = \challenges pubkey challenge ->
14781504 do c <- challenges; where c.pubkey != pubkey || c.challenge != challenge; yield c
14791505
@@ -1544,8 +1570,8 @@ handleChallenge = \pubkey clientIp respond -> do
15441570 yield (respond {challenge: "", ok: False {}, expiresAt: 0})
15451571 else do
15461572 challenge <- genToken
1573+ pr <- maybePrune t
15471574 result <- atomic do
1548− pr <- maybePrune t
15491575 challenges <- *challenges
15501576 let active = do c <- challenges; where c.expiresAt > t; yield c
15511577 let activeForKey = count (do c <- active; where c.pubkey == pubkey; yield c)
@@ -1555,7 +1581,7 @@ handleChallenge = \pubkey clientIp respond -> do
15551581 union others (evictOneOldest forKey (\c -> c.expiresAt))
15561582 else active
15571583 if count cappedActive >= maxChallenges
1558− then yield {pr: pr, accepted: False {}, rateLimited: False {}}
1584+ then yield {accepted: False {}, rateLimited: False {}}
15591585 else do
15601586 challengeRates <- *challengeRates
15611587 let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
@@ -1563,16 +1589,16 @@ handleChallenge = \pubkey clientIp respond -> do
15631589 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
15641590 maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
15651591 if rateResult.limited
1566− then yield {pr: pr, accepted: False {}, rateLimited: True {}}
1592+ then yield {accepted: False {}, rateLimited: True {}}
15671593 else do
15681594 set *challengeRates = rateResult.entries
15691595 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
1570− yield {pr: pr, accepted: True {}, rateLimited: False {}}
1596+ yield {accepted: True {}, rateLimited: False {}}
15711597 when result.rateLimited
15721598 (println ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
15731599 when (not result.accepted && not result.rateLimited)
15741600 (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
1575− forkBroadcastOffline t result.pr
1601+ forkBroadcastOffline t pr
15761602 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
15771603
15781604-- POST /auth/verify: Verify Ed25519 signature over challenge string.
@@ -1602,15 +1628,16 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
16021628verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do
16031629 let expiry = t + sessionExpiry
16041630 let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
1631+ pr <- maybePrune t
16051632 validation <- atomic do
1606− pr <- maybePrune t
16071633 challenges <- *challenges
16081634 let hasMatch = challengeMatch challenges pubkey challenge clientIp t
1609− -- Consume challenge on any verify attempt (prevents signature brute-force)
1610− when hasMatch (set *challenges = challengesExcluding challenges pubkey challenge)
1635+ -- Always consume challenge if it exists, even on IP mismatch (prevents brute-force from other IPs)
1636+ let exists = challengeExistsForKey challenges pubkey challenge t
1637+ when exists (set *challenges = challengesExcluding challenges pubkey challenge)
16111638 if not hasMatch || not sigValid
1612− then yield {pr: pr, valid: False {}}
1613− else yield {pr: pr, valid: True {}}
1639+ then yield {valid: False {}}
1640+ else yield {valid: True {}}
16141641 if validation.valid
16151642 then do
16161643 token <- genToken
@@ -1627,14 +1654,14 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
16271654 yield (if federationEnabled
16281655 then collectPeerServers updatedPresence (\p -> p.server)
16291656 else [])
1630− let expiredLocal = do p <- validation.pr.expiredLocal; where p.pubkey != pubkey; yield p
1657+ let expiredLocal = do p <- pr.expiredLocal; where p.pubkey != pubkey; yield p
16311658 fork (do
16321659 when federationEnabled
16331660 (sendBatchedGossip peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops)
1634− broadcastOfflineGossip t validation.pr.peerServers expiredLocal)
1661+ broadcastOfflineGossip t pr.peerServers expiredLocal)
16351662 yield (respond {token: token, expiresAt: expiry})
16361663 else do
1637− forkBroadcastOffline t validation.pr
1664+ forkBroadcastOffline t pr
16381665 yield (respond {token: "", expiresAt: 0})
16391666
16401667-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor
@@ -1671,16 +1698,16 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
16711698 else do
16721699 effectiveMsgId <- if messageId != "" then yield messageId else genToken
16731700 let clampedTs = clampTimestamp timestamp t
1701+ pr <- maybePrune t
16741702 result <- atomic do
1675− pr <- maybePrune t
16761703 withSessionAuth authorization t
1677− (yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}, senderRateLimited: False {}})
1704+ (yield {send: SendFailed {status: MsUnauthorized {}}, senderRateLimited: False {}, globalCapReached: False {}})
16781705 (\senderKey -> do
16791706 messages <- *messages
16801707 ackedIds <- *ackedIds
16811708 deliveredIds <- *deliveredIds
16821709 if messageId != "" && (hasMessage messageId to messages || hasMessage messageId to ackedIds || hasMessage messageId to deliveredIds)
1683− then yield {pr: pr, send: SendDuplicate {}, senderRateLimited: False {}}
1710+ then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}}
16841711 else do
16851712 sendRates <- *sendRates
16861713 let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
@@ -1688,7 +1715,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
16881715 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
16891716 maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
16901717 if rateResult.limited
1691− then yield {pr: pr, send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}}
1718+ then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}, globalCapReached: False {}}
16921719 else do
16931720 set *sendRates = rateResult.entries
16941721 recvRates <- *recvRates
@@ -1697,13 +1724,14 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
16971724 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
16981725 maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
16991726 if recvResult.limited
1700− then yield {pr: pr, send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}}
1727+ then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}}
17011728 else do
17021729 set *recvRates = recvResult.entries
1703− if count messages >= maxMessages
1704− then yield {pr: pr, send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}}
1705− else if countMessagesFor messages to t >= maxMessagesPerRecipient
1706− then yield {pr: pr, send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}}
1730+ let msgStats = messageCounts messages to t
1731+ if msgStats.total >= maxMessages
1732+ then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: True {}}
1733+ else if msgStats.forKey >= maxMessagesPerRecipient
1734+ then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}}
17071735 else do
17081736 presence <- *presence
17091737 seqCounter <- *seqCounter
@@ -1725,9 +1753,11 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
17251753 let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}
17261754 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
17271755 set *forwards = union forwards cappedFwds)
1728− yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}})
1756+ yield {send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}, globalCapReached: False {}})
17291757 when result.senderRateLimited
17301758 (println ("sendMessage: rate limited (" ++ show maxSendRatePerKey ++ "/s per key)"))
1759+ when result.globalCapReached
1760+ (println ("sendMessage: global message cap reached (" ++ show maxMessages ++ "), rejecting"))
17311761 case result.send of
17321762 SendDone {delivery, remoteServers} ->
17331763 fork (do
@@ -1738,14 +1768,14 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
17381768 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
17391769 fromServer: serverHostN, authorization: federationAuth})
17401770 case fwdResult of
1741− Err _ -> do
1742− println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)
1771+ Err e -> do
1772+ println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
17431773 recordFailure server
17441774 Ok _ -> do
17451775 removeForward effectiveMsgId server
17461776 clearFailure server))
1747− broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
1748− _ -> forkBroadcastOffline t result.pr
1777+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
1778+ _ -> forkBroadcastOffline t pr
17491779 yield (case result.send of
17501780 SendFailed {status} -> respond {status: status, messageId: ""}
17511781 SendDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId}
@@ -1761,10 +1791,10 @@ handleLookupPresence = \pubkey authorization respond -> do
17611791 globalRateLimited <- checkGlobalRate t
17621792 if globalRateLimited then yield (respond {server: "", authorized: False {}})
17631793 else do
1794+ pr <- maybePrune t
17641795 result <- atomic do
1765− pr <- maybePrune t
17661796 withSessionAuth authorization t
1767− (yield {pr: pr, server: "", authorized: False {}})
1797+ (yield {server: "", authorized: False {}})
17681798 (\senderKey -> do
17691799 lookupRates <- *lookupRates
17701800 let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == senderKey)
@@ -1773,15 +1803,15 @@ handleLookupPresence = \pubkey authorization respond -> do
17731803 maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
17741804 set *lookupRates = rateResult.entries
17751805 if rateResult.limited
1776− then yield {pr: pr, server: "", authorized: True {}}
1806+ then yield {server: "", authorized: True {}}
17771807 else do
17781808 presence <- *presence
17791809 let foundServer = bestServerForKey (presenceForKey presence pubkey t)
17801810 -- Mask remote server names: clients only need to know local/remote/unknown,
17811811 -- not which specific peer server a key is homed on (prevents metadata leakage)
17821812 let maskedServer = if foundServer == "" then "" else if isLocalServer foundServer then foundServer else "remote"
1783− yield {pr: pr, server: maskedServer, authorized: True {}})
1784− forkBroadcastOffline t result.pr
1813+ yield {server: maskedServer, authorized: True {}})
1814+ forkBroadcastOffline t pr
17851815 yield (respond {server: result.server, authorized: result.authorized})
17861816
17871817-- POST /federation/gossip: Receive presence gossip from a peer server.
@@ -1798,12 +1828,12 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do
17981828 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
17991829 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
18001830 let deduped = mergeByPubkeyServer [] filteredOnline
1831+ pr <- maybePrune t
18011832 gossipResult <- atomic do
1802− pr <- maybePrune t
18031833 presence <- *presence
18041834 let gm = mergeGossipPresence presence deduped trustedOffline fromServer t
18051835 set *presence = gm.finalPresence
1806− yield {gm: gm, pr: pr}
1836+ yield {gm: gm}
18071837 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
18081838 when (not gossipResult.gm.withinCapacity)
18091839 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
@@ -1820,7 +1850,7 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do
18201850 fork (do
18211851 when (cappedHops > 1 && federationEnabled && count freshEvents > 0) (
18221852 sendBatchedGossip (gossipTargets gossipResult.gm.gossipPeerServers fromServer) freshEvents (cappedHops - 1))
1823− broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal
1853+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal
18241854 when (federationEnabled && count newlyOnlineRemote > 0)
18251855 (forwardQueuedMessages newlyOnlineRemote t))
18261856 yield (respond {ok: True {}, error: ""}))
@@ -1832,40 +1862,45 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
18321862 t <- now
18331863 withFederationAuth authorization t respond fromServer (not (isLocalServer originServer) && fromServer == originServer) (do
18341864 let clampedTs = clampTimestamp timestamp t
1865+ pr <- maybePrune t
18351866 result <- atomic do
1836− pr <- maybePrune t
18371867 messages <- *messages
18381868 ackedIds <- *ackedIds
18391869 deliveredIds <- *deliveredIds
18401870 if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds
1841− then yield {pr: pr, forward: ForwardDuplicate {}}
1842− else if count messages >= maxMessages
1843− then yield {pr: pr, forward: ForwardRejected {}}
1844− else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1845− then yield {pr: pr, forward: ForwardRejected {}}
1846− else do
1847− recvRates <- *recvRates
1848− let recvResult = checkRateWindow recvRates (\r -> r.pubkey == toKey)
1849− (\ts -> {pubkey: toKey, windowStart: ts, count: 1})
1850− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1851− maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
1852− if recvResult.limited
1853− then yield {pr: pr, forward: ForwardRejected {}}
1854− else do
1855− set *recvRates = recvResult.entries
1856− seqCounter <- *seqCounter
1857− let seqTs = nextSeqTs t seqCounter
1858− set *seqCounter = seqTs
1859− let updatedMessages = union messages
1860− [{messageId: messageId, toKey: toKey,
1861− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1862− set *messages = updatedMessages
1863− signalMessageForKey toKey seqTs
1864− yield {pr: pr, forward: ForwardStored {}}
1871+ then yield {forward: ForwardDuplicate {}, globalCapReached: False {}}
1872+ else do
1873+ let msgStats = messageCounts messages toKey t
1874+ if msgStats.total >= maxMessages
1875+ then yield {forward: ForwardRejected {}, globalCapReached: True {}}
1876+ else if msgStats.forKey >= maxMessagesPerRecipient
1877+ then yield {forward: ForwardRejected {}, globalCapReached: False {}}
1878+ else do
1879+ recvRates <- *recvRates
1880+ let recvResult = checkRateWindow recvRates (\r -> r.pubkey == toKey)
1881+ (\ts -> {pubkey: toKey, windowStart: ts, count: 1})
1882+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1883+ maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
1884+ if recvResult.limited
1885+ then yield {forward: ForwardRejected {}, globalCapReached: False {}}
1886+ else do
1887+ set *recvRates = recvResult.entries
1888+ seqCounter <- *seqCounter
1889+ let seqTs = nextSeqTs t seqCounter
1890+ set *seqCounter = seqTs
1891+ let updatedMessages = union messages
1892+ [{messageId: messageId, toKey: toKey,
1893+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1894+ set *messages = updatedMessages
1895+ signalMessageForKey toKey seqTs
1896+ yield {forward: ForwardStored {}, globalCapReached: False {}}
18651897 case result.forward of
1866− ForwardRejected {} -> println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
1898+ ForwardRejected {} ->
1899+ if result.globalCapReached
1900+ then println ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
1901+ else println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
18671902 _ -> yield {}
1868− forkBroadcastOffline t result.pr
1903+ forkBroadcastOffline t pr
18691904 yield (case result.forward of
18701905 ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}
18711906 _ -> respond {ok: True {}, error: ""}))
@@ -1877,8 +1912,8 @@ handleFederationAck = \messageIds originServer fromServer authorization respond
18771912 t <- now
18781913 withFederationAuth authorization t respond fromServer (isLocalServer originServer && not (isLocalServer fromServer)) (do
18791914 let dedupedIds = dedupMessageIds messageIds
1880− pruneResult <- atomic do
1881− pr <- maybePrune t
1915+ pr <- maybePrune t
1916+ atomic do
18821917 forwards <- *forwards
18831918 let ackedForwards = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
18841919 let updatedForwards = forwardsExcluding forwards fromServer dedupedIds
@@ -1891,8 +1926,7 @@ handleFederationAck = \messageIds originServer fromServer authorization respond
18911926 where none presence (\p -> p.pubkey == f.toKey && isLocalServer p.server && p.expiresAt > t)
18921927 yield f
18931928 set *messages = do m <- messages; where none removable (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m
1894− yield pr
1895− forkBroadcastOffline t pruneResult
1929+ forkBroadcastOffline t pr
18961930 yield (respond {ok: True {}, error: ""}))
18971931
18981932-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).