7eba17e · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Implement 14 security, performance, resilience, and operational improvements

7eba17eiko6 months ago

server.knot+314−112

compare with master

unifiedsplit

@@ -66,6 +66,7 @@ clockSkewTolerance : Int<Ms>

66clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers

67maxForwardsEntries = 10000

68maxForwardsPerKey = 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)

69maxServerNameLen = 128

70maxServerNameLen = 128

70maxMessageIdLen = 128

71maxMessageIdLen = 128

71maxCorrelationIdLen = 128

72maxCorrelationIdLen = 128

@@ -184,7 +185,7 @@ federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after thi

184

185

185-- Server build version: included in /health and /stats responses for deployment verification.

186-- Server build version: included in /health and /stats responses for deployment verification.

186-- Update this string when deploying a new build so operators can confirm which version is running.

187-- 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"

188

189

189-- Federation protocol version: included in gossip/forward/ack requests.

190-- Federation protocol version: included in gossip/forward/ack requests.

190-- Peers log warnings when they receive requests from older versions.

191-- Peers log warnings when they receive requests from older versions.

@@ -299,8 +300,12 @@ maxCancelRetries = 10

299-- Verify callback caching: cache successful verifications per-server to reduce

300-- Verify callback caching: cache successful verifications per-server to reduce

300-- outbound HTTP calls. A peer sending N requests/second forces N verify callbacks;

301-- outbound HTTP calls. A peer sending N requests/second forces N verify callbacks;

301-- caching the success for a short window amortizes the cost.

302-- caching the success for a short window amortizes the cost.

302−verificationCacheMs : Int<Ms>

303+-- Adaptive: scales from verificationCacheMinMs to verificationCacheMaxMs based on

303−verificationCacheMs = 2000 -- 2 seconds; short enough to limit replay window

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

304maxVerificationCache = 100

309maxVerificationCache = 100

305-- Health endpoint rate limiting (global, since /health has no clientIp)

310-- Health endpoint rate limiting (global, since /health has no clientIp)

306maxHealthRate = 100 -- max health checks per window

311maxHealthRate = 100 -- max health checks per window

@@ -313,11 +318,14 @@ healthHeartbeatMaxAge : Int<Ms>

313healthHeartbeatMaxAge = pollTimeoutMs * 3

318healthHeartbeatMaxAge = pollTimeoutMs * 3

314healthWatchdogMaxAge : Int<Ms>

319healthWatchdogMaxAge : Int<Ms>

315healthWatchdogMaxAge = backgroundPruneInterval * 15 -- watchdog runs every ~5 prune intervals

320healthWatchdogMaxAge = backgroundPruneInterval * 15 -- watchdog runs every ~5 prune intervals

316−-- Forward queue age health threshold: flag unhealthy when the oldest pending forward

321+-- Forward queue age health threshold: flag unhealthy when multiple destination servers

317−-- is older than this. Indicates a destination server is unreachable and forward entries

322+-- have pending forwards older than this. A single stuck peer shouldn't mark the whole

318−-- are accumulating without being delivered or retried successfully.

323+-- server unhealthy — only widespread queue buildup indicates a systemic problem.

319healthForwardQueueMaxAge : Int<Ms>

324healthForwardQueueMaxAge : Int<Ms>

320healthForwardQueueMaxAge = retryBackoffMax * 3 -- ~24 minutes; 3x max backoff gives retries time to succeed

325healthForwardQueueMaxAge = 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

321

329

322-- ============================================

330-- ============================================

323-- Types

331-- Types

@@ -409,21 +417,23 @@ type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp

409type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}

417type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}

410-- Tracks consecutive failures per federation peer for exponential backoff

418-- Tracks consecutive failures per federation peer for exponential backoff

411type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFailedAt: Timestamp}

419type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFailedAt: Timestamp}

412−-- Sliding-window rate limit state per pubkey

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.

413−type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

423+type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}

414−type RecvRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

424+type RecvRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}

415−type PairRateEntry = {sender: PubkeyHex, recipient: PubkeyHex, windowStart: Timestamp, count: Int}

425+type PairRateEntry = {sender: PubkeyHex, recipient: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}

416−type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

426+type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}

417-- Sliding-window rate limit state per federation peer server

427-- Sliding-window rate limit state per federation peer server

418−type FederationRateEntry = {server: Text, windowStart: Timestamp, count: Int}

428+type FederationRateEntry = {server: Text, windowStart: Timestamp, count: Int, prevCount: Int}

419−type FederationPairRateEntry = {server: Text, recipient: PubkeyHex, windowStart: Timestamp, count: Int}

429+type FederationPairRateEntry = {server: Text, recipient: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}

420-- Tracks active long-poll connections to enforce per-key concurrency limits

430-- Tracks active long-poll connections to enforce per-key concurrency limits

421type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}

431type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}

422data AuthResult = Authorized {pubkey: PubkeyHex, ip: Text} | Unauthorized {}

432data AuthResult = Authorized {pubkey: PubkeyHex, ip: Text} | Unauthorized {}

423data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

433data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}

424

434

425−type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int}

435+type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int, prevCount: Int}

426−type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

436+type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}

427-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles

437-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles

428type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}

438type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}

429type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool, pendingForwards: Int, avgLatencyMs: Int, maxLatencyMs: Int, latencySamples: Int}

439type 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}

432type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}

442type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}

433-- Tracks recent federation request (server, timestamp) pairs for replay detection

443-- Tracks recent federation request (server, timestamp) pairs for replay detection

434type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}

444type 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}

436-- Tracks unique pubkeys authenticated per IP (key-rotation abuse detection)

446-- Tracks unique pubkeys authenticated per IP (key-rotation abuse detection)

437type UniqueKeyIpEntry = {ip: Text, keys: [PubkeyHex], windowStart: Timestamp}

447type UniqueKeyIpEntry = {ip: Text, keys: [PubkeyHex], windowStart: Timestamp}

438

448

@@ -472,8 +482,12 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe

472-- Hot-path auth scans only one bucket (~1/16 of sessions).

482-- Hot-path auth scans only one bucket (~1/16 of sessions).

473-- TODO: If Knot adds indexed arrays or maps, replace the 16 manual bucket

483-- TODO: If Knot adds indexed arrays or maps, replace the 16 manual bucket

474-- refs and their if-else dispatch chains with a single indexed collection.

484-- refs and their if-else dispatch chains with a single indexed collection.

475−-- This would eliminate ~1000 lines of mechanical boilerplate across sessions,

485+-- This would eliminate ~1500 lines of mechanical boilerplate:

476−-- messages, presence, signal, and recipientCount buckets.

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

477*sessionBucket0 : [Session]

491*sessionBucket0 : [Session]

478*sessionBucket1 : [Session]

492*sessionBucket1 : [Session]

479*sessionBucket2 : [Session]

493*sessionBucket2 : [Session]

@@ -656,12 +670,14 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe

656*federationPairRates : [FederationPairRateEntry]

670*federationPairRates : [FederationPairRateEntry]

657

671

658*globalRateCount : Int

672*globalRateCount : Int

673+*globalRatePrevCount : Int

659

674

660*globalRateWindowStart : Timestamp

675*globalRateWindowStart : Timestamp

661

676

662-- Health endpoint rate limiting state (separate from global rate to avoid starving real requests)

677-- Health endpoint rate limiting state (separate from global rate to avoid starving real requests)

