@@ -555,8 +555,7 @@ pruneExpired = \t -> do
556 yield {expiredLocal: (do
556 yield {expiredLocal: (do
557 p <- presenceBefore
557 p <- presenceBefore
558− where p.expiresAt <= t
558+ where p.expiresAt <= t && p.server == serverHostN
559− where p.server == serverHostN
560 where none presenceAfter (\q -> q.pubkey == p.pubkey && q.server == serverHostN)
559 where none presenceAfter (\q -> q.pubkey == p.pubkey && q.server == serverHostN)
561 yield p), peerServers: peerServersBefore}
560 yield p), peerServers: peerServersBefore}
@@ -576,17 +575,17 @@ forEachBatch = \items batchSize process ->
576 process batch
575 process batch
577 forEachBatch rest batchSize process
576 forEachBatch rest batchSize process
579−sendBatchedGossipToServer : Text -> [GossipEvent] -> IO {network} {}
578+sendBatchedGossipToServer : Text -> [GossipEvent] -> Int -> IO {network} {}
580−sendBatchedGossipToServer = \server allEvents ->
579+sendBatchedGossipToServer = \server allEvents hops ->
581 forEachBatch allEvents maxGossipEvents (\batch -> do
580 forEachBatch allEvents maxGossipEvents (\batch -> do
582 _ <- fetch (federationScheme ++ server) (RecvGossip {
581 _ <- fetch (federationScheme ++ server) (RecvGossip {
583 authorization: "Bearer " ++ federationSecret,
582 authorization: "Bearer " ++ federationSecret,
584− events: batch, hopsRemaining: 2, fromServer: serverHostN})
583+ events: batch, hopsRemaining: hops, fromServer: serverHostN})
585 yield {})
584 yield {})
587−sendBatchedGossip : [Text] -> [GossipEvent] -> IO {network} {}
586+sendBatchedGossip : [Text] -> [GossipEvent] -> Int -> IO {network} {}
588−sendBatchedGossip = \peerServers allEvents ->
587+sendBatchedGossip = \peerServers allEvents hops ->
589− sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents)
588+ sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents hops)
591broadcastOfflineGossip = \t peerServers expiredLocal ->
590broadcastOfflineGossip = \t peerServers expiredLocal ->
592 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do
591 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do
@@ -597,7 +596,7 @@ broadcastOfflineGossip = \t peerServers expiredLocal ->
597 let gossipEvents = do
596 let gossipEvents = do
598 e <- expiredEntries
597 e <- expiredEntries
599 yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}
598 yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}
600− sendBatchedGossip peerServers gossipEvents)
599+ sendBatchedGossip peerServers gossipEvents 2)
602forkBroadcastOffline = \t result ->
601forkBroadcastOffline = \t result ->
603 fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
602 fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
@@ -765,18 +764,20 @@ serve = \req -> case req of
765 then collectPeerServers updatedPresence (\p -> p.server)
764 then collectPeerServers updatedPresence (\p -> p.server)
767 yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}
766 yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}
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
768− fork (case result.verify of
771+ case result.verify of
769− VerifyOk {peerServers} -> do
772+ VerifyOk {peerServers} ->
770− when federationEnabled
773+ when federationEnabled
771− (sendToEach peerServers (\server -> do
774+ (sendToEach peerServers (\server -> do
772− _ <- fetch (federationScheme ++ server) (RecvGossip {
775+ _ <- fetch (federationScheme ++ server) (RecvGossip {
773− authorization: "Bearer " ++ federationSecret,
776+ authorization: "Bearer " ++ federationSecret,
774− events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
777+ events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
775− yield {}))
778+ yield {}))
776− broadcastOfflineGossip t result.pr.peerServers
779+ _ -> yield {}
777− (do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p)
779− broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
780+ broadcastOfflineGossip t result.pr.peerServers expiredLocal)
780 yield (case result.verify of
781 yield (case result.verify of
781 VerifyFailed _ -> respond {token: "", expiresAt: 0}
782 VerifyFailed _ -> respond {token: "", expiresAt: 0}
782 VerifyOk _ -> respond {token: token, expiresAt: expiry})
783 VerifyOk _ -> respond {token: token, expiresAt: expiry})
@@ -867,8 +868,7 @@ serve = \req -> case req of
867 let ackedMessages = do
868 let ackedMessages = do
868 m <- messages
869 m <- messages
869 mid <- dedupedIds
870 mid <- dedupedIds
870− where m.messageId == mid
871+ where m.messageId == mid && m.toKey == pubkey
871− where m.toKey == pubkey
873 set *messages = diff messages ackedMessages
873 set *messages = diff messages ackedMessages
874 let ackedFederated = do
874 let ackedFederated = do
@@ -877,12 +877,9 @@ serve = \req -> case req of
877 yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
877 yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
878 ackedIds <- *ackedIds
878 ackedIds <- *ackedIds
879 let combinedAckedIds = union ackedIds ackedFederated
879 let combinedAckedIds = union ackedIds ackedFederated
880− let excessCount = count combinedAckedIds - maxAckedIds
880+ let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
881− let afterEviction = if excessCount <= 0 then combinedAckedIds
881+ if excessCount <= 0 then [] else boundedSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
882− else evictOldestAckedIds combinedAckedIds excessCount
882+ set *ackedIds = diff combinedAckedIds evictedEntries
883− let evictedEntries = if excessCount <= 0 then []
884− else diff combinedAckedIds afterEviction
885− set *ackedIds = afterEviction
886 yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}}
883 yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}}
888 case result.ack of
885 case result.ack of
@@ -1021,16 +1018,12 @@ serve = \req -> case req of
1021 when (not gossipResult.withinCapacity)
1018 when (not gossipResult.withinCapacity)
1022 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
1019 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
1023 fork (do
1020 fork (do
1024− when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (do
1021+ when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (
1025 let targets = do
1022 let targets = do
1026 s <- gossipResult.gossipPeerServers
1023 s <- gossipResult.gossipPeerServers
1027 where not (s == fromServer)
1024 where not (s == fromServer)
1029− sendToEach targets (\server -> do
1030− _ <- fetch (federationScheme ++ server) (RecvGossip {
1031− authorization: "Bearer " ++ federationSecret,
1032− events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
1026+ in sendBatchedGossip targets effectiveEvents (cappedHops - 1))
1034 let pr = gossipResult.pr
1027 let pr = gossipResult.pr
1035 broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
1028 broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
1036 yield (respond {ok: True {}})
1029 yield (respond {ok: True {}})