@@ -664,7 +664,6 @@ pruneAndRetry = \t -> do
664664 let ackOriginServers = collectPeerServers result.pendingAcks (\a -> a.originServer)
665665 sendAcksAndClean result.pendingAcks ackOriginServers)
666666
667−pollOnce : Text -> Int -> IO {clock} {authorized: Bool, cursor: Int, events: [PollEvent]}
668667pollOnce = \token cursor -> do
669668 t <- now
670669 atomic do
@@ -672,7 +671,7 @@ pollOnce = \token cursor -> do
672671 let auth = authenticateSession sessions token t
673672 case auth of
674673 Unauthorized {} ->
675− yield {events: [], cursor: -1, authorized: False {}}
674+ yield PollUnauthorized {}
676675 Authorized {pubkey} -> do
677676 seqCounter <- *seqCounter
678677 messages <- *messages
@@ -692,19 +691,21 @@ pollOnce = \token cursor -> do
692691 encryptedBlob: m.encryptedBlob,
693692 timestamp: m.timestamp}}
694693 let lastReceivedAt = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) clampedCursor limitedMsgs
695− yield {events: evts, cursor: lastReceivedAt, authorized: True {}}
694+ yield PollAuthorized {events: evts, cursor: lastReceivedAt}
696695
697−pollLoop : Text -> Int -> Int -> IO {clock} {authorized: Bool, cursor: Int, events: [PollEvent]}
698696pollLoop = \token cursor deadline -> do
699697 result <- pollOnce token cursor
700698 t <- now
701− if result.authorized && count result.events == 0 && t < deadline
702− then do
703− let remaining = deadline - t
704− let sleepMs = if remaining > 500 then 500<Ms> else remaining
705− sleep sleepMs
706− pollLoop token cursor deadline
707− else yield result
699+ case result of
700+ PollUnauthorized {} -> yield result
701+ PollAuthorized {events} ->
702+ if count events == 0 && t < deadline
703+ then do
704+ let remaining = deadline - t
705+ let sleepMs = if remaining > 500 then 500<Ms> else remaining
706+ sleep sleepMs
707+ pollLoop token cursor deadline
708+ else yield result
708709
709710-- ============================================
710711-- Request Handler
@@ -749,7 +750,7 @@ serve = \req -> case req of
749750 Nothing {} -> False {}
750751 _ -> verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
751752 if not sigValid
752− then yield VerifyFailed {pr: pr}
753+ then yield {pr: pr, verify: VerifyFailed {}}
753754 else do
754755 set *challenges = do
755756 c <- challenges
@@ -769,22 +770,25 @@ serve = \req -> case req of
769770 let peerServers = if federationEnabled
770771 then collectPeerServers updatedPresence (\p -> p.server)
771772 else []
772− yield VerifyOk {peerServers: peerServers, pr: pr}
773− fork (case result of
774− VerifyFailed {pr} ->
775− broadcastOfflineGossip t pr.peerServers pr.expiredLocal
776− VerifyOk {peerServers, pr} -> do
777− when federationEnabled
778− (sendToEach peerServers (\server -> do
779− _ <- fetch (federationScheme ++ server) (RecvGossip {
780− authorization: "Bearer " ++ federationSecret,
781− events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
782− yield {}))
783− broadcastOfflineGossip t pr.peerServers (do
784− p <- pr.expiredLocal
773+ yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}
774+ fork (do
775+ let expiredLocal = case result.verify of
776+ VerifyOk _ -> do
777+ p <- result.pr.expiredLocal
785778 where not (p.pubkey == pubkey)
786− yield p))
787− yield (case result of
779+ yield p
780+ _ -> result.pr.expiredLocal
781+ case result.verify of
782+ VerifyOk {peerServers} ->
783+ when federationEnabled
784+ (sendToEach peerServers (\server -> do
785+ _ <- fetch (federationScheme ++ server) (RecvGossip {
786+ authorization: "Bearer " ++ federationSecret,
787+ events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
788+ yield {}))
789+ _ -> yield {}
790+ broadcastOfflineGossip t result.pr.peerServers expiredLocal)
791+ yield (case result.verify of
788792 VerifyFailed _ -> respond {token: "", expiresAt: 0}
789793 VerifyOk _ -> respond {token: token, expiresAt: expiry})
790794
@@ -798,7 +802,9 @@ serve = \req -> case req of
798802 pruneResult <- atomic do
799803 maybePrune tAfterPoll
800804 forkBroadcastOffline tAfterPoll pruneResult
801− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
805+ yield (case result of
806+ PollUnauthorized {} -> respond {events: [], cursor: -1, authorized: False {}}
807+ PollAuthorized {events, cursor: c} -> respond {events: events, cursor: c, authorized: True {}})
802808
803809 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
804810 let to = normalizeHex to
@@ -812,11 +818,11 @@ serve = \req -> case req of
812818 let auth = authenticateSession sessions (extractToken authorization) t
813819 case auth of
814820 Unauthorized {} ->
815− yield SendUnauthorized {pr: pr}
821+ yield {pr: pr, send: SendUnauthorized {}}
816822 Authorized {pubkey: fromKey} -> do
817823 messages <- *messages
818824 if countMessagesFor messages to t >= maxMessagesPerRecipient
819− then yield SendRejected {pr: pr}
825+ then yield {pr: pr, send: SendRejected {}}
820826 else do
821827 presence <- *presence
822828 seqCounter <- *seqCounter
@@ -836,26 +842,24 @@ serve = \req -> case req of
836842 set *forwards = if federationEnabled && count forwardServers > 0
837843 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
838844 else forwards
839− yield SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey, pr: pr}
840− fork (case result of
841− SendUnauthorized {pr} ->
842− broadcastOfflineGossip t pr.peerServers pr.expiredLocal
843− SendRejected {pr} ->
844− broadcastOfflineGossip t pr.peerServers pr.expiredLocal
845− SendDone {status, remoteServers, fromKey, pr} -> do
846− when (count remoteServers > 0 && federationEnabled)
847− (sendToEach remoteServers (\server -> do
848− fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
849− toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
850− timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
851− authorization: "Bearer " ++ federationSecret})
852− case fwdResult of
853− Ok _ -> removeForward msgId server
854− Err _ -> yield {}))
855− when (status == "queued" && count remoteServers > 0 && federationEnabled)
856− (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking"))
857− broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
858− yield (case result of
845+ yield {pr: pr, send: SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey}}
846+ fork (do
847+ case result.send of
848+ SendDone {status, remoteServers, fromKey} -> do
849+ when (count remoteServers > 0 && federationEnabled)
850+ (sendToEach remoteServers (\server -> do
851+ fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
852+ toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
853+ timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
854+ authorization: "Bearer " ++ federationSecret})
855+ case fwdResult of
856+ Ok _ -> removeForward msgId server
857+ Err _ -> yield {}))
858+ when (status == "queued" && count remoteServers > 0 && federationEnabled)
859+ (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking"))
860+ _ -> yield {}
861+ broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
862+ yield (case result.send of
859863 SendUnauthorized _ -> respond {status: "unauthorized", messageId: ""}
860864 SendRejected _ -> respond {status: "rejected", messageId: ""}
861865 SendDone {status} -> respond {status: status, messageId: msgId})
@@ -869,7 +873,7 @@ serve = \req -> case req of
869873 let auth = authenticateSession sessions (extractToken authorization) t
870874 case auth of
871875 Unauthorized {} ->
872− yield AckUnauthorized {pr: pr}
876+ yield {pr: pr, ack: AckUnauthorized {}}
873877 Authorized {pubkey} -> do
874878 messages <- *messages
875879 let ackedMessages = do
@@ -891,41 +895,41 @@ serve = \req -> case req of
891895 let evictedEntries = if excessCount <= 0 then []
892896 else diff combinedAckedIds afterEviction
893897 set *ackedIds = afterEviction
894− yield AckDone {pendingAcks: ackedFederated, pr: pr, evictedEntries: evictedEntries}
895− fork (case result of
896− AckUnauthorized {pr} ->
897− broadcastOfflineGossip t pr.peerServers pr.expiredLocal
898− AckDone {pendingAcks, pr, evictedEntries} -> do
899− let originServers = collectPeerServers pendingAcks (\a -> a.originServer)
900− when (count originServers > 0 && federationEnabled)
901− (sendAcksAndClean pendingAcks originServers)
902− let evictedToRetry = do
903− a <- evictedEntries
904− where none pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)
905− yield a
906− when (count evictedToRetry > 0 && federationEnabled) (do
907− let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
908− sendToEach evictedOriginServers (\server -> do
909− fetchResult <- fetch (federationScheme ++ server) (FederationAck {
910− authorization: "Bearer " ++ federationSecret,
911− messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
912− originServer: server, fromServer: serverHostN})
913− case fetchResult of
914− Err _ -> do
915− dropped <- atomic do
916− ackedIds <- *ackedIds
917− let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
918− let combined = union ackedIds reinsert
919− let excess = count combined - maxAckedIds
920− let afterEviction = if excess <= 0 then combined
921− else evictOldestAckedIds combined excess
922− set *ackedIds = afterEviction
923− yield excess
924− when (dropped > 0)
925− (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
926− _ -> yield {}))
927− broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
928− yield (case result of
898+ yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}}
899+ fork (do
900+ case result.ack of
901+ AckDone {pendingAcks, evictedEntries} -> do
902+ let originServers = collectPeerServers pendingAcks (\a -> a.originServer)
903+ when (count originServers > 0 && federationEnabled)
904+ (sendAcksAndClean pendingAcks originServers)
905+ let evictedToRetry = do
906+ a <- evictedEntries
907+ where none pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)
908+ yield a
909+ when (count evictedToRetry > 0 && federationEnabled) (do
910+ let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
911+ sendToEach evictedOriginServers (\server -> do
912+ fetchResult <- fetch (federationScheme ++ server) (FederationAck {
913+ authorization: "Bearer " ++ federationSecret,
914+ messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
915+ originServer: server, fromServer: serverHostN})
916+ case fetchResult of
917+ Err _ -> do
918+ dropped <- atomic do
919+ ackedIds <- *ackedIds
920+ let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
921+ let combined = union ackedIds reinsert
922+ let excess = count combined - maxAckedIds
923+ let afterEviction = if excess <= 0 then combined
924+ else evictOldestAckedIds combined excess
925+ set *ackedIds = afterEviction
926+ yield excess
927+ when (dropped > 0)
928+ (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
929+ _ -> yield {}))
930+ _ -> yield {}
931+ broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
932+ yield (case result.ack of
929933 AckUnauthorized _ -> respond {ok: False {}}
930934 AckDone _ -> respond {ok: True {}})
931935
@@ -938,7 +942,7 @@ serve = \req -> case req of
938942 let auth = authenticateSession sessions (extractToken authorization) t
939943 case auth of
940944 Unauthorized {} ->
941− yield {foundServer: "", authorized: False {}, pr: pr}
945+ yield {pr: pr, lookup: LookupUnauthorized {}}
942946 Authorized _ -> do
943947 presence <- *presence
944948 let valid = do
@@ -949,16 +953,18 @@ serve = \req -> case req of
949953 let foundServer = if any valid (\p -> p.server == serverHostN)
950954 then serverHostN
951955 else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
952− yield {foundServer: foundServer, authorized: True {}, pr: pr}
956+ yield {pr: pr, lookup: LookupDone {foundServer: foundServer}}
953957 forkBroadcastOffline t result.pr
954− yield (respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized})
958+ yield (case result.lookup of
959+ LookupUnauthorized _ -> respond {online: False {}, server: "", authorized: False {}}
960+ LookupDone {foundServer} -> respond {online: not (foundServer == ""), server: foundServer, authorized: True {}})
955961
956962 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
957963 t <- now
958964 let authorized = verifyFederationAuth authorization
959965 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining
960966 let fromServer = normalizeServerName fromServer
961− if not (authorized && cappedHops >= 1 && not (fromServer == serverHostN) && count events <= maxGossipEvents)
967+ if not authorized || cappedHops < 1 || fromServer == serverHostN || count events > maxGossipEvents
962968 then do
963969 pruneResult <- atomic do maybePrune t
964970 forkBroadcastOffline t pruneResult