@@ -107,6 +107,10 @@ federationFetchTimeoutMs : Int<Ms>
107federationFetchTimeoutMs = 30000 -- 30 seconds; consider server unresponsive if fetch hasn't returned
107federationFetchTimeoutMs = 30000 -- 30 seconds; consider server unresponsive if fetch hasn't returned
108maxFederationInFlight = 100
108maxFederationInFlight = 100
110+-- Federation latency metric window
111+federationLatencyWindowMs : Int<Ms>
112+federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
110-- Background task intervals
114-- Background task intervals
111backgroundPruneInterval : Int<Ms>
115backgroundPruneInterval : Int<Ms>
112backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
116backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
@@ -274,7 +278,24 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
274*sessionBucket14 : [Session]
278*sessionBucket14 : [Session]
275*sessionBucket15 : [Session]
279*sessionBucket15 : [Session]
277−*messages : [Message]
281+-- Messages sharded into 16 buckets by first hex char of recipient pubkey (toKey).
282+-- Hot-path operations (poll, send, ack) scan only one bucket (~1/16 of messages).
283+*messageBucket0 : [Message]
284+*messageBucket1 : [Message]
285+*messageBucket2 : [Message]
286+*messageBucket3 : [Message]
287+*messageBucket4 : [Message]
288+*messageBucket5 : [Message]
289+*messageBucket6 : [Message]
290+*messageBucket7 : [Message]
291+*messageBucket8 : [Message]
292+*messageBucket9 : [Message]
293+*messageBucket10 : [Message]
294+*messageBucket11 : [Message]
295+*messageBucket12 : [Message]
296+*messageBucket13 : [Message]
297+*messageBucket14 : [Message]
298+*messageBucket15 : [Message]
279*totalMessageCount : Int
300*totalMessageCount : Int
@@ -374,6 +395,12 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
374-- Server instance generation (epoch timestamp, for cursor staleness detection)
395-- Server instance generation (epoch timestamp, for cursor staleness detection)
375*serverGeneration : Int
396*serverGeneration : Int
398+-- Timestamp when drain federation ops completed (0 if in progress or not draining)
399+*drainCompletedAt : Timestamp
401+-- Last time federation latency metrics were reset (for windowed averages)
402+*federationLatencyResetAt : Timestamp
377-- ============================================
404-- ============================================
379-- ============================================
406-- ============================================
@@ -444,7 +471,7 @@ route HealthApi where
444-- Returns counts of in-memory collections (requires federation or maintenance secret)
471-- Returns counts of in-memory collections (requires federation or maintenance secret)
445route StatsApi where
472route StatsApi where
446 GET /stats headers {authorization: Text}
473 GET /stats headers {authorization: Text}
447− -> {sessions: Int, sessionsMax: Int, messages: Int, messagesMax: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pruneCount: Int, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, federationEnabled: Bool, recentGossip: Int, totalMessageCount: Int, recipientCounts: Int, federationInFlight: Int, messagesPct: Int, presencePct: Int, forwardsPct: Int, sessionsPct: Int, federationAvgLatencyMs: Int, federationMaxLatencyMs: Int, federationCallCount: Int, ok: Bool}
474+ -> {sessions: Int, sessionsMax: Int, messages: Int, messagesMax: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pruneCount: Int, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, federationEnabled: Bool, recentGossip: Int, totalMessageCount: Int, recipientCounts: Int, federationInFlight: Int, messagesPct: Int, presencePct: Int, forwardsPct: Int, sessionsPct: Int, federationAvgLatencyMs: Int, federationMaxLatencyMs: Int, federationCallCount: Int, drainCompletedAt: Timestamp, ok: Bool}
449 GET /stats/peers headers {authorization: Text}
476 GET /stats/peers headers {authorization: Text}
450 -> {peers: [PeerStatus], ok: Bool}
477 -> {peers: [PeerStatus], ok: Bool}
@@ -683,6 +710,9 @@ isBadServerName = \sn ->
683 || containsSuffix ".home" sn
710 || containsSuffix ".home" sn
684 || containsSuffix ".lan" sn
711 || containsSuffix ".lan" sn
685 || containsSuffix ".intranet" sn
712 || containsSuffix ".intranet" sn
713+ || containsSuffix ".test" sn
714+ || containsSuffix ".example" sn
715+ || containsSuffix ".invalid" sn
687-- Look up a non-expired session matching the given token (constant-time comparison).
717-- Look up a non-expired session matching the given token (constant-time comparison).
688-- Caller passes the appropriate session bucket (sharded by token prefix).
718-- Caller passes the appropriate session bucket (sharded by token prefix).
@@ -841,6 +871,100 @@ transformAllSessionBuckets = \f -> do
841 b14 <- *sessionBucket14; set *sessionBucket14 = f b14
871 b14 <- *sessionBucket14; set *sessionBucket14 = f b14
842 b15 <- *sessionBucket15; set *sessionBucket15 = f b15
872 b15 <- *sessionBucket15; set *sessionBucket15 = f b15
874+-- ============================================
875+-- Message Bucket Helpers
876+-- ============================================
877+-- Messages are sharded into 16 buckets by the first hex char of the recipient pubkey.
878+-- This reduces poll/send/ack scans from O(n) to O(n/16) on the hot path.
880+readMessageBucket : Int -> IO [Message]
881+readMessageBucket = \b ->
882+ if b == 0 then do s <- *messageBucket0; yield s
883+ else if b == 1 then do s <- *messageBucket1; yield s
884+ else if b == 2 then do s <- *messageBucket2; yield s
885+ else if b == 3 then do s <- *messageBucket3; yield s
886+ else if b == 4 then do s <- *messageBucket4; yield s
887+ else if b == 5 then do s <- *messageBucket5; yield s
888+ else if b == 6 then do s <- *messageBucket6; yield s
889+ else if b == 7 then do s <- *messageBucket7; yield s
890+ else if b == 8 then do s <- *messageBucket8; yield s
891+ else if b == 9 then do s <- *messageBucket9; yield s
892+ else if b == 10 then do s <- *messageBucket10; yield s
893+ else if b == 11 then do s <- *messageBucket11; yield s
894+ else if b == 12 then do s <- *messageBucket12; yield s
895+ else if b == 13 then do s <- *messageBucket13; yield s
896+ else if b == 14 then do s <- *messageBucket14; yield s
897+ else do s <- *messageBucket15; yield s
899+writeMessageBucket : Int -> [Message] -> IO ()
900+writeMessageBucket = \b msgs ->
901+ if b == 0 then set *messageBucket0 = msgs
902+ else if b == 1 then set *messageBucket1 = msgs
903+ else if b == 2 then set *messageBucket2 = msgs
904+ else if b == 3 then set *messageBucket3 = msgs
905+ else if b == 4 then set *messageBucket4 = msgs
906+ else if b == 5 then set *messageBucket5 = msgs
907+ else if b == 6 then set *messageBucket6 = msgs
908+ else if b == 7 then set *messageBucket7 = msgs
909+ else if b == 8 then set *messageBucket8 = msgs
910+ else if b == 9 then set *messageBucket9 = msgs
911+ else if b == 10 then set *messageBucket10 = msgs
912+ else if b == 11 then set *messageBucket11 = msgs
913+ else if b == 12 then set *messageBucket12 = msgs
914+ else if b == 13 then set *messageBucket13 = msgs
915+ else if b == 14 then set *messageBucket14 = msgs
916+ else set *messageBucket15 = msgs
918+-- Read all 16 message buckets (for cold-path operations: federation ack, forward candidates, stats)
919+readAllMessages : IO [Message]
921+ m0 <- *messageBucket0
922+ m1 <- *messageBucket1
923+ m2 <- *messageBucket2
924+ m3 <- *messageBucket3
925+ m4 <- *messageBucket4
926+ m5 <- *messageBucket5
927+ m6 <- *messageBucket6
928+ m7 <- *messageBucket7
929+ m8 <- *messageBucket8
930+ m9 <- *messageBucket9
931+ m10 <- *messageBucket10
932+ m11 <- *messageBucket11
933+ m12 <- *messageBucket12
934+ m13 <- *messageBucket13
935+ m14 <- *messageBucket14
936+ m15 <- *messageBucket15
937+ yield (union m0 (union m1 (union m2 (union m3 (union m4 (union m5 (union m6 (union m7 (union m8 (union m9 (union m10 (union m11 (union m12 (union m13 (union m14 m15)))))))))))))))
939+-- Apply a filter/transform to every message bucket
940+transformAllMessageBuckets : ([Message] -> [Message]) -> IO ()
941+transformAllMessageBuckets = \f -> do
942+ b0 <- *messageBucket0; set *messageBucket0 = f b0
943+ b1 <- *messageBucket1; set *messageBucket1 = f b1
944+ b2 <- *messageBucket2; set *messageBucket2 = f b2
945+ b3 <- *messageBucket3; set *messageBucket3 = f b3
946+ b4 <- *messageBucket4; set *messageBucket4 = f b4
947+ b5 <- *messageBucket5; set *messageBucket5 = f b5
948+ b6 <- *messageBucket6; set *messageBucket6 = f b6
949+ b7 <- *messageBucket7; set *messageBucket7 = f b7
950+ b8 <- *messageBucket8; set *messageBucket8 = f b8
951+ b9 <- *messageBucket9; set *messageBucket9 = f b9
952+ b10 <- *messageBucket10; set *messageBucket10 = f b10
953+ b11 <- *messageBucket11; set *messageBucket11 = f b11
954+ b12 <- *messageBucket12; set *messageBucket12 = f b12
955+ b13 <- *messageBucket13; set *messageBucket13 = f b13
956+ b14 <- *messageBucket14; set *messageBucket14 = f b14
957+ b15 <- *messageBucket15; set *messageBucket15 = f b15
959+-- Prune a single message bucket and return kept count + per-recipient counts.
960+-- Since messages are sharded by recipient pubkey, each bucket's recipient counts
961+-- are disjoint from other buckets — simple union merges them.
962+pruneMessageBucketCounting = \b messageCutoff -> atomic do
963+ bucket <- readMessageBucket b
964+ let kept = do m <- bucket; where m.receivedAt > messageCutoff; yield m
965+ writeMessageBucket b kept
966+ yield {count: count kept, recipients: recomputeRecipientCounts kept}
844-- ============================================
968-- ============================================
845-- Recipient Count Helpers
969-- Recipient Count Helpers
846-- ============================================
970-- ============================================
@@ -1140,16 +1264,31 @@ pruneExpired = \t -> do
1140 challenges <- *challenges
1264 challenges <- *challenges
1141 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
1265 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
1142 transformAllSessionBuckets (\bucket -> do x <- bucket; where x.expiresAt > t; yield x)
1266 transformAllSessionBuckets (\bucket -> do x <- bucket; where x.expiresAt > t; yield x)
1143− -- Prune messages (largest collection, isolated for minimal lock time)
1267+ -- Prune messages across all 16 buckets. Each bucket prune is atomic (short lock).
1144− prunedMessages <- atomic do
1268+ -- Recipient counts are rebuilt incrementally per-bucket — since messages are sharded
1269+ -- by recipient pubkey, each bucket's counts are disjoint, so simple union merges them.
1145− let messageCutoff = t - unackedMessageMaxAge
1270+ let messageCutoff = t - unackedMessageMaxAge
1146− messages <- *messages
1271+ r0 <- pruneMessageBucketCounting 0 messageCutoff
1147− let prunedMessages = do m <- messages; where m.receivedAt > messageCutoff; yield m
1272+ r1 <- pruneMessageBucketCounting 1 messageCutoff
1148− set *messages = prunedMessages
1273+ r2 <- pruneMessageBucketCounting 2 messageCutoff
1274+ r3 <- pruneMessageBucketCounting 3 messageCutoff
1275+ r4 <- pruneMessageBucketCounting 4 messageCutoff
1276+ r5 <- pruneMessageBucketCounting 5 messageCutoff
1277+ r6 <- pruneMessageBucketCounting 6 messageCutoff
1278+ r7 <- pruneMessageBucketCounting 7 messageCutoff
1279+ r8 <- pruneMessageBucketCounting 8 messageCutoff
1280+ r9 <- pruneMessageBucketCounting 9 messageCutoff
1281+ r10 <- pruneMessageBucketCounting 10 messageCutoff
1282+ r11 <- pruneMessageBucketCounting 11 messageCutoff
1283+ r12 <- pruneMessageBucketCounting 12 messageCutoff
1284+ r13 <- pruneMessageBucketCounting 13 messageCutoff
1285+ r14 <- pruneMessageBucketCounting 14 messageCutoff
1286+ r15 <- pruneMessageBucketCounting 15 messageCutoff
1287+ let prunedTotal = r0.count + r1.count + r2.count + r3.count + r4.count + r5.count + r6.count + r7.count + r8.count + r9.count + r10.count + r11.count + r12.count + r13.count + r14.count + r15.count
1288+ let prunedRecipients = union r0.recipients (union r1.recipients (union r2.recipients (union r3.recipients (union r4.recipients (union r5.recipients (union r6.recipients (union r7.recipients (union r8.recipients (union r9.recipients (union r10.recipients (union r11.recipients (union r12.recipients (union r13.recipients (union r14.recipients r15.recipients))))))))))))))
1149− set *totalMessageCount = count prunedMessages
1290+ set *totalMessageCount = prunedTotal
1150− yield prunedMessages
1151− -- Recompute recipient counts from pruned messages (outside main message lock)
1152− atomic do set *recipientMsgCounts = recomputeRecipientCounts prunedMessages
1291+ set *recipientMsgCounts = prunedRecipients
1153 -- Prune federation state
1292 -- Prune federation state
1154 atomic do
1293 atomic do
1155 let retainCutoff = t - messageMaxAge
1294 let retainCutoff = t - messageMaxAge
@@ -1183,9 +1322,11 @@ pruneExpired = \t -> do
1183 set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r
1322 set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r
1184 recentGossip <- *recentGossip
1323 recentGossip <- *recentGossip
1185 set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
1324 set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
1186− -- Clean up stale in-flight entries (hung fetches that never completed)
1325+ -- Clean up stale in-flight entries: remove zeroed entries and reset leaked slots
1326+ -- where the forked task died without calling clearFederationInFlight.
1327+ -- Uses 3x timeout threshold to allow the skip-hung-server logic to take effect first.
1187 inFlight <- *federationInFlight
1328 inFlight <- *federationInFlight
1188− set *federationInFlight = do r <- inFlight; where r.count > 0; yield r
1329+ set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < federationFetchTimeoutMs * 3); yield r
1189 yield presenceResult
1330 yield presenceResult
@@ -1407,12 +1548,12 @@ retryPendingForwards = \pending -> do
1407-- When a remote key comes online, create forward entries for queued messages and send them
1548-- When a remote key comes online, create forward entries for queued messages and send them
1408forwardQueuedMessages = \targets t -> do
1549forwardQueuedMessages = \targets t -> do
1409 newForwards <- atomic do
1550 newForwards <- atomic do
1410− messages <- *messages
1551+ allMessages <- readAllMessages
1411 forwards <- *forwards
1552 forwards <- *forwards
1412 presence <- *presence
1553 presence <- *presence
1413 let localOnline = localOnlineKeys presence t
1554 let localOnline = localOnlineKeys presence t
1414 let remoteTargets = remoteOnlyTargets targets localOnline
1555 let remoteTargets = remoteOnlyTargets targets localOnline
1415− let candidates = forwardCandidates remoteTargets messages forwards t
1556+ let candidates = forwardCandidates remoteTargets allMessages forwards t
1416 let capped = takeRelation (maxForwardsEntries - count forwards) candidates
1557 let capped = takeRelation (maxForwardsEntries - count forwards) candidates
1417 let updatedForwards = union forwards capped
1558 let updatedForwards = union forwards capped
1418 set *forwards = updatedForwards
1559 set *forwards = updatedForwards
@@ -1549,6 +1690,15 @@ pruneAndRetry = \t -> do
1549 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []
1690 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []
1550 when shouldRefresh (set *lastPresenceRefresh = t)
1691 when shouldRefresh (set *lastPresenceRefresh = t)
1551 yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
1692 yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
1693+ -- Reset federation latency metrics periodically for windowed averages (prevents
1694+ -- lifetime-cumulative average from masking current conditions)
1696+ resetAt <- *federationLatencyResetAt
1697+ when (t - resetAt >= federationLatencyWindowMs) (do
1698+ set *federationLatencyTotal = 0
1699+ set *federationLatencyCount = 0
1700+ set *federationLatencyMax = 0
1701+ set *federationLatencyResetAt = t)
1552 let result = {pendingForwards: pendingForwards, pendingAcks: ackState.pendingAcks, pendingOverflow: ackState.pendingOverflow, extendedActive: refreshResult.extendedActive, nearExpiryActive: refreshResult.nearExpiryActive, refreshPeers: refreshResult.refreshPeers, shouldRefresh: refreshResult.shouldRefresh}
1702 let result = {pendingForwards: pendingForwards, pendingAcks: ackState.pendingAcks, pendingOverflow: ackState.pendingOverflow, extendedActive: refreshResult.extendedActive, nearExpiryActive: refreshResult.nearExpiryActive, refreshPeers: refreshResult.refreshPeers, shouldRefresh: refreshResult.shouldRefresh}
1553 -- Fork federation I/O so hung peers don't block next prune cycle
1703 -- Fork federation I/O so hung peers don't block next prune cycle
1554 fork (do
1704 fork (do
@@ -1621,12 +1771,13 @@ readMessageSignal = \pubkey ->
1621-- and create ack entries for federated messages (to notify origin servers).
1771-- and create ack entries for federated messages (to notify origin servers).
1622-- Returns pending acks and any evicted entries that exceeded ackedIds capacity.
1772-- Returns pending acks and any evicted entries that exceeded ackedIds capacity.
1623ackPolledMessages = \pubkey cursor t -> do
1773ackPolledMessages = \pubkey cursor t -> do
1624− messages <- *messages
1774+ let b = pubkeyBucket pubkey
1775+ bucket <- readMessageBucket b
1625− let toAck = do m <- messages; where m.toKey == pubkey && m.receivedAt <= cursor; yield m
1776+ let toAck = do m <- bucket; where m.toKey == pubkey && m.receivedAt <= cursor; yield m
1626 let ackCount = count toAck
1777 let ackCount = count toAck
1627 if ackCount > 0
1778 if ackCount > 0
1629− set *messages = diff messages toAck
1780+ writeMessageBucket b (diff bucket toAck)
1630 totalCount <- *totalMessageCount
1781 totalCount <- *totalMessageCount
1631 set *totalMessageCount = totalCount - ackCount
1782 set *totalMessageCount = totalCount - ackCount
1632 recipientCounts <- *recipientMsgCounts
1783 recipientCounts <- *recipientMsgCounts
@@ -1644,8 +1795,8 @@ ackPolledMessages = \pubkey cursor t -> do
1644 else yield {pendingAcks: [], evictedEntries: []}
1795 else yield {pendingAcks: [], evictedEntries: []}
1646pollNewEvents = \pubkey clampedCursor t -> do
1797pollNewEvents = \pubkey clampedCursor t -> do
1647− messages <- *messages
1798+ bucket <- readMessageBucket (pubkeyBucket pubkey)
1648− let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages messages pubkey clampedCursor t))
1799+ let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages bucket pubkey clampedCursor t))
1649 let evts = do
1800 let evts = do
1650 m <- limitedMsgs
1801 m <- limitedMsgs
1651 yield {id: m.messageId, eventType: "message",
1802 yield {id: m.messageId, eventType: "message",
@@ -1704,8 +1855,8 @@ pollOnce = \token cursor clientIp -> do
1704 pollResult <- do
1855 pollResult <- do
1705 -- Phase 1: check for existing messages (atomic read)
1856 -- Phase 1: check for existing messages (atomic read)
1706 initialResult <- atomic do
1857 initialResult <- atomic do
1707− messages <- *messages
1858+ bucket <- readMessageBucket (pubkeyBucket s.pubkey)
1708− if count (pollableMessages messages s.pubkey s.clampedCursor t) > 0
1859+ if count (pollableMessages bucket s.pubkey s.clampedCursor t) > 0
1710 r <- pollNewEvents s.pubkey s.clampedCursor t
1861 r <- pollNewEvents s.pubkey s.clampedCursor t
1711 yield (Just {value: {events: r.events, cursor: r.cursor}})
1862 yield (Just {value: {events: r.events, cursor: r.cursor}})
@@ -1923,9 +2074,10 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
1923 validation <- atomic do
2074 validation <- atomic do
1924 challenges <- *challenges
2075 challenges <- *challenges
1925 let hasMatch = challengeMatch challenges pubkey challenge clientIp t
2076 let hasMatch = challengeMatch challenges pubkey challenge clientIp t
1926− -- Only consume challenge on IP match (rate limiter handles brute-force from same IP;
2077+ -- Consume only the matched challenge on IP match (rate limiter handles brute-force from same IP;
1927− -- not consuming on IP mismatch prevents remote attackers from burning challenges)
2078+ -- not consuming on IP mismatch prevents remote attackers from burning challenges).
2079+ -- Other challenges for the same pubkey are preserved (concurrent auth from multiple devices).
1928− when hasMatch (set *challenges = do c <- challenges; where c.pubkey != pubkey; yield c)
2080+ when hasMatch (set *challenges = challengesExcluding challenges pubkey challenge)
1929 if not hasMatch || not sigValid
2081 if not hasMatch || not sigValid
1930 then yield {valid: False {}}
2082 then yield {valid: False {}}
1931 else yield {valid: True {}}
2083 else yield {valid: True {}}
@@ -2020,11 +2172,12 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2020 -- Reject self-sends (sender == recipient wastes a mailbox slot)
2172 -- Reject self-sends (sender == recipient wastes a mailbox slot)
2021 if senderKey == to then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}}
2173 if senderKey == to then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}}
2023− messages <- *messages
2175+ let msgBucket = pubkeyBucket to
2176+ bucket <- readMessageBucket msgBucket
2024 ackedIds <- *ackedIds
2177 ackedIds <- *ackedIds
2025 deliveredIds <- *deliveredIds
2178 deliveredIds <- *deliveredIds
2026− -- Dedup check (only when messageId is non-empty; skips O(n) scan otherwise)
2179+ -- Dedup check (only when messageId is non-empty; scans only the recipient's bucket)
2027− let isDup = checkMessageDup messages messageId to || (messageId != "" && (hasMessage messageId to ackedIds || hasMessage messageId to deliveredIds))
2180+ let isDup = checkMessageDup bucket messageId to || (messageId != "" && (hasMessage messageId to ackedIds || hasMessage messageId to deliveredIds))
2028 if isDup
2181 if isDup
2029 then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}}
2182 then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}}
@@ -2060,9 +2213,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2060 seqCounter <- *seqCounter
2213 seqCounter <- *seqCounter
2061 let seqTs = nextSeqTs t seqCounter
2214 let seqTs = nextSeqTs t seqCounter
2062 set *seqCounter = seqTs
2215 set *seqCounter = seqTs
2063− set *messages = union messages
2216+ writeMessageBucket msgBucket (union bucket
2064 [{messageId: effectiveMsgId, toKey: to,
2217 [{messageId: effectiveMsgId, toKey: to,
2065− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
2218+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])
2066 set *totalMessageCount = totalMsgCount + 1
2219 set *totalMessageCount = totalMsgCount + 1
2067 set *recipientMsgCounts = adjustRecipientCount recipientCounts to 1
2220 set *recipientMsgCounts = adjustRecipientCount recipientCounts to 1
2068 signalMessageForKey to seqTs
2221 signalMessageForKey to seqTs
@@ -2150,6 +2303,16 @@ handleRevoke = \targetToken authorization clientIp respond -> do
2150 withSessionAuth authorization t clientIp
2303 withSessionAuth authorization t clientIp
2151 (yield (respond {ok: False {}, authorized: False {}}))
2304 (yield (respond {ok: False {}, authorized: False {}}))
2152 (\callerPubkey -> do
2305 (\callerPubkey -> do
2306+ revokeLimited <- atomic do
2307+ challengeRates <- *challengeRates
2308+ let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == callerPubkey)
2309+ (\ts -> {pubkey: callerPubkey, windowStart: ts, count: 1})
2310+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
2311+ maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
2312+ set *challengeRates = rateResult.entries
2313+ yield rateResult.limited
2314+ if revokeLimited then yield (respond {ok: False {}, authorized: True {}})
2153 let targetBucket = tokenBucket targetToken
2316 let targetBucket = tokenBucket targetToken
2154 removed <- atomic do
2317 removed <- atomic do
2155 sessions <- readSessionBucket targetBucket
2318 sessions <- readSessionBucket targetBucket
@@ -2210,12 +2373,19 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do
2210handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer authorization respond -> do
2373handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer authorization respond -> do
2211 t <- now
2374 t <- now
2212 withFederationAuth authorization t respond fromServer (not (isLocalServer originServer) && fromServer == originServer) (do
2375 withFederationAuth authorization t respond fromServer (not (isLocalServer originServer) && fromServer == originServer) (do
2376+ -- Replay protection: reject forwards with timestamps older than messageMaxAge.
2377+ -- Legitimate queued messages have timestamps within this window. Replayed captures
2378+ -- become stale once dedup state (messages + ackedIds + deliveredIds) is pruned (~30 days).
2379+ if t - timestamp > messageMaxAge
2380+ then yield (respond {ok: False {}, error: "stale_timestamp"})
2213 let clampedTs = clampTimestamp timestamp t
2382 let clampedTs = clampTimestamp timestamp t
2214 result <- atomic do
2383 result <- atomic do
2215− messages <- *messages
2384+ let msgBucket = pubkeyBucket toKey
2385+ bucket <- readMessageBucket msgBucket
2216 ackedIds <- *ackedIds
2386 ackedIds <- *ackedIds
2217 deliveredIds <- *deliveredIds
2387 deliveredIds <- *deliveredIds
2218− let isDup = checkMessageDup messages messageId toKey || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds
2388+ let isDup = checkMessageDup bucket messageId toKey || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds
2219 if isDup
2389 if isDup
2220 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}}
2390 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}}
@@ -2240,10 +2410,9 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
2240 seqCounter <- *seqCounter
2410 seqCounter <- *seqCounter
2241 let seqTs = nextSeqTs t seqCounter
2411 let seqTs = nextSeqTs t seqCounter
2242 set *seqCounter = seqTs
2412 set *seqCounter = seqTs
2243− let updatedMessages = union messages
2413+ writeMessageBucket msgBucket (union bucket
2244 [{messageId: messageId, toKey: toKey,
2414 [{messageId: messageId, toKey: toKey,
2245− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
2415+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}])
2246− set *messages = updatedMessages
2247 set *totalMessageCount = totalMsgCount + 1
2416 set *totalMessageCount = totalMsgCount + 1
2248 set *recipientMsgCounts = adjustRecipientCount recipientCounts toKey 1
2417 set *recipientMsgCounts = adjustRecipientCount recipientCounts toKey 1
2249 signalMessageForKey toKey seqTs
2418 signalMessageForKey toKey seqTs
@@ -2272,7 +2441,8 @@ handleFederationAck = \messageIds originServer fromServer authorization respond
2272 let ackedForwards = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
2441 let ackedForwards = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
2273 let updatedForwards = forwardsExcluding forwards fromServer dedupedIds
2442 let updatedForwards = forwardsExcluding forwards fromServer dedupedIds
2274 set *forwards = updatedForwards
2443 set *forwards = updatedForwards
2275− messages <- *messages
2444+ -- Read all message buckets (cold path: federation ack handles at most maxMessageIds entries)
2445+ allMessages <- readAllMessages
2276 presence <- *presence
2446 presence <- *presence
2277 let removable = do
2447 let removable = do
2278 f <- ackedForwards
2448 f <- ackedForwards
@@ -2280,20 +2450,18 @@ handleFederationAck = \messageIds originServer fromServer authorization respond
2280 where none presence (\p -> p.pubkey == f.toKey && isLocalServer p.server && p.expiresAt > t)
2450 where none presence (\p -> p.pubkey == f.toKey && isLocalServer p.server && p.expiresAt > t)
2282 -- Only remove messages stored long enough for local clients to have polled them (grace period)
2452 -- Only remove messages stored long enough for local clients to have polled them (grace period)
2283− let pastGrace = do f <- removable; where any messages (\m -> m.messageId == f.messageId && m.toKey == f.toKey && t - m.receivedAt >= ackGracePeriod); yield f
2453+ let pastGrace = do f <- removable; where any allMessages (\m -> m.messageId == f.messageId && m.toKey == f.toKey && t - m.receivedAt >= ackGracePeriod); yield f
2284− -- Single-pass fold: filter messages and count removals together
2454+ let removed = count pastGrace
2285− let msgResult = fold (\acc m ->
2455+ -- Remove matched messages from their buckets (pastGrace is small, so scanning all 16 buckets is fast)
2456+ when (removed > 0) (do
2457+ transformAllMessageBuckets (\bucket ->
2286− if none pastGrace (\f -> m.messageId == f.messageId && m.toKey == f.toKey)
2458+ do m <- bucket; where none pastGrace (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m)
2287− then {kept: union acc.kept [m], removed: acc.removed}
2288− else {kept: acc.kept, removed: acc.removed + 1}
2289− ) {kept: [], removed: 0} messages
2290− set *messages = msgResult.kept
2291− totalMsgCount <- *totalMessageCount
2459+ totalMsgCount <- *totalMessageCount
2292− set *totalMessageCount = totalMsgCount - msgResult.removed
2460+ set *totalMessageCount = totalMsgCount - removed
2293− -- Adjust per-recipient counts for removed messages
2461+ -- Adjust per-recipient counts for removed messages
2294− recipientCounts <- *recipientMsgCounts
2462+ recipientCounts <- *recipientMsgCounts
2295− let adjustedCounts = fold (\acc f -> adjustRecipientCount acc f.toKey (0 - 1)) recipientCounts pastGrace
2463+ let adjustedCounts = fold (\acc f -> adjustRecipientCount acc f.toKey (0 - 1)) recipientCounts pastGrace
2296− set *recipientMsgCounts = adjustedCounts
2464+ set *recipientMsgCounts = adjustedCounts)
2297 yield (respond {ok: True {}, error: ""}))
2465 yield (respond {ok: True {}, error: ""}))
2299-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
2467-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
@@ -2317,10 +2485,14 @@ handleDrain = \authorization respond -> do
2317 let pendingForwardCount = count forwards
2485 let pendingForwardCount = count forwards
2318 let pendingAckCount = count ackedIds
2486 let pendingAckCount = count ackedIds
2319 -- Fork gossip so drain responds immediately (hung peers can't block the response).
2487 -- Fork gossip so drain responds immediately (hung peers can't block the response).
2320− -- Operator should allow a few seconds after response before killing the process.
2488+ -- Operator can poll /stats for drainCompletedAt > 0 to know when it's safe to stop.
2489+ atomic do set *drainCompletedAt = 0
2321− fork (when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
2491+ when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
2322− let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
2492+ let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
2323− sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops))
2493+ sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
2495+ atomic do set *drainCompletedAt = t2)
2324 println ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
2496 println ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
2325 when (pendingForwardCount > 0)
2497 when (pendingForwardCount > 0)
2326 (println ("drain: warning: " ++ show pendingForwardCount ++ " pending forwards will not be retried"))
2498 (println ("drain: warning: " ++ show pendingForwardCount ++ " pending forwards will not be retried"))
@@ -2362,12 +2534,12 @@ handleHealth = \respond -> do
2362 forwards <- *forwards
2534 forwards <- *forwards
2363 sessionCount <- countAllSessions
2535 sessionCount <- countAllSessions
2364 activePolls <- *activePolls
2536 activePolls <- *activePolls
2365− messages <- *messages
2537+ tmc <- *totalMessageCount
2366 lastFedSuccess <- *lastFederationSuccessAt
2538 lastFedSuccess <- *lastFederationSuccessAt
2367 forwardFailures <- *forwardFailures
2539 forwardFailures <- *forwardFailures
2368 let warmup = t - sa < healthPruneMaxAge
2540 let warmup = t - sa < healthPruneMaxAge
2369 -- Capacity pressure: flag when any capped collection exceeds 90% full
2541 -- Capacity pressure: flag when any capped collection exceeds 90% full
2370− let capacityOk = count messages * 10 <= maxMessages * 9
2542+ let capacityOk = tmc * 10 <= maxMessages * 9
2371 && count presence * 10 <= maxPresenceEntries * 9
2543 && count presence * 10 <= maxPresenceEntries * 9
2372 && count forwards * 10 <= maxForwardsEntries * 9
2544 && count forwards * 10 <= maxForwardsEntries * 9
2373 && sessionCount * 10 <= maxSessions * 9
2545 && sessionCount * 10 <= maxSessions * 9
@@ -2386,7 +2558,7 @@ handleHealth = \respond -> do
2386 capacityHealthy: result.capacityOk,
2558 capacityHealthy: result.capacityOk,
2387 federationHealthy: result.federationOk})
2559 federationHealthy: result.federationOk})
2389−zeroStats = {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pruneCount: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, ok: False {}}
2561+zeroStats = {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pruneCount: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, drainCompletedAt: 0, ok: False {}}
2391handleStats = \authorization respond -> do
2563handleStats = \authorization respond -> do
2392 let authorized = verifyMaintenanceAuth authorization
2564 let authorized = verifyMaintenanceAuth authorization
@@ -2394,9 +2566,10 @@ handleStats = \authorization respond -> do
2395 t <- now
2567 t <- now
2396 -- Read all refs atomically for a consistent snapshot, count outside to minimize lock time
2568 -- Read all refs atomically for a consistent snapshot, count outside to minimize lock time
2569+ -- Read all refs atomically for a consistent snapshot. Messages use totalMessageCount
2570+ -- (maintained counter) instead of scanning all 16 message buckets.
2397 snapshot <- atomic do
2571 snapshot <- atomic do
2398 sc <- countAllSessions
2572 sc <- countAllSessions
2400 p <- *presence
2573 p <- *presence
2401 f <- *forwards
2574 f <- *forwards
2402 c <- *challenges
2575 c <- *challenges
@@ -2425,14 +2598,15 @@ handleStats = \authorization respond -> do
2425 flt <- *federationLatencyTotal
2598 flt <- *federationLatencyTotal
2426 flc <- *federationLatencyCount
2599 flc <- *federationLatencyCount
2427 flm <- *federationLatencyMax
2600 flm <- *federationLatencyMax
2601+ dca <- *drainCompletedAt
2428− yield {sc: sc, m: m, p: p, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, sr: sr, cr: cr, fr: fr, grc: grc, ir: ir, lr: lr, rr: rr, pc: pc, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, rg: rg, tmc: tmc, rc: rc, fi: fi, flt: flt, flc: flc, flm: flm}
2602+ yield {sc: sc, p: p, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, sr: sr, cr: cr, fr: fr, grc: grc, ir: ir, lr: lr, rr: rr, pc: pc, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, rg: rg, tmc: tmc, rc: rc, fi: fi, flt: flt, flc: flc, flm: flm, dca: dca}
2429− let result = {sessions: snapshot.sc, messages: count snapshot.m, presence: count snapshot.p, forwards: count snapshot.f, challenges: count snapshot.c, ackedIds: count snapshot.a, ackOverflow: count snapshot.ao, deliveredIds: count snapshot.d, forwardFailures: count snapshot.ff, activePolls: count snapshot.ap, sendRates: count snapshot.sr, challengeRates: count snapshot.cr, federationRates: count snapshot.fr, globalRateCount: snapshot.grc, ipRates: count snapshot.ir, lookupRates: count snapshot.lr, recvRates: count snapshot.rr, pruneCount: snapshot.pc, heartbeatCount: snapshot.hc, uptime: t - snapshot.sa, lastPruneAt: snapshot.lp, lastHeartbeatAt: snapshot.lh, draining: snapshot.dr, recentGossip: count snapshot.rg, totalMessageCount: snapshot.tmc, recipientCounts: count snapshot.rc, federationInFlight: count snapshot.fi, federationLatencyTotal: snapshot.flt, federationLatencyCount: snapshot.flc, federationLatencyMax: snapshot.flm}
2603+ let result = {sessions: snapshot.sc, messages: snapshot.tmc, presence: count snapshot.p, forwards: count snapshot.f, challenges: count snapshot.c, ackedIds: count snapshot.a, ackOverflow: count snapshot.ao, deliveredIds: count snapshot.d, forwardFailures: count snapshot.ff, activePolls: count snapshot.ap, sendRates: count snapshot.sr, challengeRates: count snapshot.cr, federationRates: count snapshot.fr, globalRateCount: snapshot.grc, ipRates: count snapshot.ir, lookupRates: count snapshot.lr, recvRates: count snapshot.rr, pruneCount: snapshot.pc, heartbeatCount: snapshot.hc, uptime: t - snapshot.sa, lastPruneAt: snapshot.lp, lastHeartbeatAt: snapshot.lh, draining: snapshot.dr, recentGossip: count snapshot.rg, totalMessageCount: snapshot.tmc, recipientCounts: count snapshot.rc, federationInFlight: count snapshot.fi, federationLatencyTotal: snapshot.flt, federationLatencyCount: snapshot.flc, federationLatencyMax: snapshot.flm, drainCompletedAt: snapshot.dca}
2430 let msgPct = if maxMessages > 0 then result.messages * 100 / maxMessages else 0
2604 let msgPct = if maxMessages > 0 then result.messages * 100 / maxMessages else 0
2431 let presPct = if maxPresenceEntries > 0 then result.presence * 100 / maxPresenceEntries else 0
2605 let presPct = if maxPresenceEntries > 0 then result.presence * 100 / maxPresenceEntries else 0
2432 let fwdPct = if maxForwardsEntries > 0 then result.forwards * 100 / maxForwardsEntries else 0
2606 let fwdPct = if maxForwardsEntries > 0 then result.forwards * 100 / maxForwardsEntries else 0
2433 let sessPct = if maxSessions > 0 then result.sessions * 100 / maxSessions else 0
2607 let sessPct = if maxSessions > 0 then result.sessions * 100 / maxSessions else 0
2434 let fedAvg = if result.federationLatencyCount > 0 then result.federationLatencyTotal / result.federationLatencyCount else 0
2608 let fedAvg = if result.federationLatencyCount > 0 then result.federationLatencyTotal / result.federationLatencyCount else 0
2435− yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, draining: result.draining, federationEnabled: federationEnabled, recentGossip: result.recentGossip, totalMessageCount: result.totalMessageCount, recipientCounts: result.recipientCounts, federationInFlight: result.federationInFlight, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, ok: True {}})
2609+ yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, draining: result.draining, federationEnabled: federationEnabled, recentGossip: result.recentGossip, totalMessageCount: result.totalMessageCount, recipientCounts: result.recipientCounts, federationInFlight: result.federationInFlight, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, drainCompletedAt: result.drainCompletedAt, ok: True {}})
2436 else yield (respond zeroStats)
2610 else yield (respond zeroStats)
2438handleStatsPeers = \authorization respond -> do
2612handleStatsPeers = \authorization respond -> do
@@ -2451,12 +2625,12 @@ handleStatsKey = \pubkey authorization respond -> do
2452 t <- now
2626 t <- now
2453 result <- atomic do
2627 result <- atomic do
2454− messages <- *messages
2628+ keyBucket <- readMessageBucket (pubkeyBucket pubkey)
2455 presence <- *presence
2629 presence <- *presence
2456 allSessions <- readAllSessions
2630 allSessions <- readAllSessions
2457 forwards <- *forwards
2631 forwards <- *forwards
2458 activePolls <- *activePolls
2632 activePolls <- *activePolls
2459− let keyMessages = count (do m <- messages; where m.toKey == pubkey; yield m)
2633+ let keyMessages = count (do m <- keyBucket; where m.toKey == pubkey; yield m)
2460 let keyPresence = do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield {server: p.server, expiresAt: p.expiresAt}
2634 let keyPresence = do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield {server: p.server, expiresAt: p.expiresAt}
2461 let keySessions = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
2635 let keySessions = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
2462 let keyForwards = count (do f <- forwards; where f.toKey == pubkey; yield f)
2636 let keyForwards = count (do f <- forwards; where f.toKey == pubkey; yield f)