@@ -433,7 +433,7 @@ maybePrune = \t -> do
433433 set *lastPrunedAt = t
434434 pruneExpired t
435435
436−sendBatchedGossip = \peerServers allEvents ->
436+sendBatchedGossipToServer = \server allEvents ->
437437 if count allEvents == 0
438438 then yield {}
439439 else do
@@ -441,21 +441,28 @@ sendBatchedGossip = \peerServers allEvents ->
441441 if count acc >= maxGossipEvents then acc
442442 else union acc [e]) [] allEvents
443443 let rest = diff allEvents batch
444− sendToEach peerServers (\server ->
445− fetch (federationScheme ++ server) (RecvGossip {
446− authorization: "Bearer " ++ federationSecret,
447− events: batch, hopsRemaining: 2, fromServer: serverHostN}))
448− sendBatchedGossip peerServers rest
444+ fetch (federationScheme ++ server) (RecvGossip {
445+ authorization: "Bearer " ++ federationSecret,
446+ events: batch, hopsRemaining: 2, fromServer: serverHostN})
447+ sendBatchedGossipToServer server rest
448+
449+sendBatchedGossip = \peerServers allEvents ->
450+ if count allEvents == 0
451+ then yield {}
452+ else sendToEach peerServers (\server ->
453+ sendBatchedGossipToServer server allEvents)
449454
450455broadcastOfflineGossip = \t peerServers expiredLocal -> do
451456 let expiredEntries = fold (\acc p ->
452− if count (do a <- acc; where a.pubkey == p.pubkey; yield a) > 0 then acc
453− else union acc [{pubkey: p.pubkey, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000)}]) [] expiredLocal
457+ if count (do a <- acc; where a.pubkey == p.pubkey; where a.expiresAt >= p.expiresAt; yield a) > 0 then acc
458+ else union
459+ (do a <- acc; where not (a.pubkey == p.pubkey); yield a)
460+ [{pubkey: p.pubkey, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000), expiresAt: p.expiresAt}]) [] expiredLocal
454461 if not (federationSecret == "") && count expiredEntries > 0 && count peerServers > 0
455462 then do
456463 let gossipEvents = do
457464 e <- expiredEntries
458− yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: t, ttl: e.ttl}
465+ yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}
459466 sendBatchedGossip peerServers gossipEvents
460467 else yield {}
461468
@@ -520,9 +527,21 @@ pollOnce = \token cursor ->
520527 where m.receivedAt >= t - messageMaxAge
521528 where m.receivedAt > clampedCursor
522529 yield m
523− let limitedMsgs = fold (\acc m ->
524− if count acc >= maxPollEvents then acc
525− else union acc [m]) [] matchingMsgs
530+ let limitedMsgs = (fold (\acc m ->
531+ if acc.size < maxPollEvents
532+ then
533+ let newHeap = union acc.heap [m]
534+ {size: count newHeap, heap: newHeap, heapMax: if m.receivedAt > acc.heapMax then m.receivedAt else acc.heapMax}
535+ else if m.receivedAt < acc.heapMax
536+ then
537+ let removed = fold (\r x ->
538+ if not r.done && x.receivedAt == acc.heapMax
539+ then {done: True {}, items: r.items}
540+ else {done: r.done, items: union r.items [x]}) {done: False {}, items: []} acc.heap
541+ let newHeap = union removed.items [m]
542+ let newMax = fold (\best x -> if x.receivedAt > best then x.receivedAt else best) -1 newHeap
543+ {size: acc.size, heap: newHeap, heapMax: newMax}
544+ else acc) {size: 0, heap: [], heapMax: -1} matchingMsgs).heap
526545 let authorized = case session of
527546 Nothing {} -> False {}
528547 _ -> True {}
@@ -566,22 +585,25 @@ serve = \req -> case req of
566585 pr <- maybePrune t
567586 challenges <- *challenges
568587 let existingCount = count (do c <- challenges; where c.pubkey == pubkey; where c.expiresAt > t; yield c)
569− let oldestChallengeExpiry = fold (\acc c2 -> if c2.pubkey == pubkey && (acc == -1 || c2.expiresAt < acc) then c2.expiresAt else acc) -1 challenges
588+ let oldestChallengeExpiry = fold (\acc c2 -> if c2.pubkey == pubkey && c2.expiresAt > t && (acc == -1 || c2.expiresAt < acc) then c2.expiresAt else acc) -1 challenges
570589 let cappedChallenges = if existingCount >= maxChallengesPerKey && oldestChallengeExpiry >= 0
571590 then fold (\acc c ->
572591 if not acc.removed && c.pubkey == pubkey && c.expiresAt == oldestChallengeExpiry
573592 then {removed: True {}, items: acc.items}
574593 else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} challenges
575594 else {removed: False {}, items: challenges}
576− let globalOldestExpiry = if count cappedChallenges.items >= maxChallenges
577− then fold (\acc c -> if acc == -1 || c.expiresAt < acc then c.expiresAt else acc) -1 cappedChallenges.items
595+ let withoutExpired = if count cappedChallenges.items >= maxChallenges
596+ then do c <- cappedChallenges.items; where c.expiresAt > t; yield c
597+ else cappedChallenges.items
598+ let globalOldestExpiry = if count withoutExpired >= maxChallenges
599+ then fold (\acc c -> if acc == -1 || c.expiresAt < acc then c.expiresAt else acc) -1 withoutExpired
578600 else -1
579601 let globalCapped = if globalOldestExpiry >= 0
580602 then (fold (\acc c ->
581603 if not acc.removed && c.expiresAt == globalOldestExpiry
582604 then {removed: True {}, items: acc.items}
583− else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} cappedChallenges.items).items
584− else cappedChallenges.items
605+ else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} withoutExpired).items
606+ else withoutExpired
585607 set *challenges = union
586608 globalCapped
587609 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
@@ -852,13 +874,18 @@ serve = \req -> case req of
852874 originServer: server, fromServer: serverHostN})
853875 yield {}) (\err -> do
854876 println ("federation: error sending evicted ack to " ++ server ++ ": " ++ show err)
855− atomic do
877+ evicted <- atomic do
856878 ackedIds <- *ackedIds
857879 let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
858− let available = maxAckedIds - count ackedIds
859− let toReinsert = if available <= 0 then []
860− else fold (\acc a -> if count acc >= available then acc else union acc [a]) [] reinsert
861− set *ackedIds = union ackedIds toReinsert))
880+ let combined = union ackedIds reinsert
881+ let excess = count combined - maxAckedIds
882+ let afterEviction = if excess <= 0 then combined
883+ else evictOldestAckedIds combined excess
884+ set *ackedIds = afterEviction
885+ yield (count combined - count afterEviction)
886+ if evicted > 0
887+ then println ("federation: evicted " ++ show evicted ++ " old ack entries to reinsert failed acks for " ++ server)
888+ else yield {}))
862889 else yield {}
863890 broadcastOfflineGossip t result.peerServers result.expiredLocal
864891
@@ -874,7 +901,7 @@ serve = \req -> case req of
874901 let authorized = case session of
875902 Nothing {} -> False {}
876903 _ -> True {}
877− let valid = do
904+ let valid = if not validPubkey then [] else do
878905 p <- presence
879906 where p.pubkey == pubkey
880907 where p.expiresAt > t
@@ -1151,22 +1178,25 @@ serve = \req -> case req of
11511178 messages <- *messages
11521179 seqCounter <- *seqCounter
11531180 let seqTs = if t > seqCounter then t else seqCounter + 1
1154− let recipientMsgCount = count (do
1155− m <- messages
1156− where m.toKey == toKey
1157− where m.receivedAt >= t - messageMaxAge
1158− yield m)
1159− let existsInMessages = count (do
1160− m <- messages
1161− where m.messageId == messageId
1162− where m.toKey == toKey
1163− yield m) > 0
1181+ let recipientMsgCount = if not (authorized && validOrigin && validFields) then 0
1182+ else count (do
1183+ m <- messages
1184+ where m.toKey == toKey
1185+ where m.receivedAt >= t - messageMaxAge
1186+ yield m)
1187+ let existsInMessages = if not (authorized && validOrigin && validFields) then False {}
1188+ else count (do
1189+ m <- messages
1190+ where m.messageId == messageId
1191+ where m.toKey == toKey
1192+ yield m) > 0
11641193 ackedIds <- *ackedIds
1165− let existsInAcked = count (do
1166− a <- ackedIds
1167− where a.messageId == messageId
1168− where a.toKey == toKey
1169− yield a) > 0
1194+ let existsInAcked = if not (authorized && validOrigin && validFields) then False {}
1195+ else count (do
1196+ a <- ackedIds
1197+ where a.messageId == messageId
1198+ where a.toKey == toKey
1199+ yield a) > 0
11701200 let exists = existsInMessages || existsInAcked
11711201 let shouldInsert = authorized && validOrigin && validFields && not exists && recipientMsgCount < maxMessagesPerRecipient
11721202 set *seqCounter = if shouldInsert then seqTs else seqCounter