@@ -313,6 +313,11 @@ healthHeartbeatMaxAge : Int<Ms>
313313healthHeartbeatMaxAge = pollTimeoutMs * 3
314314healthWatchdogMaxAge : Int<Ms>
315315healthWatchdogMaxAge = backgroundPruneInterval * 15 -- watchdog runs every ~5 prune intervals
316+-- Forward queue age health threshold: flag unhealthy when the oldest pending forward
317+-- is older than this. Indicates a destination server is unreachable and forward entries
318+-- are accumulating without being delivered or retried successfully.
319+healthForwardQueueMaxAge : Int<Ms>
320+healthForwardQueueMaxAge = retryBackoffMax * 3 -- ~24 minutes; 3x max backoff gives retries time to succeed
316321
317322-- ============================================
318323-- Types
@@ -672,6 +677,11 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
672677-- phase 1 lets active clients finish while preventing new connections.
673678*drainPhase : Int
674679
680+-- Drain generation: monotonically increasing counter, bumped on each drain initiation.
681+-- Flush forks check this before writing to prevent stale drain operations (started before
682+-- an undrain+redrain cycle) from interfering with the current drain's state.
683+*drainGeneration : Int
684+
675685-- Shutdown requested: when True, background loops exit after drain completes.
676686-- Set by /maintenance/shutdown; cannot be undone (requires process restart).
677687*shutdownRequested : Bool
@@ -885,6 +895,7 @@ maxPeerLatencyEntries = 100
885895*cachedHealthWatchdog : Bool
886896*cachedHealthNonce : Bool
887897*cachedHealthCapacityWarning : Bool
898+*cachedHealthForwardQueue : Bool
888899-- Cached count of federation peers with active failure records (degraded connectivity)
889900*cachedHealthDegradedPeers : Int
890901
@@ -996,6 +1007,11 @@ route FederationApi where
9961007 -- Cancel (retract) a previously forwarded message. The origin server sends this
9971008 -- when the sender cancels a message via /messages/cancel. The receiving server
9981009 -- deletes the message if it hasn't been polled by the recipient yet.
1010+ -- Security: the handler checks msg.originServer == fromServer, and fromServer is
1011+ -- authenticated via the verify callback (the receiver calls back to fromServer to
1012+ -- confirm the token). A relay peer cannot forge fromServer because the callback
1013+ -- would go to the wrong server, which wouldn't have the token. This prevents a
1014+ -- malicious relay from cancelling messages it forwarded on behalf of the origin.
9991015 POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /cancel headers {authorization: Text}
10001016 -> {ok: Bool, error: Text, cancelled: Bool}
10011017 = FederationCancel
@@ -1114,7 +1130,7 @@ route MaintenanceApi where
11141130 = RotateSecret
11151131
11161132route HealthApi where
1117− GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, draining: Bool, degradedPeers: Int} = Health
1133+ GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, forwardQueueHealthy: Bool, draining: Bool, degradedPeers: Int} = Health
11181134
11191135-- Returns counts of in-memory collections (requires maintenance secret)
11201136-- TODO: If Knot adds raw text response support, add a /metrics endpoint
@@ -1137,6 +1153,11 @@ route StatsApi where
11371153 POST {limit: Int} /stats/top-recipients headers {authorization: Text}
11381154 -> {recipients: [RecipientCount], ok: Bool, total: Int}
11391155 = TopRecipients
1156+ -- Per-server forward queue breakdown: shows pending forwards grouped by destination server.
1157+ -- Use to identify which peers have queue buildup (unreachable or slow peers).
1158+ GET /stats/forwards headers {authorization: Text}
1159+ -> {servers: [{server: Text, count: Int, oldestAge: Int, totalRetries: Int}], ok: Bool, total: Int}
1160+ = StatsForwards
11401161
11411162-- Readiness probe: returns false during startup warmup (before first prune completes),
11421163-- true once the server is fully operational. Unlike /health, this is unauthenticated
@@ -1885,6 +1906,27 @@ writeAllSessions = \sessions -> do
18851906 writeSessionBucket 14 partitioned.b14
18861907 writeSessionBucket 15 partitioned.b15
18871908
1909+-- Evict the single oldest session across all 16 buckets, writing only the affected bucket.
1910+-- Two-pass: pass 1 finds the bucket containing the global minimum expiresAt,
1911+-- pass 2 removes the oldest entry from that bucket. Avoids readAllSessions + writeAllSessions.
1912+evictOldestSessionFromBuckets : Int -> Maybe {bucketIdx: Int, minExpiry: Timestamp} -> IO ()
1913+evictOldestSessionFromBuckets = \b candidate ->
1914+ if b > 15
1915+ then case candidate of
1916+ Nothing {} -> yield {}
1917+ Just {value: c} -> do
1918+ bucket <- readSessionBucket c.bucketIdx
1919+ writeSessionBucket c.bucketIdx (evictOneOldest bucket (\s -> s.expiresAt))
1920+ else do
1921+ bucket <- readSessionBucket b
1922+ let bucketMin = minBy bucket (\s -> s.expiresAt)
1923+ let newCandidate = case bucketMin of
1924+ Nothing {} -> candidate
1925+ Just {value: s} -> case candidate of
1926+ Nothing {} -> Just {bucketIdx: b, minExpiry: s.expiresAt}
1927+ Just {value: c} -> if s.expiresAt < c.minExpiry then Just {bucketIdx: b, minExpiry: s.expiresAt} else candidate
1928+ evictOldestSessionFromBuckets (b + 1) newCandidate
1929+
18881930-- ============================================
18891931-- Presence Bucket Helpers
18901932-- ============================================
@@ -2208,12 +2250,12 @@ pruneMessageBucket = \b t -> atomic do
22082250 let combined = union se perKeyCapped
22092251 let overCap = count combined > maxSenderExpirations
22102252 if overCap then do
2211− let evicted = extractOldest combined (\e -> e.expiredAt)
2212− set *senderExpirations = evicted.items
2253+ -- Batch eviction: evict 2% of capacity to amortize O(n) scan cost
2254+ let evictCount = max 1 (count combined - maxSenderExpirations + maxSenderExpirations / 50)
2255+ let eviction = evictSmallest combined evictCount (\e -> e.expiredAt)
2256+ set *senderExpirations = eviction.kept
22132257 den <- *droppedExpirationNotifs
2214− let denAfterCap = case evicted.candidate of
2215− Nothing {} -> den
2216− Just {value: ev} -> incrementDropCounter den ev.senderKey 1
2258+ let denAfterCap = fold (\acc ev -> incrementDropCounter acc ev.senderKey 1) den eviction.evicted
22172259 -- Also record per-key drops for senders that hit maxSenderExpirationsPerKey
22182260 let perKeyDroppedEntries = do e <- localExpired; where none perKeyCapped (\c -> c.messageId == e.messageId && c.toKey == e.toKey); yield e
22192261 set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1) denAfterCap perKeyDroppedEntries
@@ -2642,6 +2684,16 @@ computeForwardStatsInner = \b t acc ->
26422684 let newMaxDepth = fold (\mx s -> max mx s.count) acc.maxDepth updatedServerCounts
26432685 computeForwardStatsInner (b + 1) t {totalCount: acc.totalCount + bucketCount, oldestAge: max acc.oldestAge bucketOldest, serverCount: count updatedServerCounts, maxDepth: newMaxDepth, serverCounts: updatedServerCounts}
26442686
2687+-- Lightweight scan: find the oldest forward entry age across all 16 buckets.
2688+-- O(n) total, no allocation (just tracks max age). Used by health check.
2689+oldestForwardAgeAcrossBuckets : Int -> Timestamp -> Int -> IO Int
2690+oldestForwardAgeAcrossBuckets = \b t acc ->
2691+ if b > 15 then yield acc
2692+ else do
2693+ bucket <- readFwdBucket b
2694+ 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)
2696+
26452697-- Remove forward entries matching a server and message IDs, scanning per-bucket.
26462698-- Only rewrites buckets that actually changed (avoids writeAllForwards repartitioning).
26472699-- Returns all removed forward entries across all buckets.
@@ -3013,8 +3065,9 @@ matchIp = \clientIp sessionIp ->
30133065-- Generic sliding-window rate limiter. Returns whether the request is rate-limited,
30143066-- the updated entries list, and retryAfterMs (time until the current window resets, 0 if not limited).
30153067-- Callers provide match/create/update functions for their entry type.
3068+-- Uses a direct fold for entry lookup (avoids materializing a filtered intermediate list).
30163069checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntries t ->
3017− let entry = headOf (do r <- entries; where matchKey r; yield r) in
3070+ let entry = fold (\acc r -> case acc of Just _ -> acc; Nothing {} -> if matchKey r then Just {value: r} else acc) Nothing {} entries in
30183071 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= windowMs in
30193072 let rateCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count in
30203073 if rateCount >= maxRate
@@ -3128,13 +3181,17 @@ evictByCutoff = \items cutoff getKey ->
31283181 do x <- items; where getKey x >= cutoff; yield x
31293182
31303183-- Evict nonces with fallback: first try expired-nonce eviction (O(n)), then if still
3131−-- over capacity, evict oldest 1% via evictSmallest (small k keeps O(n*k) bounded).
3184+-- over capacity, evict oldest entries to create 5% headroom via evictSmallest.
3185+-- Headroom prevents re-triggering eviction on the next insert: by evicting down to 95%
3186+-- of maxCount, we amortize the O(n*k) evictSmallest cost across ~5% of capacity inserts
3187+-- instead of paying it on every single insert when at the boundary.
31323188-- Minimizes replay window: expired nonces are removed first (no window), and the
3133−-- 1% fallback creates a window of only ~1000 entries vs ~10000 with the old 10%.
3189+-- headroom-based fallback creates a bounded window proportional to the eviction count.
31343190evictNonces = \items maxCount retention t ->
31353191 let afterExpired = evictByCutoff items (t - retention) (\n -> n.seenAt) in
31363192 if count afterExpired <= maxCount then afterExpired
3137− else let evictCount = max 1 (count afterExpired / 100) in
3193+ else let headroom = max 1 (maxCount / 20) in -- 5% headroom
3194+ let evictCount = count afterExpired - maxCount + headroom in
31383195 (evictSmallest afterExpired evictCount (\n -> n.seenAt)).kept
31393196
31403197-- Enforce per-key capacity: if a key has >= maxPerKey items, evict the oldest for that key.
@@ -3225,12 +3282,18 @@ hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey
32253282
32263283-- Validate incoming gossip events: reject local-server events, future timestamps,
32273284-- and already-expired entries. Clamp TTLs and split into online/offline buckets.
3285+-- Timestamp clamping: compute expiresAt from min(timestamp, now) instead of raw timestamp.
3286+-- A peer can send timestamp close to t + clockSkewTolerance (within tolerance), which would
3287+-- inflate effective presence lifetime by up to clockSkewTolerance ms. Clamping to now removes
3288+-- this inflation: the peer's declared TTL is honored, but the expiry clock starts from now,
3289+-- not from a future timestamp.
32283290validGossipEvents : [GossipEvent] -> Timestamp -> {online: [ValidGossipEntry], offline: [ValidGossipEntry]}
32293291validGossipEvents = \events t ->
32303292 let validate = \e ->
32313293 if not (isLocalServer e.server) && e.timestamp <= t + clockSkewTolerance
32323294 then let ttl = clampTtl e.ttl in
3233− let expiresAt = e.timestamp + ttl * 1000 in
3295+ let effectiveTs = min e.timestamp t in
3296+ let expiresAt = effectiveTs + ttl * 1000 in
32343297 if expiresAt > t
32353298 then [{pubkey: e.pubkey, server: e.server, expiresAt: expiresAt, timestamp: e.timestamp, ttl: ttl}]
32363299 else []
@@ -3740,7 +3803,7 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
37403803 globalLimited <- checkGlobalRate t
37413804 if globalLimited
37423805 then do
3743− logWarn ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)
3806+ logEvent "warn" "rate_limited_federation_global" ("peer=" ++ fromServer ++ " limit=" ++ show maxGlobalRequestRate)
37443807 -- Encode retryAfterMs in error string so peers can make smarter retry decisions.
37453808 -- Format: "rate_limited:NNNms" — peers that don't parse it still see "rate_limited:...".
37463809 yield (respond {ok: False {}, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
@@ -3748,7 +3811,7 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
37483811 rateLimited <- checkFederationRate fromServer t
37493812 if rateLimited
37503813 then do
3751− logWarn ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
3814+ logEvent "warn" "rate_limited_federation_per_server" ("peer=" ++ fromServer ++ " limit=" ++ show maxFederationRatePerServer)
37523815 yield (respond {ok: False {}, error: "rate_limited:" ++ show federationRateWindowMs ++ "ms"})
37533816 else do
37543817 when clearOnAuth (clearFailure fromServer)
@@ -3931,7 +3994,7 @@ retryForwardsForServer = \entries remaining ->
39313994 let server = case headOf entries of Just {value: f} -> f.toServer; Nothing {} -> ""
39323995 forEachBatch entries maxBatchSize (\batch -> do
39333996 t <- now
3934− auth <- generateFederationAuth "forward" server
3997+ auth <- generateFederationAuth "forward-batch" server
39353998 corrId <- genTokenN 1
39363999 let batchPayload = do f <- batch; yield {toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId, expiresAt: f.expiresAt, priority: f.priority}
39374000 result <- fetchFederation server (FederationBatchForward {
@@ -3976,7 +4039,7 @@ retryForwardsSingle = \batch server ->
39764039 Just {value: f} -> do
39774040 let rest = do x <- batch; where x.messageId != f.messageId; yield x
39784041 t <- now
3979− auth <- generateFederationAuth "forward" server
4042+ auth <- generateFederationAuth ("forward:" ++ f.messageId) server
39804043 corrId <- genTokenN 1
39814044 result <- fetchFederation server (ForwardMessage {
39824045 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
@@ -4317,7 +4380,7 @@ retryPendingCancels = do
43174380 let serverCancels = do c <- pending; where c.toServer == server; yield c
43184381 forEachBatch serverCancels maxBatchSize (\batch -> do
43194382 cancelT <- now
4320− auth <- generateFederationAuth "cancel" server
4383+ auth <- generateFederationAuth "cancel-batch" server
43214384 corrId <- genTokenN 1
43224385 let payload = do c <- batch; yield {messageId: c.messageId, toKey: c.toKey}
43234386 result <- fetchFederation server (FederationBatchCancel {
@@ -4638,13 +4701,17 @@ ackPolledMessages = \pubkey cursor t -> do
46384701 let combined = union sd deduped
46394702 let overCap = count combined > maxSenderDeliveries
46404703 if overCap then do
4641− let evicted = extractOldest combined (\d -> d.deliveredAt)
4642− set *senderDeliveries = evicted.items
4643− case evicted.candidate of
4644− Nothing {} -> yield {}
4645− Just {value: ev} -> do
4646− ddr <- *droppedDeliveryReceipts
4647− set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1
4704+ -- Batch eviction: evict 2% of capacity instead of 1 entry at a time.
4705+ -- Amortizes the O(n) scan cost across multiple inserts, reducing overhead
4706+ -- under sustained capacity pressure from O(n) per insert to O(n) per 2% fills.
4707+ let evictCount = max 1 (count combined - maxSenderDeliveries + maxSenderDeliveries / 50)
4708+ let eviction = evictSmallest combined evictCount (\d -> d.deliveredAt)
4709+ set *senderDeliveries = eviction.kept
4710+ -- Track per-sender drop counts for all evicted entries
4711+ let dropCounts = fold (\acc e -> incrementDropCounter acc e.senderKey 1) [] eviction.evicted
4712+ when (count dropCounts > 0) (do
4713+ ddr <- *droppedDeliveryReceipts
4714+ set *droppedDeliveryReceipts = fold (\acc dc -> incrementDropCounter acc dc.senderKey dc.count) ddr dropCounts)
46484715 else set *senderDeliveries = combined)
46494716 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
46504717 else yield {pendingAcks: [], evictedEntries: []}
@@ -4930,7 +4997,7 @@ handleChallenge = \pubkey clientIp respond -> do
49304997 preAuth <- checkPreAuth clientIp t
49314998 if preAuth != ""
49324999 then do
4933− when (preAuth == "ip_rate_limited") (logWarn ("challenge: IP rate limited for " ++ clientIp))
5000+ when (preAuth == "ip_rate_limited") (logEvent "warn" "rate_limited_ip" ("endpoint=challenge ip=" ++ clientIp))
49345001 yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: preAuth})
49355002 else do
49365003 authLimited <- checkAuthGlobalRate t
@@ -4959,7 +5026,7 @@ handleChallenge = \pubkey clientIp respond -> do
49595026 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
49605027 yield {accepted: True {}, rateLimited: False {}, retryAfterMs: 0, error: ""}
49615028 when result.rateLimited
4962− (logWarn ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
5029+ (logEvent "warn" "rate_limited_per_key" ("endpoint=challenge key=" ++ pubkey ++ " limit=" ++ show maxChallengeRatePerKey))
49635030 when (not result.accepted && not result.rateLimited)
49645031 (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
49655032 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0, retryAfterMs: result.retryAfterMs, error: result.error})
@@ -4972,7 +5039,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
49725039 preAuth <- checkPreAuth clientIp t
49735040 if preAuth != ""
49745041 then do
4975− when (preAuth == "ip_rate_limited") (logWarn ("verify: IP rate limited for " ++ clientIp))
5042+ when (preAuth == "ip_rate_limited") (logEvent "warn" "rate_limited_ip" ("endpoint=verify ip=" ++ clientIp))
49765043 yield (respond {token: "", expiresAt: 0, error: preAuth})
49775044 else do
49785045 authRateLimited <- atomic do
@@ -4982,7 +5049,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
49825049 set *challengeRates = rateResult.entries
49835050 yield rateResult.limited
49845051 if authRateLimited then do
4985− logWarn ("verify: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey)
5052+ logEvent "warn" "rate_limited_per_key" ("endpoint=verify key=" ++ pubkey ++ " limit=" ++ show maxChallengeRatePerKey)
49865053 yield (respond {token: "", expiresAt: 0, error: "rate_limited"})
49875054 else do
49885055 authGlobalLimited <- checkAuthGlobalRate t
@@ -4997,7 +5064,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
49975064 else result.entries
49985065 yield result.limited
49995066 if uniqueKeyLimited then do
5000− logWarn ("verify: unique key per IP rate limited (" ++ show maxUniqueKeysPerIp ++ " keys/" ++ show uniqueKeyIpWindowMs ++ "ms) for IP " ++ clientIp)
5067+ logEvent "warn" "rate_limited_unique_key_ip" ("ip=" ++ clientIp ++ " limit=" ++ show maxUniqueKeysPerIp)
50015068 yield (respond {token: "", expiresAt: 0, error: "unique_key_limited"})
50025069 else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
50035070
@@ -5078,10 +5145,10 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
50785145 let perKeyEvicted = if not revokeOthers && sessionsForKey >= maxSessionsPerKey then 1 else 0
50795146 let estimatedTotal = tsc - revokedCount - perKeyEvicted
50805147 when (estimatedTotal >= maxSessions)
5081− (do -- Find bucket with oldest session and evict from it
5082− allSess <- readAllSessions
5083− let globalCapped = capGlobal allSess maxSessions (\s -> s.expiresAt)
5084− writeAllSessions globalCapped)
5148+ (do -- Find the bucket containing the oldest session and evict from just that bucket.
5149+ -- Avoids materializing all 16 session buckets into a single list (O(n) union chain).
5150+ -- Instead, reads each bucket to find the global minimum, then writes only one bucket.
5151+ evictOldestSessionFromBuckets 0 Nothing {})
50855152 currentBucket <- readSessionBucket nb
50865153 writeSessionBucket nb (union currentBucket [newSession])
50875154 -- Final count: estimatedTotal + 1 (new session) - globalEviction (0 or 1)
@@ -5282,7 +5349,7 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
52825349 Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt, fwdPriority: fe.priority}
52835350 when fwdInfo.pending (do
52845351 fwdT <- now
5285− fwdAuth <- generateFederationAuth "forward" server
5352+ fwdAuth <- generateFederationAuth ("forward:" ++ effectiveMsgId) server
52865353 fwdCorrId <- genTokenN 1
52875354 fwdResult <- fetchFederation server (ForwardMessage {
52885355 toKey: to, encryptedBlob: encryptedBlob,
@@ -5316,7 +5383,7 @@ sendBatchForwardsSequential = \msgs server ->
53165383 yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))
53175384 when stillPending (do
53185385 fwdT <- now
5319− fwdAuth <- generateFederationAuth "forward" server
5386+ fwdAuth <- generateFederationAuth ("forward:" ++ r.messageId) server
53205387 fwdCorrId <- genTokenN 1
53215388 fwdResult <- fetchFederation server (ForwardMessage {
53225389 toKey: r.toKey, encryptedBlob: r.blob,
@@ -5362,7 +5429,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
53625429 (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
53635430 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs clampedPriority t)
53645431 case result.status of
5365− MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
5432+ MsRateLimited {} -> logEvent "warn" "rate_limited_send" ("messageId=" ++ effectiveMsgId)
53665433 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (global capacity full)"))
53675434 _ -> yield {}
53685435 forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.remoteServers
@@ -5586,7 +5653,7 @@ handleVerifyToken = \token scope clientIp respond -> do
55865653 set *verifyIpRates = result.entries
55875654 yield result.limited
55885655 if verifyLimited then do
5589− logWarn ("federation/verify: rate limited for IP " ++ clientIp)
5656+ logEvent "warn" "rate_limited_verify_ip" ("ip=" ++ clientIp)
55905657 yield (respond {ok: False {}})
55915658 else do
55925659 found <- atomic do
@@ -5641,7 +5708,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
56415708 set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs; yield r) else newEntries
56425709 yield (False {})
56435710 if eventRateLimited then do
5644− logWarn ("gossip: event volume rate limited for " ++ fromServer ++ " (" ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms)")
5711+ logEvent "warn" "rate_limited_gossip_volume" ("peer=" ++ fromServer ++ " limit=" ++ show maxGossipEventsPerServer)
56455712 yield (respond {ok: False {}, error: "event_rate_limited:" ++ show gossipEventRateWindowMs ++ "ms"})
56465713 else do
56475714 let validated = validGossipEvents cappedEvents t
@@ -5738,7 +5805,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
57385805 t <- now
57395806 let corrId = sanitizeCorrelationId correlationId
57405807 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward " ++ messageId ++ " from " ++ fromServer))
5741− withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (True {}) (do
5808+ withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (True {}) (do
57425809 when (fromServer != originServer)
57435810 (logInfo ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
57445811 -- Relay origin verification: for relay forwards, verify:
@@ -5845,7 +5912,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
58455912 if result.globalCapReached
58465913 then logWarn ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
58475914 else if result.recvRateLimited
5848− then logWarn ("forward: recv rate limited (" ++ show maxRecvRatePerKey ++ "/s per key), rejecting for " ++ toKey)
5915+ then logEvent "warn" "rate_limited_recv" ("key=" ++ toKey ++ " limit=" ++ show maxRecvRatePerKey)
58495916 else logWarn ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
58505917 _ -> yield {}
58515918 yield (case result.forward of
@@ -5974,7 +6041,7 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
59746041 let corrId = sanitizeCorrelationId correlationId
59756042 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel " ++ messageId ++ " from " ++ fromServer))
59766043 let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}}
5977− withFederationAuth authorization t cancelRespond fromServer "cancel" requestTimestamp (not (isLocalServer fromServer)) (False {}) (do
6044+ withFederationAuth authorization t cancelRespond fromServer ("cancel:" ++ messageId) requestTimestamp (not (isLocalServer fromServer)) (False {}) (do
59786045 result <- atomic do
59796046 let b = pubkeyBucket toKey
59806047 bucket <- readMessageBucket b
@@ -6015,7 +6082,7 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
60156082 let corrId = sanitizeCorrelationId correlationId
60166083 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel-batch (" ++ show (count cancels) ++ " entries) from " ++ fromServer))
60176084 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: []}
6018− withFederationAuth authorization t batchRespond fromServer "cancel" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (False {}) (do
6085+ withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (False {}) (do
60196086 let capped = takeRelation maxBatchSize cancels
60206087 results <- cancelBatchInner capped fromServer t []
60216088 let cancelledCount = count (do r <- results; where r.cancelled; yield r)
@@ -6088,7 +6155,7 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp correlation
60886155 let corrId = sanitizeCorrelationId correlationId
60896156 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward-batch (" ++ show (count forwards) ++ " entries) from " ++ fromServer))
60906157 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: []}
6091− withFederationAuth authorization t batchRespond fromServer "forward" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (True {}) (do
6158+ withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (True {}) (do
60926159 let capped = takeRelation maxBatchSize forwards
60936160 -- Pre-deduplicate (messageId, toKey) pairs: a malicious peer could send 100 entries
60946161 -- for the same pair, each triggering a full atomic block (presence check, 5 bucket reads).
@@ -6249,7 +6316,7 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
62496316 let serverCancels = do c <- results.fedCancels; where any c.servers (\s -> s == server); yield {messageId: c.messageId, toKey: c.toKey}
62506317 when (count serverCancels > 0) (do
62516318 cancelT <- now
6252− cancelAuth <- generateFederationAuth "cancel" server
6319+ cancelAuth <- generateFederationAuth "cancel-batch" server
62536320 cancelCorrId <- genTokenN 1
62546321 cancelResult <- fetchFederation server (FederationBatchCancel {
62556322 cancels: serverCancels, fromServer: serverHostN,
@@ -6355,16 +6422,26 @@ handleDrain = \phase authorization respond -> do
63556422 let pendingAckCount = count flushState.pendingAcks
63566423 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
63576424 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
6358− atomic do set *drainCompletedAt = 0
6425+ -- Drain generation: capture the current generation before forking. Flush forks check
6426+ -- this before writing drainCompletedAt to prevent stale drain operations (started before
6427+ -- an undrain+redrain cycle) from interfering with the current drain's state.
6428+ thisGeneration <- atomic do
6429+ dg <- *drainGeneration
6430+ let newDg = dg + 1
6431+ set *drainGeneration = newDg
6432+ set *drainCompletedAt = 0
6433+ yield newDg
63596434 -- Fork a deadline timer: force-complete drain if flush ops hang on slow/dead peers.
63606435 -- Without this, operators waiting on drainCompletedAt > 0 could wait forever.
63616436 fork (do
63626437 sleep drainTimeoutMs
63636438 tTimeout <- now
6364− completed <- atomic do dc <- *drainCompletedAt; yield dc
6365− when (completed == 0) (do
6366− logWarn ("drain: timeout after " ++ show drainTimeoutMs ++ "ms, force-completing")
6367− atomic do set *drainCompletedAt = tTimeout))
6439+ atomic do
6440+ dg <- *drainGeneration
6441+ dc <- *drainCompletedAt
6442+ when (dg == thisGeneration && dc == 0) (do
6443+ set *drainCompletedAt = tTimeout)
6444+ when (True {}) (logWarn ("drain: timeout after " ++ show drainTimeoutMs ++ "ms, force-completing")))
63686445 fork (do
63696446 -- Best-effort flush of pending forwards and acks before stopping
63706447 when (federationEnabled && pendingForwardCount > 0) (do
@@ -6379,10 +6456,11 @@ handleDrain = \phase authorization respond -> do
63796456 -- Wait for in-flight federation requests to complete (timeout: 2x fetch timeout)
63806457 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))
63816458 t2 <- now
6382− -- Only set if deadline timer hasn't already force-completed
6459+ -- Only set if deadline timer hasn't already force-completed and drain generation matches
63836460 atomic do
6461+ dg <- *drainGeneration
63846462 dc <- *drainCompletedAt
6385− when (dc == 0) (set *drainCompletedAt = t2))
6463+ when (dg == thisGeneration && dc == 0) (set *drainCompletedAt = t2))
63866464 logInfo ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
63876465 when (pendingForwardCount > 0)
63886466 (logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))
@@ -6506,6 +6584,10 @@ computeAndCacheHealth = \t -> do
65066584 let nonceOk = nonceCount * 10 <= maxForwardNonces * 9
65076585 -- Count peers with active failure records (degraded federation connectivity)
65086586 let degradedPeerCount = count (do f <- forwardFailures; where t - f.firstFailedAt < maxServerDowntime; yield f)
6587+ -- Forward queue age health: find oldest pending forward across all buckets.
6588+ -- Healthy when no forward is older than healthForwardQueueMaxAge (or no forwards exist).
6589+ fwdOldestAge <- oldestForwardAgeAcrossBuckets 0 t 0
6590+ let fwdQueueOk = fwdOldestAge <= healthForwardQueueMaxAge
65096591 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
65106592 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
65116593 capacityOk: capacityOk,
@@ -6513,9 +6595,10 @@ computeAndCacheHealth = \t -> do
65136595 federationOk: fedOk,
65146596 watchdogOk: wdOk,
65156597 nonceOk: nonceOk,
6598+ forwardQueueOk: fwdQueueOk,
65166599 isDraining: dp > 0,
65176600 degradedPeers: degradedPeerCount}
6518− let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk
6601+ let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk && result.forwardQueueOk
65196602 atomic do
65206603 set *cachedHealthOk = allOk
65216604 set *cachedHealthPrune = result.pruneOk
@@ -6525,6 +6608,7 @@ computeAndCacheHealth = \t -> do
65256608 set *cachedHealthFederation = result.federationOk
65266609 set *cachedHealthWatchdog = result.watchdogOk
65276610 set *cachedHealthNonce = result.nonceOk
6611+ set *cachedHealthForwardQueue = result.forwardQueueOk
65286612 set *cachedHealthDegradedPeers = result.degradedPeers
65296613 yield result
65306614
@@ -6544,15 +6628,16 @@ handleHealth = \authorization respond -> do
65446628 cF <- *cachedHealthFederation
65456629 cW <- *cachedHealthWatchdog
65466630 cN <- *cachedHealthNonce
6631+ cFQ <- *cachedHealthForwardQueue
65476632 cCW <- *cachedHealthCapacityWarning
65486633 cDP <- *cachedHealthDegradedPeers
65496634 dp <- *drainPhase
6550− yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, capacityWarning: cCW, federationHealthy: cF, watchdogHealthy: cW, nonceHealthy: cN, draining: dp > 0, degradedPeers: cDP}
6635+ yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, capacityWarning: cCW, federationHealthy: cF, watchdogHealthy: cW, nonceHealthy: cN, forwardQueueHealthy: cFQ, draining: dp > 0, degradedPeers: cDP}
65516636 yield (respond cached)
65526637 else do
65536638 isAuthed <- verifyMaintenanceAuth authorization
65546639 result <- computeAndCacheHealth t
6555− let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk
6640+ let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk && result.forwardQueueOk
65566641 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
65576642 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
65586643 -- Draining status is always visible (load balancers need it for routing decisions).
@@ -6564,6 +6649,7 @@ handleHealth = \authorization respond -> do
65646649 federationHealthy: result.federationOk,
65656650 watchdogHealthy: result.watchdogOk,
65666651 nonceHealthy: result.nonceOk,
6652+ forwardQueueHealthy: result.forwardQueueOk,
65676653 draining: result.isDraining,
65686654 degradedPeers: result.degradedPeers})
65696655 else yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: True {},
@@ -6573,6 +6659,7 @@ handleHealth = \authorization respond -> do
65736659 federationHealthy: True {},
65746660 watchdogHealthy: True {},
65756661 nonceHealthy: True {},
6662+ forwardQueueHealthy: True {},
65766663 draining: result.isDraining,
65776664 degradedPeers: 0})
65786665
@@ -6751,6 +6838,37 @@ handleTopRecipients = \limit authorization respond -> do
67516838 yield (respond {recipients: top, ok: True {}, total: totalRc})
67526839 else yield (respond {recipients: [], ok: False {}, total: 0})
67536840
6841+-- GET /stats/forwards: Per-server forward queue breakdown.
6842+-- Shows pending forwards grouped by destination server with oldest age and total retries.
6843+-- Use to identify which peers have queue buildup (unreachable or slow peers).
6844+handleStatsForwards = \authorization respond -> do
6845+ t <- now
6846+ globalRateLimited <- checkGlobalRate t
6847+ if globalRateLimited then yield (respond {servers: [], ok: False {}, total: 0})
6848+ else do
6849+ authorized <- verifyMaintenanceAuth authorization
6850+ if authorized
6851+ then do
6852+ fwdStats <- computeForwardStats 0 t
6853+ -- Enrich per-server counts with oldest age and total retries
6854+ let enriched = do sc <- fwdStats.serverCounts; yield {server: sc.server, count: sc.count, oldestAge: 0, totalRetries: 0}
6855+ enrichedStats <- enrichForwardServerStats enriched 0 t
6856+ yield (respond {servers: enrichedStats, ok: True {}, total: fwdStats.totalCount})
6857+ else yield (respond {servers: [], ok: False {}, total: 0})
6858+
6859+-- Enrich per-server forward stats with oldest age and total retries.
6860+-- Scans one forward bucket at a time, accumulating per-server metrics.
6861+enrichForwardServerStats = \servers b t ->
6862+ if b > 15 then yield servers
6863+ else do
6864+ bucket <- readFwdBucket b
6865+ let updated = do s <- servers; yield (
6866+ let serverFwds = do f <- bucket; where f.toServer == s.server; yield f in
6867+ let bucketOldest = case minBy serverFwds (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt in
6868+ let bucketRetries = fold (\acc f -> acc + f.retries) 0 serverFwds in
6869+ {server: s.server, count: s.count, oldestAge: max s.oldestAge bucketOldest, totalRetries: s.totalRetries + bucketRetries})
6870+ enrichForwardServerStats updated (b + 1) t
6871+
67546872-- GET /ready: Readiness probe — returns true once the server has completed startup.
67556873-- Unlike /health, this is unauthenticated and doesn't check subsystem liveness.
67566874-- Returns false until the first prune cycle completes (server fully initialized).
@@ -7030,7 +7148,7 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
70307148 fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)
70317149 (sendToEach cancelResult.cancelServers (\server -> do
70327150 cancelT <- now
7033− cancelAuth <- generateFederationAuth "cancel" server
7151+ cancelAuth <- generateFederationAuth "cancel-batch" server
70347152 cancelCorrId <- genTokenN 1
70357153 cancelFedResult <- fetchFederation server (FederationBatchCancel {
70367154 cancels: [{messageId: messageId, toKey: to}], fromServer: serverHostN,
@@ -7149,7 +7267,7 @@ logBatchResults = \results ->
71497267 Just {value: r} -> do
71507268 let rest = do x <- results; where x != r; yield x
71517269 case r.status of
7152− MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ r.messageId)
7270+ MsRateLimited {} -> logEvent "warn" "rate_limited_send" ("messageId=" ++ r.messageId ++ " batch=true")
71537271 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ r.messageId ++ " (capacity or mailbox full)")
71547272 _ -> yield {}
71557273 logBatchResults rest
@@ -7580,6 +7698,7 @@ requestPath = \req -> case req of
75807698 StatsPeers {} -> "GET /stats/peers"
75817699 StatsKey {} -> "GET /stats/key"
75827700 TopRecipients {} -> "POST /stats/top-recipients"
7701+ StatsForwards {} -> "GET /stats/forwards"
75837702 Ready {} -> "GET /ready"
75847703
75857704-- Extract client IP for access logging (empty for federation/maintenance routes)
@@ -7632,6 +7751,7 @@ requestClientIp = \req -> case req of
76327751 StatsPeers {} -> ""
76337752 StatsKey {} -> ""
76347753 TopRecipients {} -> ""
7754+ StatsForwards {} -> ""
76357755 Ready {} -> ""
76367756
76377757-- Slow request threshold for access logging (ms)
@@ -7693,6 +7813,7 @@ serveInner = \req -> case req of
76937813 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
76947814 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
76957815 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond
7816+ StatsForwards {authorization, respond} -> handleStatsForwards authorization respond
76967817 Ready {respond} -> handleReady respond
76977818
76987819-- Backpressure: check if the server is overloaded (too many concurrent requests).
@@ -7737,6 +7858,7 @@ serveOverloaded = \req -> case req of
77377858 FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}
77387859 VerifyToken {respond} -> respond {ok: False {}}
77397860 TopRecipients {respond} -> respond {recipients: [], ok: False {}, total: 0}
7861+ StatsForwards {respond} -> respond {servers: [], ok: False {}, total: 0}
77407862 BlockPeer {respond} -> respond {ok: False {}, blocked: 0}
77417863 UnblockPeer {respond} -> respond {ok: False {}, unblocked: False {}}
77427864 ListBlocked {respond} -> respond {peers: [], ok: False {}}
@@ -7915,14 +8037,25 @@ backgroundPrune = do
79158037 af <- *activePruneForks
79168038 set *activePruneForks = af + 1
79178039 fork (do
7918− pruneAndRetry t
7919− -- Only clear running flag if this is still the latest generation
7920− -- (prevents a stuck old fork from clearing the flag after a recovery cycle started)
7921− atomic do
7922− currentGen <- *pruneGeneration
7923− when (currentGen == status.gen) (set *pruneRunning = 0)
7924− af <- *activePruneForks
7925− set *activePruneForks = max 0 (af - 1))
8040+ -- Pre-flight generation check: if a newer prune cycle was scheduled between
8041+ -- our fork and when the runtime schedules this task, skip to avoid overlapping
8042+ -- prune cycles that could double-count counter decrements.
8043+ preFlightGen <- *pruneGeneration
8044+ if preFlightGen != status.gen
8045+ then do
8046+ logWarn "backgroundPrune: stale fork detected (generation changed), skipping"
8047+ atomic do
8048+ af <- *activePruneForks
8049+ set *activePruneForks = max 0 (af - 1)
8050+ else do
8051+ pruneAndRetry t
8052+ -- Only clear running flag if this is still the latest generation
8053+ -- (prevents a stuck old fork from clearing the flag after a recovery cycle started)
8054+ atomic do
8055+ currentGen <- *pruneGeneration
8056+ when (currentGen == status.gen) (set *pruneRunning = 0)
8057+ af <- *activePruneForks
8058+ set *activePruneForks = max 0 (af - 1))
79268059 else logWarn "backgroundPrune: skipping (previous cycle still running)"
79278060 backgroundPrune
79288061
@@ -8094,6 +8227,8 @@ main = do
80948227 (logWarn "maintenanceSecret is shorter than 32 characters — 32+ character secrets recommended for production (128+ bits of entropy)")
80958228 when (maintenanceSecret != "" && federationEnabled && not (tooLong maintenanceSecret 15))
80968229 (logError "SECURITY: maintenanceSecret is weak and federation is enabled — maintenance endpoints control drain and peer blocking. Use a secret of at least 32 characters.")
8230+ when (maintenanceSecret != "")
8231+ (logWarn "SECURITY: maintenanceSecret is set as a compile-time constant. Avoid committing secrets to source control. Prefer setting maintenanceSecret to empty and using /maintenance/rotate-secret at startup via deployment scripts for zero-knowledge secret management.")
80978232 when (matchesHostnameOrSubdomain serverHostN "localhost")
80988233 (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
80998234 when (federationEnabled && isBadServerName serverHostN)