@@ -460,7 +460,8 @@ pruneExpired = \t -> atomic do
460 -- presence: remember locally-expired keys for offline gossip
460 -- presence: remember locally-expired keys for offline gossip
461 pres <- *presence
461 pres <- *presence
462 *presence = filter (\p -> p.expiresAt > t) pres
462 *presence = filter (\p -> p.expiresAt > t) pres
463+ let liveLocalPubkeys = map (\p -> p.pubkey) (filter (\p -> isLocalServer p.server && p.expiresAt > t) pres)
463− yield (map (\p -> p.pubkey) (filter (\p -> p.expiresAt <= t && isLocalServer p.server && not (hasLocalPresence pres p.pubkey t)) pres))
464+ yield (map (\p -> p.pubkey) (filter (\p -> p.expiresAt <= t && isLocalServer p.server && not elem p.pubkey liveLocalPubkeys) pres))
465-- ============================================
466-- ============================================
466-- Federation gossip
467-- Federation gossip
@@ -728,25 +729,23 @@ handleSendMessages = \messages authorization clientIp -> do
728-- stays so `retryPendingForwards` can re-attempt with backoff.
729-- stays so `retryPendingForwards` can re-attempt with backoff.
729forkForwardsTo : PubkeyHex -> BlobHex -> [Text] -> IO {r *presence, rw *forwardFailures, rw *forwards, rw *messages, network, clock} {}
730forkForwardsTo : PubkeyHex -> BlobHex -> [Text] -> IO {r *presence, rw *forwardFailures, rw *forwards, rw *messages, network, clock} {}
730forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
731forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
731− fwds <- *forwards
732+ t <- now
733+ result <- fetch (federationUrl server) (ForwardMessage {toKey: to, encryptedBlob})
732 let same = forwardMatches encryptedBlob server
734 let same = forwardMatches encryptedBlob server
733− when (any same fwds) (do
735− result <- fetch (federationUrl server) (ForwardMessage {toKey: to, encryptedBlob})
736− case result of
735+ case result of
737− Err _ -> recordFailure server
736+ Err _ -> recordFailure server
738− Ok _ -> do
737+ Ok _ -> do
739− clearFailure server
738+ clearFailure server
740− atomic (do
739+ atomic (do
741− fwds2 <- *forwards
740+ fwds <- *forwards
742− let remaining = filter (\x -> not same x) fwds2
741+ let remaining = filter (\x -> not same x) fwds
743− *forwards = remaining
742+ *forwards = remaining
744− pres <- *presence
743+ pres <- *presence
745− let stillLocal = hasLocalPresence pres to t
744+ let stillLocal = hasLocalPresence pres to t
746− let othersPending = any (\x -> x.encryptedBlob == encryptedBlob) remaining
745+ let othersPending = any (\x -> x.encryptedBlob == encryptedBlob) remaining
747− when (not (stillLocal || othersPending)) (do
746+ when (not (stillLocal || othersPending)) (do
748− msgs <- *messages
747+ msgs <- *messages
749− *messages = filter (\m -> m.encryptedBlob != encryptedBlob) msgs))))
748+ *messages = filter (\m -> m.encryptedBlob != encryptedBlob) msgs)))
751-- ============================================
750-- ============================================
752-- Handler: federation receive
751-- Handler: federation receive
@@ -769,16 +768,16 @@ handleRecvGossip = \events fromServer -> do
770 newlyOnline <- atomic (do
769 newlyOnline <- atomic (do
771 pres <- *presence
770 pres <- *presence
772− let touchedPubkeys = fold (\acc e -> case e of
771+ let collected = fold (\acc e -> case e of
773− GossipOnline {pubkey} -> union acc [pubkey]
772+ GossipOnline {pubkey} -> {
774− GossipOffline {pubkey} -> union acc [pubkey]
773+ touched: union acc.touched [pubkey],
776− let added = fold (\acc e -> case e of
777− GossipOnline {pubkey} -> union acc [{pubkey, server: fromServer, expiresAt: t + onlineGossipTtl}]
774+ added: union acc.added [{pubkey, server: fromServer, expiresAt: t + onlineGossipTtl}]
776+ GossipOffline {pubkey} -> {acc.added, touched: union acc.touched [pubkey]}
778− _ -> acc) [] events
777+ _ -> acc) {touched: [], added: []} events
779− let kept = filter (\p -> p.server != fromServer || not elem p.pubkey touchedPubkeys) pres
778+ let kept = filter (\p -> p.server != fromServer || not elem p.pubkey collected.touched) pres
780− *presence = union kept added
779+ *presence = union kept collected.added
781− yield (filter (\pk -> not any (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t) pres) (map (\p -> p.pubkey) added)))
780+ yield (filter (\pk -> not any (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t) pres) (map (\p -> p.pubkey) collected.added)))
782 fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer))
781 fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer))
783 yield (Ok {value: {}})
782 yield (Ok {value: {}})