@@ -180,7 +180,6 @@ type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: T
180180type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFailedAt: Timestamp}
181181type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
182182type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
183−data AuthResult = Authorized {pubkey: PubkeyHex, ip: Text} | Unauthorized {}
184183
185184type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}
186185type 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
498497-- Read entries matching `pred` (transformed by `extract`) and prune them in one step.
499498collectAndPrune = \ref pred extract -> do
500499 items <- *ref
501− let mine = mapWhere items pred extract
502500 pruneRef ref (\x -> not (pred x))
503− yield mine
501+ yield (mapWhere items pred extract)
504502
505503-- Common ack/delete tail: pull matching messages, drop them and their forwards.
506504removeForKey = \pred -> do
@@ -620,20 +618,13 @@ gateRated = \checkDrain auth ip rateRef maxRate windowMs t respond mkErr rateRes
620618 lim <- checkPubkeyRateGate rateRef pk maxRate windowMs t
621619 if lim then yield (respond rateResp) else onAuth pk)
622620
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}
628−
629−withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
621+withSessionAuth = \authorization t clientIp onUnauth onAuth ->
630622 case extractToken authorization of
631623 Nothing {} -> onUnauth
632624 Just {value: token} -> do
633− sessions <- *sessions
634− case authenticateSession sessions token t of
635− Authorized {pubkey, ip} ->
636− if matchIp clientIp ip then onAuth pubkey else onUnauth
625+ ses <- *sessions
626+ case findFirst ses (\s -> s.expiresAt > t && s.token == token && matchIp clientIp s.ip) of
627+ Just {value: s} -> onAuth s.pubkey
637628 _ -> onUnauth
638629
639630generateFederationAuth : Text -> Timestamp -> IO {network} Text
@@ -681,16 +672,14 @@ verifyFederationAuth = \authorization fromServer scope requestTimestamp -> do
681672withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do
682673 blocked <- *blockedPeers
683674 allowed <- *allowedPeers
684− let preErr =
685− if elem blocked fromServer then "blocked"
686− else if count allowed > 0 && not (elem allowed fromServer) then "not_allowed"
687− else if not precondition then "invalid_request"
688− else if requestTimestamp > 0 && abs (requestTimestamp - t) > clockSkewTolerance then "clock_skew"
689− else ""
690− if preErr != "" then yield (respond {ok: False {}, error: preErr, retryAfterMs: 0})
675+ let err = \e -> respond {ok: False {}, error: e, retryAfterMs: 0}
676+ if elem blocked fromServer then yield (err "blocked")
677+ else if count allowed > 0 && not (elem allowed fromServer) then yield (err "not_allowed")
678+ else if not precondition then yield (err "invalid_request")
679+ else if requestTimestamp > 0 && abs (requestTimestamp - t) > clockSkewTolerance then yield (err "clock_skew")
691680 else do
692681 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")
694683 else do
695684 limited <- checkServerRateGate fromServer t
696685 if limited then yield (respond {ok: False {}, error: "rate_limited", retryAfterMs: federationRateWindowMs})
@@ -989,15 +978,13 @@ refreshLocalPresence = \t -> do
989978
990979handleChallenge = \pubkey clientIp respond -> do
991980 t <- now
992− let mkResp = \err -> respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: err}
993981 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})
995983 else do
996984 challenge <- genToken
997985 result <- atomic do
998986 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)
1001988 rates <- *challengeRates
1002989 let r = checkRate rates pubkey maxChallengeRatePerKey challengeRateWindowMs 5000 t
1003990 if count cappedActive >= maxChallenges then yield {ok: False {}, error: "capacity"}
@@ -1018,16 +1005,14 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
10181005 result <- atomic do
10191006 ch <- *challenges
10201007 dp <- *drainPhase
1021− let hasMatch = any 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)
1023− if dp >= 1 || not hasMatch || not sigOk then yield {valid: False {}, peers: []}
1008+ if dp >= 1
1009+ || none ch (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && matchIp clientIp c.ip)
1010+ || not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))
1011+ then yield {valid: False {}, peers: []}
10241012 else do
10251013 set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)
10261014 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)
10311016 pres <- *presence
10321017 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)
10331018 set *presence = newPres
@@ -1090,15 +1075,13 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
10901075pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -> do
10911076 setup <- atomic do
10921077 ses <- *sessions
1093− case authenticateSession ses token t of
1094− Unauthorized {} -> yield Nothing {}
1095− Authorized {pubkey, ip} ->
1096− if not (matchIp clientIp ip) then yield Nothing {}
1097− else do
1098− let clampedCursor = clamp 0 t cursor
1099− ackResult <- if clampedCursor > 0 then ackPolledMessages pubkey clampedCursor t else yield {pendingAcks: []}
1100− startWakeup <- *pollWakeup
1101− yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, pendingAcks: ackResult.pendingAcks}})
1078+ case findFirst ses (\s -> s.expiresAt > t && s.token == token && matchIp clientIp s.ip) of
1079+ Nothing {} -> yield Nothing {}
1080+ Just {value: s} -> do
1081+ let clampedCursor = clamp 0 t cursor
1082+ ackResult <- if clampedCursor > 0 then ackPolledMessages s.pubkey clampedCursor t else yield {pendingAcks: []}
1083+ startWakeup <- *pollWakeup
1084+ yield (Just {value: {pubkey: s.pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, pendingAcks: ackResult.pendingAcks}})
11021085 case setup of
11031086 Nothing {} -> yield (respond (emptyPollResp 0 (False {}) (False {}) (False {}) serverGen 0 cursorWasReset))
11041087 Just {value: s} -> do
@@ -1146,8 +1129,9 @@ doPoll = \s t -> do
11461129collectPollEvents = \messages pubkey cursor t ->
11471130 let myMsgs = filter messages (\m -> m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor) in
11481131 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
1150− {events: evts, cursor: fold (\acc m -> max acc m.receivedAt) cursor limited, hasMore: count myMsgs > count limited}
1132+ {events: pluck limited (\m -> {id: m.messageId, eventType: "message", payload: {toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, expiresAt: m.expiresAt}}),
1133+ cursor: fold (\acc m -> max acc m.receivedAt) cursor limited,
1134+ hasMore: count myMsgs > count limited}
11511135
11521136ackPolledMessages = \pubkey cursor t -> do
11531137 msgs <- *messages
@@ -1164,7 +1148,6 @@ ackPolledMessages = \pubkey cursor t -> do
11641148-- ============================================
11651149
11661150handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority authorization clientIp respond -> do
1167− let clampedPriority = clamp 0 maxMessagePriority priority
11681151 t <- now
11691152 dp <- *drainPhase
11701153 g <- checkGlobalRate t
@@ -1177,7 +1160,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
11771160 result <- atomic do
11781161 withSessionAuth authorization t clientIp
11791162 (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)
11811164 when (federationEnabled && count result.remoteServers > 0)
11821165 (fork (forkForwardsTo effMsgId to encryptedBlob clampedTs result.remoteServers))
11831166 let ok = case result.status of
@@ -1339,8 +1322,7 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
13391322 else do
13401323 fwdServers <- applyCancel msgs matches to messageId t
13411324 pres <- *presence
1342− let presServers = collectPeers (presenceForKey pres to t) (\p -> p.server)
1343− yield {cancelled: True {}, cancelServers: unique (union presServers fwdServers)}
1325+ yield {cancelled: True {}, cancelServers: unique (union (collectPeers (presenceForKey pres to t) (\p -> p.server)) fwdServers)}
13441326 when (result.cancelled && count result.cancelServers > 0 && federationEnabled)
13451327 (fork (sendCancelToEach messageId to result.cancelServers))
13461328 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
@@ -1501,8 +1483,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
15011483 && none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)
15021484 && none remainingFwds (\f -> f.messageId == m.messageId && f.toKey == m.toKey))
15031485 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)})
1505− yield {ackedCount: count removed, results: perResult}
1486+ yield {ackedCount: count removed, results: pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})}
15061487 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
15071488
15081489handleFederationCancel = \messageId toKey fromServer requestTimestamp correlationId authorization respond -> do
@@ -1561,12 +1542,11 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
15611542 yield {messageId: mp.v, toKey: kp.v}
15621543 atomic do
15631544 msgs <- *messages
1564− let receipts = do
1545+ appendCapped senderDeliveries (do
15651546 p <- pairs
15661547 m <- msgs
15671548 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}
1569− appendCapped senderDeliveries receipts 10000 (\d -> d.deliveredAt)
1549+ yield {senderKey: m.senderKey, messageId: p.messageId, toKey: p.toKey, deliveredAt: t}) 10000 (\d -> d.deliveredAt)
15701550 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
15711551
15721552-- Index a list (1-based by virtue of `count` of the partial fold).
@@ -1626,8 +1606,7 @@ handleHealth = \authorization respond -> do
16261606 let warmup = t - sa < backgroundPruneInterval * 3
16271607 let pruneOk = warmup || t - lastPrune < backgroundPruneInterval * 3
16281608 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})
16311610
16321611handleReady = \respond -> do
16331612 sa <- *startedAt
@@ -1699,8 +1678,7 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
16991678 let mine = filter msgs (\m -> m.toKey == pubkey)
17001679 pruneRef messages (\m -> m.toKey != pubkey)
17011680 pruneRef forwards (\f -> f.toKey != pubkey)
1702− let federated = mapWhere mine (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
1703− appendCapped ackedIds federated maxAckedIds (\a -> a.ackedAt)
1681+ appendCapped ackedIds (mapWhere mine (\m -> m.originServer != "") (\m -> mkAckedEntry m t)) maxAckedIds (\a -> a.ackedAt)
17041682 yield (count mine)
17051683 yield (respond {ok: True {}, purgedCount: purged}))
17061684
@@ -1708,12 +1686,11 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
17081686 (yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0}))
17091687 (if ip == "" then yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0})
17101688 else do
1711− result <- atomic do
1689+ keys <- atomic do
17121690 ses <- *sessions
1713− let myKeys = mapWhere ses (\s -> s.ip == ip) (\s -> s.pubkey)
17141691 pruneRef sessions (\s -> s.ip != ip)
1715− yield {revoked: count myKeys, keys: count (unique myKeys)}
1716− yield (respond {ok: True {}, revokedCount: result.revoked, affectedKeys: result.keys}))
1692+ yield (mapWhere ses (\s -> s.ip == ip) (\s -> s.pubkey))
1693+ yield (respond {ok: True {}, revokedCount: count keys, affectedKeys: count (unique keys)}))
17171694
17181695handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization
17191696 (yield (respond {sessions: [], ok: False {}, total: 0}))
@@ -1727,8 +1704,7 @@ handleAdminListAllSessions = \limit authorization respond -> withMaintAuth autho
17271704 count: count mine,
17281705 ips: unique (mapWhere mine (\s -> s.ip != "") (\s -> s.ip)),
17291706 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}))
17321708
17331709handleGetRateLimits = \authorization respond -> do
17341710 ok <- verifyMaintenanceAuth authorization
@@ -1829,11 +1805,12 @@ handleStatsForwards = \authorization respond -> withMaintAuth authorization
18291805 (do
18301806 t <- now
18311807 fwds <- *forwards
1832− let servers = collectPeers fwds (\f -> f.toServer)
1833− let result = pluck servers (\s ->
1834− 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})
1836− yield (respond {servers: result, ok: True {}, total: count fwds}))
1808+ yield (respond {
1809+ servers: pluck (collectPeers fwds (\f -> f.toServer)) (\s ->
1810+ let svFwds = filter fwds (\f -> f.toServer == s) in
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}),
1812+ ok: True {},
1813+ total: count fwds}))
18371814
18381815handleStatsRates = \authorization respond -> do
18391816 ok <- verifyMaintenanceAuth authorization
@@ -1849,16 +1826,14 @@ handleMetrics = \authorization respond -> do
18491826 aps <- *activePolls
18501827 sa <- *startedAt
18511828 dp <- *drainPhase
1852− let rows = [
1829+ yield (respond {metrics: fold (\acc r -> acc ++ r.k ++ " " ++ show r.v ++ "\n") "" [
18531830 {k: "skrepka_messages_total", v: count msgs},
18541831 {k: "skrepka_sessions_total", v: count ses},
18551832 {k: "skrepka_presence_total", v: count pres},
18561833 {k: "skrepka_forwards_total", v: count fwds},
18571834 {k: "skrepka_active_polls", v: count aps},
18581835 {k: "skrepka_uptime_seconds", v: (t - sa) / 1000},
1859− {k: "skrepka_draining", v: if dp > 0 then 1 else 0}]
1860− let metrics = fold (\acc r -> acc ++ r.k ++ " " ++ show r.v ++ "\n") "" rows
1861− yield (respond {metrics: metrics, ok: True {}})
1836+ {k: "skrepka_draining", v: if dp > 0 then 1 else 0}], ok: True {}})
18621837
18631838-- ============================================
18641839-- Dispatch
@@ -1928,16 +1903,13 @@ serve = \req -> case req of
19281903backgroundPrune = do
19291904 sleep backgroundPruneInterval
19301905 dp <- *drainPhase
1931− if dp >= 2 then do
1932− sleep backgroundPruneInterval
1933− backgroundPrune
1934− else do
1906+ when (dp < 2) (do
19351907 t <- now
19361908 atomic do
19371909 pc <- *pruneCount
19381910 set *pruneCount = pc + 1
1939− pruneAndRetry t
1940− backgroundPrune
1911+ pruneAndRetry t)
1912+ backgroundPrune
19411913
19421914pollHeartbeat = do
19431915 sleep pollTimeoutMs