@@ -313,6 +313,11 @@ healthHeartbeatMaxAge : Int<Ms>
313healthHeartbeatMaxAge = pollTimeoutMs * 3
313healthHeartbeatMaxAge = pollTimeoutMs * 3
314healthWatchdogMaxAge : Int<Ms>
314healthWatchdogMaxAge : Int<Ms>
315healthWatchdogMaxAge = backgroundPruneInterval * 15 -- watchdog runs every ~5 prune intervals
315healthWatchdogMaxAge = 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
317-- ============================================
322-- ============================================
@@ -672,6 +677,11 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
672-- phase 1 lets active clients finish while preventing new connections.
677-- phase 1 lets active clients finish while preventing new connections.
673*drainPhase : Int
678*drainPhase : Int
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
675-- Shutdown requested: when True, background loops exit after drain completes.
685-- Shutdown requested: when True, background loops exit after drain completes.
676-- Set by /maintenance/shutdown; cannot be undone (requires process restart).
686-- Set by /maintenance/shutdown; cannot be undone (requires process restart).
677*shutdownRequested : Bool
687*shutdownRequested : Bool
@@ -885,6 +895,7 @@ maxPeerLatencyEntries = 100
885*cachedHealthWatchdog : Bool
895*cachedHealthWatchdog : Bool
886*cachedHealthNonce : Bool
896*cachedHealthNonce : Bool
887*cachedHealthCapacityWarning : Bool
897*cachedHealthCapacityWarning : Bool
898+*cachedHealthForwardQueue : Bool
888-- Cached count of federation peers with active failure records (degraded connectivity)
899-- Cached count of federation peers with active failure records (degraded connectivity)
889*cachedHealthDegradedPeers : Int
900*cachedHealthDegradedPeers : Int
@@ -996,6 +1007,11 @@ route FederationApi where
996 -- Cancel (retract) a previously forwarded message. The origin server sends this
1007 -- Cancel (retract) a previously forwarded message. The origin server sends this
997 -- when the sender cancels a message via /messages/cancel. The receiving server
1008 -- when the sender cancels a message via /messages/cancel. The receiving server
998 -- deletes the message if it hasn't been polled by the recipient yet.
1009 -- 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.
999 POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /cancel headers {authorization: Text}
1015 POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /cancel headers {authorization: Text}
1000 -> {ok: Bool, error: Text, cancelled: Bool}
1016 -> {ok: Bool, error: Text, cancelled: Bool}
1001 = FederationCancel
1017 = FederationCancel
@@ -1114,7 +1130,7 @@ route MaintenanceApi where
1114 = RotateSecret
1130 = RotateSecret
1116route HealthApi where
1132route 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
1119-- Returns counts of in-memory collections (requires maintenance secret)
1135-- Returns counts of in-memory collections (requires maintenance secret)
1120-- TODO: If Knot adds raw text response support, add a /metrics endpoint
1136-- TODO: If Knot adds raw text response support, add a /metrics endpoint
@@ -1137,6 +1153,11 @@ route StatsApi where
1137 POST {limit: Int} /stats/top-recipients headers {authorization: Text}
1153 POST {limit: Int} /stats/top-recipients headers {authorization: Text}
1138 -> {recipients: [RecipientCount], ok: Bool, total: Int}
1154 -> {recipients: [RecipientCount], ok: Bool, total: Int}
1139 = TopRecipients
1155 = 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}
1141-- Readiness probe: returns false during startup warmup (before first prune completes),
1162-- Readiness probe: returns false during startup warmup (before first prune completes),
1142-- true once the server is fully operational. Unlike /health, this is unauthenticated
1163-- true once the server is fully operational. Unlike /health, this is unauthenticated
@@ -1885,6 +1906,27 @@ writeAllSessions = \sessions -> do
1885 writeSessionBucket 14 partitioned.b14
1906 writeSessionBucket 14 partitioned.b14
1886 writeSessionBucket 15 partitioned.b15
1907 writeSessionBucket 15 partitioned.b15
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 ->
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))
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
1888-- ============================================
1930-- ============================================
1889-- Presence Bucket Helpers
1931-- Presence Bucket Helpers
1890-- ============================================
1932-- ============================================
@@ -2208,12 +2250,12 @@ pruneMessageBucket = \b t -> atomic do
2208 let combined = union se perKeyCapped
2250 let combined = union se perKeyCapped
2209 let overCap = count combined > maxSenderExpirations
2251 let overCap = count combined > maxSenderExpirations
2210 if overCap then do
2252 if overCap then do
2253+ -- Batch eviction: evict 2% of capacity to amortize O(n) scan cost
2254+ let evictCount = max 1 (count combined - maxSenderExpirations + maxSenderExpirations / 50)
2211− let evicted = extractOldest combined (\e -> e.expiredAt)
2255+ let eviction = evictSmallest combined evictCount (\e -> e.expiredAt)
2212− set *senderExpirations = evicted.items
2256+ set *senderExpirations = eviction.kept
2213 den <- *droppedExpirationNotifs
2257 den <- *droppedExpirationNotifs
2214− let denAfterCap = case evicted.candidate of
2216− Just {value: ev} -> incrementDropCounter den ev.senderKey 1
2258+ let denAfterCap = fold (\acc ev -> incrementDropCounter acc ev.senderKey 1) den eviction.evicted
2217 -- Also record per-key drops for senders that hit maxSenderExpirationsPerKey
2259 -- Also record per-key drops for senders that hit maxSenderExpirationsPerKey
2218 let perKeyDroppedEntries = do e <- localExpired; where none perKeyCapped (\c -> c.messageId == e.messageId && c.toKey == e.toKey); yield e
2260 let perKeyDroppedEntries = do e <- localExpired; where none perKeyCapped (\c -> c.messageId == e.messageId && c.toKey == e.toKey); yield e
2219 set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1) denAfterCap perKeyDroppedEntries
2261 set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1) denAfterCap perKeyDroppedEntries
@@ -2642,6 +2684,16 @@ computeForwardStatsInner = \b t acc ->
2642 let newMaxDepth = fold (\mx s -> max mx s.count) acc.maxDepth updatedServerCounts
2684 let newMaxDepth = fold (\mx s -> max mx s.count) acc.maxDepth updatedServerCounts
2643 computeForwardStatsInner (b + 1) t {totalCount: acc.totalCount + bucketCount, oldestAge: max acc.oldestAge bucketOldest, serverCount: count updatedServerCounts, maxDepth: newMaxDepth, serverCounts: updatedServerCounts}
2685 computeForwardStatsInner (b + 1) t {totalCount: acc.totalCount + bucketCount, oldestAge: max acc.oldestAge bucketOldest, serverCount: count updatedServerCounts, maxDepth: newMaxDepth, serverCounts: updatedServerCounts}
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
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)
2645-- Remove forward entries matching a server and message IDs, scanning per-bucket.
2697-- Remove forward entries matching a server and message IDs, scanning per-bucket.
2646-- Only rewrites buckets that actually changed (avoids writeAllForwards repartitioning).
2698-- Only rewrites buckets that actually changed (avoids writeAllForwards repartitioning).
2647-- Returns all removed forward entries across all buckets.
2699-- Returns all removed forward entries across all buckets.
@@ -3013,8 +3065,9 @@ matchIp = \clientIp sessionIp ->
3013-- Generic sliding-window rate limiter. Returns whether the request is rate-limited,
3065-- Generic sliding-window rate limiter. Returns whether the request is rate-limited,
3014-- the updated entries list, and retryAfterMs (time until the current window resets, 0 if not limited).
3066-- the updated entries list, and retryAfterMs (time until the current window resets, 0 if not limited).
3015-- Callers provide match/create/update functions for their entry type.
3067-- Callers provide match/create/update functions for their entry type.
3068+-- Uses a direct fold for entry lookup (avoids materializing a filtered intermediate list).
3016checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntries t ->
3069checkRateWindow = \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
3018 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= windowMs in
3071 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= windowMs in
3019 let rateCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count in
3072 let rateCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count in
3020 if rateCount >= maxRate
3073 if rateCount >= maxRate
@@ -3128,13 +3181,17 @@ evictByCutoff = \items cutoff getKey ->
3128 do x <- items; where getKey x >= cutoff; yield x
3181 do x <- items; where getKey x >= cutoff; yield x
3130-- Evict nonces with fallback: first try expired-nonce eviction (O(n)), then if still
3183-- 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.
3132-- Minimizes replay window: expired nonces are removed first (no window), and the
3188-- 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.
3134evictNonces = \items maxCount retention t ->
3190evictNonces = \items maxCount retention t ->
3135 let afterExpired = evictByCutoff items (t - retention) (\n -> n.seenAt) in
3191 let afterExpired = evictByCutoff items (t - retention) (\n -> n.seenAt) in
3136 if count afterExpired <= maxCount then afterExpired
3192 if count afterExpired <= maxCount then afterExpired
3193+ else let headroom = max 1 (maxCount / 20) in -- 5% headroom
3137− else let evictCount = max 1 (count afterExpired / 100) in
3194+ let evictCount = count afterExpired - maxCount + headroom in
3138 (evictSmallest afterExpired evictCount (\n -> n.seenAt)).kept
3195 (evictSmallest afterExpired evictCount (\n -> n.seenAt)).kept
3140-- Enforce per-key capacity: if a key has >= maxPerKey items, evict the oldest for that key.
3197-- 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
3226-- Validate incoming gossip events: reject local-server events, future timestamps,
3283-- Validate incoming gossip events: reject local-server events, future timestamps,
3227-- and already-expired entries. Clamp TTLs and split into online/offline buckets.
3284-- 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.
3228validGossipEvents : [GossipEvent] -> Timestamp -> {online: [ValidGossipEntry], offline: [ValidGossipEntry]}
3290validGossipEvents : [GossipEvent] -> Timestamp -> {online: [ValidGossipEntry], offline: [ValidGossipEntry]}
3229validGossipEvents = \events t ->
3291validGossipEvents = \events t ->
3230 let validate = \e ->
3292 let validate = \e ->
3231 if not (isLocalServer e.server) && e.timestamp <= t + clockSkewTolerance
3293 if not (isLocalServer e.server) && e.timestamp <= t + clockSkewTolerance
3232 then let ttl = clampTtl e.ttl in
3294 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
3234 if expiresAt > t
3297 if expiresAt > t
3235 then [{pubkey: e.pubkey, server: e.server, expiresAt: expiresAt, timestamp: e.timestamp, ttl: ttl}]
3298 then [{pubkey: e.pubkey, server: e.server, expiresAt: expiresAt, timestamp: e.timestamp, ttl: ttl}]
@@ -3740,7 +3803,7 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
3740 globalLimited <- checkGlobalRate t
3803 globalLimited <- checkGlobalRate t
3741 if globalLimited
3804 if globalLimited
3743− logWarn ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)
3806+ logEvent "warn" "rate_limited_federation_global" ("peer=" ++ fromServer ++ " limit=" ++ show maxGlobalRequestRate)
3744 -- Encode retryAfterMs in error string so peers can make smarter retry decisions.
3807 -- Encode retryAfterMs in error string so peers can make smarter retry decisions.
3745 -- Format: "rate_limited:NNNms" — peers that don't parse it still see "rate_limited:...".
3808 -- Format: "rate_limited:NNNms" — peers that don't parse it still see "rate_limited:...".
3746 yield (respond {ok: False {}, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
3809 yield (respond {ok: False {}, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
@@ -3748,7 +3811,7 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
3748 rateLimited <- checkFederationRate fromServer t
3811 rateLimited <- checkFederationRate fromServer t
3749 if rateLimited
3812 if rateLimited
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)
3752 yield (respond {ok: False {}, error: "rate_limited:" ++ show federationRateWindowMs ++ "ms"})
3815 yield (respond {ok: False {}, error: "rate_limited:" ++ show federationRateWindowMs ++ "ms"})
3754 when clearOnAuth (clearFailure fromServer)
3817 when clearOnAuth (clearFailure fromServer)
@@ -3931,7 +3994,7 @@ retryForwardsForServer = \entries remaining ->
3931 let server = case headOf entries of Just {value: f} -> f.toServer; Nothing {} -> ""
3994 let server = case headOf entries of Just {value: f} -> f.toServer; Nothing {} -> ""
3932 forEachBatch entries maxBatchSize (\batch -> do
3995 forEachBatch entries maxBatchSize (\batch -> do
3933 t <- now
3996 t <- now
3934− auth <- generateFederationAuth "forward" server
3997+ auth <- generateFederationAuth "forward-batch" server
3935 corrId <- genTokenN 1
3998 corrId <- genTokenN 1
3936 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}
3999 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}
3937 result <- fetchFederation server (FederationBatchForward {
4000 result <- fetchFederation server (FederationBatchForward {
@@ -3976,7 +4039,7 @@ retryForwardsSingle = \batch server ->
3976 Just {value: f} -> do
4039 Just {value: f} -> do
3977 let rest = do x <- batch; where x.messageId != f.messageId; yield x
4040 let rest = do x <- batch; where x.messageId != f.messageId; yield x
3978 t <- now
4041 t <- now
3979− auth <- generateFederationAuth "forward" server
4042+ auth <- generateFederationAuth ("forward:" ++ f.messageId) server
3980 corrId <- genTokenN 1
4043 corrId <- genTokenN 1
3981 result <- fetchFederation server (ForwardMessage {
4044 result <- fetchFederation server (ForwardMessage {
3982 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
4045 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
@@ -4317,7 +4380,7 @@ retryPendingCancels = do
4317 let serverCancels = do c <- pending; where c.toServer == server; yield c
4380 let serverCancels = do c <- pending; where c.toServer == server; yield c
4318 forEachBatch serverCancels maxBatchSize (\batch -> do
4381 forEachBatch serverCancels maxBatchSize (\batch -> do
4319 cancelT <- now
4382 cancelT <- now
4320− auth <- generateFederationAuth "cancel" server
4383+ auth <- generateFederationAuth "cancel-batch" server
4321 corrId <- genTokenN 1
4384 corrId <- genTokenN 1
4322 let payload = do c <- batch; yield {messageId: c.messageId, toKey: c.toKey}
4385 let payload = do c <- batch; yield {messageId: c.messageId, toKey: c.toKey}
4323 result <- fetchFederation server (FederationBatchCancel {
4386 result <- fetchFederation server (FederationBatchCancel {
@@ -4638,13 +4701,17 @@ ackPolledMessages = \pubkey cursor t -> do
4638 let combined = union sd deduped
4701 let combined = union sd deduped
4639 let overCap = count combined > maxSenderDeliveries
4702 let overCap = count combined > maxSenderDeliveries
4640 if overCap then do
4703 if overCap then do
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)
4641− let evicted = extractOldest combined (\d -> d.deliveredAt)
4708+ let eviction = evictSmallest combined evictCount (\d -> d.deliveredAt)
4642− set *senderDeliveries = evicted.items
4709+ set *senderDeliveries = eviction.kept
4643− case evicted.candidate of
4710+ -- Track per-sender drop counts for all evicted entries
4644− Nothing {} -> yield {}
4711+ let dropCounts = fold (\acc e -> incrementDropCounter acc e.senderKey 1) [] eviction.evicted
4645− Just {value: ev} -> do
4712+ when (count dropCounts > 0) (do
4646− ddr <- *droppedDeliveryReceipts
4713+ ddr <- *droppedDeliveryReceipts
4647− set *droppedDeliveryReceipts = incrementDropCounter ddr ev.senderKey 1
4714+ set *droppedDeliveryReceipts = fold (\acc dc -> incrementDropCounter acc dc.senderKey dc.count) ddr dropCounts)
4648 else set *senderDeliveries = combined)
4715 else set *senderDeliveries = combined)
4649 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
4716 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
4650 else yield {pendingAcks: [], evictedEntries: []}
4717 else yield {pendingAcks: [], evictedEntries: []}
@@ -4930,7 +4997,7 @@ handleChallenge = \pubkey clientIp respond -> do
4930 preAuth <- checkPreAuth clientIp t
4997 preAuth <- checkPreAuth clientIp t
4931 if preAuth != ""
4998 if preAuth != ""
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))
4934 yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: preAuth})
5001 yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: preAuth})
4936 authLimited <- checkAuthGlobalRate t
5003 authLimited <- checkAuthGlobalRate t
@@ -4959,7 +5026,7 @@ handleChallenge = \pubkey clientIp respond -> do
4959 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
5026 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
4960 yield {accepted: True {}, rateLimited: False {}, retryAfterMs: 0, error: ""}
5027 yield {accepted: True {}, rateLimited: False {}, retryAfterMs: 0, error: ""}
4961 when result.rateLimited
5028 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))
4963 when (not result.accepted && not result.rateLimited)
5030 when (not result.accepted && not result.rateLimited)
4964 (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
5031 (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
4965 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})
5032 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 ->
4972 preAuth <- checkPreAuth clientIp t
5039 preAuth <- checkPreAuth clientIp t
4973 if preAuth != ""
5040 if preAuth != ""
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))
4976 yield (respond {token: "", expiresAt: 0, error: preAuth})
5043 yield (respond {token: "", expiresAt: 0, error: preAuth})
4978 authRateLimited <- atomic do
5045 authRateLimited <- atomic do
@@ -4982,7 +5049,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
4982 set *challengeRates = rateResult.entries
5049 set *challengeRates = rateResult.entries
4983 yield rateResult.limited
5050 yield rateResult.limited
4984 if authRateLimited then do
5051 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)
4986 yield (respond {token: "", expiresAt: 0, error: "rate_limited"})
5053 yield (respond {token: "", expiresAt: 0, error: "rate_limited"})
4988 authGlobalLimited <- checkAuthGlobalRate t
5055 authGlobalLimited <- checkAuthGlobalRate t
@@ -4997,7 +5064,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
4997 else result.entries
5064 else result.entries
4998 yield result.limited
5065 yield result.limited
4999 if uniqueKeyLimited then do
5066 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)
5001 yield (respond {token: "", expiresAt: 0, error: "unique_key_limited"})
5068 yield (respond {token: "", expiresAt: 0, error: "unique_key_limited"})
5002 else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
5069 else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
@@ -5078,10 +5145,10 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
5078 let perKeyEvicted = if not revokeOthers && sessionsForKey >= maxSessionsPerKey then 1 else 0
5145 let perKeyEvicted = if not revokeOthers && sessionsForKey >= maxSessionsPerKey then 1 else 0
5079 let estimatedTotal = tsc - revokedCount - perKeyEvicted
5146 let estimatedTotal = tsc - revokedCount - perKeyEvicted
5080 when (estimatedTotal >= maxSessions)
5147 when (estimatedTotal >= maxSessions)
5081− (do -- Find bucket with oldest session and evict from it
5148+ (do -- Find the bucket containing the oldest session and evict from just that bucket.
5082− allSess <- readAllSessions
5149+ -- Avoids materializing all 16 session buckets into a single list (O(n) union chain).
5083− let globalCapped = capGlobal allSess maxSessions (\s -> s.expiresAt)
5150+ -- Instead, reads each bucket to find the global minimum, then writes only one bucket.
5084− writeAllSessions globalCapped)
5151+ evictOldestSessionFromBuckets 0 Nothing {})
5085 currentBucket <- readSessionBucket nb
5152 currentBucket <- readSessionBucket nb
5086 writeSessionBucket nb (union currentBucket [newSession])
5153 writeSessionBucket nb (union currentBucket [newSession])
5087 -- Final count: estimatedTotal + 1 (new session) - globalEviction (0 or 1)
5154 -- Final count: estimatedTotal + 1 (new session) - globalEviction (0 or 1)
@@ -5282,7 +5349,7 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
5282 Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt, fwdPriority: fe.priority}
5349 Just {value: fe} -> yield {pending: True {}, fwdExpiresAt: fe.expiresAt, fwdPriority: fe.priority}
5283 when fwdInfo.pending (do
5350 when fwdInfo.pending (do
5284 fwdT <- now
5351 fwdT <- now
5285− fwdAuth <- generateFederationAuth "forward" server
5352+ fwdAuth <- generateFederationAuth ("forward:" ++ effectiveMsgId) server
5286 fwdCorrId <- genTokenN 1
5353 fwdCorrId <- genTokenN 1
5287 fwdResult <- fetchFederation server (ForwardMessage {
5354 fwdResult <- fetchFederation server (ForwardMessage {
5288 toKey: to, encryptedBlob: encryptedBlob,
5355 toKey: to, encryptedBlob: encryptedBlob,
@@ -5316,7 +5383,7 @@ sendBatchForwardsSequential = \msgs server ->
5316 yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))
5383 yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))
5317 when stillPending (do
5384 when stillPending (do
5318 fwdT <- now
5385 fwdT <- now
5319− fwdAuth <- generateFederationAuth "forward" server
5386+ fwdAuth <- generateFederationAuth ("forward:" ++ r.messageId) server
5320 fwdCorrId <- genTokenN 1
5387 fwdCorrId <- genTokenN 1
5321 fwdResult <- fetchFederation server (ForwardMessage {
5388 fwdResult <- fetchFederation server (ForwardMessage {
5322 toKey: r.toKey, encryptedBlob: r.blob,
5389 toKey: r.toKey, encryptedBlob: r.blob,
@@ -5362,7 +5429,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
5362 (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
5429 (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
5363 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs clampedPriority t)
5430 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs clampedPriority t)
5364 case result.status of
5431 case result.status of
5365− MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
5432+ MsRateLimited {} -> logEvent "warn" "rate_limited_send" ("messageId=" ++ effectiveMsgId)
5366 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (global capacity full)"))
5433 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (global capacity full)"))
5367 _ -> yield {}
5434 _ -> yield {}
5368 forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.remoteServers
5435 forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.remoteServers
@@ -5586,7 +5653,7 @@ handleVerifyToken = \token scope clientIp respond -> do
5586 set *verifyIpRates = result.entries
5653 set *verifyIpRates = result.entries
5587 yield result.limited
5654 yield result.limited
5588 if verifyLimited then do
5655 if verifyLimited then do
5589− logWarn ("federation/verify: rate limited for IP " ++ clientIp)
5656+ logEvent "warn" "rate_limited_verify_ip" ("ip=" ++ clientIp)
5590 yield (respond {ok: False {}})
5657 yield (respond {ok: False {}})
5592 found <- atomic do
5659 found <- atomic do
@@ -5641,7 +5708,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
5641 set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs; yield r) else newEntries
5708 set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs; yield r) else newEntries
5642 yield (False {})
5709 yield (False {})
5643 if eventRateLimited then do
5710 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)
5645 yield (respond {ok: False {}, error: "event_rate_limited:" ++ show gossipEventRateWindowMs ++ "ms"})
5712 yield (respond {ok: False {}, error: "event_rate_limited:" ++ show gossipEventRateWindowMs ++ "ms"})
5647 let validated = validGossipEvents cappedEvents t
5714 let validated = validGossipEvents cappedEvents t
@@ -5738,7 +5805,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
5738 t <- now
5805 t <- now
5739 let corrId = sanitizeCorrelationId correlationId
5806 let corrId = sanitizeCorrelationId correlationId
5740 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward " ++ messageId ++ " from " ++ fromServer))
5807 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
5742 when (fromServer != originServer)
5809 when (fromServer != originServer)
5743 (logInfo ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
5810 (logInfo ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
5744 -- Relay origin verification: for relay forwards, verify:
5811 -- Relay origin verification: for relay forwards, verify:
@@ -5845,7 +5912,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
5845 if result.globalCapReached
5912 if result.globalCapReached
5846 then logWarn ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
5913 then logWarn ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
5847 else if result.recvRateLimited
5914 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)
5849 else logWarn ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
5916 else logWarn ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
5850 _ -> yield {}
5917 _ -> yield {}
5851 yield (case result.forward of
5918 yield (case result.forward of
@@ -5974,7 +6041,7 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
5974 let corrId = sanitizeCorrelationId correlationId
6041 let corrId = sanitizeCorrelationId correlationId
5975 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel " ++ messageId ++ " from " ++ fromServer))
6042 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel " ++ messageId ++ " from " ++ fromServer))
5976 let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}}
6043 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
5978 result <- atomic do
6045 result <- atomic do
5979 let b = pubkeyBucket toKey
6046 let b = pubkeyBucket toKey
5980 bucket <- readMessageBucket b
6047 bucket <- readMessageBucket b
@@ -6015,7 +6082,7 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
6015 let corrId = sanitizeCorrelationId correlationId
6082 let corrId = sanitizeCorrelationId correlationId
6016 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel-batch (" ++ show (count cancels) ++ " entries) from " ++ fromServer))
6083 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel-batch (" ++ show (count cancels) ++ " entries) from " ++ fromServer))
6017 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: []}
6084 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
6019 let capped = takeRelation maxBatchSize cancels
6086 let capped = takeRelation maxBatchSize cancels
6020 results <- cancelBatchInner capped fromServer t []
6087 results <- cancelBatchInner capped fromServer t []
6021 let cancelledCount = count (do r <- results; where r.cancelled; yield r)
6088 let cancelledCount = count (do r <- results; where r.cancelled; yield r)
@@ -6088,7 +6155,7 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp correlation
6088 let corrId = sanitizeCorrelationId correlationId
6155 let corrId = sanitizeCorrelationId correlationId
6089 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward-batch (" ++ show (count forwards) ++ " entries) from " ++ fromServer))
6156 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward-batch (" ++ show (count forwards) ++ " entries) from " ++ fromServer))
6090 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: []}
6157 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
6092 let capped = takeRelation maxBatchSize forwards
6159 let capped = takeRelation maxBatchSize forwards
6093 -- Pre-deduplicate (messageId, toKey) pairs: a malicious peer could send 100 entries
6160 -- Pre-deduplicate (messageId, toKey) pairs: a malicious peer could send 100 entries
6094 -- for the same pair, each triggering a full atomic block (presence check, 5 bucket reads).
6161 -- 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
6249 let serverCancels = do c <- results.fedCancels; where any c.servers (\s -> s == server); yield {messageId: c.messageId, toKey: c.toKey}
6316 let serverCancels = do c <- results.fedCancels; where any c.servers (\s -> s == server); yield {messageId: c.messageId, toKey: c.toKey}
6250 when (count serverCancels > 0) (do
6317 when (count serverCancels > 0) (do
6251 cancelT <- now
6318 cancelT <- now
6252− cancelAuth <- generateFederationAuth "cancel" server
6319+ cancelAuth <- generateFederationAuth "cancel-batch" server
6253 cancelCorrId <- genTokenN 1
6320 cancelCorrId <- genTokenN 1
6254 cancelResult <- fetchFederation server (FederationBatchCancel {
6321 cancelResult <- fetchFederation server (FederationBatchCancel {
6255 cancels: serverCancels, fromServer: serverHostN,
6322 cancels: serverCancels, fromServer: serverHostN,
@@ -6355,16 +6422,26 @@ handleDrain = \phase authorization respond -> do
6355 let pendingAckCount = count flushState.pendingAcks
6422 let pendingAckCount = count flushState.pendingAcks
6356 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
6423 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
6357 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
6424 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
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
6431+ set *drainGeneration = newDg
6358− atomic do set *drainCompletedAt = 0
6432+ set *drainCompletedAt = 0
6359 -- Fork a deadline timer: force-complete drain if flush ops hang on slow/dead peers.
6434 -- Fork a deadline timer: force-complete drain if flush ops hang on slow/dead peers.
6360 -- Without this, operators waiting on drainCompletedAt > 0 could wait forever.
6435 -- Without this, operators waiting on drainCompletedAt > 0 could wait forever.
6361 fork (do
6436 fork (do
6362 sleep drainTimeoutMs
6437 sleep drainTimeoutMs
6363 tTimeout <- now
6438 tTimeout <- now
6440+ dg <- *drainGeneration
6364− completed <- atomic do dc <- *drainCompletedAt; yield dc
6441+ dc <- *drainCompletedAt
6365− when (completed == 0) (do
6442+ when (dg == thisGeneration && dc == 0) (do
6443+ set *drainCompletedAt = tTimeout)
6366− logWarn ("drain: timeout after " ++ show drainTimeoutMs ++ "ms, force-completing")
6444+ when (True {}) (logWarn ("drain: timeout after " ++ show drainTimeoutMs ++ "ms, force-completing")))
6367− atomic do set *drainCompletedAt = tTimeout))
6368 fork (do
6445 fork (do
6369 -- Best-effort flush of pending forwards and acks before stopping
6446 -- Best-effort flush of pending forwards and acks before stopping
6370 when (federationEnabled && pendingForwardCount > 0) (do
6447 when (federationEnabled && pendingForwardCount > 0) (do
@@ -6379,10 +6456,11 @@ handleDrain = \phase authorization respond -> do
6379 -- Wait for in-flight federation requests to complete (timeout: 2x fetch timeout)
6456 -- Wait for in-flight federation requests to complete (timeout: 2x fetch timeout)
6380 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))
6457 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))
6381 t2 <- now
6458 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
6383 atomic do
6460 atomic do
6461+ dg <- *drainGeneration
6384 dc <- *drainCompletedAt
6462 dc <- *drainCompletedAt
6385− when (dc == 0) (set *drainCompletedAt = t2))
6463+ when (dg == thisGeneration && dc == 0) (set *drainCompletedAt = t2))
6386 logInfo ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
6464 logInfo ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
6387 when (pendingForwardCount > 0)
6465 when (pendingForwardCount > 0)
6388 (logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))
6466 (logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))
@@ -6506,6 +6584,10 @@ computeAndCacheHealth = \t -> do
6506 let nonceOk = nonceCount * 10 <= maxForwardNonces * 9
6584 let nonceOk = nonceCount * 10 <= maxForwardNonces * 9
6507 -- Count peers with active failure records (degraded federation connectivity)
6585 -- Count peers with active failure records (degraded federation connectivity)
6508 let degradedPeerCount = count (do f <- forwardFailures; where t - f.firstFailedAt < maxServerDowntime; yield f)
6586 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
6509 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
6591 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
6510 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
6592 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
6511 capacityOk: capacityOk,
6593 capacityOk: capacityOk,
@@ -6513,9 +6595,10 @@ computeAndCacheHealth = \t -> do
6513 federationOk: fedOk,
6595 federationOk: fedOk,
6514 watchdogOk: wdOk,
6596 watchdogOk: wdOk,
6515 nonceOk: nonceOk,
6597 nonceOk: nonceOk,
6598+ forwardQueueOk: fwdQueueOk,
6516 isDraining: dp > 0,
6599 isDraining: dp > 0,
6517 degradedPeers: degradedPeerCount}
6600 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
6519 atomic do
6602 atomic do
6520 set *cachedHealthOk = allOk
6603 set *cachedHealthOk = allOk
6521 set *cachedHealthPrune = result.pruneOk
6604 set *cachedHealthPrune = result.pruneOk
@@ -6525,6 +6608,7 @@ computeAndCacheHealth = \t -> do
6525 set *cachedHealthFederation = result.federationOk
6608 set *cachedHealthFederation = result.federationOk
6526 set *cachedHealthWatchdog = result.watchdogOk
6609 set *cachedHealthWatchdog = result.watchdogOk
6527 set *cachedHealthNonce = result.nonceOk
6610 set *cachedHealthNonce = result.nonceOk
6611+ set *cachedHealthForwardQueue = result.forwardQueueOk
6528 set *cachedHealthDegradedPeers = result.degradedPeers
6612 set *cachedHealthDegradedPeers = result.degradedPeers
6529 yield result
6613 yield result
@@ -6544,15 +6628,16 @@ handleHealth = \authorization respond -> do
6544 cF <- *cachedHealthFederation
6628 cF <- *cachedHealthFederation
6545 cW <- *cachedHealthWatchdog
6629 cW <- *cachedHealthWatchdog
6546 cN <- *cachedHealthNonce
6630 cN <- *cachedHealthNonce
6631+ cFQ <- *cachedHealthForwardQueue
6547 cCW <- *cachedHealthCapacityWarning
6632 cCW <- *cachedHealthCapacityWarning
6548 cDP <- *cachedHealthDegradedPeers
6633 cDP <- *cachedHealthDegradedPeers
6549 dp <- *drainPhase
6634 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}
6551 yield (respond cached)
6636 yield (respond cached)
6553 isAuthed <- verifyMaintenanceAuth authorization
6638 isAuthed <- verifyMaintenanceAuth authorization
6554 result <- computeAndCacheHealth t
6639 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
6556 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
6641 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
6557 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
6642 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
6558 -- Draining status is always visible (load balancers need it for routing decisions).
6643 -- Draining status is always visible (load balancers need it for routing decisions).
@@ -6564,6 +6649,7 @@ handleHealth = \authorization respond -> do
6564 federationHealthy: result.federationOk,
6649 federationHealthy: result.federationOk,
6565 watchdogHealthy: result.watchdogOk,
6650 watchdogHealthy: result.watchdogOk,
6566 nonceHealthy: result.nonceOk,
6651 nonceHealthy: result.nonceOk,
6652+ forwardQueueHealthy: result.forwardQueueOk,
6567 draining: result.isDraining,
6653 draining: result.isDraining,
6568 degradedPeers: result.degradedPeers})
6654 degradedPeers: result.degradedPeers})
6569 else yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: True {},
6655 else yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: True {},
@@ -6573,6 +6659,7 @@ handleHealth = \authorization respond -> do
6573 federationHealthy: True {},
6659 federationHealthy: True {},
6574 watchdogHealthy: True {},
6660 watchdogHealthy: True {},
6575 nonceHealthy: True {},
6661 nonceHealthy: True {},
6662+ forwardQueueHealthy: True {},
6576 draining: result.isDraining,
6663 draining: result.isDraining,
6577 degradedPeers: 0})
6664 degradedPeers: 0})
@@ -6751,6 +6838,37 @@ handleTopRecipients = \limit authorization respond -> do
6751 yield (respond {recipients: top, ok: True {}, total: totalRc})
6838 yield (respond {recipients: top, ok: True {}, total: totalRc})
6752 else yield (respond {recipients: [], ok: False {}, total: 0})
6839 else yield (respond {recipients: [], ok: False {}, total: 0})
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
6846+ globalRateLimited <- checkGlobalRate t
6847+ if globalRateLimited then yield (respond {servers: [], ok: False {}, total: 0})
6849+ authorized <- verifyMaintenanceAuth authorization
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})
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
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
6754-- GET /ready: Readiness probe — returns true once the server has completed startup.
6872-- GET /ready: Readiness probe — returns true once the server has completed startup.
6755-- Unlike /health, this is unauthenticated and doesn't check subsystem liveness.
6873-- Unlike /health, this is unauthenticated and doesn't check subsystem liveness.
6756-- Returns false until the first prune cycle completes (server fully initialized).
6874-- Returns false until the first prune cycle completes (server fully initialized).
@@ -7030,7 +7148,7 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
7030 fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)
7148 fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)
7031 (sendToEach cancelResult.cancelServers (\server -> do
7149 (sendToEach cancelResult.cancelServers (\server -> do
7032 cancelT <- now
7150 cancelT <- now
7033− cancelAuth <- generateFederationAuth "cancel" server
7151+ cancelAuth <- generateFederationAuth "cancel-batch" server
7034 cancelCorrId <- genTokenN 1
7152 cancelCorrId <- genTokenN 1
7035 cancelFedResult <- fetchFederation server (FederationBatchCancel {
7153 cancelFedResult <- fetchFederation server (FederationBatchCancel {
7036 cancels: [{messageId: messageId, toKey: to}], fromServer: serverHostN,
7154 cancels: [{messageId: messageId, toKey: to}], fromServer: serverHostN,
@@ -7149,7 +7267,7 @@ logBatchResults = \results ->
7149 Just {value: r} -> do
7267 Just {value: r} -> do
7150 let rest = do x <- results; where x != r; yield x
7268 let rest = do x <- results; where x != r; yield x
7151 case r.status of
7269 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")
7153 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ r.messageId ++ " (capacity or mailbox full)")
7271 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ r.messageId ++ " (capacity or mailbox full)")
7154 _ -> yield {}
7272 _ -> yield {}
7155 logBatchResults rest
7273 logBatchResults rest
@@ -7580,6 +7698,7 @@ requestPath = \req -> case req of
7580 StatsPeers {} -> "GET /stats/peers"
7698 StatsPeers {} -> "GET /stats/peers"
7581 StatsKey {} -> "GET /stats/key"
7699 StatsKey {} -> "GET /stats/key"
7582 TopRecipients {} -> "POST /stats/top-recipients"
7700 TopRecipients {} -> "POST /stats/top-recipients"
7701+ StatsForwards {} -> "GET /stats/forwards"
7583 Ready {} -> "GET /ready"
7702 Ready {} -> "GET /ready"
7585-- Extract client IP for access logging (empty for federation/maintenance routes)
7704-- Extract client IP for access logging (empty for federation/maintenance routes)
@@ -7632,6 +7751,7 @@ requestClientIp = \req -> case req of
7632 StatsPeers {} -> ""
7751 StatsPeers {} -> ""
7633 StatsKey {} -> ""
7752 StatsKey {} -> ""
7634 TopRecipients {} -> ""
7753 TopRecipients {} -> ""
7754+ StatsForwards {} -> ""
7635 Ready {} -> ""
7755 Ready {} -> ""
7637-- Slow request threshold for access logging (ms)
7757-- Slow request threshold for access logging (ms)
@@ -7693,6 +7813,7 @@ serveInner = \req -> case req of
7693 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
7813 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
7694 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
7814 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
7695 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond
7815 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond
7816+ StatsForwards {authorization, respond} -> handleStatsForwards authorization respond
7696 Ready {respond} -> handleReady respond
7817 Ready {respond} -> handleReady respond
7698-- Backpressure: check if the server is overloaded (too many concurrent requests).
7819-- Backpressure: check if the server is overloaded (too many concurrent requests).
@@ -7737,6 +7858,7 @@ serveOverloaded = \req -> case req of
7737 FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}
7858 FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}
7738 VerifyToken {respond} -> respond {ok: False {}}
7859 VerifyToken {respond} -> respond {ok: False {}}
7739 TopRecipients {respond} -> respond {recipients: [], ok: False {}, total: 0}
7860 TopRecipients {respond} -> respond {recipients: [], ok: False {}, total: 0}
7861+ StatsForwards {respond} -> respond {servers: [], ok: False {}, total: 0}
7740 BlockPeer {respond} -> respond {ok: False {}, blocked: 0}
7862 BlockPeer {respond} -> respond {ok: False {}, blocked: 0}
7741 UnblockPeer {respond} -> respond {ok: False {}, unblocked: False {}}
7863 UnblockPeer {respond} -> respond {ok: False {}, unblocked: False {}}
7742 ListBlocked {respond} -> respond {peers: [], ok: False {}}
7864 ListBlocked {respond} -> respond {peers: [], ok: False {}}
@@ -7915,14 +8037,25 @@ backgroundPrune = do
7915 af <- *activePruneForks
8037 af <- *activePruneForks
7916 set *activePruneForks = af + 1
8038 set *activePruneForks = af + 1
7917 fork (do
8039 fork (do
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
8046+ logWarn "backgroundPrune: stale fork detected (generation changed), skipping"
8048+ af <- *activePruneForks
8049+ set *activePruneForks = max 0 (af - 1)
7918− pruneAndRetry t
8051+ pruneAndRetry t
7919− -- Only clear running flag if this is still the latest generation
8052+ -- 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)
8053+ -- (prevents a stuck old fork from clearing the flag after a recovery cycle started)
7921− atomic do
8054+ atomic do
7922− currentGen <- *pruneGeneration
8055+ currentGen <- *pruneGeneration
7923− when (currentGen == status.gen) (set *pruneRunning = 0)
8056+ when (currentGen == status.gen) (set *pruneRunning = 0)
7924− af <- *activePruneForks
8057+ af <- *activePruneForks
7925− set *activePruneForks = max 0 (af - 1))
8058+ set *activePruneForks = max 0 (af - 1))
7926 else logWarn "backgroundPrune: skipping (previous cycle still running)"
8059 else logWarn "backgroundPrune: skipping (previous cycle still running)"
7927 backgroundPrune
8060 backgroundPrune
@@ -8094,6 +8227,8 @@ main = do
8094 (logWarn "maintenanceSecret is shorter than 32 characters — 32+ character secrets recommended for production (128+ bits of entropy)")
8227 (logWarn "maintenanceSecret is shorter than 32 characters — 32+ character secrets recommended for production (128+ bits of entropy)")
8095 when (maintenanceSecret != "" && federationEnabled && not (tooLong maintenanceSecret 15))
8228 when (maintenanceSecret != "" && federationEnabled && not (tooLong maintenanceSecret 15))
8096 (logError "SECURITY: maintenanceSecret is weak and federation is enabled — maintenance endpoints control drain and peer blocking. Use a secret of at least 32 characters.")
8229 (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.")
8097 when (matchesHostnameOrSubdomain serverHostN "localhost")
8232 when (matchesHostnameOrSubdomain serverHostN "localhost")
8098 (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
8233 (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
8099 when (federationEnabled && isBadServerName serverHostN)
8234 when (federationEnabled && isBadServerName serverHostN)