@@ -56,21 +56,21 @@ type Timestamp = Int<Ms>
5656-- Persisted State
5757-- ============================================
5858
59−*challenges : [{pubkey: Text, challenge: Text, expiresAt: Int}]
59+*challenges : [{pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp}]
6060
61−*sessions : [{token: Text, pubkey: Text, expiresAt: Int}]
61+*sessions : [{token: Text, pubkey: PubkeyHex, expiresAt: Timestamp}]
6262
63−*messages : [{messageId: Text, toKey: Text, fromKey: Text, encryptedBlob: Text, timestamp: Timestamp, originServer: Text, receivedAt: Int}]
63+*messages : [{messageId: MessageIdHex, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp}]
6464
65−*presence : [{pubkey: Text, server: Text, expiresAt: Int, timestamp: Timestamp}]
65+*presence : [{pubkey: PubkeyHex, server: Text, expiresAt: Timestamp, timestamp: Timestamp}]
6666
67−*seqCounter : Int
67+*seqCounter : Timestamp
6868
69−*forwards : [{messageId: Text, toServer: Text, createdAt: Int, toKey: Text, fromKey: Text, encryptedBlob: Text, timestamp: Timestamp}]
69+*forwards : [{messageId: MessageIdHex, toServer: Text, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}]
7070
71−*ackedIds : [{messageId: Text, toKey: Text, ackedAt: Int, originServer: Text}]
71+*ackedIds : [{messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: Text}]
7272
73−*lastPrunedAt : Int
73+*lastPrunedAt : Timestamp
7474
7575-- ============================================
7676-- Routes
@@ -108,7 +108,7 @@ route FederationApi where
108108 POST {toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex} /forward headers {authorization: Text}
109109 -> {ok: Bool}
110110 = ForwardMessage
111− POST {messageIds: [Text], originServer: Text, fromServer: ServerName} /ack headers {authorization: Text}
111+ POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName} /ack headers {authorization: Text}
112112 -> {ok: Bool}
113113 = FederationAck
114114
@@ -345,13 +345,7 @@ lookupSession = \sessions token t ->
345345 else acc) Nothing {} sessions
346346
347347extractToken = \authHeader ->
348− if (take 1 authHeader == "B" || take 1 authHeader == "b")
349− && (take 1 (drop 1 authHeader) == "e" || take 1 (drop 1 authHeader) == "E")
350− && (take 1 (drop 2 authHeader) == "a" || take 1 (drop 2 authHeader) == "A")
351− && (take 1 (drop 3 authHeader) == "r" || take 1 (drop 3 authHeader) == "R")
352− && (take 1 (drop 4 authHeader) == "e" || take 1 (drop 4 authHeader) == "E")
353− && (take 1 (drop 5 authHeader) == "r" || take 1 (drop 5 authHeader) == "R")
354− && take 1 (drop 6 authHeader) == " "
348+ if toLower (take 7 authHeader) == "bearer "
355349 && not (tooLong (drop 7 authHeader) maxTokenLen)
356350 then drop 7 authHeader
357351 else ""
@@ -405,6 +399,21 @@ safeCompareLoop = \s1 s2 n result ->
405399safeCompare = \a b ->
406400 safeCompareLoop a b maxTokenLen 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0
407401
402+evictOneOldest = \items getKey ->
403+ let oldest = fold (\acc x -> let k = getKey x in if acc == -1 || k < acc then k else acc) (-1) items in
404+ if oldest == -1 then items
405+ else (fold (\acc x ->
406+ if not acc.removed && getKey x == oldest
407+ then {removed: True {}, items: acc.items}
408+ else {removed: acc.removed, items: union acc.items [x]}) {removed: False {}, items: []} items).items
409+
410+collectPeerServers = \items getServer ->
411+ fold (\acc item ->
412+ let s = getServer item in
413+ if s == serverHostN then acc
414+ else if count (do a <- acc; where a == s; yield a) > 0 then acc
415+ else union acc [s]) [] items
416+
408417sendToEach : [Text] -> (Text -> (a {})) -> a {}
409418sendToEach = \servers action ->
410419 if count servers == 0
@@ -428,10 +437,7 @@ pruneExpired = \t -> do
428437 where s.expiresAt > t
429438 yield s
430439 presenceBefore <- *presence
431− let peerServersBefore = fold (\acc p ->
432− if p.server == serverHostN then acc
433− else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
434− else union acc [p.server]) [] presenceBefore
440+ let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)
435441 let presenceAfter = do
436442 p <- presenceBefore
437443 where p.expiresAt > t
@@ -534,9 +540,7 @@ retryForwardsForServer = \entries remaining ->
534540
535541retryPendingForwards : [{encryptedBlob: Text, fromKey: Text, messageId: Text, timestamp: Timestamp, toKey: Text, toServer: Text}] -> IO {network} {}
536542retryPendingForwards = \pending -> do
537− let servers = fold (\acc x ->
538− if count (do s <- acc; where s == x.toServer; yield s) > 0 then acc
539− else union acc [x.toServer]) [] pending
543+ let servers = collectPeerServers pending (\x -> x.toServer)
540544 sendToEach servers (\server ->
541545 retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)
542546 yield {}
@@ -567,16 +571,33 @@ sendBatchedAcksToServer = \serverAcks server ->
567571
568572sendAcksAndClean : [{messageId: Text, toKey: Text, originServer: Text, | a}] -> [Text] -> IO {network} {}
569573sendAcksAndClean = \pendingAcks servers ->
570− if count servers == 0
571− then yield {}
572− else do
573− let server = fold (\acc s -> if acc == "" then s else acc) "" servers
574− let rest = do s <- servers; where not (s == server); yield s
575− if server == ""
576− then yield {}
577− else do
578− fork (sendBatchedAcksToServer (do a <- pendingAcks; where a.originServer == server; yield a) server)
579− sendAcksAndClean pendingAcks rest
574+ sendToEach servers (\server ->
575+ sendBatchedAcksToServer (do a <- pendingAcks; where a.originServer == server; yield a) server)
576+
577+pruneAndRetry = \t -> do
578+ result <- atomic do
579+ set *lastPrunedAt = t
580+ pr <- pruneExpired t
581+ forwards <- *forwards
582+ let pendingForwards = do
583+ f <- forwards
584+ yield {toServer: f.toServer, toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, messageId: f.messageId}
585+ ackedIds <- *ackedIds
586+ let pendingAcks = do
587+ a <- ackedIds
588+ where not (a.originServer == "")
589+ where not (a.originServer == serverHostN)
590+ yield a
591+ yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
592+ broadcastOfflineGossip t result.peerServers result.expiredLocal
593+ if not (federationSecret == "") && count result.pendingForwards > 0
594+ then retryPendingForwards result.pendingForwards
595+ else yield {}
596+ if not (federationSecret == "") && count result.pendingAcks > 0
597+ then do
598+ let ackOriginServers = collectPeerServers result.pendingAcks (\a -> a.originServer)
599+ sendAcksAndClean result.pendingAcks ackOriginServers
600+ else yield {}
580601
581602pollOnce : Text -> Int -> IO {clock} {authorized: Bool, cursor: Int, events: [PollEvent]}
582603pollOnce = \token cursor -> do
@@ -655,24 +676,16 @@ serve = \req -> case req of
655676 pr <- maybePrune t
656677 challenges <- *challenges
657678 let existingCount = count (do c <- challenges; where c.pubkey == pubkey; where c.expiresAt > t; yield c)
658− let oldestChallengeExpiry = fold (\acc c2 -> if c2.pubkey == pubkey && c2.expiresAt > t && (acc == -1 || c2.expiresAt < acc) then c2.expiresAt else acc) (-1) challenges
659− let cappedChallenges = if existingCount >= maxChallengesPerKey && oldestChallengeExpiry >= 0
660− then fold (\acc c ->
661− if not acc.removed && c.pubkey == pubkey && c.expiresAt == oldestChallengeExpiry
662− then {removed: True {}, items: acc.items}
663− else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} challenges
664− else {removed: False {}, items: challenges}
665− let withoutExpired = if count cappedChallenges.items >= maxChallenges
666− then do c <- cappedChallenges.items; where c.expiresAt > t; yield c
667− else cappedChallenges.items
668− let globalOldestExpiry = if count withoutExpired >= maxChallenges
669− then fold (\acc c -> if acc == -1 || c.expiresAt < acc then c.expiresAt else acc) (-1) withoutExpired
670− else -1
671− let globalCapped = if globalOldestExpiry >= 0
672− then (fold (\acc c ->
673− if not acc.removed && c.expiresAt == globalOldestExpiry
674− then {removed: True {}, items: acc.items}
675− else {removed: acc.removed, items: union acc.items [c]}) {removed: False {}, items: []} withoutExpired).items
679+ let forKey = do c <- challenges; where c.pubkey == pubkey; yield c
680+ let others = do c <- challenges; where not (c.pubkey == pubkey); yield c
681+ let challengesCapped = if existingCount >= maxChallengesPerKey
682+ then union others (evictOneOldest forKey (\c -> c.expiresAt))
683+ else challenges
684+ let withoutExpired = if count challengesCapped >= maxChallenges
685+ then do c <- challengesCapped; where c.expiresAt > t; yield c
686+ else challengesCapped
687+ let globalCapped = if count withoutExpired >= maxChallenges
688+ then evictOneOldest withoutExpired (\c -> c.expiresAt)
676689 else withoutExpired
677690 set *challenges = union
678691 globalCapped
@@ -708,28 +721,15 @@ serve = \req -> case req of
708721 yield c
709722 Nothing {} -> challenges
710723 else challenges
711− let existingForKey = do
712− s <- sessions
713− where s.pubkey == pubkey
714− where s.expiresAt > t
715− yield s
716− let oldestExpiry = fold (\acc s -> if acc == -1 || s.expiresAt < acc then s.expiresAt else acc) (-1) existingForKey
717− let cappedFold = fold (\acc s ->
718− if not acc.removed && s.pubkey == pubkey && s.expiresAt == oldestExpiry
719− then {removed: True {}, sessions: acc.sessions}
720− else {removed: acc.removed, sessions: union acc.sessions [s]}) {removed: False {}, sessions: []} sessions
721− let sessionsCapped = if count existingForKey >= maxSessionsPerKey && oldestExpiry >= 0
722− then cappedFold.sessions
724+ let existingForKey = do s <- sessions; where s.pubkey == pubkey; where s.expiresAt > t; yield s
725+ let forKey = do s <- sessions; where s.pubkey == pubkey; yield s
726+ let others = do s <- sessions; where not (s.pubkey == pubkey); yield s
727+ let sessionsCapped = if count existingForKey >= maxSessionsPerKey
728+ then union others (evictOneOldest forKey (\s -> s.expiresAt))
723729 else sessions
724730 let activeSessionsCapped = do s <- sessionsCapped; where s.expiresAt > t; yield s
725− let globalOldestSessionExpiry = if count activeSessionsCapped >= maxSessions
726− then fold (\acc s -> if acc == -1 || s.expiresAt < acc then s.expiresAt else acc) (-1) activeSessionsCapped
727− else -1
728− let globalSessionsCapped = if globalOldestSessionExpiry >= 0
729− then (fold (\acc s ->
730− if not acc.removed && s.expiresAt == globalOldestSessionExpiry
731− then {removed: True {}, items: acc.items}
732− else {removed: acc.removed, items: union acc.items [s]}) {removed: False {}, items: []} activeSessionsCapped).items
731+ let globalSessionsCapped = if count activeSessionsCapped >= maxSessions
732+ then evictOneOldest activeSessionsCapped (\s -> s.expiresAt)
733733 else activeSessionsCapped
734734 set *sessions = if sigValid
735735 then union globalSessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]
@@ -738,22 +738,15 @@ serve = \req -> case req of
738738 p <- presence
739739 where not (p.pubkey == pubkey && p.server == serverHostN)
740740 yield p
741− let oldestPresenceExp = fold (\acc p -> if acc == -1 || p.expiresAt < acc then p.expiresAt else acc) (-1) presenceWithout
742− let presenceEvicted = if count presenceWithout >= maxPresenceEntries && oldestPresenceExp >= 0
743− then (fold (\acc p ->
744− if not acc.removed && p.expiresAt == oldestPresenceExp
745− then {removed: True {}, items: acc.items}
746− else {removed: acc.removed, items: union acc.items [p]}) {removed: False {}, items: []} presenceWithout).items
741+ let presenceEvicted = if count presenceWithout >= maxPresenceEntries
742+ then evictOneOldest presenceWithout (\p -> p.expiresAt)
747743 else presenceWithout
748744 let updatedPresence = if sigValid
749745 then union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
750746 else presence
751747 set *presence = updatedPresence
752748 let peerServers = if sigValid && not (federationSecret == "")
753− then fold (\acc p ->
754− if p.server == serverHostN then acc
755− else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
756− else union acc [p.server]) [] updatedPresence
749+ then collectPeerServers updatedPresence (\p -> p.server)
757750 else []
758751 yield {sigValid: sigValid, peerServers: peerServers, expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}
759752 fork (do
@@ -921,9 +914,7 @@ serve = \req -> case req of
921914 where not (m.originServer == "")
922915 where not (m.originServer == serverHostN)
923916 yield m
924− let originServers = fold (\acc m ->
925− if count (do a <- acc; where a == m.originServer; yield a) > 0 then acc
926− else union acc [m.originServer]) [] fedMessages
917+ let originServers = collectPeerServers fedMessages (\m -> m.originServer)
927918 fork (do
928919 if result.ok && count originServers > 0 && not (federationSecret == "")
929920 then do
@@ -940,9 +931,7 @@ serve = \req -> case req of
940931 yield a
941932 if count evictedToRetry > 0 && not (federationSecret == "")
942933 then do
943− let evictedOriginServers = fold (\acc a ->
944− if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
945− else union acc [a.originServer]) [] evictedToRetry
934+ let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
946935 sendToEach evictedOriginServers (\server -> do
947936 result <- fetch (federationScheme ++ server) (FederationAck {
948937 authorization: "Bearer " ++ federationSecret,
@@ -1189,10 +1178,7 @@ serve = \req -> case req of
11891178 then if withinCapacity then afterOffline else afterReclaim
11901179 else presence
11911180 set *presence = finalPresence
1192− let gossipPeerServers = fold (\acc p ->
1193− if p.server == serverHostN then acc
1194− else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
1195− else union acc [p.server]) [] finalPresence
1181+ let gossipPeerServers = collectPeerServers finalPresence (\p -> p.server)
11961182 yield {appliedKeys: (if withinCapacity then union appliedOnline appliedOffline else union appliedOnlineUpdates appliedOffline), expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
11971183 let validEvents = do
11981184 e <- events
@@ -1295,16 +1281,14 @@ serve = \req -> case req of
12951281
12961282 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
12971283 let fromServer = normalizeServerName fromServer
1298− let originServer = if tooLong originServer maxServerNameLen then originServer else normalizeServerName originServer
1284+ let originServer = normalizeServerName originServer
12991285 t <- now
13001286 let authorized = verifyFederationAuth authorization
13011287 let dedupedIds = fold (\acc mid ->
1302− if mid == "" then acc
1303− else if count acc >= maxMessageIds then acc
1304− else if tooLong mid maxMessageIdLen then acc
1305− else if not (isValidHex mid) then acc
1306− else if count (do a <- acc; where a == normalizeHex mid; yield a) > 0 then acc
1307− else union acc [normalizeHex mid]) [] messageIds
1288+ let nmid = normalizeHex mid in
1289+ if count acc >= maxMessageIds then acc
1290+ else if count (do a <- acc; where a == nmid; yield a) > 0 then acc
1291+ else union acc [nmid]) [] messageIds
13081292 let validAckFields = authorized && originServer == serverHostN && not (fromServer == serverHostN)
13091293 pruneResult <- atomic do
13101294 pr <- maybePrune t
@@ -1328,32 +1312,7 @@ serve = \req -> case req of
13281312 let authorized = verifyMaintenanceAuth authorization
13291313 if authorized
13301314 then do
1331− result <- atomic do
1332− set *lastPrunedAt = t
1333− pr <- pruneExpired t
1334− forwards <- *forwards
1335− let pendingForwards = do
1336− f <- forwards
1337− yield {toServer: f.toServer, toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, messageId: f.messageId}
1338− ackedIds <- *ackedIds
1339− let pendingAcks = do
1340− a <- ackedIds
1341− where not (a.originServer == "")
1342− where not (a.originServer == serverHostN)
1343− yield a
1344− yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
1345− fork (do
1346− broadcastOfflineGossip t result.peerServers result.expiredLocal
1347− if not (federationSecret == "") && count result.pendingForwards > 0
1348− then retryPendingForwards result.pendingForwards
1349− else yield {}
1350− if not (federationSecret == "") && count result.pendingAcks > 0
1351− then do
1352− let ackOriginServers = fold (\acc a ->
1353− if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1354− else union acc [a.originServer]) [] result.pendingAcks
1355− sendAcksAndClean result.pendingAcks ackOriginServers
1356− else yield {})
1315+ fork (pruneAndRetry t)
13571316 yield (respond {ok: True {}})
13581317 else yield (respond {ok: False {}})
13591318
@@ -1364,31 +1323,7 @@ serve = \req -> case req of
13641323backgroundPrune = do
13651324 sleep 30000
13661325 t <- now
1367− result <- atomic do
1368− set *lastPrunedAt = t
1369− pr <- pruneExpired t
1370− forwards <- *forwards
1371− let pendingForwards = do
1372− f <- forwards
1373− yield {toServer: f.toServer, toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, messageId: f.messageId}
1374− ackedIds <- *ackedIds
1375− let pendingAcks = do
1376− a <- ackedIds
1377− where not (a.originServer == "")
1378− where not (a.originServer == serverHostN)
1379− yield a
1380− yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
1381− broadcastOfflineGossip t result.peerServers result.expiredLocal
1382− if not (federationSecret == "") && count result.pendingForwards > 0
1383− then retryPendingForwards result.pendingForwards
1384− else yield {}
1385− if not (federationSecret == "") && count result.pendingAcks > 0
1386− then do
1387− let ackOriginServers = fold (\acc a ->
1388− if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1389− else union acc [a.originServer]) [] result.pendingAcks
1390− sendAcksAndClean result.pendingAcks ackOriginServers
1391− else yield {}
1326+ pruneAndRetry t
13921327 fork backgroundPrune
13931328
13941329main = do