@@ -34,6 +34,7 @@ challengeExpiry = 60000 -- 1 minute in milliseconds
3434sessionExpiry = 3600000 -- 1 hour in milliseconds
3535onlineGossipTtl = 3600 -- 1 hour in seconds
3636offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events in seconds
37+maxForwardRetriesPerServer = 100
3738
3839-- ============================================
3940-- Types
@@ -158,15 +159,25 @@ isValidHex = \s -> not (s == "") && isValidHexPairs s
158159
159160normalizeHex = \s -> bytesToHex (bytesFromHex s)
160161
162+isDigitChar = \c -> c == "0" || c == "1" || c == "2" || c == "3" || c == "4" || c == "5" || c == "6" || c == "7" || c == "8" || c == "9"
163+
164+isAllowedServerChar = \c ->
165+ isDigitChar c || c == "-" || c == "." || c == ":"
166+ || c == "a" || c == "b" || c == "c" || c == "d" || c == "e" || c == "f"
167+ || c == "g" || c == "h" || c == "i" || c == "j" || c == "k" || c == "l"
168+ || c == "m" || c == "n" || c == "o" || c == "p" || c == "q" || c == "r"
169+ || c == "s" || c == "t" || c == "u" || c == "v" || c == "w" || c == "x"
170+ || c == "y" || c == "z"
171+ || c == "A" || c == "B" || c == "C" || c == "D" || c == "E" || c == "F"
172+ || c == "G" || c == "H" || c == "I" || c == "J" || c == "K" || c == "L"
173+ || c == "M" || c == "N" || c == "O" || c == "P" || c == "Q" || c == "R"
174+ || c == "S" || c == "T" || c == "U" || c == "V" || c == "W" || c == "X"
175+ || c == "Y" || c == "Z"
176+
161177containsBadUrlChar = \s ->
162178 if s == "" then False {}
163− else if take 1 s == "/" || take 1 s == "?" || take 1 s == "#" || take 1 s == " " || take 1 s == "@" || take 1 s == "%" || take 1 s == "\\"
164− || take 1 (bytesToHex (textToBytes (take 1 s))) == "0" || take 1 (bytesToHex (textToBytes (take 1 s))) == "1" || bytesToHex (textToBytes (take 1 s)) == "7f"
165− || not (drop 2 (bytesToHex (textToBytes (take 1 s))) == "")
166− then True {}
167− else containsBadUrlChar (drop 1 s)
168−
169−isDigitChar = \c -> c == "0" || c == "1" || c == "2" || c == "3" || c == "4" || c == "5" || c == "6" || c == "7" || c == "8" || c == "9"
179+ else if not (isAllowedServerChar (take 1 s)) then True {}
180+ else containsBadUrlChar (drop 1 s)
170181
171182isCgnatRange = \s ->
172183 if not (take 4 s == "100.") then False {}
@@ -318,11 +329,13 @@ isBadServerName = \s ->
318329 || take 4 sn == "252." || take 4 sn == "253." || take 4 sn == "254." || take 4 sn == "255."
319330
320331lookupSession = \sessions token t ->
321− single (do
322− s <- sessions
323− where safeCompare s.token token
324− where s.expiresAt > t
325− yield s)
332+ fold (\acc s ->
333+ let matches = safeCompare s.token token in
334+ if matches && s.expiresAt > t
335+ then case acc of
336+ Nothing {} -> Just {value: s}
337+ _ -> acc
338+ else acc) Nothing {} sessions
326339
327340extractToken = \authHeader ->
328341 if (take 1 authHeader == "B" || take 1 authHeader == "b")
@@ -385,6 +398,7 @@ safeCompareLoop = \s1 s2 n result ->
385398safeCompare = \a b ->
386399 safeCompareLoop a b maxTokenLen 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0
387400
401+sendToEach : [Text] -> (Text -> (a {})) -> a {}
388402sendToEach = \servers action ->
389403 if count servers == 0
390404 then yield {}
@@ -393,10 +407,7 @@ sendToEach = \servers action ->
393407 let rest = do s <- servers; where not (s == server); yield s
394408 if server == ""
395409 then yield {}
396− else do
397− catch (action server) (\err ->
398− println ("federation: error reaching " ++ server ++ ": " ++ show err))
399− sendToEach rest action
410+ else action server
400411
401412pruneExpired = \t -> do
402413 challenges <- *challenges
@@ -453,6 +464,7 @@ maybePrune = \t -> do
453464 set *lastPrunedAt = t
454465 pruneExpired t
455466
467+sendBatchedGossipToServer : Text -> [{eventType: Text, pubkey: Text, server: Text, timestamp: Int, ttl: Int}] -> IO {network} {}
456468sendBatchedGossipToServer = \server allEvents ->
457469 if count allEvents == 0
458470 then yield {}
@@ -466,11 +478,12 @@ sendBatchedGossipToServer = \server allEvents ->
466478 events: batch, hopsRemaining: 2, fromServer: serverHostN})
467479 sendBatchedGossipToServer server rest
468480
481+sendBatchedGossip : [Text] -> [{eventType: Text, pubkey: Text, server: Text, timestamp: Int, ttl: Int}] -> IO {} {}
469482sendBatchedGossip = \peerServers allEvents ->
470483 if count allEvents == 0
471484 then yield {}
472− else sendToEach peerServers (\server ->
473− sendBatchedGossipToServer server allEvents)
485+ else
486+ sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents)
474487
475488broadcastOfflineGossip = \t peerServers expiredLocal -> do
476489 let expiredEntries = fold (\acc p ->
@@ -486,29 +499,42 @@ broadcastOfflineGossip = \t peerServers expiredLocal -> do
486499 sendBatchedGossip peerServers gossipEvents
487500 else yield {}
488501
489−retryPendingForwards = \pending ->
490− if count pending == 0
502+retryForwardsForServer : [{encryptedBlob: Text, fromKey: Text, messageId: Text, timestamp: Int, toKey: Text, toServer: Text}] -> Int -> IO {network} {}
503+retryForwardsForServer = \entries remaining ->
504+ if count entries == 0 || remaining <= 0
491505 then yield {}
492506 else do
493− let f = fold (\acc x -> if acc.messageId == "" then x else acc) {messageId: "", toServer: "", toKey: "", fromKey: "", encryptedBlob: "", timestamp: 0} pending
494− let rest = do x <- pending; where not (x.messageId == f.messageId && x.toServer == f.toServer); yield x
507+ let f = fold (\acc x -> if acc.messageId == "" then x else acc) {messageId: "", toServer: "", toKey: "", fromKey: "", encryptedBlob: "", timestamp: 0} entries
508+ let rest = do x <- entries; where not (x.messageId == f.messageId); yield x
495509 if f.messageId == ""
496510 then yield {}
497511 else do
498− catch (do
499− fetch (federationScheme ++ f.toServer) (ForwardMessage {
500− toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob,
501− timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
502− authorization: "Bearer " ++ federationSecret})
503− atomic do
504− forwards <- *forwards
505− set *forwards = do
506− x <- forwards
507− where not (x.messageId == f.messageId && x.toServer == f.toServer)
508− yield x) (\err ->
509− println ("prune: retry forward to " ++ f.toServer ++ " failed: " ++ show err))
510− retryPendingForwards rest
511−
512+ result <- fetch (federationScheme ++ f.toServer) (ForwardMessage {
513+ toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob,
514+ timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
515+ authorization: "Bearer " ++ federationSecret})
516+ case result of
517+ Ok _ -> do
518+ atomic do
519+ forwards <- *forwards
520+ set *forwards = do
521+ x <- forwards
522+ where not (x.messageId == f.messageId && x.toServer == f.toServer)
523+ yield x
524+ yield {}
525+ Err _ -> yield {}
526+ retryForwardsForServer rest (remaining - 1)
527+
528+retryPendingForwards : [{encryptedBlob: Text, fromKey: Text, messageId: Text, timestamp: Int, toKey: Text, toServer: Text}] -> IO {network} {}
529+retryPendingForwards = \pending -> do
530+ let servers = fold (\acc x ->
531+ if count (do s <- acc; where s == x.toServer; yield s) > 0 then acc
532+ else union acc [x.toServer]) [] pending
533+ sendToEach servers (\server ->
534+ retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)
535+ yield {}
536+
537+sendAcksAndClean : [{messageId: Text, originServer: Text, | a}] -> [Text] -> IO {network} {}
512538sendAcksAndClean = \pendingAcks servers ->
513539 if count servers == 0
514540 then yield {}
@@ -519,24 +545,25 @@ sendAcksAndClean = \pendingAcks servers ->
519545 then yield {}
520546 else do
521547 let serverAcks = do a <- pendingAcks; where a.originServer == server; yield a
522− catch (do
523− fetch (federationScheme ++ server) (FederationAck {
524− authorization: "Bearer " ++ federationSecret,
525− messageIds: (do a <- serverAcks; yield a.messageId),
526− originServer: server, fromServer: serverHostN})
527− atomic do
528− ackedIds <- *ackedIds
529− let ackMsgIds = do a <- serverAcks; yield a.messageId
530− set *ackedIds = do
531− a <- ackedIds
532− where not (a.originServer == server && count (do mid <- ackMsgIds; where mid == a.messageId; yield mid) > 0)
533− yield a) (\err ->
534− println ("federation: error sending ack to " ++ server ++ ": " ++ show err))
548+ result <- fetch (federationScheme ++ server) (FederationAck {
549+ authorization: "Bearer " ++ federationSecret,
550+ messageIds: (do a <- serverAcks; yield a.messageId),
551+ originServer: server, fromServer: serverHostN})
552+ case result of
553+ Ok _ ->
554+ atomic do
555+ ackedIds <- *ackedIds
556+ let ackMsgIds = do a <- serverAcks; yield a.messageId
557+ set *ackedIds = do
558+ a <- ackedIds
559+ where not (a.originServer == server && count (do mid <- ackMsgIds; where mid == a.messageId; yield mid) > 0)
560+ yield a
561+ Err _ -> yield {}
535562 sendAcksAndClean pendingAcks rest
536563
537−pollOnce = \token cursor ->
564+pollOnce = \token cursor -> do
565+ t <- now
538566 atomic do
539− t <- now
540567 sessions <- *sessions
541568 messages <- *messages
542569 seqCounter <- *seqCounter
@@ -584,6 +611,7 @@ pollOnce = \token cursor ->
584611 _ -> lastReceivedAt
585612 yield {events: evts, cursor: newCursor, authorized: authorized}
586613
614+pollLoop : Text -> Int -> Int -> IO {clock} {authorized: Bool, cursor: Int, events: [{eventType: Text, id: Text, payload: Text}]}
587615pollLoop = \token cursor deadline -> do
588616 result <- pollOnce token cursor
589617 t <- now
@@ -635,9 +663,9 @@ serve = \req -> case req of
635663 globalCapped
636664 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
637665 yield pr
638− respond {challenge: challenge}
639666 broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal
640− else respond {challenge: ""}
667+ yield (respond {challenge: challenge})
668+ else yield (respond {challenge: ""})
641669
642670 VerifyAuth {pubkey, challenge, signature, respond} -> do
643671 let pubkey = if isValidHex pubkey then normalizeHex pubkey else pubkey
@@ -718,19 +746,22 @@ serve = \req -> case req of
718746 else union acc [p.server]) [] updatedPresence
719747 else []
720748 yield {sigValid: sigValid, peerServers: peerServers, expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}
721− if result.sigValid
722− then respond {token: token, expiresAt: expiry}
723− else respond {token: "", expiresAt: 0}
724749 if result.sigValid && not (federationSecret == "")
725− then sendToEach result.peerServers (\server ->
750+ then sendToEach result.peerServers (\server -> do
726751 fetch (federationScheme ++ server) (RecvGossip {
727752 authorization: "Bearer " ++ federationSecret,
728− events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN}))
753+ events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
754+ yield {}
755+ )
756+
729757 else yield {}
730758 broadcastOfflineGossip t result.offlinePeers (do
731759 p <- result.expiredLocal
732− where not (p.pubkey == pubkey)
760+ where not (result.sigValid && p.pubkey == pubkey)
733761 yield p)
762+ yield (if result.sigValid
763+ then respond {token: token, expiresAt: expiry}
764+ else respond {token: "", expiresAt: 0})
734765
735766 Poll {cursor, timeout, authorization, respond} -> do
736767 t <- now
@@ -738,11 +769,11 @@ serve = \req -> case req of
738769 let deadline = t + clampedTimeout
739770 let token = extractToken authorization
740771 result <- pollLoop token cursor deadline
741− respond {events: result.events, cursor: result.cursor, authorized: result.authorized}
742772 tAfterPoll <- now
743773 pruneResult <- atomic do
744774 maybePrune tAfterPoll
745775 broadcastOfflineGossip tAfterPoll pruneResult.peerServers pruneResult.expiredLocal
776+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
746777
747778 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
748779 let validInput = not (to == "") && not (encryptedBlob == "") && not (tooLong to maxPubkeyLen) && not (tooLong encryptedBlob maxBlobLen) && isValidHex to && isValidHex encryptedBlob
@@ -808,7 +839,6 @@ serve = \req -> case req of
808839 else forwards
809840 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}
810841 let finalStatus = if result.status == "federated" && (federationSecret == "" || result.forwardsAtCapacity) then "queued" else result.status
811− respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else if result.status == "invalid" then "" else if result.status == "rejected" then "" else msgId)}
812842 if count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (result.status == "rejected") && not (federationSecret == "")
813843 then sendToEach result.remoteServers (\server -> do
814844 fetch (federationScheme ++ server) (ForwardMessage {
@@ -826,6 +856,7 @@ serve = \req -> case req of
826856 then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
827857 else yield {}
828858 broadcastOfflineGossip t result.peerServers result.expiredLocal
859+ yield (respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else if result.status == "invalid" then "" else if result.status == "rejected" then "" else msgId)})
829860
830861 AckMessages {messageIds, authorization, respond} -> do
831862 t <- now
@@ -877,7 +908,6 @@ serve = \req -> case req of
877908 let originServers = fold (\acc m ->
878909 if count (do a <- acc; where a == m.originServer; yield a) > 0 then acc
879910 else union acc [m.originServer]) [] fedMessages
880− respond {ok: result.ok}
881911 if result.ok && count originServers > 0 && not (federationSecret == "")
882912 then do
883913 let pendingAcks = do
@@ -896,29 +926,27 @@ serve = \req -> case req of
896926 let evictedOriginServers = fold (\acc a ->
897927 if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
898928 else union acc [a.originServer]) [] evictedToRetry
899− sendToEach evictedOriginServers (\server ->
900− catch (do
901− fetch (federationScheme ++ server) (FederationAck {
929+ sendToEach evictedOriginServers (\server -> do
930+ result <- fetch (federationScheme ++ server) (FederationAck {
902931 authorization: "Bearer " ++ federationSecret,
903932 messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
904933 originServer: server, fromServer: serverHostN})
905− yield {}) (\err -> do
906− println ("federation: error sending evicted ack to " ++ server ++ ": " ++ show err)
907− evicted <- 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 (count combined - count afterEviction)
916− if evicted > 0
917− then println ("federation: evicted " ++ show evicted ++ " old ack entries to reinsert failed acks for " ++ server)
918− else yield {}))
934+ case result of
935+ Err _ -> atomic do
936+ ackedIds <- *ackedIds
937+ let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
938+ let combined = union ackedIds reinsert
939+ let excess = count combined - maxAckedIds
940+ let afterEviction = if excess <= 0 then combined
941+ else evictOldestAckedIds combined excess
942+ set *ackedIds = afterEviction
943+ yield {}
944+ _ -> yield {}
945+ )
919946 else yield {}
920947 else yield {}
921948 broadcastOfflineGossip t result.peerServers result.expiredLocal
949+ yield (respond {ok: result.ok})
922950
923951 LookupPresence {pubkey, authorization, respond} -> do
924952 let validPubkey = not (pubkey == "") && not (tooLong pubkey maxPubkeyLen) && isValidHex pubkey
@@ -946,8 +974,8 @@ serve = \req -> case req of
946974 then serverHostN
947975 else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
948976 yield {foundServer: foundServer, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
949− respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized}
950977 broadcastOfflineGossip t result.peerServers result.expiredLocal
978+ yield (respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized})
951979
952980 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
953981 t <- now
@@ -1176,7 +1204,6 @@ serve = \req -> case req of
11761204 e <- validEvents
11771205 where count (do a <- gossipResult.appliedKeys; where a.pubkey == normalizeHex e.pubkey; where a.server == normalizeServerName e.server; yield a) > 0
11781206 yield {eventType: e.eventType, pubkey: normalizeHex e.pubkey, server: normalizeServerName e.server, timestamp: e.timestamp, ttl: clampTtl e.ttl}
1179− respond {ok: validRequest}
11801207 if not gossipResult.withinCapacity
11811208 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)
11821209 else yield {}
@@ -1186,12 +1213,15 @@ serve = \req -> case req of
11861213 s <- gossipResult.gossipPeerServers
11871214 where not (s == fromServer)
11881215 yield s
1189− sendToEach targets (\server ->
1216+ sendToEach targets (\server -> do
11901217 fetch (federationScheme ++ server) (RecvGossip {
11911218 authorization: "Bearer " ++ federationSecret,
1192− events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN}))
1219+ events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
1220+ yield {}
1221+ )
11931222 else yield {}
11941223 broadcastOfflineGossip t gossipResult.peerServers gossipResult.expiredLocal
1224+ yield (respond {ok: validRequest})
11951225
11961226 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
11971227 let validFields = not (toKey == "") && not (fromKey == "") && not (encryptedBlob == "") && not (messageId == "") && not (tooLong toKey maxPubkeyLen) && not (tooLong fromKey maxPubkeyLen) && not (tooLong encryptedBlob maxBlobLen) && not (tooLong messageId maxMessageIdLen) && isValidHex toKey && isValidHex fromKey && isValidHex encryptedBlob && isValidHex messageId
@@ -1237,8 +1267,8 @@ serve = \req -> case req of
12371267 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
12381268 else messages
12391269 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, inserted: shouldInsert, alreadyExists: exists}
1240− respond {ok: result.inserted || (authorized && validOrigin && validFields && result.alreadyExists)}
12411270 broadcastOfflineGossip t result.peerServers result.expiredLocal
1271+ yield (respond {ok: result.inserted || (authorized && validOrigin && validFields && result.alreadyExists)})
12421272
12431273 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
12441274 let fromServer = if tooLong fromServer maxServerNameLen then fromServer else normalizeServerName fromServer
@@ -1267,8 +1297,8 @@ serve = \req -> case req of
12671297 then diff forwards forwardsToRemove
12681298 else forwards
12691299 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}
1270− respond {ok: pruneResult.validAck}
12711300 broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal
1301+ yield (respond {ok: pruneResult.validAck})
12721302
12731303 Prune {authorization, respond} -> do
12741304 t <- now
@@ -1289,22 +1319,19 @@ serve = \req -> case req of
12891319 where not (a.originServer == serverHostN)
12901320 yield a
12911321 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
1292− respond {ok: True {}}
1293− catch (broadcastOfflineGossip t result.peerServers result.expiredLocal) (\err ->
1294− println ("prune: gossip failed: " ++ show err))
1295− catch (if not (federationSecret == "") && count result.pendingForwards > 0
1322+ broadcastOfflineGossip t result.peerServers result.expiredLocal
1323+ if not (federationSecret == "") && count result.pendingForwards > 0
12961324 then retryPendingForwards result.pendingForwards
1297− else yield {}) (\err ->
1298− println ("prune: forward retry failed: " ++ show err))
1299− catch (if not (federationSecret == "") && count result.pendingAcks > 0
1325+ else yield {}
1326+ if not (federationSecret == "") && count result.pendingAcks > 0
13001327 then do
13011328 let ackOriginServers = fold (\acc a ->
13021329 if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
13031330 else union acc [a.originServer]) [] result.pendingAcks
13041331 sendAcksAndClean result.pendingAcks ackOriginServers
1305− else yield {}) (\err ->
1306− println ("prune: ack send failed: " ++ show err))
1307− else respond {ok: False {}}
1332+ else yield {}
1333+ yield (respond {ok: True {}})
1334+ else yield (respond {ok: False {}})
13081335
13091336-- ============================================
13101337-- Entry Point
@@ -1313,39 +1340,34 @@ serve = \req -> case req of
13131340backgroundPrune = do
13141341 sleep 30000
13151342 t <- now
1316− catch (do
1317− result <- atomic do
1318− set *lastPrunedAt = t
1319− pr <- pruneExpired t
1320− forwards <- *forwards
1321− let pendingForwards = do
1322− f <- forwards
1323− yield {toServer: f.toServer, toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, messageId: f.messageId}
1324− ackedIds <- *ackedIds
1325− let pendingAcks = do
1326− a <- ackedIds
1327− where not (a.originServer == "")
1328− where not (a.originServer == serverHostN)
1329− yield a
1330− yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
1331− catch (broadcastOfflineGossip t result.peerServers result.expiredLocal) (\err ->
1332− println ("background prune: gossip failed: " ++ show err))
1333− catch (if not (federationSecret == "") && count result.pendingForwards > 0
1334− then retryPendingForwards result.pendingForwards
1335− else yield {}) (\err ->
1336− println ("background prune: forward retry failed: " ++ show err))
1337− catch (if not (federationSecret == "") && count result.pendingAcks > 0
1338− then do
1339− let ackOriginServers = fold (\acc a ->
1340− if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1341− else union acc [a.originServer]) [] result.pendingAcks
1342− sendAcksAndClean result.pendingAcks ackOriginServers
1343− else yield {}) (\err ->
1344− println ("background prune: ack send failed: " ++ show err))) (\err ->
1345− println ("background prune: " ++ show err))
1346− spawn backgroundPrune
1343+ result <- atomic do
1344+ set *lastPrunedAt = t
1345+ pr <- pruneExpired t
1346+ forwards <- *forwards
1347+ let pendingForwards = do
1348+ f <- forwards
1349+ yield {toServer: f.toServer, toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, messageId: f.messageId}
1350+ ackedIds <- *ackedIds
1351+ let pendingAcks = do
1352+ a <- ackedIds
1353+ where not (a.originServer == "")
1354+ where not (a.originServer == serverHostN)
1355+ yield a
1356+ yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
1357+ broadcastOfflineGossip t result.peerServers result.expiredLocal
1358+ if not (federationSecret == "") && count result.pendingForwards > 0
1359+ then retryPendingForwards result.pendingForwards
1360+ else yield {}
1361+ if not (federationSecret == "") && count result.pendingAcks > 0
1362+ then do
1363+ let ackOriginServers = fold (\acc a ->
1364+ if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1365+ else union acc [a.originServer]) [] result.pendingAcks
1366+ sendAcksAndClean result.pendingAcks ackOriginServers
1367+ else yield {}
1368+ fork backgroundPrune
13471369
13481370main = do
13491371 println "Skrepka relay server starting on port 8080..."
1350− spawn backgroundPrune
1372+ fork backgroundPrune
13511373 listen 8080 serve