@@ -506,10 +506,8 @@ countMessagesFor = \messages toKey t ->
506506validGossipEvents : [GossipEvent] -> Timestamp -> [ValidGossipEntry]
507507validGossipEvents = \events t -> do
508508 e <- events
509− where not (tooLong e.pubkey maxPubkeyLen)
510− where not (tooLong e.server maxServerNameLen)
511− where not (isBadServerName e.server)
512− where isValidHex e.pubkey
509+ where isValidHex e.pubkey && not (tooLong e.pubkey maxPubkeyLen)
510+ where not (tooLong e.server maxServerNameLen) && not (isBadServerName e.server)
513511 let server = normalizeServerName e.server
514512 where not (server == serverHostN)
515513 where e.timestamp <= t + clockSkewTolerance
@@ -590,12 +588,12 @@ sendBatchedGossip : [Text] -> [GossipEvent] -> IO {network} {}
590588sendBatchedGossip = \peerServers allEvents ->
591589 sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents)
592590
593−broadcastOfflineGossip = \t peerServers expiredLocal -> do
594− let expiredEntries = mergeByKey (\a b -> a.pubkey == b.pubkey) [] (do
595− p <- expiredLocal
596− let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
597− yield {pubkey: p.pubkey, ttl: ttl, expiresAt: p.expiresAt})
598− when (federationEnabled && count expiredEntries > 0 && count peerServers > 0) (do
591+broadcastOfflineGossip = \t peerServers expiredLocal ->
592+ when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do
593+ let expiredEntries = mergeByKey (\a b -> a.pubkey == b.pubkey) [] (do
594+ p <- expiredLocal
595+ let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
596+ yield {pubkey: p.pubkey, ttl: ttl, expiresAt: p.expiresAt})
599597 let gossipEvents = do
600598 e <- expiredEntries
601599 yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}
@@ -767,20 +765,18 @@ serve = \req -> case req of
767765 then collectPeerServers updatedPresence (\p -> p.server)
768766 else []
769767 yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}
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− let expiredLocal = case result.verify of
781− VerifyOk _ -> do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p
782− _ -> result.pr.expiredLocal
783− broadcastOfflineGossip t result.pr.peerServers expiredLocal)
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)
784780 yield (case result.verify of
785781 VerifyFailed _ -> respond {token: "", expiresAt: 0}
786782 VerifyOk _ -> respond {token: token, expiresAt: expiry})
@@ -800,13 +796,14 @@ serve = \req -> case req of
800796 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
801797 let to = normalizeHex to
802798 let encryptedBlob = normalizeHex encryptedBlob
799+ let token = extractToken authorization
803800 t <- now
804801 msgId <- genToken
805802 let clampedTs = clampTimestamp timestamp t
806803 result <- atomic do
807804 pr <- maybePrune t
808805 sessions <- *sessions
809− let auth = authenticateSession sessions (extractToken authorization) t
806+ let auth = authenticateSession sessions token t
810807 case auth of
811808 Unauthorized {} ->
812809 yield {pr: pr, send: SendUnauthorized {}}
@@ -855,12 +852,13 @@ serve = \req -> case req of
855852 SendDone {status} -> respond {status: status, messageId: msgId})
856853
857854 AckMessages {messageIds, authorization, respond} -> do
855+ let token = extractToken authorization
858856 t <- now
859857 let dedupedIds = dedupMessageIds messageIds
860858 result <- atomic do
861859 pr <- maybePrune t
862860 sessions <- *sessions
863− let auth = authenticateSession sessions (extractToken authorization) t
861+ let auth = authenticateSession sessions token t
864862 case auth of
865863 Unauthorized {} ->
866864 yield {pr: pr, ack: AckUnauthorized {}}
@@ -888,35 +886,36 @@ serve = \req -> case req of
888886 yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}}
889887 fork (do
890888 case result.ack of
891− AckDone {pendingAcks, evictedEntries} -> do
892− let originServers = collectPeerServers pendingAcks (\a -> a.originServer)
893− when (count originServers > 0 && federationEnabled)
894− (sendAcksAndClean pendingAcks originServers)
895− let evictedToRetry = do
896− a <- evictedEntries
897− where none pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)
898− yield a
899− when (count evictedToRetry > 0 && federationEnabled) (do
900− let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
901− sendToEach evictedOriginServers (\server -> do
902− fetchResult <- fetch (federationScheme ++ server) (FederationAck {
903− authorization: "Bearer " ++ federationSecret,
904− messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
905− originServer: server, fromServer: serverHostN})
906− case fetchResult of
907− Err _ -> do
908− dropped <- atomic do
909− ackedIds <- *ackedIds
910− let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
911− let combined = union ackedIds reinsert
912− let excess = count combined - maxAckedIds
913− let afterEviction = if excess <= 0 then combined
914− else evictOldestAckedIds combined excess
915− set *ackedIds = afterEviction
916− yield excess
917− when (dropped > 0)
918− (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
919− _ -> yield {}))
889+ AckDone {pendingAcks, evictedEntries} ->
890+ when federationEnabled (do
891+ let originServers = collectPeerServers pendingAcks (\a -> a.originServer)
892+ when (count originServers > 0)
893+ (sendAcksAndClean pendingAcks originServers)
894+ let evictedToRetry = do
895+ a <- evictedEntries
896+ where none pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)
897+ yield a
898+ when (count evictedToRetry > 0) (do
899+ let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
900+ sendToEach evictedOriginServers (\server -> do
901+ fetchResult <- fetch (federationScheme ++ server) (FederationAck {
902+ authorization: "Bearer " ++ federationSecret,
903+ messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
904+ originServer: server, fromServer: serverHostN})
905+ case fetchResult of
906+ Err _ -> do
907+ dropped <- atomic do
908+ ackedIds <- *ackedIds
909+ let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
910+ let combined = union ackedIds reinsert
911+ let excess = count combined - maxAckedIds
912+ let afterEviction = if excess <= 0 then combined
913+ else evictOldestAckedIds combined excess
914+ set *ackedIds = afterEviction
915+ yield excess
916+ when (dropped > 0)
917+ (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
918+ _ -> yield {})))
920919 _ -> yield {}
921920 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
922921 yield (case result.ack of
@@ -925,11 +924,12 @@ serve = \req -> case req of
925924
926925 LookupPresence {pubkey, authorization, respond} -> do
927926 let pubkey = normalizeHex pubkey
927+ let token = extractToken authorization
928928 t <- now
929929 result <- atomic do
930930 pr <- maybePrune t
931931 sessions <- *sessions
932− let auth = authenticateSession sessions (extractToken authorization) t
932+ let auth = authenticateSession sessions token t
933933 case auth of
934934 Unauthorized {} ->
935935 yield {pr: pr, lookup: LookupUnauthorized {}}
@@ -953,7 +953,7 @@ serve = \req -> case req of
953953 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
954954 t <- now
955955 let authorized = verifyFederationAuth authorization
956− let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining
956+ let cappedHops = clamp 0 5 hopsRemaining
957957 let fromServer = normalizeServerName fromServer
958958 if not authorized || cappedHops < 1 || fromServer == serverHostN || count events > maxGossipEvents
959959 then do