@@ -66,6 +66,7 @@ clockSkewTolerance : Int<Ms>
6666clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers
6767maxForwardsEntries = 10000
6868maxForwardsPerKey = 50 -- max pending forwards per recipient pubkey (prevents one key from starving others)
69+maxForwardsPerServer = 2000 -- max pending forwards per destination server (prevents one unreachable peer from starving others)
6970maxServerNameLen = 128
7071maxMessageIdLen = 128
7172maxCorrelationIdLen = 128
@@ -184,7 +185,7 @@ federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after thi
184185
185186-- Server build version: included in /health and /stats responses for deployment verification.
186187-- Update this string when deploying a new build so operators can confirm which version is running.
187−serverBuildVersion = "2026-04-26c"
188+serverBuildVersion = "2026-04-26d"
188189
189190-- Federation protocol version: included in gossip/forward/ack requests.
190191-- Peers log warnings when they receive requests from older versions.
@@ -299,8 +300,12 @@ maxCancelRetries = 10
299300-- Verify callback caching: cache successful verifications per-server to reduce
300301-- outbound HTTP calls. A peer sending N requests/second forces N verify callbacks;
301302-- caching the success for a short window amortizes the cost.
302−verificationCacheMs : Int<Ms>
303−verificationCacheMs = 2000 -- 2 seconds; short enough to limit replay window
303+-- Adaptive: scales from verificationCacheMinMs to verificationCacheMaxMs based on
304+-- recent request volume from the peer (higher volume = longer cache = fewer callbacks).
305+verificationCacheMinMs : Int<Ms>
306+verificationCacheMinMs = 2000 -- 2 seconds; base cache for low-traffic peers
307+verificationCacheMaxMs : Int<Ms>
308+verificationCacheMaxMs = 10000 -- 10 seconds; max cache for high-traffic peers
304309maxVerificationCache = 100
305310-- Health endpoint rate limiting (global, since /health has no clientIp)
306311maxHealthRate = 100 -- max health checks per window
@@ -313,11 +318,14 @@ healthHeartbeatMaxAge : Int<Ms>
313318healthHeartbeatMaxAge = pollTimeoutMs * 3
314319healthWatchdogMaxAge : Int<Ms>
315320healthWatchdogMaxAge = backgroundPruneInterval * 15 -- watchdog runs every ~5 prune intervals
316−-- Forward queue age health threshold: flag unhealthy when the oldest pending forward
317−-- is older than this. Indicates a destination server is unreachable and forward entries
318−-- are accumulating without being delivered or retried successfully.
321+-- Forward queue age health threshold: flag unhealthy when multiple destination servers
322+-- have pending forwards older than this. A single stuck peer shouldn't mark the whole
323+-- server unhealthy — only widespread queue buildup indicates a systemic problem.
319324healthForwardQueueMaxAge : Int<Ms>
320325healthForwardQueueMaxAge = retryBackoffMax * 3 -- ~24 minutes; 3x max backoff gives retries time to succeed
326+-- Minimum number of destination servers with stale queues before flagging unhealthy.
327+-- 1 = old behavior (any stale server triggers unhealthy). 2+ = tolerate isolated peer issues.
328+healthForwardQueueMinStaleServers = 2
321329
322330-- ============================================
323331-- Types
@@ -409,21 +417,23 @@ type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp
409417type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
410418-- Tracks consecutive failures per federation peer for exponential backoff
411419type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFailedAt: Timestamp}
412−-- Sliding-window rate limit state per pubkey
413−type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
414−type RecvRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
415−type PairRateEntry = {sender: PubkeyHex, recipient: PubkeyHex, windowStart: Timestamp, count: Int}
416−type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
420+-- Sliding-window rate limit state per pubkey.
421+-- prevCount: count from the previous window, used for sliding-window interpolation.
422+-- Effective count = prevCount * (1 - elapsed/windowMs) + count, preventing 2x burst at window boundaries.
423+type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}
424+type RecvRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}
425+type PairRateEntry = {sender: PubkeyHex, recipient: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}
426+type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}
417427-- Sliding-window rate limit state per federation peer server
418−type FederationRateEntry = {server: Text, windowStart: Timestamp, count: Int}
419−type FederationPairRateEntry = {server: Text, recipient: PubkeyHex, windowStart: Timestamp, count: Int}
428+type FederationRateEntry = {server: Text, windowStart: Timestamp, count: Int, prevCount: Int}
429+type FederationPairRateEntry = {server: Text, recipient: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}
420430-- Tracks active long-poll connections to enforce per-key concurrency limits
421431type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
422432data AuthResult = Authorized {pubkey: PubkeyHex, ip: Text} | Unauthorized {}
423433data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
424434
425−type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int}
426−type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
435+type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int, prevCount: Int}
436+type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}
427437-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles
428438type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}
429439type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool, pendingForwards: Int, avgLatencyMs: Int, maxLatencyMs: Int, latencySamples: Int}
@@ -432,7 +442,7 @@ type RecipientCount = {pubkey: PubkeyHex, count: Int}
432442type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
433443-- Tracks recent federation request (server, timestamp) pairs for replay detection
434444type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
435−type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
445+type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int, prevCount: Int}
436446-- Tracks unique pubkeys authenticated per IP (key-rotation abuse detection)
437447type UniqueKeyIpEntry = {ip: Text, keys: [PubkeyHex], windowStart: Timestamp}
438448
@@ -472,8 +482,12 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
472482-- Hot-path auth scans only one bucket (~1/16 of sessions).
473483-- TODO: If Knot adds indexed arrays or maps, replace the 16 manual bucket
474484-- refs and their if-else dispatch chains with a single indexed collection.
475−-- This would eliminate ~1000 lines of mechanical boilerplate across sessions,
476−-- messages, presence, signal, and recipientCount buckets.
485+-- This would eliminate ~1500 lines of mechanical boilerplate:
486+-- - 10 sharded collections (sessions, messages, presence, forwards, acked,
487+-- delivered, nonces, recipientCounts, messageSignals, fwdBuckets)
488+-- - Each has: 16 mutable refs, read/write/readAll/writeAll/countAll/transform functions
489+-- - writeAll* functions share an identical 16-way partition fold pattern
490+-- - The partition fold and 16-write pattern could become: partitionByBucket items getKey writeBucket
477491*sessionBucket0 : [Session]
478492*sessionBucket1 : [Session]
479493*sessionBucket2 : [Session]
@@ -656,12 +670,14 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
656670*federationPairRates : [FederationPairRateEntry]
657671
658672*globalRateCount : Int
673+*globalRatePrevCount : Int
659674
660675*globalRateWindowStart : Timestamp
661676
662677-- Health endpoint rate limiting state (separate from global rate to avoid starving real requests)
663678*healthRateCount : Int
664679*healthRateWindowStart : Timestamp
680+*healthRatePrevCount : Int
665681
666682-- Background task cycle counters (exposed via /stats for health monitoring)
667683*pruneCount : Int
@@ -759,6 +775,7 @@ maxPeerLatencyEntries = 100
759775
760776-- Auth-tier global rate limiting state
761777*authGlobalRateCount : Int
778+*authGlobalRatePrevCount : Int
762779*authGlobalRateWindowStart : Timestamp
763780
764781-- Max concurrent background prune forks (prevents resource exhaustion from stuck recovery)
@@ -1158,6 +1175,12 @@ route StatsApi where
11581175 GET /stats/forwards headers {authorization: Text}
11591176 -> {servers: [{server: Text, count: Int, oldestAge: Int, totalRetries: Int}], ok: Bool, total: Int}
11601177 = StatsForwards
1178+ -- Effective rate limits after adaptive scaling: shows the current scale factor and
1179+ -- each rate limit as it's actually applied (compile-time default * runtime override * adaptive scale).
1180+ -- Helps operators understand why requests are being rejected under load.
1181+ GET /stats/rates headers {authorization: Text}
1182+ -> {ok: Bool, scalePct: Int, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}
1183+ = StatsRates
11611184
11621185-- Readiness probe: returns false during startup warmup (before first prune completes),
11631186-- true once the server is fully operational. Unlike /health, this is unauthenticated
@@ -1630,6 +1653,19 @@ generateFederationAuth = \scope server -> do
16301653-- No retry: a failed verify rejects the request. The peer's federation rate limit + circuit breaker
16311654-- handle unreliable peers; retrying here doubles verify traffic and blocks with a 1s sleep per op,
16321655-- which a malicious peer can exploit for amplification.
1656+-- Adaptive verify cache TTL: scales with recent request volume from a peer.
1657+-- Peers sending many requests/second benefit from longer cache (fewer callbacks).
1658+-- Uses peerLatencies.count as a proxy for request volume (tracks verify calls per window).
1659+-- Returns TTL in ms between verificationCacheMinMs and verificationCacheMaxMs.
1660+adaptiveVerifyCacheTtl : Text -> [{server: Text, totalMs: Int, count: Int, maxMs: Int, windowStart: Timestamp}] -> Int<Ms>
1661+adaptiveVerifyCacheTtl = \server peerLats ->
1662+ case headOf (do p <- peerLats; where p.server == server; yield p) of
1663+ Nothing {} -> verificationCacheMinMs
1664+ Just {value: p} ->
1665+ -- Scale: 0-5 calls = min, 50+ calls = max, linear interpolation between
1666+ let scale = clamp 0 100 (p.count * 100 / 50)
1667+ in verificationCacheMinMs + (verificationCacheMaxMs - verificationCacheMinMs) * scale / 100
1668+
16331669verifyFederationCallback : Text -> Text -> Text -> IO {network} Bool
16341670verifyFederationCallback = \fromServer authorization scope ->
16351671 case extractToken authorization of
@@ -1639,7 +1675,9 @@ verifyFederationCallback = \fromServer authorization scope ->
16391675 t <- now
16401676 cached <- atomic do
16411677 cache <- *verificationCache
1642− yield (any cache (\c -> c.server == fromServer && c.scope == scope && t - c.verifiedAt < verificationCacheMs))
1678+ peerLats <- *peerLatencies
1679+ let cacheTtl = adaptiveVerifyCacheTtl fromServer peerLats
1680+ yield (any cache (\c -> c.server == fromServer && c.scope == scope && t - c.verifiedAt < cacheTtl))
16431681 if cached then yield (True {})
16441682 else do
16451683 startTime <- now
@@ -1661,7 +1699,9 @@ verifyFederationCallback = \fromServer authorization scope ->
16611699 Ok r -> do
16621700 when r.ok (atomic do
16631701 cache <- *verificationCache
1664− let cleaned = do c <- cache; where t - c.verifiedAt < verificationCacheMs; yield c
1702+ peerLats <- *peerLatencies
1703+ let cacheTtl = adaptiveVerifyCacheTtl fromServer peerLats
1704+ let cleaned = do c <- cache; where t - c.verifiedAt < cacheTtl; yield c
16651705 let updated = union cleaned [{server: fromServer, scope: scope, verifiedAt: t}]
16661706 set *verificationCache = if count updated > maxVerificationCache
16671707 then evictOneOldest updated (\c -> c.verifiedAt)
@@ -2288,17 +2328,29 @@ pruneMessageBucketsConcurrent = \b t ->
22882328-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one
22892329-- bucket, so per-bucket recipient counts are disjoint — each bucket's counts are written
22902330-- to the corresponding rcBucket without overlap.
2291−-- NOTE: TOCTOU window — messages can arrive between per-bucket reads, so the final
2292−-- totalMessageCount write may slightly overwrite concurrent updates from sendMessageCore.
2293−-- This is acceptable: (a) reconciliation runs on drift detection or every 10th prune cycle,
2294−-- (b) the next prune re-reconciles, (c) capacity checks on the hot path use per-bucket
2295−-- rcBucket counts (not totalMessageCount) so message acceptance isn't affected.
2331+-- Delta-based counter adjustment: snapshots totalMessageCount before scanning, computes the
2332+-- drift (actual - snapshot), then applies the drift atomically to the live counter. This is
2333+-- commutative with concurrent sendMessageCore: if sendMessageCore increments by N between our
2334+-- snapshot and final adjustment, the live counter becomes (snapshot + N + drift) = (actual + N),
2335+-- which is correct. The old approach overwrote with the absolute recount, losing concurrent N.
2336+-- Delta-based reconciliation: snapshots totalMessageCount before scanning, then applies
2337+-- the computed drift atomically. This preserves concurrent sendMessageCore increments:
2338+-- if sendMessageCore adds N messages between snapshot and final write, the live counter
2339+-- becomes (snapshot + N) + (actual - snapshot) = actual + N, which is correct.
2340+-- Old approach: set totalMessageCount = actual (overwrites concurrent N).
2341+reconcileMessageCountsDelta : IO Int
2342+reconcileMessageCountsDelta = do
2343+ snapshot <- *totalMessageCount
2344+ actual <- reconcileMessageCounts 0 0
2345+ atomic do
2346+ stored <- *totalMessageCount
2347+ let drift = actual - snapshot
2348+ set *totalMessageCount = max 0 (stored + drift)
2349+ yield drift
2350+
22962351reconcileMessageCounts = \b totalAcc ->
22972352 if b > 15 then do
2298− atomic do
2299− stored <- *totalMessageCount
2300− set *totalMessageCount = totalAcc
2301− yield (totalAcc - stored)
2353+ yield totalAcc
23022354 else do
23032355 bucketTotal <- atomic do
23042356 bucket <- readMessageBucket b
@@ -2694,6 +2746,23 @@ oldestForwardAgeAcrossBuckets = \b t acc ->
26942746 let bucketOldest = case minBy bucket (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
26952747 oldestForwardAgeAcrossBuckets (b + 1) t (max acc bucketOldest)
26962748
2749+-- Count distinct destination servers with stale forward queues (oldest entry > healthForwardQueueMaxAge).
2750+-- Scans all 16 buckets, tracking unique stale servers across buckets.
2751+-- O(total_forwards) scan with O(servers) memory. Used by health check for per-server awareness.
2752+countStaleForwardServers : Int -> Timestamp -> [Text] -> IO Int
2753+countStaleForwardServers = \b t seen ->
2754+ if b > 15
2755+ then yield (count seen)
2756+ else do
2757+ bucket <- readFwdBucket b
2758+ -- Find servers in this bucket with entries older than the threshold
2759+ let newStale = fold (\sacc f ->
2760+ let age = t - f.createdAt in
2761+ if age > healthForwardQueueMaxAge && none sacc (\s -> s == f.toServer) && none seen (\s -> s == f.toServer)
2762+ then union [f.toServer] sacc
2763+ else sacc) [] bucket
2764+ countStaleForwardServers (b + 1) t (union seen newStale)
2765+
26972766-- Remove forward entries matching a server and message IDs, scanning per-bucket.
26982767-- Only rewrites buckets that actually changed (avoids writeAllForwards repartitioning).
26992768-- Returns all removed forward entries across all buckets.
@@ -3062,20 +3131,28 @@ matchIp = \clientIp sessionIp ->
30623131 else if clientIp == "" then False {} -- session has IP but request doesn't — proxy disappeared
30633132 else safeCompareLoop clientIp sessionIp maxIpLen 0 0 == 0
30643133
3065−-- Generic sliding-window rate limiter. Returns whether the request is rate-limited,
3066−-- the updated entries list, and retryAfterMs (time until the current window resets, 0 if not limited).
3067−-- Callers provide match/create/update functions for their entry type.
3134+-- Generic sliding-window rate limiter using the sliding window counter algorithm.
3135+-- Instead of a fixed window that resets abruptly (allowing 2x burst at boundaries),
3136+-- interpolates between the current and previous window's counts:
3137+-- effectiveCount = prevCount * (1 - elapsed/windowMs) + count
3138+-- This smoothly transitions between windows, preventing boundary bursts.
3139+-- Callers provide match/create/update/rotate functions for their entry type.
30683140-- Uses a direct fold for entry lookup (avoids materializing a filtered intermediate list).
3069−checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntries t ->
3141+checkRateWindow = \entries matchKey makeNew updateCount rotateWindow getCount getPrevCount maxRate windowMs maxEntries t ->
30703142 let entry = fold (\acc r -> case acc of Just _ -> acc; Nothing {} -> if matchKey r then Just {value: r} else acc) Nothing {} entries in
30713143 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= windowMs in
3072− let rateCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count in
3073− if rateCount >= maxRate
3074− then let retryMs = case entry of Nothing {} -> windowMs; Just {value: r} -> max 0 (windowMs - (t - r.windowStart))
3144+ -- Sliding window interpolation: weight previous window's count by remaining fraction
3145+ let elapsed = case entry of Nothing {} -> 0; Just {value: r} -> t - r.windowStart in
3146+ let prevWeight = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else getPrevCount r * (windowMs - elapsed) / windowMs in
3147+ let currentCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else getCount r in
3148+ let effectiveCount = prevWeight + currentCount in
3149+ if effectiveCount >= maxRate
3150+ then let retryMs = case entry of Nothing {} -> windowMs; Just {value: r} -> max 0 (windowMs - elapsed)
30753151 in {limited: True {}, entries: entries, retryAfterMs: retryMs}
30763152 else
30773153 let newEntries = if isNewWindow
3078− then union (do r <- entries; where not (matchKey r); yield r) [makeNew t]
3154+ then let prevCount = case entry of Nothing {} -> 0; Just {value: r} -> getCount r
3155+ in union (do r <- entries; where not (matchKey r); yield r) [rotateWindow (makeNew t) prevCount]
30793156 else do r <- entries; yield (if matchKey r then updateCount r else r)
30803157 in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs; yield r) else newEntries, retryAfterMs: 0}
30813158
@@ -3083,26 +3160,34 @@ checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntr
30833160-- Each returns {limited, entries, retryAfterMs} — caller reads/writes the mutable ref.
30843161checkPubkeyRate = \rates pubkey maxRate windowMs maxEntries t ->
30853162 checkRateWindow rates (\r -> r.pubkey == pubkey)
3086− (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
3087− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3163+ (\ts -> {pubkey: pubkey, windowStart: ts, count: 1, prevCount: 0})
3164+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1, prevCount: r.prevCount})
3165+ (\r pc -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count, prevCount: pc})
3166+ (\r -> r.count) (\r -> r.prevCount)
30883167 maxRate windowMs maxEntries t
30893168
30903169checkIpRate = \rates ip maxRate windowMs maxEntries t ->
30913170 checkRateWindow rates (\r -> r.ip == ip)
3092− (\ts -> {ip: ip, windowStart: ts, count: 1})
3093− (\r -> {ip: r.ip, windowStart: r.windowStart, count: r.count + 1})
3171+ (\ts -> {ip: ip, windowStart: ts, count: 1, prevCount: 0})
3172+ (\r -> {ip: r.ip, windowStart: r.windowStart, count: r.count + 1, prevCount: r.prevCount})
3173+ (\r pc -> {ip: r.ip, windowStart: r.windowStart, count: r.count, prevCount: pc})
3174+ (\r -> r.count) (\r -> r.prevCount)
30943175 maxRate windowMs maxEntries t
30953176
30963177checkServerRate = \rates server maxRate windowMs maxEntries t ->
30973178 checkRateWindow rates (\r -> r.server == server)
3098− (\ts -> {server: server, windowStart: ts, count: 1})
3099− (\r -> {server: r.server, windowStart: r.windowStart, count: r.count + 1})
3179+ (\ts -> {server: server, windowStart: ts, count: 1, prevCount: 0})
3180+ (\r -> {server: r.server, windowStart: r.windowStart, count: r.count + 1, prevCount: r.prevCount})
3181+ (\r pc -> {server: r.server, windowStart: r.windowStart, count: r.count, prevCount: pc})
3182+ (\r -> r.count) (\r -> r.prevCount)
31003183 maxRate windowMs maxEntries t
31013184
31023185checkPairRate = \rates sender recipient maxRate windowMs maxEntries t ->
31033186 checkRateWindow rates (\r -> r.sender == sender && r.recipient == recipient)
3104− (\ts -> {sender: sender, recipient: recipient, windowStart: ts, count: 1})
3105− (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
3187+ (\ts -> {sender: sender, recipient: recipient, windowStart: ts, count: 1, prevCount: 0})
3188+ (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1, prevCount: r.prevCount})
3189+ (\r pc -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count, prevCount: pc})
3190+ (\r -> r.count) (\r -> r.prevCount)
31063191 maxRate windowMs maxEntries t
31073192
31083193-- Per-IP unique key rate check: tracks distinct pubkeys per IP per window.
@@ -3126,8 +3211,10 @@ checkUniqueKeyIpRate = \entries ip pubkey t ->
31263211
31273212checkServerRecipientRate = \rates server recipient maxRate windowMs maxEntries t ->
31283213 checkRateWindow rates (\r -> r.server == server && r.recipient == recipient)
3129− (\ts -> {server: server, recipient: recipient, windowStart: ts, count: 1})
3130− (\r -> {server: r.server, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
3214+ (\ts -> {server: server, recipient: recipient, windowStart: ts, count: 1, prevCount: 0})
3215+ (\r -> {server: r.server, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1, prevCount: r.prevCount})
3216+ (\r pc -> {server: r.server, recipient: r.recipient, windowStart: r.windowStart, count: r.count, prevCount: pc})
3217+ (\r -> r.count) (\r -> r.prevCount)
31313218 maxRate windowMs maxEntries t
31323219
31333220-- Extract the item with the smallest key and return both the item and remaining list
@@ -3675,54 +3762,70 @@ checkFederationRate = \server t -> atomic do
36753762checkGlobalRate = \t -> atomic do
36763763 windowStart <- *globalRateWindowStart
36773764 rateCount <- *globalRateCount
3765+ prevCount <- *globalRatePrevCount
36783766 effectiveRate <- effectiveGlobalRequestRate
3679− if t - windowStart >= globalRateWindowMs
3767+ let elapsed = t - windowStart
3768+ if elapsed >= globalRateWindowMs
36803769 then do
36813770 set *globalRateWindowStart = t
3771+ set *globalRatePrevCount = rateCount
36823772 set *globalRateCount = 1
36833773 yield (False {})
3684− else if rateCount >= effectiveRate
3685− then do
3686− rlg <- *rateLimitedGlobal
3687− set *rateLimitedGlobal = rlg + 1
3688− yield (True {})
3689− else do
3690− set *globalRateCount = rateCount + 1
3691− yield (False {})
3774+ else do
3775+ -- Sliding window: interpolate previous window's count
3776+ let prevWeight = prevCount * (globalRateWindowMs - elapsed) / globalRateWindowMs
3777+ if prevWeight + rateCount >= effectiveRate
3778+ then do
3779+ rlg <- *rateLimitedGlobal
3780+ set *rateLimitedGlobal = rlg + 1
3781+ yield (True {})
3782+ else do
3783+ set *globalRateCount = rateCount + 1
3784+ yield (False {})
36923785
36933786-- Auth-tier global rate limiting (shared budget for challenge + verify + revoke)
36943787checkAuthGlobalRate = \t -> atomic do
36953788 windowStart <- *authGlobalRateWindowStart
36963789 rateCount <- *authGlobalRateCount
3790+ prevCount <- *authGlobalRatePrevCount
36973791 authRate <- effectiveAuthGlobalRate
3698− if t - windowStart >= authGlobalRateWindowMs
3792+ let elapsed = t - windowStart
3793+ if elapsed >= authGlobalRateWindowMs
36993794 then do
37003795 set *authGlobalRateWindowStart = t
3796+ set *authGlobalRatePrevCount = rateCount
37013797 set *authGlobalRateCount = 1
37023798 yield (False {})
3703− else if rateCount >= authRate
3704− then do
3705− rla <- *rateLimitedAuth
3706− set *rateLimitedAuth = rla + 1
3707− yield (True {})
3708− else do
3709− set *authGlobalRateCount = rateCount + 1
3710− yield (False {})
3799+ else do
3800+ let prevWeight = prevCount * (authGlobalRateWindowMs - elapsed) / authGlobalRateWindowMs
3801+ if prevWeight + rateCount >= authRate
3802+ then do
3803+ rla <- *rateLimitedAuth
3804+ set *rateLimitedAuth = rla + 1
3805+ yield (True {})
3806+ else do
3807+ set *authGlobalRateCount = rateCount + 1
3808+ yield (False {})
37113809
37123810-- Health endpoint rate limiting (separate from global rate so monitoring doesn't starve real requests)
37133811checkHealthRate = \t -> atomic do
37143812 windowStart <- *healthRateWindowStart
37153813 rateCount <- *healthRateCount
3716− if t - windowStart >= healthRateWindowMs
3814+ prevCount <- *healthRatePrevCount
3815+ let elapsed = t - windowStart
3816+ if elapsed >= healthRateWindowMs
37173817 then do
37183818 set *healthRateWindowStart = t
3819+ set *healthRatePrevCount = rateCount
37193820 set *healthRateCount = 1
37203821 yield (False {})
3721− else if rateCount >= maxHealthRate
3722− then yield (True {})
3723− else do
3724− set *healthRateCount = rateCount + 1
3725− yield (False {})
3822+ else do
3823+ let prevWeight = prevCount * (healthRateWindowMs - elapsed) / healthRateWindowMs
3824+ if prevWeight + rateCount >= maxHealthRate
3825+ then yield (True {})
3826+ else do
3827+ set *healthRateCount = rateCount + 1
3828+ yield (False {})
37263829
37273830-- IP-based rate limiting; skips if clientIp header is empty (no reverse proxy)
37283831checkClientIpRate = \ip t ->
@@ -3991,8 +4094,16 @@ retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
39914094retryForwardsForServer = \entries remaining ->
39924095 if remaining <= 0 || count entries == 0 then yield {}
39934096 else do
3994− let server = case headOf entries of Just {value: f} -> f.toServer; Nothing {} -> ""
3995− forEachBatch entries maxBatchSize (\batch -> do
4097+ t <- now
4098+ -- Pre-filter expired forwards: avoids wasting bandwidth on messages the recipient
4099+ -- would discard. Expired entries are cleaned up by the next prune cycle.
4100+ let liveEntries = do f <- entries; where f.expiresAt > t; yield f
4101+ let expiredCount = count entries - count liveEntries
4102+ when (expiredCount > 0) (logInfo ("federation: skipped " ++ show expiredCount ++ " expired forwards during retry"))
4103+ if count liveEntries == 0 then yield {}
4104+ else do
4105+ let server = case headOf liveEntries of Just {value: f} -> f.toServer; Nothing {} -> ""
4106+ forEachBatch liveEntries maxBatchSize (\batch -> do
39964107 t <- now
39974108 auth <- generateFederationAuth "forward-batch" server
39984109 corrId <- genTokenN 1
@@ -4093,6 +4204,30 @@ collectEligibleForwards = \b eligibleServers ->
40934204-- Circuit breaker: skips entirely when too many peers are failing (likely local network issue).
40944205-- Per-bucket: reads forward buckets one at a time, collecting only entries for eligible servers,
40954206-- instead of materializing all forwards into memory via readAllForwards.
4207+-- Half-open circuit breaker probe: when the circuit is open, pick the peer with the
4208+-- oldest failure and send a lightweight gossip probe. If it succeeds, clear the failure
4209+-- record and re-evaluate the circuit. This allows faster recovery than waiting for
4210+-- maxServerDowntime to expire on each peer independently.
4211+circuitBreakerProbe : [ForwardFailure] -> Timestamp -> IO {network} ()
4212+circuitBreakerProbe = \failures t -> do
4213+ -- Pick the peer whose last failure is oldest (most likely to have recovered)
4214+ let activeFailures = do f <- failures; where t - f.firstFailedAt < maxServerDowntime; yield f
4215+ case minBy activeFailures (\f -> f.failedAt) of
4216+ Nothing {} -> yield {}
4217+ Just {value: candidate} -> do
4218+ logInfo ("federation: circuit breaker half-open probe to " ++ candidate.server)
4219+ probeStart <- now
4220+ auth <- generateFederationAuth "gossip" candidate.server
4221+ result <- fetchFederation candidate.server (RecvGossip {
4222+ authorization: auth,
4223+ events: [], hopsRemaining: 1, fromServer: serverHostN,
4224+ requestTimestamp: probeStart, correlationId: "", protocolVersion: federationProtocolVersion})
4225+ case result of
4226+ Err e -> logWarn ("federation: half-open probe to " ++ candidate.server ++ " failed: " ++ show e)
4227+ Ok _ -> do
4228+ logInfo ("federation: half-open probe to " ++ candidate.server ++ " succeeded, clearing failure")
4229+ clearFailure candidate.server
4230+
40964231retryPendingForwardsFromBuckets : IO {network} {}
40974232retryPendingForwardsFromBuckets = do
40984233 t <- now
@@ -4101,6 +4236,8 @@ retryPendingForwardsFromBuckets = do
41014236 then do
41024237 logWarn ("federation: circuit breaker open (" ++ show (count failures) ++ " peers failing), skipping forward retries")
41034238 logEvent "warn" "circuit_breaker_open" (show (count failures) ++ " peers failing")
4239+ -- Half-open: probe one peer to detect recovery
4240+ circuitBreakerProbe failures t
41044241 else do
41054242 inFlight <- *federationInFlight
41064243 blocked <- *blockedPeers
@@ -4113,12 +4250,27 @@ retryPendingForwardsFromBuckets = do
41134250 else do
41144251 -- Read only entries for eligible servers (per-bucket, streaming)
41154252 eligibleForwards <- collectEligibleForwards 0 eligibleServers
4253+ -- Ramp-up for recovering peers: limit forwards per server based on recent failure history.
4254+ -- A server with N prior failures gets min(maxBatchSize * 2^(clearCycles), all) forwards per cycle,
4255+ -- where clearCycles = time since last failure / prune interval. This prevents overwhelming
4256+ -- a just-recovered peer with thousands of queued forwards in a single burst.
4257+ let rampUpLimit = \server ->
4258+ case headOf (do f <- failures; where f.server == server; yield f) of
4259+ Nothing {} -> maxForwardsPerServer -- no failure record: full speed
4260+ Just {value: f} ->
4261+ let cyclesSinceFailure = (t - f.failedAt) / backgroundPruneInterval in
4262+ min maxForwardsPerServer (maxBatchSize * pow2 (min 10 cyclesSinceFailure))
4263+ -- Apply per-server ramp-up: cap entries per server
4264+ let rampedForwards = fold (\acc server ->
4265+ let serverFwds = do f <- eligibleForwards; where f.toServer == server; yield f
4266+ in let limit = rampUpLimit server
4267+ in union acc (takeRelation limit serverFwds)) [] eligibleServers
41164268 -- Prioritize forwards for recipients with active polls (currently online).
41174269 -- These keys will receive the message sooner, improving delivery latency.
41184270 -- Forwards for offline keys are retried after online-key forwards complete.
41194271 activePolls <- *activePolls
4120− let onlineForwards = do f <- eligibleForwards; where any activePolls (\p -> p.pubkey == f.toKey); yield f
4121− let offlineForwards = do f <- eligibleForwards; where none activePolls (\p -> p.pubkey == f.toKey); yield f
4272+ let onlineForwards = do f <- rampedForwards; where any activePolls (\p -> p.pubkey == f.toKey); yield f
4273+ let offlineForwards = do f <- rampedForwards; where none activePolls (\p -> p.pubkey == f.toKey); yield f
41224274 -- Sort by effective priority descending (higher priority first) within each group.
41234275 -- Penalty by retry count: messages with many retries are deprioritized so fresh
41244276 -- messages aren't blocked behind likely-failing ones. A fresh priority-5 (score -5)
@@ -4149,11 +4301,13 @@ forwardQueuedMessages = \targets t -> do
41494301 relevantMessages <- readSelectedBuckets neededBuckets
41504302 let candidates = forwardCandidates remoteTargets relevantMessages forwards t
41514303 totalFwdCount <- *totalForwardCount
4152− -- Per-key cap: filter candidates to respect maxForwardsPerKey per recipient
4304+ -- Per-key and per-server caps: filter candidates to respect both limits
41534305 let perKeyCapped = fold (\acc c ->
41544306 let existingForKey = count (do f <- forwards; where f.toKey == c.toKey; yield f)
41554307 + count (do a <- acc; where a.toKey == c.toKey; yield a)
4156− in if existingForKey >= maxForwardsPerKey then acc else union acc [c]) [] candidates
4308+ in let existingForServer = count (do f <- forwards; where f.toServer == c.toServer; yield f)
4309+ + count (do a <- acc; where a.toServer == c.toServer; yield a)
4310+ in if existingForKey >= maxForwardsPerKey || existingForServer >= maxForwardsPerServer then acc else union acc [c]) [] candidates
41574311 let capped = takeRelation (maxForwardsEntries - totalFwdCount) perKeyCapped
41584312 -- Write back only the affected forward buckets (not all 16)
41594313 let combined = union forwards capped
@@ -4502,7 +4656,7 @@ pruneAndRetry = \t -> do
45024656 let absDrift = if cheapMsgDrift < 0 then 0 - cheapMsgDrift else cheapMsgDrift
45034657 if absDrift > reconciliationDriftThreshold || isFullCycle
45044658 then do
4505− msgDrift <- reconcileMessageCounts 0 0
4659+ msgDrift <- reconcileMessageCountsDelta
45064660 when (msgDrift != 0)
45074661 (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))
45084662 else when (cheapMsgDrift != 0) (atomic do
@@ -4568,7 +4722,7 @@ pruneAndRetry = \t -> do
45684722 set *rateLimitedFederation = 0
45694723 -- Prune expired verify cache entries
45704724 cache <- *verificationCache
4571− set *verificationCache = do c <- cache; where t - c.verifiedAt < verificationCacheMs; yield c
4725+ set *verificationCache = do c <- cache; where t - c.verifiedAt < verificationCacheMaxMs; yield c
45724726 -- Reset per-peer latency tracking with global window
45734727 set *peerLatencies = []
45744728 -- Reset client operation latency metrics (windowed with federation metrics)
@@ -4594,8 +4748,10 @@ pruneAndRetry = \t -> do
45944748 broadcastOfflineGossip t pr.peerServers pr.expiredLocal
45954749 when federationEnabled (do
45964750 retryPendingForwardsFromBuckets
4597− sendAcksAndClean result.pendingAcks
4598− retryOverflowAcks result.pendingOverflow
4751+ -- Coalesce regular and overflow acks: merge into a single send pass so acks to
4752+ -- the same origin server share HTTP round-trips instead of two separate send cycles.
4753+ let coalescedAcks = union result.pendingAcks result.pendingOverflow
4754+ sendAcksAndClean coalescedAcks
45994755 retryPendingCancels))
46004756 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
46014757 -- trigger offline gossip when they expire (avoids wasted federation traffic)
@@ -5020,7 +5176,7 @@ handleChallenge = \pubkey clientIp respond -> do
50205176 chalRate <- effectiveChallengeRate
50215177 let rateResult = checkPubkeyRate challengeRates pubkey chalRate challengeRateWindowMs maxChallengeRateEntries t
50225178 if rateResult.limited
5023− then yield {accepted: False {}, rateLimited: True {}, retryAfterMs: rateResult.retryAfterMs, error: "rate_limited"}
5179+ then yield {accepted: False {}, rateLimited: True {}, retryAfterMs: rateResult.retryAfterMs, error: "rate_limited:" ++ show challengeRateWindowMs ++ "ms"}
50245180 else do
50255181 set *challengeRates = rateResult.entries
50265182 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
@@ -5050,7 +5206,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
50505206 yield rateResult.limited
50515207 if authRateLimited then do
50525208 logEvent "warn" "rate_limited_per_key" ("endpoint=verify key=" ++ pubkey ++ " limit=" ++ show maxChallengeRatePerKey)
5053− yield (respond {token: "", expiresAt: 0, error: "rate_limited"})
5209+ yield (respond {token: "", expiresAt: 0, error: "rate_limited:" ++ show challengeRateWindowMs ++ "ms"})
50545210 else do
50555211 authGlobalLimited <- checkAuthGlobalRate t
50565212 if authGlobalLimited then yield (respond {token: "", expiresAt: 0, error: "auth_rate_limited"})
@@ -5321,7 +5477,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
53215477 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
53225478 let fbi = pubkeyBucket to
53235479 fwdBkt <- readFwdBucket fbi
5324− let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: msgExpiresAt, priority: msgPriority, senderKey: senderKey}
5480+ -- Per-server cap: prevents one unreachable peer from consuming the entire forward queue
5481+ let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s) && count (do f <- fwdBkt; where f.toServer == s; yield f) < maxForwardsPerServer; yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: msgExpiresAt, priority: msgPriority, senderKey: senderKey}
53255482 totalFwdCount <- *totalForwardCount
53265483 let globalCap = maxForwardsEntries - totalFwdCount
53275484 -- Per-key cap: prevents one key from consuming disproportionate forward slots
@@ -5413,7 +5570,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
54135570 if dp >= 2 then yield (respond {status: MsDraining {}, messageId: "", retryAfterMs: 0, error: "draining"})
54145571 else do
54155572 globalRateLimited <- checkGlobalRate t
5416− if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0, error: "rate_limited"})
5573+ if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
54175574 else do
54185575 effectiveMsgId <- if messageId != "" then yield messageId else genToken
54195576 if t - timestamp > messageMaxAge
@@ -5445,7 +5602,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
54455602handleLookupPresence = \pubkey authorization clientIp respond -> do
54465603 t <- now
54475604 globalRateLimited <- checkGlobalRate t
5448− if globalRateLimited then yield (respond {server: "", authorized: False {}, error: "rate_limited"})
5605+ if globalRateLimited then yield (respond {server: "", authorized: False {}, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
54495606 else do
54505607 result <- atomic do
54515608 dp <- *drainPhase
@@ -5458,7 +5615,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
54585615 let rateResult = checkPubkeyRate lookupRates senderKey lkRate lookupRateWindowMs maxLookupRateEntries t
54595616 set *lookupRates = rateResult.entries
54605617 if rateResult.limited
5461− then yield {server: "", authorized: True {}, error: "rate_limited"}
5618+ then yield {server: "", authorized: True {}, error: "rate_limited:" ++ show lookupRateWindowMs ++ "ms"}
54625619 else do
54635620 -- Hot path: read only the target key's presence bucket
54645621 lookupBkt <- readPresenceBucket (pubkeyBucket pubkey)
@@ -5475,7 +5632,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
54755632handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
54765633 t <- now
54775634 globalRateLimited <- checkGlobalRate t
5478− if globalRateLimited then yield (respond {results: [], authorized: False {}, error: "rate_limited"})
5635+ if globalRateLimited then yield (respond {results: [], authorized: False {}, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
54795636 else do
54805637 dp <- *drainPhase
54815638 if dp >= 2 then yield (respond {results: [], authorized: False {}, error: ""})
@@ -5489,7 +5646,7 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
54895646 let rateResult = checkPubkeyRate lookupRates senderKey lkRate lookupRateWindowMs maxLookupRateEntries t
54905647 set *lookupRates = rateResult.entries
54915648 yield rateResult.limited
5492− if lookupRateLimited then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
5649+ if lookupRateLimited then yield (respond {results: [], authorized: True {}, error: "rate_limited:" ++ show lookupRateWindowMs ++ "ms"})
54935650 else do
54945651 -- Deduplicate input pubkeys to avoid redundant presence bucket reads
54955652 -- (a client could send 50 copies of the same pubkey, each triggering a scan)
@@ -5699,12 +5856,16 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
56995856 let entry = headOf (do r <- ger; where r.server == fromServer; yield r)
57005857 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= gossipEventRateWindowMs
57015858 let currentCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count
5702− if currentCount + eventCount > maxGossipEventsPerServer
5859+ -- Sliding window: interpolate previous window's count for smoother rate limiting
5860+ let elapsed = case entry of Nothing {} -> 0; Just {value: r} -> t - r.windowStart
5861+ let prevWeight = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.prevCount * (gossipEventRateWindowMs - elapsed) / gossipEventRateWindowMs
5862+ if prevWeight + currentCount + eventCount > maxGossipEventsPerServer
57035863 then yield (True {})
57045864 else do
57055865 let newEntries = if isNewWindow
5706− then union (do r <- ger; where r.server != fromServer; yield r) [{server: fromServer, windowStart: t, count: eventCount}]
5707− else do r <- ger; yield (if r.server == fromServer then {server: r.server, windowStart: r.windowStart, count: r.count + eventCount} else r)
5866+ then let pc = case entry of Nothing {} -> 0; Just {value: r} -> r.count
5867+ in union (do r <- ger; where r.server != fromServer; yield r) [{server: fromServer, windowStart: t, count: eventCount, prevCount: pc}]
5868+ else do r <- ger; yield (if r.server == fromServer then {server: r.server, windowStart: r.windowStart, count: r.count + eventCount, prevCount: r.prevCount} else r)
57085869 set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs; yield r) else newEntries
57095870 yield (False {})
57105871 if eventRateLimited then do
@@ -6293,7 +6454,7 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
62936454 if dp >= 2 then yield (respond {results: [], authorized: False {}, error: ""})
62946455 else do
62956456 globalRateLimited <- checkGlobalRate t
6296− if globalRateLimited then yield (respond {results: [], authorized: False {}, error: "rate_limited"})
6457+ if globalRateLimited then yield (respond {results: [], authorized: False {}, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
62976458 else
62986459 withSessionAuth authorization t clientIp
62996460 (yield (respond {results: [], authorized: False {}, error: "unauthorized"}))
@@ -6304,7 +6465,7 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
63046465 let rateResult = checkPubkeyRate cwr senderKey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t
63056466 set *clientWriteRates = rateResult.entries
63066467 yield rateResult.limited
6307− if cancelRateLimited then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
6468+ if cancelRateLimited then yield (respond {results: [], authorized: True {}, error: "rate_limited:" ++ show clientWriteRateWindowMs ++ "ms"})
63086469 else do
63096470 let capped = takeRelation maxBatchSize cancels
63106471 -- Process each cancel, collecting results and per-server cancel targets
@@ -6541,7 +6702,7 @@ handleReconcile = \authorization respond -> do
65416702 stored <- *totalForwardCount
65426703 set *totalForwardCount = actual
65436704 yield (actual - stored)
6544− msgDrift <- reconcileMessageCounts 0 0
6705+ msgDrift <- reconcileMessageCountsDelta
65456706 when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0 || msgDrift != 0)
65466707 (logInfo ("reconcile: corrected drift — messages: " ++ show msgDrift ++ ", presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift))
65476708 yield (respond {ok: True {}, messageDrift: msgDrift, presenceDrift: presDrift, sessionDrift: sessDrift})
@@ -6584,10 +6745,11 @@ computeAndCacheHealth = \t -> do
65846745 let nonceOk = nonceCount * 10 <= maxForwardNonces * 9
65856746 -- Count peers with active failure records (degraded federation connectivity)
65866747 let degradedPeerCount = count (do f <- forwardFailures; where t - f.firstFailedAt < maxServerDowntime; yield f)
6587− -- Forward queue age health: find oldest pending forward across all buckets.
6588− -- Healthy when no forward is older than healthForwardQueueMaxAge (or no forwards exist).
6589− fwdOldestAge <- oldestForwardAgeAcrossBuckets 0 t 0
6590− let fwdQueueOk = fwdOldestAge <= healthForwardQueueMaxAge
6748+ -- Forward queue age health: per-server awareness. Only flag unhealthy when
6749+ -- multiple destination servers have stale queues (oldest forward > healthForwardQueueMaxAge).
6750+ -- A single stuck peer shouldn't mark the server unhealthy — it's an isolated issue.
6751+ fwdStaleCount <- countStaleForwardServers 0 t []
6752+ let fwdQueueOk = fwdStaleCount < healthForwardQueueMinStaleServers
65916753 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
65926754 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
65936755 capacityOk: capacityOk,
@@ -6869,6 +7031,34 @@ enrichForwardServerStats = \servers b t ->
68697031 {server: s.server, count: s.count, oldestAge: max s.oldestAge bucketOldest, totalRetries: s.totalRetries + bucketRetries})
68707032 enrichForwardServerStats updated (b + 1) t
68717033
7034+-- GET /stats/rates: Show effective rate limits after adaptive scaling.
7035+-- Returns the current scale factor (0-100%) and each rate limit as actually applied.
7036+handleStatsRates = \authorization respond -> do
7037+ t <- now
7038+ globalRateLimited <- checkGlobalRate t
7039+ if globalRateLimited then yield (respond {ok: False {}, scalePct: 0, globalRequest: 0, ipRate: 0, sendRate: 0, federationRate: 0, authGlobal: 0, recvRate: 0, challengeRate: 0, pairRate: 0, lookupRate: 0, clientWriteRate: 0})
7040+ else do
7041+ authorized <- verifyMaintenanceAuth authorization
7042+ if authorized
7043+ then do
7044+ scale <- atomic do
7045+ tmc <- *totalMessageCount
7046+ sc <- *totalSessionCount
7047+ pc <- *totalPresenceCount
7048+ yield (adaptiveRateScale tmc sc pc)
7049+ effGlobal <- effectiveGlobalRequestRate
7050+ effIp <- effectiveIpRate
7051+ effSend <- effectiveSendRate
7052+ effFed <- effectiveFederationRate
7053+ effAuth <- effectiveAuthGlobalRate
7054+ effRecv <- effectiveRecvRate
7055+ effChal <- effectiveChallengeRate
7056+ effPair <- effectivePairRate
7057+ effLookup <- effectiveLookupRate
7058+ effCw <- effectiveClientWriteRate
7059+ yield (respond {ok: True {}, scalePct: scale, globalRequest: effGlobal, ipRate: effIp, sendRate: effSend, federationRate: effFed, authGlobal: effAuth, recvRate: effRecv, challengeRate: effChal, pairRate: effPair, lookupRate: effLookup, clientWriteRate: effCw})
7060+ else yield (respond {ok: False {}, scalePct: 0, globalRequest: 0, ipRate: 0, sendRate: 0, federationRate: 0, authGlobal: 0, recvRate: 0, challengeRate: 0, pairRate: 0, lookupRate: 0, clientWriteRate: 0})
7061+
68727062-- GET /ready: Readiness probe — returns true once the server has completed startup.
68737063-- Unlike /health, this is unauthenticated and doesn't check subsystem liveness.
68747064-- Returns false until the first prune cycle completes (server fully initialized).
@@ -6897,7 +7087,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
68977087 if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "draining"})
68987088 else do
68997089 globalRateLimited <- checkGlobalRate t
6900− if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "rate_limited"})
7090+ if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
69017091 else
69027092 withSessionAuth authorization t clientIp
69037093 (yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "unauthorized"}))
@@ -6909,7 +7099,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
69097099 let rateResult = checkPubkeyRate cwr pubkey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t
69107100 set *clientWriteRates = rateResult.entries
69117101 yield rateResult.limited
6912− if ackRateLimited then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})
7102+ if ackRateLimited then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited:" ++ show clientWriteRateWindowMs ++ "ms"})
69137103 else do
69147104 -- Cap input size to prevent expensive scans from unbounded messageIds lists
69157105 let dedupedIds = dedupMessageIds (takeRelation maxMessageIds messageIds)
@@ -6967,7 +7157,7 @@ handleListMessages = \offset authorization clientIp respond -> do
69677157 if dp >= 2 then yield (respond {messages: [], authorized: False {}, count: 0, error: ""})
69687158 else do
69697159 globalRateLimited <- checkGlobalRate t
6970− if globalRateLimited then yield (respond {messages: [], authorized: False {}, count: 0, error: "rate_limited"})
7160+ if globalRateLimited then yield (respond {messages: [], authorized: False {}, count: 0, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
69717161 else
69727162 withSessionAuth authorization t clientIp
69737163 (yield (respond {messages: [], authorized: False {}, count: 0, error: ""}))
@@ -6978,7 +7168,7 @@ handleListMessages = \offset authorization clientIp respond -> do
69787168 let rateResult = checkPubkeyRate lookupRates pubkey lkRate lookupRateWindowMs maxLookupRateEntries t
69797169 set *lookupRates = rateResult.entries
69807170 yield rateResult.limited
6981− if listRateLimited then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})
7171+ if listRateLimited then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited:" ++ show lookupRateWindowMs ++ "ms"})
69827172 else do
69837173 result <- atomic do
69847174 bucket <- readMessageBucket (pubkeyBucket pubkey)
@@ -7003,7 +7193,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
70037193 if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: ""})
70047194 else do
70057195 globalRateLimited <- checkGlobalRate t
7006− if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: "rate_limited"})
7196+ if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
70077197 else
70087198 withSessionAuth authorization t clientIp
70097199 (yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: "unauthorized"}))
@@ -7015,7 +7205,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
70157205 let rateResult = checkPubkeyRate cwr pubkey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t
70167206 set *clientWriteRates = rateResult.entries
70177207 yield rateResult.limited
7018− if deleteRateLimited then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})
7208+ if deleteRateLimited then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited:" ++ show clientWriteRateWindowMs ++ "ms"})
70197209 else do
70207210 -- Cap input size to prevent expensive scans from unbounded messageIds lists
70217211 let dedupedIds = dedupMessageIds (takeRelation maxMessageIds messageIds)
@@ -7087,7 +7277,7 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
70877277 if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: ""})
70887278 else do
70897279 globalRateLimited <- checkGlobalRate t
7090− if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: "rate_limited"})
7280+ if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
70917281 else
70927282 withSessionAuth authorization t clientIp
70937283 (yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: "unauthorized"}))
@@ -7099,7 +7289,7 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
70997289 let rateResult = checkPubkeyRate cwr senderKey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t
71007290 set *clientWriteRates = rateResult.entries
71017291 yield rateResult.limited
7102− if cancelRateLimited then yield (respond {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited"})
7292+ if cancelRateLimited then yield (respond {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited:" ++ show clientWriteRateWindowMs ++ "ms"})
71037293 else do
71047294 cancelResult <- atomic do
71057295 let b = pubkeyBucket to
@@ -7814,6 +8004,7 @@ serveInner = \req -> case req of
78148004 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
78158005 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond
78168006 StatsForwards {authorization, respond} -> handleStatsForwards authorization respond
8007+ StatsRates {authorization, respond} -> handleStatsRates authorization respond
78178008 Ready {respond} -> handleReady respond
78188009
78198010-- Backpressure: check if the server is overloaded (too many concurrent requests).
@@ -7830,15 +8021,21 @@ isOverloadExempt = \req -> case req of
78308021 Stats {} -> True {}
78318022 StatsPeers {} -> True {}
78328023 StatsKey {} -> True {}
8024+ StatsRates {} -> True {}
78338025 _ -> False {}
78348026
78358027-- Reject overloaded requests with appropriate empty/error responses per route type.
78368028-- Each response matches the handler's return shape with zeroed/empty fields.
8029+-- NOTE: This exhaustive case match is required by Knot's type system — each route has a
8030+-- different response shape. If Knot adds a default response constructor or row-polymorphic
8031+-- error responses, this can be collapsed to a single default branch.
78378032serveOverloaded = \req -> case req of
8033+ -- Auth routes
78388034 ReqChallenge {respond} -> respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 1000, error: "overloaded"}
78398035 VerifyAuth {respond} -> respond {token: "", expiresAt: 0, error: "overloaded"}
78408036 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}
78418037 Logout {respond} -> respond {ok: False {}, authorized: False {}}
8038+ -- Client routes
78428039 Poll {respond} -> respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: 0, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 1000, cursorReset: False {}, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0}
78438040 SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded"}
78448041 MessageStatus {respond} -> respond {status: "", authorized: False {}}
@@ -7850,6 +8047,7 @@ serveOverloaded = \req -> case req of
78508047 CancelBatch {respond} -> respond {results: [], authorized: False {}, error: "overloaded"}
78518048 LookupPresence {respond} -> respond {server: "", authorized: False {}, error: "overloaded"}
78528049 BatchLookupPresence {respond} -> respond {results: [], authorized: False {}, error: "overloaded"}
8050+ -- Federation routes
78538051 RecvGossip {respond} -> respond {ok: False {}, error: "overloaded:1000ms"}
78548052 ForwardMessage {respond} -> respond {ok: False {}, error: "overloaded:1000ms"}
78558053 FederationAck {respond} -> respond {ok: False {}, error: "overloaded:1000ms", ackedCount: 0, results: []}
@@ -7857,8 +8055,10 @@ serveOverloaded = \req -> case req of
78578055 FederationBatchCancel {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}
78588056 FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}
78598057 VerifyToken {respond} -> respond {ok: False {}}
8058+ -- Stats/admin routes (exempt routes handled by isOverloadExempt; these are fallbacks)
78608059 TopRecipients {respond} -> respond {recipients: [], ok: False {}, total: 0}
78618060 StatsForwards {respond} -> respond {servers: [], ok: False {}, total: 0}
8061+ StatsRates {respond} -> respond {ok: False {}, scalePct: 0, globalRequest: 0, ipRate: 0, sendRate: 0, federationRate: 0, authGlobal: 0, recvRate: 0, challengeRate: 0, pairRate: 0, lookupRate: 0, clientWriteRate: 0}
78628062 BlockPeer {respond} -> respond {ok: False {}, blocked: 0}
78638063 UnblockPeer {respond} -> respond {ok: False {}, unblocked: False {}}
78648064 ListBlocked {respond} -> respond {peers: [], ok: False {}}
@@ -7934,14 +8134,15 @@ serve = \req -> do
79348134 _ -> yield {}
79358135 let path = requestPath req
79368136 let ip = requestClientIp req
7937− let ipSuffix = if ip != "" then " " ++ ip else ""
8137+ -- Structured access log: machine-parseable JSON for monitoring pipelines.
8138+ -- Polls only logged when slow (> slowRequestThreshold) to avoid log volume from long-poll traffic.
79388139 if path != ""
79398140 then do
7940− logInfo ("[ACCESS] " ++ reqId ++ " " ++ path ++ " " ++ show latency ++ "ms" ++ ipSuffix)
8141+ println ("{\"level\":\"info\",\"event\":\"access\",\"req\":\"" ++ escapeJsonString reqId ++ "\",\"path\":\"" ++ escapeJsonString path ++ "\",\"ms\":" ++ show latency ++ (if ip != "" then ",\"ip\":\"" ++ escapeJsonString ip ++ "\"" else "") ++ "}")
79418142 when (latency > slowRequestThreshold)
79428143 (logEvent "warn" "slow_request" (path ++ " " ++ show latency ++ "ms"))
79438144 else when (latency > slowRequestThreshold) (do
7944− logInfo ("[SLOW] " ++ reqId ++ " poll " ++ show latency ++ "ms" ++ ipSuffix)
8145+ println ("{\"level\":\"warn\",\"event\":\"access\",\"req\":\"" ++ escapeJsonString reqId ++ "\",\"path\":\"poll\",\"ms\":" ++ show latency ++ (if ip != "" then ",\"ip\":\"" ++ escapeJsonString ip ++ "\"" else "") ++ ",\"slow\":true}")
79458146 logEvent "warn" "slow_request" ("poll " ++ show latency ++ "ms"))
79468147
79478148-- ============================================
@@ -8197,6 +8398,7 @@ main = do
81978398 && clockSkewTolerance < challengeExpiry
81988399 && maxMessagesPerRecipient <= maxMessages
81998400 && maxForwardsPerKey <= maxForwardsEntries
8401+ && maxForwardsPerServer <= maxForwardsEntries
82008402 && maxSessionsPerKey <= maxSessions
82018403 && maxChallengesPerKey <= maxChallenges
82028404 && maxPollsPerKey <= maxActivePolls
@@ -8213,7 +8415,7 @@ main = do
82138415 if not configOk
82148416 then logError "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
82158417 else if not configRelationsOk
8216− then logError "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, sessionExpiry <= maxSessionLifetime, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, per-key limits <= global limits), server not started"
8418+ then logError "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, sessionExpiry <= maxSessionLifetime, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, per-key/per-server limits <= global limits), server not started"
82178419 else do
82188420 t <- now
82198421 atomic do
@@ -8244,7 +8446,7 @@ main = do
82448446 logInfo (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")
82458447 logInfo (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")
82468448 logInfo (" max presence: " ++ show maxPresenceEntries)
8247− logInfo (" max forwards: " ++ show maxForwardsEntries ++ " (per key: " ++ show maxForwardsPerKey ++ ")")
8449+ logInfo (" max forwards: " ++ show maxForwardsEntries ++ " (per key: " ++ show maxForwardsPerKey ++ ", per server: " ++ show maxForwardsPerServer ++ ")")
82488450 logInfo (" session expiry: " ++ show sessionExpiry ++ "ms (max lifetime: " ++ show maxSessionLifetime ++ "ms)")
82498451 logInfo (" message max age: " ++ show messageMaxAge ++ "ms")
82508452 logInfo (" max active requests: " ++ if maxActiveRequests > 0 then show maxActiveRequests else "unlimited")