@@ -857,6 +857,12 @@ maxPeerLatencyEntries = 100
857-- Populated during ackPolledMessages and handleAckMessages for locally-originated messages.
857-- Populated during ackPolledMessages and handleAckMessages for locally-originated messages.
858*senderDeliveries : [SenderDelivery]
858*senderDeliveries : [SenderDelivery]
860+-- Tracks how many sender delivery receipts / expiration notifications were dropped
861+-- due to capacity eviction before the sender could poll them. Reset per-sender on poll
862+-- (drained with the notifications). Lets clients know they may have missed receipts.
863+*droppedDeliveryReceipts : Int
864+*droppedExpirationNotifs : Int
860-- Runtime maintenance secret: when non-empty, used instead of compile-time maintenanceSecret.
866-- Runtime maintenance secret: when non-empty, used instead of compile-time maintenanceSecret.
861-- Set via /maintenance/rotate-secret to enable zero-downtime secret rotation.
867-- Set via /maintenance/rotate-secret to enable zero-downtime secret rotation.
862-- Empty = use compile-time maintenanceSecret (default).
868-- Empty = use compile-time maintenanceSecret (default).
@@ -947,7 +953,7 @@ route AuthApi where
947route ClientApi where
953route ClientApi where
948 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
954 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
949 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
955 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
950− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}]}
956+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}], droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int}
952 /messages
958 /messages
953 -- Send an E2E encrypted message to a recipient pubkey.
959 -- Send an E2E encrypted message to a recipient pubkey.
@@ -1043,6 +1049,13 @@ route FederationApi where
1044-- Federation callback verification: the receiving server calls back to the sender
1050-- Federation callback verification: the receiving server calls back to the sender
1045-- to confirm a request token is genuine. No shared secret needed.
1051-- to confirm a request token is genuine. No shared secret needed.
1052+-- NOTE: Tokens are not bound to the caller's IP. A MITM that intercepts a token
1053+-- can replay it from a different IP within the 30-second TTL (verificationTokenExpiry).
1054+-- Mitigations: (1) tokens are single-use (consumed on first verify), (2) per-IP rate
1055+-- limiting (verifyIpRates) constrains brute-force probing, (3) the verify cache
1056+-- (verificationCacheMs=2s) limits the replay window after the first legitimate verify.
1057+-- Binding to IP would break federation through NAT/proxy topologies where the
1058+-- verifying server's egress IP differs from the initial request's source IP.
1046route FederationVerifyApi where
1059route FederationVerifyApi where
1047 POST {token: Text, scope: Text} /federation/verify headers {clientIp: Text} -> {ok: Bool} = VerifyToken
1060 POST {token: Text, scope: Text} /federation/verify headers {clientIp: Text} -> {ok: Bool} = VerifyToken
@@ -2223,9 +2236,13 @@ pruneMessageBucket = \b t -> atomic do
2223 when (count localExpired > 0) (do
2236 when (count localExpired > 0) (do
2224 se <- *senderExpirations
2237 se <- *senderExpirations
2225 let combined = union se localExpired
2238 let combined = union se localExpired
2239+ let overCap = count combined > maxSenderExpirations
2226− set *senderExpirations = if count combined > maxSenderExpirations
2240+ set *senderExpirations = if overCap
2227 then evictOneOldest combined (\e -> e.expiredAt)
2241 then evictOneOldest combined (\e -> e.expiredAt)
2228− else combined)
2242+ else combined
2244+ den <- *droppedExpirationNotifs
2245+ set *droppedExpirationNotifs = den + 1))
2229 else yield {}
2246 else yield {}
2231-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
2248-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
@@ -3487,6 +3504,11 @@ sendBatchedGossipToServer = \server allEvents hops ->
3487 Err e -> logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
3504 Err e -> logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
3488 Ok _ -> yield {})
3505 Ok _ -> yield {})
3507+-- Sends gossip events to each peer server. Per-server seenBy filtering happens inside
3508+-- sendBatchedGossipToServer: if all events for a server are already in its seenBy list,
3509+-- the function returns immediately without making an HTTP call. Pre-filtering the target
3510+-- list here would save only function call overhead and complicate the seenBy set logic
3511+-- (seenBy is per-event, not per-batch), so we rely on the per-server early-exit instead.
3490sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}
3512sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}
3491sendBatchedGossip = \peerServers allEvents hops ->
3513sendBatchedGossip = \peerServers allEvents hops ->
3492 sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents hops)
3514 sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents hops)
@@ -3769,7 +3791,10 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
3769 -- since pubkey was already verified by authenticateSession above.
3791 -- since pubkey was already verified by authenticateSession above.
3770 let needsRefresh = any sessions (\s -> s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 2)
3792 let needsRefresh = any sessions (\s -> s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 2)
3771 when needsRefresh
3793 when needsRefresh
3794+ -- Pubkey pre-filter on safeCompare: only runs constant-time comparison for
3795+ -- sessions matching the authenticated pubkey (typically 1-2 per bucket),
3796+ -- skipping the expensive safeCompare for all unrelated sessions (~600/bucket).
3772− (writeSessionBucket bucket (do s <- sessions; yield (if s.pubkey == pubkey && safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip, createdAt: s.createdAt} else s)))
3797+ (writeSessionBucket bucket (do s <- sessions; yield (if s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 2 && safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip, createdAt: s.createdAt} else s)))
3773 onAuth pubkey
3798 onAuth pubkey
3775-- Check if a pubkey has no remaining sessions and, if so, clean up local presence.
3800-- Check if a pubkey has no remaining sessions and, if so, clean up local presence.
@@ -4511,9 +4536,13 @@ ackPolledMessages = \pubkey cursor t -> do
4511 when (count localDelivered > 0) (do
4536 when (count localDelivered > 0) (do
4512 sd <- *senderDeliveries
4537 sd <- *senderDeliveries
4513 let combined = union sd localDelivered
4538 let combined = union sd localDelivered
4539+ let overCap = count combined > maxSenderDeliveries
4514− set *senderDeliveries = if count combined > maxSenderDeliveries
4540+ set *senderDeliveries = if overCap
4515 then evictOneOldest combined (\d -> d.deliveredAt)
4541 then evictOneOldest combined (\d -> d.deliveredAt)
4516− else combined)
4542+ else combined
4544+ ddr <- *droppedDeliveryReceipts
4545+ set *droppedDeliveryReceipts = ddr + 1))
4517 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
4546 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
4518 else yield {pendingAcks: [], evictedEntries: []}
4547 else yield {pendingAcks: [], evictedEntries: []}
@@ -4976,10 +5005,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
4976 -- Cursor staleness: cursor is older than message retention, so messages were missed.
5005 -- Cursor staleness: cursor is older than message retention, so messages were missed.
4977 -- Different from cursorReset (server restart): staleness means client was offline too long.
5006 -- Different from cursorReset (server restart): staleness means client was offline too long.
4978 let cursorStale = effectiveCursor > 0 && t - effectiveCursor > unackedMessageMaxAge
5007 let cursorStale = effectiveCursor > 0 && t - effectiveCursor > unackedMessageMaxAge
4979− if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: []})
5008+ if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0})
4981 globalRateLimited <- checkGlobalRate t
5010 globalRateLimited <- checkGlobalRate t
4982− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: []})
5011+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0})
4984 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
5013 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
4985 startPresenceVersion <- *presenceVersion
5014 startPresenceVersion <- *presenceVersion
@@ -5030,7 +5059,18 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
5030 yield capped
5059 yield capped
5031 -- Poll slot exhaustion: signal clients to back off for one poll cycle
5060 -- Poll slot exhaustion: signal clients to back off for one poll cycle
5032 let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
5061 let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
5062+ -- Read and reset drop counters so clients know if they missed receipts/expirations.
5063+ -- Counters are global (not per-key) — a non-zero value means *some* receipts were
5064+ -- dropped due to capacity, though not necessarily for this sender. This is acceptable:
5065+ -- the client treats it as advisory ("some receipts may have been lost") rather than precise.
5066+ drops <- if not result.authorized then yield {dr: 0, de: 0}
5068+ dr <- *droppedDeliveryReceipts
5069+ de <- *droppedExpirationNotifs
5070+ when (dr > 0) (set *droppedDeliveryReceipts = 0)
5071+ when (de > 0) (set *droppedExpirationNotifs = 0)
5072+ yield {dr: dr, de: de}
5033− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, hasMore: result.hasMore, cursorStale: cursorStale, messageExpirations: expirations, messageDeliveries: deliveries})
5073+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, hasMore: result.hasMore, cursorStale: cursorStale, messageExpirations: expirations, messageDeliveries: deliveries, droppedDeliveryReceipts: drops.dr, droppedExpirationNotifs: drops.de})
5035-- Core send logic: dedup, rate limiting, store message, create forwards.
5075-- Core send logic: dedup, rate limiting, store message, create forwards.
5036-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
5076-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
@@ -5726,9 +5766,13 @@ fedDeliveryReceipts = \bucketIds forwards t -> case headOf bucketIds of
5726 when (count deliveries > 0) (atomic do
5766 when (count deliveries > 0) (atomic do
5727 sd <- *senderDeliveries
5767 sd <- *senderDeliveries
5728 let combined = union sd deliveries
5768 let combined = union sd deliveries
5769+ let overCap = count combined > maxSenderDeliveries
5729− set *senderDeliveries = if count combined > maxSenderDeliveries
5770+ set *senderDeliveries = if overCap
5730 then evictOneOldest combined (\d -> d.deliveredAt)
5771 then evictOneOldest combined (\d -> d.deliveredAt)
5731− else combined)
5772+ else combined
5774+ ddr <- *droppedDeliveryReceipts
5775+ set *droppedDeliveryReceipts = ddr + 1))
5732 fedDeliveryReceipts rest forwards t
5776 fedDeliveryReceipts rest forwards t
5734-- POST /federation/ack: Peer server confirms delivery of messages we originated.
5778-- POST /federation/ack: Peer server confirms delivery of messages we originated.
@@ -5874,42 +5918,52 @@ cancelBatchInner = \entries fromServer t acc ->
5874 -- Collect all entries targeting the same bucket
5918 -- Collect all entries targeting the same bucket
5875 let sameBucket = do e <- entries; where pubkeyBucket e.toKey == b; yield e
5919 let sameBucket = do e <- entries; where pubkeyBucket e.toKey == b; yield e
5876 let otherBuckets = do e <- entries; where pubkeyBucket e.toKey != b; yield e
5920 let otherBuckets = do e <- entries; where pubkeyBucket e.toKey != b; yield e
5877− -- Process the entire bucket group in one atomic block
5921+ -- Process the entire bucket group in one atomic block.
5922+ -- Read buckets once upfront and thread state through the recursion.
5878− results <- atomic do cancelBucketGroup sameBucket fromServer b t []
5923+ results <- atomic do
5924+ bucket <- readMessageBucket b
5925+ fwdBkt <- readFwdBucket b
5926+ nonceBkt <- readNonceBucket b
5927+ rc <- readRcBucket b
5928+ r <- cancelBucketGroupInner sameBucket fromServer b t [] bucket fwdBkt nonceBkt rc 0 0
5929+ -- Write back all modified buckets once after processing all entries
5930+ writeMessageBucket b r.bucket
5931+ writeFwdBucket b r.fwdBkt
5932+ writeNonceBucket b r.nonceBkt
5933+ writeRcBucket b r.rc
5934+ -- Batch-adjust shared counters once (not per-entry)
5935+ when (r.msgRemoved > 0) (do
5936+ tc <- *totalMessageCount
5937+ set *totalMessageCount = max 0 (tc - r.msgRemoved))
5938+ when (r.fwdRemoved > 0) (do
5939+ tfc <- *totalForwardCount
5940+ set *totalForwardCount = max 0 (tfc - r.fwdRemoved))
5879 cancelBatchInner otherBuckets fromServer t (union acc results)
5942 cancelBatchInner otherBuckets fromServer t (union acc results)
5881-- Process all cancel entries for a single bucket within one atomic block.
5944-- Process all cancel entries for a single bucket within one atomic block.
5882−-- Reads messageBucket/fwdBucket once and applies all cancels, avoiding redundant reads.
5945+-- Threads bucket state through recursion to avoid redundant per-entry reads.
5883−cancelBucketGroup = \entries fromServer b t acc ->
5946+-- Accumulates removal counts for batch counter adjustment after the loop.
5947+cancelBucketGroupInner = \entries fromServer b t acc bucket fwdBkt nonceBkt rc msgRemoved fwdRemoved ->
5884 case headOf entries of
5948 case headOf entries of
5885− Nothing {} -> yield acc
5949+ Nothing {} -> yield {acc: acc, bucket: bucket, fwdBkt: fwdBkt, nonceBkt: nonceBkt, rc: rc, msgRemoved: msgRemoved, fwdRemoved: fwdRemoved}
5886 Just {value: entry} -> do
5950 Just {value: entry} -> do
5887 let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
5951 let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
5888− bucket <- readMessageBucket b
5889 let toCancel = headOf (do m <- bucket; where m.messageId == entry.messageId && m.toKey == entry.toKey && m.originServer == fromServer; yield m)
5952 let toCancel = headOf (do m <- bucket; where m.messageId == entry.messageId && m.toKey == entry.toKey && m.originServer == fromServer; yield m)
5890− result <- case toCancel of
5953+ case toCancel of
5891− Nothing {} -> yield {cancelled: False {}}
5954+ Nothing {} ->
5955+ cancelBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, cancelled: False {}}]) bucket fwdBkt nonceBkt rc msgRemoved fwdRemoved
5892 Just {} -> do
5956 Just {} -> do
5893− writeMessageBucket b (do m <- bucket; where not (m.messageId == entry.messageId && m.toKey == entry.toKey); yield m)
5957+ let newBucket = do m <- bucket; where not (m.messageId == entry.messageId && m.toKey == entry.toKey); yield m
5894− tc <- *totalMessageCount
5895− set *totalMessageCount = max 0 (tc - 1)
5896− rc <- readRcBucket b
5897− writeRcBucket b (adjustRecipientCount rc entry.toKey (0 - 1))
5958+ let newRc = adjustRecipientCount rc entry.toKey (0 - 1)
5898− fwdBkt <- readFwdBucket b
5899− let keptFwds = do f <- fwdBkt; where not (f.messageId == entry.messageId && f.toKey == entry.toKey); yield f
5959+ let newFwdBkt = do f <- fwdBkt; where not (f.messageId == entry.messageId && f.toKey == entry.toKey); yield f
5900− writeFwdBucket b keptFwds
5901− let fwdRemoved = count fwdBkt - count keptFwds
5960+ let entryFwdRemoved = count fwdBkt - count newFwdBkt
5902− when (fwdRemoved > 0) (do
5903− tfc <- *totalForwardCount
5904− set *totalForwardCount = max 0 (tfc - fwdRemoved))
5905− -- Record cancel nonce for each cancelled message in the batch
5961+ -- Record cancel nonce for replay protection
5906− nonceBkt <- readNonceBucket b
5907 let updatedNonces = union nonceBkt [{messageId: entry.messageId, toKey: entry.toKey, seenAt: t}]
5962 let updatedNonces = union nonceBkt [{messageId: entry.messageId, toKey: entry.toKey, seenAt: t}]
5908− writeNonceBucket b (if count updatedNonces > maxNoncesPerBucket
5963+ let newNonceBkt = if count updatedNonces > maxNoncesPerBucket
5909 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5964 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5910− else updatedNonces)
5965+ else updatedNonces
5911− yield {cancelled: True {}}
5912− cancelBucketGroup rest fromServer b t (union acc [{messageId: entry.messageId, cancelled: result.cancelled}])
5966+ cancelBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, cancelled: True {}}]) newBucket newFwdBkt newNonceBkt newRc (msgRemoved + 1) (fwdRemoved + entryFwdRemoved)
5914-- POST /federation/forward-batch: Receive multiple forwarded messages in one request.
5968-- POST /federation/forward-batch: Receive multiple forwarded messages in one request.
5915-- Each message is processed independently (same logic as handleForwardMessage).
5969-- Each message is processed independently (same logic as handleForwardMessage).
@@ -5939,33 +5993,33 @@ forwardBatchInner = \entries fromServer t acc ->
5939 Nothing {} -> yield acc
5993 Nothing {} -> yield acc
5940 Just {value: first} -> do
5994 Just {value: first} -> do
5941 -- Pre-validate all entries: reject stale timestamps and untrusted relay origins.
5995 -- Pre-validate all entries: reject stale timestamps and untrusted relay origins.
5942− -- Reads global state once (knownPeers, blocked, allowed) outside atomic blocks.
5996+ -- Read global state once before the loop (not per-entry) to avoid redundant IO.
5997+ knownPeers <- *knownPeerServers
5998+ blocked <- *blockedPeers
5999+ allowed <- *allowedPeers
5943− validated <- prevalidateForwardBatch entries fromServer t [] []
6000+ let validated = prevalidateForwardBatch entries fromServer t knownPeers blocked allowed [] []
5944 -- Group validated entries by bucket
6001 -- Group validated entries by bucket
5945 forwardBatchByBucket validated.valid fromServer t (union acc validated.rejected)
6002 forwardBatchByBucket validated.valid fromServer t (union acc validated.rejected)
5947-- Pre-validate forward entries: reject stale timestamps and untrusted relay origins.
6004-- Pre-validate forward entries: reject stale timestamps and untrusted relay origins.
5948-- Returns {valid: entries to store, rejected: per-message results for failed entries}.
6005-- Returns {valid: entries to store, rejected: per-message results for failed entries}.
6006+-- Pure function: global state (knownPeers, blocked, allowed) is passed in by caller.
5949−prevalidateForwardBatch = \entries fromServer t validAcc rejectedAcc ->
6007+prevalidateForwardBatch = \entries fromServer t knownPeers blocked allowed validAcc rejectedAcc ->
5950 case headOf entries of
6008 case headOf entries of
5951− Nothing {} -> yield {valid: validAcc, rejected: rejectedAcc}
6009+ Nothing {} -> {valid: validAcc, rejected: rejectedAcc}
5952− Just {value: entry} -> do
6010+ Just {value: entry} ->
5953− let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
6011+ let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e in
5954 if isLocalServer entry.originServer || t - entry.timestamp > messageMaxAge
6012 if isLocalServer entry.originServer || t - entry.timestamp > messageMaxAge
5955− then prevalidateForwardBatch rest fromServer t validAcc (union rejectedAcc [{messageId: entry.messageId, ok: False {}}])
6013+ then prevalidateForwardBatch rest fromServer t knownPeers blocked allowed validAcc (union rejectedAcc [{messageId: entry.messageId, ok: False {}}])
5957 -- Relay origin verification (same as handleForwardMessage)
6015 -- Relay origin verification (same as handleForwardMessage)
5958− originOk <- if fromServer == entry.originServer then yield (True {})
6016+ let originOk = if fromServer == entry.originServer then True {}
5960− knownPeers <- *knownPeerServers
5961− blocked <- *blockedPeers
5962− allowed <- *allowedPeers
5963− yield (any knownPeers (\p -> p == entry.originServer)
6017+ else any knownPeers (\p -> p == entry.originServer)
5964− && not (isBlockedPeer entry.originServer blocked)
6018+ && not (isBlockedPeer entry.originServer blocked)
5965− && isAllowedByAllowlist entry.originServer allowed)
6019+ && isAllowedByAllowlist entry.originServer allowed
5966− if not originOk
6020+ in if not originOk
5967− then prevalidateForwardBatch rest fromServer t validAcc (union rejectedAcc [{messageId: entry.messageId, ok: False {}}])
6021+ then prevalidateForwardBatch rest fromServer t knownPeers blocked allowed validAcc (union rejectedAcc [{messageId: entry.messageId, ok: False {}}])
5968− else prevalidateForwardBatch rest fromServer t (union validAcc [entry]) rejectedAcc
6022+ else prevalidateForwardBatch rest fromServer t knownPeers blocked allowed (union validAcc [entry]) rejectedAcc
5970-- Group validated entries by pubkeyBucket and process each group in one atomic block.
6024-- Group validated entries by pubkeyBucket and process each group in one atomic block.
5971forwardBatchByBucket = \entries fromServer t acc ->
6025forwardBatchByBucket = \entries fromServer t acc ->
@@ -5975,66 +6029,79 @@ forwardBatchByBucket = \entries fromServer t acc ->
5975 let b = pubkeyBucket first.toKey
6029 let b = pubkeyBucket first.toKey
5976 let sameBucket = do e <- entries; where pubkeyBucket e.toKey == b; yield e
6030 let sameBucket = do e <- entries; where pubkeyBucket e.toKey == b; yield e
5977 let otherBuckets = do e <- entries; where pubkeyBucket e.toKey != b; yield e
6031 let otherBuckets = do e <- entries; where pubkeyBucket e.toKey != b; yield e
5978− -- Process the entire bucket group in one atomic block
6032+ -- Process the entire bucket group in one atomic block.
6033+ -- Read bucket-local state once upfront and thread through the recursion.
5979− results <- atomic do forwardBucketGroup sameBucket fromServer b t []
6034+ results <- atomic do
6035+ presenceBkt <- readPresenceBucket b
6036+ bucket <- readMessageBucket b
6037+ ackedBkt <- readAckedBucket b
6038+ delivBkt <- readDelivBucket b
6039+ nonceBkt <- readNonceBucket b
6040+ rc <- readRcBucket b
6041+ r <- forwardBucketGroupInner sameBucket fromServer b t [] presenceBkt bucket ackedBkt delivBkt nonceBkt rc 0
6042+ -- Write back modified bucket-local state once after processing all entries
6043+ writeMessageBucket b r.bucket
6044+ writeNonceBucket b r.nonceBkt
6045+ writeRcBucket b r.rc
6046+ -- Batch-adjust totalMessageCount once (not per-entry)
6047+ when (r.msgStored > 0) (do
6048+ tc <- *totalMessageCount
6049+ set *totalMessageCount = tc + r.msgStored)
5980 forwardBatchByBucket otherBuckets fromServer t (union acc results)
6051 forwardBatchByBucket otherBuckets fromServer t (union acc results)
5982-- Process all forward entries for a single bucket within one atomic block.
6053-- Process all forward entries for a single bucket within one atomic block.
5983−-- Reads presenceBucket/messageBucket/ackedBucket/deliveredBucket/nonceBucket once
6054+-- Threads bucket-local state through recursion to avoid redundant per-entry reads.
5984−-- and applies all forwards, avoiding redundant per-entry bucket reads.
6055+-- Shared state (recvRates, federationPairRates, seqCounter) is read per-entry since
5985−forwardBucketGroup = \entries fromServer b t acc ->
6056+-- it's modified by each successful store and affects subsequent entries.
6057+forwardBucketGroupInner = \entries fromServer b t acc presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored ->
5986 case headOf entries of
6058 case headOf entries of
5987− Nothing {} -> yield acc
6059+ Nothing {} -> yield {acc: acc, bucket: bucket, nonceBkt: nonceBkt, rc: rc, msgStored: msgStored}
5988 Just {value: entry} -> do
6060 Just {value: entry} -> do
5989 let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
6061 let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
5990 let clampedTs = clampTimestamp entry.timestamp t
6062 let clampedTs = clampTimestamp entry.timestamp t
5991− presenceBkt <- readPresenceBucket b
5992 let hasPresence = any presenceBkt (\p -> p.pubkey == entry.toKey && p.expiresAt > t)
6063 let hasPresence = any presenceBkt (\p -> p.pubkey == entry.toKey && p.expiresAt > t)
5993− result <- if not hasPresence then yield (False {})
6064+ if not hasPresence
6065+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
5995− bucket <- readMessageBucket b
5996− ackedBkt <- readAckedBucket b
5997− delivBkt <- readDelivBucket b
5998− nonceBkt <- readNonceBucket b
5999 let isDup = checkMessageDup bucket entry.messageId entry.toKey || hasMessage entry.messageId entry.toKey ackedBkt || hasMessage entry.messageId entry.toKey delivBkt || any nonceBkt (\n -> n.messageId == entry.messageId && n.toKey == entry.toKey)
6067 let isDup = checkMessageDup bucket entry.messageId entry.toKey || hasMessage entry.messageId entry.toKey ackedBkt || hasMessage entry.messageId entry.toKey delivBkt || any nonceBkt (\n -> n.messageId == entry.messageId && n.toKey == entry.toKey)
6000− if isDup then yield (True {})
6068+ if isDup
6069+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
6001− else do
6070+ else do
6002− totalMsgCount <- *totalMessageCount
6071+ totalMsgCount <- *totalMessageCount
6003− recipientCounts <- readRcBucket b
6004− let recipientCount = getRecipientCount entry.toKey recipientCounts
6072+ let recipientCount = getRecipientCount entry.toKey rc
6005− if totalMsgCount >= maxMessages || recipientCount >= maxMessagesPerRecipient
6073+ if totalMsgCount + msgStored >= maxMessages || recipientCount >= maxMessagesPerRecipient
6006− then yield (False {})
6074+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
6008− -- Per-recipient receive rate limiting (matches handleForwardMessage)
6009− recvRates <- *recvRates
6010− let recvResult = checkPubkeyRate recvRates entry.toKey maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
6011− if recvResult.limited then yield (False {})
6076+ -- Per-recipient receive rate limiting (matches handleForwardMessage)
6077+ recvRates <- *recvRates
6078+ let recvResult = checkPubkeyRate recvRates entry.toKey maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
6079+ if recvResult.limited
6080+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
6013− set *recvRates = recvResult.entries
6082+ set *recvRates = recvResult.entries
6014− -- Per-server-per-recipient rate limiting (matches handleForwardMessage)
6083+ -- Per-server-per-recipient rate limiting (matches handleForwardMessage)
6015− federationPairRates <- *federationPairRates
6084+ federationPairRates <- *federationPairRates
6016− let fedPairResult = checkServerRecipientRate federationPairRates fromServer entry.toKey maxFederationPairRatePerPair federationPairRateWindowMs maxFederationPairRateEntries t
6085+ let fedPairResult = checkServerRecipientRate federationPairRates fromServer entry.toKey maxFederationPairRatePerPair federationPairRateWindowMs maxFederationPairRateEntries t
6017− if fedPairResult.limited then yield (False {})
6086+ if fedPairResult.limited
6087+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
6018− else do
6088+ else do
6019− set *federationPairRates = fedPairResult.entries
6089+ set *federationPairRates = fedPairResult.entries
6020− seqCounter <- *seqCounter
6090+ seqCounter <- *seqCounter
6021− let seqTs = nextSeqTs t seqCounter
6091+ let seqTs = nextSeqTs t seqCounter
6022− set *seqCounter = seqTs
6092+ set *seqCounter = seqTs
6023− let fwdExpiresAt = if entry.expiresAt > 0 && entry.expiresAt > t then min entry.expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
6093+ let fwdExpiresAt = if entry.expiresAt > 0 && entry.expiresAt > t then min entry.expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
6024− let batchClampedPriority = clamp 0 maxMessagePriority entry.priority
6094+ let batchClampedPriority = clamp 0 maxMessagePriority entry.priority
6025− writeMessageBucket b (union bucket
6095+ let newBucket = union bucket
6026− [{messageId: entry.messageId, toKey: entry.toKey,
6096+ [{messageId: entry.messageId, toKey: entry.toKey,
6027− encryptedBlob: entry.encryptedBlob, timestamp: clampedTs, originServer: entry.originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: batchClampedPriority}])
6097+ encryptedBlob: entry.encryptedBlob, timestamp: clampedTs, originServer: entry.originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: batchClampedPriority}]
6028− set *totalMessageCount = totalMsgCount + 1
6029− writeRcBucket b (adjustRecipientCount recipientCounts entry.toKey 1)
6098+ let newRc = adjustRecipientCount rc entry.toKey 1
6030− signalMessageForKey entry.toKey seqTs
6099+ signalMessageForKey entry.toKey seqTs
6031− fwdNonces <- readNonceBucket b
6032− let updatedNonces = union fwdNonces [{messageId: entry.messageId, toKey: entry.toKey, seenAt: t}]
6100+ let updatedNonces = union nonceBkt [{messageId: entry.messageId, toKey: entry.toKey, seenAt: t}]
6033− writeNonceBucket b (if count updatedNonces > maxNoncesPerBucket
6101+ let newNonceBkt = if count updatedNonces > maxNoncesPerBucket
6034− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6102+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6035− else updatedNonces)
6103+ else updatedNonces
6036− yield (True {})
6104+ forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt newBucket ackedBkt delivBkt newNonceBkt newRc (msgStored + 1)
6037− forwardBucketGroup rest fromServer b t (union acc [{messageId: entry.messageId, ok: result}])
6039-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
6106-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
6040-- Auth checked once; each cancel processed independently. Partial success possible.
6107-- Auth checked once; each cancel processed independently. Partial success possible.
@@ -6781,7 +6848,8 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
6781 yield rateResult.limited
6848 yield rateResult.limited
6782 if ackRateLimited then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})
6849 if ackRateLimited then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})
6851+ -- Cap input size to prevent expensive scans from unbounded messageIds lists
6784− let dedupedIds = dedupMessageIds messageIds
6852+ let dedupedIds = dedupMessageIds (takeRelation maxMessageIds messageIds)
6785 result <- atomic do
6853 result <- atomic do
6786 let b = pubkeyBucket pubkey
6854 let b = pubkeyBucket pubkey
6787 bucket <- readMessageBucket b
6855 bucket <- readMessageBucket b
@@ -6808,9 +6876,13 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
6808 when (count localDelivered > 0) (do
6876 when (count localDelivered > 0) (do
6809 sd <- *senderDeliveries
6877 sd <- *senderDeliveries
6810 let combined = union sd localDelivered
6878 let combined = union sd localDelivered
6879+ let overCap = count combined > maxSenderDeliveries
6811− set *senderDeliveries = if count combined > maxSenderDeliveries
6880+ set *senderDeliveries = if overCap
6812 then evictOneOldest combined (\d -> d.deliveredAt)
6881 then evictOneOldest combined (\d -> d.deliveredAt)
6813− else combined)
6882+ else combined
6884+ ddr <- *droppedDeliveryReceipts
6885+ set *droppedDeliveryReceipts = ddr + 1))
6814 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
6886 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
6815 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}
6887 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}
6816 fork (when federationEnabled (do
6888 fork (when federationEnabled (do
@@ -6877,7 +6949,8 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
6877 yield rateResult.limited
6949 yield rateResult.limited
6878 if deleteRateLimited then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})
6950 if deleteRateLimited then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})
6952+ -- Cap input size to prevent expensive scans from unbounded messageIds lists
6880− let dedupedIds = dedupMessageIds messageIds
6953+ let dedupedIds = dedupMessageIds (takeRelation maxMessageIds messageIds)
6881 deleteResult <- atomic do
6954 deleteResult <- atomic do
6882 let b = pubkeyBucket pubkey
6955 let b = pubkeyBucket pubkey
6883 bucket <- readMessageBucket b
6956 bucket <- readMessageBucket b
@@ -6902,9 +6975,13 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
6902 when (count localDeleted > 0) (do
6975 when (count localDeleted > 0) (do
6903 sd <- *senderDeliveries
6976 sd <- *senderDeliveries
6904 let combinedDel = union sd localDeleted
6977 let combinedDel = union sd localDeleted
6978+ let overCapDel = count combinedDel > maxSenderDeliveries
6905− set *senderDeliveries = if count combinedDel > maxSenderDeliveries
6979+ set *senderDeliveries = if overCapDel
6906 then evictOneOldest combinedDel (\d -> d.deliveredAt)
6980 then evictOneOldest combinedDel (\d -> d.deliveredAt)
6907− else combinedDel)
6981+ else combinedDel
6982+ when overCapDel (do
6983+ ddr <- *droppedDeliveryReceipts
6984+ set *droppedDeliveryReceipts = ddr + 1))
6908 if count ackedFederated > 0 then do
6985 if count ackedFederated > 0 then do
6909 ackedBkt <- readAckedBucket b
6986 ackedBkt <- readAckedBucket b
6910 let combinedAcked = union ackedBkt ackedFederated
6987 let combinedAcked = union ackedBkt ackedFederated
@@ -7686,7 +7763,7 @@ serveOverloaded = \req -> case req of
7686 VerifyAuth {respond} -> respond {token: "", expiresAt: 0, error: "overloaded"}
7763 VerifyAuth {respond} -> respond {token: "", expiresAt: 0, error: "overloaded"}
7687 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}
7764 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}
7688 Logout {respond} -> respond {ok: False {}, authorized: False {}}
7765 Logout {respond} -> respond {ok: False {}, authorized: False {}}
7689− Poll {respond} -> respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: 0, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 1000, cursorReset: False {}, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: []}
7766+ Poll {respond} -> respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: 0, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 1000, cursorReset: False {}, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0}
7690 SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded"}
7767 SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded"}
7691 MessageStatus {respond} -> respond {status: "", authorized: False {}}
7768 MessageStatus {respond} -> respond {status: "", authorized: False {}}
7692 AckMessages {respond} -> respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "overloaded"}
7769 AckMessages {respond} -> respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "overloaded"}
@@ -7937,6 +8014,11 @@ backgroundSnapshot = do
7937 lastFp <- *lastSnapshotFingerprint
8014 lastFp <- *lastSnapshotFingerprint
7938 let stateChanged = lastFp == 0 || fingerprint != lastFp
8015 let stateChanged = lastFp == 0 || fingerprint != lastFp
7939 when (hasState && stateChanged && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do
8016 when (hasState && stateChanged && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do
8017+ -- NOTE: TOCTOU between the fingerprint/hasState check above and the atomic snapshot
8018+ -- below. State could change between the two reads (e.g., a message arrives, presence
8019+ -- is pruned). This is acceptable: worst case is one extra export (fingerprint changed
8020+ -- after we decided to export) or one skipped export (state became empty after we
8021+ -- decided to export). The next backgroundSnapshot cycle corrects either case.
7940 -- Single atomic read for consistent [SNAPSHOT_DATA] output (used for restore).
8022 -- Single atomic read for consistent [SNAPSHOT_DATA] output (used for restore).
7941 -- Previous per-collection reads caused inconsistencies (e.g., ackedIds referencing
8023 -- Previous per-collection reads caused inconsistencies (e.g., ackedIds referencing
7942 -- messageIds from a different snapshot window). The contention cost is acceptable
8024 -- messageIds from a different snapshot window). The contention cost is acceptable