@@ -725,8 +725,8 @@ chargeRate = \rateRef key maxRate windowMs t -> do
725checkPubkeyRateGate = \rateRef pubkey maxRate windowMs t -> atomic
725checkPubkeyRateGate = \rateRef pubkey maxRate windowMs t -> atomic
726 (chargeRate rateRef pubkey maxRate windowMs t)
726 (chargeRate rateRef pubkey maxRate windowMs t)
728−checkServerRateGate = \server t ->
728+checkServerRateGate = \server ->
729− checkPubkeyRateGate federationRates server maxFederationRatePerServer federationRateWindowMs t
729+ checkPubkeyRateGate federationRates server maxFederationRatePerServer federationRateWindowMs
731-- Pre-auth gate: drain + global + ip + auth-global. Returns "" when ok, else an error code.
731-- Pre-auth gate: drain + global + ip + auth-global. Returns "" when ok, else an error code.
732authPreCheck = \clientIp t -> do
732authPreCheck = \clientIp t -> do
@@ -755,8 +755,9 @@ recordFailure = \server -> do
756 atomic do
756 atomic do
757 failures <- *forwardFailures
757 failures <- *forwardFailures
758+ let same = \f -> f.server == server
758− set *forwardFailures = union (removeWhere failures (\f -> f.server == server))
759+ set *forwardFailures = union (removeWhere failures same)
759− [case findFirst failures (\f -> f.server == server) of
760+ [case findFirst failures same of
760 Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
761 Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
761 Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}]
762 Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}]
@@ -1126,9 +1127,10 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
1126 dp <- *drainPhase
1127 dp <- *drainPhase
1127 g <- checkGlobalRate t
1128 g <- checkGlobalRate t
1128 effMsgId <- if messageId != "" then yield messageId else genToken
1129 effMsgId <- if messageId != "" then yield messageId else genToken
1130+ let err = \s e -> yield (respond (sendErrResp s e))
1129− if dp >= 2 then yield (respond (sendErrResp (MsDraining {}) "draining"))
1131+ if dp >= 2 then err (MsDraining {}) "draining"
1130− else if g then yield (respond (sendErrResp (MsRateLimited {}) "rate_limited"))
1132+ else if g then err (MsRateLimited {}) "rate_limited"
1131− else if t - timestamp > messageMaxAge then yield (respond (sendErrResp (MsRejected {}) "stale_timestamp"))
1133+ else if t - timestamp > messageMaxAge then err (MsRejected {}) "stale_timestamp"
1133 let clampedTs = clampTimestamp timestamp t
1135 let clampedTs = clampTimestamp timestamp t
1134 result <- atomic do
1136 result <- atomic do
@@ -1149,11 +1151,12 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
1149 srLim <- chargeRate sendRates senderKey maxSendRatePerKey sendRateWindowMs t
1151 srLim <- chargeRate sendRates senderKey maxSendRatePerKey sendRateWindowMs t
1150 prLim <- chargeRate pairRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t
1152 prLim <- chargeRate pairRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t
1151 rrLim <- chargeRate recvRates to maxRecvRatePerKey recvRateWindowMs t
1153 rrLim <- chargeRate recvRates to maxRecvRatePerKey recvRateWindowMs t
1154+ let reject = \s -> yield {status: s, remoteServers: []}
1152− if senderKey == to then yield {status: MsRejected {}, remoteServers: []}
1155+ if senderKey == to then reject (MsRejected {})
1153− else if hasMessage effMsgId to msgs || hasMessage effMsgId to aids || hasMessage effMsgId to dids then yield {status: MsDuplicate {}, remoteServers: []}
1156+ else if hasMessage effMsgId to msgs || hasMessage effMsgId to aids || hasMessage effMsgId to dids then reject (MsDuplicate {})
1154− else if srLim || prLim || rrLim then yield {status: MsRateLimited {}, remoteServers: []}
1157+ else if srLim || prLim || rrLim then reject (MsRateLimited {})
1155− else if count msgs >= maxMessages then yield {status: MsRejected {}, remoteServers: []}
1158+ else if count msgs >= maxMessages then reject (MsRejected {})
1156− else if countWhere msgs (\m -> m.toKey == to) >= maxMessagesPerRecipient then yield {status: MsQueued {}, remoteServers: []}
1159+ else if countWhere msgs (\m -> m.toKey == to) >= maxMessagesPerRecipient then reject (MsQueued {})
1158 pres <- *presence
1161 pres <- *presence
1159 seq <- *seqCounter
1162 seq <- *seqCounter
@@ -1175,7 +1178,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
1175forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
1178forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
1176 forEach servers id (\server -> do
1179 forEach servers id (\server -> do
1177 fwds <- *forwards
1180 fwds <- *forwards
1178− case findFirst fwds (\f -> f.messageId == effMsgId && f.toServer == server) of
1181+ let same = \x -> x.messageId == effMsgId && x.toServer == server
1182+ case findFirst fwds same of
1179 Nothing {} -> yield {}
1183 Nothing {} -> yield {}
1180 Just {value: f} -> do
1184 Just {value: f} -> do
1181 t <- now
1185 t <- now
@@ -1190,7 +1194,7 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
1190 Err _ -> recordFailure server
1194 Err _ -> recordFailure server
1191 Ok r -> when r.ok (do
1195 Ok r -> when r.ok (do
1192 clearFailure server
1196 clearFailure server
1193− atomic (pruneRef forwards (\x -> not (x.messageId == effMsgId && x.toServer == server)))))
1197+ atomic (pruneRef forwards (\x -> not (same x)))))
1195-- ============================================
1199-- ============================================
1196-- Handler: lookup
1200-- Handler: lookup
@@ -1255,9 +1259,7 @@ handleListMessages = \offset authorization clientIp respond -> do
1255 {messages: [], authorized: True {}, count: 0, error: "rate_limited"} (\pubkey -> do
1259 {messages: [], authorized: True {}, count: 0, error: "rate_limited"} (\pubkey -> do
1256 msgs <- *messages
1260 msgs <- *messages
1257 let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))
1261 let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))
1258− let after = drop (clamp 0 (count myMsgs) offset) myMsgs
1259− let result = pluck (takeRelation maxPollEvents after) (\m -> {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt})
1262+ yield (respond {messages: pluck (takeRelation maxPollEvents (drop (clamp 0 (count myMsgs) offset) myMsgs)) (\m -> {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt}), authorized: True {}, count: count myMsgs, error: ""}))
1260− yield (respond {messages: result, authorized: True {}, count: count myMsgs, error: ""}))
1262handleDeleteMessages = \messageIds authorization clientIp respond -> do
1264handleDeleteMessages = \messageIds authorization clientIp respond -> do
1263 t <- now
1265 t <- now
@@ -1345,8 +1347,9 @@ handleVerifyToken = \token scope clientIp respond -> do
1346 found <- atomic do
1348 found <- atomic do
1347 pending <- *pendingVerifications
1349 pending <- *pendingVerifications
1350+ let same = \p -> p.scope == scope && p.token == token
1348− let match = any pending (\p -> p.scope == scope && p.token == token && t - p.createdAt < verificationTokenExpiry)
1351+ let match = any pending (\p -> same p && t - p.createdAt < verificationTokenExpiry)
1349− when match (set *pendingVerifications = removeWhere pending (\p -> p.scope == scope && p.token == token))
1352+ when match (set *pendingVerifications = removeWhere pending same)
1350 yield match
1353 yield match
1351 yield (respond {ok: found})
1354 yield (respond {ok: found})
@@ -1424,7 +1427,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
1424 && t - m.receivedAt >= ackGracePeriod
1427 && t - m.receivedAt >= ackGracePeriod
1425 && none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)
1428 && none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)
1426 && none remainingFwds (\f -> f.messageId == m.messageId && f.toKey == m.toKey))
1429 && none remainingFwds (\f -> f.messageId == m.messageId && f.toKey == m.toKey))
1427− set *messages = removeWhere msgs (\m -> any toReap (\r -> r.messageId == m.messageId && r.toKey == m.toKey))
1430+ set *messages = removeWhere msgs (\m -> hasMessage m.messageId m.toKey toReap)
1428 yield {ackedCount: count removed, results: pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})}
1431 yield {ackedCount: count removed, results: pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})}
1429 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
1432 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
@@ -1560,11 +1563,13 @@ handleUnblockPeer = \server authorization respond -> withMaintAuth authorization
1560 (yield (respond {ok: False {}, unblocked: False {}}))
1563 (yield (respond {ok: False {}, unblocked: False {}}))
1561 (do removed <- removeServerFromList blockedPeers server; yield (respond {ok: True {}, unblocked: removed}))
1564 (do removed <- removeServerFromList blockedPeers server; yield (respond {ok: True {}, unblocked: removed}))
1563−handleListBlocked = \authorization respond -> do
1566+handleListPeersResp = \listRef authorization respond -> do
1564 ok <- verifyMaintenanceAuth authorization
1567 ok <- verifyMaintenanceAuth authorization
1565− list <- *blockedPeers
1568+ list <- *listRef
1566 yield (respond {peers: if ok then list else [], ok: ok})
1569 yield (respond {peers: if ok then list else [], ok: ok})
1571+handleListBlocked = handleListPeersResp blockedPeers
1568handleAllowPeer = \server authorization respond -> withMaintAuth authorization
1573handleAllowPeer = \server authorization respond -> withMaintAuth authorization
1569 (yield (respond {ok: False {}, allowed: 0}))
1574 (yield (respond {ok: False {}, allowed: 0}))
1570 (do n <- addServerToList allowedPeers server; yield (respond {ok: True {}, allowed: n}))
1575 (do n <- addServerToList allowedPeers server; yield (respond {ok: True {}, allowed: n}))
@@ -1573,10 +1578,7 @@ handleUnallowPeer = \server authorization respond -> withMaintAuth authorization
1573 (yield (respond {ok: False {}, removed: False {}}))
1578 (yield (respond {ok: False {}, removed: False {}}))
1574 (do removed <- removeServerFromList allowedPeers server; yield (respond {ok: True {}, removed: removed}))
1579 (do removed <- removeServerFromList allowedPeers server; yield (respond {ok: True {}, removed: removed}))
1576−handleListAllowed = \authorization respond -> do
1581+handleListAllowed = handleListPeersResp allowedPeers
1577− ok <- verifyMaintenanceAuth authorization
1578− list <- *allowedPeers
1579− yield (respond {peers: if ok then list else [], ok: ok})
1581handlePingPeer = \server authorization respond -> withMaintAuth authorization
1583handlePingPeer = \server authorization respond -> withMaintAuth authorization
1582 (yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"}))
1584 (yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"}))
@@ -1603,7 +1605,7 @@ handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth autho
1604 revoked <- atomic do
1606 revoked <- atomic do
1605 ses <- *sessions
1607 ses <- *sessions
1606− pruneRef sessions (\s -> s.pubkey != pubkey)
1608+ set *sessions = removeWhere ses (\s -> s.pubkey == pubkey)
1607 pruneRef presence (\p -> p.pubkey != pubkey || not (isLocalServer p.server))
1609 pruneRef presence (\p -> p.pubkey != pubkey || not (isLocalServer p.server))
1608 yield (countWhere ses (\s -> s.pubkey == pubkey))
1610 yield (countWhere ses (\s -> s.pubkey == pubkey))
1609 yield (respond {ok: True {}, revokedCount: revoked}))
1611 yield (respond {ok: True {}, revokedCount: revoked}))
@@ -1615,7 +1617,7 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
1615 purged <- atomic do
1617 purged <- atomic do
1616 msgs <- *messages
1618 msgs <- *messages
1617 let mine = filter msgs (\m -> m.toKey == pubkey)
1619 let mine = filter msgs (\m -> m.toKey == pubkey)
1618− pruneRef messages (\m -> m.toKey != pubkey)
1620+ set *messages = removeWhere msgs (\m -> m.toKey == pubkey)
1619 pruneRef forwards (\f -> f.toKey != pubkey)
1621 pruneRef forwards (\f -> f.toKey != pubkey)
1620 appendCapped ackedIds (mapWhere mine (\m -> m.originServer != "") (\m -> mkAckedEntry m t)) maxAckedIds (\a -> a.ackedAt)
1622 appendCapped ackedIds (mapWhere mine (\m -> m.originServer != "") (\m -> mkAckedEntry m t)) maxAckedIds (\a -> a.ackedAt)
1621 yield (count mine)
1623 yield (count mine)
@@ -1626,7 +1628,7 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
1627 keys <- if ip == "" then yield [] else atomic do
1629 keys <- if ip == "" then yield [] else atomic do
1628 ses <- *sessions
1630 ses <- *sessions
1629− pruneRef sessions (\s -> s.ip != ip)
1631+ set *sessions = removeWhere ses (\s -> s.ip == ip)
1630 yield (mapWhere ses (\s -> s.ip == ip) (\s -> s.pubkey))
1632 yield (mapWhere ses (\s -> s.ip == ip) (\s -> s.pubkey))
1631 yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count (unique keys)}))
1633 yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count (unique keys)}))