@@ -148,6 +148,8 @@ gossipEventRateWindowMs = 60000 -- 1 minute
148148maxGossipEventRateEntries = 100
149149-- Global federation connection cap (bounds total concurrent outbound connections across all paths)
150150maxGlobalFederationConcurrency = 30
151+-- Peer blocklist: max blocked servers tracked at runtime (managed via /maintenance/block, /maintenance/unblock)
152+maxBlockedPeers = 100
151153
152154-- Background task intervals
153155backgroundPruneInterval : Int<Ms>
@@ -170,6 +172,8 @@ federationEnabled = True {}
170172-- ServerName validation time. A hostname passing isBadServerName could rebind to a
171173-- private IP between validation and fetch. Mitigate at the network layer (e.g.,
172174-- firewall rules blocking outbound connections to RFC1918 ranges from the server process).
175+-- TODO: If Knot adds a resolved-IP introspection API (e.g., fetchWithResolvedIp),
176+-- validate the resolved IP against isBadServerName before completing the request.
173177federationUrl = \server -> federationScheme ++ server
174178-- Callback verification: outbound federation requests include a random single-use token;
175179-- the receiver calls back to /federation/verify to confirm the sender actually issued it.
@@ -199,6 +203,8 @@ ackGracePeriod : Int<Ms>
199203ackGracePeriod = 30000 -- 30 seconds
200204-- Deferred ack cleanup capacity
201205maxPendingAckCleanups = 10000
206+-- Per-key deferred ack cleanup cap (prevents one noisy key from dominating the queue)
207+maxPendingAckCleanupsPerKey = 100
202208
203209-- Health check thresholds (background task freshness)
204210healthPruneMaxAge : Int<Ms>
@@ -332,6 +338,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
332338
333339-- Sessions sharded into 16 buckets by first hex char of token.
334340-- Hot-path auth scans only one bucket (~1/16 of sessions).
341+-- TODO: If Knot adds indexed arrays or maps, replace the 16 manual bucket
342+-- refs and their if-else dispatch chains with a single indexed collection.
343+-- This would eliminate ~800 lines of mechanical boilerplate across sessions,
344+-- messages, presence, and signal buckets.
335345*sessionBucket0 : [Session]
336346*sessionBucket1 : [Session]
337347*sessionBucket2 : [Session]
@@ -533,6 +543,9 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
533543-- Global federation connection counter (tracks total concurrent outbound connections)
534544*federationActiveConnections : Int
535545
546+-- Blocked federation peers (managed at runtime via /maintenance/block and /maintenance/unblock)
547+*blockedPeers : [Text]
548+
536549-- Deferred ack cleanups: origin message reclamation queued by federation ack handlers,
537550-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
538551*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
@@ -640,11 +653,38 @@ route MaintenanceApi where
640653 POST {version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce]} /maintenance/restore headers {authorization: Text}
641654 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
642655 = Restore
656+ -- Block a federation peer (rejects all inbound/outbound federation with this server)
657+ POST {server: ServerName} /maintenance/block headers {authorization: Text}
658+ -> {ok: Bool, blocked: Int}
659+ = BlockPeer
660+ -- Unblock a previously blocked peer
661+ POST {server: ServerName} /maintenance/unblock headers {authorization: Text}
662+ -> {ok: Bool, unblocked: Bool}
663+ = UnblockPeer
664+ -- List currently blocked peers
665+ GET /maintenance/blocked headers {authorization: Text}
666+ -> {peers: [Text], ok: Bool}
667+ = ListBlocked
668+ -- Test federation connectivity to a specific peer via verify callback round-trip
669+ POST {server: ServerName} /maintenance/ping headers {authorization: Text}
670+ -> {ok: Bool, latencyMs: Int, error: Text}
671+ = PingPeer
672+ -- List sessions for a specific pubkey (admin debugging)
673+ GET /maintenance/sessions/{pubkey: PubkeyHex} headers {authorization: Text}
674+ -> {sessions: [{token: Text, expiresAt: Timestamp, ip: Text, createdAt: Timestamp}], count: Int, ok: Bool}
675+ = AdminListSessions
676+ -- Revoke all sessions for a specific pubkey (abuse response)
677+ POST {pubkey: PubkeyHex} /maintenance/sessions/revoke-all headers {authorization: Text}
678+ -> {ok: Bool, revokedCount: Int}
679+ = AdminRevokeSessions
643680
644681route HealthApi where
645− GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool} = Health
682+ GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool} = Health
646683
647684-- Returns counts of in-memory collections (requires maintenance secret)
685+-- TODO: If Knot adds raw text response support, add a /metrics endpoint
686+-- returning Prometheus exposition format for direct integration with standard
687+-- monitoring stacks. Currently /stats returns JSON which requires a custom scraper.
648688route StatsApi where
649689 GET /stats headers {authorization: Text}
650690 -> {sessions: Int, sessionsMax: Int, messages: Int, messagesMax: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pairRates: Int, federationPairRates: Int, pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, federationEnabled: Bool, recentGossip: Int, totalMessageCount: Int, recipientCounts: Int, federationInFlight: Int, hungFederationSlots: Int, messagesPct: Int, presencePct: Int, forwardsPct: Int, sessionsPct: Int, federationAvgLatencyMs: Int, federationMaxLatencyMs: Int, federationCallCount: Int, drainCompletedAt: Timestamp, oldestForwardAge: Int, oldestAckAge: Int, ok: Bool}
@@ -934,6 +974,10 @@ verifyAuthWith = \check authorization ->
934974verifyMaintenanceAuth : Text -> Bool
935975verifyMaintenanceAuth = verifyAuthWith (\token -> verifySecret token maintenanceSecret)
936976
977+-- Check if a server is on the blocklist
978+isBlockedPeer : Text -> [Text] -> Bool
979+isBlockedPeer = \server blocked -> any blocked (\b -> b == server)
980+
937981-- Generate a single-use token for outbound federation auth.
938982-- The receiving server calls back to /federation/verify to confirm we issued it.
939983-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
@@ -1405,6 +1449,9 @@ reconcileMessageCounts = \b acc ->
14051449-- Recipient Count Helpers
14061450-- ============================================
14071451-- Per-recipient message counts avoid O(n) scan of all messages on send/forward.
1452+-- TODO: If Knot adds a map/dictionary type, replace [RecipientCount] with a
1453+-- Map<PubkeyHex, Int> for O(1) lookups. Currently getRecipientCount and
1454+-- adjustRecipientCount are O(unique_recipients) per call.
14081455
14091456getRecipientCount : PubkeyHex -> [RecipientCount] -> Int
14101457getRecipientCount = \pubkey counts ->
@@ -1427,6 +1474,8 @@ adjustRecipientCount = \counts pubkey delta ->
14271474
14281475-- Count items per toKey in a single pass: fold accumulates {pubkey, count} pairs,
14291476-- incrementing existing entries or inserting new ones.
1477+-- NOTE: O(n*k) where k = unique recipients due to `any acc` check inside fold.
1478+-- Acceptable for per-bucket data (n ≤ maxMessages/16). A map type would make this O(n).
14301479countByRecipient = \items ->
14311480 fold (\acc x ->
14321481 if any acc (\c -> c.pubkey == x.toKey)
@@ -1575,6 +1624,38 @@ checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntr
15751624 else do r <- entries; yield (if matchKey r then updateCount r else r)
15761625 in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs; yield r) else newEntries, retryAfterMs: 0}
15771626
1627+-- Rate limiter helpers: wrap common checkRateWindow patterns to reduce boilerplate.
1628+-- Each returns {limited, entries, retryAfterMs} — caller reads/writes the mutable ref.
1629+checkPubkeyRate = \rates pubkey maxRate windowMs maxEntries t ->
1630+ checkRateWindow rates (\r -> r.pubkey == pubkey)
1631+ (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
1632+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
1633+ maxRate windowMs maxEntries t
1634+
1635+checkIpRate = \rates ip maxRate windowMs maxEntries t ->
1636+ checkRateWindow rates (\r -> r.ip == ip)
1637+ (\ts -> {ip: ip, windowStart: ts, count: 1})
1638+ (\r -> {ip: r.ip, windowStart: r.windowStart, count: r.count + 1})
1639+ maxRate windowMs maxEntries t
1640+
1641+checkServerRate = \rates server maxRate windowMs maxEntries t ->
1642+ checkRateWindow rates (\r -> r.server == server)
1643+ (\ts -> {server: server, windowStart: ts, count: 1})
1644+ (\r -> {server: r.server, windowStart: r.windowStart, count: r.count + 1})
1645+ maxRate windowMs maxEntries t
1646+
1647+checkPairRate = \rates sender recipient maxRate windowMs maxEntries t ->
1648+ checkRateWindow rates (\r -> r.sender == sender && r.recipient == recipient)
1649+ (\ts -> {sender: sender, recipient: recipient, windowStart: ts, count: 1})
1650+ (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
1651+ maxRate windowMs maxEntries t
1652+
1653+checkServerRecipientRate = \rates server recipient maxRate windowMs maxEntries t ->
1654+ checkRateWindow rates (\r -> r.server == server && r.recipient == recipient)
1655+ (\ts -> {server: server, recipient: recipient, windowStart: ts, count: 1})
1656+ (\r -> {server: r.server, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
1657+ maxRate windowMs maxEntries t
1658+
15781659-- Extract the item with the smallest key and return both the item and remaining list
15791660extractOldest = \items getKey ->
15801661 fold (\acc x ->
@@ -1838,10 +1919,12 @@ pruneExpired = \t -> do
18381919 set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
18391920 -- Clean up stale in-flight entries: remove zeroed entries and reset leaked slots
18401921 -- where the forked task died without calling clearFederationInFlight.
1841− -- Uses 1.5x timeout threshold so leaked slots are reclaimed promptly
1922+ -- Uses 1.25x timeout threshold so leaked slots are reclaimed promptly
18421923 -- after the skip-hung-server logic (at 1x) has already taken effect.
1924+ -- Tighter than the original 1.5x to reduce the window where panicked
1925+ -- fetches hold federation slots hostage.
18431926 inFlight <- *federationInFlight
1844− let staleThreshold = federationFetchTimeoutMs * 3 / 2
1927+ let staleThreshold = federationFetchTimeoutMs * 5 / 4
18451928 let staleInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= staleThreshold; yield r
18461929 when (count staleInFlight > 0)
18471930 (logWarn ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))
@@ -1931,10 +2014,7 @@ mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
19312014
19322015checkFederationRate = \server t -> atomic do
19332016 federationRates <- *federationRates
1934− let result = checkRateWindow federationRates (\r -> r.server == server)
1935− (\ts -> {server: server, windowStart: ts, count: 1})
1936− (\r -> {server: r.server, windowStart: r.windowStart, count: r.count + 1})
1937− maxFederationRatePerServer federationRateWindowMs maxFederationRateEntries t
2017+ let result = checkServerRate federationRates server maxFederationRatePerServer federationRateWindowMs maxFederationRateEntries t
19382018 set *federationRates = result.entries
19392019 yield result.limited
19402020
@@ -1972,10 +2052,7 @@ checkIpRate = \ip t ->
19722052 if ip == "" then yield (False {})
19732053 else atomic do
19742054 ipRates <- *ipRates
1975− let result = checkRateWindow ipRates (\r -> r.ip == ip)
1976− (\ts -> {ip: ip, windowStart: ts, count: 1})
1977− (\r -> {ip: r.ip, windowStart: r.windowStart, count: r.count + 1})
1978− maxIpRatePerIp ipRateWindowMs maxIpRateEntries t
2055+ let result = checkIpRate ipRates ip maxIpRatePerIp ipRateWindowMs maxIpRateEntries t
19792056 set *ipRates = result.entries
19802057 yield result.limited
19812058
@@ -1992,6 +2069,13 @@ checkPreAuth = \clientIp t -> do
19922069 else yield ""
19932070
19942071withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do
2072+ -- Check blocklist before spending resources on verification callback
2073+ blocked <- *blockedPeers
2074+ if isBlockedPeer fromServer blocked
2075+ then do
2076+ logWarn ("federation: blocked peer " ++ fromServer ++ " rejected")
2077+ yield (respond {ok: False {}, error: "blocked"})
2078+ else do
19952079 verified <- verifyFederationCallback fromServer authorization scope
19962080 if not verified
19972081 then yield (respond {ok: False {}, error: "unauthorized"})
@@ -2099,10 +2183,10 @@ shouldRetryServer = \server t failures ->
20992183 Nothing {} -> True {}
21002184 Just {value: f} -> t - f.firstFailedAt < maxServerDowntime && t - f.failedAt >= retryBackoff f.failures
21012185
2102−-- Extended retry check: also skip servers with hung in-flight requests
2103−shouldRetryServerEx : Text -> Timestamp -> [ForwardFailure] -> [FederationInFlightEntry] -> Bool
2104−shouldRetryServerEx = \server t failures inFlight ->
2105− shouldRetryServer server t failures && not (hasHungFederationRequest server t inFlight)
2186+-- Extended retry check: also skip servers with hung in-flight requests or on the blocklist
2187+shouldRetryServerEx : Text -> Timestamp -> [ForwardFailure] -> [FederationInFlightEntry] -> [Text] -> Bool
2188+shouldRetryServerEx = \server t failures inFlight blocked ->
2189+ not (isBlockedPeer server blocked) && shouldRetryServer server t failures && not (hasHungFederationRequest server t inFlight)
21062190
21072191-- Global circuit breaker: returns True when too many peers are failing simultaneously,
21082192-- indicating a likely local network issue rather than individual peer problems.
@@ -2180,7 +2264,8 @@ retryPendingForwards = \pending -> do
21802264 then logWarn ("federation: circuit breaker open (" ++ show (count failures) ++ " peers failing), skipping forward retries")
21812265 else do
21822266 inFlight <- *federationInFlight
2183− let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2267+ blocked <- *blockedPeers
2268+ let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
21842269 sendToEach eligibleServers (\server ->
21852270 retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)
21862271 yield {}
@@ -2208,7 +2293,8 @@ forwardQueuedMessages = \targets t -> do
22082293 then logWarn ("federation: circuit breaker open, skipping queued message forwarding")
22092294 else do
22102295 inFlight <- *federationInFlight
2211− let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2296+ blocked <- *blockedPeers
2297+ let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
22122298 sendToEach eligibleServers (\server ->
22132299 retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)
22142300
@@ -2263,7 +2349,8 @@ sendAcksAndClean = \pendingAcks -> do
22632349 then logWarn ("federation: circuit breaker open, skipping ack retries")
22642350 else do
22652351 inFlight <- *federationInFlight
2266− let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2352+ blocked <- *blockedPeers
2353+ let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
22672354 sendToEach eligibleServers (\server ->
22682355 sendBatchedAcksToServer (acksForServer pendingAcks server) server)
22692356
@@ -2322,7 +2409,8 @@ retryOverflowAcks = \overflow -> do
23222409 then logWarn ("federation: circuit breaker open, skipping overflow ack retries")
23232410 else do
23242411 inFlight <- *federationInFlight
2325− let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2412+ blocked <- *blockedPeers
2413+ let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
23262414 sendToEach eligibleServers (\server ->
23272415 forEachBatch (acksForServer overflow server) maxMessageIds (\batch -> do
23282416 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
@@ -2797,10 +2885,7 @@ handleChallenge = \pubkey clientIp respond -> do
27972885 then yield {accepted: False {}, rateLimited: False {}, retryAfterMs: 0}
27982886 else do
27992887 challengeRates <- *challengeRates
2800− let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
2801− (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
2802− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
2803− maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
2888+ let rateResult = checkPubkeyRate challengeRates pubkey maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
28042889 if rateResult.limited
28052890 then yield {accepted: False {}, rateLimited: True {}, retryAfterMs: rateResult.retryAfterMs}
28062891 else do
@@ -2826,10 +2911,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
28262911 else do
28272912 authRateLimited <- atomic do
28282913 challengeRates <- *challengeRates
2829− let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
2830− (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
2831− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
2832− maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
2914+ let rateResult = checkPubkeyRate challengeRates pubkey maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
28332915 set *challengeRates = rateResult.entries
28342916 yield rateResult.limited
28352917 if authRateLimited then do
@@ -3018,28 +3100,19 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
30183100 then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}
30193101 else do
30203102 sendRates <- *sendRates
3021− let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
3022− (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})
3023− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3024− maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
3103+ let rateResult = checkPubkeyRate sendRates senderKey maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
30253104 if rateResult.limited
30263105 then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}, globalCapReached: False {}, retryAfterMs: rateResult.retryAfterMs}
30273106 else do
30283107 set *sendRates = rateResult.entries
30293108 pairRates <- *pairRates
3030− let pairResult = checkRateWindow pairRates (\r -> r.sender == senderKey && r.recipient == to)
3031− (\ts -> {sender: senderKey, recipient: to, windowStart: ts, count: 1})
3032− (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
3033− maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
3109+ let pairResult = checkPairRate pairRates senderKey to maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
30343110 set *pairRates = pairResult.entries
30353111 if pairResult.limited
30363112 then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: pairResult.retryAfterMs}
30373113 else do
30383114 recvRates <- *recvRates
3039− let recvResult = checkRateWindow recvRates (\r -> r.pubkey == to)
3040− (\ts -> {pubkey: to, windowStart: ts, count: 1})
3041− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3042− maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
3115+ let recvResult = checkPubkeyRate recvRates to maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
30433116 if recvResult.limited
30443117 then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: recvResult.retryAfterMs}
30453118 else do
@@ -3132,10 +3205,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
31323205 (yield {server: "", authorized: False {}})
31333206 (\senderKey -> do
31343207 lookupRates <- *lookupRates
3135− let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == senderKey)
3136− (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})
3137− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3138− maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
3208+ let rateResult = checkPubkeyRate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
31393209 set *lookupRates = rateResult.entries
31403210 if rateResult.limited
31413211 then yield {server: "", authorized: True {}}
@@ -3164,10 +3234,7 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
31643234 -- Per-key rate limiting (reuses lookup rate limiter — both are info lookups)
31653235 statusRateLimited <- atomic do
31663236 lookupRates <- *lookupRates
3167− let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == senderKey)
3168− (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})
3169− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3170− maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
3237+ let rateResult = checkPubkeyRate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
31713238 set *lookupRates = rateResult.entries
31723239 yield rateResult.limited
31733240 if statusRateLimited then yield (respond {status: "", authorized: True {}})
@@ -3215,10 +3282,7 @@ handleRevoke = \targetToken authorization clientIp respond -> do
32153282 else do
32163283 revokeLimited <- atomic do
32173284 challengeRates <- *challengeRates
3218− let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == callerPubkey)
3219− (\ts -> {pubkey: callerPubkey, windowStart: ts, count: 1})
3220− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3221− maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
3285+ let rateResult = checkPubkeyRate challengeRates callerPubkey maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
32223286 set *challengeRates = rateResult.entries
32233287 yield rateResult.limited
32243288 if revokeLimited then yield (respond {ok: False {}, authorized: True {}})
@@ -3306,10 +3370,7 @@ handleVerifyToken = \token scope clientIp respond -> do
33063370 verifyLimited <- if clientIp == "" then yield (False {})
33073371 else atomic do
33083372 verifyIpRates <- *verifyIpRates
3309− let result = checkRateWindow verifyIpRates (\r -> r.ip == clientIp)
3310− (\ts -> {ip: clientIp, windowStart: ts, count: 1})
3311− (\r -> {ip: r.ip, windowStart: r.windowStart, count: r.count + 1})
3312− maxVerifyRatePerIp verifyRateWindowMs maxVerifyRateEntries t
3373+ let result = checkIpRate verifyIpRates clientIp maxVerifyRatePerIp verifyRateWindowMs maxVerifyRateEntries t
33133374 set *verifyIpRates = result.entries
33143375 yield result.limited
33153376 if verifyLimited then do
@@ -3475,19 +3536,13 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
34753536 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
34763537 else do
34773538 recvRates <- *recvRates
3478− let recvResult = checkRateWindow recvRates (\r -> r.pubkey == toKey)
3479− (\ts -> {pubkey: toKey, windowStart: ts, count: 1})
3480− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3481− maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
3539+ let recvResult = checkPubkeyRate recvRates toKey maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
34823540 if recvResult.limited
34833541 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: True {}}
34843542 else do
34853543 set *recvRates = recvResult.entries
34863544 federationPairRates <- *federationPairRates
3487− let fedPairResult = checkRateWindow federationPairRates (\r -> r.server == fromServer && r.recipient == toKey)
3488− (\ts -> {server: fromServer, recipient: toKey, windowStart: ts, count: 1})
3489− (\r -> {server: r.server, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
3490− maxFederationPairRatePerPair federationPairRateWindowMs maxFederationPairRateEntries t
3545+ let fedPairResult = checkServerRecipientRate federationPairRates fromServer toKey maxFederationPairRatePerPair federationPairRateWindowMs maxFederationPairRateEntries t
34913546 if fedPairResult.limited
34923547 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
34933548 else do
@@ -3504,14 +3559,13 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
35043559 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
35053560 fwdNonces <- *forwardNonces
35063561 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
3507− -- Per-origin eviction: when at capacity, only evict nonces from the
3508− -- origin server that's causing the overflow. Prevents one peer from
3509− -- force-evicting another peer's nonces via targeted flooding.
3562+ -- Global age-based eviction: evict oldest nonces regardless of origin.
3563+ -- Previous per-originServer partitioning could be gamed by a relay peer
3564+ -- forging varied originServer values to spread nonces across partitions
3565+ -- and crowd out legitimate entries.
35103566 set *forwardNonces = if count updatedNonces > maxForwardNonces
3511− then let fromOrigin = do n <- updatedNonces; where n.originServer == originServer; yield n in
3512− let others = do n <- updatedNonces; where n.originServer != originServer; yield n in
3513− let evictCount = max 1 (count fromOrigin / 10) in
3514− union others (evictSmallest fromOrigin evictCount (\n -> n.seenAt)).kept
3567+ then let evictCount = max 1 (count updatedNonces / 10) in
3568+ (evictSmallest updatedNonces evictCount (\n -> n.seenAt)).kept
35153569 else updatedNonces
35163570 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
35173571 case result.forward of
@@ -3549,8 +3603,12 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp autho
35493603 when (ackedCount > 0) (atomic do
35503604 cleanups <- *pendingAckCleanups
35513605 let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
3552− let combined = union cleanups newCleanups
3553− set *pendingAckCleanups = if count combined > maxPendingAckCleanups then takeRelation maxPendingAckCleanups combined else combined)
3606+ -- Per-key cap: limit cleanups per recipient to prevent one noisy key from dominating the queue
3607+ let perKeyCapped = fold (\acc c ->
3608+ let existingForKey = count (do x <- acc; where x.toKey == c.toKey; yield x)
3609+ in if existingForKey >= maxPendingAckCleanupsPerKey then acc else union acc [c])
3610+ cleanups newCleanups
3611+ set *pendingAckCleanups = if count perKeyCapped > maxPendingAckCleanups then takeRelation maxPendingAckCleanups perKeyCapped else perKeyCapped)
35543612 yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
35553613
35563614-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
@@ -3645,6 +3703,7 @@ handleHealth = \authorization respond -> do
36453703 lastFedSuccess <- *lastFederationSuccessAt
36463704 forwardFailures <- *forwardFailures
36473705 pruneRun <- *pruneRunning
3706+ lastSnap <- *lastSnapshotExportAt
36483707 let warmup = t - sa < healthPruneMaxAge
36493708 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
36503709 -- Capacity pressure: flag when any capped collection exceeds 90% full
@@ -3657,21 +3716,30 @@ handleHealth = \authorization respond -> do
36573716 let fedOk = not federationEnabled || warmup
36583717 || count forwardFailures == 0
36593718 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
3719+ -- Snapshot staleness: unhealthy if maintenance secret is configured but no snapshot
3720+ -- has been exported within the staleness window (federation state will be lost on restart)
3721+ let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup
3722+ || (lastSnap > 0 && t - lastSnap <= snapshotStalenessWarning)
3723+ || (tmc == 0 && presenceCount == 0 && count forwards == 0)
36603724 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
36613725 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
36623726 capacityOk: capacityOk,
3663− federationOk: fedOk}
3664− let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk
3665− -- Unauthenticated: return only aggregate ok flag to avoid leaking capacity/federation details
3727+ federationOk: fedOk,
3728+ snapshotOk: snapOk}
3729+ let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk
3730+ -- Unauthenticated: always return ok:true to prevent information leakage.
3731+ -- Attackers could correlate ok:false transitions with targeted load to infer subsystem pressure.
36663732 if isAuthed
36673733 then yield (respond {ok: allOk, pruneHealthy: result.pruneOk,
36683734 heartbeatHealthy: result.heartbeatOk,
36693735 capacityHealthy: result.capacityOk,
3670− federationHealthy: result.federationOk})
3671− else yield (respond {ok: allOk, pruneHealthy: allOk,
3672− heartbeatHealthy: allOk,
3673− capacityHealthy: allOk,
3674− federationHealthy: allOk})
3736+ federationHealthy: result.federationOk,
3737+ snapshotHealthy: result.snapshotOk})
3738+ else yield (respond {ok: True {}, pruneHealthy: True {},
3739+ heartbeatHealthy: True {},
3740+ capacityHealthy: True {},
3741+ federationHealthy: True {},
3742+ snapshotHealthy: True {}})
36753743
36763744zeroStats = {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pairRates: 0, federationPairRates: 0, pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, hungFederationSlots: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, drainCompletedAt: 0, oldestForwardAge: 0, oldestAckAge: 0, ok: False {}}
36773745
@@ -3879,10 +3947,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
38793947 (\pubkey -> do
38803948 ackRateLimited <- atomic do
38813949 lookupRates <- *lookupRates
3882− let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == pubkey)
3883− (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
3884− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3885− maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
3950+ let rateResult = checkPubkeyRate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
38863951 set *lookupRates = rateResult.entries
38873952 yield rateResult.limited
38883953 if ackRateLimited then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0})
@@ -3928,10 +3993,7 @@ handleListMessages = \authorization clientIp respond -> do
39283993 (\pubkey -> do
39293994 listRateLimited <- atomic do
39303995 lookupRates <- *lookupRates
3931− let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == pubkey)
3932− (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
3933− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3934− maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
3996+ let rateResult = checkPubkeyRate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
39353997 set *lookupRates = rateResult.entries
39363998 yield rateResult.limited
39373999 if listRateLimited then yield (respond {messages: [], authorized: True {}, count: 0})
@@ -4018,27 +4080,18 @@ sendBatchInner = \senderKey messages t acc ->
40184080 if isDup then yield {status: MsDuplicate {}, fwdServer: "", needsFwd: False {}, retryMs: 0}
40194081 else do
40204082 sendRates <- *sendRates
4021− let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
4022− (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})
4023− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
4024− maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
4083+ let rateResult = checkPubkeyRate sendRates senderKey maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
40254084 if rateResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: rateResult.retryAfterMs}
40264085 else do
40274086 set *sendRates = rateResult.entries
40284087 recvRates <- *recvRates
4029− let recvResult = checkRateWindow recvRates (\r -> r.pubkey == msg.to)
4030− (\ts -> {pubkey: msg.to, windowStart: ts, count: 1})
4031− (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
4032− maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
4088+ let recvResult = checkPubkeyRate recvRates msg.to maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
40334089 if recvResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: recvResult.retryAfterMs}
40344090 else do
40354091 set *recvRates = recvResult.entries
40364092 -- Per-sender-per-recipient pair rate limiting (prevents one sender filling a recipient's mailbox)
40374093 pairRates <- *pairRates
4038− let pairResult = checkRateWindow pairRates (\r -> r.sender == senderKey && r.recipient == msg.to)
4039− (\ts -> {sender: senderKey, recipient: msg.to, windowStart: ts, count: 1})
4040− (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
4041− maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
4094+ let pairResult = checkPairRate pairRates senderKey msg.to maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
40424095 if pairResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: pairResult.retryAfterMs}
40434096 else do
40444097 set *pairRates = pairResult.entries
@@ -4080,6 +4133,142 @@ sendBatchInner = \senderKey messages t acc ->
40804133 yield {messageId: effectiveMsgId, status: sendResult.status, needsForward: sendResult.needsFwd, forwardServer: sendResult.fwdServer, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryMs}
40814134 sendBatchInner senderKey split.rest t (union acc [result])
40824135
4136+-- POST /maintenance/block: Add a server to the federation blocklist.
4137+-- Blocked peers are rejected from all inbound and outbound federation operations.
4138+handleBlockPeer = \server authorization respond -> do
4139+ let authorized = verifyMaintenanceAuth authorization
4140+ if not authorized then yield (respond {ok: False {}, blocked: 0})
4141+ else do
4142+ result <- atomic do
4143+ blocked <- *blockedPeers
4144+ if isBlockedPeer server blocked then yield {alreadyBlocked: True {}, total: count blocked}
4145+ else do
4146+ let updated = if count blocked >= maxBlockedPeers
4147+ then union (evictOneOldest blocked (\_ -> 0)) [server]
4148+ else union blocked [server]
4149+ set *blockedPeers = updated
4150+ yield {alreadyBlocked: False {}, total: count updated}
4151+ when (not result.alreadyBlocked) (logInfo ("block: added " ++ server ++ " to blocklist (" ++ show result.total ++ " total)"))
4152+ yield (respond {ok: True {}, blocked: result.total})
4153+
4154+-- POST /maintenance/unblock: Remove a server from the federation blocklist.
4155+handleUnblockPeer = \server authorization respond -> do
4156+ let authorized = verifyMaintenanceAuth authorization
4157+ if not authorized then yield (respond {ok: False {}, unblocked: False {}})
4158+ else do
4159+ removed <- atomic do
4160+ blocked <- *blockedPeers
4161+ let had = isBlockedPeer server blocked
4162+ set *blockedPeers = do b <- blocked; where b != server; yield b
4163+ yield had
4164+ when removed (logInfo ("unblock: removed " ++ server ++ " from blocklist"))
4165+ yield (respond {ok: True {}, unblocked: removed})
4166+
4167+-- GET /maintenance/blocked: List currently blocked federation peers.
4168+handleListBlocked = \authorization respond -> do
4169+ let authorized = verifyMaintenanceAuth authorization
4170+ if not authorized then yield (respond {peers: [], ok: False {}})
4171+ else do
4172+ blocked <- *blockedPeers
4173+ yield (respond {peers: blocked, ok: True {}})
4174+
4175+-- POST /maintenance/ping: Test federation connectivity to a specific peer.
4176+-- Does a verify callback round-trip to measure latency and confirm reachability.
4177+handlePingPeer = \server authorization respond -> do
4178+ let authorized = verifyMaintenanceAuth authorization
4179+ if not authorized then yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"})
4180+ else do
4181+ token <- genToken
4182+ t <- now
4183+ atomic do
4184+ pending <- *pendingVerifications
4185+ set *pendingVerifications = union pending [{token: token, createdAt: t, scope: "ping"}]
4186+ startTime <- now
4187+ result <- fetch (federationUrl server) (VerifyToken {token: token, scope: "ping"})
4188+ endTime <- now
4189+ let latency = endTime - startTime
4190+ case result of
4191+ Err e -> do
4192+ logWarn ("ping: " ++ server ++ " failed: " ++ show e)
4193+ yield (respond {ok: False {}, latencyMs: latency, error: show e})
4194+ Ok r -> yield (respond {ok: r.ok, latencyMs: latency, error: ""})
4195+
4196+-- GET /maintenance/sessions/{pubkey}: List sessions for a specific pubkey.
4197+-- Returns masked tokens (first 8 chars only) for identification without exposing full tokens.
4198+handleAdminListSessions = \pubkey authorization respond -> do
4199+ let authorized = verifyMaintenanceAuth authorization
4200+ if not authorized then yield (respond {sessions: [], count: 0, ok: False {}})
4201+ else do
4202+ result <- atomic do
4203+ allSessions <- readAllSessions
4204+ let keySessions = do s <- allSessions; where s.pubkey == pubkey; yield {token: take 8 s.token ++ "...", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt}
4205+ yield {sessions: keySessions, total: count keySessions}
4206+ yield (respond {sessions: result.sessions, count: result.total, ok: True {}})
4207+
4208+-- POST /maintenance/sessions/revoke-all: Revoke all sessions for a pubkey (abuse response).
4209+-- Also clears local presence and gossips offline if federation is enabled.
4210+handleAdminRevokeSessions = \pubkey authorization respond -> do
4211+ let authorized = verifyMaintenanceAuth authorization
4212+ if not authorized then yield (respond {ok: False {}, revokedCount: 0})
4213+ else do
4214+ t <- now
4215+ result <- atomic do
4216+ allSessions <- readAllSessions
4217+ let revokedCount = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
4218+ let remaining = do s <- allSessions; where s.pubkey != pubkey; yield s
4219+ -- Single-pass partition by token bucket
4220+ let partitioned = fold (\acc s ->
4221+ let b = tokenBucket s.token in
4222+ {b0: if b == 0 then union [s] acc.b0 else acc.b0,
4223+ b1: if b == 1 then union [s] acc.b1 else acc.b1,
4224+ b2: if b == 2 then union [s] acc.b2 else acc.b2,
4225+ b3: if b == 3 then union [s] acc.b3 else acc.b3,
4226+ b4: if b == 4 then union [s] acc.b4 else acc.b4,
4227+ b5: if b == 5 then union [s] acc.b5 else acc.b5,
4228+ b6: if b == 6 then union [s] acc.b6 else acc.b6,
4229+ b7: if b == 7 then union [s] acc.b7 else acc.b7,
4230+ b8: if b == 8 then union [s] acc.b8 else acc.b8,
4231+ b9: if b == 9 then union [s] acc.b9 else acc.b9,
4232+ b10: if b == 10 then union [s] acc.b10 else acc.b10,
4233+ b11: if b == 11 then union [s] acc.b11 else acc.b11,
4234+ b12: if b == 12 then union [s] acc.b12 else acc.b12,
4235+ b13: if b == 13 then union [s] acc.b13 else acc.b13,
4236+ b14: if b == 14 then union [s] acc.b14 else acc.b14,
4237+ b15: if b == 15 then union [s] acc.b15 else acc.b15})
4238+ {b0: [], b1: [], b2: [], b3: [], b4: [], b5: [], b6: [], b7: [], b8: [], b9: [], b10: [], b11: [], b12: [], b13: [], b14: [], b15: []} remaining
4239+ writeSessionBucket 0 partitioned.b0
4240+ writeSessionBucket 1 partitioned.b1
4241+ writeSessionBucket 2 partitioned.b2
4242+ writeSessionBucket 3 partitioned.b3
4243+ writeSessionBucket 4 partitioned.b4
4244+ writeSessionBucket 5 partitioned.b5
4245+ writeSessionBucket 6 partitioned.b6
4246+ writeSessionBucket 7 partitioned.b7
4247+ writeSessionBucket 8 partitioned.b8
4248+ writeSessionBucket 9 partitioned.b9
4249+ writeSessionBucket 10 partitioned.b10
4250+ writeSessionBucket 11 partitioned.b11
4251+ writeSessionBucket 12 partitioned.b12
4252+ writeSessionBucket 13 partitioned.b13
4253+ writeSessionBucket 14 partitioned.b14
4254+ writeSessionBucket 15 partitioned.b15
4255+ set *totalSessionCount = count remaining
4256+ -- Clear local presence
4257+ let pb = pubkeyBucket pubkey
4258+ presenceBkt <- readPresenceBucket pb
4259+ let presenceRemoved = count presenceBkt - count (do p <- presenceBkt; where p.pubkey != pubkey || not (isLocalServer p.server); yield p)
4260+ writePresenceBucket pb (do p <- presenceBkt; where p.pubkey != pubkey || not (isLocalServer p.server); yield p)
4261+ tpc <- *totalPresenceCount
4262+ set *totalPresenceCount = max 0 (tpc - presenceRemoved)
4263+ when (presenceRemoved > 0) (set *presenceVersion = t)
4264+ peerServers <- *knownPeerServers
4265+ yield {revokedCount: revokedCount, peerServers: peerServers, hadPresence: presenceRemoved > 0}
4266+ when (result.revokedCount > 0)
4267+ (logInfo ("admin: revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
4268+ when (result.hadPresence && federationEnabled && count result.peerServers > 0)
4269+ (fork (sendBatchedGossip result.peerServers [{eventType: GossipOffline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))
4270+ yield (respond {ok: True {}, revokedCount: result.revokedCount})
4271+
40834272serve = \req -> case req of
40844273 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
40854274 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
@@ -4101,6 +4290,12 @@ serve = \req -> case req of
41014290 Undrain {authorization, respond} -> handleUndrain authorization respond
41024291 Snapshot {authorization, respond} -> handleSnapshot authorization respond
41034292 Restore {version, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, authorization, respond} -> handleRestore version presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces authorization respond
4293+ BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
4294+ UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
4295+ ListBlocked {authorization, respond} -> handleListBlocked authorization respond
4296+ PingPeer {server, authorization, respond} -> handlePingPeer server authorization respond
4297+ AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
4298+ AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
41044299 Health {authorization, respond} -> handleHealth authorization respond
41054300 Stats {authorization, respond} -> handleStats authorization respond
41064301 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
@@ -4200,6 +4395,9 @@ pollHeartbeat = do
42004395
42014396-- Background loop: periodically logs a state summary to stdout.
42024397-- Operators can capture stdout to detect state drift; use /maintenance/snapshot for full export.
4398+-- TODO: If Knot adds file I/O primitives, auto-export snapshots to a file
4399+-- (e.g., /var/lib/skrepka/snapshot.json) in addition to logging. Currently
4400+-- a crash between manual snapshot exports loses all federation state.
42034401backgroundSnapshot = do
42044402 sleep autoSnapshotInterval
42054403 draining <- *draining
@@ -4249,6 +4447,15 @@ backgroundWatchdog = do
42494447 when (t - lastHb > healthHeartbeatMaxAge * 3) (do
42504448 logError "watchdog: heartbeat loop appears dead, restarting"
42514449 fork pollHeartbeat)
4450+ -- Recover leaked federation slots: if prune is alive but in-flight counters
4451+ -- are stuck, force-reconcile. This catches edge cases where panicked fetches
4452+ -- leaked slots but prune's stale threshold hasn't reclaimed them yet.
4453+ fedActive <- *federationActiveConnections
4454+ when (fedActive > maxGlobalFederationConcurrency) (do
4455+ logWarn ("watchdog: federation connection counter (" ++ show fedActive ++ ") exceeds max (" ++ show maxGlobalFederationConcurrency ++ "), reconciling")
4456+ atomic do
4457+ inFlight <- *federationInFlight
4458+ set *federationActiveConnections = fold (\acc r -> acc + r.count) 0 inFlight)
42524459 backgroundWatchdog
42534460
42544461main = do