@@ -180,7 +180,6 @@ type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: T
180type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFailedAt: Timestamp}
180type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFailedAt: Timestamp}
181type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
181type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
182type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
182type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
183−data AuthResult = Authorized {pubkey: PubkeyHex, ip: Text} | Unauthorized {}
185type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}
184type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}
186type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
185type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
@@ -498,9 +497,8 @@ pct = \cur maxN -> if maxN > 0 then cur * 100 / maxN else 0
498-- Read entries matching `pred` (transformed by `extract`) and prune them in one step.
497-- Read entries matching `pred` (transformed by `extract`) and prune them in one step.
499collectAndPrune = \ref pred extract -> do
498collectAndPrune = \ref pred extract -> do
500 items <- *ref
499 items <- *ref
501− let mine = mapWhere items pred extract
502 pruneRef ref (\x -> not (pred x))
500 pruneRef ref (\x -> not (pred x))
503− yield mine
501+ yield (mapWhere items pred extract)
505-- Common ack/delete tail: pull matching messages, drop them and their forwards.
503-- Common ack/delete tail: pull matching messages, drop them and their forwards.
506removeForKey = \pred -> do
504removeForKey = \pred -> do
@@ -620,20 +618,13 @@ gateRated = \checkDrain auth ip rateRef maxRate windowMs t respond mkErr rateRes
620 lim <- checkPubkeyRateGate rateRef pk maxRate windowMs t
618 lim <- checkPubkeyRateGate rateRef pk maxRate windowMs t
621 if lim then yield (respond rateResp) else onAuth pk)
619 if lim then yield (respond rateResp) else onAuth pk)
623−authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
624−authenticateSession = \sessions token t ->
625− case findFirst sessions (\s -> s.expiresAt > t && s.token == token) of
626− Nothing {} -> Unauthorized {}
627− Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}
629−withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
621+withSessionAuth = \authorization t clientIp onUnauth onAuth ->
630 case extractToken authorization of
622 case extractToken authorization of
631 Nothing {} -> onUnauth
623 Nothing {} -> onUnauth
632 Just {value: token} -> do
624 Just {value: token} -> do
633− sessions <- *sessions
625+ ses <- *sessions
634− case authenticateSession sessions token t of
626+ case findFirst ses (\s -> s.expiresAt > t && s.token == token && matchIp clientIp s.ip) of
635− Authorized {pubkey, ip} ->
627+ Just {value: s} -> onAuth s.pubkey
636− if matchIp clientIp ip then onAuth pubkey else onUnauth
637 _ -> onUnauth
628 _ -> onUnauth
639generateFederationAuth : Text -> Timestamp -> IO {network} Text
630generateFederationAuth : Text -> Timestamp -> IO {network} Text
@@ -681,16 +672,14 @@ verifyFederationAuth = \authorization fromServer scope requestTimestamp -> do
681withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do
672withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do
682 blocked <- *blockedPeers
673 blocked <- *blockedPeers
683 allowed <- *allowedPeers
674 allowed <- *allowedPeers
684− let preErr =
675+ let err = \e -> respond {ok: False {}, error: e, retryAfterMs: 0}
685− if elem blocked fromServer then "blocked"
676+ if elem blocked fromServer then yield (err "blocked")
686− else if count allowed > 0 && not (elem allowed fromServer) then "not_allowed"
677+ else if count allowed > 0 && not (elem allowed fromServer) then yield (err "not_allowed")
687− else if not precondition then "invalid_request"
678+ else if not precondition then yield (err "invalid_request")
688− else if requestTimestamp > 0 && abs (requestTimestamp - t) > clockSkewTolerance then "clock_skew"
679+ else if requestTimestamp > 0 && abs (requestTimestamp - t) > clockSkewTolerance then yield (err "clock_skew")
690− if preErr != "" then yield (respond {ok: False {}, error: preErr, retryAfterMs: 0})
692 verified <- verifyFederationAuth authorization fromServer scope requestTimestamp
681 verified <- verifyFederationAuth authorization fromServer scope requestTimestamp
693− if not verified then yield (respond {ok: False {}, error: "unauthorized", retryAfterMs: 0})
682+ if not verified then yield (err "unauthorized")
695 limited <- checkServerRateGate fromServer t
684 limited <- checkServerRateGate fromServer t
696 if limited then yield (respond {ok: False {}, error: "rate_limited", retryAfterMs: federationRateWindowMs})
685 if limited then yield (respond {ok: False {}, error: "rate_limited", retryAfterMs: federationRateWindowMs})
@@ -989,15 +978,13 @@ refreshLocalPresence = \t -> do
990handleChallenge = \pubkey clientIp respond -> do
979handleChallenge = \pubkey clientIp respond -> do
992− let mkResp = \err -> respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: err}
993 preErr <- if requireClientIp && clientIp == "" then yield "client_ip_required" else authPreCheck clientIp t
981 preErr <- if requireClientIp && clientIp == "" then yield "client_ip_required" else authPreCheck clientIp t
994− if preErr != "" then yield (mkResp preErr)
982+ if preErr != "" then yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: preErr})
996 challenge <- genToken
984 challenge <- genToken
997 result <- atomic do
985 result <- atomic do
998 ch <- *challenges
986 ch <- *challenges
999− let active = filter ch (\c -> c.expiresAt > t)
1000− let cappedActive = capPerKey active (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt)
987+ let cappedActive = capPerKey (filter ch (\c -> c.expiresAt > t)) (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt)
1001 rates <- *challengeRates
988 rates <- *challengeRates
1002 let r = checkRate rates pubkey maxChallengeRatePerKey challengeRateWindowMs 5000 t
989 let r = checkRate rates pubkey maxChallengeRatePerKey challengeRateWindowMs 5000 t
1003 if count cappedActive >= maxChallenges then yield {ok: False {}, error: "capacity"}
990 if count cappedActive >= maxChallenges then yield {ok: False {}, error: "capacity"}
@@ -1018,16 +1005,14 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
1018 result <- atomic do
1005 result <- atomic do
1019 ch <- *challenges
1006 ch <- *challenges
1020 dp <- *drainPhase
1007 dp <- *drainPhase
1021− let hasMatch = any ch (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && matchIp clientIp c.ip)
1009+ || none ch (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && matchIp clientIp c.ip)
1022− let sigOk = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
1010+ || not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))
1023− if dp >= 1 || not hasMatch || not sigOk then yield {valid: False {}, peers: []}
1011+ then yield {valid: False {}, peers: []}
1025 set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)
1013 set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)
1026 ses <- *sessions
1014 ses <- *sessions
1027− let cleaned = if revokeOthers
1028− then removeWhere ses (\s -> s.pubkey == pubkey)
1029− else capPerKey ses (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt)
1030− set *sessions = capList (union cleaned [{token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}]) maxSessions (\s -> s.expiresAt)
1015+ set *sessions = capList (union (if revokeOthers then removeWhere ses (\s -> s.pubkey == pubkey) else capPerKey ses (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt)) [{token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}]) maxSessions (\s -> s.expiresAt)
1031 pres <- *presence
1016 pres <- *presence
1032 let newPres = capList (union (removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)) [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1017 let newPres = capList (union (removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)) [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1033 set *presence = newPres
1018 set *presence = newPres
@@ -1090,15 +1075,13 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
1090pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -> do
1075pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -> do
1091 setup <- atomic do
1076 setup <- atomic do
1092 ses <- *sessions
1077 ses <- *sessions
1093− case authenticateSession ses token t of
1078+ case findFirst ses (\s -> s.expiresAt > t && s.token == token && matchIp clientIp s.ip) of
1094− Unauthorized {} -> yield Nothing {}
1095− Authorized {pubkey, ip} ->
1096− if not (matchIp clientIp ip) then yield Nothing {}
1079+ Nothing {} -> yield Nothing {}
1097− else do
1080+ Just {value: s} -> do
1098− let clampedCursor = clamp 0 t cursor
1081+ let clampedCursor = clamp 0 t cursor
1099− ackResult <- if clampedCursor > 0 then ackPolledMessages pubkey clampedCursor t else yield {pendingAcks: []}
1082+ ackResult <- if clampedCursor > 0 then ackPolledMessages s.pubkey clampedCursor t else yield {pendingAcks: []}
1100− startWakeup <- *pollWakeup
1083+ startWakeup <- *pollWakeup
1101− yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, pendingAcks: ackResult.pendingAcks}})
1084+ yield (Just {value: {pubkey: s.pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, pendingAcks: ackResult.pendingAcks}})
1102 case setup of
1085 case setup of
1103 Nothing {} -> yield (respond (emptyPollResp 0 (False {}) (False {}) (False {}) serverGen 0 cursorWasReset))
1086 Nothing {} -> yield (respond (emptyPollResp 0 (False {}) (False {}) (False {}) serverGen 0 cursorWasReset))
1104 Just {value: s} -> do
1087 Just {value: s} -> do
@@ -1146,8 +1129,9 @@ doPoll = \s t -> do
1146collectPollEvents = \messages pubkey cursor t ->
1129collectPollEvents = \messages pubkey cursor t ->
1147 let myMsgs = filter messages (\m -> m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor) in
1130 let myMsgs = filter messages (\m -> m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor) in
1148 let limited = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs) in
1131 let limited = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs) in
1149− let evts = pluck limited (\m -> {id: m.messageId, eventType: "message", payload: {toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, expiresAt: m.expiresAt}}) in
1132+ {events: pluck limited (\m -> {id: m.messageId, eventType: "message", payload: {toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, expiresAt: m.expiresAt}}),
1150− {events: evts, cursor: fold (\acc m -> max acc m.receivedAt) cursor limited, hasMore: count myMsgs > count limited}
1133+ cursor: fold (\acc m -> max acc m.receivedAt) cursor limited,
1134+ hasMore: count myMsgs > count limited}
1152ackPolledMessages = \pubkey cursor t -> do
1136ackPolledMessages = \pubkey cursor t -> do
1153 msgs <- *messages
1137 msgs <- *messages
@@ -1164,7 +1148,6 @@ ackPolledMessages = \pubkey cursor t -> do
1164-- ============================================
1148-- ============================================
1166handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority authorization clientIp respond -> do
1150handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority authorization clientIp respond -> do
1167− let clampedPriority = clamp 0 maxMessagePriority priority
1168 t <- now
1151 t <- now
1169 dp <- *drainPhase
1152 dp <- *drainPhase
1170 g <- checkGlobalRate t
1153 g <- checkGlobalRate t
@@ -1177,7 +1160,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
1177 result <- atomic do
1160 result <- atomic do
1178 withSessionAuth authorization t clientIp
1161 withSessionAuth authorization t clientIp
1179 (yield {status: MsUnauthorized {}, remoteServers: []})
1162 (yield {status: MsUnauthorized {}, remoteServers: []})
1180− (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effMsgId expiresInMs clampedPriority t)
1163+ (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effMsgId expiresInMs (clamp 0 maxMessagePriority priority) t)
1181 when (federationEnabled && count result.remoteServers > 0)
1164 when (federationEnabled && count result.remoteServers > 0)
1182 (fork (forkForwardsTo effMsgId to encryptedBlob clampedTs result.remoteServers))
1165 (fork (forkForwardsTo effMsgId to encryptedBlob clampedTs result.remoteServers))
1183 let ok = case result.status of
1166 let ok = case result.status of
@@ -1339,8 +1322,7 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
1340 fwdServers <- applyCancel msgs matches to messageId t
1323 fwdServers <- applyCancel msgs matches to messageId t
1341 pres <- *presence
1324 pres <- *presence
1342− let presServers = collectPeers (presenceForKey pres to t) (\p -> p.server)
1325+ yield {cancelled: True {}, cancelServers: unique (union (collectPeers (presenceForKey pres to t) (\p -> p.server)) fwdServers)}
1343− yield {cancelled: True {}, cancelServers: unique (union presServers fwdServers)}
1344 when (result.cancelled && count result.cancelServers > 0 && federationEnabled)
1326 when (result.cancelled && count result.cancelServers > 0 && federationEnabled)
1345 (fork (sendCancelToEach messageId to result.cancelServers))
1327 (fork (sendCancelToEach messageId to result.cancelServers))
1346 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
1328 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
@@ -1501,8 +1483,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
1501 && none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)
1483 && none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)
1502 && none remainingFwds (\f -> f.messageId == m.messageId && f.toKey == m.toKey))
1484 && none remainingFwds (\f -> f.messageId == m.messageId && f.toKey == m.toKey))
1503 set *messages = removeWhere msgs (\m -> any toReap (\r -> r.messageId == m.messageId && r.toKey == m.toKey))
1485 set *messages = removeWhere msgs (\m -> any toReap (\r -> r.messageId == m.messageId && r.toKey == m.toKey))
1504− let perResult = pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})
1486+ yield {ackedCount: count removed, results: pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})}
1505− yield {ackedCount: count removed, results: perResult}
1506 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
1487 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
1508handleFederationCancel = \messageId toKey fromServer requestTimestamp correlationId authorization respond -> do
1489handleFederationCancel = \messageId toKey fromServer requestTimestamp correlationId authorization respond -> do
@@ -1561,12 +1542,11 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
1561 yield {messageId: mp.v, toKey: kp.v}
1542 yield {messageId: mp.v, toKey: kp.v}
1562 atomic do
1543 atomic do
1563 msgs <- *messages
1544 msgs <- *messages
1564− let receipts = do
1545+ appendCapped senderDeliveries (do
1565 p <- pairs
1546 p <- pairs
1566 m <- msgs
1547 m <- msgs
1567 where m.messageId == p.messageId && m.toKey == p.toKey && m.senderKey != "" && m.originServer == ""
1548 where m.messageId == p.messageId && m.toKey == p.toKey && m.senderKey != "" && m.originServer == ""
1568− yield {senderKey: m.senderKey, messageId: p.messageId, toKey: p.toKey, deliveredAt: t}
1549+ yield {senderKey: m.senderKey, messageId: p.messageId, toKey: p.toKey, deliveredAt: t}) 10000 (\d -> d.deliveredAt)
1569− appendCapped senderDeliveries receipts 10000 (\d -> d.deliveredAt)
1570 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
1550 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
1572-- Index a list (1-based by virtue of `count` of the partial fold).
1552-- Index a list (1-based by virtue of `count` of the partial fold).
@@ -1626,8 +1606,7 @@ handleHealth = \authorization respond -> do
1626 let warmup = t - sa < backgroundPruneInterval * 3
1606 let warmup = t - sa < backgroundPruneInterval * 3
1627 let pruneOk = warmup || t - lastPrune < backgroundPruneInterval * 3
1607 let pruneOk = warmup || t - lastPrune < backgroundPruneInterval * 3
1628 let heartbeatOk = warmup || t - lastHb < pollTimeoutMs * 3
1608 let heartbeatOk = warmup || t - lastHb < pollTimeoutMs * 3
1629− let allOk = dp == 0 && pruneOk && heartbeatOk
1630− yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: pruneOk, heartbeatHealthy: heartbeatOk, capacityHealthy: True {}, capacityWarning: False {}, federationHealthy: True {}, watchdogHealthy: True {}, nonceHealthy: True {}, forwardQueueHealthy: True {}, requestPressureHealthy: True {}, draining: dp > 0, degradedPeers: 0})
1609+ yield (respond {ok: dp == 0 && pruneOk && heartbeatOk, version: serverBuildVersion, pruneHealthy: pruneOk, heartbeatHealthy: heartbeatOk, capacityHealthy: True {}, capacityWarning: False {}, federationHealthy: True {}, watchdogHealthy: True {}, nonceHealthy: True {}, forwardQueueHealthy: True {}, requestPressureHealthy: True {}, draining: dp > 0, degradedPeers: 0})
1632handleReady = \respond -> do
1611handleReady = \respond -> do
1633 sa <- *startedAt
1612 sa <- *startedAt
@@ -1699,8 +1678,7 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
1699 let mine = filter msgs (\m -> m.toKey == pubkey)
1678 let mine = filter msgs (\m -> m.toKey == pubkey)
1700 pruneRef messages (\m -> m.toKey != pubkey)
1679 pruneRef messages (\m -> m.toKey != pubkey)
1701 pruneRef forwards (\f -> f.toKey != pubkey)
1680 pruneRef forwards (\f -> f.toKey != pubkey)
1702− let federated = mapWhere mine (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
1681+ appendCapped ackedIds (mapWhere mine (\m -> m.originServer != "") (\m -> mkAckedEntry m t)) maxAckedIds (\a -> a.ackedAt)
1703− appendCapped ackedIds federated maxAckedIds (\a -> a.ackedAt)
1704 yield (count mine)
1682 yield (count mine)
1705 yield (respond {ok: True {}, purgedCount: purged}))
1683 yield (respond {ok: True {}, purgedCount: purged}))
@@ -1708,12 +1686,11 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
1708 (yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0}))
1686 (yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0}))
1709 (if ip == "" then yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0})
1687 (if ip == "" then yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0})
1711− result <- atomic do
1689+ keys <- atomic do
1712 ses <- *sessions
1690 ses <- *sessions
1713− let myKeys = mapWhere ses (\s -> s.ip == ip) (\s -> s.pubkey)
1714 pruneRef sessions (\s -> s.ip != ip)
1691 pruneRef sessions (\s -> s.ip != ip)
1715− yield {revoked: count myKeys, keys: count (unique myKeys)}
1692+ yield (mapWhere ses (\s -> s.ip == ip) (\s -> s.pubkey))
1716− yield (respond {ok: True {}, revokedCount: result.revoked, affectedKeys: result.keys}))
1693+ yield (respond {ok: True {}, revokedCount: count keys, affectedKeys: count (unique keys)}))
1718handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization
1695handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization
1719 (yield (respond {sessions: [], ok: False {}, total: 0}))
1696 (yield (respond {sessions: [], ok: False {}, total: 0}))
@@ -1727,8 +1704,7 @@ handleAdminListAllSessions = \limit authorization respond -> withMaintAuth autho
1727 count: count mine,
1704 count: count mine,
1728 ips: unique (mapWhere mine (\s -> s.ip != "") (\s -> s.ip)),
1705 ips: unique (mapWhere mine (\s -> s.ip != "") (\s -> s.ip)),
1729 latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine})
1706 latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine})
1730− let sorted = sortBy (\g -> 0 - g.count) grouped
1731− yield (respond {sessions: takeRelation (if limit <= 0 then 100 else clamp 1 500 limit) sorted, ok: True {}, total: count grouped}))
1707+ yield (respond {sessions: takeRelation (if limit <= 0 then 100 else clamp 1 500 limit) (sortBy (\g -> 0 - g.count) grouped), ok: True {}, total: count grouped}))
1733handleGetRateLimits = \authorization respond -> do
1709handleGetRateLimits = \authorization respond -> do
1734 ok <- verifyMaintenanceAuth authorization
1710 ok <- verifyMaintenanceAuth authorization
@@ -1829,11 +1805,12 @@ handleStatsForwards = \authorization respond -> withMaintAuth authorization
1830 t <- now
1806 t <- now
1831 fwds <- *forwards
1807 fwds <- *forwards
1832− let servers = collectPeers fwds (\f -> f.toServer)
1809+ servers: pluck (collectPeers fwds (\f -> f.toServer)) (\s ->
1833− let result = pluck servers (\s ->
1834− let svFwds = filter fwds (\f -> f.toServer == s) in
1810+ let svFwds = filter fwds (\f -> f.toServer == s) in
1835− {server: s, count: count svFwds, oldestAge: t - fold (\acc f -> min acc f.createdAt) t svFwds, totalRetries: fold (\acc f -> acc + f.retries) 0 svFwds})
1811+ {server: s, count: count svFwds, oldestAge: t - fold (\acc f -> min acc f.createdAt) t svFwds, totalRetries: fold (\acc f -> acc + f.retries) 0 svFwds}),
1836− yield (respond {servers: result, ok: True {}, total: count fwds}))
1813+ total: count fwds}))
1838handleStatsRates = \authorization respond -> do
1815handleStatsRates = \authorization respond -> do
1839 ok <- verifyMaintenanceAuth authorization
1816 ok <- verifyMaintenanceAuth authorization
@@ -1849,16 +1826,14 @@ handleMetrics = \authorization respond -> do
1849 aps <- *activePolls
1826 aps <- *activePolls
1850 sa <- *startedAt
1827 sa <- *startedAt
1851 dp <- *drainPhase
1828 dp <- *drainPhase
1852− let rows = [
1829+ yield (respond {metrics: fold (\acc r -> acc ++ r.k ++ " " ++ show r.v ++ "\n") "" [
1853 {k: "skrepka_messages_total", v: count msgs},
1830 {k: "skrepka_messages_total", v: count msgs},
1854 {k: "skrepka_sessions_total", v: count ses},
1831 {k: "skrepka_sessions_total", v: count ses},
1855 {k: "skrepka_presence_total", v: count pres},
1832 {k: "skrepka_presence_total", v: count pres},
1856 {k: "skrepka_forwards_total", v: count fwds},
1833 {k: "skrepka_forwards_total", v: count fwds},
1857 {k: "skrepka_active_polls", v: count aps},
1834 {k: "skrepka_active_polls", v: count aps},
1858 {k: "skrepka_uptime_seconds", v: (t - sa) / 1000},
1835 {k: "skrepka_uptime_seconds", v: (t - sa) / 1000},
1859− {k: "skrepka_draining", v: if dp > 0 then 1 else 0}]
1836+ {k: "skrepka_draining", v: if dp > 0 then 1 else 0}], ok: True {}})
1860− let metrics = fold (\acc r -> acc ++ r.k ++ " " ++ show r.v ++ "\n") "" rows
1861− yield (respond {metrics: metrics, ok: True {}})
1863-- ============================================
1838-- ============================================
1864-- Dispatch
1839-- Dispatch
@@ -1928,16 +1903,13 @@ serve = \req -> case req of
1928backgroundPrune = do
1903backgroundPrune = do
1929 sleep backgroundPruneInterval
1904 sleep backgroundPruneInterval
1930 dp <- *drainPhase
1905 dp <- *drainPhase
1931− if dp >= 2 then do
1906+ when (dp < 2) (do
1932− sleep backgroundPruneInterval
1935 t <- now
1907 t <- now
1936 atomic do
1908 atomic do
1937 pc <- *pruneCount
1909 pc <- *pruneCount
1938 set *pruneCount = pc + 1
1910 set *pruneCount = pc + 1
1939− pruneAndRetry t
1911+ pruneAndRetry t)
1940− backgroundPrune
1912+ backgroundPrune
1942pollHeartbeat = do
1914pollHeartbeat = do
1943 sleep pollTimeoutMs
1915 sleep pollTimeoutMs