663*healthRateCount : Int

678*healthRateCount : Int

664*healthRateWindowStart : Timestamp

679*healthRateWindowStart : Timestamp

680+*healthRatePrevCount : Int

665

681

666-- Background task cycle counters (exposed via /stats for health monitoring)

682-- Background task cycle counters (exposed via /stats for health monitoring)

667*pruneCount : Int

683*pruneCount : Int

@@ -759,6 +775,7 @@ maxPeerLatencyEntries = 100

759

775

760-- Auth-tier global rate limiting state

776-- Auth-tier global rate limiting state

761*authGlobalRateCount : Int

777*authGlobalRateCount : Int

778+*authGlobalRatePrevCount : Int

762*authGlobalRateWindowStart : Timestamp

779*authGlobalRateWindowStart : Timestamp

763

780

764-- Max concurrent background prune forks (prevents resource exhaustion from stuck recovery)

781-- Max concurrent background prune forks (prevents resource exhaustion from stuck recovery)

@@ -1158,6 +1175,12 @@ route StatsApi where

1158 GET /stats/forwards headers {authorization: Text}

1175 GET /stats/forwards headers {authorization: Text}

1159 -> {servers: [{server: Text, count: Int, oldestAge: Int, totalRetries: Int}], ok: Bool, total: Int}

1176 -> {servers: [{server: Text, count: Int, oldestAge: Int, totalRetries: Int}], ok: Bool, total: Int}

1160 = StatsForwards

1177 = 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

1161

1184

1162-- Readiness probe: returns false during startup warmup (before first prune completes),

1185-- Readiness probe: returns false during startup warmup (before first prune completes),

1163-- true once the server is fully operational. Unlike /health, this is unauthenticated

1186-- true once the server is fully operational. Unlike /health, this is unauthenticated

@@ -1630,6 +1653,19 @@ generateFederationAuth = \scope server -> do

1630-- No retry: a failed verify rejects the request. The peer's federation rate limit + circuit breaker

1653-- No retry: a failed verify rejects the request. The peer's federation rate limit + circuit breaker

1631-- handle unreliable peers; retrying here doubles verify traffic and blocks with a 1s sleep per op,

1654-- handle unreliable peers; retrying here doubles verify traffic and blocks with a 1s sleep per op,

1632-- which a malicious peer can exploit for amplification.

1655-- 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+

1633verifyFederationCallback : Text -> Text -> Text -> IO {network} Bool

1669verifyFederationCallback : Text -> Text -> Text -> IO {network} Bool

1634verifyFederationCallback = \fromServer authorization scope ->

1670verifyFederationCallback = \fromServer authorization scope ->

1635 case extractToken authorization of

1671 case extractToken authorization of

@@ -1639,7 +1675,9 @@ verifyFederationCallback = \fromServer authorization scope ->

1639 t <- now

1675 t <- now

1640 cached <- atomic do

1676 cached <- atomic do

1641 cache <- *verificationCache

1677 cache <- *verificationCache

1678+ peerLats <- *peerLatencies

1679+ let cacheTtl = adaptiveVerifyCacheTtl fromServer peerLats

1642− yield (any cache (\c -> c.server == fromServer && c.scope == scope && t - c.verifiedAt < verificationCacheMs))

1680+ yield (any cache (\c -> c.server == fromServer && c.scope == scope && t - c.verifiedAt < cacheTtl))

1643 if cached then yield (True {})

1681 if cached then yield (True {})

1644 else do

1682 else do

1645 startTime <- now

1683 startTime <- now

@@ -1661,7 +1699,9 @@ verifyFederationCallback = \fromServer authorization scope ->

1661 Ok r -> do

1699 Ok r -> do

1662 when r.ok (atomic do

1700 when r.ok (atomic do

1663 cache <- *verificationCache

1701 cache <- *verificationCache

1702+ peerLats <- *peerLatencies

1703+ let cacheTtl = adaptiveVerifyCacheTtl fromServer peerLats

1664− let cleaned = do c <- cache; where t - c.verifiedAt < verificationCacheMs; yield c

1704+ let cleaned = do c <- cache; where t - c.verifiedAt < cacheTtl; yield c

1665 let updated = union cleaned [{server: fromServer, scope: scope, verifiedAt: t}]

1705 let updated = union cleaned [{server: fromServer, scope: scope, verifiedAt: t}]

1666 set *verificationCache = if count updated > maxVerificationCache

1706 set *verificationCache = if count updated > maxVerificationCache

1667 then evictOneOldest updated (\c -> c.verifiedAt)

1707 then evictOneOldest updated (\c -> c.verifiedAt)

@@ -2288,17 +2328,29 @@ pruneMessageBucketsConcurrent = \b t ->

2288-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one

2328-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one

2289-- bucket, so per-bucket recipient counts are disjoint — each bucket's counts are written

2329-- bucket, so per-bucket recipient counts are disjoint — each bucket's counts are written

2290-- to the corresponding rcBucket without overlap.

2330-- to the corresponding rcBucket without overlap.

2291−-- NOTE: TOCTOU window — messages can arrive between per-bucket reads, so the final

2331+-- Delta-based counter adjustment: snapshots totalMessageCount before scanning, computes the

2292−-- totalMessageCount write may slightly overwrite concurrent updates from sendMessageCore.

2332+-- drift (actual - snapshot), then applies the drift atomically to the live counter. This is

2293−-- This is acceptable: (a) reconciliation runs on drift detection or every 10th prune cycle,

2333+-- commutative with concurrent sendMessageCore: if sendMessageCore increments by N between our

2294−-- (b) the next prune re-reconciles, (c) capacity checks on the hot path use per-bucket

2334+-- snapshot and final adjustment, the live counter becomes (snapshot + N + drift) = (actual + N),

2295−-- rcBucket counts (not totalMessageCount) so message acceptance isn't affected.

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+

2296reconcileMessageCounts = \b totalAcc ->

2351reconcileMessageCounts = \b totalAcc ->

2297 if b > 15 then do

2352 if b > 15 then do

2298− atomic do

2299− stored <- *totalMessageCount

2300− set *totalMessageCount = totalAcc

2301− yield (totalAcc - stored)

2353+ yield totalAcc

2302 else do

2354 else do

2303 bucketTotal <- atomic do

2355 bucketTotal <- atomic do

2304 bucket <- readMessageBucket b

2356 bucket <- readMessageBucket b

@@ -2694,6 +2746,23 @@ oldestForwardAgeAcrossBuckets = \b t acc ->

2694 let bucketOldest = case minBy bucket (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt

2746 let bucketOldest = case minBy bucket (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt

2695 oldestForwardAgeAcrossBuckets (b + 1) t (max acc bucketOldest)

2747 oldestForwardAgeAcrossBuckets (b + 1) t (max acc bucketOldest)

2696

2748

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+

2697-- Remove forward entries matching a server and message IDs, scanning per-bucket.

2766-- Remove forward entries matching a server and message IDs, scanning per-bucket.

2698-- Only rewrites buckets that actually changed (avoids writeAllForwards repartitioning).

2767-- Only rewrites buckets that actually changed (avoids writeAllForwards repartitioning).

2699-- Returns all removed forward entries across all buckets.

2768-- Returns all removed forward entries across all buckets.

@@ -3062,20 +3131,28 @@ matchIp = \clientIp sessionIp ->

3062 else if clientIp == "" then False {} -- session has IP but request doesn't — proxy disappeared

3131 else if clientIp == "" then False {} -- session has IP but request doesn't — proxy disappeared

3063 else safeCompareLoop clientIp sessionIp maxIpLen 0 0 == 0

3132 else safeCompareLoop clientIp sessionIp maxIpLen 0 0 == 0

3064

3133

3065−-- Generic sliding-window rate limiter. Returns whether the request is rate-limited,

3134+-- Generic sliding-window rate limiter using the sliding window counter algorithm.

3066−-- the updated entries list, and retryAfterMs (time until the current window resets, 0 if not limited).

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.

3067−-- Callers provide match/create/update functions for their entry type.

3139+-- Callers provide match/create/update/rotate functions for their entry type.

3068-- Uses a direct fold for entry lookup (avoids materializing a filtered intermediate list).

3140-- 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 ->

3070 let entry = fold (\acc r -> case acc of Just _ -> acc; Nothing {} -> if matchKey r then Just {value: r} else acc) Nothing {} entries in

3142 let entry = fold (\acc r -> case acc of Just _ -> acc; Nothing {} -> if matchKey r then Just {value: r} else acc) Nothing {} entries in

3071 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= windowMs in

3143 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

3144+ -- Sliding window interpolation: weight previous window's count by remaining fraction

3073− if rateCount >= maxRate

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

3074− then let retryMs = case entry of Nothing {} -> windowMs; Just {value: r} -> max 0 (windowMs - (t - r.windowStart))

3150+ then let retryMs = case entry of Nothing {} -> windowMs; Just {value: r} -> max 0 (windowMs - elapsed)

3075 in {limited: True {}, entries: entries, retryAfterMs: retryMs}

3151 in {limited: True {}, entries: entries, retryAfterMs: retryMs}

3076 else

3152 else

3077 let newEntries = if isNewWindow

3153 let newEntries = if isNewWindow

3154+ then let prevCount = case entry of Nothing {} -> 0; Just {value: r} -> getCount r

3078− then union (do r <- entries; where not (matchKey r); yield r) [makeNew t]

3155+ in union (do r <- entries; where not (matchKey r); yield r) [rotateWindow (makeNew t) prevCount]

3079 else do r <- entries; yield (if matchKey r then updateCount r else r)

3156 else do r <- entries; yield (if matchKey r then updateCount r else r)

3080 in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs; yield r) else newEntries, retryAfterMs: 0}

3157 in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs; yield r) else newEntries, retryAfterMs: 0}

