@@ -427,6 +427,19 @@ evictOneOldest = \items getKey ->
427427 then {removed: True {}, items: acc.items}
428428 else {removed: acc.removed, items: union acc.items [x]}) {removed: False {}, items: []} items).items
429429
430+capPerKey = \items matchKey maxPerKey getExpiry t ->
431+ let activeCount = count (do x <- items; where matchKey x; where getExpiry x > t; yield x) in
432+ if activeCount >= maxPerKey
433+ then let forKey = do x <- items; where matchKey x; yield x in
434+ let others = do x <- items; where not (matchKey x); yield x in
435+ union others (evictOneOldest forKey getExpiry)
436+ else items
437+
438+capGlobal = \items maxCount getKey ->
439+ if count items >= maxCount
440+ then evictOneOldest items getKey
441+ else items
442+
430443collectPeerServers = \items getServer ->
431444 fold (\acc item ->
432445 let s = getServer item in
@@ -434,6 +447,31 @@ collectPeerServers = \items getServer ->
434447 else if any acc (\a -> a == s) then acc
435448 else union acc [s]) [] items
436449
450+samePubkeyServer = \a b -> a.pubkey == b.pubkey && a.server == b.server
451+
452+mergeByPubkeyServer = \initial items ->
453+ fold (\acc e ->
454+ if any acc (\a -> samePubkeyServer a e && a.expiresAt >= e.expiresAt)
455+ then acc
456+ else union
457+ (do a <- acc; where not (samePubkeyServer a e); yield a)
458+ [e]) initial items
459+
460+removeOffline = \presenceSet offlineOnlyKeys trustedOffline t ->
461+ diff presenceSet (do
462+ p <- presenceSet
463+ k <- offlineOnlyKeys
464+ where samePubkeyServer p k
465+ where any trustedOffline (\o -> samePubkeyServer o k && (if o.timestamp > t then t else o.timestamp) > p.timestamp)
466+ yield p)
467+
468+countMessagesFor = \messages toKey t ->
469+ count (do
470+ m <- messages
471+ where m.toKey == toKey
472+ where m.receivedAt >= t - messageMaxAge
473+ yield m)
474+
437475validGossipEvents = \events validRequest t eventType ->
438476 do
439477 e <- events
@@ -688,18 +726,11 @@ serve = \req -> case req of
688726 pruneResult <- atomic do
689727 pr <- maybePrune t
690728 challenges <- *challenges
691− let existingCount = count (do c <- challenges; where c.pubkey == pubkey; where c.expiresAt > t; yield c)
692− let forKey = do c <- challenges; where c.pubkey == pubkey; yield c
693− let others = do c <- challenges; where not (c.pubkey == pubkey); yield c
694− let challengesCapped = if existingCount >= maxChallengesPerKey
695− then union others (evictOneOldest forKey (\c -> c.expiresAt))
696− else challenges
729+ let challengesCapped = capPerKey challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t
697730 let withoutExpired = if count challengesCapped >= maxChallenges
698731 then do c <- challengesCapped; where c.expiresAt > t; yield c
699732 else challengesCapped
700− let globalCapped = if count withoutExpired >= maxChallenges
701− then evictOneOldest withoutExpired (\c -> c.expiresAt)
702− else withoutExpired
733+ let globalCapped = capGlobal withoutExpired maxChallenges (\c -> c.expiresAt)
703734 set *challenges = union
704735 globalCapped
705736 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
@@ -734,16 +765,9 @@ serve = \req -> case req of
734765 yield c
735766 Nothing {} -> challenges
736767 else challenges
737− let existingForKey = do s <- sessions; where s.pubkey == pubkey; where s.expiresAt > t; yield s
738− let forKey = do s <- sessions; where s.pubkey == pubkey; yield s
739− let others = do s <- sessions; where not (s.pubkey == pubkey); yield s
740− let sessionsCapped = if count existingForKey >= maxSessionsPerKey
741− then union others (evictOneOldest forKey (\s -> s.expiresAt))
742− else sessions
768+ let sessionsCapped = capPerKey sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t
743769 let activeSessionsCapped = do s <- sessionsCapped; where s.expiresAt > t; yield s
744− let globalSessionsCapped = if count activeSessionsCapped >= maxSessions
745− then evictOneOldest activeSessionsCapped (\s -> s.expiresAt)
746− else activeSessionsCapped
770+ let globalSessionsCapped = capGlobal activeSessionsCapped maxSessions (\s -> s.expiresAt)
747771 set *sessions = if sigValid
748772 then union globalSessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]
749773 else sessions
@@ -751,9 +775,7 @@ serve = \req -> case req of
751775 p <- presence
752776 where not (p.pubkey == pubkey && p.server == serverHostN)
753777 yield p
754− let presenceEvicted = if count presenceWithout >= maxPresenceEntries
755− then evictOneOldest presenceWithout (\p -> p.expiresAt)
756− else presenceWithout
778+ let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
757779 let updatedPresence = if sigValid
758780 then union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
759781 else presence
@@ -811,11 +833,7 @@ serve = \req -> case req of
811833 let authorized = isJust session
812834 seqCounter <- *seqCounter
813835 let seqTs = if t > seqCounter then t else seqCounter + 1
814− let recipientMsgCount = count (do
815− m <- messages
816− where m.toKey == to
817− where m.receivedAt >= t - messageMaxAge
818− yield m)
836+ let recipientMsgCount = countMessagesFor messages to t
819837 let shouldInsert = authorized && recipientMsgCount < maxMessagesPerRecipient
820838 set *seqCounter = if shouldInsert then seqTs else seqCounter
821839 set *messages = if shouldInsert
@@ -990,7 +1008,7 @@ serve = \req -> case req of
9901008 let offlineEntries = validGossipEvents events validRequest t "offline"
9911009 let filteredOnline = do
9921010 e <- onlineEntries
993− where none offlineEntries (\o -> o.server == fromServer && o.pubkey == e.pubkey && o.server == e.server && o.timestamp > e.timestamp)
1011+ where none offlineEntries (\o -> o.server == fromServer && samePubkeyServer o e && o.timestamp > e.timestamp)
9941012 yield {pubkey: e.pubkey, server: e.server, expiresAt: e.expiresAt, timestamp: e.timestamp, ttl: e.ttl}
9951013 let affectedKeys = union
9961014 (do
@@ -999,15 +1017,7 @@ serve = \req -> case req of
9991017 (do
10001018 e <- offlineEntries
10011019 yield {pubkey: e.pubkey, server: e.server})
1002− let deduped = fold (\acc e ->
1003− if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server && a.expiresAt >= e.expiresAt)
1004− then acc
1005− else union
1006− (do
1007− a <- acc
1008− where not (a.pubkey == e.pubkey && a.server == e.server)
1009− yield a)
1010− [e]) [] filteredOnline
1020+ let deduped = mergeByPubkeyServer [] filteredOnline
10111021 gossipResult <- atomic do
10121022 pr <- maybePrune t
10131023 presence <- *presence
@@ -1021,59 +1031,36 @@ serve = \req -> case req of
10211031 yield e
10221032 let appliedOnline = do
10231033 e <- trustedDeduped
1024− where none presence (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)
1034+ where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
10251035 yield {pubkey: e.pubkey, server: e.server}
1026− let merged = fold (\acc e ->
1027− if any acc (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)
1028− then acc
1029− else union
1030− (do
1031− p <- acc
1032− where not (p.pubkey == e.pubkey && p.server == e.server)
1033− yield p)
1034− [e]) presence trustedDeduped
1036+ let merged = mergeByPubkeyServer presence trustedDeduped
10351037 let offlineOnlyKeys = do
10361038 k <- affectedKeys
1037− where none trustedDeduped (\d -> d.pubkey == k.pubkey && d.server == k.server)
1039+ where none trustedDeduped (\d -> samePubkeyServer d k)
10381040 yield k
10391041 let appliedOffline = do
10401042 k <- offlineOnlyKeys
1041− where any presence (\p -> p.pubkey == k.pubkey && p.server == k.server && any trustedOffline (\o -> o.pubkey == k.pubkey && o.server == k.server && (if o.timestamp > t then t else o.timestamp) > p.timestamp))
1043+ where any presence (\p -> samePubkeyServer p k && any trustedOffline (\o -> samePubkeyServer o k && (if o.timestamp > t then t else o.timestamp) > p.timestamp))
10421044 yield {pubkey: k.pubkey, server: k.server}
1043− let afterOffline = diff merged (do
1044− p <- merged
1045− k <- offlineOnlyKeys
1046− where p.pubkey == k.pubkey
1047− where p.server == k.server
1048− where any trustedOffline (\o -> o.pubkey == k.pubkey && o.server == k.server && (if o.timestamp > t then t else o.timestamp) > p.timestamp)
1049− yield p)
1045+ let afterOffline = removeOffline merged offlineOnlyKeys trustedOffline t
10501046 let withinCapacity = count afterOffline <= maxPresenceEntries
10511047 let updatesOnlyMerged = if withinCapacity then afterOffline
10521048 else fold (\acc e ->
1053− if any acc (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)
1049+ if any acc (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
10541050 then acc
1055− else if any presence (\p -> p.pubkey == e.pubkey && p.server == e.server)
1051+ else if any presence (\p -> samePubkeyServer p e)
10561052 then union
1057− (do
1058− p <- acc
1059− where not (p.pubkey == e.pubkey && p.server == e.server)
1060− yield p)
1053+ (do p <- acc; where not (samePubkeyServer p e); yield p)
10611054 [e]
10621055 else acc) presence trustedDeduped
10631056 let updatesOnlyAfterOffline = if withinCapacity then afterOffline
1064− else diff updatesOnlyMerged (do
1065− p <- updatesOnlyMerged
1066− k <- offlineOnlyKeys
1067− where p.pubkey == k.pubkey
1068− where p.server == k.server
1069− where any trustedOffline (\o -> o.pubkey == k.pubkey && o.server == k.server && (if o.timestamp > t then t else o.timestamp) > p.timestamp)
1070− yield p)
1057+ else removeOffline updatesOnlyMerged offlineOnlyKeys trustedOffline t
10711058 let reclaimCapacity = if withinCapacity then 0 else maxPresenceEntries - count updatesOnlyAfterOffline
10721059 let newEntries = if withinCapacity then []
10731060 else do
10741061 e <- trustedDeduped
1075− where none presence (\p -> p.pubkey == e.pubkey && p.server == e.server)
1076− where none updatesOnlyAfterOffline (\p -> p.pubkey == e.pubkey && p.server == e.server)
1062+ where none presence (\p -> samePubkeyServer p e)
1063+ where none updatesOnlyAfterOffline (\p -> samePubkeyServer p e)
10771064 yield e
10781065 let reclaimedEntries = if withinCapacity then []
10791066 else fold (\acc e ->
@@ -1085,8 +1072,8 @@ serve = \req -> case req of
10851072 else union
10861073 (do
10871074 e <- trustedDeduped
1088− where any presence (\p -> p.pubkey == e.pubkey && p.server == e.server)
1089− where none presence (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)
1075+ where any presence (\p -> samePubkeyServer p e)
1076+ where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
10901077 yield {pubkey: e.pubkey, server: e.server})
10911078 (do
10921079 e <- reclaimedEntries
@@ -1099,13 +1086,13 @@ serve = \req -> case req of
10991086 yield {appliedKeys: (if withinCapacity then union appliedOnline appliedOffline else union appliedOnlineUpdates appliedOffline), expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
11001087 let filteredOffline = do
11011088 e <- offlineEntries
1102− where none onlineEntries (\o -> o.pubkey == e.pubkey && o.server == e.server && o.timestamp > e.timestamp)
1089+ where none onlineEntries (\o -> samePubkeyServer o e && o.timestamp > e.timestamp)
11031090 yield e
11041091 let effectiveEvents = do
11051092 e <- union
11061093 (do e <- filteredOnline; yield {eventType: "online", pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl})
11071094 (do e <- filteredOffline; yield {eventType: "offline", pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl})
1108− where any gossipResult.appliedKeys (\a -> a.pubkey == e.pubkey && a.server == e.server)
1095+ where any gossipResult.appliedKeys (\a -> samePubkeyServer a e)
11091096 yield e
11101097 if not gossipResult.withinCapacity
11111098 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)
@@ -1145,11 +1132,7 @@ serve = \req -> case req of
11451132 seqCounter <- *seqCounter
11461133 let seqTs = if t > seqCounter then t else seqCounter + 1
11471134 let recipientMsgCount = if not (authorized && validOrigin) then 0
1148− else count (do
1149− m <- messages
1150− where m.toKey == toKey
1151− where m.receivedAt >= t - messageMaxAge
1152− yield m)
1135+ else countMessagesFor messages toKey t
11531136 let existsInMessages = if not (authorized && validOrigin) then False {}
11541137 else any messages (\m -> m.messageId == messageId && m.toKey == toKey)
11551138 ackedIds <- *ackedIds