@@ -1212,6 +1212,14 @@ logError = \msg -> println ("[ERROR] " ++ msg)
12121212-- when msg contains user-controlled data (e.g., error strings, server names).
12131213logEvent = \level event msg -> println ("{\"level\":\"" ++ escapeJsonString level ++ "\",\"event\":\"" ++ escapeJsonString event ++ "\",\"msg\":\"" ++ escapeJsonString msg ++ "\"}")
12141214
1215+-- Structured key=value logging: formats message as "k1=v1 k2=v2 ..." for consistent
1216+-- machine-parseable log entries. Wraps logEvent with key-value pair formatting.
1217+-- Use for events where individual fields need to be extracted by log pipelines.
1218+logKV = \level event kvs ->
1219+ logEvent level event (joinKVPairs kvs)
1220+
1221+joinKVPairs = \kvs -> fold (\acc kv -> if acc == "" then kv else acc ++ " " ++ kv) "" kvs
1222+
12151223-- Log a warning when a federation peer sends a request with an older protocol version.
12161224-- protocolVersion == 0 means pre-versioning peer (field missing). Only logs once per
12171225-- request — callers invoke this before delegating to the handler.
@@ -4025,6 +4033,9 @@ ackCleanupForwards = \entries t ->
40254033 let bucketIds = uniqueBucketsFor entries (\e -> e.toKey) in
40264034 ackCleanupByBucket bucketIds entries t
40274035
4036+-- Two-pass ack cleanup: first filter entries to those eligible for cleanup,
4037+-- then single-pass filter messages. Avoids O(entries * messages) nested scan
4038+-- from the fold-with-shrinking-accumulator pattern.
40284039ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of
40294040 Nothing {} -> yield {}
40304041 Just {value: b} -> do
@@ -4037,17 +4048,20 @@ ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of
40374048 msgs <- readMessageBucket b
40384049 tc <- *totalMessageCount
40394050 rc <- readRcBucket b
4040− let cleanup = fold (\acc rf ->
4041− let hasLocal = any bucketPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in
4042− let hasOtherFwds = any fwdBkt (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey) in
4043− if hasLocal || hasOtherFwds then acc
4044− else case headOf (do m <- acc.msgs; where m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod; yield m) of
4045− Nothing {} -> acc
4046− Just {} -> {msgs: (do m <- acc.msgs; where not (m.messageId == rf.messageId && m.toKey == rf.toKey); yield m), removed: acc.removed + 1, rc: adjustRecipientCount acc.rc rf.toKey (0 - 1)}
4047− ) {msgs: msgs, removed: 0, rc: rc} bucketEntries
4048− writeMessageBucket b cleanup.msgs
4049− set *totalMessageCount = max 0 (tc - cleanup.removed)
4050− writeRcBucket b cleanup.rc
4051+ -- Phase 1: filter to entries eligible for cleanup (no local presence, no pending forwards)
4052+ let eligible = do rf <- bucketEntries
4053+ where not (any bucketPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t))
4054+ where not (any fwdBkt (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey))
4055+ yield rf
4056+ -- Phase 2: single pass over messages — remove those matching any eligible entry past grace period
4057+ let shouldRemove = \m -> any eligible (\rf -> m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod)
4058+ let cleanMsgs = do m <- msgs; where not (shouldRemove m); yield m
4059+ let removedMsgs = do m <- msgs; where shouldRemove m; yield m
4060+ let removedCount = count removedMsgs
4061+ let newRc = fold (\accRc m -> adjustRecipientCount accRc m.toKey (0 - 1)) rc removedMsgs
4062+ writeMessageBucket b cleanMsgs
4063+ set *totalMessageCount = max 0 (tc - removedCount)
4064+ writeRcBucket b newRc
40514065 ackCleanupByBucket remainingBuckets entries t
40524066
40534067pow2 : Int -> Int
@@ -4088,12 +4102,18 @@ isFederationCircuitOpen = \failures t ->
40884102
40894103recordFailure = \server -> do
40904104 t <- now
4105+ -- Random jitter (0-25% of base backoff) prevents synchronized retries when
4106+ -- multiple operations to the same peer fail in quick succession. Shifts failedAt
4107+ -- forward, delaying when shouldRetryServer allows the next attempt. Combined with
4108+ -- the deterministic per-server spread in retryBackoffFor, this breaks both
4109+ -- cross-server and same-server retry correlation (thundering herd).
4110+ jitter <- randomInt (retryBackoffBase / 4)
40914111 atomic do
40924112 failures <- *forwardFailures
40934113 let existing = headOf (do f <- failures; where f.server == server; yield f)
40944114 let updated = case existing of
4095− Nothing {} -> union failures [{server: server, failedAt: t, failures: 1, firstFailedAt: t}]
4096− Just {value: f} -> union (do x <- failures; where x.server != server; yield x) [{server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}]
4115+ Nothing {} -> union failures [{server: server, failedAt: t + jitter, failures: 1, firstFailedAt: t}]
4116+ Just {value: f} -> union (do x <- failures; where x.server != server; yield x) [{server: server, failedAt: t + jitter, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}]
40974117 set *forwardFailures = if count updated > maxForwardFailures then evictOneOldest updated (\x -> x.failedAt) else updated
40984118
40994119clearFailure = \server -> do
@@ -4113,7 +4133,7 @@ incrementForwardRetries = \messageId toServer toKey -> atomic do
41134133 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toServer == toServer); yield f)
41144134 tfc <- *totalForwardCount
41154135 set *totalForwardCount = max 0 (tfc - 1)
4116− logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
4136+ logKV "warn" "federation_forward_dropped" ["messageId=" ++ messageId, "server=" ++ toServer, "retries=" ++ show maxForwardRetries]
41174137 else writeFwdBucket b (do f <- fwdBkt; yield (if f.messageId == messageId && f.toServer == toServer then {messageId: f.messageId, toServer: f.toServer, createdAt: f.createdAt, toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, retries: f.retries + 1, expiresAt: f.expiresAt, priority: f.priority, senderKey: f.senderKey} else f))
41184138
41194139-- Retry forwards for a server using batch forward (1 HTTP round-trip per batch of maxBatchSize).
@@ -4141,7 +4161,7 @@ retryForwardsForServer = \entries remaining ->
41414161 requestTimestamp: t, authorization: auth, correlationId: corrId, protocolVersion: federationProtocolVersion})
41424162 case result of
41434163 Err e -> do
4144− logWarn ("federation: batch forward to " ++ server ++ " failed: " ++ show e)
4164+ logKV "warn" "federation_batch_forward_failed" ["server=" ++ server, "batchSize=" ++ show (count batch), "error=" ++ show e]
41454165 recordFailure server
41464166 -- HTTP-level failure (connectivity, timeout): don't increment per-message retries.
41474167 -- The messages weren't rejected — the server was unreachable. Server backoff
@@ -4154,7 +4174,7 @@ retryForwardsForServer = \entries remaining ->
41544174 -- Process per-message results: remove succeeded, increment retries for failed
41554175 retryForwardsBatchResults batch response.results server
41564176 else do
4157− logWarn ("federation: batch forward to " ++ server ++ " rejected: " ++ response.error ++ ", falling back to single-message forwards")
4177+ logKV "warn" "federation_batch_forward_rejected" ["server=" ++ server, "error=" ++ response.error]
41584178 -- Application-level batch rejection (old peer, invalid request): fall back to
41594179 -- sequential single-message forwards instead of burning retries. The peer likely
41604180 -- doesn't support /federation/forward-batch, so individual forwards may succeed.
@@ -4188,7 +4208,7 @@ retryForwardsSingle = \batch server ->
41884208 expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})
41894209 case result of
41904210 Err e -> do
4191− logWarn ("federation: single forward to " ++ server ++ " failed for " ++ f.messageId ++ ": " ++ show e)
4211+ logKV "warn" "federation_single_forward_failed" ["server=" ++ server, "messageId=" ++ f.messageId, "error=" ++ show e]
41924212 recordFailure server
41934213 Ok response ->
41944214 if response.ok then do
@@ -5499,9 +5519,15 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
54995519 let nb = pubkeyBucket to
55005520 fwdNonces <- readNonceBucket nb
55015521 let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, seenAt: t}]
5502− writeNonceBucket nb (if count updatedNonces > maxNoncesPerBucket
5503− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5504− else updatedNonces)
5522+ let evictedNonces = if count updatedNonces > maxNoncesPerBucket
5523+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5524+ else updatedNonces
5525+ -- Log LRU nonce eviction: evicted nonces create a bounded replay window
5526+ -- where messages with those nonces could be re-accepted until natural expiry.
5527+ let lruEvicted = count updatedNonces - count evictedNonces
5528+ when (lruEvicted > 0)
5529+ (logKV "warn" "nonce_lru_eviction" ["bucket=" ++ show nb, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket])
5530+ writeNonceBucket nb evictedNonces
55055531 let toPresence = presenceForKey recipientPresence to t
55065532 let localOnline = hasLocalPresence toPresence
55075533 let remoteServers = collectPeerServers toPresence (\p -> p.server)
@@ -5547,14 +5573,14 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
55475573 expiresAt: fwdInfo.fwdExpiresAt, priority: fwdInfo.fwdPriority, protocolVersion: federationProtocolVersion})
55485574 case fwdResult of
55495575 Err e -> do
5550− logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
5576+ logKV "warn" "federation_forward_failed" ["server=" ++ server, "messageId=" ++ effectiveMsgId, "error=" ++ show e]
55515577 recordFailure server
55525578 Ok response ->
55535579 if response.ok then do
55545580 removeForward effectiveMsgId server to
55555581 clearFailure server
55565582 else
5557− logWarn ("federation: forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))
5583+ logKV "warn" "federation_forward_rejected" ["server=" ++ server, "messageId=" ++ effectiveMsgId, "error=" ++ response.error]))))
55585584
55595585-- Send all batch forwards for one server sequentially (no per-message fork overhead).
55605586-- Used by handleSendBatch; reduces fork count from O(messages * servers) to O(servers).
@@ -5581,14 +5607,14 @@ sendBatchForwardsSequential = \msgs server ->
55815607 expiresAt: r.fwdExpiresAt, priority: r.fwdPriority, protocolVersion: federationProtocolVersion})
55825608 case fwdResult of
55835609 Err e -> do
5584− logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
5610+ logKV "warn" "federation_forward_failed" ["server=" ++ server, "messageId=" ++ r.messageId, "error=" ++ show e]
55855611 recordFailure server
55865612 Ok response ->
55875613 if response.ok then do
55885614 removeForward r.messageId server r.toKey
55895615 clearFailure server
55905616 else
5591− logWarn ("federation: forward to " ++ server ++ " rejected for " ++ r.messageId ++ ": " ++ response.error))
5617+ logKV "warn" "federation_forward_rejected" ["server=" ++ server, "messageId=" ++ r.messageId, "error=" ++ response.error])
55925618 sendBatchForwardsSequential rest server
55935619
55945620-- POST /messages: Send an encrypted message to a recipient.
@@ -6092,9 +6118,13 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
60926118 -- Sharded: writes only to the bucket for this recipient's pubkey.
60936119 fwdNonces <- readNonceBucket bkt
60946120 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, seenAt: t}]
6095− writeNonceBucket bkt (if count updatedNonces > maxNoncesPerBucket
6096− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6097− else updatedNonces)
6121+ let evictedNonces = if count updatedNonces > maxNoncesPerBucket
6122+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6123+ else updatedNonces
6124+ let lruEvicted = count updatedNonces - count evictedNonces
6125+ when (lruEvicted > 0)
6126+ (logKV "warn" "nonce_lru_eviction" ["bucket=" ++ show bkt, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket])
6127+ writeNonceBucket bkt evictedNonces
60986128 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
60996129 if result.noPresence then do
61006130 logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
@@ -7922,9 +7952,12 @@ requestPath = \req -> case req of
79227952 StatsKey {} -> "GET /stats/key"
79237953 TopRecipients {} -> "POST /stats/top-recipients"
79247954 StatsForwards {} -> "GET /stats/forwards"
7955+ StatsRates {} -> "GET /stats/rates"
79257956 Ready {} -> "GET /ready"
79267957
7927−-- Extract client IP for access logging (empty for federation/maintenance routes)
7958+-- Extract client IP for access logging (empty for federation/maintenance/stats routes).
7959+-- Client-facing routes extract clientIp; all others use catch-all to avoid enumerating
7960+-- every federation/maintenance/stats route (reduces maintenance burden on new routes).
79287961requestClientIp = \req -> case req of
79297962 ReqChallenge {clientIp} -> clientIp
79307963 VerifyAuth {clientIp} -> clientIp
@@ -7942,40 +7975,7 @@ requestClientIp = \req -> case req of
79427975 LookupPresence {clientIp} -> clientIp
79437976 BatchLookupPresence {clientIp} -> clientIp
79447977 VerifyToken {clientIp} -> clientIp
7945− RecvGossip {} -> ""
7946− ForwardMessage {} -> ""
7947− FederationAck {} -> ""
7948− FederationCancel {} -> ""
7949− FederationBatchCancel {} -> ""
7950− FederationBatchForward {} -> ""
7951− Prune {} -> ""
7952− Drain {} -> ""
7953− Shutdown {} -> ""
7954− Undrain {} -> ""
7955− BlockPeer {} -> ""
7956− UnblockPeer {} -> ""
7957− ListBlocked {} -> ""
7958− PingPeer {} -> ""
7959− AdminListSessions {} -> ""
7960− AdminRevokeSessions {} -> ""
7961− AdminListAllSessions {} -> ""
7962− AdminPurgeMessages {} -> ""
7963− AdminRevokeIp {} -> ""
7964− GetRateLimits {} -> ""
7965− SetRateLimits {} -> ""
7966− FlushNonces {} -> ""
7967− AllowPeer {} -> ""
7968− UnallowPeer {} -> ""
7969− ListAllowed {} -> ""
7970− Reconcile {} -> ""
7971− RotateSecret {} -> ""
7972− Health {} -> ""
7973− Stats {} -> ""
7974− StatsPeers {} -> ""
7975− StatsKey {} -> ""
7976− TopRecipients {} -> ""
7977− StatsForwards {} -> ""
7978− Ready {} -> ""
7978+ _ -> ""
79797979
79807980-- Slow request threshold for access logging (ms)
79817981slowRequestThreshold : Int<Ms>
@@ -8295,9 +8295,21 @@ backgroundPrune = do
82958295
82968296-- Background loop: periodically bumps pollWakeup to unblock long-polls that have been
82978297-- waiting too long (heartbeat). Ensures polls return within pollTimeoutMs even if no messages arrive.
8298+-- Adaptive: under high load (>80% capacity), the heartbeat fires more frequently
8299+-- (down to 5s at peak) to release poll slots faster and reduce connection hold time.
8300+-- Normal load uses the full pollTimeoutMs (~25s).
82988301pollHeartbeat = do
8302+ -- Adaptive poll timeout: scale heartbeat interval based on server load.
8303+ -- adaptiveRateScale returns 10-100 (percentage). At 100% (normal): full pollTimeoutMs.
8304+ -- At 10% (peak load): max(5000, pollTimeoutMs * 10 / 100) = 5000ms.
8305+ scale <- atomic do
8306+ tmc <- *totalMessageCount
8307+ sc <- *totalSessionCount
8308+ pc <- *totalPresenceCount
8309+ yield (adaptiveRateScale tmc sc pc)
8310+ let effectiveTimeout = max 5000 (pollTimeoutMs * scale / 100)
82998311 jitter <- randomInt 2000
8300− sleep (pollTimeoutMs + jitter)
8312+ sleep (effectiveTimeout + jitter)
83018313 dp <- *drainPhase
83028314 shutdownReq <- *shutdownRequested
83038315 dca <- *drainCompletedAt