3081

3158

@@ -3083,26 +3160,34 @@ checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntr

3083-- Each returns {limited, entries, retryAfterMs} — caller reads/writes the mutable ref.

3160-- Each returns {limited, entries, retryAfterMs} — caller reads/writes the mutable ref.

3084checkPubkeyRate = \rates pubkey maxRate windowMs maxEntries t ->

3161checkPubkeyRate = \rates pubkey maxRate windowMs maxEntries t ->

3085 checkRateWindow rates (\r -> r.pubkey == pubkey)

3162 checkRateWindow rates (\r -> r.pubkey == pubkey)

3086− (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})

3163+ (\ts -> {pubkey: pubkey, windowStart: ts, count: 1, prevCount: 0})

3087− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})

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)

3088 maxRate windowMs maxEntries t

3167 maxRate windowMs maxEntries t

3089

3168

3090checkIpRate = \rates ip maxRate windowMs maxEntries t ->

3169checkIpRate = \rates ip maxRate windowMs maxEntries t ->

3091 checkRateWindow rates (\r -> r.ip == ip)

3170 checkRateWindow rates (\r -> r.ip == ip)

3092− (\ts -> {ip: ip, windowStart: ts, count: 1})

3171+ (\ts -> {ip: ip, windowStart: ts, count: 1, prevCount: 0})

3093− (\r -> {ip: r.ip, windowStart: r.windowStart, count: r.count + 1})

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)

3094 maxRate windowMs maxEntries t

3175 maxRate windowMs maxEntries t

3095

3176

3096checkServerRate = \rates server maxRate windowMs maxEntries t ->

3177checkServerRate = \rates server maxRate windowMs maxEntries t ->

3097 checkRateWindow rates (\r -> r.server == server)

3178 checkRateWindow rates (\r -> r.server == server)

3098− (\ts -> {server: server, windowStart: ts, count: 1})

3179+ (\ts -> {server: server, windowStart: ts, count: 1, prevCount: 0})

3099− (\r -> {server: r.server, windowStart: r.windowStart, count: r.count + 1})

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)

3100 maxRate windowMs maxEntries t

3183 maxRate windowMs maxEntries t

3101

3184

3102checkPairRate = \rates sender recipient maxRate windowMs maxEntries t ->

3185checkPairRate = \rates sender recipient maxRate windowMs maxEntries t ->

3103 checkRateWindow rates (\r -> r.sender == sender && r.recipient == recipient)

3186 checkRateWindow rates (\r -> r.sender == sender && r.recipient == recipient)

3104− (\ts -> {sender: sender, recipient: recipient, windowStart: ts, count: 1})

3187+ (\ts -> {sender: sender, recipient: recipient, windowStart: ts, count: 1, prevCount: 0})

3105− (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})

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)

3106 maxRate windowMs maxEntries t

3191 maxRate windowMs maxEntries t

3107

3192

3108-- Per-IP unique key rate check: tracks distinct pubkeys per IP per window.

3193-- Per-IP unique key rate check: tracks distinct pubkeys per IP per window.

@@ -3126,8 +3211,10 @@ checkUniqueKeyIpRate = \entries ip pubkey t ->

3126

3211

3127checkServerRecipientRate = \rates server recipient maxRate windowMs maxEntries t ->

3212checkServerRecipientRate = \rates server recipient maxRate windowMs maxEntries t ->

3128 checkRateWindow rates (\r -> r.server == server && r.recipient == recipient)

3213 checkRateWindow rates (\r -> r.server == server && r.recipient == recipient)

3129− (\ts -> {server: server, recipient: recipient, windowStart: ts, count: 1})

3214+ (\ts -> {server: server, recipient: recipient, windowStart: ts, count: 1, prevCount: 0})

3130− (\r -> {server: r.server, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})

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)

3131 maxRate windowMs maxEntries t

3218 maxRate windowMs maxEntries t

3132

3219

3133-- Extract the item with the smallest key and return both the item and remaining list

3220-- Extract the item with the smallest key and return both the item and remaining list

@@ -3675,54 +3762,70 @@ checkFederationRate = \server t -> atomic do

3675checkGlobalRate = \t -> atomic do

3762checkGlobalRate = \t -> atomic do

3676 windowStart <- *globalRateWindowStart

3763 windowStart <- *globalRateWindowStart

3677 rateCount <- *globalRateCount

3764 rateCount <- *globalRateCount

3765+ prevCount <- *globalRatePrevCount

3678 effectiveRate <- effectiveGlobalRequestRate

3766 effectiveRate <- effectiveGlobalRequestRate

3767+ let elapsed = t - windowStart

3679− if t - windowStart >= globalRateWindowMs

3768+ if elapsed >= globalRateWindowMs

3680 then do

3769 then do

3681 set *globalRateWindowStart = t

3770 set *globalRateWindowStart = t

3771+ set *globalRatePrevCount = rateCount

3682 set *globalRateCount = 1

