@@ -185,6 +185,11 @@ maxIpRatePerIp = 10 -- max auth requests per IP per window
185ipRateWindowMs : Int<Ms>
185ipRateWindowMs : Int<Ms>
186ipRateWindowMs = 1000
186ipRateWindowMs = 1000
187maxIpRateEntries = 5000
187maxIpRateEntries = 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
188-- Lookup rate limiting (per-key, on presence lookups)
193-- Lookup rate limiting (per-key, on presence lookups)
189maxLookupRatePerKey = 5 -- max lookups per key per window
194maxLookupRatePerKey = 5 -- max lookups per key per window
190lookupRateWindowMs : Int<Ms>
195lookupRateWindowMs : Int<Ms>
@@ -318,6 +323,8 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
318-- Tracks recent federation request (server, timestamp) pairs for replay detection
323-- Tracks recent federation request (server, timestamp) pairs for replay detection
319type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
324type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
320type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
325type 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}
322-- Forward nonce for extended replay protection (persisted in snapshots)
329-- Forward nonce for extended replay protection (persisted in snapshots)
323type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}
330type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}
@@ -338,8 +345,8 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
338-- Hot-path auth scans only one bucket (~1/16 of sessions).
345-- Hot-path auth scans only one bucket (~1/16 of sessions).
339-- TODO: If Knot adds indexed arrays or maps, replace the 16 manual bucket
346-- TODO: If Knot adds indexed arrays or maps, replace the 16 manual bucket
340-- refs and their if-else dispatch chains with a single indexed collection.
347-- refs and their if-else dispatch chains with a single indexed collection.
341−-- This would eliminate ~800 lines of mechanical boilerplate across sessions,
348+-- This would eliminate ~1000 lines of mechanical boilerplate across sessions,
342−-- messages, presence, and signal buckets.
349+-- messages, presence, signal, and recipientCount buckets.
343*sessionBucket0 : [Session]
350*sessionBucket0 : [Session]
344*sessionBucket1 : [Session]
351*sessionBucket1 : [Session]
345*sessionBucket2 : [Session]
352*sessionBucket2 : [Session]
@@ -486,8 +493,24 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
487*lastFederationSuccessAt : Timestamp
494*lastFederationSuccessAt : Timestamp
489−-- Per-recipient message counts (avoids O(n) scan of all messages on send/forward)
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]
490−*recipientMsgCounts : [RecipientCount]
513+*rcBucket15 : [RecipientCount]
492-- Tracks outstanding federation fetches per server (detects hung connections)
515-- Tracks outstanding federation fetches per server (detects hung connections)
493*federationInFlight : [FederationInFlightEntry]
516*federationInFlight : [FederationInFlightEntry]
@@ -548,6 +571,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
548-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
571-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
549*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
572*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
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]
551-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
578-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
552-- Polls compare against this to skip expensive presence lookups when nothing changed.
579-- Polls compare against this to skip expensive presence lookups when nothing changed.
553*presenceVersion : Timestamp
580*presenceVersion : Timestamp
@@ -686,9 +713,13 @@ route HealthApi where
686-- TODO: If Knot adds raw text response support, add a /metrics endpoint
713-- TODO: If Knot adds raw text response support, add a /metrics endpoint
687-- returning Prometheus exposition format for direct integration with standard
714-- returning Prometheus exposition format for direct integration with standard
688-- monitoring stacks. Currently /stats returns JSON which requires a custom scraper.
715-- 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.
689route StatsApi where
720route StatsApi where
690 GET /stats headers {authorization: Text}
721 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}
693 GET /stats/peers headers {authorization: Text}
724 GET /stats/peers headers {authorization: Text}
694 -> {peers: [PeerStatus], ok: Bool}
725 -> {peers: [PeerStatus], ok: Bool}
@@ -1060,7 +1091,10 @@ writeSessionBucket = \b sessions ->
1060 else if b == 14 then set *sessionBucket14 = sessions
1091 else if b == 14 then set *sessionBucket14 = sessions
1061 else set *sessionBucket15 = sessions
1092 else set *sessionBucket15 = sessions
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)).
1064readAllSessions : IO [Session]
1098readAllSessions : IO [Session]
1065readAllSessions = do
1099readAllSessions = do
1066 s0 <- *sessionBucket0
1100 s0 <- *sessionBucket0
@@ -1454,7 +1488,7 @@ uniqueBucketsFor = \items getKey ->
1454 if any acc (\i -> i == b) then acc else union acc [b]) [] items
1488 if any acc (\i -> i == b) then acc else union acc [b]) [] items
1456-- Prune a single message bucket and atomically adjust shared counters.
1490-- 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
1458-- what was removed, preserving concurrent modifications from poll-ack or send.
1492-- what was removed, preserving concurrent modifications from poll-ack or send.
1459pruneMessageBucket = \b messageCutoff -> atomic do
1493pruneMessageBucket = \b messageCutoff -> atomic do
1460 bucket <- readMessageBucket b
1494 bucket <- readMessageBucket b
@@ -1464,35 +1498,48 @@ pruneMessageBucket = \b messageCutoff -> atomic do
1464 if removedCount > 0 then do
1498 if removedCount > 0 then do
1465 tc <- *totalMessageCount
1499 tc <- *totalMessageCount
1466 set *totalMessageCount = max 0 (tc - removedCount)
1500 set *totalMessageCount = max 0 (tc - removedCount)
1467− rc <- *recipientMsgCounts
1501+ rc <- readRcBucket b
1468 let removed = do m <- bucket; where m.receivedAt <= messageCutoff; yield m
1502 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))
1470 else yield {}
1504 else yield {}
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
1472-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16
1516-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16
1473-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one
1517-- 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.
1518+-- bucket, so per-bucket recipient counts are disjoint — each bucket's counts are written
1519+-- to the corresponding rcBucket without overlap.
1475−reconcileMessageCounts = \b acc ->
1520+reconcileMessageCounts = \b totalAcc ->
1476 if b > 15 then do
1521 if b > 15 then do
1477 atomic do
1522 atomic do
1478− let total = fold (\a c -> a + c.count) 0 acc
1479 stored <- *totalMessageCount
1523 stored <- *totalMessageCount
1480− set *totalMessageCount = total
1524+ set *totalMessageCount = totalAcc
1481− set *recipientMsgCounts = acc
1482− yield (total - stored)
1525+ yield (totalAcc - stored)
1484− bucketCounts <- atomic do
1527+ bucketTotal <- atomic do
1485 bucket <- readMessageBucket b
1528 bucket <- readMessageBucket b
1486− yield (countByRecipient bucket)
1529+ let counts = countByRecipient bucket
1530+ writeRcBucket b counts
1531+ yield (fold (\a c -> a + c.count) 0 counts)
1487− reconcileMessageCounts (b + 1) (union acc bucketCounts)
1532+ reconcileMessageCounts (b + 1) (totalAcc + bucketTotal)
1489-- ============================================
1534-- ============================================
1490-- Recipient Count Helpers
1535-- Recipient Count Helpers
1491-- ============================================
1536-- ============================================
1492-- Per-recipient message counts avoid O(n) scan of all messages on send/forward.
1537-- 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.
1493-- TODO: If Knot adds a map/dictionary type, replace [RecipientCount] with a
1541-- TODO: If Knot adds a map/dictionary type, replace [RecipientCount] with a
1494−-- Map<PubkeyHex, Int> for O(1) lookups. Currently getRecipientCount and
1542+-- Map<PubkeyHex, Int> for O(1) lookups.
1495−-- adjustRecipientCount are O(unique_recipients) per call.
1497getRecipientCount : PubkeyHex -> [RecipientCount] -> Int
1544getRecipientCount : PubkeyHex -> [RecipientCount] -> Int
1498getRecipientCount = \pubkey counts ->
1545getRecipientCount = \pubkey counts ->
@@ -1529,7 +1576,55 @@ batchDecrementRecipientCounts : [RecipientCount] -> [RecipientCount] -> [Recipie
1529batchDecrementRecipientCounts = \counts removedCounts ->
1576batchDecrementRecipientCounts = \counts removedCounts ->
1530 do c <- counts; let adj = c.count - getRecipientCount c.pubkey removedCounts; where adj > 0; yield {pubkey: c.pubkey, count: adj}
1577 do c <- counts; let adj = c.count - getRecipientCount c.pubkey removedCounts; where adj > 0; yield {pubkey: c.pubkey, count: adj}
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
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
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
1623+ bucket <- readRcBucket b
1624+ rest <- countAllRcEntries (b + 1)
1625+ yield (count bucket + rest)
1532−-- Check message dedup only (without counting — count comes from recipientMsgCounts)
1627+-- Check message dedup only (without counting — count comes from rcBuckets)
1533checkMessageDup : [Message] -> Text -> PubkeyHex -> Bool
1628checkMessageDup : [Message] -> Text -> PubkeyHex -> Bool
1534checkMessageDup = \messages messageId toKey ->
1629checkMessageDup = \messages messageId toKey ->
1535 messageId != "" && any messages (\m -> m.messageId == messageId && m.toKey == toKey)
1630 messageId != "" && any messages (\m -> m.messageId == messageId && m.toKey == toKey)
@@ -1609,8 +1704,13 @@ dedupMessageIds = \messageIds ->
1609 else union acc [mid]) [] capped
1704 else union acc [mid]) [] capped
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).
1612-- Constant-time string comparison to prevent timing side-channels.
1712-- 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;
1614-- rejects empty strings. Tracks both character mismatches and length
1714-- rejects empty strings. Tracks both character mismatches and length
1615-- mismatch (one string ended before the other) within the loop itself,
1715-- mismatch (one string ended before the other) within the loop itself,
1616-- avoiding variable-time count(chars) calls that would leak string length.
1716-- avoiding variable-time count(chars) calls that would leak string length.
@@ -1628,7 +1728,7 @@ safeCompareLoop = \s1 s2 n diff lenDiff ->
1628safeCompare : Text -> Text -> Bool
1728safeCompare : Text -> Text -> Bool
1629safeCompare = \a b ->
1729safeCompare = \a b ->
1630 if take 1 a == "" || take 1 b == "" then False {}
1730 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
1633-- Constant-time IP binding check: returns True if IPs match or no binding is configured.
1733-- Constant-time IP binding check: returns True if IPs match or no binding is configured.
1634-- When a session was created with an IP (behind a proxy), requests without an IP header
1734-- 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 ->
1682 (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
1782 (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
1683 maxRate windowMs maxEntries t
1783 maxRate windowMs maxEntries t
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}
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
1794+ then {limited: False {}, entries: union (do r <- entries; where r.ip != ip; yield r) [{ip: ip, keys: [pubkey], windowStart: t}]}
1796+ Nothing {} -> {limited: False {}, entries: entries}
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)}
1685checkServerRecipientRate = \rates server recipient maxRate windowMs maxEntries t ->
1804checkServerRecipientRate = \rates server recipient maxRate windowMs maxEntries t ->
1686 checkRateWindow rates (\r -> r.server == server && r.recipient == recipient)
1805 checkRateWindow rates (\r -> r.server == server && r.recipient == recipient)
1687 (\ts -> {server: server, recipient: recipient, windowStart: ts, count: 1})
1806 (\ts -> {server: server, recipient: recipient, windowStart: ts, count: 1})
@@ -1883,26 +2002,12 @@ pruneExpired = \t -> do
1883 transformAllSessionBuckets (\bucket -> do x <- bucket; where x.expiresAt > t && t - x.createdAt < maxSessionLifetime; yield x)
2002 transformAllSessionBuckets (\bucket -> do x <- bucket; where x.expiresAt > t && t - x.createdAt < maxSessionLifetime; yield x)
1884 sessionCount <- countAllSessions
2003 sessionCount <- countAllSessions
1885 set *totalSessionCount = sessionCount
2004 set *totalSessionCount = sessionCount
1886− -- Prune messages across all 16 buckets. Each bucket prune is atomic and adjusts
2005+ -- Prune messages across all 16 buckets concurrently. Each bucket prune is atomic
1887− -- shared counters (totalMessageCount, recipientMsgCounts) by delta, preserving
2006+ -- and touches only its own message bucket + rcBucket. The only shared write is
1888− -- concurrent modifications from poll-ack or send-message operations.
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.
1889 let messageCutoff = t - unackedMessageMaxAge
2009 let messageCutoff = t - unackedMessageMaxAge
1890− pruneMessageBucket 0 messageCutoff
2010+ pruneMessageBucketsConcurrent 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
1906 -- Prune federation state
2011 -- Prune federation state
1907 atomic do
2012 atomic do
1908 let retainCutoff = t - messageMaxAge
2013 let retainCutoff = t - messageMaxAge
@@ -1944,6 +2049,8 @@ pruneExpired = \t -> do
1944 set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs; yield r
2049 set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs; yield r
1945 verifyIpRates <- *verifyIpRates
2050 verifyIpRates <- *verifyIpRates
1946 set *verifyIpRates = do r <- verifyIpRates; where t - r.windowStart < verifyRateWindowMs; yield r
2051 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
1947 atomic do
2054 atomic do
1948 activePolls <- *activePolls
2055 activePolls <- *activePolls
1949 set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p
2056 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
2187 forwards <- *forwards
2294 forwards <- *forwards
2188 msgs <- readMessageBucket b
2295 msgs <- readMessageBucket b
2189 tc <- *totalMessageCount
2296 tc <- *totalMessageCount
2190− rc <- *recipientMsgCounts
2297+ rc <- readRcBucket b
2191 let cleanup = fold (\acc rf ->
2298 let cleanup = fold (\acc rf ->
2192 let hasLocal = any bucketPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in
2299 let hasLocal = any bucketPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in
2193 let hasOtherFwds = any forwards (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey) in
2300 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
2198 ) {msgs: msgs, removed: 0, rc: rc} bucketEntries
2305 ) {msgs: msgs, removed: 0, rc: rc} bucketEntries
2199 writeMessageBucket b cleanup.msgs
2306 writeMessageBucket b cleanup.msgs
2200 set *totalMessageCount = max 0 (tc - cleanup.removed)
2307 set *totalMessageCount = max 0 (tc - cleanup.removed)
2201− set *recipientMsgCounts = cleanup.rc
2308+ writeRcBucket b cleanup.rc
2202 ackCleanupByBucket remainingBuckets entries t
2309 ackCleanupByBucket remainingBuckets entries t
2204pow2 : Int -> Int
2311pow2 : Int -> Int
@@ -2583,6 +2690,9 @@ pruneAndRetry = \t -> do
2583 when (count result.refreshPeers > 0 && count refreshEvents > 0)
2690 when (count result.refreshPeers > 0 && count refreshEvents > 0)
2584 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
2691 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
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.
2586hexCharIndex : Text -> Int
2696hexCharIndex : Text -> Int
2587hexCharIndex = \c ->
2697hexCharIndex = \c ->
2588 if c == "0" then 0 else if c == "1" then 1 else if c == "2" then 2 else if c == "3" then 3
2698 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
2649 writeMessageBucket b (diff bucket toAck)
2759 writeMessageBucket b (diff bucket toAck)
2650 totalCount <- *totalMessageCount
2760 totalCount <- *totalMessageCount
2651 set *totalMessageCount = max 0 (totalCount - ackCount)
2761 set *totalMessageCount = max 0 (totalCount - ackCount)
2652− recipientCounts <- *recipientMsgCounts
2762+ recipientCounts <- readRcBucket b
2653− set *recipientMsgCounts = adjustRecipientCount recipientCounts pubkey (0 - ackCount)
2763+ writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
2654 let ackedMsgIds = do m <- toAck; yield m.messageId
2764 let ackedMsgIds = do m <- toAck; yield m.messageId
2655 forwards <- *forwards
2765 forwards <- *forwards
2656 set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
2766 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 ->
2953 authGlobalLimited <- checkAuthGlobalRate t
3063 authGlobalLimited <- checkAuthGlobalRate t
2954 if authGlobalLimited then yield (respond {token: "", expiresAt: 0})
3064 if authGlobalLimited then yield (respond {token: "", expiresAt: 0})
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})
2955 else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
3077 else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
2957verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do
3079verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do
@@ -3097,7 +3219,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
3097 totalMsgCount <- *totalMessageCount
3219 totalMsgCount <- *totalMessageCount
3098 if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
3220 if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
3100− recipientCounts <- *recipientMsgCounts
3222+ recipientCounts <- readRcBucket msgBucket
3101 let recipientCount = getRecipientCount to recipientCounts
3223 let recipientCount = getRecipientCount to recipientCounts
3102 if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
3224 if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
@@ -3109,7 +3231,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
3109 [{messageId: effectiveMsgId, toKey: to,
3231 [{messageId: effectiveMsgId, toKey: to,
3110 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])
3232 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])
3111 set *totalMessageCount = totalMsgCount + 1
3233 set *totalMessageCount = totalMsgCount + 1
3112− set *recipientMsgCounts = adjustRecipientCount recipientCounts to 1
3234+ writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)
3113 signalMessageForKey to seqTs
3235 signalMessageForKey to seqTs
3114 let toPresence = presenceForKey recipientPresence to t
3236 let toPresence = presenceForKey recipientPresence to t
3115 let localOnline = hasLocalPresence toPresence
3237 let localOnline = hasLocalPresence toPresence
@@ -3449,9 +3571,10 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
3449 freshEvents <- atomic do
3571 freshEvents <- atomic do
3450 recentGossip <- *recentGossip
3572 recentGossip <- *recentGossip
3451 let active = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
3573 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)
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).
3453− let batchPubkeys = fold (\acc e -> if any acc (\a -> a == e.pubkey) then acc else union acc [e.pubkey]) [] effectiveEvents
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
3454− let relevantActive = do r <- active; where any batchPubkeys (\pk -> pk == r.pubkey); yield r
3577+ let relevantActive = do r <- active; where any batchPairs (\p -> p.pubkey == r.pubkey && p.server == r.server); yield r
3455 let fresh = do e <- effectiveEvents; where none relevantActive (\r -> r.pubkey == e.pubkey && r.server == e.server && r.timestamp == e.timestamp); yield e
3578 let fresh = do e <- effectiveEvents; where none relevantActive (\r -> r.pubkey == e.pubkey && r.server == e.server && r.timestamp == e.timestamp); yield e
3456 let newRecent = do e <- fresh; yield {pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, seenAt: t}
3579 let newRecent = do e <- fresh; yield {pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, seenAt: t}
3457 let updated = union active newRecent
3580 let updated = union active newRecent
@@ -3524,7 +3647,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3524 if totalMsgCount >= maxMessages
3647 if totalMsgCount >= maxMessages
3525 then yield {forward: ForwardRejected {}, globalCapReached: True {}, recvRateLimited: False {}}
3648 then yield {forward: ForwardRejected {}, globalCapReached: True {}, recvRateLimited: False {}}
3527− recipientCounts <- *recipientMsgCounts
3650+ recipientCounts <- readRcBucket msgBucket
3528 let recipientCount = getRecipientCount toKey recipientCounts
3651 let recipientCount = getRecipientCount toKey recipientCounts
3529 if recipientCount >= maxMessagesPerRecipient
3652 if recipientCount >= maxMessagesPerRecipient
3530 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
3653 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}}
@@ -3548,7 +3671,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3548 [{messageId: messageId, toKey: toKey,
3671 [{messageId: messageId, toKey: toKey,
3549 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}])
3672 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}])
3550 set *totalMessageCount = totalMsgCount + 1
3673 set *totalMessageCount = totalMsgCount + 1
3551− set *recipientMsgCounts = adjustRecipientCount recipientCounts toKey 1
3674+ writeRcBucket msgBucket (adjustRecipientCount recipientCounts toKey 1)
3552 signalMessageForKey toKey seqTs
3675 signalMessageForKey toKey seqTs
3553 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3676 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3554 fwdNonces <- *forwardNonces
3677 fwdNonces <- *forwardNonces
@@ -3632,6 +3755,19 @@ handleDrain = \authorization respond -> do
3632 yield {forwards: f, pendingAcks: pa}
3755 yield {forwards: f, pendingAcks: pa}
3633 let pendingForwardCount = count flushState.forwards
3756 let pendingForwardCount = count flushState.forwards
3634 let pendingAckCount = count flushState.pendingAcks
3757 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
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)")
3635 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
3771 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
3636 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
3772 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
3637 atomic do set *drainCompletedAt = 0
3773 atomic do set *drainCompletedAt = 0
@@ -3725,8 +3861,8 @@ handleHealth = \authorization respond -> do
3725 snapshotOk: snapOk,
3861 snapshotOk: snapOk,
3726 watchdogOk: wdOk}
3862 watchdogOk: wdOk}
3727 let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
3863 let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
3728− -- Unauthenticated: always return ok:true to prevent information leakage.
3864+ -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
3729− -- Attackers could correlate ok:false transitions with targeted load to infer subsystem pressure.
3865+ -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
3730 if isAuthed
3866 if isAuthed
3731 then yield (respond {ok: allOk, pruneHealthy: result.pruneOk,
3867 then yield (respond {ok: allOk, pruneHealthy: result.pruneOk,
3732 heartbeatHealthy: result.heartbeatOk,
3868 heartbeatHealthy: result.heartbeatOk,
@@ -3734,14 +3870,14 @@ handleHealth = \authorization respond -> do
3734 federationHealthy: result.federationOk,
3870 federationHealthy: result.federationOk,
3735 snapshotHealthy: result.snapshotOk,
3871 snapshotHealthy: result.snapshotOk,
3736 watchdogHealthy: result.watchdogOk})
3872 watchdogHealthy: result.watchdogOk})
3737− else yield (respond {ok: True {}, pruneHealthy: True {},
3873+ else yield (respond {ok: allOk, pruneHealthy: True {},
3738 heartbeatHealthy: True {},
3874 heartbeatHealthy: True {},
3739 capacityHealthy: True {},
3875 capacityHealthy: True {},
3740 federationHealthy: True {},
3876 federationHealthy: True {},
3741 snapshotHealthy: True {},
3877 snapshotHealthy: True {},
3742 watchdogHealthy: True {}})
3878 watchdogHealthy: True {}})
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 {}}
3746handleStats = \authorization respond -> do
3882handleStats = \authorization respond -> do
3747 t <- now
3883 t <- now
@@ -3764,7 +3900,7 @@ handleStats = \authorization respond -> do
3764 ff <- *forwardFailures
3900 ff <- *forwardFailures
3765 ap <- *activePolls
3901 ap <- *activePolls
3766 tmc <- *totalMessageCount
3902 tmc <- *totalMessageCount
3767− rc <- *recipientMsgCounts
3903+ rc <- countAllRcEntries 0
3768 fi <- *federationInFlight
3904 fi <- *federationInFlight
3769 rg <- *recentGossip
3905 rg <- *recentGossip
3770 fn <- *forwardNonces
3906 fn <- *forwardNonces
@@ -3772,7 +3908,8 @@ handleStats = \authorization respond -> do
3772 bp <- *blockedPeers
3908 bp <- *blockedPeers
3773 ger <- *gossipEventRates
3909 ger <- *gossipEventRates
3774 vir <- *verifyIpRates
3910 vir <- *verifyIpRates
3911+ ukir <- *uniqueKeyIpRates
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}
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}
3776 counters <- atomic do
3913 counters <- atomic do
3777 sr <- *sendRates
3914 sr <- *sendRates
3778 cr <- *challengeRates
3915 cr <- *challengeRates
@@ -3804,7 +3941,7 @@ handleStats = \authorization respond -> do
3804 let hungFedSlots = count (do r <- core.fi; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs; yield r)
3941 let hungFedSlots = count (do r <- core.fi; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs; yield r)
3805 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
3942 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
3806 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt
3943 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 {}})
3808 else yield (respond zeroStats)
3945 else yield (respond zeroStats)
3810handleStatsPeers = \authorization respond -> do
3947handleStatsPeers = \authorization respond -> do
@@ -3973,8 +4110,8 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
3973 writeMessageBucket b (diff bucket toAck)
4110 writeMessageBucket b (diff bucket toAck)
3974 totalCount <- *totalMessageCount
4111 totalCount <- *totalMessageCount
3975 set *totalMessageCount = max 0 (totalCount - ackCount)
4112 set *totalMessageCount = max 0 (totalCount - ackCount)
3976− recipientCounts <- *recipientMsgCounts
4113+ recipientCounts <- readRcBucket b
3977− set *recipientMsgCounts = adjustRecipientCount recipientCounts pubkey (0 - ackCount)
4114+ writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
3978 let ackedMsgIds = do m <- toAck; yield m.messageId
4115 let ackedMsgIds = do m <- toAck; yield m.messageId
3979 forwards <- *forwards
4116 forwards <- *forwards
3980 set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
4117 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
4139 result <- atomic do
4276 result <- atomic do
4140 allSessions <- readAllSessions
4277 allSessions <- readAllSessions
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.
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}
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}
4142 yield {sessions: keySessions, total: count keySessions}
4281 yield {sessions: keySessions, total: count keySessions}
4143 yield (respond {sessions: result.sessions, count: result.total, ok: True {}})
4282 yield (respond {sessions: result.sessions, count: result.total, ok: True {}})
@@ -4171,7 +4310,44 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do
4171 (fork (sendBatchedGossip result.peerServers [{eventType: GossipOffline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))
4310 (fork (sendBatchedGossip result.peerServers [{eventType: GossipOffline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))
4172 yield (respond {ok: True {}, revokedCount: result.revokedCount})
4311 yield (respond {ok: True {}, revokedCount: result.revokedCount})
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"
4346+-- Slow request threshold for access logging (ms)
4347+slowRequestThreshold : Int<Ms>
4348+slowRequestThreshold = 1000
4174−serve = \req -> case req of
4350+serveInner = \req -> case req of
4175 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
4351 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
4176 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
4352 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
4177 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
4353 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
@@ -4203,6 +4379,19 @@ serve = \req -> case req of
4203 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
4379 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
4204 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
4380 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
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.
4389+ let latency = endTime - startTime
4390+ let path = requestPath req
4392+ then logInfo ("[ACCESS] " ++ path ++ " " ++ show latency ++ "ms")
4393+ else when (latency > slowRequestThreshold) (logInfo ("[SLOW] poll " ++ show latency ++ "ms"))
4206-- ============================================
4395-- ============================================
4207-- Entry Point
4396-- Entry Point
4208-- ============================================
4397-- ============================================
@@ -4212,6 +4401,11 @@ serve = \req -> case req of
4212-- NOTE: This function (and all background loops below) recurse via tail calls.
4401-- NOTE: This function (and all background loops below) recurse via tail calls.
4213-- Knot compiles tail calls to loops, so stack growth is bounded. If Knot's tail-call
4402-- Knot compiles tail calls to loops, so stack growth is bounded. If Knot's tail-call
4214-- optimization is ever disabled, these loops will overflow the stack over time.
4403-- 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.
4215backgroundPrune = do
4409backgroundPrune = do
4216 jitter <- randomInt 5000
4410 jitter <- randomInt 5000
4217 sleep (backgroundPruneInterval + jitter)
4411 sleep (backgroundPruneInterval + jitter)
@@ -4297,9 +4491,11 @@ pollHeartbeat = do
4298-- Background loop: periodically logs a state summary to stdout.
4492-- Background loop: periodically logs a state summary to stdout.
4299-- Operators can capture stdout to detect state drift; use /maintenance/snapshot for full export.
4493-- 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.
4300-- TODO: If Knot adds file I/O primitives, auto-export snapshots to a file
4497-- 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
4498+-- (e.g., /var/lib/skrepka/snapshot.json) in addition to logging.
4302−-- a crash between manual snapshot exports loses all federation state.
4303backgroundSnapshot = do
4499backgroundSnapshot = do
4304 sleep autoSnapshotInterval
4500 sleep autoSnapshotInterval
4305 draining <- *draining
4501 draining <- *draining
@@ -4361,6 +4557,35 @@ backgroundWatchdog = do
4361 set *federationActiveConnections = fold (\acc r -> acc + r.count) 0 inFlight)
4557 set *federationActiveConnections = fold (\acc r -> acc + r.count) 0 inFlight)
4362 backgroundWatchdog
4558 backgroundWatchdog
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
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
4575+ auth <- generateFederationAuth "gossip"
4576+ result <- fetchFederation server (RecvGossip {
4577+ authorization: auth,
4578+ events: [], hopsRemaining: 1, fromServer: serverHostN,
4579+ requestTimestamp: startTime})
4581+ let latency = endTime - startTime
4583+ Err e -> logWarn ("startup: peer " ++ server ++ " unreachable (" ++ show e ++ ") — check DNS/firewall/TLS")
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)
4364main = do
4589main = do
4365 let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
4590 let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
4366 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
4591 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
@@ -4412,6 +4637,8 @@ main = do
4412 when (matchesHostnameOrSubdomain serverHostN "localhost")
4637 when (matchesHostnameOrSubdomain serverHostN "localhost")
4413 (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
4638 (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
4414 logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
4639 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.")
4415 logInfo "Skrepka relay server starting on port 8080..."
4642 logInfo "Skrepka relay server starting on port 8080..."
4416 logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")
4643 logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")
4417 logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
4644 logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
@@ -4437,4 +4664,11 @@ main = do
4437 fork pollHeartbeat
4664 fork pollHeartbeat
4438 fork backgroundWatchdog
4665 fork backgroundWatchdog
4439 when (autoSnapshotInterval > 0) (fork backgroundSnapshot)
4666 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))
4440 listen 8080 serve
4674 listen 8080 serve