@@ -49,7 +49,7 @@ type GossipEventType = Text where \s -> s == "online" || s == "offline"
4949
5050type GossipEvent = {eventType: GossipEventType, pubkey: Text, server: Text, timestamp: Timestamp, ttl: Int<S>}
5151
52−type ValidGossipEntry = {eventType: GossipEventType, pubkey: PubkeyHex, server: Text, expiresAt: Timestamp, timestamp: Timestamp, ttl: Int<S>}
52+type ValidGossipEntry = {eventType: GossipEventType, pubkey: PubkeyHex, server: ServerName, expiresAt: Timestamp, timestamp: Timestamp, ttl: Int<S>}
5353
5454type MessageStatus = Text where \s -> s == "unauthorized" || s == "rejected" || s == "delivered" || s == "federated" || s == "queued"
5555
@@ -145,6 +145,7 @@ route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi
145145
146146any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
147147none = \xs pred -> not (any xs pred)
148+when = \cond action -> if cond then action else yield {}
148149headOf = \items -> fold (\acc x -> case acc of
149150 Nothing {} -> Just {value: x}
150151 _ -> acc) Nothing {} items
@@ -583,13 +584,11 @@ broadcastOfflineGossip = \t peerServers expiredLocal -> do
583584 let expiredEntries = mergeByKey (\a b -> a.pubkey == b.pubkey) [] (do
584585 p <- expiredLocal
585586 yield {pubkey: p.pubkey, ttl: (let rawTtl = (p.expiresAt - p.timestamp) / 1000 in if rawTtl < offlineGossipMinTtl then offlineGossipMinTtl else if rawTtl > maxPresenceTtl then maxPresenceTtl else rawTtl), expiresAt: p.expiresAt})
586− if federationEnabled && count expiredEntries > 0 && count peerServers > 0
587− then do
588− let gossipEvents = do
589− e <- expiredEntries
590− yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}
591− sendBatchedGossip peerServers gossipEvents
592− else yield {}
587+ when (federationEnabled && count expiredEntries > 0 && count peerServers > 0) (do
588+ let gossipEvents = do
589+ e <- expiredEntries
590+ yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}
591+ sendBatchedGossip peerServers gossipEvents)
593592
594593forkBroadcastOffline = \t result ->
595594 fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
@@ -658,14 +657,11 @@ pruneAndRetry = \t -> do
658657 let pendingAcks = do a <- ackedIds; where isFederatedOrigin a.originServer; yield a
659658 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks}
660659 broadcastOfflineGossip t result.peerServers result.expiredLocal
661− if federationEnabled && count result.pendingForwards > 0
662− then retryPendingForwards result.pendingForwards
663− else yield {}
664− if federationEnabled && count result.pendingAcks > 0
665− then do
666− let ackOriginServers = collectPeerServers result.pendingAcks (\a -> a.originServer)
667− sendAcksAndClean result.pendingAcks ackOriginServers
668− else yield {}
660+ when (federationEnabled && count result.pendingForwards > 0)
661+ (retryPendingForwards result.pendingForwards)
662+ when (federationEnabled && count result.pendingAcks > 0) (do
663+ let ackOriginServers = collectPeerServers result.pendingAcks (\a -> a.originServer)
664+ sendAcksAndClean result.pendingAcks ackOriginServers)
669665
670666pollOnce : Text -> Int -> IO {clock} {authorized: Bool, cursor: Int, events: [PollEvent]}
671667pollOnce = \token cursor -> do
@@ -751,40 +747,35 @@ serve = \req -> case req of
751747 let sigValid = case matched of
752748 Nothing {} -> False {}
753749 _ -> verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
754− set *challenges = if sigValid
755− then do
756− c <- challenges
757− where not (c.pubkey == pubkey && c.challenge == challenge)
758− yield c
759− else challenges
760− let sessionsCapped = capPerKey sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t
761− let activeSessionsCapped = do s <- sessionsCapped; where s.expiresAt > t; yield s
762− let globalSessionsCapped = capGlobal activeSessionsCapped maxSessions (\s -> s.expiresAt)
763− set *sessions = if sigValid
764− then union globalSessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]
765− else sessions
766− let presenceWithout = do
767− p <- presence
768− where not (p.pubkey == pubkey && p.server == serverHostN)
769− yield p
770− let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
771− let updatedPresence = if sigValid
772− then union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
773− else presence
774− set *presence = updatedPresence
775− let peerServers = if sigValid && federationEnabled
776− then collectPeerServers updatedPresence (\p -> p.server)
777− else []
778− yield {sigValid: sigValid, peerServers: peerServers, expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}
750+ if not sigValid
751+ then yield {sigValid: False {}, peerServers: [], expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}
752+ else do
753+ set *challenges = do
754+ c <- challenges
755+ where not (c.pubkey == pubkey && c.challenge == challenge)
756+ yield c
757+ let sessionsCapped = capPerKey sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t
758+ let activeSessionsCapped = do s <- sessionsCapped; where s.expiresAt > t; yield s
759+ let globalSessionsCapped = capGlobal activeSessionsCapped maxSessions (\s -> s.expiresAt)
760+ set *sessions = union globalSessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]
761+ let presenceWithout = do
762+ p <- presence
763+ where not (p.pubkey == pubkey && p.server == serverHostN)
764+ yield p
765+ let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
766+ let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
767+ set *presence = updatedPresence
768+ let peerServers = if federationEnabled
769+ then collectPeerServers updatedPresence (\p -> p.server)
770+ else []
771+ yield {sigValid: True {}, peerServers: peerServers, expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}
779772 fork (do
780− if result.sigValid && federationEnabled
781− then sendToEach result.peerServers (\server -> do
773+ when (result.sigValid && federationEnabled)
774+ (sendToEach result.peerServers (\server -> do
782775 _ <- fetch (federationScheme ++ server) (RecvGossip {
783776 authorization: "Bearer " ++ federationSecret,
784777 events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
785− yield {}
786− )
787− else yield {}
778+ yield {}))
788779 broadcastOfflineGossip t result.offlinePeers (do
789780 p <- result.expiredLocal
790781 where not (result.sigValid && p.pubkey == pubkey)
@@ -846,20 +837,17 @@ serve = \req -> case req of
846837 else forwards
847838 yield {status: status, shouldInsert: shouldInsert, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}
848839 fork (do
849− if count result.remoteServers > 0 && result.shouldInsert && federationEnabled
850− then sendToEach result.remoteServers (\server -> do
840+ when (count result.remoteServers > 0 && result.shouldInsert && federationEnabled)
841+ (sendToEach result.remoteServers (\server -> do
851842 fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
852843 toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
853844 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
854845 authorization: "Bearer " ++ federationSecret})
855846 case fwdResult of
856847 Ok _ -> removeForward msgId server
857− Err _ -> yield {}
858− )
859− else yield {}
860− if result.forwardsAtCapacity && count result.remoteServers > 0 && result.shouldInsert
861− then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
862− else yield {}
848+ Err _ -> yield {}))
849+ when (result.forwardsAtCapacity && count result.remoteServers > 0 && result.shouldInsert)
850+ (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking"))
863851 broadcastOfflineGossip t result.peerServers result.expiredLocal)
864852 yield (respond {status: result.status, messageId: if result.shouldInsert then msgId else ""})
865853
@@ -897,41 +885,34 @@ serve = \req -> case req of
897885 yield {ok: True {}, pendingAcks: ackedFederated, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, evictedEntries: evictedEntries}
898886 let originServers = collectPeerServers result.pendingAcks (\a -> a.originServer)
899887 fork (do
900− if result.ok
901− then do
902− if count originServers > 0 && federationEnabled
903− then sendAcksAndClean result.pendingAcks originServers
904− else yield {}
905− let evictedToRetry = do
906− a <- result.evictedEntries
907− where none result.pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)
908− yield a
909− if count evictedToRetry > 0 && federationEnabled
910− then do
911− let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
912− sendToEach evictedOriginServers (\server -> do
913− fetchResult <- fetch (federationScheme ++ server) (FederationAck {
914− authorization: "Bearer " ++ federationSecret,
915− messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
916− originServer: server, fromServer: serverHostN})
917− case fetchResult of
918− Err _ -> do
919− dropped <- atomic do
920− ackedIds <- *ackedIds
921− let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
922− let combined = union ackedIds reinsert
923− let excess = count combined - maxAckedIds
924− let afterEviction = if excess <= 0 then combined
925− else evictOldestAckedIds combined excess
926− set *ackedIds = afterEviction
927− yield excess
928− if dropped > 0
929− then println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server)
930− else yield {}
931− _ -> yield {}
932− )
933− else yield {}
934− else yield {}
888+ when result.ok (do
889+ when (count originServers > 0 && federationEnabled)
890+ (sendAcksAndClean result.pendingAcks originServers)
891+ let evictedToRetry = do
892+ a <- result.evictedEntries
893+ where none result.pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)
894+ yield a
895+ when (count evictedToRetry > 0 && federationEnabled) (do
896+ let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
897+ sendToEach evictedOriginServers (\server -> do
898+ fetchResult <- fetch (federationScheme ++ server) (FederationAck {
899+ authorization: "Bearer " ++ federationSecret,
900+ messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
901+ originServer: server, fromServer: serverHostN})
902+ case fetchResult of
903+ Err _ -> do
904+ dropped <- atomic do
905+ ackedIds <- *ackedIds
906+ let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
907+ let combined = union ackedIds reinsert
908+ let excess = count combined - maxAckedIds
909+ let afterEviction = if excess <= 0 then combined
910+ else evictOldestAckedIds combined excess
911+ set *ackedIds = afterEviction
912+ yield excess
913+ when (dropped > 0)
914+ (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
915+ _ -> yield {})))
935916 broadcastOfflineGossip t result.peerServers result.expiredLocal)
936917 yield (respond {ok: result.ok})
937918
@@ -973,7 +954,7 @@ serve = \req -> case req of
973954 where none offlineEntries (\o -> o.server == fromServer && samePubkeyServer o e && o.timestamp > e.timestamp)
974955 yield e
975956 let affectedKeys = do
976− e <- union onlineEntries offlineEntries
957+ e <- allValidEntries
977958 yield {pubkey: e.pubkey, server: e.server}
978959 let deduped = mergeByPubkeyServer [] filteredOnline
979960 gossipResult <- atomic do
@@ -1027,23 +1008,19 @@ serve = \req -> case req of
10271008 e <- union filteredOnline filteredOffline
10281009 where any gossipResult.appliedKeys (\a -> samePubkeyServer a e)
10291010 yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
1030− if not gossipResult.withinCapacity
1031− then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)
1032− else yield {}
1011+ when (not gossipResult.withinCapacity)
1012+ (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
10331013 fork (do
1034− if validRequest && cappedHops > 1 && federationEnabled && count effectiveEvents > 0
1035− then do
1036− let targets = do
1037− s <- gossipResult.gossipPeerServers
1038− where not (s == fromServer)
1039− yield s
1040− sendToEach targets (\server -> do
1041− _ <- fetch (federationScheme ++ server) (RecvGossip {
1042− authorization: "Bearer " ++ federationSecret,
1043− events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
1044− yield {}
1045− )
1046− else yield {}
1014+ when (validRequest && cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (do
1015+ let targets = do
1016+ s <- gossipResult.gossipPeerServers
1017+ where not (s == fromServer)
1018+ yield s
1019+ sendToEach targets (\server -> do
1020+ _ <- fetch (federationScheme ++ server) (RecvGossip {
1021+ authorization: "Bearer " ++ federationSecret,
1022+ events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
1023+ yield {}))
10471024 broadcastOfflineGossip t gossipResult.peerServers gossipResult.expiredLocal)
10481025 yield (respond {ok: validRequest})
10491026
@@ -1063,10 +1040,11 @@ serve = \req -> case req of
10631040 seqCounter <- *seqCounter
10641041 let seqTs = nextSeqTs t seqCounter
10651042 ackedIds <- *ackedIds
1066− let exists = authorized && validOrigin
1043+ let valid = authorized && validOrigin
1044+ let alreadyExists = valid
10671045 && (any messages (\m -> m.messageId == messageId && m.toKey == toKey)
10681046 || any ackedIds (\a -> a.messageId == messageId && a.toKey == toKey))
1069− let shouldInsert = authorized && validOrigin && not exists
1047+ let shouldInsert = valid && not alreadyExists
10701048 && countMessagesFor messages toKey t < maxMessagesPerRecipient
10711049 set *seqCounter = if shouldInsert then seqTs else seqCounter
10721050 set *messages = if shouldInsert
@@ -1074,7 +1052,7 @@ serve = \req -> case req of
10741052 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
10751053 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
10761054 else messages
1077− yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, ok: shouldInsert || exists}
1055+ yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, ok: shouldInsert || alreadyExists}
10781056 forkBroadcastOffline t result
10791057 yield (respond {ok: result.ok})
10801058
@@ -1104,7 +1082,7 @@ serve = \req -> case req of
11041082 Prune {authorization, respond} -> do
11051083 t <- now
11061084 let authorized = verifyMaintenanceAuth authorization
1107− if authorized then fork (pruneAndRetry t) else yield {}
1085+ when authorized (fork (pruneAndRetry t))
11081086 yield (respond {ok: authorized})
11091087
11101088-- ============================================