3772 set *globalRateCount = 1

3683 yield (False {})

3773 yield (False {})

3774+ else do

3775+ -- Sliding window: interpolate previous window's count

3776+ let prevWeight = prevCount * (globalRateWindowMs - elapsed) / globalRateWindowMs

3684− else if rateCount >= effectiveRate

3777+ if prevWeight + rateCount >= effectiveRate

3685− then do

3778+ then do

3686− rlg <- *rateLimitedGlobal

3779+ rlg <- *rateLimitedGlobal

3687− set *rateLimitedGlobal = rlg + 1

3780+ set *rateLimitedGlobal = rlg + 1

3688− yield (True {})

3781+ yield (True {})

3689− else do

3782+ else do

3690− set *globalRateCount = rateCount + 1

3783+ set *globalRateCount = rateCount + 1

3691− yield (False {})

3784+ yield (False {})

3692

3785

3693-- Auth-tier global rate limiting (shared budget for challenge + verify + revoke)

3786-- Auth-tier global rate limiting (shared budget for challenge + verify + revoke)

3694checkAuthGlobalRate = \t -> atomic do

3787checkAuthGlobalRate = \t -> atomic do

3695 windowStart <- *authGlobalRateWindowStart

3788 windowStart <- *authGlobalRateWindowStart

3696 rateCount <- *authGlobalRateCount

3789 rateCount <- *authGlobalRateCount

3790+ prevCount <- *authGlobalRatePrevCount

3697 authRate <- effectiveAuthGlobalRate

3791 authRate <- effectiveAuthGlobalRate

3792+ let elapsed = t - windowStart

3698− if t - windowStart >= authGlobalRateWindowMs

3793+ if elapsed >= authGlobalRateWindowMs

3699 then do

3794 then do

3700 set *authGlobalRateWindowStart = t

3795 set *authGlobalRateWindowStart = t

3796+ set *authGlobalRatePrevCount = rateCount

3701 set *authGlobalRateCount = 1

3797 set *authGlobalRateCount = 1

3702 yield (False {})

3798 yield (False {})

3799+ else do

3800+ let prevWeight = prevCount * (authGlobalRateWindowMs - elapsed) / authGlobalRateWindowMs

3703− else if rateCount >= authRate

3801+ if prevWeight + rateCount >= authRate

3704− then do

3802+ then do

3705− rla <- *rateLimitedAuth

3803+ rla <- *rateLimitedAuth

3706− set *rateLimitedAuth = rla + 1

3804+ set *rateLimitedAuth = rla + 1

3707− yield (True {})

3805+ yield (True {})

3708− else do

3806+ else do

3709− set *authGlobalRateCount = rateCount + 1

3807+ set *authGlobalRateCount = rateCount + 1

3710− yield (False {})

3808+ yield (False {})

3711

3809

3712-- Health endpoint rate limiting (separate from global rate so monitoring doesn't starve real requests)

3810-- Health endpoint rate limiting (separate from global rate so monitoring doesn't starve real requests)

3713checkHealthRate = \t -> atomic do

3811checkHealthRate = \t -> atomic do

3714 windowStart <- *healthRateWindowStart

3812 windowStart <- *healthRateWindowStart

3715 rateCount <- *healthRateCount

3813 rateCount <- *healthRateCount

3814+ prevCount <- *healthRatePrevCount

3815+ let elapsed = t - windowStart

3716− if t - windowStart >= healthRateWindowMs

3816+ if elapsed >= healthRateWindowMs

3717 then do

3817 then do

3718 set *healthRateWindowStart = t

3818 set *healthRateWindowStart = t

3819+ set *healthRatePrevCount = rateCount

3719 set *healthRateCount = 1

3820 set *healthRateCount = 1

3720 yield (False {})

3821 yield (False {})

3822+ else do

3823+ let prevWeight = prevCount * (healthRateWindowMs - elapsed) / healthRateWindowMs

3721− else if rateCount >= maxHealthRate

3824+ if prevWeight + rateCount >= maxHealthRate

3722− then yield (True {})

3825+ then yield (True {})

3723− else do

3826+ else do

3724− set *healthRateCount = rateCount + 1

3827+ set *healthRateCount = rateCount + 1

3725− yield (False {})

3828+ yield (False {})

3726

3829

3727-- IP-based rate limiting; skips if clientIp header is empty (no reverse proxy)

3830-- IP-based rate limiting; skips if clientIp header is empty (no reverse proxy)

3728checkClientIpRate = \ip t ->

3831checkClientIpRate = \ip t ->

@@ -3991,8 +4094,16 @@ retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}

3991retryForwardsForServer = \entries remaining ->

4094retryForwardsForServer = \entries remaining ->

3992 if remaining <= 0 || count entries == 0 then yield {}

4095 if remaining <= 0 || count entries == 0 then yield {}

3993 else do

4096 else 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

3994− let server = case headOf entries of Just {value: f} -> f.toServer; Nothing {} -> ""

4105+ let server = case headOf liveEntries of Just {value: f} -> f.toServer; Nothing {} -> ""

3995− forEachBatch entries maxBatchSize (\batch -> do

4106+ forEachBatch liveEntries maxBatchSize (\batch -> do

3996 t <- now

4107 t <- now

3997 auth <- generateFederationAuth "forward-batch" server

4108 auth <- generateFederationAuth "forward-batch" server

3998 corrId <- genTokenN 1

4109 corrId <- genTokenN 1

@@ -4093,6 +4204,30 @@ collectEligibleForwards = \b eligibleServers ->

4093-- Circuit breaker: skips entirely when too many peers are failing (likely local network issue).

4204-- Circuit breaker: skips entirely when too many peers are failing (likely local network issue).

4094-- Per-bucket: reads forward buckets one at a time, collecting only entries for eligible servers,

4205-- Per-bucket: reads forward buckets one at a time, collecting only entries for eligible servers,

4095-- instead of materializing all forwards into memory via readAllForwards.

4206-- 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+

4096retryPendingForwardsFromBuckets : IO {network} {}

4231retryPendingForwardsFromBuckets : IO {network} {}

4097retryPendingForwardsFromBuckets = do

4232retryPendingForwardsFromBuckets = do

4098 t <- now

4233 t <- now

@@ -4101,6 +4236,8 @@ retryPendingForwardsFromBuckets = do

4101 then do

4236 then do

4102 logWarn ("federation: circuit breaker open (" ++ show (count failures) ++ " peers failing), skipping forward retries")

4237 logWarn ("federation: circuit breaker open (" ++ show (count failures) ++ " peers failing), skipping forward retries")

4103 logEvent "warn" "circuit_breaker_open" (show (count failures) ++ " peers failing")

4238 logEvent "warn" "circuit_breaker_open" (show (count failures) ++ " peers failing")

4239+ -- Half-open: probe one peer to detect recovery

4240+ circuitBreakerProbe failures t

4104 else do

4241 else do

4105 inFlight <- *federationInFlight

4242 inFlight <- *federationInFlight

4106 blocked <- *blockedPeers

4243 blocked <- *blockedPeers

@@ -4113,12 +4250,27 @@ retryPendingForwardsFromBuckets = do

4113 else do

4250 else do

4114 -- Read only entries for eligible servers (per-bucket, streaming)

4251 -- Read only entries for eligible servers (per-bucket, streaming)

4115 eligibleForwards <- collectEligibleForwards 0 eligibleServers

4252 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

4116 -- Prioritize forwards for recipients with active polls (currently online).

4268 -- Prioritize forwards for recipients with active polls (currently online).

4117 -- These keys will receive the message sooner, improving delivery latency.

4269 -- These keys will receive the message sooner, improving delivery latency.

4118 -- Forwards for offline keys are retried after online-key forwards complete.

4270 -- Forwards for offline keys are retried after online-key forwards complete.

4119 activePolls <- *activePolls

4271 activePolls <- *activePolls

4120− let onlineForwards = do f <- eligibleForwards; where any activePolls (\p -> p.pubkey == f.toKey); yield f

4272+ let onlineForwards = do f <- rampedForwards; 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

4273+ let offlineForwards = do f <- rampedForwards; where none activePolls (\p -> p.pubkey == f.toKey); yield f

4122 -- Sort by effective priority descending (higher priority first) within each group.

4274 -- Sort by effective priority descending (higher priority first) within each group.

4123 -- Penalty by retry count: messages with many retries are deprioritized so fresh

4275 -- Penalty by retry count: messages with many retries are deprioritized so fresh

4124 -- messages aren't blocked behind likely-failing ones. A fresh priority-5 (score -5)

4276 -- messages aren't blocked behind likely-failing ones. A fresh priority-5 (score -5)

@@ -4149,11 +4301,13 @@ forwardQueuedMessages = \targets t -> do

4149 relevantMessages <- readSelectedBuckets neededBuckets

4301 relevantMessages <- readSelectedBuckets neededBuckets

4150 let candidates = forwardCandidates remoteTargets relevantMessages forwards t

4302 let candidates = forwardCandidates remoteTargets relevantMessages forwards t

4151 totalFwdCount <- *totalForwardCount

4303 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

4153 let perKeyCapped = fold (\acc c ->

4305 let perKeyCapped = fold (\acc c ->

4154 let existingForKey = count (do f <- forwards; where f.toKey == c.toKey; yield f)

4306 let existingForKey = count (do f <- forwards; where f.toKey == c.toKey; yield f)

4155 + count (do a <- acc; where a.toKey == c.toKey; yield a)

4307 + count (do a <- acc; where a.toKey == c.toKey; yield a)

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)

4156− in if existingForKey >= maxForwardsPerKey then acc else union acc [c]) [] candidates

4310+ in if existingForKey >= maxForwardsPerKey || existingForServer >= maxForwardsPerServer then acc else union acc [c]) [] candidates

4157 let capped = takeRelation (maxForwardsEntries - totalFwdCount) perKeyCapped

4311 let capped = takeRelation (maxForwardsEntries - totalFwdCount) perKeyCapped

4158 -- Write back only the affected forward buckets (not all 16)

4312 -- Write back only the affected forward buckets (not all 16)

4159 let combined = union forwards capped

4313 let combined = union forwards capped

@@ -4502,7 +4656,7 @@ pruneAndRetry = \t -> do

4502 let absDrift = if cheapMsgDrift < 0 then 0 - cheapMsgDrift else cheapMsgDrift

4656 let absDrift = if cheapMsgDrift < 0 then 0 - cheapMsgDrift else cheapMsgDrift

4503 if absDrift > reconciliationDriftThreshold || isFullCycle

4657 if absDrift > reconciliationDriftThreshold || isFullCycle

4504 then do

4658 then do

4505− msgDrift <- reconcileMessageCounts 0 0

4659+ msgDrift <- reconcileMessageCountsDelta

4506 when (msgDrift != 0)

4660 when (msgDrift != 0)

4507 (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))

4661 (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))

