@@ -479,8 +479,9 @@ sendBatchedGossipToServer = \server allEvents ->
479479 authorization: "Bearer " ++ federationSecret,
480480 events: batch, hopsRemaining: 2, fromServer: serverHostN})
481481 case batchResult of
482− Ok _ -> sendBatchedGossipToServer server rest
482+ Ok _ -> yield {}
483483 Err _ -> yield {}
484+ sendBatchedGossipToServer server rest
484485
485486sendBatchedGossip : [Text] -> [{eventType: Text, pubkey: Text, server: Text, timestamp: Int, ttl: Int}] -> IO {} {}
486487sendBatchedGossip = \peerServers allEvents ->
@@ -667,7 +668,7 @@ serve = \req -> case req of
667668 globalCapped
668669 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
669670 yield pr
670− broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal
671+ fork (broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal)
671672 yield (respond {challenge: challenge})
672673 else yield (respond {challenge: ""})
673674
@@ -750,19 +751,21 @@ serve = \req -> case req of
750751 else union acc [p.server]) [] updatedPresence
751752 else []
752753 yield {sigValid: sigValid, peerServers: peerServers, expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}
753− if result.sigValid && not (federationSecret == "")
754− then sendToEach result.peerServers (\server -> do
755− fetch (federationScheme ++ server) (RecvGossip {
756− authorization: "Bearer " ++ federationSecret,
757− events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
758− yield {}
759− )
760−
761− else yield {}
762− broadcastOfflineGossip t result.offlinePeers (do
763− p <- result.expiredLocal
764− where not (result.sigValid && p.pubkey == pubkey)
765− yield p)
754+ fork (do
755+ if result.sigValid && not (federationSecret == "")
756+ then sendToEach result.peerServers (\server -> do
757+ fetchResult <- fetch (federationScheme ++ server) (RecvGossip {
758+ authorization: "Bearer " ++ federationSecret,
759+ events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
760+ case fetchResult of
761+ Ok _ -> yield {}
762+ Err _ -> yield {}
763+ )
764+ else yield {}
765+ broadcastOfflineGossip t result.offlinePeers (do
766+ p <- result.expiredLocal
767+ where not (result.sigValid && p.pubkey == pubkey)
768+ yield p))
766769 yield (if result.sigValid
767770 then respond {token: token, expiresAt: expiry}
768771 else respond {token: "", expiresAt: 0})
@@ -776,7 +779,7 @@ serve = \req -> case req of
776779 tAfterPoll <- now
777780 pruneResult <- atomic do
778781 maybePrune tAfterPoll
779− broadcastOfflineGossip tAfterPoll pruneResult.peerServers pruneResult.expiredLocal
782+ fork (broadcastOfflineGossip tAfterPoll pruneResult.peerServers pruneResult.expiredLocal)
780783 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
781784
782785 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
@@ -838,30 +841,31 @@ serve = \req -> case req of
838841 else "queued"
839842 forwards <- *forwards
840843 let forwardsAtCapacity = count forwards >= maxForwardsEntries
841− set *forwards = if shouldInsert && not forwardsAtCapacity && count remoteServers > 0
844+ set *forwards = if shouldInsert && not forwardsAtCapacity && count remoteServers > 0 && not (federationSecret == "")
842845 then union forwards (do s <- remoteServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
843846 else forwards
844847 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}
845848 let finalStatus = if result.status == "federated" && (federationSecret == "" || result.forwardsAtCapacity) then "queued" else result.status
846− if count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (result.status == "rejected") && not (federationSecret == "")
847− then sendToEach result.remoteServers (\server -> do
848− fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
849− toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
850− timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
851− authorization: "Bearer " ++ federationSecret})
852− case fwdResult of
853− Ok _ -> atomic do
854− forwards <- *forwards
855− set *forwards = do
856− x <- forwards
857− where not (x.messageId == msgId && x.toServer == server)
858− yield x
859− Err _ -> yield {})
860− else yield {}
861− if result.forwardsAtCapacity && count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (result.status == "rejected")
862− then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
863− else yield {}
864− broadcastOfflineGossip t result.peerServers result.expiredLocal
849+ fork (do
850+ if count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (result.status == "rejected") && not (federationSecret == "")
851+ then sendToEach result.remoteServers (\server -> do
852+ fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
853+ toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
854+ timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
855+ authorization: "Bearer " ++ federationSecret})
856+ case fwdResult of
857+ Ok _ -> atomic do
858+ forwards <- *forwards
859+ set *forwards = do
860+ x <- forwards
861+ where not (x.messageId == msgId && x.toServer == server)
862+ yield x
863+ Err _ -> yield {})
864+ else yield {}
865+ if result.forwardsAtCapacity && count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (result.status == "rejected")
866+ then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
867+ else yield {}
868+ broadcastOfflineGossip t result.peerServers result.expiredLocal)
865869 yield (respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else if result.status == "invalid" then "" else if result.status == "rejected" then "" else msgId)})
866870
867871 AckMessages {messageIds, authorization, respond} -> do
@@ -914,44 +918,45 @@ serve = \req -> case req of
914918 let originServers = fold (\acc m ->
915919 if count (do a <- acc; where a == m.originServer; yield a) > 0 then acc
916920 else union acc [m.originServer]) [] fedMessages
917− if result.ok && count originServers > 0 && not (federationSecret == "")
918− then do
919− let pendingAcks = do
920− m <- fedMessages
921− yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
922− sendAcksAndClean pendingAcks originServers
923− else yield {}
924− if result.ok
925− then do
926− let evictedToRetry = do
927− a <- result.evictedEntries
928− where count (do m <- fedMessages; where m.messageId == a.messageId; where m.toKey == a.toKey; yield m) == 0
929− yield a
930− if count evictedToRetry > 0 && not (federationSecret == "")
931− then do
932− let evictedOriginServers = fold (\acc a ->
933− if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
934− else union acc [a.originServer]) [] evictedToRetry
935− sendToEach evictedOriginServers (\server -> do
936− result <- fetch (federationScheme ++ server) (FederationAck {
937− authorization: "Bearer " ++ federationSecret,
938− messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
939− originServer: server, fromServer: serverHostN})
940− case result of
941− Err _ -> atomic do
942− ackedIds <- *ackedIds
943− let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
944− let combined = union ackedIds reinsert
945− let excess = count combined - maxAckedIds
946− let afterEviction = if excess <= 0 then combined
947− else evictOldestAckedIds combined excess
948− set *ackedIds = afterEviction
949− yield {}
950− _ -> yield {}
951− )
952− else yield {}
953− else yield {}
954− broadcastOfflineGossip t result.peerServers result.expiredLocal
921+ fork (do
922+ if result.ok && count originServers > 0 && not (federationSecret == "")
923+ then do
924+ let pendingAcks = do
925+ m <- fedMessages
926+ yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
927+ sendAcksAndClean pendingAcks originServers
928+ else yield {}
929+ if result.ok
930+ then do
931+ let evictedToRetry = do
932+ a <- result.evictedEntries
933+ where count (do m <- fedMessages; where m.messageId == a.messageId; where m.toKey == a.toKey; yield m) == 0
934+ yield a
935+ if count evictedToRetry > 0 && not (federationSecret == "")
936+ then do
937+ let evictedOriginServers = fold (\acc a ->
938+ if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
939+ else union acc [a.originServer]) [] evictedToRetry
940+ sendToEach evictedOriginServers (\server -> do
941+ result <- fetch (federationScheme ++ server) (FederationAck {
942+ authorization: "Bearer " ++ federationSecret,
943+ messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
944+ originServer: server, fromServer: serverHostN})
945+ case result of
946+ Err _ -> atomic do
947+ ackedIds <- *ackedIds
948+ let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
949+ let combined = union ackedIds reinsert
950+ let excess = count combined - maxAckedIds
951+ let afterEviction = if excess <= 0 then combined
952+ else evictOldestAckedIds combined excess
953+ set *ackedIds = afterEviction
954+ yield {}
955+ _ -> yield {}
956+ )
957+ else yield {}
958+ else yield {}
959+ broadcastOfflineGossip t result.peerServers result.expiredLocal)
955960 yield (respond {ok: result.ok})
956961
957962 LookupPresence {pubkey, authorization, respond} -> do
@@ -980,7 +985,7 @@ serve = \req -> case req of
980985 then serverHostN
981986 else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
982987 yield {foundServer: foundServer, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
983− broadcastOfflineGossip t result.peerServers result.expiredLocal
988+ fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
984989 yield (respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized})
985990
986991 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
@@ -1212,20 +1217,23 @@ serve = \req -> case req of
12121217 if not gossipResult.withinCapacity
12131218 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)
12141219 else yield {}
1215− if validRequest && cappedHops > 1 && not (federationSecret == "") && count effectiveEvents > 0
1216− then do
1217− let targets = do
1218− s <- gossipResult.gossipPeerServers
1219− where not (s == fromServer)
1220− yield s
1221− sendToEach targets (\server -> do
1222− fetch (federationScheme ++ server) (RecvGossip {
1223− authorization: "Bearer " ++ federationSecret,
1224− events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
1225− yield {}
1226− )
1227− else yield {}
1228− broadcastOfflineGossip t gossipResult.peerServers gossipResult.expiredLocal
1220+ fork (do
1221+ if validRequest && cappedHops > 1 && not (federationSecret == "") && count effectiveEvents > 0
1222+ then do
1223+ let targets = do
1224+ s <- gossipResult.gossipPeerServers
1225+ where not (s == fromServer)
1226+ yield s
1227+ sendToEach targets (\server -> do
1228+ fetchResult <- fetch (federationScheme ++ server) (RecvGossip {
1229+ authorization: "Bearer " ++ federationSecret,
1230+ events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
1231+ case fetchResult of
1232+ Ok _ -> yield {}
1233+ Err _ -> yield {}
1234+ )
1235+ else yield {}
1236+ broadcastOfflineGossip t gossipResult.peerServers gossipResult.expiredLocal)
12291237 yield (respond {ok: validRequest})
12301238
12311239 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
@@ -1272,7 +1280,7 @@ serve = \req -> case req of
12721280 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
12731281 else messages
12741282 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, inserted: shouldInsert, alreadyExists: exists}
1275− broadcastOfflineGossip t result.peerServers result.expiredLocal
1283+ fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
12761284 yield (respond {ok: result.inserted || (authorized && validOrigin && validFields && result.alreadyExists)})
12771285
12781286 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
@@ -1302,7 +1310,7 @@ serve = \req -> case req of
13021310 then diff forwards forwardsToRemove
13031311 else forwards
13041312 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}
1305− broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal
1313+ fork (broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal)
13061314 yield (respond {ok: pruneResult.validAck})
13071315
13081316 Prune {authorization, respond} -> do
@@ -1324,17 +1332,18 @@ serve = \req -> case req of
13241332 where not (a.originServer == serverHostN)
13251333 yield a
13261334 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
1327− broadcastOfflineGossip t result.peerServers result.expiredLocal
1328− if not (federationSecret == "") && count result.pendingForwards > 0
1329− then retryPendingForwards result.pendingForwards
1330− else yield {}
1331− if not (federationSecret == "") && count result.pendingAcks > 0
1332− then do
1333− let ackOriginServers = fold (\acc a ->
1334− if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1335− else union acc [a.originServer]) [] result.pendingAcks
1336− sendAcksAndClean result.pendingAcks ackOriginServers
1337− else yield {}
1335+ fork (do
1336+ broadcastOfflineGossip t result.peerServers result.expiredLocal
1337+ if not (federationSecret == "") && count result.pendingForwards > 0
1338+ then retryPendingForwards result.pendingForwards
1339+ else yield {}
1340+ if not (federationSecret == "") && count result.pendingAcks > 0
1341+ then do
1342+ let ackOriginServers = fold (\acc a ->
1343+ if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1344+ else union acc [a.originServer]) [] result.pendingAcks
1345+ sendAcksAndClean result.pendingAcks ackOriginServers
1346+ else yield {})
13381347 yield (respond {ok: True {}})
13391348 else yield (respond {ok: False {}})
13401349