@@ -56,6 +56,11 @@ maxSendRatePerKey = 20 -- max messages sent per key per window
5656sendRateWindowMs : Int<Ms>
5757sendRateWindowMs = 1000
5858maxSendRateEntries = 1000
59+-- Per-recipient receive rate limiting
60+maxRecvRatePerKey = 10 -- max messages received by a key per window (across all senders)
61+recvRateWindowMs : Int<Ms>
62+recvRateWindowMs = 1000
63+maxRecvRateEntries = 1000
5964-- Federation rate limiting (per-server)
6065maxFederationRatePerServer = 50 -- max federation requests per server per window
6166federationRateWindowMs : Int<Ms>
@@ -98,6 +103,24 @@ presenceRefreshThreshold = presenceRefreshInterval * 2 -- refresh entries expiri
98103federationEnabled = federationSecret != ""
99104federationAuth = "Bearer " ++ federationSecret
100105federationUrl = \server -> federationScheme ++ server
106+-- IP-based rate limiting (reverse proxy sets ClientIp header)
107+maxIpRatePerIp = 10 -- max auth requests per IP per window
108+ipRateWindowMs : Int<Ms>
109+ipRateWindowMs = 1000
110+maxIpRateEntries = 5000
111+-- Lookup rate limiting (per-key, on presence lookups)
112+maxLookupRatePerKey = 5 -- max lookups per key per window
113+lookupRateWindowMs : Int<Ms>
114+lookupRateWindowMs = 1000
115+maxLookupRateEntries = 1000
116+-- Unacked message aging (messages pruned after this regardless of ack status)
117+unackedMessageMaxAge : Int<Ms>
118+unackedMessageMaxAge = 604800000 -- 7 days
119+-- Health check thresholds (background task freshness)
120+healthPruneMaxAge : Int<Ms>
121+healthPruneMaxAge = backgroundPruneInterval * 3
122+healthHeartbeatMaxAge : Int<Ms>
123+healthHeartbeatMaxAge = pollTimeoutMs * 3
101124
102125-- ============================================
103126-- Types
@@ -166,7 +189,7 @@ type AnyServerName = Text where \s -> not (tooLong s maxServerNameLen) && s == n
166189type Timestamp = Int<Ms>
167190
168191-- Pending challenge for Ed25519 challenge-response auth
169−type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp}
192+type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp, ip: Text}
170193-- Authenticated session; token is a random hex string
171194type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp}
172195-- Stored message; originServer is "" for locally-originated messages, set to sender server for federated forwards
@@ -183,6 +206,7 @@ type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: T
183206type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFailedAt: Timestamp}
184207-- Sliding-window rate limit state per pubkey
185208type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
209+type RecvRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
186210type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
187211-- Sliding-window rate limit state per federation peer server
188212type FederationRateEntry = {server: Text, windowStart: Timestamp, count: Int}
@@ -192,6 +216,10 @@ data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
192216data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName]} | SendDuplicate {}
193217data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
194218
219+type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int}
220+type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
221+type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool}
222+
195223-- ============================================
196224-- Persisted State
197225-- ============================================
@@ -270,6 +298,16 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
270298
271299*lastHeartbeatAt : Timestamp
272300
301+*draining : Bool
302+
303+*pruneRunning : Timestamp
304+
305+*ipRates : [IpRateEntry]
306+
307+*lookupRates : [LookupRateEntry]
308+
309+*recvRates : [RecvRateEntry]
310+
273311-- ============================================
274312-- Routes
275313-- ============================================
@@ -280,10 +318,10 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
280318-- 3. Server verifies signature, returns a session token
281319route AuthApi where
282320 /auth
283− POST {pubkey: PubkeyHex} /challenge
321+ POST {pubkey: PubkeyHex} /challenge headers {clientIp: Text}
284322 -> {challenge: Text, ok: Bool, expiresAt: Timestamp}
285323 = ReqChallenge
286− POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex} /verify
324+ POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /verify headers {clientIp: Text}
287325 -> {token: Text, expiresAt: Timestamp}
288326 = VerifyAuth
289327
@@ -308,15 +346,15 @@ route FederationApi where
308346 /federation
309347 -- Receive presence gossip events from a peer server
310348 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName} /gossip headers {authorization: Text}
311− -> {ok: Bool}
349+ -> {ok: Bool, error: Text}
312350 = RecvGossip
313351 -- Receive a forwarded message from a peer server on behalf of a sender
314352 POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName} /forward headers {authorization: Text}
315− -> {ok: Bool}
353+ -> {ok: Bool, error: Text}
316354 = ForwardMessage
317355 -- Acknowledge delivery of messages back to the origin server
318356 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName} /ack headers {authorization: Text}
319− -> {ok: Bool}
357+ -> {ok: Bool, error: Text}
320358 = FederationAck
321359
322360-- Triggers pruning of expired state and retry of pending federation ops
@@ -329,13 +367,16 @@ route MaintenanceApi where
329367 = Drain
330368
331369route HealthApi where
332− GET /health -> {ok: Bool} = Health
370+ GET /health -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool} = Health
333371
334372-- Returns counts of in-memory collections (requires federation or maintenance secret)
335373route StatsApi where
336374 GET /stats headers {authorization: Text}
337− -> {sessions: Int, messages: Int, presence: Int, forwards: Int, challenges: Int, ackedIds: Int, ackOverflow: Int, deliveredIds: Int, forwardFailures: Int, activePolls: Int, sendRates: Int, challengeRates: Int, federationRates: Int, pruneCount: Int, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, ok: Bool}
375+ -> {sessions: Int, sessionsMax: Int, messages: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pruneCount: Int, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, ok: Bool}
338376 = Stats
377+ GET /stats/peers headers {authorization: Text}
378+ -> {peers: [PeerStatus], ok: Bool}
379+ = StatsPeers
339380
340381route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi | HealthApi | StatsApi
341382
@@ -541,7 +582,7 @@ isBadServerName = \sn ->
541582 || matchesHostname sn "localhost.localdomain"
542583 || matchesHostnameOrSubdomain sn "local"
543584 || matchesHostnameOrSubdomain sn "internal"
544− || matchesHostname sn "metadata"
585+ || matchesHostnameOrSubdomain sn "metadata"
545586 || take 4 sn == "127." || take 3 sn == "10."
546587 || take 8 sn == "192.168." || take 8 sn == "169.254."
547588 || take 2 sn == "0." || take 1 sn == "["
@@ -638,7 +679,7 @@ safeCompare : Text -> Text -> Bool
638679safeCompare = \a b ->
639680 let lenA = count (chars a) in
640681 let lenB = count (chars b) in
641− safeCompareLoop a b maxTokenLen 0
682+ safeCompareLoop a b (max maxTokenLen (max lenA lenB)) 0
642683 + (if lenA != lenB then 1 else 0)
643684 + (if a == "" then 1 else 0)
644685 + (if b == "" then 1 else 0) == 0
@@ -692,12 +733,9 @@ capPerKeyAndGlobal = \items matchKey maxPerKey getExpiry t maxGlobal ->
692733 capGlobal (capPerKey items matchKey maxPerKey getExpiry t) maxGlobal getExpiry
693734
694735removeOneForKey = \items pubkey ->
695− let result = fold (\acc x ->
696− if not acc.removed && x.pubkey == pubkey
697− then {removed: True {}, items: acc.items}
698− else {removed: acc.removed, items: union acc.items [x]}
699− ) {removed: False {}, items: []} items
700− in result.items
736+ let forKey = do x <- items; where x.pubkey == pubkey; yield x in
737+ let others = do x <- items; where x.pubkey != pubkey; yield x in
738+ union others (evictOneOldest forKey (\x -> x.startedAt))
701739
702740-- Extract unique remote (non-local) server names from a list of items
703741collectPeerServers : [a] -> (a -> Text) -> [Text]
@@ -799,13 +837,14 @@ pruneExpired = \t -> do
799837 let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)
800838 let presenceAfter = do x <- presenceBefore; where x.expiresAt > t; yield x
801839 set *presence = presenceAfter
802− let messageCutoff = t - messageMaxAge
840+ let messageCutoff = t - unackedMessageMaxAge
841+ let retainCutoff = t - messageMaxAge
803842 messages <- *messages
804843 set *messages = do m <- messages; where m.receivedAt > messageCutoff; yield m
805844 forwards <- *forwards
806− set *forwards = do f <- forwards; where f.createdAt > messageCutoff; yield f
845+ set *forwards = do f <- forwards; where f.createdAt > retainCutoff; yield f
807846 ackedIds <- *ackedIds
808− set *ackedIds = do a <- ackedIds; where a.ackedAt > messageCutoff; yield a
847+ set *ackedIds = do a <- ackedIds; where a.ackedAt > retainCutoff; yield a
809848 sendRates <- *sendRates
810849 set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r
811850 forwardFailures <- *forwardFailures
@@ -816,10 +855,16 @@ pruneExpired = \t -> do
816855 set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs; yield r
817856 activePolls <- *activePolls
818857 set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p
858+ ipRates <- *ipRates
859+ set *ipRates = do r <- ipRates; where t - r.windowStart < ipRateWindowMs; yield r
860+ lookupRates <- *lookupRates
861+ set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs; yield r
862+ recvRates <- *recvRates
863+ set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r
819864 ackOverflow <- *ackOverflow
820− set *ackOverflow = do a <- ackOverflow; where a.ackedAt > messageCutoff; yield a
865+ set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a
821866 deliveredIds <- *deliveredIds
822− set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > messageCutoff; yield d
867+ set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > retainCutoff; yield d
823868 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
824869
825870-- Opportunistic pruning: runs at most once per second, triggered by request handlers
@@ -910,25 +955,53 @@ checkGlobalRate = \t -> atomic do
910955 set *globalRateCount = rateCount + 1
911956 yield (False {})
912957
958+-- IP-based rate limiting; skips if clientIp header is empty (no reverse proxy)
959+checkIpRate = \ip t ->
960+ if ip == "" then yield (False {})
961+ else atomic do
962+ ipRates <- *ipRates
963+ let result = checkRateWindow ipRates (\r -> r.ip == ip)
964+ (\ts -> {ip: ip, windowStart: ts, count: 1})
965+ (\r -> {ip: r.ip, windowStart: r.windowStart, count: r.count + 1})
966+ maxIpRatePerIp ipRateWindowMs maxIpRateEntries t
967+ set *ipRates = result.entries
968+ yield result.limited
969+
970+-- Combined pre-auth gate: drain check, global rate, IP rate. Returns "" if all pass.
971+checkPreAuth = \clientIp t -> do
972+ draining <- *draining
973+ if draining then yield "draining"
974+ else do
975+ globalLimited <- checkGlobalRate t
976+ if globalLimited then yield "global_rate_limited"
977+ else do
978+ ipLimited <- checkIpRate clientIp t
979+ if ipLimited then yield "ip_rate_limited"
980+ else yield ""
981+
913982withFederationAuth = \authorization t respond fromServer precondition action ->
914− if not (verifyFederationAuth authorization) || not precondition
983+ if not (verifyFederationAuth authorization)
915984 then do
916985 maybePruneAndBroadcast t
917− yield (respond {ok: False {}})
986+ yield (respond {ok: False {}, error: "unauthorized"})
987+ else if not precondition
988+ then do
989+ maybePruneAndBroadcast t
990+ yield (respond {ok: False {}, error: "invalid_request"})
918991 else do
919992 globalLimited <- checkGlobalRate t
920993 if globalLimited
921994 then do
922995 println ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)
923996 maybePruneAndBroadcast t
924− yield (respond {ok: False {}})
997+ yield (respond {ok: False {}, error: "rate_limited"})
925998 else do
926999 rateLimited <- checkFederationRate fromServer t
9271000 if rateLimited
9281001 then do
9291002 println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
9301003 maybePruneAndBroadcast t
931− yield (respond {ok: False {}})
1004+ yield (respond {ok: False {}, error: "rate_limited"})
9321005 else action
9331006
9341007withSessionAuth = \authorization t onUnauth onAuth -> do
@@ -1357,8 +1430,8 @@ forwardCandidates = \targets messages forwards t ->
13571430 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp})) [] targets
13581431
13591432
1360−challengeMatch = \challenges pubkey challenge t ->
1361− any challenges (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t)
1433+challengeMatch = \challenges pubkey challenge ip t ->
1434+ any challenges (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && (ip == "" || c.ip == "" || c.ip == ip))
13621435
13631436challengesExcluding = \challenges pubkey challenge ->
13641437 do c <- challenges; where c.pubkey != pubkey || c.challenge != challenge; yield c
@@ -1387,9 +1460,11 @@ gossipTargets = \peerServers fromServer ->
13871460newlyOnlineRemoteKeys = \events appliedKeys ->
13881461 do e <- events; where any appliedKeys (\a -> samePubkeyServer a e); where not (isLocalServer e.server); yield {pubkey: e.pubkey, server: e.server}
13891462
1390−-- Trust gossip entries if they're from the sender or about a server we already know about
1463+-- Trust gossip entries if they're first-party (server == fromServer) or updating
1464+-- an existing (pubkey, server) presence entry. Third-party gossip cannot create
1465+-- new entries for keys we haven't seen — only the key's home server can do that.
13911466trustedGossipEntries = \deduped presence fromServer t ->
1392− do e <- deduped; where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t); yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp}
1467+ do e <- deduped; where e.server == fromServer || any presence (\p -> samePubkeyServer p e && p.expiresAt > t); yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp}
13931468
13941469-- Keys whose presence was actually updated (new or extended expiry)
13951470appliedOnlineKeys = \entries presence ->
@@ -1419,71 +1494,77 @@ remoteOnlyTargets = \targets localOnline ->
14191494
14201495-- POST /auth/challenge: Generate a random challenge for Ed25519 authentication.
14211496-- Rate-limited per key; globally capped. Challenge expires in 1 minute.
1422−handleChallenge = \pubkey respond -> do
1497+handleChallenge = \pubkey clientIp respond -> do
14231498 t <- now
1424− globalRateLimited <- checkGlobalRate t
1425− if globalRateLimited then yield (respond {challenge: "", ok: False {}, expiresAt: 0})
1426− else do
1427− challenge <- genToken
1428− result <- atomic do
1429− pr <- maybePrune t
1430− challenges <- *challenges
1431− let active = do c <- challenges; where c.expiresAt > t; yield c
1432− let activeForKey = count (do c <- active; where c.pubkey == pubkey; yield c)
1433− let cappedActive = if activeForKey >= maxChallengesPerKey
1434− then let forKey = do c <- active; where c.pubkey == pubkey; yield c in
1435− let others = do c <- active; where c.pubkey != pubkey; yield c in
1436− union others (evictOneOldest forKey (\c -> c.expiresAt))
1437− else active
1438− if count cappedActive >= maxChallenges
1439− then yield {pr: pr, accepted: False {}, rateLimited: False {}}
1440− else do
1441− challengeRates <- *challengeRates
1442− let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
1443− (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
1444− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1445− maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
1446− if rateResult.limited
1447− then yield {pr: pr, accepted: False {}, rateLimited: True {}}
1448− else do
1449− set *challengeRates = rateResult.entries
1450− set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
1451− yield {pr: pr, accepted: True {}, rateLimited: False {}}
1452− when result.rateLimited
1453− (println ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
1454− when (not result.accepted && not result.rateLimited)
1455− (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
1456− forkBroadcastOffline t result.pr
1457− yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
1499+ preAuth <- checkPreAuth clientIp t
1500+ if preAuth != ""
1501+ then do
1502+ when (preAuth == "ip_rate_limited") (println ("challenge: IP rate limited for " ++ clientIp))
1503+ yield (respond {challenge: "", ok: False {}, expiresAt: 0})
1504+ else do
1505+ challenge <- genToken
1506+ result <- atomic do
1507+ pr <- maybePrune t
1508+ challenges <- *challenges
1509+ let active = do c <- challenges; where c.expiresAt > t; yield c
1510+ let activeForKey = count (do c <- active; where c.pubkey == pubkey; yield c)
1511+ let cappedActive = if activeForKey >= maxChallengesPerKey
1512+ then let forKey = do c <- active; where c.pubkey == pubkey; yield c in
1513+ let others = do c <- active; where c.pubkey != pubkey; yield c in
1514+ union others (evictOneOldest forKey (\c -> c.expiresAt))
1515+ else active
1516+ if count cappedActive >= maxChallenges
1517+ then yield {pr: pr, accepted: False {}, rateLimited: False {}}
1518+ else do
1519+ challengeRates <- *challengeRates
1520+ let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
1521+ (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
1522+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1523+ maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
1524+ if rateResult.limited
1525+ then yield {pr: pr, accepted: False {}, rateLimited: True {}}
1526+ else do
1527+ set *challengeRates = rateResult.entries
1528+ set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
1529+ yield {pr: pr, accepted: True {}, rateLimited: False {}}
1530+ when result.rateLimited
1531+ (println ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
1532+ when (not result.accepted && not result.rateLimited)
1533+ (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
1534+ forkBroadcastOffline t result.pr
1535+ yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
14581536
14591537-- POST /auth/verify: Verify Ed25519 signature over challenge string.
14601538-- On success: creates a session, registers local presence, and gossips online status.
14611539-- Rate-limited globally and per-key (shares auth rate window with challenge requests).
1462−handleVerifyAuth = \pubkey challenge signature respond -> do
1540+handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond -> do
14631541 t <- now
1464− globalRateLimited <- checkGlobalRate t
1465− if globalRateLimited then yield (respond {token: "", expiresAt: 0})
1466− else do
1467− authRateLimited <- atomic do
1468− challengeRates <- *challengeRates
1469− let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
1470− (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
1471− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1472− maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
1473− set *challengeRates = rateResult.entries
1474− yield rateResult.limited
1475− if authRateLimited then do
1476− println ("verify: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey)
1542+ preAuth <- checkPreAuth clientIp t
1543+ if preAuth != ""
1544+ then do
1545+ when (preAuth == "ip_rate_limited") (println ("verify: IP rate limited for " ++ clientIp))
14771546 yield (respond {token: "", expiresAt: 0})
1478− else verifyAuthInner pubkey challenge signature respond t
1479−
1480−verifyAuthInner = \pubkey challenge signature respond t -> do
1547+ else do
1548+ authRateLimited <- atomic do
1549+ challengeRates <- *challengeRates
1550+ let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
1551+ (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
1552+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1553+ maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
1554+ set *challengeRates = rateResult.entries
1555+ yield rateResult.limited
1556+ if authRateLimited then do
1557+ println ("verify: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey)
1558+ yield (respond {token: "", expiresAt: 0})
1559+ else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
1560+
1561+verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do
14811562 let expiry = t + sessionExpiry
14821563 let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
14831564 validation <- atomic do
14841565 pr <- maybePrune t
14851566 challenges <- *challenges
1486− let hasMatch = challengeMatch challenges pubkey challenge t
1567+ let hasMatch = challengeMatch challenges pubkey challenge clientIp t
14871568 -- Consume challenge on any verify attempt (prevents signature brute-force)
14881569 when hasMatch (set *challenges = challengesExcluding challenges pubkey challenge)
14891570 if not hasMatch || not sigValid
@@ -1495,7 +1576,8 @@ verifyAuthInner = \pubkey challenge signature respond t -> do
14951576 peerServers <- atomic do
14961577 sessions <- *sessions
14971578 presence <- *presence
1498− let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
1579+ let afterRevoke = if revokeOthers then do s <- sessions; where s.pubkey != pubkey; yield s else sessions
1580+ let capped = capPerKeyAndGlobal afterRevoke (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
14991581 set *sessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
15001582 let presenceWithout = presenceWithoutLocalKey presence pubkey
15011583 let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
@@ -1545,13 +1627,13 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
15451627 result <- atomic do
15461628 pr <- maybePrune t
15471629 withSessionAuth authorization t
1548− (yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}})
1630+ (yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}, senderRateLimited: False {}})
15491631 (\senderKey -> do
15501632 messages <- *messages
15511633 ackedIds <- *ackedIds
15521634 deliveredIds <- *deliveredIds
15531635 if messageId != "" && (hasMessage messageId to messages || hasMessage messageId to ackedIds || hasMessage messageId to deliveredIds)
1554− then yield {pr: pr, send: SendDuplicate {}}
1636+ then yield {pr: pr, send: SendDuplicate {}, senderRateLimited: False {}}
15551637 else do
15561638 sendRates <- *sendRates
15571639 let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
@@ -1559,33 +1641,44 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
15591641 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
15601642 maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
15611643 if rateResult.limited
1562− then yield {pr: pr, send: SendFailed {status: MsRateLimited {}}}
1644+ then yield {pr: pr, send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}}
15631645 else do
15641646 set *sendRates = rateResult.entries
1565− if countMessagesFor messages to t >= maxMessagesPerRecipient
1566− then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
1647+ recvRates <- *recvRates
1648+ let recvResult = checkRateWindow recvRates (\r -> r.pubkey == to)
1649+ (\ts -> {pubkey: to, windowStart: ts, count: 1})
1650+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1651+ maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
1652+ if recvResult.limited
1653+ then yield {pr: pr, send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}}
15671654 else do
1568− presence <- *presence
1569− seqCounter <- *seqCounter
1570− let seqTs = nextSeqTs t seqCounter
1571− set *seqCounter = seqTs
1572− set *messages = union messages
1573− [{messageId: effectiveMsgId, toKey: to,
1574− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
1575− signalMessageForKey to seqTs
1576− let toPresence = presenceForKey presence to t
1577− let localOnline = hasLocalPresence toPresence
1578− let remoteServers = collectPeerServers toPresence (\p -> p.server)
1579− let delivery = if localOnline then DsDelivered {}
1580− else if federationEnabled && count remoteServers > 0 then DsFederated {}
1581− else DsQueued {}
1582− -- Create forward entries atomically with message storage (prevents race with ack)
1583− when (federationEnabled && count remoteServers > 0) (do
1584− forwards <- *forwards
1585− let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}
1586− let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
1587− set *forwards = union forwards cappedFwds)
1588− yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers}})
1655+ set *recvRates = recvResult.entries
1656+ if countMessagesFor messages to t >= maxMessagesPerRecipient
1657+ then yield {pr: pr, send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}}
1658+ else do
1659+ presence <- *presence
1660+ seqCounter <- *seqCounter
1661+ let seqTs = nextSeqTs t seqCounter
1662+ set *seqCounter = seqTs
1663+ set *messages = union messages
1664+ [{messageId: effectiveMsgId, toKey: to,
1665+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
1666+ signalMessageForKey to seqTs
1667+ let toPresence = presenceForKey presence to t
1668+ let localOnline = hasLocalPresence toPresence
1669+ let remoteServers = collectPeerServers toPresence (\p -> p.server)
1670+ let delivery = if localOnline then DsDelivered {}
1671+ else if federationEnabled && count remoteServers > 0 then DsFederated {}
1672+ else DsQueued {}
1673+ -- Create forward entries atomically with message storage (prevents race with ack)
1674+ when (federationEnabled && count remoteServers > 0) (do
1675+ forwards <- *forwards
1676+ let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs}
1677+ let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
1678+ set *forwards = union forwards cappedFwds)
1679+ yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}})
1680+ when result.senderRateLimited
1681+ (println ("sendMessage: rate limited (" ++ show maxSendRatePerKey ++ "/s per key)"))
15891682 case result.send of
15901683 SendDone {delivery, remoteServers} ->
15911684 fork (do
@@ -1620,10 +1713,19 @@ handleLookupPresence = \pubkey authorization respond -> do
16201713 pr <- maybePrune t
16211714 withSessionAuth authorization t
16221715 (yield {pr: pr, server: "", authorized: False {}})
1623− (\_ -> do
1624− presence <- *presence
1625− let foundServer = bestServerForKey (presenceForKey presence pubkey t)
1626− yield {pr: pr, server: foundServer, authorized: True {}})
1716+ (\senderKey -> do
1717+ lookupRates <- *lookupRates
1718+ let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == senderKey)
1719+ (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})
1720+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1721+ maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
1722+ set *lookupRates = rateResult.entries
1723+ if rateResult.limited
1724+ then yield {pr: pr, server: "", authorized: True {}}
1725+ else do
1726+ presence <- *presence
1727+ let foundServer = bestServerForKey (presenceForKey presence pubkey t)
1728+ yield {pr: pr, server: foundServer, authorized: True {}})
16271729 forkBroadcastOffline t result.pr
16281730 yield (respond {server: result.server, authorized: result.authorized})
16291731
@@ -1657,7 +1759,7 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do
16571759 broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal
16581760 when (federationEnabled && count newlyOnlineRemote > 0)
16591761 (forwardQueuedMessages newlyOnlineRemote t))
1660− yield (respond {ok: True {}}))
1762+ yield (respond {ok: True {}, error: ""}))
16611763
16621764-- POST /federation/forward: Receive a message forwarded from a peer server.
16631765-- Only accepts messages where fromServer == originServer (no cascade forwarding).
@@ -1689,9 +1791,9 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
16891791 ForwardRejected {} -> println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
16901792 _ -> yield {}
16911793 forkBroadcastOffline t result.pr
1692− yield (respond {ok: case result.forward of
1693− ForwardRejected {} -> False {}
1694− _ -> True {}}))
1794+ yield (case result.forward of
1795+ ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}
1796+ _ -> respond {ok: True {}, error: ""}))
16951797
16961798-- POST /federation/ack: Peer server confirms delivery of messages we originated.
16971799-- Removes corresponding forward entries. Also cleans up the original message if the
@@ -1716,7 +1818,7 @@ handleFederationAck = \messageIds originServer fromServer authorization respond
17161818 set *messages = do m <- messages; where none removable (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m
17171819 yield pr
17181820 forkBroadcastOffline t pruneResult
1719− yield (respond {ok: True {}}))
1821+ yield (respond {ok: True {}, error: ""}))
17201822
17211823-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
17221824-- Call before stopping the server so peers learn keys went offline immediately.
@@ -1727,16 +1829,25 @@ handleDrain = \authorization respond -> do
17271829 if not authorized then yield (respond {ok: False {}, drained: 0})
17281830 else do
17291831 drainResult <- atomic do
1832+ set *draining = True {}
17301833 presence <- *presence
17311834 let localEntries = do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p
17321835 set *presence = do p <- presence; where not (isLocalServer p.server) || p.expiresAt <= t; yield p
17331836 set *pollWakeup = t
17341837 yield {localEntries: localEntries, peerServers: collectPeerServers presence (\p -> p.server)}
17351838 let drainedCount = count drainResult.localEntries
1736− when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
1839+ forwards <- *forwards
1840+ ackedIds <- *ackedIds
1841+ let pendingForwardCount = count forwards
1842+ let pendingAckCount = count ackedIds
1843+ fork (when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
17371844 let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
1738− sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
1845+ sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops))
17391846 println ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
1847+ when (pendingForwardCount > 0)
1848+ (println ("drain: warning: " ++ show pendingForwardCount ++ " pending forwards will not be retried"))
1849+ when (pendingAckCount > 0)
1850+ (println ("drain: warning: " ++ show pendingAckCount ++ " pending acks will not be sent"))
17401851 yield (respond {ok: True {}, drained: drainedCount})
17411852
17421853-- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).
@@ -1747,7 +1858,18 @@ handlePrune = \authorization respond -> do
17471858 when authorized (fork (pruneAndRetry t))
17481859 yield (respond {ok: authorized})
17491860
1750−handleHealth = \respond -> yield (respond {ok: True {}})
1861+handleHealth = \respond -> do
1862+ t <- now
1863+ result <- atomic do
1864+ lastPrune <- *lastPrunedAt
1865+ lastHb <- *lastHeartbeatAt
1866+ sa <- *startedAt
1867+ let warmup = t - sa < healthPruneMaxAge
1868+ yield {pruneOk: warmup || t - lastPrune < healthPruneMaxAge,
1869+ heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge}
1870+ yield (respond {ok: result.pruneOk && result.heartbeatOk,
1871+ pruneHealthy: result.pruneOk,
1872+ heartbeatHealthy: result.heartbeatOk})
17511873
17521874handleStats = \authorization respond -> do
17531875 let authorized = verifyMaintenanceAuth authorization
@@ -1768,18 +1890,32 @@ handleStats = \authorization respond -> do
17681890 sr <- *sendRates
17691891 cr <- *challengeRates
17701892 fr <- *federationRates
1893+ grc <- *globalRateCount
1894+ ir <- *ipRates
1895+ lr <- *lookupRates
1896+ rr <- *recvRates
17711897 pc <- *pruneCount
17721898 hc <- *heartbeatCount
17731899 sa <- *startedAt
17741900 lp <- *lastPrunedAt
17751901 lh <- *lastHeartbeatAt
1776− yield {sessions: count s, messages: count m, presence: count p, forwards: count f, challenges: count c, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff, activePolls: count ap, sendRates: count sr, challengeRates: count cr, federationRates: count fr, pruneCount: pc, heartbeatCount: hc, uptime: t - sa, lastPruneAt: lp, lastHeartbeatAt: lh}
1777− yield (respond {sessions: result.sessions, messages: result.messages, presence: result.presence, forwards: result.forwards, challenges: result.challenges, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, ok: True {}})
1778− else yield (respond {sessions: 0, messages: 0, presence: 0, forwards: 0, challenges: 0, ackedIds: 0, ackOverflow: 0, deliveredIds: 0, forwardFailures: 0, activePolls: 0, sendRates: 0, challengeRates: 0, federationRates: 0, pruneCount: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, ok: False {}})
1902+ yield {sessions: count s, messages: count m, presence: count p, forwards: count f, challenges: count c, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff, activePolls: count ap, sendRates: count sr, challengeRates: count cr, federationRates: count fr, globalRateCount: grc, ipRates: count ir, lookupRates: count lr, recvRates: count rr, pruneCount: pc, heartbeatCount: hc, uptime: t - sa, lastPruneAt: lp, lastHeartbeatAt: lh}
1903+ yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, ok: True {}})
1904+ else yield (respond {sessions: 0, sessionsMax: 0, messages: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pruneCount: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, ok: False {}})
1905+
1906+handleStatsPeers = \authorization respond -> do
1907+ let authorized = verifyMaintenanceAuth authorization
1908+ if authorized
1909+ then do
1910+ t <- now
1911+ failures <- *forwardFailures
1912+ let peers = do f <- failures; yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoff f.failures, retriable: shouldRetryServer f.server t failures}
1913+ yield (respond {peers: peers, ok: True {}})
1914+ else yield (respond {peers: [], ok: False {}})
17791915
17801916serve = \req -> case req of
1781− ReqChallenge {pubkey, respond} -> handleChallenge pubkey respond
1782− VerifyAuth {pubkey, challenge, signature, respond} -> handleVerifyAuth pubkey challenge signature respond
1917+ ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
1918+ VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
17831919 Poll {cursor, authorization, respond} -> handlePoll cursor authorization respond
17841920 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization respond
17851921 LookupPresence {pubkey, authorization, respond} -> handleLookupPresence pubkey authorization respond
@@ -1790,6 +1926,7 @@ serve = \req -> case req of
17901926 Drain {authorization, respond} -> handleDrain authorization respond
17911927 Health {respond} -> handleHealth respond
17921928 Stats {authorization, respond} -> handleStats authorization respond
1929+ StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
17931930
17941931-- ============================================
17951932-- Entry Point
@@ -1800,11 +1937,24 @@ serve = \req -> case req of
18001937backgroundPrune = do
18011938 jitter <- randomInt 5000
18021939 sleep (backgroundPruneInterval + jitter)
1940+ draining <- *draining
1941+ if draining then println "backgroundPrune: stopping (server draining)"
1942+ else do
18031943 t <- now
1804− atomic do
1805− pc <- *pruneCount
1806− set *pruneCount = pc + 1
1807− fork (pruneAndRetry t)
1944+ canRun <- atomic do
1945+ running <- *pruneRunning
1946+ if running > 0 && t - running < backgroundPruneInterval * 5
1947+ then yield (False {})
1948+ else do
1949+ set *pruneRunning = t
1950+ pc <- *pruneCount
1951+ set *pruneCount = pc + 1
1952+ yield (True {})
1953+ if canRun
1954+ then fork (do
1955+ pruneAndRetry t
1956+ atomic do set *pruneRunning = 0)
1957+ else println "backgroundPrune: skipping (previous cycle still running)"
18081958 backgroundPrune
18091959
18101960-- Background loop: periodically bumps pollWakeup to unblock long-polls that have been
@@ -1812,13 +1962,19 @@ backgroundPrune = do
18121962pollHeartbeat = do
18131963 jitter <- randomInt 2000
18141964 sleep (pollTimeoutMs + jitter)
1815− t <- now
1816− atomic do
1817− set *pollWakeup = t
1818− set *lastHeartbeatAt = t
1819− hc <- *heartbeatCount
1820− set *heartbeatCount = hc + 1
1821− pollHeartbeat
1965+ draining <- *draining
1966+ if draining then do
1967+ t <- now
1968+ atomic do set *pollWakeup = t
1969+ println "pollHeartbeat: stopping (server draining)"
1970+ else do
1971+ t <- now
1972+ atomic do
1973+ set *pollWakeup = t
1974+ set *lastHeartbeatAt = t
1975+ hc <- *heartbeatCount
1976+ set *heartbeatCount = hc + 1
1977+ pollHeartbeat
18221978
18231979main = do
18241980 t <- now