4508 else when (cheapMsgDrift != 0) (atomic do

4662 else when (cheapMsgDrift != 0) (atomic do

@@ -4568,7 +4722,7 @@ pruneAndRetry = \t -> do

4568 set *rateLimitedFederation = 0

4722 set *rateLimitedFederation = 0

4569 -- Prune expired verify cache entries

4723 -- Prune expired verify cache entries

4570 cache <- *verificationCache

4724 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

4572 -- Reset per-peer latency tracking with global window

4726 -- Reset per-peer latency tracking with global window

4573 set *peerLatencies = []

4727 set *peerLatencies = []

4574 -- Reset client operation latency metrics (windowed with federation metrics)

4728 -- Reset client operation latency metrics (windowed with federation metrics)

@@ -4594,8 +4748,10 @@ pruneAndRetry = \t -> do

4594 broadcastOfflineGossip t pr.peerServers pr.expiredLocal

4748 broadcastOfflineGossip t pr.peerServers pr.expiredLocal

4595 when federationEnabled (do

4749 when federationEnabled (do

4596 retryPendingForwardsFromBuckets

4750 retryPendingForwardsFromBuckets

4597− sendAcksAndClean result.pendingAcks

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.

4598− retryOverflowAcks result.pendingOverflow

4753+ let coalescedAcks = union result.pendingAcks result.pendingOverflow

4754+ sendAcksAndClean coalescedAcks

4599 retryPendingCancels))

4755 retryPendingCancels))

4600 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally

4756 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally

4601 -- trigger offline gossip when they expire (avoids wasted federation traffic)

4757 -- trigger offline gossip when they expire (avoids wasted federation traffic)

@@ -5020,7 +5176,7 @@ handleChallenge = \pubkey clientIp respond -> do

5020 chalRate <- effectiveChallengeRate

5176 chalRate <- effectiveChallengeRate

5021 let rateResult = checkPubkeyRate challengeRates pubkey chalRate challengeRateWindowMs maxChallengeRateEntries t

5177 let rateResult = checkPubkeyRate challengeRates pubkey chalRate challengeRateWindowMs maxChallengeRateEntries t

5022 if rateResult.limited

5178 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"}

5024 else do

5180 else do

5025 set *challengeRates = rateResult.entries

5181 set *challengeRates = rateResult.entries

5026 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]

5182 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]

@@ -5050,7 +5206,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->

5050 yield rateResult.limited

5206 yield rateResult.limited

5051 if authRateLimited then do

5207 if authRateLimited then do

5052 logEvent "warn" "rate_limited_per_key" ("endpoint=verify key=" ++ pubkey ++ " limit=" ++ show maxChallengeRatePerKey)

5208 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"})

5054 else do

5210 else do

5055 authGlobalLimited <- checkAuthGlobalRate t

5211 authGlobalLimited <- checkAuthGlobalRate t

5056 if authGlobalLimited then yield (respond {token: "", expiresAt: 0, error: "auth_rate_limited"})

5212 if authGlobalLimited then yield (respond {token: "", expiresAt: 0, error: "auth_rate_limited"})

@@ -5321,7 +5477,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn

5321 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do

5477 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do

5322 let fbi = pubkeyBucket to

5478 let fbi = pubkeyBucket to

5323 fwdBkt <- readFwdBucket fbi

5479 fwdBkt <- readFwdBucket fbi

5480+ -- Per-server cap: prevents one unreachable peer from consuming the entire forward queue

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}

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}

5325 totalFwdCount <- *totalForwardCount

5482 totalFwdCount <- *totalForwardCount

5326 let globalCap = maxForwardsEntries - totalFwdCount

5483 let globalCap = maxForwardsEntries - totalFwdCount

5327 -- Per-key cap: prevents one key from consuming disproportionate forward slots

5484 -- Per-key cap: prevents one key from consuming disproportionate forward slots

