@@ -31,6 +31,7 @@ maxPubkeyLen = 256
31maxBlobLen = 65536
31maxBlobLen = 65536
32maxSignatureLen = 256
32maxSignatureLen = 256
33maxMessageIds = 100
33maxMessageIds = 100
34+maxBatchSize = 100 -- max messages per batch request (bounds recursion depth in sendBatchInner)
34maxGossipEvents = 100
35maxGossipEvents = 100
35maxPollEvents = 50
36maxPollEvents = 50
36maxPresenceEntries = 10000
37maxPresenceEntries = 10000
@@ -2339,15 +2340,32 @@ evictOneOldest = \items getKey ->
2339 Nothing {} -> items
2340 Nothing {} -> items
2340 Just {} -> r.items
2341 Just {} -> r.items
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
2342−-- Remove the N items with the smallest key value. O(n*k) — efficient when k is small.
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.
2343evictSmallest = \items n getKey ->
2354evictSmallest = \items n getKey ->
2344 if n <= 0 then {kept: items, evicted: []}
2355 if n <= 0 then {kept: items, evicted: []}
2345− else let r = extractOldest items getKey in
2356+ else if count items <= n then {kept: [], evicted: items}
2346− case r.candidate of
2357+ else
2347− Nothing {} -> {kept: items, evicted: []}
2358+ let result = fold (\acc x ->
2348− Just {value: c} ->
2359+ if count acc.evict < n
2349− let rest = evictSmallest r.items (n - 1) getKey in
2360+ then {evict: union acc.evict [x], maxKey: if getKey x > acc.maxKey then getKey x else acc.maxKey}
2350− {kept: rest.kept, evicted: union [c] rest.evicted}
2361+ else if getKey x >= acc.maxKey then acc
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}
2352-- Single-pass eviction by timestamp cutoff: keeps items where getKey >= cutoff.
2370-- Single-pass eviction by timestamp cutoff: keeps items where getKey >= cutoff.
2353-- O(n) vs O(n*k) for evictSmallest. Used for forward nonce eviction on the hot path.
2371-- 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 ->
3866 bucket <- readMessageBucket msgBucket
3884 bucket <- readMessageBucket msgBucket
3867 ackedBkt <- readAckedBucket msgBucket
3885 ackedBkt <- readAckedBucket msgBucket
3868 delivBkt <- readDelivBucket msgBucket
3886 delivBkt <- readDelivBucket msgBucket
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
3869− let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt
3891+ let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt || any nonceBkt (\n -> n.messageId == effectiveMsgId && n.toKey == to)
3870 if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}
3892 if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}
3872 sendRates <- *sendRates
3894 sendRates <- *sendRates
@@ -4256,26 +4278,28 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
4256 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
4278 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
4257 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
4279 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
4258 let deduped = mergeByPubkeyServer [] filteredOnline
4280 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.
4260 -- A peer announcing presence for a key we have a live session for could be a presence
4282 -- A peer announcing presence for a key we have a live session for could be a presence
4261 -- hijack attempt: the attacker's server would receive forwarded message copies.
4283 -- 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.
4263 localClashes <- atomic do
4286 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)
4266 localPresence <- readSelectedPresenceBuckets targetBuckets
4288 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)
4268 when (count localClashes > 0)
4290 when (count localClashes > 0)
4269 (logWarn ("gossip: " ++ fromServer ++ " claims " ++ show (count localClashes) ++ " locally-authenticated key(s) — filtering from merge"))
4291 (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
4292+ -- Remove clashing entries from both deduped and trustedOffline before merge
4271− -- locally-authenticated keys; third-party relays are handled by trustedGossipEntries)
4272 let safeDeduped = if count localClashes > 0
4293 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
4274 else deduped
4295 else deduped
4275 -- Per-bucket gossip merge: read only presence buckets affected by this gossip batch
4296 -- Per-bucket gossip merge: read only presence buckets affected by this gossip batch
4276 -- instead of all 16. All entries for a pubkey hash to the same bucket, so per-bucket
4297 -- instead of all 16. All entries for a pubkey hash to the same bucket, so per-bucket
4277 -- merge is equivalent to global merge. Reduces lock scope for small gossip batches.
4298 -- merge is equivalent to global merge. Reduces lock scope for small gossip batches.
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
4278− let allGossipEntries = union safeDeduped trustedOffline
4302+ let allGossipEntries = union safeDeduped safeTrustedOffline
4279 let affectedBuckets = uniqueBucketsFor allGossipEntries (\e -> e.pubkey)
4303 let affectedBuckets = uniqueBucketsFor allGossipEntries (\e -> e.pubkey)
4280 gossipResult <- atomic do
4304 gossipResult <- atomic do
4281 selectedPresence <- readSelectedPresenceBuckets affectedBuckets
4305 selectedPresence <- readSelectedPresenceBuckets affectedBuckets
@@ -4285,7 +4309,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
4285 let capacityDeduped = if tpc >= maxPresenceEntries
4309 let capacityDeduped = if tpc >= maxPresenceEntries
4286 then do d <- safeDeduped; where any selectedPresence (\p -> p.pubkey == d.pubkey); yield d
4310 then do d <- safeDeduped; where any selectedPresence (\p -> p.pubkey == d.pubkey); yield d
4287 else safeDeduped
4311 else safeDeduped
4288− let gm = mergeGossipPresence selectedPresence capacityDeduped trustedOffline fromServer t
4312+ let gm = mergeGossipPresence selectedPresence capacityDeduped safeTrustedOffline fromServer t
4289 -- Write back only affected buckets (not writeAllPresence, which would zero unaffected buckets)
4313 -- Write back only affected buckets (not writeAllPresence, which would zero unaffected buckets)
4290 writeSelectedPresenceBuckets affectedBuckets gm.finalPresence
4314 writeSelectedPresenceBuckets affectedBuckets gm.finalPresence
4291 let delta = count gm.finalPresence - count selectedPresence
4315 let delta = count gm.finalPresence - count selectedPresence
@@ -4747,9 +4771,10 @@ handleStats = \authorization respond -> do
4747 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
4771 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
4748 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt
4772 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt
4749 -- Per-server forward queue summary: count unique servers and max queue depth
4773 -- Per-server forward queue summary: count unique servers and max queue depth
4750− let fwdCounts = fold (\acc f ->
4774+ -- Two-pass aggregation: collect unique servers, then count per server.
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)
4775+ -- Avoids O(n*k) accumulator rebuilding in the fold (no list comprehension per step).
4752− else union acc [{server: f.toServer, count: 1}]) [] core.f
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)}
4753 let fwdServerCount = count fwdCounts
4778 let fwdServerCount = count fwdCounts
4754 let maxFwdDepth = fold (\acc c -> max acc c.count) 0 fwdCounts
4779 let maxFwdDepth = fold (\acc c -> max acc c.count) 0 fwdCounts
4755 yield (respond {ok: True {},
4780 yield (respond {ok: True {},
@@ -4770,9 +4795,9 @@ handleStatsPeers = \authorization respond -> do
4770 failures <- *forwardFailures
4795 failures <- *forwardFailures
4771 forwards <- readAllForwards
4796 forwards <- readAllForwards
4772 -- Count pending forwards per server in a single pass
4797 -- Count pending forwards per server in a single pass
4773− let fwdCounts = fold (\acc f ->
4798+ -- Two-pass aggregation: collect unique servers, then count per server.
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)
4799+ let fwdServers = fold (\acc f -> if any acc (\s -> s == f.toServer) then acc else union acc [f.toServer]) [] forwards
4775− else union acc [{server: f.toServer, count: 1}]) [] forwards
4800+ let fwdCounts = do s <- fwdServers; yield {server: s, count: count (do f <- forwards; where f.toServer == s; yield f)}
4776 let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
4801 let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
4777 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}
4802 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}
4778 -- Include servers with pending forwards but no failure records
4803 -- Include servers with pending forwards but no failure records
@@ -4863,7 +4888,9 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
4863 -- snapshot could inject presence pointing to private IPs, causing forward attempts
4888 -- snapshot could inject presence pointing to private IPs, causing forward attempts
4864 -- to those addresses.
4889 -- to those addresses.
4865 let validPresence = do p <- presenceData; where not (isLocalServer p.server) && not (isBadServerName p.server); yield p
4890 let validPresence = do p <- presenceData; where not (isLocalServer p.server) && not (isBadServerName p.server); yield p
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
4866− let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow) + (count presenceData - count validPresence)
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)
4867 when (rejectedCount > 0) (logWarn ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))
4894 when (rejectedCount > 0) (logWarn ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))
4868 result <- atomic do
4895 result <- atomic do
4869 -- Merge presence (existing entries win on conflict via freshness)
4896 -- Merge presence (existing entries win on conflict via freshness)
@@ -4883,12 +4910,12 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
4883 let mergedOverflow = union ackOverflow validOverflow
4910 let mergedOverflow = union ackOverflow validOverflow
4884 set *ackOverflow = takeRelation maxAckOverflow mergedOverflow
4911 set *ackOverflow = takeRelation maxAckOverflow mergedOverflow
4885 deliveredIds <- readAllDelivered
4912 deliveredIds <- readAllDelivered
4886− let mergedDelivered = union deliveredIds deliveredIdsData
4913+ let mergedDelivered = union deliveredIds validDeliveredIds
4887 let cappedDelivered = takeRelation maxDeliveredIds mergedDelivered
4914 let cappedDelivered = takeRelation maxDeliveredIds mergedDelivered
4888 writeAllDelivered cappedDelivered
4915 writeAllDelivered cappedDelivered
4889 -- Restore peer failure records
4916 -- Restore peer failure records
4890 forwardFailures <- *forwardFailures
4917 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
4892 let cappedFailures = takeRelation maxForwardFailures mergedFailures
4919 let cappedFailures = takeRelation maxForwardFailures mergedFailures
4893 set *forwardFailures = cappedFailures
4920 set *forwardFailures = cappedFailures
4894 -- Restore forward entries (dedup by messageId + toServer, keep fresher on conflict; using validated data)
4921 -- Restore forward entries (dedup by messageId + toServer, keep fresher on conflict; using validated data)
@@ -5056,7 +5083,12 @@ handleSendBatch = \messages authorization clientIp respond -> do
5056 withSessionAuth authorization t clientIp
5083 withSessionAuth authorization t clientIp
5057 (yield (respond {results: [], authorized: False {}, retryAfterMs: 0}))
5084 (yield (respond {results: [], authorized: False {}, retryAfterMs: 0}))
5058 (\senderKey -> do
5085 (\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
5060 results <- sendBatchInner senderKey capped t []
5092 results <- sendBatchInner senderKey capped t []
5061 -- Fork federation forwarding: one forked task per target server (not per message).
5093 -- Fork federation forwarding: one forked task per target server (not per message).
5062 -- Within each task, forwards are sent sequentially to avoid per-message fork overhead.
5094 -- Within each task, forwards are sent sequentially to avoid per-message fork overhead.