@@ -185,6 +185,11 @@ maxIpRatePerIp = 10 -- max auth requests per IP per window
185185ipRateWindowMs : Int<Ms>
186186ipRateWindowMs = 1000
187187maxIpRateEntries = 5000
188+-- Per-IP unique key tracking (detects key-rotation abuse: single IP authenticating many distinct pubkeys)
189+maxUniqueKeysPerIp = 20 -- max distinct pubkeys per IP per window before throttling
190+uniqueKeyIpWindowMs : Int<Ms>
191+uniqueKeyIpWindowMs = 60000 -- 1 minute
192+maxUniqueKeyIpEntries = 5000
188193-- Lookup rate limiting (per-key, on presence lookups)
189194maxLookupRatePerKey = 5 -- max lookups per key per window
190195lookupRateWindowMs : Int<Ms>
@@ -318,6 +323,8 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
318323-- Tracks recent federation request (server, timestamp) pairs for replay detection
319324type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
320325type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
326+-- Tracks unique pubkeys authenticated per IP (key-rotation abuse detection)
327+type UniqueKeyIpEntry = {ip: Text, keys: [PubkeyHex], windowStart: Timestamp}
321328
322329-- Forward nonce for extended replay protection (persisted in snapshots)
323330type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}
@@ -338,8 +345,8 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
338345-- Hot-path auth scans only one bucket (~1/16 of sessions).
339346-- TODO: If Knot adds indexed arrays or maps, replace the 16 manual bucket
340347-- refs and their if-else dispatch chains with a single indexed collection.
341−-- This would eliminate ~800 lines of mechanical boilerplate across sessions,
342−-- messages, presence, and signal buckets.
348+-- This would eliminate ~1000 lines of mechanical boilerplate across sessions,
349+-- messages, presence, signal, and recipientCount buckets.
343350*sessionBucket0 : [Session]
344351*sessionBucket1 : [Session]
345352*sessionBucket2 : [Session]
@@ -486,8 +493,24 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
486493
487494*lastFederationSuccessAt : Timestamp
488495
489−-- Per-recipient message counts (avoids O(n) scan of all messages on send/forward)
490−*recipientMsgCounts : [RecipientCount]
496+-- Per-recipient message counts sharded into 16 buckets by pubkeyBucket.
497+-- Each pubkey maps to exactly one bucket, so lookups scan ~1/16 of entries.
498+*rcBucket0 : [RecipientCount]
499+*rcBucket1 : [RecipientCount]
500+*rcBucket2 : [RecipientCount]
501+*rcBucket3 : [RecipientCount]
502+*rcBucket4 : [RecipientCount]
503+*rcBucket5 : [RecipientCount]
504+*rcBucket6 : [RecipientCount]
505+*rcBucket7 : [RecipientCount]
506+*rcBucket8 : [RecipientCount]
507+*rcBucket9 : [RecipientCount]
508+*rcBucket10 : [RecipientCount]
509+*rcBucket11 : [RecipientCount]
510+*rcBucket12 : [RecipientCount]
511+*rcBucket13 : [RecipientCount]
512+*rcBucket14 : [RecipientCount]
513+*rcBucket15 : [RecipientCount]
491514
492515-- Tracks outstanding federation fetches per server (detects hung connections)
493516*federationInFlight : [FederationInFlightEntry]
@@ -548,6 +571,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
548571-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
549572*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
550573
574+-- Per-IP unique key tracking: detects when a single IP authenticates many distinct pubkeys
575+-- (key-rotation abuse bypassing per-key rate limits). Pruned with other rate limiters.
576+*uniqueKeyIpRates : [UniqueKeyIpEntry]
577+
551578-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
552579-- Polls compare against this to skip expensive presence lookups when nothing changed.
553580*presenceVersion : Timestamp
@@ -686,9 +713,13 @@ route HealthApi where
686713-- TODO: If Knot adds raw text response support, add a /metrics endpoint
687714-- returning Prometheus exposition format for direct integration with standard
688715-- monitoring stacks. Currently /stats returns JSON which requires a custom scraper.
716+-- TODO: Consider grouping the flat 60+ field response into nested objects
717+-- (e.g., {capacity: {messages, sessions, ...}, federation: {avgLatency, ...}})
718+-- for better monitoring dashboard ergonomics. This is a wire format change
719+-- that would require coordinating with any existing scrapers.
689720route StatsApi where
690721 GET /stats headers {authorization: Text}
691− -> {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, forwardNonces: Int, forwardNoncesMax: Int, pendingAckCleanups: Int, blockedPeers: Int, gossipEventRates: Int, verifyIpRates: Int, ok: Bool}
722+ -> {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, forwardNonces: Int, forwardNoncesMax: Int, pendingAckCleanups: Int, blockedPeers: Int, gossipEventRates: Int, verifyIpRates: Int, uniqueKeyIpRates: Int, ok: Bool}
692723 = Stats
693724 GET /stats/peers headers {authorization: Text}
694725 -> {peers: [PeerStatus], ok: Bool}
@@ -1060,7 +1091,10 @@ writeSessionBucket = \b sessions ->
10601091 else if b == 14 then set *sessionBucket14 = sessions
10611092 else set *sessionBucket15 = sessions
10621093
1063−-- Read all 16 session buckets (for per-key operations, prune, stats)
1094+-- Read all 16 session buckets (for per-key operations, prune, stats).
1095+-- Right-nested union is intentional: union copies the left operand, so right-nesting
1096+-- copies small buckets onto the growing accumulator (O(n) total), while left-nesting
1097+-- would copy the growing accumulator repeatedly (O(n * 16)).
10641098readAllSessions : IO [Session]
10651099readAllSessions = do
10661100 s0 <- *sessionBucket0
@@ -1454,7 +1488,7 @@ uniqueBucketsFor = \items getKey ->
14541488 if any acc (\i -> i == b) then acc else union acc [b]) [] items
14551489
14561490-- Prune a single message bucket and atomically adjust shared counters.
1457−-- Each bucket prune decrements totalMessageCount and recipientMsgCounts by exactly
1491+-- Each bucket prune decrements totalMessageCount and its rcBucket by exactly
14581492-- what was removed, preserving concurrent modifications from poll-ack or send.
14591493pruneMessageBucket = \b messageCutoff -> atomic do
14601494 bucket <- readMessageBucket b
@@ -1464,35 +1498,48 @@ pruneMessageBucket = \b messageCutoff -> atomic do
14641498 if removedCount > 0 then do
14651499 tc <- *totalMessageCount
14661500 set *totalMessageCount = max 0 (tc - removedCount)
1467− rc <- *recipientMsgCounts
1501+ rc <- readRcBucket b
14681502 let removed = do m <- bucket; where m.receivedAt <= messageCutoff; yield m
1469− set *recipientMsgCounts = batchDecrementRecipientCounts rc (countByRecipient removed)
1503+ writeRcBucket b (batchDecrementRecipientCounts rc (countByRecipient removed))
14701504 else yield {}
14711505
1506+-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
1507+-- Each bucket's atomic block only contends on *totalMessageCount (a single Int);
1508+-- rcBuckets and message buckets are per-bucket and independent.
1509+pruneMessageBucketsConcurrent = \b messageCutoff ->
1510+ if b > 15 then yield {}
1511+ else if b < 15 then do
1512+ fork (pruneMessageBucket b messageCutoff)
1513+ pruneMessageBucketsConcurrent (b + 1) messageCutoff
1514+ else pruneMessageBucket b messageCutoff
1515+
14721516-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16
14731517-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one
1474−-- bucket, so per-bucket recipient counts are disjoint and can be merged without overlap.
1475−reconcileMessageCounts = \b acc ->
1518+-- bucket, so per-bucket recipient counts are disjoint — each bucket's counts are written
1519+-- to the corresponding rcBucket without overlap.
1520+reconcileMessageCounts = \b totalAcc ->
14761521 if b > 15 then do
14771522 atomic do
1478− let total = fold (\a c -> a + c.count) 0 acc
14791523 stored <- *totalMessageCount
1480− set *totalMessageCount = total
1481− set *recipientMsgCounts = acc
1482− yield (total - stored)
1524+ set *totalMessageCount = totalAcc
1525+ yield (totalAcc - stored)
14831526 else do
1484− bucketCounts <- atomic do
1527+ bucketTotal <- atomic do
14851528 bucket <- readMessageBucket b
1486− yield (countByRecipient bucket)
1487− reconcileMessageCounts (b + 1) (union acc bucketCounts)
1529+ let counts = countByRecipient bucket
1530+ writeRcBucket b counts
1531+ yield (fold (\a c -> a + c.count) 0 counts)
1532+ reconcileMessageCounts (b + 1) (totalAcc + bucketTotal)
14881533
14891534-- ============================================
14901535-- Recipient Count Helpers
14911536-- ============================================
14921537-- Per-recipient message counts avoid O(n) scan of all messages on send/forward.
1538+-- Sharded into 16 buckets (rcBucket0-15) by pubkeyBucket, reducing scan from
1539+-- O(unique_recipients) to O(unique_recipients/16). Each pubkey maps to exactly
1540+-- one bucket, so lookups never cross shards.
14931541-- TODO: If Knot adds a map/dictionary type, replace [RecipientCount] with a
1494−-- Map<PubkeyHex, Int> for O(1) lookups. Currently getRecipientCount and
1495−-- adjustRecipientCount are O(unique_recipients) per call.
1542+-- Map<PubkeyHex, Int> for O(1) lookups.
14961543
14971544getRecipientCount : PubkeyHex -> [RecipientCount] -> Int
14981545getRecipientCount = \pubkey counts ->
@@ -1529,7 +1576,55 @@ batchDecrementRecipientCounts : [RecipientCount] -> [RecipientCount] -> [Recipie
15291576batchDecrementRecipientCounts = \counts removedCounts ->
15301577 do c <- counts; let adj = c.count - getRecipientCount c.pubkey removedCounts; where adj > 0; yield {pubkey: c.pubkey, count: adj}
15311578
1532−-- Check message dedup only (without counting — count comes from recipientMsgCounts)
1579+-- Recipient count bucket helpers (sharded by pubkeyBucket, same as messages)
1580+readRcBucket : Int -> IO [RecipientCount]
1581+readRcBucket = \b ->
1582+ if b == 0 then do s <- *rcBucket0; yield s
1583+ else if b == 1 then do s <- *rcBucket1; yield s
1584+ else if b == 2 then do s <- *rcBucket2; yield s
1585+ else if b == 3 then do s <- *rcBucket3; yield s
1586+ else if b == 4 then do s <- *rcBucket4; yield s
1587+ else if b == 5 then do s <- *rcBucket5; yield s
1588+ else if b == 6 then do s <- *rcBucket6; yield s
1589+ else if b == 7 then do s <- *rcBucket7; yield s
1590+ else if b == 8 then do s <- *rcBucket8; yield s
1591+ else if b == 9 then do s <- *rcBucket9; yield s
1592+ else if b == 10 then do s <- *rcBucket10; yield s
1593+ else if b == 11 then do s <- *rcBucket11; yield s
1594+ else if b == 12 then do s <- *rcBucket12; yield s
1595+ else if b == 13 then do s <- *rcBucket13; yield s
1596+ else if b == 14 then do s <- *rcBucket14; yield s
1597+ else do s <- *rcBucket15; yield s
1598+
1599+writeRcBucket : Int -> [RecipientCount] -> IO ()
1600+writeRcBucket = \b counts ->
1601+ if b == 0 then set *rcBucket0 = counts
1602+ else if b == 1 then set *rcBucket1 = counts
1603+ else if b == 2 then set *rcBucket2 = counts
1604+ else if b == 3 then set *rcBucket3 = counts
1605+ else if b == 4 then set *rcBucket4 = counts
1606+ else if b == 5 then set *rcBucket5 = counts
1607+ else if b == 6 then set *rcBucket6 = counts
1608+ else if b == 7 then set *rcBucket7 = counts
1609+ else if b == 8 then set *rcBucket8 = counts
1610+ else if b == 9 then set *rcBucket9 = counts
1611+ else if b == 10 then set *rcBucket10 = counts
1612+ else if b == 11 then set *rcBucket11 = counts
1613+ else if b == 12 then set *rcBucket12 = counts
1614+ else if b == 13 then set *rcBucket13 = counts
1615+ else if b == 14 then set *rcBucket14 = counts
1616+ else set *rcBucket15 = counts
1617+
1618+-- Count total recipients across all 16 rc buckets (for stats)
1619+countAllRcEntries : Int -> IO Int
1620+countAllRcEntries = \b ->
1621+ if b > 15 then yield 0
1622+ else do
1623+ bucket <- readRcBucket b
1624+ rest <- countAllRcEntries (b + 1)
1625+ yield (count bucket + rest)
1626+
1627+-- Check message dedup only (without counting — count comes from rcBuckets)
15331628checkMessageDup : [Message] -> Text -> PubkeyHex -> Bool
15341629checkMessageDup = \messages messageId toKey ->
15351630 messageId != "" && any messages (\m -> m.messageId == messageId && m.toKey == toKey)
@@ -1609,8 +1704,13 @@ dedupMessageIds = \messageIds ->
16091704 else union acc [mid]) [] capped
16101705
16111706
1707+-- Constant-time comparison iteration count. Must be >= the longest secret
1708+-- compared (tokens = 48 hex chars, maintenance secret = variable). Using 64
1709+-- provides headroom without the waste of maxTokenLen (256 = 5x actual length).
1710+safeCompareLen = 64
1711+
16121712-- Constant-time string comparison to prevent timing side-channels.
1613−-- Always iterates maxTokenLen characters regardless of input length;
1713+-- Always iterates safeCompareLen characters regardless of input length;
16141714-- rejects empty strings. Tracks both character mismatches and length
16151715-- mismatch (one string ended before the other) within the loop itself,
16161716-- avoiding variable-time count(chars) calls that would leak string length.
@@ -1628,7 +1728,7 @@ safeCompareLoop = \s1 s2 n diff lenDiff ->
16281728safeCompare : Text -> Text -> Bool
16291729safeCompare = \a b ->
16301730 if take 1 a == "" || take 1 b == "" then False {}
1631− else safeCompareLoop a b maxTokenLen 0 0 == 0
1731+ else safeCompareLoop a b safeCompareLen 0 0 == 0
16321732
16331733-- Constant-time IP binding check: returns True if IPs match or no binding is configured.
16341734-- When a session was created with an IP (behind a proxy), requests without an IP header
@@ -1682,6 +1782,25 @@ checkPairRate = \rates sender recipient maxRate windowMs maxEntries t ->
16821782 (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
16831783 maxRate windowMs maxEntries t
16841784
1785+-- Per-IP unique key rate check: tracks distinct pubkeys per IP per window.
1786+-- Returns True if the IP has authenticated too many unique keys (key-rotation abuse).
1787+checkUniqueKeyIpRate : [UniqueKeyIpEntry] -> Text -> PubkeyHex -> Timestamp -> {limited: Bool, entries: [UniqueKeyIpEntry]}
1788+checkUniqueKeyIpRate = \entries ip pubkey t ->
1789+ if ip == "" then {limited: False {}, entries: entries}
1790+ else
1791+ let entry = headOf (do r <- entries; where r.ip == ip; yield r) in
1792+ let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= uniqueKeyIpWindowMs in
1793+ if isNewWindow
1794+ then {limited: False {}, entries: union (do r <- entries; where r.ip != ip; yield r) [{ip: ip, keys: [pubkey], windowStart: t}]}
1795+ else case entry of
1796+ Nothing {} -> {limited: False {}, entries: entries}
1797+ Just {value: r} ->
1798+ let alreadyTracked = any r.keys (\k -> k == pubkey) in
1799+ if alreadyTracked then {limited: False {}, entries: entries}
1800+ else if count r.keys >= maxUniqueKeysPerIp
1801+ then {limited: True {}, entries: entries}
1802+ else {limited: False {}, entries: do e <- entries; yield (if e.ip == ip then {ip: e.ip, keys: union e.keys [pubkey], windowStart: e.windowStart} else e)}
1803+
16851804checkServerRecipientRate = \rates server recipient maxRate windowMs maxEntries t ->
16861805 checkRateWindow rates (\r -> r.server == server && r.recipient == recipient)
16871806 (\ts -> {server: server, recipient: recipient, windowStart: ts, count: 1})
@@ -1883,26 +2002,12 @@ pruneExpired = \t -> do
18832002 transformAllSessionBuckets (\bucket -> do x <- bucket; where x.expiresAt > t && t - x.createdAt < maxSessionLifetime; yield x)
18842003 sessionCount <- countAllSessions
18852004 set *totalSessionCount = sessionCount
1886− -- Prune messages across all 16 buckets. Each bucket prune is atomic and adjusts
1887− -- shared counters (totalMessageCount, recipientMsgCounts) by delta, preserving
1888− -- concurrent modifications from poll-ack or send-message operations.
2005+ -- Prune messages across all 16 buckets concurrently. Each bucket prune is atomic
2006+ -- and touches only its own message bucket + rcBucket. The only shared write is
2007+ -- *totalMessageCount (a single Int), so STM contention is minimal.
2008+ -- Fork first 15 buckets; run the last one in the foreground to bound concurrency.
18892009 let messageCutoff = t - unackedMessageMaxAge
1890− pruneMessageBucket 0 messageCutoff
1891− pruneMessageBucket 1 messageCutoff
1892− pruneMessageBucket 2 messageCutoff
1893− pruneMessageBucket 3 messageCutoff
1894− pruneMessageBucket 4 messageCutoff
1895− pruneMessageBucket 5 messageCutoff
1896− pruneMessageBucket 6 messageCutoff
1897− pruneMessageBucket 7 messageCutoff
1898− pruneMessageBucket 8 messageCutoff
1899− pruneMessageBucket 9 messageCutoff
1900− pruneMessageBucket 10 messageCutoff
1901− pruneMessageBucket 11 messageCutoff
1902− pruneMessageBucket 12 messageCutoff
1903− pruneMessageBucket 13 messageCutoff
1904− pruneMessageBucket 14 messageCutoff
1905− pruneMessageBucket 15 messageCutoff
2010+ pruneMessageBucketsConcurrent 0 messageCutoff
19062011 -- Prune federation state
19072012 atomic do
19082013 let retainCutoff = t - messageMaxAge
@@ -1944,6 +2049,8 @@ pruneExpired = \t -> do
19442049 set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs; yield r
19452050 verifyIpRates <- *verifyIpRates
19462051 set *verifyIpRates = do r <- verifyIpRates; where t - r.windowStart < verifyRateWindowMs; yield r
2052+ uniqueKeyIpRates <- *uniqueKeyIpRates
2053+ set *uniqueKeyIpRates = do r <- uniqueKeyIpRates; where t - r.windowStart < uniqueKeyIpWindowMs; yield r
19472054 atomic do
19482055 activePolls <- *activePolls
19492056 set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p
@@ -2187,7 +2294,7 @@ ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of
21872294 forwards <- *forwards
21882295 msgs <- readMessageBucket b
21892296 tc <- *totalMessageCount
2190− rc <- *recipientMsgCounts
2297+ rc <- readRcBucket b
21912298 let cleanup = fold (\acc rf ->
21922299 let hasLocal = any bucketPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in
21932300 let hasOtherFwds = any forwards (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey) in
@@ -2198,7 +2305,7 @@ ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of
21982305 ) {msgs: msgs, removed: 0, rc: rc} bucketEntries
21992306 writeMessageBucket b cleanup.msgs
22002307 set *totalMessageCount = max 0 (tc - cleanup.removed)
2201− set *recipientMsgCounts = cleanup.rc
2308+ writeRcBucket b cleanup.rc
22022309 ackCleanupByBucket remainingBuckets entries t
22032310
22042311pow2 : Int -> Int
@@ -2583,6 +2690,9 @@ pruneAndRetry = \t -> do
25832690 when (count result.refreshPeers > 0 && count refreshEvents > 0)
25842691 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
25852692
2693+-- Map a single hex character to its integer value (0-15).
2694+-- 16-way if-else chain is mechanical but fine: Knot lacks char-to-int or lookup-table
2695+-- primitives. Clamped to 0-15 by callers (pubkeyBucket, tokenBucket) as a safety net.
25862696hexCharIndex : Text -> Int
25872697hexCharIndex = \c ->
25882698 if c == "0" then 0 else if c == "1" then 1 else if c == "2" then 2 else if c == "3" then 3
@@ -2649,8 +2759,8 @@ ackPolledMessages = \pubkey cursor t -> do
26492759 writeMessageBucket b (diff bucket toAck)
26502760 totalCount <- *totalMessageCount
26512761 set *totalMessageCount = max 0 (totalCount - ackCount)
2652− recipientCounts <- *recipientMsgCounts
2653− set *recipientMsgCounts = adjustRecipientCount recipientCounts pubkey (0 - ackCount)
2762+ recipientCounts <- readRcBucket b
2763+ writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
26542764 let ackedMsgIds = do m <- toAck; yield m.messageId
26552765 forwards <- *forwards
26562766 set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
@@ -2952,6 +3062,18 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
29523062 else do
29533063 authGlobalLimited <- checkAuthGlobalRate t
29543064 if authGlobalLimited then yield (respond {token: "", expiresAt: 0})
3065+ else do
3066+ -- Per-IP unique key abuse detection: reject IPs authenticating too many distinct pubkeys
3067+ uniqueKeyLimited <- atomic do
3068+ ukr <- *uniqueKeyIpRates
3069+ let result = checkUniqueKeyIpRate ukr clientIp pubkey t
3070+ set *uniqueKeyIpRates = if count result.entries > maxUniqueKeyIpEntries
3071+ then do r <- result.entries; where t - r.windowStart < uniqueKeyIpWindowMs; yield r
3072+ else result.entries
3073+ yield result.limited
3074+ if uniqueKeyLimited then do
3075+ logWarn ("verify: unique key per IP rate limited (" ++ show maxUniqueKeysPerIp ++ " keys/" ++ show uniqueKeyIpWindowMs ++ "ms) for IP " ++ clientIp)
3076+ yield (respond {token: "", expiresAt: 0})
29553077 else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
29563078
29573079verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do
@@ -3097,7 +3219,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
30973219 totalMsgCount <- *totalMessageCount
30983220 if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
30993221 else do
3100− recipientCounts <- *recipientMsgCounts
3222+ recipientCounts <- readRcBucket msgBucket
31013223 let recipientCount = getRecipientCount to recipientCounts
31023224 if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
31033225 else do
@@ -3109,7 +3231,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
31093231 [{messageId: effectiveMsgId, toKey: to,
31103232 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])
31113233 set *totalMessageCount = totalMsgCount + 1
3112− set *recipientMsgCounts = adjustRecipientCount recipientCounts to 1
3234+ writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)
31133235 signalMessageForKey to seqTs
31143236 let toPresence = presenceForKey recipientPresence to t
31153237 let localOnline = hasLocalPresence toPresence
@@ -3449,9 +3571,10 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
34493571 freshEvents <- atomic do
34503572 recentGossip <- *recentGossip
34513573 let active = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
3452− -- Pre-filter active entries to only those matching pubkeys in this batch (reduces O(n*m) inner scan)
3453− let batchPubkeys = fold (\acc e -> if any acc (\a -> a == e.pubkey) then acc else union acc [e.pubkey]) [] effectiveEvents
3454− let relevantActive = do r <- active; where any batchPubkeys (\pk -> pk == r.pubkey); yield r
3574+ -- Pre-filter active entries to only those matching (pubkey, server) pairs in this batch.
3575+ -- Reduces the inner dedup scan from O(n*m) to O(n*k) where k << m (k = matching pairs).
3576+ let batchPairs = fold (\acc e -> if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server) then acc else union acc [{pubkey: e.pubkey, server: e.server}]) [] effectiveEvents
3577+ let relevantActive = do r <- active; where any batchPairs (\p -> p.pubkey == r.pubkey && p.server == r.server); yield r
34553578 let fresh = do e <- effectiveEvents; where none relevantActive (\r -> r.pubkey == e.pubkey && r.server == e.server && r.timestamp == e.timestamp); yield e
34563579 let newRecent = do e <- fresh; yield {pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, seenAt: t}
34573580 let updated = union active newRecent
@@ -3524,7 +3647,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
35243647 if totalMsgCount >= maxMessages
35253648 then yield {forward: ForwardRejected {}, globalCapReached: True {}, recvRateLimited: False {}}
35263649 else do
3527− recipientCounts <- *recipientMsgCounts
3650+ recipientCounts <- readRcBucket msgBucket
35283651 let recipientCount = getRecipientCount toKey recipientCounts
35293652 if recipientCount >= maxMessagesPerRecipient
35303653 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
@@ -3548,7 +3671,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
35483671 [{messageId: messageId, toKey: toKey,
35493672 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}])
35503673 set *totalMessageCount = totalMsgCount + 1
3551− set *recipientMsgCounts = adjustRecipientCount recipientCounts toKey 1
3674+ writeRcBucket msgBucket (adjustRecipientCount recipientCounts toKey 1)
35523675 signalMessageForKey toKey seqTs
35533676 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
35543677 fwdNonces <- *forwardNonces
@@ -3632,6 +3755,19 @@ handleDrain = \authorization respond -> do
36323755 yield {forwards: f, pendingAcks: pa}
36333756 let pendingForwardCount = count flushState.forwards
36343757 let pendingAckCount = count flushState.pendingAcks
3758+ -- Auto-snapshot on drain: export federation state so a crash during drain doesn't lose it.
3759+ -- Output as structured JSON with [SNAPSHOT_DATA] prefix for log-shipping pipelines.
3760+ -- This is the last chance to persist state before the server stops.
3761+ drainSnapshot <- atomic do
3762+ a <- *ackedIds
3763+ ao <- *ackOverflow
3764+ d <- *deliveredIds
3765+ ff <- *forwardFailures
3766+ fn <- *forwardNonces
3767+ bp <- *blockedPeers
3768+ yield {version: snapshotVersion, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp}
3769+ println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
3770+ logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
36353771 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
36363772 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
36373773 atomic do set *drainCompletedAt = 0
@@ -3725,8 +3861,8 @@ handleHealth = \authorization respond -> do
37253861 snapshotOk: snapOk,
37263862 watchdogOk: wdOk}
37273863 let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
3728− -- Unauthenticated: always return ok:true to prevent information leakage.
3729− -- Attackers could correlate ok:false transitions with targeted load to infer subsystem pressure.
3864+ -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
3865+ -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
37303866 if isAuthed
37313867 then yield (respond {ok: allOk, pruneHealthy: result.pruneOk,
37323868 heartbeatHealthy: result.heartbeatOk,
@@ -3734,14 +3870,14 @@ handleHealth = \authorization respond -> do
37343870 federationHealthy: result.federationOk,
37353871 snapshotHealthy: result.snapshotOk,
37363872 watchdogHealthy: result.watchdogOk})
3737− else yield (respond {ok: True {}, pruneHealthy: True {},
3873+ else yield (respond {ok: allOk, pruneHealthy: True {},
37383874 heartbeatHealthy: True {},
37393875 capacityHealthy: True {},
37403876 federationHealthy: True {},
37413877 snapshotHealthy: True {},
37423878 watchdogHealthy: True {}})
37433879
3744−zeroStats = {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, forwardNonces: 0, forwardNoncesMax: 0, pendingAckCleanups: 0, blockedPeers: 0, gossipEventRates: 0, verifyIpRates: 0, ok: False {}}
3880+zeroStats = {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, forwardNonces: 0, forwardNoncesMax: 0, pendingAckCleanups: 0, blockedPeers: 0, gossipEventRates: 0, verifyIpRates: 0, uniqueKeyIpRates: 0, ok: False {}}
37453881
37463882handleStats = \authorization respond -> do
37473883 t <- now
@@ -3764,7 +3900,7 @@ handleStats = \authorization respond -> do
37643900 ff <- *forwardFailures
37653901 ap <- *activePolls
37663902 tmc <- *totalMessageCount
3767− rc <- *recipientMsgCounts
3903+ rc <- countAllRcEntries 0
37683904 fi <- *federationInFlight
37693905 rg <- *recentGossip
37703906 fn <- *forwardNonces
@@ -3772,7 +3908,8 @@ handleStats = \authorization respond -> do
37723908 bp <- *blockedPeers
37733909 ger <- *gossipEventRates
37743910 vir <- *verifyIpRates
3775− yield {sc: sc, pc: pc, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg, fn: fn, pac: pac, bp: bp, ger: ger, vir: vir}
3911+ ukir <- *uniqueKeyIpRates
3912+ yield {sc: sc, pc: pc, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg, fn: fn, pac: pac, bp: bp, ger: ger, vir: vir, ukir: ukir}
37763913 counters <- atomic do
37773914 sr <- *sendRates
37783915 cr <- *challengeRates
@@ -3804,7 +3941,7 @@ handleStats = \authorization respond -> do
38043941 let hungFedSlots = count (do r <- core.fi; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs; yield r)
38053942 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
38063943 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt
3807− yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pairRates: result.pairRates, federationPairRates: result.federationPairRates, pruneCount: result.pruneCount, pruneRunning: result.pruneRunning, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, draining: result.draining, federationEnabled: federationEnabled, recentGossip: result.recentGossip, totalMessageCount: result.totalMessageCount, recipientCounts: result.recipientCounts, federationInFlight: result.federationInFlight, hungFederationSlots: hungFedSlots, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, drainCompletedAt: result.drainCompletedAt, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, forwardNonces: count core.fn, forwardNoncesMax: maxForwardNonces, pendingAckCleanups: count core.pac, blockedPeers: count core.bp, gossipEventRates: count core.ger, verifyIpRates: count core.vir, ok: True {}})
3944+ yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pairRates: result.pairRates, federationPairRates: result.federationPairRates, pruneCount: result.pruneCount, pruneRunning: result.pruneRunning, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, draining: result.draining, federationEnabled: federationEnabled, recentGossip: result.recentGossip, totalMessageCount: result.totalMessageCount, recipientCounts: result.recipientCounts, federationInFlight: result.federationInFlight, hungFederationSlots: hungFedSlots, messagesPct: msgPct, presencePct: presPct, forwardsPct: fwdPct, sessionsPct: sessPct, federationAvgLatencyMs: fedAvg, federationMaxLatencyMs: result.federationLatencyMax, federationCallCount: result.federationLatencyCount, drainCompletedAt: result.drainCompletedAt, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, forwardNonces: count core.fn, forwardNoncesMax: maxForwardNonces, pendingAckCleanups: count core.pac, blockedPeers: count core.bp, gossipEventRates: count core.ger, verifyIpRates: count core.vir, uniqueKeyIpRates: count core.ukir, ok: True {}})
38083945 else yield (respond zeroStats)
38093946
38103947handleStatsPeers = \authorization respond -> do
@@ -3973,8 +4110,8 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
39734110 writeMessageBucket b (diff bucket toAck)
39744111 totalCount <- *totalMessageCount
39754112 set *totalMessageCount = max 0 (totalCount - ackCount)
3976− recipientCounts <- *recipientMsgCounts
3977− set *recipientMsgCounts = adjustRecipientCount recipientCounts pubkey (0 - ackCount)
4113+ recipientCounts <- readRcBucket b
4114+ writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
39784115 let ackedMsgIds = do m <- toAck; yield m.messageId
39794116 forwards <- *forwards
39804117 set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
@@ -4138,7 +4275,9 @@ handleAdminListSessions = \pubkey authorization respond -> do
41384275 else do
41394276 result <- atomic do
41404277 allSessions <- readAllSessions
4141− let keySessions = do s <- allSessions; where s.pubkey == pubkey; yield {token: take 8 s.token ++ "...", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt}
4278+ -- Show only shard prefix (4 hex chars = 16 bits) for identification without aiding brute-force.
4279+ -- Previous 8-char prefix exposed 32 bits, narrowing brute-force from 192 to 160 bits.
4280+ let keySessions = do s <- allSessions; where s.pubkey == pubkey; yield {token: take 4 s.token ++ "...", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt}
41424281 yield {sessions: keySessions, total: count keySessions}
41434282 yield (respond {sessions: result.sessions, count: result.total, ok: True {}})
41444283
@@ -4171,7 +4310,44 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do
41714310 (fork (sendBatchedGossip result.peerServers [{eventType: GossipOffline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))
41724311 yield (respond {ok: True {}, revokedCount: result.revokedCount})
41734312
4174−serve = \req -> case req of
4313+-- Extract request path for access logging (excludes poll to reduce log volume)
4314+requestPath = \req -> case req of
4315+ ReqChallenge {} -> "POST /auth/challenge"
4316+ VerifyAuth {} -> "POST /auth/verify"
4317+ RevokeSession {} -> "POST /auth/revoke"
4318+ Logout {} -> "POST /auth/logout"
4319+ Poll {} -> "" -- suppress: high volume, logged on slow path only
4320+ SendMessage {} -> "POST /messages"
4321+ MessageStatus {} -> "POST /messages/status"
4322+ AckMessages {} -> "POST /messages/ack"
4323+ SendBatch {} -> "POST /messages/batch"
4324+ ListMessages {} -> "GET /messages/list"
4325+ LookupPresence {} -> "GET /lookup"
4326+ RecvGossip {} -> "POST /federation/gossip"
4327+ ForwardMessage {} -> "POST /federation/forward"
4328+ FederationAck {} -> "POST /federation/ack"
4329+ VerifyToken {} -> "POST /federation/verify"
4330+ Prune {} -> "POST /maintenance/prune"
4331+ Drain {} -> "POST /maintenance/drain"
4332+ Undrain {} -> "POST /maintenance/undrain"
4333+ Snapshot {} -> "GET /maintenance/snapshot"
4334+ Restore {} -> "POST /maintenance/restore"
4335+ BlockPeer {} -> "POST /maintenance/block"
4336+ UnblockPeer {} -> "POST /maintenance/unblock"
4337+ ListBlocked {} -> "GET /maintenance/blocked"
4338+ PingPeer {} -> "POST /maintenance/ping"
4339+ AdminListSessions {} -> "GET /maintenance/sessions"
4340+ AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"
4341+ Health {} -> "GET /health"
4342+ Stats {} -> "GET /stats"
4343+ StatsPeers {} -> "GET /stats/peers"
4344+ StatsKey {} -> "GET /stats/key"
4345+
4346+-- Slow request threshold for access logging (ms)
4347+slowRequestThreshold : Int<Ms>
4348+slowRequestThreshold = 1000
4349+
4350+serveInner = \req -> case req of
41754351 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
41764352 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
41774353 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
@@ -4203,6 +4379,19 @@ serve = \req -> case req of
42034379 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
42044380 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
42054381
4382+-- Access-logged request handler: wraps serveInner with timing.
4383+-- Logs all non-poll requests. Polls are only logged when slow (> slowRequestThreshold)
4384+-- to avoid overwhelming logs with high-frequency long-poll traffic.
4385+serve = \req -> do
4386+ startTime <- now
4387+ serveInner req
4388+ endTime <- now
4389+ let latency = endTime - startTime
4390+ let path = requestPath req
4391+ if path != ""
4392+ then logInfo ("[ACCESS] " ++ path ++ " " ++ show latency ++ "ms")
4393+ else when (latency > slowRequestThreshold) (logInfo ("[SLOW] poll " ++ show latency ++ "ms"))
4394+
42064395-- ============================================
42074396-- Entry Point
42084397-- ============================================
@@ -4212,6 +4401,11 @@ serve = \req -> case req of
42124401-- NOTE: This function (and all background loops below) recurse via tail calls.
42134402-- Knot compiles tail calls to loops, so stack growth is bounded. If Knot's tail-call
42144403-- optimization is ever disabled, these loops will overflow the stack over time.
4404+-- The backgroundWatchdog can restart backgroundPrune and pollHeartbeat if they die,
4405+-- but cannot restart itself or backgroundSnapshot. A TCO failure in any loop would
4406+-- eventually surface as a health check failure (pruneHealthy/heartbeatHealthy/watchdogHealthy).
4407+-- TODO: If Knot adds an iterative construct (e.g., loop/while), migrate background loops
4408+-- to avoid the TCO dependency entirely.
42154409backgroundPrune = do
42164410 jitter <- randomInt 5000
42174411 sleep (backgroundPruneInterval + jitter)
@@ -4297,9 +4491,11 @@ pollHeartbeat = do
42974491
42984492-- Background loop: periodically logs a state summary to stdout.
42994493-- Operators can capture stdout to detect state drift; use /maintenance/snapshot for full export.
4494+-- The drain handler now auto-exports a [SNAPSHOT_DATA] JSON line for log-shipping pipelines,
4495+-- but a crash outside drain still loses state. Mitigate with frequent /maintenance/snapshot calls
4496+-- or a sidecar that watches for [SNAPSHOT_DATA] lines and writes them to disk.
43004497-- TODO: If Knot adds file I/O primitives, auto-export snapshots to a file
4301−-- (e.g., /var/lib/skrepka/snapshot.json) in addition to logging. Currently
4302−-- a crash between manual snapshot exports loses all federation state.
4498+-- (e.g., /var/lib/skrepka/snapshot.json) in addition to logging.
43034499backgroundSnapshot = do
43044500 sleep autoSnapshotInterval
43054501 draining <- *draining
@@ -4361,6 +4557,35 @@ backgroundWatchdog = do
43614557 set *federationActiveConnections = fold (\acc r -> acc + r.count) 0 inFlight)
43624558 backgroundWatchdog
43634559
4560+-- Startup connectivity probe: send an empty gossip to each known peer and log results.
4561+-- Surfaces DNS, firewall, and TLS issues before the first real gossip cycle.
4562+-- Probes up to 10 peers sequentially (not forked) to avoid overwhelming the network.
4563+startupProbePeers = \probed -> do
4564+ knownPeers <- *knownPeerServers
4565+ let remaining = do p <- knownPeers; yield p
4566+ startupProbeNext remaining 0
4567+
4568+startupProbeNext = \peers probed ->
4569+ if probed >= 10 then logInfo ("startup: probed " ++ show probed ++ " peers")
4570+ else case headOf peers of
4571+ Nothing {} -> logInfo ("startup: probed " ++ show probed ++ " peers")
4572+ Just {value: server} -> do
4573+ let rest = do p <- peers; where p != server; yield p
4574+ startTime <- now
4575+ auth <- generateFederationAuth "gossip"
4576+ result <- fetchFederation server (RecvGossip {
4577+ authorization: auth,
4578+ events: [], hopsRemaining: 1, fromServer: serverHostN,
4579+ requestTimestamp: startTime})
4580+ endTime <- now
4581+ let latency = endTime - startTime
4582+ case result of
4583+ Err e -> logWarn ("startup: peer " ++ server ++ " unreachable (" ++ show e ++ ") — check DNS/firewall/TLS")
4584+ Ok r -> if r.ok
4585+ then logInfo ("startup: peer " ++ server ++ " ok (" ++ show latency ++ "ms)")
4586+ else logWarn ("startup: peer " ++ server ++ " rejected (" ++ r.error ++ ", " ++ show latency ++ "ms)")
4587+ startupProbeNext rest (probed + 1)
4588+
43644589main = do
43654590 let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
43664591 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
@@ -4412,6 +4637,8 @@ main = do
44124637 when (matchesHostnameOrSubdomain serverHostN "localhost")
44134638 (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
44144639 logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
4640+ when federationEnabled
4641+ (logWarn "SECURITY: DNS rebinding risk — federation URLs resolve DNS at fetch time, not at validation time. A peer hostname passing isBadServerName could rebind to a private IP between validation and fetch. Mitigate by blocking outbound connections to RFC1918/link-local/CGNAT ranges at the firewall level.")
44154642 logInfo "Skrepka relay server starting on port 8080..."
44164643 logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")
44174644 logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
@@ -4437,4 +4664,11 @@ main = do
44374664 fork pollHeartbeat
44384665 fork backgroundWatchdog
44394666 when (autoSnapshotInterval > 0) (fork backgroundSnapshot)
4667+ -- Startup peer connectivity check: probe known peers after a short delay
4668+ -- to surface federation misconfigurations immediately (DNS, firewall, TLS).
4669+ -- Runs as a forked task so it doesn't delay server startup.
4670+ when federationEnabled (fork (do
4671+ sleep 5000 -- wait for listener to be ready
4672+ logInfo "startup: probing federation peers..."
4673+ startupProbePeers 0))
44404674 listen 8080 serve