@@ -5413,7 +5570,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a

5413 if dp >= 2 then yield (respond {status: MsDraining {}, messageId: "", retryAfterMs: 0, error: "draining"})

5570 if dp >= 2 then yield (respond {status: MsDraining {}, messageId: "", retryAfterMs: 0, error: "draining"})

5414 else do

5571 else do

5415 globalRateLimited <- checkGlobalRate t

5572 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"})

5417 else do

5574 else do

5418 effectiveMsgId <- if messageId != "" then yield messageId else genToken

5575 effectiveMsgId <- if messageId != "" then yield messageId else genToken

5419 if t - timestamp > messageMaxAge

5576 if t - timestamp > messageMaxAge

@@ -5445,7 +5602,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a

5445handleLookupPresence = \pubkey authorization clientIp respond -> do

5602handleLookupPresence = \pubkey authorization clientIp respond -> do

5446 t <- now

5603 t <- now

5447 globalRateLimited <- checkGlobalRate t

5604 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"})

5449 else do

5606 else do

5450 result <- atomic do

5607 result <- atomic do

5451 dp <- *drainPhase

5608 dp <- *drainPhase

@@ -5458,7 +5615,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do

5458 let rateResult = checkPubkeyRate lookupRates senderKey lkRate lookupRateWindowMs maxLookupRateEntries t

5615 let rateResult = checkPubkeyRate lookupRates senderKey lkRate lookupRateWindowMs maxLookupRateEntries t

5459 set *lookupRates = rateResult.entries

5616 set *lookupRates = rateResult.entries

5460 if rateResult.limited

5617 if rateResult.limited

5461− then yield {server: "", authorized: True {}, error: "rate_limited"}

5618+ then yield {server: "", authorized: True {}, error: "rate_limited:" ++ show lookupRateWindowMs ++ "ms"}

5462 else do

5619 else do

5463 -- Hot path: read only the target key's presence bucket

5620 -- Hot path: read only the target key's presence bucket

5464 lookupBkt <- readPresenceBucket (pubkeyBucket pubkey)

5621 lookupBkt <- readPresenceBucket (pubkeyBucket pubkey)

@@ -5475,7 +5632,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do

5475handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do

5632handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do

5476 t <- now

5633 t <- now

5477 globalRateLimited <- checkGlobalRate t

5634 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"})

5479 else do

5636 else do

5480 dp <- *drainPhase

5637 dp <- *drainPhase

5481 if dp >= 2 then yield (respond {results: [], authorized: False {}, error: ""})

5638 if dp >= 2 then yield (respond {results: [], authorized: False {}, error: ""})

@@ -5489,7 +5646,7 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do

5489 let rateResult = checkPubkeyRate lookupRates senderKey lkRate lookupRateWindowMs maxLookupRateEntries t

5646 let rateResult = checkPubkeyRate lookupRates senderKey lkRate lookupRateWindowMs maxLookupRateEntries t

5490 set *lookupRates = rateResult.entries

5647 set *lookupRates = rateResult.entries

5491 yield rateResult.limited

5648 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"})

5493 else do

5650 else do

5494 -- Deduplicate input pubkeys to avoid redundant presence bucket reads

5651 -- Deduplicate input pubkeys to avoid redundant presence bucket reads

5495 -- (a client could send 50 copies of the same pubkey, each triggering a scan)

5652 -- (a client could send 50 copies of the same pubkey, each triggering a scan)

@@ -5699,12 +5856,16 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation

5699 let entry = headOf (do r <- ger; where r.server == fromServer; yield r)

5856 let entry = headOf (do r <- ger; where r.server == fromServer; yield r)

5700 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= gossipEventRateWindowMs

5857 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= gossipEventRateWindowMs

5701 let currentCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count

5858 let currentCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count

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

5702− if currentCount + eventCount > maxGossipEventsPerServer

5862+ if prevWeight + currentCount + eventCount > maxGossipEventsPerServer

5703 then yield (True {})

5863 then yield (True {})

5704 else do

5864 else do

5705 let newEntries = if isNewWindow

5865 let newEntries = if isNewWindow

5866+ then let pc = case entry of Nothing {} -> 0; Just {value: r} -> r.count

5706− then union (do r <- ger; where r.server != fromServer; yield r) [{server: fromServer, windowStart: t, count: eventCount}]

5867+ in union (do r <- ger; where r.server != fromServer; yield r) [{server: fromServer, windowStart: t, count: eventCount, prevCount: pc}]

5707− else do r <- ger; yield (if r.server == fromServer then {server: r.server, windowStart: r.windowStart, count: r.count + eventCount} else r)

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)

5708 set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs; yield r) else newEntries

5869 set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs; yield r) else newEntries

5709 yield (False {})

5870 yield (False {})

5710 if eventRateLimited then do

5871 if eventRateLimited then do

@@ -6293,7 +6454,7 @@ handleCancelBatch = \cancels authorization clientIp respond -> do

6293 if dp >= 2 then yield (respond {results: [], authorized: False {}, error: ""})

6454 if dp >= 2 then yield (respond {results: [], authorized: False {}, error: ""})

6294 else do

6455 else do

6295 globalRateLimited <- checkGlobalRate t

6456 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"})

6297 else

6458 else

6298 withSessionAuth authorization t clientIp

6459 withSessionAuth authorization t clientIp

6299 (yield (respond {results: [], authorized: False {}, error: "unauthorized"}))

6460 (yield (respond {results: [], authorized: False {}, error: "unauthorized"}))

@@ -6304,7 +6465,7 @@ handleCancelBatch = \cancels authorization clientIp respond -> do

6304 let rateResult = checkPubkeyRate cwr senderKey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t

6465 let rateResult = checkPubkeyRate cwr senderKey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t

6305 set *clientWriteRates = rateResult.entries

6466 set *clientWriteRates = rateResult.entries

6306 yield rateResult.limited

6467 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"})

6308 else do

6469 else do

6309 let capped = takeRelation maxBatchSize cancels

6470 let capped = takeRelation maxBatchSize cancels

6310 -- Process each cancel, collecting results and per-server cancel targets

6471 -- Process each cancel, collecting results and per-server cancel targets

@@ -6541,7 +6702,7 @@ handleReconcile = \authorization respond -> do

6541 stored <- *totalForwardCount

6702 stored <- *totalForwardCount

6542 set *totalForwardCount = actual

6703 set *totalForwardCount = actual

6543 yield (actual - stored)

6704 yield (actual - stored)

6544− msgDrift <- reconcileMessageCounts 0 0

6705+ msgDrift <- reconcileMessageCountsDelta

6545 when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0 || msgDrift != 0)

6706 when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0 || msgDrift != 0)

