@@ -31,6 +31,7 @@ maxPubkeyLen = 256
3131maxBlobLen = 65536
3232maxSignatureLen = 256
3333maxMessageIds = 100
34+maxBatchSize = 100 -- max messages per batch request (bounds recursion depth in sendBatchInner)
3435maxGossipEvents = 100
3536maxPollEvents = 50
3637maxPresenceEntries = 10000
@@ -2339,15 +2340,32 @@ evictOneOldest = \items getKey ->
23392340 Nothing {} -> items
23402341 Just {} -> r.items
23412342
2342−-- Remove the N items with the smallest key value. O(n*k) — efficient when k is small.
2343+-- Remove the first item matching a predicate. O(n) single pass.
2344+removeFirstMatch = \items pred ->
2345+ let r = fold (\acc x ->
2346+ if not acc.done && pred x then {items: acc.items, done: True {}}
2347+ else {items: union acc.items [x], done: acc.done}
2348+ ) {items: [], done: False {}} items
2349+ in r.items
2350+
2351+-- Remove the N items with the smallest key value.
2352+-- Single-pass fold maintaining N candidates with smallest keys, then diff.
2353+-- O(n*k) time — avoids the O(k*n²) cost of recursive extractOldest + list rebuilding.
23432354evictSmallest = \items n getKey ->
23442355 if n <= 0 then {kept: items, evicted: []}
2345− else let r = extractOldest items getKey in
2346− case r.candidate of
2347− Nothing {} -> {kept: items, evicted: []}
2348− Just {value: c} ->
2349− let rest = evictSmallest r.items (n - 1) getKey in
2350− {kept: rest.kept, evicted: union [c] rest.evicted}
2356+ else if count items <= n then {kept: [], evicted: items}
2357+ else
2358+ let result = fold (\acc x ->
2359+ if count acc.evict < n
2360+ then {evict: union acc.evict [x], maxKey: if getKey x > acc.maxKey then getKey x else acc.maxKey}
2361+ else if getKey x >= acc.maxKey then acc
2362+ else
2363+ let without = removeFirstMatch acc.evict (\e -> getKey e == acc.maxKey) in
2364+ let newEvict = union without [x] in
2365+ let newMax = fold (\m e -> if getKey e > m then getKey e else m) (getKey x) newEvict in
2366+ {evict: newEvict, maxKey: newMax}
2367+ ) {evict: [], maxKey: 0} items
2368+ in {kept: diff items result.evict, evicted: result.evict}
23512369
23522370-- Single-pass eviction by timestamp cutoff: keeps items where getKey >= cutoff.
23532371-- O(n) vs O(n*k) for evictSmallest. Used for forward nonce eviction on the hot path.
@@ -3866,7 +3884,11 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
38663884 bucket <- readMessageBucket msgBucket
38673885 ackedBkt <- readAckedBucket msgBucket
38683886 delivBkt <- readDelivBucket msgBucket
3869− let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt
3887+ -- Nonce check closes the dedup gap between message expiry (30 days) and nonce retention
3888+ -- (60 days): without this, a client could replay a messageId after messages/ackedIds/deliveredIds
3889+ -- expire but before the nonce expires.
3890+ nonceBkt <- readNonceBucket msgBucket
3891+ let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt || any nonceBkt (\n -> n.messageId == effectiveMsgId && n.toKey == to)
38703892 if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}
38713893 else do
38723894 sendRates <- *sendRates
@@ -4256,26 +4278,28 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
42564278 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
42574279 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
42584280 let deduped = mergeByPubkeyServer [] filteredOnline
4259− -- Reject first-party gossip entries that claim keys locally authenticated on this server.
4281+ -- Reject gossip entries that claim keys locally authenticated on this server.
42604282 -- A peer announcing presence for a key we have a live session for could be a presence
42614283 -- hijack attempt: the attacker's server would receive forwarded message copies.
4262− -- Filter these out before the merge to prevent duplicate routing.
4284+ -- Covers both first-party (e.server == fromServer) and third-party relay (e.server != fromServer)
4285+ -- to prevent attackers from using relay gossip to reroute locally-authenticated keys.
42634286 localClashes <- atomic do
4264− let firstParty = do e <- deduped; where e.server == fromServer; yield e
4265− let targetBuckets = uniqueBucketsFor firstParty (\e -> e.pubkey)
4287+ let targetBuckets = uniqueBucketsFor deduped (\e -> e.pubkey)
42664288 localPresence <- readSelectedPresenceBuckets targetBuckets
4267− yield (do e <- firstParty; where any localPresence (\p -> p.pubkey == e.pubkey && isLocalServer p.server && p.expiresAt > t); yield e.pubkey)
4289+ yield (do e <- deduped; where any localPresence (\p -> p.pubkey == e.pubkey && isLocalServer p.server && p.expiresAt > t); yield e.pubkey)
42684290 when (count localClashes > 0)
42694291 (logWarn ("gossip: " ++ fromServer ++ " claims " ++ show (count localClashes) ++ " locally-authenticated key(s) — filtering from merge"))
4270− -- Remove clashing entries from deduped before merge (only first-party claims for
4271− -- locally-authenticated keys; third-party relays are handled by trustedGossipEntries)
4292+ -- Remove clashing entries from both deduped and trustedOffline before merge
42724293 let safeDeduped = if count localClashes > 0
4273− then do d <- deduped; where not (d.server == fromServer && any localClashes (\pk -> pk == d.pubkey)); yield d
4294+ then do d <- deduped; where not (any localClashes (\pk -> pk == d.pubkey)); yield d
42744295 else deduped
42754296 -- Per-bucket gossip merge: read only presence buckets affected by this gossip batch
42764297 -- instead of all 16. All entries for a pubkey hash to the same bucket, so per-bucket
42774298 -- merge is equivalent to global merge. Reduces lock scope for small gossip batches.
4278− let allGossipEntries = union safeDeduped trustedOffline
4299+ let safeTrustedOffline = if count localClashes > 0
4300+ then do d <- trustedOffline; where not (any localClashes (\pk -> pk == d.pubkey)); yield d
4301+ else trustedOffline
4302+ let allGossipEntries = union safeDeduped safeTrustedOffline
42794303 let affectedBuckets = uniqueBucketsFor allGossipEntries (\e -> e.pubkey)
42804304 gossipResult <- atomic do
42814305 selectedPresence <- readSelectedPresenceBuckets affectedBuckets
@@ -4285,7 +4309,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
42854309 let capacityDeduped = if tpc >= maxPresenceEntries
42864310 then do d <- safeDeduped; where any selectedPresence (\p -> p.pubkey == d.pubkey); yield d
42874311 else safeDeduped
4288− let gm = mergeGossipPresence selectedPresence capacityDeduped trustedOffline fromServer t
4312+ let gm = mergeGossipPresence selectedPresence capacityDeduped safeTrustedOffline fromServer t
42894313 -- Write back only affected buckets (not writeAllPresence, which would zero unaffected buckets)
42904314 writeSelectedPresenceBuckets affectedBuckets gm.finalPresence
42914315 let delta = count gm.finalPresence - count selectedPresence
@@ -4747,9 +4771,10 @@ handleStats = \authorization respond -> do
47474771 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
47484772 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt
47494773 -- Per-server forward queue summary: count unique servers and max queue depth
4750− let fwdCounts = fold (\acc f ->
4751− if any acc (\a -> a.server == f.toServer) then do a <- acc; yield (if a.server == f.toServer then {server: a.server, count: a.count + 1} else a)
4752− else union acc [{server: f.toServer, count: 1}]) [] core.f
4774+ -- Two-pass aggregation: collect unique servers, then count per server.
4775+ -- Avoids O(n*k) accumulator rebuilding in the fold (no list comprehension per step).
4776+ let fwdServers = fold (\acc f -> if any acc (\s -> s == f.toServer) then acc else union acc [f.toServer]) [] core.f
4777+ let fwdCounts = do s <- fwdServers; yield {server: s, count: count (do f <- core.f; where f.toServer == s; yield f)}
47534778 let fwdServerCount = count fwdCounts
47544779 let maxFwdDepth = fold (\acc c -> max acc c.count) 0 fwdCounts
47554780 yield (respond {ok: True {},
@@ -4770,9 +4795,9 @@ handleStatsPeers = \authorization respond -> do
47704795 failures <- *forwardFailures
47714796 forwards <- readAllForwards
47724797 -- Count pending forwards per server in a single pass
4773− let fwdCounts = fold (\acc f ->
4774− if any acc (\a -> a.server == f.toServer) then do a <- acc; yield (if a.server == f.toServer then {server: a.server, count: a.count + 1} else a)
4775− else union acc [{server: f.toServer, count: 1}]) [] forwards
4798+ -- Two-pass aggregation: collect unique servers, then count per server.
4799+ let fwdServers = fold (\acc f -> if any acc (\s -> s == f.toServer) then acc else union acc [f.toServer]) [] forwards
4800+ let fwdCounts = do s <- fwdServers; yield {server: s, count: count (do f <- forwards; where f.toServer == s; yield f)}
47764801 let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
47774802 let peers = do f <- failures; yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoffFor f.failures f.server, retriable: shouldRetryServer f.server t failures, pendingForwards: fwdCountFor f.server}
47784803 -- Include servers with pending forwards but no failure records
@@ -4863,7 +4888,9 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
48634888 -- snapshot could inject presence pointing to private IPs, causing forward attempts
48644889 -- to those addresses.
48654890 let validPresence = do p <- presenceData; where not (isLocalServer p.server) && not (isBadServerName p.server); yield p
4866− let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow) + (count presenceData - count validPresence)
4891+ let validDeliveredIds = do d <- deliveredIdsData; where not (isLocalServer d.originServer); yield d
4892+ let validFailures = do f <- forwardFailuresData; where not (isBadServerName f.server) && not (isLocalServer f.server); yield f
4893+ let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow) + (count presenceData - count validPresence) + (count deliveredIdsData - count validDeliveredIds) + (count forwardFailuresData - count validFailures)
48674894 when (rejectedCount > 0) (logWarn ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))
48684895 result <- atomic do
48694896 -- Merge presence (existing entries win on conflict via freshness)
@@ -4883,12 +4910,12 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
48834910 let mergedOverflow = union ackOverflow validOverflow
48844911 set *ackOverflow = takeRelation maxAckOverflow mergedOverflow
48854912 deliveredIds <- readAllDelivered
4886− let mergedDelivered = union deliveredIds deliveredIdsData
4913+ let mergedDelivered = union deliveredIds validDeliveredIds
48874914 let cappedDelivered = takeRelation maxDeliveredIds mergedDelivered
48884915 writeAllDelivered cappedDelivered
48894916 -- Restore peer failure records
48904917 forwardFailures <- *forwardFailures
4891− let mergedFailures = mergeByKey (\a b -> a.server == b.server) (\x -> x.failedAt) forwardFailures forwardFailuresData
4918+ let mergedFailures = mergeByKey (\a b -> a.server == b.server) (\x -> x.failedAt) forwardFailures validFailures
48924919 let cappedFailures = takeRelation maxForwardFailures mergedFailures
48934920 set *forwardFailures = cappedFailures
48944921 -- Restore forward entries (dedup by messageId + toServer, keep fresher on conflict; using validated data)
@@ -5056,7 +5083,12 @@ handleSendBatch = \messages authorization clientIp respond -> do
50565083 withSessionAuth authorization t clientIp
50575084 (yield (respond {results: [], authorized: False {}, retryAfterMs: 0}))
50585085 (\senderKey -> do
5059− let capped = takeRelation maxMessageIds messages
5086+ let sized = takeRelation maxBatchSize messages
5087+ -- Pre-deduplicate by explicit messageId to prevent within-batch duplicates.
5088+ -- Each message in a batch gets its own atomic block via sendMessageCore, so
5089+ -- dedup between messages in the same batch is not caught by STM.
5090+ -- Messages with empty messageId (auto-generated) are always kept.
5091+ let capped = fold (\acc m -> if m.messageId != "" && any acc (\a -> a.messageId == m.messageId) then acc else union acc [m]) [] sized
50605092 results <- sendBatchInner senderKey capped t []
50615093 -- Fork federation forwarding: one forked task per target server (not per message).
50625094 -- Within each task, forwards are sent sequentially to avoid per-message fork overhead.