@@ -148,6 +148,8 @@ gossipEventRateWindowMs = 60000 -- 1 minute
148maxGossipEventRateEntries = 100
148maxGossipEventRateEntries = 100
149-- Global federation connection cap (bounds total concurrent outbound connections across all paths)
149-- Global federation connection cap (bounds total concurrent outbound connections across all paths)
150maxGlobalFederationConcurrency = 30
150maxGlobalFederationConcurrency = 30
151+-- Peer blocklist: max blocked servers tracked at runtime (managed via /maintenance/block, /maintenance/unblock)
152+maxBlockedPeers = 100
152-- Background task intervals
154-- Background task intervals
153backgroundPruneInterval : Int<Ms>
155backgroundPruneInterval : Int<Ms>
@@ -170,6 +172,8 @@ federationEnabled = True {}
170-- ServerName validation time. A hostname passing isBadServerName could rebind to a
172-- ServerName validation time. A hostname passing isBadServerName could rebind to a
171-- private IP between validation and fetch. Mitigate at the network layer (e.g.,
173-- private IP between validation and fetch. Mitigate at the network layer (e.g.,
172-- firewall rules blocking outbound connections to RFC1918 ranges from the server process).
174-- 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.
173federationUrl = \server -> federationScheme ++ server
177federationUrl = \server -> federationScheme ++ server
174-- Callback verification: outbound federation requests include a random single-use token;
178-- Callback verification: outbound federation requests include a random single-use token;
175-- the receiver calls back to /federation/verify to confirm the sender actually issued it.
179-- the receiver calls back to /federation/verify to confirm the sender actually issued it.
@@ -199,6 +203,8 @@ ackGracePeriod : Int<Ms>
199ackGracePeriod = 30000 -- 30 seconds
203ackGracePeriod = 30000 -- 30 seconds
200-- Deferred ack cleanup capacity
204-- Deferred ack cleanup capacity
201maxPendingAckCleanups = 10000
205maxPendingAckCleanups = 10000
206+-- Per-key deferred ack cleanup cap (prevents one noisy key from dominating the queue)
207+maxPendingAckCleanupsPerKey = 100
203-- Health check thresholds (background task freshness)
209-- Health check thresholds (background task freshness)
204healthPruneMaxAge : Int<Ms>
210healthPruneMaxAge : Int<Ms>
@@ -332,6 +338,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
333-- Sessions sharded into 16 buckets by first hex char of token.
339-- Sessions sharded into 16 buckets by first hex char of token.
334-- Hot-path auth scans only one bucket (~1/16 of sessions).
340-- 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.
335*sessionBucket0 : [Session]
345*sessionBucket0 : [Session]
336*sessionBucket1 : [Session]
346*sessionBucket1 : [Session]
337*sessionBucket2 : [Session]
347*sessionBucket2 : [Session]
@@ -533,6 +543,9 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
533-- Global federation connection counter (tracks total concurrent outbound connections)
543-- Global federation connection counter (tracks total concurrent outbound connections)
534*federationActiveConnections : Int
544*federationActiveConnections : Int
546+-- Blocked federation peers (managed at runtime via /maintenance/block and /maintenance/unblock)
547+*blockedPeers : [Text]
536-- Deferred ack cleanups: origin message reclamation queued by federation ack handlers,
549-- Deferred ack cleanups: origin message reclamation queued by federation ack handlers,
537-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
550-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
538*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
551*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
@@ -640,11 +653,38 @@ route MaintenanceApi where
640 POST {version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce]} /maintenance/restore headers {authorization: Text}
653 POST {version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce]} /maintenance/restore headers {authorization: Text}
641 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
654 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
642 = Restore
655 = 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}
660+ -- Unblock a previously blocked peer
661+ POST {server: ServerName} /maintenance/unblock headers {authorization: Text}
662+ -> {ok: Bool, unblocked: Bool}
664+ -- List currently blocked peers
665+ GET /maintenance/blocked headers {authorization: Text}
666+ -> {peers: [Text], ok: Bool}
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}
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}
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
644route HealthApi where
681route 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
647-- Returns counts of in-memory collections (requires maintenance secret)
684-- 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.
648route StatsApi where
688route StatsApi where
649 GET /stats headers {authorization: Text}
689 GET /stats headers {authorization: Text}
650 -> {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}
690 -> {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 ->
934verifyMaintenanceAuth : Text -> Bool
974verifyMaintenanceAuth : Text -> Bool
935verifyMaintenanceAuth = verifyAuthWith (\token -> verifySecret token maintenanceSecret)
975verifyMaintenanceAuth = verifyAuthWith (\token -> verifySecret token maintenanceSecret)
977+-- Check if a server is on the blocklist
978+isBlockedPeer : Text -> [Text] -> Bool
979+isBlockedPeer = \server blocked -> any blocked (\b -> b == server)
937-- Generate a single-use token for outbound federation auth.
981-- Generate a single-use token for outbound federation auth.
938-- The receiving server calls back to /federation/verify to confirm we issued it.
982-- The receiving server calls back to /federation/verify to confirm we issued it.
939-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
983-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
@@ -1405,6 +1449,9 @@ reconcileMessageCounts = \b acc ->
1405-- Recipient Count Helpers
1449-- Recipient Count Helpers
1406-- ============================================
1450-- ============================================
1407-- Per-recipient message counts avoid O(n) scan of all messages on send/forward.
1451-- 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.
1409getRecipientCount : PubkeyHex -> [RecipientCount] -> Int
1456getRecipientCount : PubkeyHex -> [RecipientCount] -> Int
1410getRecipientCount = \pubkey counts ->
1457getRecipientCount = \pubkey counts ->
@@ -1427,6 +1474,8 @@ adjustRecipientCount = \counts pubkey delta ->
1428-- Count items per toKey in a single pass: fold accumulates {pubkey, count} pairs,
1475-- Count items per toKey in a single pass: fold accumulates {pubkey, count} pairs,
1429-- incrementing existing entries or inserting new ones.
1476-- 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).
1430countByRecipient = \items ->
1479countByRecipient = \items ->
1431 fold (\acc x ->
1480 fold (\acc x ->
1432 if any acc (\c -> c.pubkey == x.toKey)
1481 if any acc (\c -> c.pubkey == x.toKey)
@@ -1575,6 +1624,38 @@ checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntr
1575 else do r <- entries; yield (if matchKey r then updateCount r else r)
1624 else do r <- entries; yield (if matchKey r then updateCount r else r)
1576 in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs; yield r) else newEntries, retryAfterMs: 0}
1625 in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs; yield r) else newEntries, retryAfterMs: 0}
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
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
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
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
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
1578-- Extract the item with the smallest key and return both the item and remaining list
1659-- Extract the item with the smallest key and return both the item and remaining list
1579extractOldest = \items getKey ->
1660extractOldest = \items getKey ->
1580 fold (\acc x ->
1661 fold (\acc x ->
@@ -1838,10 +1919,12 @@ pruneExpired = \t -> do
1838 set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
1919 set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
1839 -- Clean up stale in-flight entries: remove zeroed entries and reset leaked slots
1920 -- Clean up stale in-flight entries: remove zeroed entries and reset leaked slots
1840 -- where the forked task died without calling clearFederationInFlight.
1921 -- 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
1842 -- after the skip-hung-server logic (at 1x) has already taken effect.
1923 -- 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.
1843 inFlight <- *federationInFlight
1926 inFlight <- *federationInFlight
1844− let staleThreshold = federationFetchTimeoutMs * 3 / 2
1927+ let staleThreshold = federationFetchTimeoutMs * 5 / 4
1845 let staleInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= staleThreshold; yield r
1928 let staleInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= staleThreshold; yield r
1846 when (count staleInFlight > 0)
1929 when (count staleInFlight > 0)
1847 (logWarn ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))
1930 (logWarn ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))
@@ -1931,10 +2014,7 @@ mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
1932checkFederationRate = \server t -> atomic do
2015checkFederationRate = \server t -> atomic do
1933 federationRates <- *federationRates
2016 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
1938 set *federationRates = result.entries
2018 set *federationRates = result.entries
1939 yield result.limited
2019 yield result.limited
@@ -1972,10 +2052,7 @@ checkIpRate = \ip t ->
1972 if ip == "" then yield (False {})
2052 if ip == "" then yield (False {})
1973 else atomic do
2053 else atomic do
1974 ipRates <- *ipRates
2054 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
1979 set *ipRates = result.entries
2056 set *ipRates = result.entries
1980 yield result.limited
2057 yield result.limited
@@ -1992,6 +2069,13 @@ checkPreAuth = \clientIp t -> do
1992 else yield ""
2069 else yield ""
1994withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do
2071withFederationAuth = \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
2076+ logWarn ("federation: blocked peer " ++ fromServer ++ " rejected")
2077+ yield (respond {ok: False {}, error: "blocked"})
1995 verified <- verifyFederationCallback fromServer authorization scope
2079 verified <- verifyFederationCallback fromServer authorization scope
1996 if not verified
2080 if not verified
1997 then yield (respond {ok: False {}, error: "unauthorized"})
2081 then yield (respond {ok: False {}, error: "unauthorized"})
@@ -2099,10 +2183,10 @@ shouldRetryServer = \server t failures ->
2099 Nothing {} -> True {}
2183 Nothing {} -> True {}
2100 Just {value: f} -> t - f.firstFailedAt < maxServerDowntime && t - f.failedAt >= retryBackoff f.failures
2184 Just {value: f} -> t - f.firstFailedAt < maxServerDowntime && t - f.failedAt >= retryBackoff f.failures
2102−-- Extended retry check: also skip servers with hung in-flight requests
2186+-- Extended retry check: also skip servers with hung in-flight requests or on the blocklist
2103−shouldRetryServerEx : Text -> Timestamp -> [ForwardFailure] -> [FederationInFlightEntry] -> Bool
2187+shouldRetryServerEx : Text -> Timestamp -> [ForwardFailure] -> [FederationInFlightEntry] -> [Text] -> Bool
2104−shouldRetryServerEx = \server t failures inFlight ->
2188+shouldRetryServerEx = \server t failures inFlight blocked ->
2105− shouldRetryServer server t failures && not (hasHungFederationRequest server t inFlight)
2189+ not (isBlockedPeer server blocked) && shouldRetryServer server t failures && not (hasHungFederationRequest server t inFlight)
2107-- Global circuit breaker: returns True when too many peers are failing simultaneously,
2191-- Global circuit breaker: returns True when too many peers are failing simultaneously,
2108-- indicating a likely local network issue rather than individual peer problems.
2192-- indicating a likely local network issue rather than individual peer problems.
@@ -2180,7 +2264,8 @@ retryPendingForwards = \pending -> do
2180 then logWarn ("federation: circuit breaker open (" ++ show (count failures) ++ " peers failing), skipping forward retries")
2264 then logWarn ("federation: circuit breaker open (" ++ show (count failures) ++ " peers failing), skipping forward retries")
2182 inFlight <- *federationInFlight
2266 inFlight <- *federationInFlight
2267+ blocked <- *blockedPeers
2183− let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2268+ let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
2184 sendToEach eligibleServers (\server ->
2269 sendToEach eligibleServers (\server ->
2185 retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)
2270 retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)
2186 yield {}
2271 yield {}
@@ -2208,7 +2293,8 @@ forwardQueuedMessages = \targets t -> do
2208 then logWarn ("federation: circuit breaker open, skipping queued message forwarding")
2293 then logWarn ("federation: circuit breaker open, skipping queued message forwarding")
2210 inFlight <- *federationInFlight
2295 inFlight <- *federationInFlight
2296+ blocked <- *blockedPeers
2211− let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2297+ let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
2212 sendToEach eligibleServers (\server ->
2298 sendToEach eligibleServers (\server ->
2213 retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)
2299 retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)
@@ -2263,7 +2349,8 @@ sendAcksAndClean = \pendingAcks -> do
2263 then logWarn ("federation: circuit breaker open, skipping ack retries")
2349 then logWarn ("federation: circuit breaker open, skipping ack retries")
2265 inFlight <- *federationInFlight
2351 inFlight <- *federationInFlight
2352+ blocked <- *blockedPeers
2266− let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2353+ let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
2267 sendToEach eligibleServers (\server ->
2354 sendToEach eligibleServers (\server ->
2268 sendBatchedAcksToServer (acksForServer pendingAcks server) server)
2355 sendBatchedAcksToServer (acksForServer pendingAcks server) server)
@@ -2322,7 +2409,8 @@ retryOverflowAcks = \overflow -> do
2322 then logWarn ("federation: circuit breaker open, skipping overflow ack retries")
2409 then logWarn ("federation: circuit breaker open, skipping overflow ack retries")
2324 inFlight <- *federationInFlight
2411 inFlight <- *federationInFlight
2412+ blocked <- *blockedPeers
2325− let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2413+ let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
2326 sendToEach eligibleServers (\server ->
2414 sendToEach eligibleServers (\server ->
2327 forEachBatch (acksForServer overflow server) maxMessageIds (\batch -> do
2415 forEachBatch (acksForServer overflow server) maxMessageIds (\batch -> do
2328 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
2416 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
@@ -2797,10 +2885,7 @@ handleChallenge = \pubkey clientIp respond -> do
2797 then yield {accepted: False {}, rateLimited: False {}, retryAfterMs: 0}
2885 then yield {accepted: False {}, rateLimited: False {}, retryAfterMs: 0}
2799 challengeRates <- *challengeRates
2887 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
2804 if rateResult.limited
2889 if rateResult.limited
2805 then yield {accepted: False {}, rateLimited: True {}, retryAfterMs: rateResult.retryAfterMs}
2890 then yield {accepted: False {}, rateLimited: True {}, retryAfterMs: rateResult.retryAfterMs}
@@ -2826,10 +2911,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
2827 authRateLimited <- atomic do
2912 authRateLimited <- atomic do
2828 challengeRates <- *challengeRates
2913 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
2833 set *challengeRates = rateResult.entries
2915 set *challengeRates = rateResult.entries
2834 yield rateResult.limited
2916 yield rateResult.limited
2835 if authRateLimited then do
2917 if authRateLimited then do
@@ -3018,28 +3100,19 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
3018 then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}
3100 then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}
3020 sendRates <- *sendRates
3102 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
3025 if rateResult.limited
3104 if rateResult.limited
3026 then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}, globalCapReached: False {}, retryAfterMs: rateResult.retryAfterMs}
3105 then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}, globalCapReached: False {}, retryAfterMs: rateResult.retryAfterMs}
3028 set *sendRates = rateResult.entries
3107 set *sendRates = rateResult.entries
3029 pairRates <- *pairRates
3108 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
3034 set *pairRates = pairResult.entries
3110 set *pairRates = pairResult.entries
3035 if pairResult.limited
3111 if pairResult.limited
3036 then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: pairResult.retryAfterMs}
3112 then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: pairResult.retryAfterMs}
3038 recvRates <- *recvRates
3114 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
3043 if recvResult.limited
3116 if recvResult.limited
3044 then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: recvResult.retryAfterMs}
3117 then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: recvResult.retryAfterMs}
@@ -3132,10 +3205,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
3132 (yield {server: "", authorized: False {}})
3205 (yield {server: "", authorized: False {}})
3133 (\senderKey -> do
3206 (\senderKey -> do
3134 lookupRates <- *lookupRates
3207 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
3139 set *lookupRates = rateResult.entries
3209 set *lookupRates = rateResult.entries
3140 if rateResult.limited
3210 if rateResult.limited
3141 then yield {server: "", authorized: True {}}
3211 then yield {server: "", authorized: True {}}
@@ -3164,10 +3234,7 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
3164 -- Per-key rate limiting (reuses lookup rate limiter — both are info lookups)
3234 -- Per-key rate limiting (reuses lookup rate limiter — both are info lookups)
3165 statusRateLimited <- atomic do
3235 statusRateLimited <- atomic do
3166 lookupRates <- *lookupRates
3236 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
3171 set *lookupRates = rateResult.entries
3238 set *lookupRates = rateResult.entries
3172 yield rateResult.limited
3239 yield rateResult.limited
3173 if statusRateLimited then yield (respond {status: "", authorized: True {}})
3240 if statusRateLimited then yield (respond {status: "", authorized: True {}})
@@ -3215,10 +3282,7 @@ handleRevoke = \targetToken authorization clientIp respond -> do
3216 revokeLimited <- atomic do
3283 revokeLimited <- atomic do
3217 challengeRates <- *challengeRates
3284 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
3222 set *challengeRates = rateResult.entries
3286 set *challengeRates = rateResult.entries
3223 yield rateResult.limited
3287 yield rateResult.limited
3224 if revokeLimited then yield (respond {ok: False {}, authorized: True {}})
3288 if revokeLimited then yield (respond {ok: False {}, authorized: True {}})
@@ -3306,10 +3370,7 @@ handleVerifyToken = \token scope clientIp respond -> do
3306 verifyLimited <- if clientIp == "" then yield (False {})
3370 verifyLimited <- if clientIp == "" then yield (False {})
3307 else atomic do
3371 else atomic do
3308 verifyIpRates <- *verifyIpRates
3372 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
3313 set *verifyIpRates = result.entries
3374 set *verifyIpRates = result.entries
3314 yield result.limited
3375 yield result.limited
3315 if verifyLimited then do
3376 if verifyLimited then do
@@ -3475,19 +3536,13 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3475 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
3536 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
3477 recvRates <- *recvRates
3538 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
3482 if recvResult.limited
3540 if recvResult.limited
3483 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: True {}}
3541 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: True {}}
3485 set *recvRates = recvResult.entries
3543 set *recvRates = recvResult.entries
3486 federationPairRates <- *federationPairRates
3544 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
3491 if fedPairResult.limited
3546 if fedPairResult.limited
3492 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
3547 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
@@ -3504,14 +3559,13 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3504 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3559 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3505 fwdNonces <- *forwardNonces
3560 fwdNonces <- *forwardNonces
3506 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
3561 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
3507− -- Per-origin eviction: when at capacity, only evict nonces from the
3562+ -- Global age-based eviction: evict oldest nonces regardless of origin.
3508− -- origin server that's causing the overflow. Prevents one peer from
3563+ -- Previous per-originServer partitioning could be gamed by a relay peer
3509− -- force-evicting another peer's nonces via targeted flooding.
3564+ -- forging varied originServer values to spread nonces across partitions
3565+ -- and crowd out legitimate entries.
3510 set *forwardNonces = if count updatedNonces > maxForwardNonces
3566 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
3567+ then let evictCount = max 1 (count updatedNonces / 10) in
3514− union others (evictSmallest fromOrigin evictCount (\n -> n.seenAt)).kept
3568+ (evictSmallest updatedNonces evictCount (\n -> n.seenAt)).kept
3515 else updatedNonces
3569 else updatedNonces
3516 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
3570 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
3517 case result.forward of
3571 case result.forward of
@@ -3549,8 +3603,12 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp autho
3549 when (ackedCount > 0) (atomic do
3603 when (ackedCount > 0) (atomic do
3550 cleanups <- *pendingAckCleanups
3604 cleanups <- *pendingAckCleanups
3551 let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
3605 let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
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])
3552− let combined = union cleanups newCleanups
3610+ cleanups newCleanups
3553− set *pendingAckCleanups = if count combined > maxPendingAckCleanups then takeRelation maxPendingAckCleanups combined else combined)
3611+ set *pendingAckCleanups = if count perKeyCapped > maxPendingAckCleanups then takeRelation maxPendingAckCleanups perKeyCapped else perKeyCapped)
3554 yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
3612 yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
3556-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
3614-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
@@ -3645,6 +3703,7 @@ handleHealth = \authorization respond -> do
3645 lastFedSuccess <- *lastFederationSuccessAt
3703 lastFedSuccess <- *lastFederationSuccessAt
3646 forwardFailures <- *forwardFailures
3704 forwardFailures <- *forwardFailures
3647 pruneRun <- *pruneRunning
3705 pruneRun <- *pruneRunning
3706+ lastSnap <- *lastSnapshotExportAt
3648 let warmup = t - sa < healthPruneMaxAge
3707 let warmup = t - sa < healthPruneMaxAge
3649 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
3708 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
3650 -- Capacity pressure: flag when any capped collection exceeds 90% full
3709 -- Capacity pressure: flag when any capped collection exceeds 90% full
@@ -3657,21 +3716,30 @@ handleHealth = \authorization respond -> do
3657 let fedOk = not federationEnabled || warmup
3716 let fedOk = not federationEnabled || warmup
3658 || count forwardFailures == 0
3717 || count forwardFailures == 0
3659 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
3718 || (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)
3660 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
3724 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
3661 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
3725 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
3662 capacityOk: capacityOk,
3726 capacityOk: capacityOk,
3663− federationOk: fedOk}
3727+ federationOk: fedOk,
3728+ snapshotOk: snapOk}
3664− let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk
3729+ let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk
3665− -- Unauthenticated: return only aggregate ok flag to avoid leaking capacity/federation details
3730+ -- Unauthenticated: always return ok:true to prevent information leakage.
3731+ -- Attackers could correlate ok:false transitions with targeted load to infer subsystem pressure.
3666 if isAuthed
3732 if isAuthed
3667 then yield (respond {ok: allOk, pruneHealthy: result.pruneOk,
3733 then yield (respond {ok: allOk, pruneHealthy: result.pruneOk,
3668 heartbeatHealthy: result.heartbeatOk,
3734 heartbeatHealthy: result.heartbeatOk,
3669 capacityHealthy: result.capacityOk,
3735 capacityHealthy: result.capacityOk,
3670− federationHealthy: result.federationOk})
3736+ federationHealthy: result.federationOk,
3737+ snapshotHealthy: result.snapshotOk})
3671− else yield (respond {ok: allOk, pruneHealthy: allOk,
3738+ else yield (respond {ok: True {}, pruneHealthy: True {},
3672− heartbeatHealthy: allOk,
3739+ heartbeatHealthy: True {},
3673− capacityHealthy: allOk,
3740+ capacityHealthy: True {},
3674− federationHealthy: allOk})
3741+ federationHealthy: True {},
3742+ snapshotHealthy: True {}})
3676zeroStats = {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 {}}
3744zeroStats = {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 {}}
@@ -3879,10 +3947,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
3879 (\pubkey -> do
3947 (\pubkey -> do
3880 ackRateLimited <- atomic do
3948 ackRateLimited <- atomic do
3881 lookupRates <- *lookupRates
3949 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
3886 set *lookupRates = rateResult.entries
3951 set *lookupRates = rateResult.entries
3887 yield rateResult.limited
3952 yield rateResult.limited
3888 if ackRateLimited then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0})
3953 if ackRateLimited then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0})
@@ -3928,10 +3993,7 @@ handleListMessages = \authorization clientIp respond -> do
3928 (\pubkey -> do
3993 (\pubkey -> do
3929 listRateLimited <- atomic do
3994 listRateLimited <- atomic do
3930 lookupRates <- *lookupRates
3995 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
3935 set *lookupRates = rateResult.entries
3997 set *lookupRates = rateResult.entries
3936 yield rateResult.limited
3998 yield rateResult.limited
3937 if listRateLimited then yield (respond {messages: [], authorized: True {}, count: 0})
3999 if listRateLimited then yield (respond {messages: [], authorized: True {}, count: 0})
@@ -4018,27 +4080,18 @@ sendBatchInner = \senderKey messages t acc ->
4018 if isDup then yield {status: MsDuplicate {}, fwdServer: "", needsFwd: False {}, retryMs: 0}
4080 if isDup then yield {status: MsDuplicate {}, fwdServer: "", needsFwd: False {}, retryMs: 0}
4020 sendRates <- *sendRates
4082 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
4025 if rateResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: rateResult.retryAfterMs}
4084 if rateResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: rateResult.retryAfterMs}
4027 set *sendRates = rateResult.entries
4086 set *sendRates = rateResult.entries
4028 recvRates <- *recvRates
4087 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
4033 if recvResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: recvResult.retryAfterMs}
4089 if recvResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: recvResult.retryAfterMs}
4035 set *recvRates = recvResult.entries
4091 set *recvRates = recvResult.entries
4036 -- Per-sender-per-recipient pair rate limiting (prevents one sender filling a recipient's mailbox)
4092 -- Per-sender-per-recipient pair rate limiting (prevents one sender filling a recipient's mailbox)
4037 pairRates <- *pairRates
4093 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
4042 if pairResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: pairResult.retryAfterMs}
4095 if pairResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: pairResult.retryAfterMs}
4044 set *pairRates = pairResult.entries
4097 set *pairRates = pairResult.entries
@@ -4080,6 +4133,142 @@ sendBatchInner = \senderKey messages t acc ->
4080 yield {messageId: effectiveMsgId, status: sendResult.status, needsForward: sendResult.needsFwd, forwardServer: sendResult.fwdServer, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryMs}
4133 yield {messageId: effectiveMsgId, status: sendResult.status, needsForward: sendResult.needsFwd, forwardServer: sendResult.fwdServer, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryMs}
4081 sendBatchInner senderKey split.rest t (union acc [result])
4134 sendBatchInner senderKey split.rest t (union acc [result])
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})
4142+ result <- atomic do
4143+ blocked <- *blockedPeers
4144+ if isBlockedPeer server blocked then yield {alreadyBlocked: True {}, total: count blocked}
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})
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 {}})
4159+ removed <- atomic do
4160+ blocked <- *blockedPeers
4161+ let had = isBlockedPeer server blocked
4162+ set *blockedPeers = do b <- blocked; where b != server; yield b
4164+ when removed (logInfo ("unblock: removed " ++ server ++ " from blocklist"))
4165+ yield (respond {ok: True {}, unblocked: removed})
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 {}})
4172+ blocked <- *blockedPeers
4173+ yield (respond {peers: blocked, ok: True {}})
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"})
4184+ pending <- *pendingVerifications
4185+ set *pendingVerifications = union pending [{token: token, createdAt: t, scope: "ping"}]
4187+ result <- fetch (federationUrl server) (VerifyToken {token: token, scope: "ping"})
4189+ let latency = endTime - startTime
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: ""})
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 {}})
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 {}})
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})
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})
4083serve = \req -> case req of
4272serve = \req -> case req of
4084 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
4273 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
4085 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
4274 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
@@ -4101,6 +4290,12 @@ serve = \req -> case req of
4101 Undrain {authorization, respond} -> handleUndrain authorization respond
4290 Undrain {authorization, respond} -> handleUndrain authorization respond
4102 Snapshot {authorization, respond} -> handleSnapshot authorization respond
4291 Snapshot {authorization, respond} -> handleSnapshot authorization respond
4103 Restore {version, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, authorization, respond} -> handleRestore version presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces authorization respond
4292 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
4104 Health {authorization, respond} -> handleHealth authorization respond
4299 Health {authorization, respond} -> handleHealth authorization respond
4105 Stats {authorization, respond} -> handleStats authorization respond
4300 Stats {authorization, respond} -> handleStats authorization respond
4106 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
4301 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
@@ -4200,6 +4395,9 @@ pollHeartbeat = do
4201-- Background loop: periodically logs a state summary to stdout.
4396-- Background loop: periodically logs a state summary to stdout.
4202-- Operators can capture stdout to detect state drift; use /maintenance/snapshot for full export.
4397-- 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.
4203backgroundSnapshot = do
4401backgroundSnapshot = do
4204 sleep autoSnapshotInterval
4402 sleep autoSnapshotInterval
4205 draining <- *draining
4403 draining <- *draining
@@ -4249,6 +4447,15 @@ backgroundWatchdog = do
4249 when (t - lastHb > healthHeartbeatMaxAge * 3) (do
4447 when (t - lastHb > healthHeartbeatMaxAge * 3) (do
4250 logError "watchdog: heartbeat loop appears dead, restarting"
4448 logError "watchdog: heartbeat loop appears dead, restarting"
4251 fork pollHeartbeat)
4449 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")
4457+ inFlight <- *federationInFlight
4458+ set *federationActiveConnections = fold (\acc r -> acc + r.count) 0 inFlight)
4252 backgroundWatchdog
4459 backgroundWatchdog
4254main = do
4461main = do