6546 (logInfo ("reconcile: corrected drift — messages: " ++ show msgDrift ++ ", presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift))

6707 (logInfo ("reconcile: corrected drift — messages: " ++ show msgDrift ++ ", presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift))

6547 yield (respond {ok: True {}, messageDrift: msgDrift, presenceDrift: presDrift, sessionDrift: sessDrift})

6708 yield (respond {ok: True {}, messageDrift: msgDrift, presenceDrift: presDrift, sessionDrift: sessDrift})

@@ -6584,10 +6745,11 @@ computeAndCacheHealth = \t -> do

6584 let nonceOk = nonceCount * 10 <= maxForwardNonces * 9

6745 let nonceOk = nonceCount * 10 <= maxForwardNonces * 9

6585 -- Count peers with active failure records (degraded federation connectivity)

6746 -- Count peers with active failure records (degraded federation connectivity)

6586 let degradedPeerCount = count (do f <- forwardFailures; where t - f.firstFailedAt < maxServerDowntime; yield f)

6747 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.

6748+ -- Forward queue age health: per-server awareness. Only flag unhealthy when

6588− -- Healthy when no forward is older than healthForwardQueueMaxAge (or no forwards exist).

6749+ -- multiple destination servers have stale queues (oldest forward > healthForwardQueueMaxAge).

6589− fwdOldestAge <- oldestForwardAgeAcrossBuckets 0 t 0

6750+ -- A single stuck peer shouldn't mark the server unhealthy — it's an isolated issue.

6590− let fwdQueueOk = fwdOldestAge <= healthForwardQueueMaxAge

6751+ fwdStaleCount <- countStaleForwardServers 0 t []

6752+ let fwdQueueOk = fwdStaleCount < healthForwardQueueMinStaleServers

6591 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),

6753 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),

6592 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,

6754 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,

6593 capacityOk: capacityOk,

6755 capacityOk: capacityOk,

@@ -6869,6 +7031,34 @@ enrichForwardServerStats = \servers b t ->

6869 {server: s.server, count: s.count, oldestAge: max s.oldestAge bucketOldest, totalRetries: s.totalRetries + bucketRetries})

7031 {server: s.server, count: s.count, oldestAge: max s.oldestAge bucketOldest, totalRetries: s.totalRetries + bucketRetries})

6870 enrichForwardServerStats updated (b + 1) t

7032 enrichForwardServerStats updated (b + 1) t

6871

7033

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+

6872-- GET /ready: Readiness probe — returns true once the server has completed startup.

7062-- GET /ready: Readiness probe — returns true once the server has completed startup.

6873-- Unlike /health, this is unauthenticated and doesn't check subsystem liveness.

7063-- Unlike /health, this is unauthenticated and doesn't check subsystem liveness.

6874-- Returns false until the first prune cycle completes (server fully initialized).

7064-- Returns false until the first prune cycle completes (server fully initialized).

@@ -6897,7 +7087,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do

6897 if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "draining"})

7087 if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "draining"})

6898 else do

7088 else do

6899 globalRateLimited <- checkGlobalRate t

7089 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"})

6901 else

7091 else

6902 withSessionAuth authorization t clientIp

7092 withSessionAuth authorization t clientIp

6903 (yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "unauthorized"}))

7093 (yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "unauthorized"}))

@@ -6909,7 +7099,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do

6909 let rateResult = checkPubkeyRate cwr pubkey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t

7099 let rateResult = checkPubkeyRate cwr pubkey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t

6910 set *clientWriteRates = rateResult.entries

7100 set *clientWriteRates = rateResult.entries

6911 yield rateResult.limited

7101 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"})

6913 else do

7103 else do

6914 -- Cap input size to prevent expensive scans from unbounded messageIds lists

7104 -- Cap input size to prevent expensive scans from unbounded messageIds lists

6915 let dedupedIds = dedupMessageIds (takeRelation maxMessageIds messageIds)

7105 let dedupedIds = dedupMessageIds (takeRelation maxMessageIds messageIds)

@@ -6967,7 +7157,7 @@ handleListMessages = \offset authorization clientIp respond -> do

6967 if dp >= 2 then yield (respond {messages: [], authorized: False {}, count: 0, error: ""})

7157 if dp >= 2 then yield (respond {messages: [], authorized: False {}, count: 0, error: ""})

6968 else do

7158 else do

6969 globalRateLimited <- checkGlobalRate t

7159 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"})

6971 else

7161 else

6972 withSessionAuth authorization t clientIp

7162 withSessionAuth authorization t clientIp

6973 (yield (respond {messages: [], authorized: False {}, count: 0, error: ""}))

7163 (yield (respond {messages: [], authorized: False {}, count: 0, error: ""}))

@@ -6978,7 +7168,7 @@ handleListMessages = \offset authorization clientIp respond -> do

6978 let rateResult = checkPubkeyRate lookupRates pubkey lkRate lookupRateWindowMs maxLookupRateEntries t

7168 let rateResult = checkPubkeyRate lookupRates pubkey lkRate lookupRateWindowMs maxLookupRateEntries t

6979 set *lookupRates = rateResult.entries

7169 set *lookupRates = rateResult.entries

6980 yield rateResult.limited

7170 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"})

6982 else do

7172 else do

6983 result <- atomic do

7173 result <- atomic do

6984 bucket <- readMessageBucket (pubkeyBucket pubkey)

7174 bucket <- readMessageBucket (pubkeyBucket pubkey)

@@ -7003,7 +7193,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do

7003 if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: ""})

7193 if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: ""})

7004 else do

7194 else do

7005 globalRateLimited <- checkGlobalRate t

7195 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"})

7007 else

7197 else

7008 withSessionAuth authorization t clientIp

7198 withSessionAuth authorization t clientIp

7009 (yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: "unauthorized"}))

7199 (yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: "unauthorized"}))

@@ -7015,7 +7205,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do

7015 let rateResult = checkPubkeyRate cwr pubkey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t

7205 let rateResult = checkPubkeyRate cwr pubkey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t

7016 set *clientWriteRates = rateResult.entries

7206 set *clientWriteRates = rateResult.entries

7017 yield rateResult.limited

7207 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"})

7019 else do

7209 else do

7020 -- Cap input size to prevent expensive scans from unbounded messageIds lists

7210 -- Cap input size to prevent expensive scans from unbounded messageIds lists

7021 let dedupedIds = dedupMessageIds (takeRelation maxMessageIds messageIds)

7211 let dedupedIds = dedupMessageIds (takeRelation maxMessageIds messageIds)

@@ -7087,7 +7277,7 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do

7087 if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: ""})

7277 if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: ""})

7088 else do

7278 else do

7089 globalRateLimited <- checkGlobalRate t

7279 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"})

7091 else

7281 else

7092 withSessionAuth authorization t clientIp

7282 withSessionAuth authorization t clientIp

7093 (yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: "unauthorized"}))

7283 (yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: "unauthorized"}))

@@ -7099,7 +7289,7 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do

7099 let rateResult = checkPubkeyRate cwr senderKey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t

7289 let rateResult = checkPubkeyRate cwr senderKey cwRate clientWriteRateWindowMs maxClientWriteRateEntries t

7100 set *clientWriteRates = rateResult.entries

7290 set *clientWriteRates = rateResult.entries

7101 yield rateResult.limited

7291 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"})

7103 else do

7293 else do

7104 cancelResult <- atomic do

7294 cancelResult <- atomic do

7105 let b = pubkeyBucket to

7295 let b = pubkeyBucket to

@@ -7814,6 +8004,7 @@ serveInner = \req -> case req of

7814 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond

8004 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond

7815 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond

8005 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond

7816 StatsForwards {authorization, respond} -> handleStatsForwards authorization respond

8006 StatsForwards {authorization, respond} -> handleStatsForwards authorization respond

8007+ StatsRates {authorization, respond} -> handleStatsRates authorization respond

7817 Ready {respond} -> handleReady respond

8008 Ready {respond} -> handleReady respond

7818

8009

7819-- Backpressure: check if the server is overloaded (too many concurrent requests).

