@@ -555,8 +555,7 @@ pruneExpired = \t -> do
555555 yield a
556556 yield {expiredLocal: (do
557557 p <- presenceBefore
558− where p.expiresAt <= t
559− where p.server == serverHostN
558+ where p.expiresAt <= t && p.server == serverHostN
560559 where none presenceAfter (\q -> q.pubkey == p.pubkey && q.server == serverHostN)
561560 yield p), peerServers: peerServersBefore}
562561
@@ -576,17 +575,17 @@ forEachBatch = \items batchSize process ->
576575 process batch
577576 forEachBatch rest batchSize process
578577
579−sendBatchedGossipToServer : Text -> [GossipEvent] -> IO {network} {}
580−sendBatchedGossipToServer = \server allEvents ->
578+sendBatchedGossipToServer : Text -> [GossipEvent] -> Int -> IO {network} {}
579+sendBatchedGossipToServer = \server allEvents hops ->
581580 forEachBatch allEvents maxGossipEvents (\batch -> do
582581 _ <- fetch (federationScheme ++ server) (RecvGossip {
583582 authorization: "Bearer " ++ federationSecret,
584− events: batch, hopsRemaining: 2, fromServer: serverHostN})
583+ events: batch, hopsRemaining: hops, fromServer: serverHostN})
585584 yield {})
586585
587−sendBatchedGossip : [Text] -> [GossipEvent] -> IO {network} {}
588−sendBatchedGossip = \peerServers allEvents ->
589− sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents)
586+sendBatchedGossip : [Text] -> [GossipEvent] -> Int -> IO {network} {}
587+sendBatchedGossip = \peerServers allEvents hops ->
588+ sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents hops)
590589
591590broadcastOfflineGossip = \t peerServers expiredLocal ->
592591 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do
@@ -597,7 +596,7 @@ broadcastOfflineGossip = \t peerServers expiredLocal ->
597596 let gossipEvents = do
598597 e <- expiredEntries
599598 yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}
600− sendBatchedGossip peerServers gossipEvents)
599+ sendBatchedGossip peerServers gossipEvents 2)
601600
602601forkBroadcastOffline = \t result ->
603602 fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
@@ -765,18 +764,20 @@ serve = \req -> case req of
765764 then collectPeerServers updatedPresence (\p -> p.server)
766765 else []
767766 yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}
768− fork (case result.verify of
769− VerifyOk {peerServers} -> do
770− when federationEnabled
771− (sendToEach peerServers (\server -> do
772− _ <- fetch (federationScheme ++ server) (RecvGossip {
773− authorization: "Bearer " ++ federationSecret,
774− events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
775− yield {}))
776− broadcastOfflineGossip t result.pr.peerServers
777− (do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p)
778− _ ->
779− broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
767+ let expiredLocal = case result.verify of
768+ VerifyOk _ -> do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p
769+ _ -> result.pr.expiredLocal
770+ fork (do
771+ case result.verify of
772+ VerifyOk {peerServers} ->
773+ when federationEnabled
774+ (sendToEach peerServers (\server -> do
775+ _ <- fetch (federationScheme ++ server) (RecvGossip {
776+ authorization: "Bearer " ++ federationSecret,
777+ events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
778+ yield {}))
779+ _ -> yield {}
780+ broadcastOfflineGossip t result.pr.peerServers expiredLocal)
780781 yield (case result.verify of
781782 VerifyFailed _ -> respond {token: "", expiresAt: 0}
782783 VerifyOk _ -> respond {token: token, expiresAt: expiry})
@@ -867,8 +868,7 @@ serve = \req -> case req of
867868 let ackedMessages = do
868869 m <- messages
869870 mid <- dedupedIds
870− where m.messageId == mid
871− where m.toKey == pubkey
871+ where m.messageId == mid && m.toKey == pubkey
872872 yield m
873873 set *messages = diff messages ackedMessages
874874 let ackedFederated = do
@@ -877,12 +877,9 @@ serve = \req -> case req of
877877 yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
878878 ackedIds <- *ackedIds
879879 let combinedAckedIds = union ackedIds ackedFederated
880− let excessCount = count combinedAckedIds - maxAckedIds
881− let afterEviction = if excessCount <= 0 then combinedAckedIds
882− else evictOldestAckedIds combinedAckedIds excessCount
883− let evictedEntries = if excessCount <= 0 then []
884− else diff combinedAckedIds afterEviction
885− set *ackedIds = afterEviction
880+ let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
881+ if excessCount <= 0 then [] else boundedSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
882+ set *ackedIds = diff combinedAckedIds evictedEntries
886883 yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}}
887884 fork (do
888885 case result.ack of
@@ -1021,16 +1018,12 @@ serve = \req -> case req of
10211018 when (not gossipResult.withinCapacity)
10221019 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
10231020 fork (do
1024− when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (do
1021+ when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (
10251022 let targets = do
10261023 s <- gossipResult.gossipPeerServers
10271024 where not (s == fromServer)
10281025 yield s
1029− sendToEach targets (\server -> do
1030− _ <- fetch (federationScheme ++ server) (RecvGossip {
1031− authorization: "Bearer " ++ federationSecret,
1032− events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
1033− yield {}))
1026+ in sendBatchedGossip targets effectiveEvents (cappedHops - 1))
10341027 let pr = gossipResult.pr
10351028 broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
10361029 yield (respond {ok: True {}})