@@ -440,7 +440,7 @@ route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | Federati
440440-- ============================================
441441
442442any : [a] -> (a -> Bool) -> Bool
443−any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
443+any = \xs pred -> count (do x <- xs; where pred x; yield x) > 0
444444none : [a] -> (a -> Bool) -> Bool
445445none = \xs pred -> not (any xs pred)
446446all : [a] -> (a -> Bool) -> Bool
@@ -461,17 +461,15 @@ pruneRef = \ref keep -> do
461461-- Add `server` to the named list ref if absent. Returns the new total.
462462addServerToList = \listRef server -> atomic do
463463 current <- *listRef
464− if any current (\x -> x == server) then yield (count current)
465− else do
466− set *listRef = union current [server]
467− yield (count current + 1)
464+ let updated = if any current (\x -> x == server) then current else union current [server]
465+ set *listRef = updated
466+ yield (count updated)
468467
469468-- Remove `server` from the named list ref. Returns whether it was present.
470469removeServerFromList = \listRef server -> atomic do
471470 current <- *listRef
472− let had = any current (\x -> x == server)
473− set *listRef = do x <- current; where x != server; yield x
474− yield had
471+ pruneRef listRef (\x -> x != server)
472+ yield (any current (\x -> x == server))
475473
476474headOf = \items -> fold (\acc x -> case acc of
477475 Nothing {} -> Just {value: x}
@@ -669,17 +667,16 @@ verifyFederationAuth = \authorization fromServer scope requestTimestamp -> do
669667withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do
670668 blocked <- *blockedPeers
671669 allowed <- *allowedPeers
672− let denyReason =
670+ let preErr =
673671 if isBlockedPeer fromServer blocked then "blocked"
674672 else if not (isAllowedByAllowlist fromServer allowed) then "not_allowed"
673+ else if not precondition then "invalid_request"
674+ else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance) then "clock_skew"
675675 else ""
676− if denyReason != "" then yield (respond {ok: False {}, error: denyReason, retryAfterMs: 0})
676+ if preErr != "" then yield (respond {ok: False {}, error: preErr, retryAfterMs: 0})
677677 else do
678678 verified <- verifyFederationAuth authorization fromServer scope requestTimestamp
679679 if not verified then yield (respond {ok: False {}, error: "unauthorized", retryAfterMs: 0})
680− else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance)
681− then yield (respond {ok: False {}, error: "clock_skew", retryAfterMs: 0})
682− else if not precondition then yield (respond {ok: False {}, error: "invalid_request", retryAfterMs: 0})
683680 else do
684681 limited <- checkServerRateGate fromServer t
685682 if limited then yield (respond {ok: False {}, error: "rate_limited", retryAfterMs: federationRateWindowMs})
@@ -701,29 +698,20 @@ checkRate = \entries key maxRate windowMs maxEntries t ->
701698 let windowStart = case cur of Just {value: r} -> r.windowStart; _ -> t in
702699 {limited: False {}, entries: capList (union others [{key: key, windowStart: windowStart, count: nextCount}]) maxEntries (\r -> r.windowStart)}
703700
704−checkGlobalRate = \t -> atomic do
705− ws <- *globalRateWindowStart
706− c <- *globalRateCount
707− if t - ws >= globalRateWindowMs then do
708− set *globalRateWindowStart = t
709− set *globalRateCount = 1
701+checkWindowRate = \windowRef countRef windowMs maxRate t -> atomic do
702+ ws <- *windowRef
703+ c <- *countRef
704+ if t - ws >= windowMs then do
705+ set *windowRef = t
706+ set *countRef = 1
710707 yield False
711− else if c >= maxGlobalRequestRate then yield True
708+ else if c >= maxRate then yield True
712709 else do
713− set *globalRateCount = c + 1
710+ set *countRef = c + 1
714711 yield False
715712
716−checkAuthGlobalRate = \t -> atomic do
717− ws <- *authGlobalRateWindowStart
718− c <- *authGlobalRateCount
719− if t - ws >= authGlobalRateWindowMs then do
720− set *authGlobalRateWindowStart = t
721− set *authGlobalRateCount = 1
722− yield False
723− else if c >= maxAuthGlobalRate then yield True
724− else do
725− set *authGlobalRateCount = c + 1
726− yield False
713+checkGlobalRate = \t -> checkWindowRate globalRateWindowStart globalRateCount globalRateWindowMs maxGlobalRequestRate t
714+checkAuthGlobalRate = \t -> checkWindowRate authGlobalRateWindowStart authGlobalRateCount authGlobalRateWindowMs maxAuthGlobalRate t
727715
728716checkIpRateGate = \ip t ->
729717 if ip == "" then yield False
@@ -740,13 +728,12 @@ checkServerRateGate = \server t ->
740728
741729checkPreAuth = \clientIp t -> do
742730 dp <- *drainPhase
743− if dp >= 1 then yield "draining"
744− else do
745− g <- checkGlobalRate t
746− if g then yield "global_rate_limited"
747− else do
748− i <- checkIpRateGate clientIp t
749− yield (if i then "ip_rate_limited" else "")
731+ g <- checkGlobalRate t
732+ i <- checkIpRateGate clientIp t
733+ yield (if dp >= 1 then "draining"
734+ else if g then "global_rate_limited"
735+ else if i then "ip_rate_limited"
736+ else "")
750737
751738-- ============================================
752739-- Federation failure tracking (simple, no circuit breaker)
@@ -766,14 +753,12 @@ recordFailure = \server -> do
766753 atomic do
767754 failures <- *forwardFailures
768755 let others = do f <- failures; where f.server != server; yield f
769− let prior = headOf (do f <- failures; where f.server == server; yield f)
770− let firstAt = case prior of Just {value: f} -> f.firstFailedAt; _ -> t
771− let nextN = case prior of Just {value: f} -> min (f.failures + 1) 10; _ -> 1
772− set *forwardFailures = union others [{server: server, failedAt: t, failures: nextN, firstFailedAt: firstAt}]
756+ let new = case headOf (do f <- failures; where f.server == server; yield f) of
757+ Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
758+ Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}
759+ set *forwardFailures = union others [new]
773760
774−clearFailure = \server -> atomic do
775− failures <- *forwardFailures
776− set *forwardFailures = do f <- failures; where f.server != server; yield f
761+clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))
777762
778763-- ============================================
779764-- Common utility queries
@@ -1135,9 +1120,7 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -
11351120 then yield (respond {events: [], cursor: cursor, authorized: True {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: pollTimeoutMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: []})
11361121 else do
11371122 pollResult <- doPoll s t
1138− atomic do
1139− aps <- *activePolls
1140− set *activePolls = do p <- aps; where p.pubkey != s.pubkey; yield p
1123+ atomic (pruneRef activePolls (\p -> p.pubkey != s.pubkey))
11411124 when (federationEnabled && count s.pendingAcks > 0)
11421125 (fork (sendAcksAndClean s.pendingAcks))
11431126 presenceUpdates <- if count watchKeys > 0 && pollResult.authorized
@@ -1231,58 +1214,50 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
12311214 MsQueued {} -> True {}; MsDuplicate {} -> True {}; _ -> False {}
12321215 yield (respond {status: result.status, messageId: if ok then effMsgId else "", retryAfterMs: 0, error: if ok then "" else "rejected", loadPct: 0})
12331216
1234−sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msgPriority t ->
1217+sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msgPriority t -> do
1218+ msgs <- *messages
1219+ aids <- *ackedIds
1220+ dids <- *deliveredIds
1221+ let isDup = (effMsgId != "" && any msgs (\m -> m.messageId == effMsgId && m.toKey == to))
1222+ || hasMessage effMsgId to aids
1223+ || hasMessage effMsgId to dids
1224+ sendRates <- *sendRates
1225+ let sr = checkRate sendRates senderKey maxSendRatePerKey sendRateWindowMs 5000 t
1226+ set *sendRates = sr.entries
1227+ pairRates <- *pairRates
1228+ let pr = checkRate pairRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs 5000 t
1229+ set *pairRates = pr.entries
1230+ recvRates <- *recvRates
1231+ let rr = checkRate recvRates to maxRecvRatePerKey recvRateWindowMs 5000 t
1232+ set *recvRates = rr.entries
1233+ let recipientCount = count (do m <- msgs; where m.toKey == to; yield m)
12351234 if senderKey == to then yield {status: MsRejected {}, remoteServers: []}
1235+ else if isDup then yield {status: MsDuplicate {}, remoteServers: []}
1236+ else if sr.limited || pr.limited || rr.limited then yield {status: MsRateLimited {}, remoteServers: []}
1237+ else if count msgs >= maxMessages then yield {status: MsRejected {}, remoteServers: []}
1238+ else if recipientCount >= maxMessagesPerRecipient then yield {status: MsQueued {}, remoteServers: []}
12361239 else do
1237− msgs <- *messages
1238− aids <- *ackedIds
1239− dids <- *deliveredIds
1240− let isDup = (effMsgId != "" && any msgs (\m -> m.messageId == effMsgId && m.toKey == to))
1241− || hasMessage effMsgId to aids
1242− || hasMessage effMsgId to dids
1243− if isDup then yield {status: MsDuplicate {}, remoteServers: []}
1244− else do
1245− sendRates <- *sendRates
1246− let sr = checkRate sendRates senderKey maxSendRatePerKey sendRateWindowMs 5000 t
1247− if sr.limited then yield {status: MsRateLimited {}, remoteServers: []}
1248− else do
1249− set *sendRates = sr.entries
1250− pairRates <- *pairRates
1251− let pr = checkRate pairRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs 5000 t
1252− set *pairRates = pr.entries
1253− if pr.limited then yield {status: MsRateLimited {}, remoteServers: []}
1254− else do
1255− recvRates <- *recvRates
1256− let rr = checkRate recvRates to maxRecvRatePerKey recvRateWindowMs 5000 t
1257− if rr.limited then yield {status: MsRateLimited {}, remoteServers: []}
1258− else do
1259− set *recvRates = rr.entries
1260− if count msgs >= maxMessages then yield {status: MsRejected {}, remoteServers: []}
1261− else
1262− let recipientCount = count (do m <- msgs; where m.toKey == to; yield m) in
1263− if recipientCount >= maxMessagesPerRecipient then yield {status: MsQueued {}, remoteServers: []}
1264− else do
1265− pres <- *presence
1266− seq <- *seqCounter
1267− let seqTs = nextSeqTs t seq
1268− set *seqCounter = seqTs
1269− let expiresAt = seqTs + effectiveMessageTtl expiresInMs
1270− set *messages = union msgs [{messageId: effMsgId, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: expiresAt, senderKey: senderKey, priority: msgPriority}]
1271− set *pollWakeup = seqTs
1272− let toPresence = presenceForKey pres to t
1273− let localOnline = any toPresence (\p -> isLocalServer p.server)
1274− let remoteServers = collectPeers toPresence (\p -> p.server)
1275− forwardsEnqueued <- if federationEnabled && count remoteServers > 0
1276− then do
1277− fwds <- *forwards
1278− let newFwds = do s <- remoteServers; where none fwds (\f -> f.messageId == effMsgId && f.toServer == s); yield {messageId: effMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: expiresAt, priority: msgPriority, senderKey: senderKey}
1279− set *forwards = capList (union fwds newFwds) maxForwardsEntries (\f -> f.createdAt)
1280− yield (count newFwds > 0)
1281− else yield False
1282− let status = if localOnline then MsDelivered {}
1283− else if forwardsEnqueued then MsFederated {}
1284− else MsQueued {}
1285− yield {status: status, remoteServers: remoteServers}
1240+ pres <- *presence
1241+ seq <- *seqCounter
1242+ let seqTs = nextSeqTs t seq
1243+ set *seqCounter = seqTs
1244+ let expiresAt = seqTs + effectiveMessageTtl expiresInMs
1245+ set *messages = union msgs [{messageId: effMsgId, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: expiresAt, senderKey: senderKey, priority: msgPriority}]
1246+ set *pollWakeup = seqTs
1247+ let toPresence = presenceForKey pres to t
1248+ let localOnline = any toPresence (\p -> isLocalServer p.server)
1249+ let remoteServers = collectPeers toPresence (\p -> p.server)
1250+ forwardsEnqueued <- if federationEnabled && count remoteServers > 0
1251+ then do
1252+ fwds <- *forwards
1253+ let newFwds = do s <- remoteServers; where none fwds (\f -> f.messageId == effMsgId && f.toServer == s); yield {messageId: effMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: expiresAt, priority: msgPriority, senderKey: senderKey}
1254+ set *forwards = capList (union fwds newFwds) maxForwardsEntries (\f -> f.createdAt)
1255+ yield (count newFwds > 0)
1256+ else yield False
1257+ let status = if localOnline then MsDelivered {}
1258+ else if forwardsEnqueued then MsFederated {}
1259+ else MsQueued {}
1260+ yield {status: status, remoteServers: remoteServers}
12861261
12871262forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
12881263 forEach servers (\s -> s) (\server -> do
@@ -1894,9 +1869,8 @@ handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth autho
18941869 revoked <- atomic do
18951870 ses <- *sessions
18961871 let mine = do s <- ses; where s.pubkey == pubkey; yield s
1897− set *sessions = do s <- ses; where s.pubkey != pubkey; yield s
1898− pres <- *presence
1899− set *presence = do p <- pres; where p.pubkey != pubkey || not (isLocalServer p.server); yield p
1872+ pruneRef sessions (\s -> s.pubkey != pubkey)
1873+ pruneRef presence (\p -> p.pubkey != pubkey || not (isLocalServer p.server))
19001874 yield (count mine)
19011875 yield (respond {ok: True {}, revokedCount: revoked}))
19021876
@@ -1907,9 +1881,8 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
19071881 purged <- atomic do
19081882 msgs <- *messages
19091883 let mine = do m <- msgs; where m.toKey == pubkey; yield m
1910− set *messages = do m <- msgs; where m.toKey != pubkey; yield m
1911− fwds <- *forwards
1912− set *forwards = do f <- fwds; where f.toKey != pubkey; yield f
1884+ pruneRef messages (\m -> m.toKey != pubkey)
1885+ pruneRef forwards (\f -> f.toKey != pubkey)
19131886 let federated = do m <- mine; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}
19141887 aids <- *ackedIds
19151888 set *ackedIds = capList (union aids federated) maxAckedIds (\a -> a.ackedAt)
@@ -1923,7 +1896,7 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
19231896 result <- atomic do
19241897 ses <- *sessions
19251898 let mine = do s <- ses; where s.ip == ip; yield s
1926− set *sessions = do s <- ses; where s.ip != ip; yield s
1899+ pruneRef sessions (\s -> s.ip != ip)
19271900 let keys = unique (do s <- mine; yield s.pubkey)
19281901 yield {revoked: count mine, keys: count keys}
19291902 yield (respond {ok: True {}, revokedCount: result.revoked, affectedKeys: result.keys}))