8010-- Backpressure: check if the server is overloaded (too many concurrent requests).

@@ -7830,15 +8021,21 @@ isOverloadExempt = \req -> case req of

7830 Stats {} -> True {}

8021 Stats {} -> True {}

7831 StatsPeers {} -> True {}

8022 StatsPeers {} -> True {}

7832 StatsKey {} -> True {}

8023 StatsKey {} -> True {}

8024+ StatsRates {} -> True {}

7833 _ -> False {}

8025 _ -> False {}

7834

8026

7835-- Reject overloaded requests with appropriate empty/error responses per route type.

8027-- Reject overloaded requests with appropriate empty/error responses per route type.

7836-- Each response matches the handler's return shape with zeroed/empty fields.

8028-- 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.

7837serveOverloaded = \req -> case req of

8032serveOverloaded = \req -> case req of

8033+ -- Auth routes

7838 ReqChallenge {respond} -> respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 1000, error: "overloaded"}

8034 ReqChallenge {respond} -> respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 1000, error: "overloaded"}

7839 VerifyAuth {respond} -> respond {token: "", expiresAt: 0, error: "overloaded"}

8035 VerifyAuth {respond} -> respond {token: "", expiresAt: 0, error: "overloaded"}

7840 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}

8036 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}

7841 Logout {respond} -> respond {ok: False {}, authorized: False {}}

8037 Logout {respond} -> respond {ok: False {}, authorized: False {}}

8038+ -- Client routes

7842 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}

8039 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}

7843 SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded"}

8040 SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded"}

7844 MessageStatus {respond} -> respond {status: "", authorized: False {}}

8041 MessageStatus {respond} -> respond {status: "", authorized: False {}}

@@ -7850,6 +8047,7 @@ serveOverloaded = \req -> case req of

7850 CancelBatch {respond} -> respond {results: [], authorized: False {}, error: "overloaded"}

8047 CancelBatch {respond} -> respond {results: [], authorized: False {}, error: "overloaded"}

7851 LookupPresence {respond} -> respond {server: "", authorized: False {}, error: "overloaded"}

8048 LookupPresence {respond} -> respond {server: "", authorized: False {}, error: "overloaded"}

7852 BatchLookupPresence {respond} -> respond {results: [], authorized: False {}, error: "overloaded"}

8049 BatchLookupPresence {respond} -> respond {results: [], authorized: False {}, error: "overloaded"}

8050+ -- Federation routes

7853 RecvGossip {respond} -> respond {ok: False {}, error: "overloaded:1000ms"}

8051 RecvGossip {respond} -> respond {ok: False {}, error: "overloaded:1000ms"}

7854 ForwardMessage {respond} -> respond {ok: False {}, error: "overloaded:1000ms"}

8052 ForwardMessage {respond} -> respond {ok: False {}, error: "overloaded:1000ms"}

7855 FederationAck {respond} -> respond {ok: False {}, error: "overloaded:1000ms", ackedCount: 0, results: []}

8053 FederationAck {respond} -> respond {ok: False {}, error: "overloaded:1000ms", ackedCount: 0, results: []}

@@ -7857,8 +8055,10 @@ serveOverloaded = \req -> case req of

7857 FederationBatchCancel {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}

8055 FederationBatchCancel {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}

7858 FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}

8056 FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}

7859 VerifyToken {respond} -> respond {ok: False {}}

8057 VerifyToken {respond} -> respond {ok: False {}}

8058+ -- Stats/admin routes (exempt routes handled by isOverloadExempt; these are fallbacks)

7860 TopRecipients {respond} -> respond {recipients: [], ok: False {}, total: 0}

8059 TopRecipients {respond} -> respond {recipients: [], ok: False {}, total: 0}

7861 StatsForwards {respond} -> respond {servers: [], ok: False {}, total: 0}

8060 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}

7862 BlockPeer {respond} -> respond {ok: False {}, blocked: 0}

8062 BlockPeer {respond} -> respond {ok: False {}, blocked: 0}

7863 UnblockPeer {respond} -> respond {ok: False {}, unblocked: False {}}

8063 UnblockPeer {respond} -> respond {ok: False {}, unblocked: False {}}

7864 ListBlocked {respond} -> respond {peers: [], ok: False {}}

8064 ListBlocked {respond} -> respond {peers: [], ok: False {}}

@@ -7934,14 +8134,15 @@ serve = \req -> do

7934 _ -> yield {}

8134 _ -> yield {}

7935 let path = requestPath req

8135 let path = requestPath req

7936 let ip = requestClientIp req

8136 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.

7938 if path != ""

8139 if path != ""

7939 then do

8140 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 "") ++ "}")

7941 when (latency > slowRequestThreshold)

8142 when (latency > slowRequestThreshold)

7942 (logEvent "warn" "slow_request" (path ++ " " ++ show latency ++ "ms"))

8143 (logEvent "warn" "slow_request" (path ++ " " ++ show latency ++ "ms"))

7943 else when (latency > slowRequestThreshold) (do

8144 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}")

7945 logEvent "warn" "slow_request" ("poll " ++ show latency ++ "ms"))

8146 logEvent "warn" "slow_request" ("poll " ++ show latency ++ "ms"))

7946

8147

7947-- ============================================

8148-- ============================================

@@ -8197,6 +8398,7 @@ main = do

8197 && clockSkewTolerance < challengeExpiry

8398 && clockSkewTolerance < challengeExpiry

8198 && maxMessagesPerRecipient <= maxMessages

8399 && maxMessagesPerRecipient <= maxMessages

8199 && maxForwardsPerKey <= maxForwardsEntries

8400 && maxForwardsPerKey <= maxForwardsEntries

8401+ && maxForwardsPerServer <= maxForwardsEntries

8200 && maxSessionsPerKey <= maxSessions

8402 && maxSessionsPerKey <= maxSessions

8201 && maxChallengesPerKey <= maxChallenges

8403 && maxChallengesPerKey <= maxChallenges

8202 && maxPollsPerKey <= maxActivePolls

8404 && maxPollsPerKey <= maxActivePolls

@@ -8213,7 +8415,7 @@ main = do

8213 if not configOk

8415 if not configOk

8214 then logError "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"

8416 then logError "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"

8215 else if not configRelationsOk

8417 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"

8217 else do

8419 else do

8218 t <- now

8420 t <- now

8219 atomic do

8421 atomic do

@@ -8244,7 +8446,7 @@ main = do

8244 logInfo (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")

8446 logInfo (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")

8245 logInfo (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")

8447 logInfo (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")

8246 logInfo (" max presence: " ++ show maxPresenceEntries)

8448 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 ++ ")")

8248 logInfo (" session expiry: " ++ show sessionExpiry ++ "ms (max lifetime: " ++ show maxSessionLifetime ++ "ms)")

8450 logInfo (" session expiry: " ++ show sessionExpiry ++ "ms (max lifetime: " ++ show maxSessionLifetime ++ "ms)")

8249 logInfo (" message max age: " ++ show messageMaxAge ++ "ms")

8451 logInfo (" message max age: " ++ show messageMaxAge ++ "ms")

8250 logInfo (" max active requests: " ++ if maxActiveRequests > 0 then show maxActiveRequests else "unlimited")

8452 logInfo (" max active requests: " ++ if maxActiveRequests > 0 then show maxActiveRequests else "unlimited")