@@ -857,6 +857,12 @@ maxPeerLatencyEntries = 100
857857-- Populated during ackPolledMessages and handleAckMessages for locally-originated messages.
858858*senderDeliveries : [SenderDelivery]
859859
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
865+
860866-- Runtime maintenance secret: when non-empty, used instead of compile-time maintenanceSecret.
861867-- Set via /maintenance/rotate-secret to enable zero-downtime secret rotation.
862868-- Empty = use compile-time maintenanceSecret (default).
@@ -947,7 +953,7 @@ route AuthApi where
947953route ClientApi where
948954 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
949955 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}
951957 = Poll
952958 /messages
953959 -- Send an E2E encrypted message to a recipient pubkey.
@@ -1043,6 +1049,13 @@ route FederationApi where
10431049
10441050-- Federation callback verification: the receiving server calls back to the sender
10451051-- 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.
10461059route FederationVerifyApi where
10471060 POST {token: Text, scope: Text} /federation/verify headers {clientIp: Text} -> {ok: Bool} = VerifyToken
10481061
@@ -2223,9 +2236,13 @@ pruneMessageBucket = \b t -> atomic do
22232236 when (count localExpired > 0) (do
22242237 se <- *senderExpirations
22252238 let combined = union se localExpired
2226− set *senderExpirations = if count combined > maxSenderExpirations
2239+ let overCap = count combined > maxSenderExpirations
2240+ set *senderExpirations = if overCap
22272241 then evictOneOldest combined (\e -> e.expiredAt)
2228− else combined)
2242+ else combined
2243+ when overCap (do
2244+ den <- *droppedExpirationNotifs
2245+ set *droppedExpirationNotifs = den + 1))
22292246 else yield {}
22302247
22312248-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
@@ -3487,6 +3504,11 @@ sendBatchedGossipToServer = \server allEvents hops ->
34873504 Err e -> logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
34883505 Ok _ -> yield {})
34893506
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.
34903512sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}
34913513sendBatchedGossip = \peerServers allEvents hops ->
34923514 sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents hops)
@@ -3769,7 +3791,10 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
37693791 -- since pubkey was already verified by authenticateSession above.
37703792 let needsRefresh = any sessions (\s -> s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 2)
37713793 when needsRefresh
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)))
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).
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)))
37733798 onAuth pubkey
37743799
37753800-- Check if a pubkey has no remaining sessions and, if so, clean up local presence.
@@ -4511,9 +4536,13 @@ ackPolledMessages = \pubkey cursor t -> do
45114536 when (count localDelivered > 0) (do
45124537 sd <- *senderDeliveries
45134538 let combined = union sd localDelivered
4514− set *senderDeliveries = if count combined > maxSenderDeliveries
4539+ let overCap = count combined > maxSenderDeliveries
4540+ set *senderDeliveries = if overCap
45154541 then evictOneOldest combined (\d -> d.deliveredAt)
4516− else combined)
4542+ else combined
4543+ when overCap (do
4544+ ddr <- *droppedDeliveryReceipts
4545+ set *droppedDeliveryReceipts = ddr + 1))
45174546 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
45184547 else yield {pendingAcks: [], evictedEntries: []}
45194548
@@ -4976,10 +5005,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
49765005 -- Cursor staleness: cursor is older than message retention, so messages were missed.
49775006 -- Different from cursorReset (server restart): staleness means client was offline too long.
49785007 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})
49805009 else do
49815010 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})
49835012 else do
49845013 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
49855014 startPresenceVersion <- *presenceVersion
@@ -5030,7 +5059,18 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
50305059 yield capped
50315060 -- Poll slot exhaustion: signal clients to back off for one poll cycle
50325061 let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
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})
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}
5067+ else atomic do
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}
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})
50345074
50355075-- Core send logic: dedup, rate limiting, store message, create forwards.
50365076-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
@@ -5726,9 +5766,13 @@ fedDeliveryReceipts = \bucketIds forwards t -> case headOf bucketIds of
57265766 when (count deliveries > 0) (atomic do
57275767 sd <- *senderDeliveries
57285768 let combined = union sd deliveries
5729− set *senderDeliveries = if count combined > maxSenderDeliveries
5769+ let overCap = count combined > maxSenderDeliveries
5770+ set *senderDeliveries = if overCap
57305771 then evictOneOldest combined (\d -> d.deliveredAt)
5731− else combined)
5772+ else combined
5773+ when overCap (do
5774+ ddr <- *droppedDeliveryReceipts
5775+ set *droppedDeliveryReceipts = ddr + 1))
57325776 fedDeliveryReceipts rest forwards t
57335777
57345778-- POST /federation/ack: Peer server confirms delivery of messages we originated.
@@ -5874,42 +5918,52 @@ cancelBatchInner = \entries fromServer t acc ->
58745918 -- Collect all entries targeting the same bucket
58755919 let sameBucket = do e <- entries; where pubkeyBucket e.toKey == b; yield e
58765920 let otherBuckets = do e <- entries; where pubkeyBucket e.toKey != b; yield e
5877− -- Process the entire bucket group in one atomic block
5878− results <- atomic do cancelBucketGroup sameBucket fromServer b t []
5921+ -- Process the entire bucket group in one atomic block.
5922+ -- Read buckets once upfront and thread state through the recursion.
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))
5941+ yield r.acc
58795942 cancelBatchInner otherBuckets fromServer t (union acc results)
58805943
58815944-- Process all cancel entries for a single bucket within one atomic block.
5882−-- Reads messageBucket/fwdBucket once and applies all cancels, avoiding redundant reads.
5883−cancelBucketGroup = \entries fromServer b t acc ->
5945+-- Threads bucket state through recursion to avoid redundant per-entry reads.
5946+-- Accumulates removal counts for batch counter adjustment after the loop.
5947+cancelBucketGroupInner = \entries fromServer b t acc bucket fwdBkt nonceBkt rc msgRemoved fwdRemoved ->
58845948 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}
58865950 Just {value: entry} -> do
58875951 let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
5888− bucket <- readMessageBucket b
58895952 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
5891− Nothing {} -> yield {cancelled: False {}}
5953+ case toCancel of
5954+ Nothing {} ->
5955+ cancelBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, cancelled: False {}}]) bucket fwdBkt nonceBkt rc msgRemoved fwdRemoved
58925956 Just {} -> do
5893− writeMessageBucket b (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))
5898− fwdBkt <- readFwdBucket b
5899− let keptFwds = 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
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
5906− nonceBkt <- readNonceBucket b
5957+ let newBucket = do m <- bucket; where not (m.messageId == entry.messageId && m.toKey == entry.toKey); yield m
5958+ let newRc = adjustRecipientCount rc entry.toKey (0 - 1)
5959+ let newFwdBkt = do f <- fwdBkt; where not (f.messageId == entry.messageId && f.toKey == entry.toKey); yield f
5960+ let entryFwdRemoved = count fwdBkt - count newFwdBkt
5961+ -- Record cancel nonce for replay protection
59075962 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
59095964 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5910− else updatedNonces)
5911− yield {cancelled: True {}}
5912− cancelBucketGroup rest fromServer b t (union acc [{messageId: entry.messageId, cancelled: result.cancelled}])
5965+ else updatedNonces
5966+ cancelBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, cancelled: True {}}]) newBucket newFwdBkt newNonceBkt newRc (msgRemoved + 1) (fwdRemoved + entryFwdRemoved)
59135967
59145968-- POST /federation/forward-batch: Receive multiple forwarded messages in one request.
59155969-- Each message is processed independently (same logic as handleForwardMessage).
@@ -5939,33 +5993,33 @@ forwardBatchInner = \entries fromServer t acc ->
59395993 Nothing {} -> yield acc
59405994 Just {value: first} -> do
59415995 -- Pre-validate all entries: reject stale timestamps and untrusted relay origins.
5942− -- Reads global state once (knownPeers, blocked, allowed) outside atomic blocks.
5943− validated <- prevalidateForwardBatch entries fromServer t [] []
5996+ -- Read global state once before the loop (not per-entry) to avoid redundant IO.
5997+ knownPeers <- *knownPeerServers
5998+ blocked <- *blockedPeers
5999+ allowed <- *allowedPeers
6000+ let validated = prevalidateForwardBatch entries fromServer t knownPeers blocked allowed [] []
59446001 -- Group validated entries by bucket
59456002 forwardBatchByBucket validated.valid fromServer t (union acc validated.rejected)
59466003
59476004-- Pre-validate forward entries: reject stale timestamps and untrusted relay origins.
59486005-- Returns {valid: entries to store, rejected: per-message results for failed entries}.
5949−prevalidateForwardBatch = \entries fromServer t validAcc rejectedAcc ->
6006+-- Pure function: global state (knownPeers, blocked, allowed) is passed in by caller.
6007+prevalidateForwardBatch = \entries fromServer t knownPeers blocked allowed validAcc rejectedAcc ->
59506008 case headOf entries of
5951− Nothing {} -> yield {valid: validAcc, rejected: rejectedAcc}
5952− Just {value: entry} -> do
5953− let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
6009+ Nothing {} -> {valid: validAcc, rejected: rejectedAcc}
6010+ Just {value: entry} ->
6011+ let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e in
59546012 if isLocalServer entry.originServer || t - entry.timestamp > messageMaxAge
5955− then prevalidateForwardBatch rest fromServer t validAcc (union rejectedAcc [{messageId: entry.messageId, ok: False {}}])
5956− else do
6013+ then prevalidateForwardBatch rest fromServer t knownPeers blocked allowed validAcc (union rejectedAcc [{messageId: entry.messageId, ok: False {}}])
6014+ else
59576015 -- Relay origin verification (same as handleForwardMessage)
5958− originOk <- if fromServer == entry.originServer then yield (True {})
5959− else do
5960− knownPeers <- *knownPeerServers
5961− blocked <- *blockedPeers
5962− allowed <- *allowedPeers
5963− yield (any knownPeers (\p -> p == entry.originServer)
5964− && not (isBlockedPeer entry.originServer blocked)
5965− && isAllowedByAllowlist entry.originServer allowed)
5966− if not originOk
5967− then prevalidateForwardBatch rest fromServer t validAcc (union rejectedAcc [{messageId: entry.messageId, ok: False {}}])
5968− else prevalidateForwardBatch rest fromServer t (union validAcc [entry]) rejectedAcc
6016+ let originOk = if fromServer == entry.originServer then True {}
6017+ else any knownPeers (\p -> p == entry.originServer)
6018+ && not (isBlockedPeer entry.originServer blocked)
6019+ && isAllowedByAllowlist entry.originServer allowed
6020+ in if not originOk
6021+ then prevalidateForwardBatch rest fromServer t knownPeers blocked allowed validAcc (union rejectedAcc [{messageId: entry.messageId, ok: False {}}])
6022+ else prevalidateForwardBatch rest fromServer t knownPeers blocked allowed (union validAcc [entry]) rejectedAcc
59696023
59706024-- Group validated entries by pubkeyBucket and process each group in one atomic block.
59716025forwardBatchByBucket = \entries fromServer t acc ->
@@ -5975,66 +6029,79 @@ forwardBatchByBucket = \entries fromServer t acc ->
59756029 let b = pubkeyBucket first.toKey
59766030 let sameBucket = do e <- entries; where pubkeyBucket e.toKey == b; yield e
59776031 let otherBuckets = do e <- entries; where pubkeyBucket e.toKey != b; yield e
5978− -- Process the entire bucket group in one atomic block
5979− results <- atomic do forwardBucketGroup sameBucket fromServer b t []
6032+ -- Process the entire bucket group in one atomic block.
6033+ -- Read bucket-local state once upfront and thread through the recursion.
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)
6050+ yield r.acc
59806051 forwardBatchByBucket otherBuckets fromServer t (union acc results)
59816052
59826053-- Process all forward entries for a single bucket within one atomic block.
5983−-- Reads presenceBucket/messageBucket/ackedBucket/deliveredBucket/nonceBucket once
5984−-- and applies all forwards, avoiding redundant per-entry bucket reads.
5985−forwardBucketGroup = \entries fromServer b t acc ->
6054+-- Threads bucket-local state through recursion to avoid redundant per-entry reads.
6055+-- Shared state (recvRates, federationPairRates, seqCounter) is read per-entry since
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 ->
59866058 case headOf entries of
5987− Nothing {} -> yield acc
6059+ Nothing {} -> yield {acc: acc, bucket: bucket, nonceBkt: nonceBkt, rc: rc, msgStored: msgStored}
59886060 Just {value: entry} -> do
59896061 let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
59906062 let clampedTs = clampTimestamp entry.timestamp t
5991− presenceBkt <- readPresenceBucket b
59926063 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
59946066 else do
5995− bucket <- readMessageBucket b
5996− ackedBkt <- readAckedBucket b
5997− delivBkt <- readDelivBucket b
5998− nonceBkt <- readNonceBucket b
59996067 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 {})
6001− else do
6002− totalMsgCount <- *totalMessageCount
6003− recipientCounts <- readRcBucket b
6004− let recipientCount = getRecipientCount entry.toKey recipientCounts
6005− if totalMsgCount >= maxMessages || recipientCount >= maxMessagesPerRecipient
6006− then yield (False {})
6007− else do
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 {})
6068+ if isDup
6069+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
6070+ else do
6071+ totalMsgCount <- *totalMessageCount
6072+ let recipientCount = getRecipientCount entry.toKey rc
6073+ if totalMsgCount + msgStored >= maxMessages || recipientCount >= maxMessagesPerRecipient
6074+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
60126075 else do
6013− set *recvRates = recvResult.entries
6014− -- Per-server-per-recipient rate limiting (matches handleForwardMessage)
6015− federationPairRates <- *federationPairRates
6016− let fedPairResult = checkServerRecipientRate federationPairRates fromServer entry.toKey maxFederationPairRatePerPair federationPairRateWindowMs maxFederationPairRateEntries t
6017− if fedPairResult.limited then yield (False {})
6018− else do
6019− set *federationPairRates = fedPairResult.entries
6020− seqCounter <- *seqCounter
6021− let seqTs = nextSeqTs t seqCounter
6022− set *seqCounter = seqTs
6023− 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
6025− writeMessageBucket b (union bucket
6026− [{messageId: entry.messageId, toKey: entry.toKey,
6027− 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)
6030− signalMessageForKey entry.toKey seqTs
6031− fwdNonces <- readNonceBucket b
6032− let updatedNonces = union fwdNonces [{messageId: entry.messageId, toKey: entry.toKey, seenAt: t}]
6033− writeNonceBucket b (if count updatedNonces > maxNoncesPerBucket
6034− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6035− else updatedNonces)
6036− yield (True {})
6037− forwardBucketGroup rest fromServer b t (union acc [{messageId: entry.messageId, ok: result}])
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
6081+ else do
6082+ set *recvRates = recvResult.entries
6083+ -- Per-server-per-recipient rate limiting (matches handleForwardMessage)
6084+ federationPairRates <- *federationPairRates
6085+ let fedPairResult = checkServerRecipientRate federationPairRates fromServer entry.toKey maxFederationPairRatePerPair federationPairRateWindowMs maxFederationPairRateEntries t
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
6088+ else do
6089+ set *federationPairRates = fedPairResult.entries
6090+ seqCounter <- *seqCounter
6091+ let seqTs = nextSeqTs t seqCounter
6092+ set *seqCounter = seqTs
6093+ let fwdExpiresAt = if entry.expiresAt > 0 && entry.expiresAt > t then min entry.expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
6094+ let batchClampedPriority = clamp 0 maxMessagePriority entry.priority
6095+ let newBucket = union bucket
6096+ [{messageId: entry.messageId, toKey: entry.toKey,
6097+ encryptedBlob: entry.encryptedBlob, timestamp: clampedTs, originServer: entry.originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: batchClampedPriority}]
6098+ let newRc = adjustRecipientCount rc entry.toKey 1
6099+ signalMessageForKey entry.toKey seqTs
6100+ let updatedNonces = union nonceBkt [{messageId: entry.messageId, toKey: entry.toKey, seenAt: t}]
6101+ let newNonceBkt = if count updatedNonces > maxNoncesPerBucket
6102+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6103+ else updatedNonces
6104+ forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt newBucket ackedBkt delivBkt newNonceBkt newRc (msgStored + 1)
60386105
60396106-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
60406107-- Auth checked once; each cancel processed independently. Partial success possible.
@@ -6781,7 +6848,8 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
67816848 yield rateResult.limited
67826849 if ackRateLimited then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})
67836850 else do
6784− let dedupedIds = dedupMessageIds messageIds
6851+ -- Cap input size to prevent expensive scans from unbounded messageIds lists
6852+ let dedupedIds = dedupMessageIds (takeRelation maxMessageIds messageIds)
67856853 result <- atomic do
67866854 let b = pubkeyBucket pubkey
67876855 bucket <- readMessageBucket b
@@ -6808,9 +6876,13 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
68086876 when (count localDelivered > 0) (do
68096877 sd <- *senderDeliveries
68106878 let combined = union sd localDelivered
6811− set *senderDeliveries = if count combined > maxSenderDeliveries
6879+ let overCap = count combined > maxSenderDeliveries
6880+ set *senderDeliveries = if overCap
68126881 then evictOneOldest combined (\d -> d.deliveredAt)
6813− else combined)
6882+ else combined
6883+ when overCap (do
6884+ ddr <- *droppedDeliveryReceipts
6885+ set *droppedDeliveryReceipts = ddr + 1))
68146886 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
68156887 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}
68166888 fork (when federationEnabled (do
@@ -6877,7 +6949,8 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
68776949 yield rateResult.limited
68786950 if deleteRateLimited then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})
68796951 else do
6880− let dedupedIds = dedupMessageIds messageIds
6952+ -- Cap input size to prevent expensive scans from unbounded messageIds lists
6953+ let dedupedIds = dedupMessageIds (takeRelation maxMessageIds messageIds)
68816954 deleteResult <- atomic do
68826955 let b = pubkeyBucket pubkey
68836956 bucket <- readMessageBucket b
@@ -6902,9 +6975,13 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
69026975 when (count localDeleted > 0) (do
69036976 sd <- *senderDeliveries
69046977 let combinedDel = union sd localDeleted
6905− set *senderDeliveries = if count combinedDel > maxSenderDeliveries
6978+ let overCapDel = count combinedDel > maxSenderDeliveries
6979+ set *senderDeliveries = if overCapDel
69066980 then evictOneOldest combinedDel (\d -> d.deliveredAt)
6907− else combinedDel)
6981+ else combinedDel
6982+ when overCapDel (do
6983+ ddr <- *droppedDeliveryReceipts
6984+ set *droppedDeliveryReceipts = ddr + 1))
69086985 if count ackedFederated > 0 then do
69096986 ackedBkt <- readAckedBucket b
69106987 let combinedAcked = union ackedBkt ackedFederated
@@ -7686,7 +7763,7 @@ serveOverloaded = \req -> case req of
76867763 VerifyAuth {respond} -> respond {token: "", expiresAt: 0, error: "overloaded"}
76877764 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}
76887765 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}
76907767 SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded"}
76917768 MessageStatus {respond} -> respond {status: "", authorized: False {}}
76927769 AckMessages {respond} -> respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "overloaded"}
@@ -7937,6 +8014,11 @@ backgroundSnapshot = do
79378014 lastFp <- *lastSnapshotFingerprint
79388015 let stateChanged = lastFp == 0 || fingerprint != lastFp
79398016 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.
79408022 -- Single atomic read for consistent [SNAPSHOT_DATA] output (used for restore).
79418023 -- Previous per-collection reads caused inconsistencies (e.g., ackedIds referencing
79428024 -- messageIds from a different snapshot window). The contention cost is acceptable