@@ -631,6 +631,12 @@ gateRated = \checkDrain auth ip rateRef maxRate windowMs t respond mkErr rateRes
631631 lim <- checkPubkeyRateGate rateRef pk maxRate windowMs t
632632 if lim then yield (respond rateResp) else onAuth pk)
633633
634+gateClientWrite = \auth ip t respond mkErr rateResp onAuth ->
635+ gateRated (True {}) auth ip clientWriteRates maxClientWriteRatePerKey clientWriteRateWindowMs t respond mkErr rateResp onAuth
636+
637+gateLookup = \auth ip t respond mkErr rateResp onAuth ->
638+ gateRated (False {}) auth ip lookupRates maxLookupRatePerKey lookupRateWindowMs t respond mkErr rateResp onAuth
639+
634640withSessionAuth = \authorization t clientIp onUnauth onAuth ->
635641 case extractToken authorization of
636642 Nothing {} -> onUnauth
@@ -760,17 +766,18 @@ retryBackoff = \failures -> min retryBackoffMax (retryBackoffBase * max 1 failur
760766
761767shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
762768shouldRetryServer = \server t failures ->
763− all failures (\f -> f.server != server || t - f.failedAt >= retryBackoff f.failures)
769+ none failures (\f -> f.server == server && t - f.failedAt < retryBackoff f.failures)
764770
765771recordFailure = \server -> do
766772 t <- now
767773 atomic do
768774 failures <- *forwardFailures
769775 let same = \f -> f.server == server
770− let entry = case findFirst failures same of
771− Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
772− Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}
773− set *forwardFailures = union (removeWhere failures same) [entry]
776+ let prev = case findFirst failures same of
777+ Just {value: f} -> {count: f.failures, firstFailedAt: f.firstFailedAt}
778+ _ -> {count: 0, firstFailedAt: t}
779+ set *forwardFailures = union (removeWhere failures same)
780+ [{server: server, failedAt: t, failures: min (prev.count + 1) 10, firstFailedAt: prev.firstFailedAt}]
774781
775782clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))
776783
@@ -806,6 +813,9 @@ maskedServer = \pres pubkey t ->
806813hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool
807814hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)
808815
816+isDuplicateMessage = \mid toKey msgs aids dids ->
817+ hasMessage mid toKey msgs || hasMessage mid toKey aids || hasMessage mid toKey dids
818+
809819collectPeers : [a] -> (a -> Text) -> [Text]
810820collectPeers = \items getServer ->
811821 unique (filter (pluck items getServer) (\s -> not (isLocalServer s)))
@@ -873,6 +883,15 @@ applyCancel = \msgs matches to mid t -> do
873883 appendCapped cancelNotifications [{toKey: to, messageId: mid, cancelledAt: t}] 10000 (\c -> c.cancelledAt)
874884 yield (collectPeers (filter fwds matches) (\f -> f.toServer))
875885
886+-- Cancel one message if `ownerPred` holds for it; returns whether a cancel happened.
887+cancelOne = \mid to ownerPred t -> atomic do
888+ let matches = \x -> x.messageId == mid && x.toKey == to
889+ msgs <- *messages
890+ if any msgs (\m -> matches m && ownerPred m) then do
891+ _ <- applyCancel msgs matches to mid t
892+ yield True
893+ else yield False
894+
876895sendBatchedGossip = \peers events hops -> forEach peers id (\server -> do
877896 t <- now
878897 failures <- *forwardFailures
@@ -1017,9 +1036,13 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
10171036 else do
10181037 set *challenges = removeWhere ch same
10191038 ses <- *sessions
1020− 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)
1039+ let mine = \s -> s.pubkey == pubkey
1040+ let pruned = if revokeOthers then removeWhere ses mine else capPerKey ses mine maxSessionsPerKey (\s -> s.expiresAt)
1041+ let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
1042+ set *sessions = capList (union pruned [newSession]) maxSessions (\s -> s.expiresAt)
10211043 pres <- *presence
1022− 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)
1044+ let prunedPres = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
1045+ let newPres = capList (union prunedPres [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
10231046 set *presence = newPres
10241047 yield {valid: True {}, peers: knownPeerServersFromPresence newPres}
10251048 if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})
@@ -1166,7 +1189,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
11661189 rrLim <- chargeRate recvRates to maxRecvRatePerKey recvRateWindowMs t
11671190 let reject = \s -> yield {status: s, remoteServers: []}
11681191 if senderKey == to then reject (MsRejected {})
1169− else if hasMessage effMsgId to msgs || hasMessage effMsgId to aids || hasMessage effMsgId to dids then reject (MsDuplicate {})
1192+ else if isDuplicateMessage effMsgId to msgs aids dids then reject (MsDuplicate {})
11701193 else if srLim || prLim || rrLim then reject (MsRateLimited {})
11711194 else if count msgs >= maxMessages then reject (MsRejected {})
11721195 else if countWhere msgs (\m -> m.toKey == to) >= maxMessagesPerRecipient then reject (MsQueued {})
@@ -1216,7 +1239,7 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
12161239handleLookupPresence = \pubkey authorization clientIp respond -> do
12171240 t <- now
12181241 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}
1219− gateRated (False {}) authorization clientIp lookupRates maxLookupRatePerKey lookupRateWindowMs t respond mkErr
1242+ gateLookup authorization clientIp t respond mkErr
12201243 {server: "", authorized: True {}, error: "rate_limited"} (\_ -> do
12211244 pres <- *presence
12221245 yield (respond {server: maskedServer pres pubkey t, authorized: True {}, error: ""}))
@@ -1224,7 +1247,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
12241247handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
12251248 t <- now
12261249 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}
1227− gateRated (False {}) authorization clientIp lookupRates maxLookupRatePerKey lookupRateWindowMs t respond mkErr
1250+ gateLookup authorization clientIp t respond mkErr
12281251 {results: [], authorized: True {}, error: "rate_limited"} (\_ -> do
12291252 pres <- *presence
12301253 let results = do
@@ -1254,7 +1277,7 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
12541277handleAckMessages = \messageIds authorization clientIp respond -> do
12551278 t <- now
12561279 let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}
1257− gateRated (True {}) authorization clientIp clientWriteRates maxClientWriteRatePerKey clientWriteRateWindowMs t respond mkErr
1280+ gateClientWrite authorization clientIp t respond mkErr
12581281 {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"} (\pubkey -> do
12591282 let dedupedIds = dedupMessageIds messageIds
12601283 let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId
@@ -1272,12 +1295,14 @@ handleListMessages = \offset authorization clientIp respond -> do
12721295 {messages: [], authorized: True {}, count: 0, error: "rate_limited"} (\pubkey -> do
12731296 msgs <- *messages
12741297 let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))
1275− 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: ""}))
1298+ let total = count myMsgs
1299+ let page = pluck (takeRelation maxPollEvents (drop (clamp 0 total offset) myMsgs)) (\m -> {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt})
1300+ yield (respond {messages: page, authorized: True {}, count: total, error: ""}))
12761301
12771302handleDeleteMessages = \messageIds authorization clientIp respond -> do
12781303 t <- now
12791304 let mkErr = \e -> {ok: False {}, authorized: False {}, deletedCount: 0, error: if e == "draining" then "" else e}
1280− gateRated (True {}) authorization clientIp clientWriteRates maxClientWriteRatePerKey clientWriteRateWindowMs t respond mkErr
1305+ gateClientWrite authorization clientIp t respond mkErr
12811306 {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"} (\pubkey -> do
12821307 let dedupedIds = dedupMessageIds messageIds
12831308 let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId
@@ -1292,7 +1317,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
12921317handleCancelMessage = \messageId to authorization clientIp respond -> do
12931318 t <- now
12941319 let mkErr = \e -> {ok: False {}, authorized: False {}, cancelled: False {}, error: if e == "draining" then "" else e}
1295− gateRated (True {}) authorization clientIp clientWriteRates maxClientWriteRatePerKey clientWriteRateWindowMs t respond mkErr
1320+ gateClientWrite authorization clientIp t respond mkErr
12961321 {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited"} (\senderKey -> do
12971322 let matches = \x -> x.messageId == messageId && x.toKey == to
12981323 result <- atomic do
@@ -1320,16 +1345,10 @@ sendCancelToEach = \messageId to servers ->
13201345handleCancelBatch = \cancels authorization clientIp respond -> do
13211346 t <- now
13221347 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "draining" then "" else e}
1323− gateRated (True {}) authorization clientIp clientWriteRates maxClientWriteRatePerKey clientWriteRateWindowMs t respond mkErr
1348+ gateClientWrite authorization clientIp t respond mkErr
13241349 {results: [], authorized: True {}, error: "rate_limited"} (\senderKey -> do
13251350 results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do
1326− let matches = \x -> x.messageId == e.messageId && x.toKey == e.to
1327− r <- atomic do
1328− msgs <- *messages
1329− if any msgs (\m -> matches m && m.originServer == "" && m.senderKey == senderKey) then do
1330− _ <- applyCancel msgs matches e.to e.messageId t
1331− yield True
1332− else yield False
1351+ r <- cancelOne e.messageId e.to (\m -> m.originServer == "" && m.senderKey == senderKey) t
13331352 yield {messageId: e.messageId, cancelled: r})
13341353 yield (respond {results: results, authorized: True {}, error: ""}))
13351354
@@ -1394,25 +1413,24 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
13941413-- "no_presence" | "global_full" | "mailbox_full" | "rate_limited" (only when checkRecvRate).
13951414storeFederated = \toKey encryptedBlob timestamp originServer messageId expiresAt priority checkRecvRate t -> do
13961415 pres <- *presence
1416+ msgs <- *messages
1417+ aids <- *ackedIds
1418+ dids <- *deliveredIds
13971419 if not (any pres (\p -> p.pubkey == toKey && p.expiresAt > t)) then yield "no_presence"
1420+ else if isDuplicateMessage messageId toKey msgs aids dids then yield ""
1421+ else if count msgs >= maxMessages then yield "global_full"
1422+ else if countWhere msgs (\m -> m.toKey == toKey) >= maxMessagesPerRecipient then yield "mailbox_full"
13981423 else do
1399− msgs <- *messages
1400− aids <- *ackedIds
1401− dids <- *deliveredIds
1402− if hasMessage messageId toKey msgs || hasMessage messageId toKey aids || hasMessage messageId toKey dids
1403− then yield ""
1404− else if count msgs >= maxMessages then yield "global_full"
1405− else if countWhere msgs (\m -> m.toKey == toKey) >= maxMessagesPerRecipient then yield "mailbox_full"
1424+ limited <- if checkRecvRate then chargeRate recvRates toKey maxRecvRatePerKey recvRateWindowMs t else yield False
1425+ if limited then yield "rate_limited"
14061426 else do
1407− limited <- if checkRecvRate then chargeRate recvRates toKey maxRecvRatePerKey recvRateWindowMs t else yield False
1408− if limited then yield "rate_limited"
1409− else do
1410− seq <- *seqCounter
1411− let seqTs = nextSeqTs t seq
1412− set *seqCounter = seqTs
1413− set *messages = union msgs [{messageId: messageId, toKey: toKey, encryptedBlob: encryptedBlob, timestamp: clampTimestamp timestamp t, originServer: originServer, receivedAt: seqTs, expiresAt: if expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge, senderKey: "", priority: clamp 0 maxMessagePriority priority}]
1414− set *pollWakeup = seqTs
1415− yield ""
1427+ seq <- *seqCounter
1428+ let seqTs = nextSeqTs t seq
1429+ let cappedExpiry = if expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
1430+ set *seqCounter = seqTs
1431+ set *messages = union msgs [{messageId: messageId, toKey: toKey, encryptedBlob: encryptedBlob, timestamp: clampTimestamp timestamp t, originServer: originServer, receivedAt: seqTs, expiresAt: cappedExpiry, senderKey: "", priority: clamp 0 maxMessagePriority priority}]
1432+ set *pollWakeup = seqTs
1433+ yield ""
14161434
14171435handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond -> do
14181436 t <- now
@@ -1450,13 +1468,7 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
14501468 t <- now
14511469 let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}, retryAfterMs: r.retryAfterMs}
14521470 withFederationAuth authorization t cancelRespond fromServer ("cancel:" ++ messageId) requestTimestamp (not (isLocalServer fromServer)) (do
1453− let matches = \x -> x.messageId == messageId && x.toKey == toKey
1454− cancelled <- atomic do
1455− msgs <- *messages
1456− if any msgs (\m -> matches m && m.originServer == fromServer) then do
1457− _ <- applyCancel msgs matches toKey messageId t
1458− yield True
1459− else yield False
1471+ cancelled <- cancelOne messageId toKey (\m -> m.originServer == fromServer) t
14601472 yield (respond {ok: True {}, error: "", cancelled: cancelled, retryAfterMs: 0}))
14611473
14621474handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId authorization respond -> do
@@ -1464,13 +1476,7 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
14641476 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
14651477 withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (do
14661478 results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do
1467− let matches = \x -> x.messageId == e.messageId && x.toKey == e.toKey
1468− cancelled <- atomic do
1469− msgs <- *messages
1470− if any msgs (\m -> matches m && m.originServer == fromServer) then do
1471− _ <- applyCancel msgs matches e.toKey e.messageId t
1472− yield True
1473− else yield False
1479+ cancelled <- cancelOne e.messageId e.toKey (\m -> m.originServer == fromServer) t
14741480 yield {messageId: e.messageId, cancelled: cancelled})
14751481 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
14761482
@@ -1648,7 +1654,8 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
16481654 let same = \s -> s.ip == ip
16491655 set *sessions = removeWhere ses same
16501656 yield (mapWhere ses same (\s -> s.pubkey))
1651− yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count (unique keys)}))
1657+ let uniqueKeys = unique keys
1658+ yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count uniqueKeys}))
16521659
16531660handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization
16541661 (yield (respond {sessions: [], ok: False {}, total: 0}))
@@ -1662,7 +1669,9 @@ handleAdminListAllSessions = \limit authorization respond -> withMaintAuth autho
16621669 count: count mine,
16631670 ips: unique (mapWhere mine (\s -> s.ip != "") (\s -> s.ip)),
16641671 latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine})
1665− 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}))
1672+ let cap = if limit <= 0 then 100 else clamp 1 500 limit
1673+ let top = takeRelation cap (sortBy (\g -> 0 - g.count) grouped)
1674+ yield (respond {sessions: top, ok: True {}, total: count grouped}))
16661675
16671676handleGetRateLimits = \authorization respond -> do
16681677 ok <- verifyMaintenanceAuth authorization
@@ -1754,7 +1763,9 @@ handleTopRecipients = \limit authorization respond -> withMaintAuth authorizatio
17541763 (do
17551764 msgs <- *messages
17561765 let counts = pluck (unique (pluck msgs (\m -> m.toKey))) (\pk -> {pubkey: pk, count: countWhere msgs (\m -> m.toKey == pk)})
1757− yield (respond {recipients: takeRelation (if limit <= 0 then 20 else clamp 1 100 limit) (sortBy (\c -> 0 - c.count) counts), ok: True {}, total: count counts}))
1766+ let cap = if limit <= 0 then 20 else clamp 1 100 limit
1767+ let top = takeRelation cap (sortBy (\c -> 0 - c.count) counts)
1768+ yield (respond {recipients: top, ok: True {}, total: count counts}))
17581769
17591770handleStatsForwards = \authorization respond -> withMaintAuth authorization
17601771 (yield (respond {servers: [], ok: False {}, total: 0}))