@@ -61,6 +61,11 @@ maxSendRatePerKey = 20 -- max messages sent per key per window
6161sendRateWindowMs : Int<Ms>
6262sendRateWindowMs = 1000
6363maxSendRateEntries = 1000
64+-- Per-key revoke rate limiting (separate from challenge to prevent challenge floods from blocking revocations)
65+maxRevokeRatePerKey = 5 -- max revoke requests per key per window
66+revokeRateWindowMs : Int<Ms>
67+revokeRateWindowMs = 1000
68+maxRevokeRateEntries = 1000
6469-- Per-recipient receive rate limiting
6570maxRecvRatePerKey = 10 -- max messages received by a key per window (across all senders)
6671recvRateWindowMs : Int<Ms>
@@ -131,8 +136,7 @@ maxRecentFederationRequests = 1000
131136maxForwardNonces = 100000
132137forwardNonceRetention : Int<Ms>
133138forwardNonceRetention = messageMaxAge * 2 -- 60 days; extends 30 days beyond message dedup
134−-- Counter reconciliation: every N prune cycles, recount from truth to self-heal drift
135−reconciliationFrequency = 10
139+-- (reconciliationFrequency removed: message counters now reconcile every prune cycle)
136140
137141-- Federation latency metric window
138142federationLatencyWindowMs : Int<Ms>
@@ -151,6 +155,9 @@ maxGlobalFederationConcurrency = 30
151155-- Peer blocklist: max blocked servers tracked at runtime (managed via /maintenance/block, /maintenance/unblock)
152156maxBlockedPeers = 100
153157
158+-- Drain timeout: max time for drain flush operations before force-completing
159+drainTimeoutMs : Int<Ms>
160+drainTimeoutMs = 120000 -- 2 minutes; covers forward retries + ack sends + gossip + in-flight wait
154161-- Background task intervals
155162backgroundPruneInterval : Int<Ms>
156163backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
@@ -212,6 +219,10 @@ maxPendingAckCleanups = 10000
212219-- Per-key deferred ack cleanup cap (prevents one noisy key from dominating the queue)
213220maxPendingAckCleanupsPerKey = 100
214221
222+-- Health endpoint rate limiting (global, since /health has no clientIp)
223+maxHealthRate = 100 -- max health checks per window
224+healthRateWindowMs : Int<Ms>
225+healthRateWindowMs = 1000
215226-- Health check thresholds (background task freshness)
216227healthPruneMaxAge : Int<Ms>
217228healthPruneMaxAge = backgroundPruneInterval * 3
@@ -428,6 +439,9 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
428439
429440*challengeRates : [ChallengeRateEntry]
430441
442+-- Separate revoke rate limiter (prevents challenge floods from exhausting revoke budget)
443+*revokeRates : [ChallengeRateEntry]
444+
431445-- Heartbeat signal: incremented periodically to wake all blocking long-polls
432446*pollWakeup : Timestamp
433447
@@ -467,6 +481,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
467481
468482*globalRateWindowStart : Timestamp
469483
484+-- Health endpoint rate limiting state (separate from global rate to avoid starving real requests)
485+*healthRateCount : Int
486+*healthRateWindowStart : Timestamp
487+
470488-- Background task cycle counters (exposed via /stats for health monitoring)
471489*pruneCount : Int
472490
@@ -2043,6 +2061,8 @@ pruneExpired = \t -> do
20432061 set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r
20442062 challengeRates <- *challengeRates
20452063 set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r
2064+ revokeRates <- *revokeRates
2065+ set *revokeRates = do r <- revokeRates; where t - r.windowStart < revokeRateWindowMs; yield r
20462066 recvRates <- *recvRates
20472067 set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r
20482068 pairRates <- *pairRates
@@ -2052,6 +2072,11 @@ pruneExpired = \t -> do
20522072 when (t - authWS >= authGlobalRateWindowMs) (do
20532073 set *authGlobalRateWindowStart = t
20542074 set *authGlobalRateCount = 0)
2075+ -- Reset health rate window
2076+ healthWS <- *healthRateWindowStart
2077+ when (t - healthWS >= healthRateWindowMs) (do
2078+ set *healthRateWindowStart = t
2079+ set *healthRateCount = 0)
20552080 atomic do
20562081 federationRates <- *federationRates
20572082 set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs; yield r
@@ -2200,6 +2225,21 @@ checkAuthGlobalRate = \t -> atomic do
22002225 set *authGlobalRateCount = rateCount + 1
22012226 yield (False {})
22022227
2228+-- Health endpoint rate limiting (separate from global rate so monitoring doesn't starve real requests)
2229+checkHealthRate = \t -> atomic do
2230+ windowStart <- *healthRateWindowStart
2231+ rateCount <- *healthRateCount
2232+ if t - windowStart >= healthRateWindowMs
2233+ then do
2234+ set *healthRateWindowStart = t
2235+ set *healthRateCount = 1
2236+ yield (False {})
2237+ else if rateCount >= maxHealthRate
2238+ then yield (True {})
2239+ else do
2240+ set *healthRateCount = rateCount + 1
2241+ yield (False {})
2242+
22032243-- IP-based rate limiting; skips if clientIp header is empty (no reverse proxy)
22042244checkClientIpRate = \ip t ->
22052245 if ip == "" then yield (False {})
@@ -2419,9 +2459,8 @@ retryPendingForwards = \pending -> do
24192459 inFlight <- *federationInFlight
24202460 blocked <- *blockedPeers
24212461 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
2422− sendToEach eligibleServers (\server ->
2423− retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)
2424− yield {}
2462+ let eligibleForwards = do f <- pending; where any eligibleServers (\s -> s == f.toServer); yield f
2463+ sendForwardsPartitioned eligibleForwards eligibleServers maxForwardRetriesPerServer
24252464
24262465-- When a remote key comes online, create forward entries for queued messages and send them
24272466forwardQueuedMessages = \targets t -> do
@@ -2448,8 +2487,8 @@ forwardQueuedMessages = \targets t -> do
24482487 inFlight <- *federationInFlight
24492488 blocked <- *blockedPeers
24502489 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
2451− sendToEach eligibleServers (\server ->
2452− retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)
2490+ let eligibleForwards = do f <- newForwards; where any eligibleServers (\s -> s == f.toServer); yield f
2491+ sendForwardsPartitioned eligibleForwards eligibleServers maxForwardRetriesPerServer
24532492
24542493fetchFederationAck = \origin messageIds -> do
24552494 t <- now
@@ -2624,9 +2663,10 @@ waitForInFlightDrain = \deadline -> do
26242663pruneAndRetry = \t -> do
26252664 atomic do set *lastPrunedAt = t
26262665 pr <- pruneExpired t
2627− -- Counter reconciliation: always reconcile presence and session counters (cheap: just
2628− -- counting list lengths across buckets). Message reconciliation stays periodic because
2629− -- it requires reading all messages and rebuilding per-recipient counts.
2666+ -- Counter reconciliation: reconcile all counters every prune cycle.
2667+ -- Presence and session counters are cheap (just counting list lengths across buckets).
2668+ -- Message reconciliation is heavier (reads all messages, rebuilds per-recipient counts)
2669+ -- but runs every cycle to minimize the drift window from concurrent bucket pruning.
26302670 presDrift <- atomic do
26312671 actual <- countAllPresence
26322672 stored <- *totalPresenceCount
@@ -2640,11 +2680,9 @@ pruneAndRetry = \t -> do
26402680 when (presDrift != 0 || sessDrift != 0)
26412681 (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))
26422682 -- Message reconciliation: per-bucket reads reduce lock contention vs reading all 16 at once
2643− pc <- *pruneCount
2644− when (pc > 0 && pc - (pc / reconciliationFrequency) * reconciliationFrequency == 0) (do
2645− msgDrift <- reconcileMessageCounts 0 []
2646− when (msgDrift != 0)
2647− (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift)))
2683+ msgDrift <- reconcileMessageCounts 0 0
2684+ when (msgDrift != 0)
2685+ (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))
26482686 pendingForwards <- atomic do f <- *forwards; yield f
26492687 ackState <- atomic do
26502688 ackedIds2 <- *ackedIds
@@ -2705,14 +2743,17 @@ pruneAndRetry = \t -> do
27052743 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
27062744
27072745-- Map a single hex character to its integer value (0-15).
2708−-- 16-way if-else chain is mechanical but fine: Knot lacks char-to-int or lookup-table
2709−-- primitives. Clamped to 0-15 by callers (pubkeyBucket, tokenBucket) as a safety net.
2746+-- Handles both lowercase and uppercase hex to prevent hot-bucket skew from
2747+-- federation peers or clients that send uppercase pubkeys.
2748+-- Clamped to 0-15 by callers (pubkeyBucket, tokenBucket) as a safety net.
27102749hexCharIndex : Text -> Int
27112750hexCharIndex = \c ->
27122751 if c == "0" then 0 else if c == "1" then 1 else if c == "2" then 2 else if c == "3" then 3
27132752 else if c == "4" then 4 else if c == "5" then 5 else if c == "6" then 6 else if c == "7" then 7
2714− else if c == "8" then 8 else if c == "9" then 9 else if c == "a" then 10 else if c == "b" then 11
2715− else if c == "c" then 12 else if c == "d" then 13 else if c == "e" then 14 else 15
2753+ else if c == "8" then 8 else if c == "9" then 9
2754+ else if c == "a" || c == "A" then 10 else if c == "b" || c == "B" then 11
2755+ else if c == "c" || c == "C" then 12 else if c == "d" || c == "D" then 13
2756+ else if c == "e" || c == "E" then 14 else 15
27162757
27172758-- Map pubkey to one of 16 signal buckets based on first hex digit.
27182759-- This partitions the signal space so storing a message for one pubkey
@@ -2963,6 +3004,19 @@ acksForServer = \acks server -> do a <- acks; where a.originServer == server; yi
29633004
29643005forwardsForServer = \forwards server -> do f <- forwards; where f.toServer == server; yield f
29653006
3007+-- Partition forwards by server and dispatch sequentially. Each iteration removes
3008+-- processed entries from the working list, so subsequent scans are smaller.
3009+-- Total work: ~n*k/2 vs n*k for the naive sendToEach + forwardsForServer pattern.
3010+sendForwardsPartitioned = \pending servers maxRetries ->
3011+ case headOf servers of
3012+ Nothing {} -> yield {}
3013+ Just {value: server} -> do
3014+ let rest = do s <- servers; where s != server; yield s
3015+ let serverForwards = do f <- pending; where f.toServer == server; yield f
3016+ let remaining = do f <- pending; where f.toServer != server; yield f
3017+ retryForwardsForServer serverForwards maxRetries
3018+ sendForwardsPartitioned remaining rest maxRetries
3019+
29663020-- Only trust offline events that claim to be from the sending server itself
29673021trustedOfflineEvents = \events fromServer ->
29683022 do e <- events; where e.server == fromServer; yield e
@@ -3247,6 +3301,16 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
32473301 set *totalMessageCount = totalMsgCount + 1
32483302 writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)
32493303 signalMessageForKey to seqTs
3304+ -- Record forward nonce for locally-originated messages: prevents replay
3305+ -- if the message expires from messages/ackedIds/deliveredIds and a federation
3306+ -- peer forwards it back. Without this, the nonce check in handleForwardMessage
3307+ -- would miss locally-originated messages entirely.
3308+ fwdNonces <- *forwardNonces
3309+ let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, originServer: "", seenAt: t}]
3310+ set *forwardNonces = if count updatedNonces > maxForwardNonces
3311+ then let evictCount = max 1 (count updatedNonces / 10) in
3312+ (evictSmallest updatedNonces evictCount (\n -> n.seenAt)).kept
3313+ else updatedNonces
32503314 let toPresence = presenceForKey recipientPresence to t
32513315 let localOnline = hasLocalPresence toPresence
32523316 let remoteServers = collectPeerServers toPresence (\p -> p.server)
@@ -3441,14 +3505,14 @@ handleRevoke = \targetToken authorization clientIp respond -> do
34413505 if authLimited then yield (respond {ok: False {}, authorized: True {}})
34423506 else do
34433507 revokeLimited <- atomic do
3444− challengeRates <- *challengeRates
3445− let rateResult = checkPubkeyRate challengeRates callerPubkey maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
3446− set *challengeRates = rateResult.entries
3508+ rr <- *revokeRates
3509+ let rateResult = checkPubkeyRate rr callerPubkey maxRevokeRatePerKey revokeRateWindowMs maxRevokeRateEntries t
3510+ set *revokeRates = rateResult.entries
34473511 yield rateResult.limited
34483512 if revokeLimited then yield (respond {ok: False {}, authorized: True {}})
34493513 else do
34503514 let targetBucket = tokenBucket targetToken
3451− removed <- atomic do
3515+ result <- atomic do
34523516 sessions <- readSessionBucket targetBucket
34533517 let hasTarget = any sessions (\s -> safeCompare s.token targetToken && s.pubkey == callerPubkey)
34543518 if hasTarget
@@ -3456,10 +3520,30 @@ handleRevoke = \targetToken authorization clientIp respond -> do
34563520 writeSessionBucket targetBucket (do s <- sessions; where not (safeCompare s.token targetToken && s.pubkey == callerPubkey); yield s)
34573521 sc <- *totalSessionCount
34583522 set *totalSessionCount = max 0 (sc - 1)
3459− yield (True {})
3460− else yield (False {})
3461− when removed (logInfo ("revoke: session revoked for " ++ callerPubkey))
3462− yield (respond {ok: removed, authorized: True {}}))
3523+ -- Check if this was the last session for this key (same logic as handleLogout)
3524+ hasRemaining <- hasSessionForKey callerPubkey 0
3525+ if not hasRemaining
3526+ then do
3527+ let pb = pubkeyBucket callerPubkey
3528+ presenceBkt <- readPresenceBucket pb
3529+ writePresenceBucket pb (do p <- presenceBkt; where p.pubkey != callerPubkey || not (isLocalServer p.server); yield p)
3530+ let removedCount = count presenceBkt - count (do p <- presenceBkt; where p.pubkey != callerPubkey || not (isLocalServer p.server); yield p)
3531+ tpc <- *totalPresenceCount
3532+ set *totalPresenceCount = max 0 (tpc - removedCount)
3533+ set *presenceVersion = t
3534+ peerServers <- *knownPeerServers
3535+ yield (Just {value: {pubkey: callerPubkey, gossipOffline: True {}, peerServers: peerServers}})
3536+ else yield (Just {value: {pubkey: callerPubkey, gossipOffline: False {}, peerServers: []}})
3537+ else yield (Nothing {})
3538+ case result of
3539+ Nothing {} -> yield (respond {ok: False {}, authorized: True {}})
3540+ Just {value: r} -> do
3541+ when r.gossipOffline (do
3542+ logInfo ("revoke: last session for " ++ r.pubkey ++ ", gossiping offline")
3543+ when (federationEnabled && count r.peerServers > 0)
3544+ (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
3545+ when (not r.gossipOffline) (logInfo ("revoke: session revoked for " ++ r.pubkey))
3546+ yield (respond {ok: True {}, authorized: True {}}))
34633547
34643548-- POST /auth/logout: Voluntary disconnect. Removes the caller's session and, if it was
34653549-- the last active session for this pubkey, clears local presence and gossips offline.
@@ -3687,7 +3771,10 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
36873771 ackedIds <- *ackedIds
36883772 deliveredIds <- *deliveredIds
36893773 forwardNonces <- *forwardNonces
3690− let isDup = checkMessageDup bucket messageId toKey || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds || any forwardNonces (\n -> n.messageId == messageId && n.toKey == toKey && n.originServer == originServer)
3774+ -- Nonce check ignores originServer: a (messageId, toKey) pair is globally unique,
3775+ -- so replays from any origin should be rejected. This also catches locally-originated
3776+ -- messages (originServer="") whose nonces were recorded by sendMessageCore.
3777+ let isDup = checkMessageDup bucket messageId toKey || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds || any forwardNonces (\n -> n.messageId == messageId && n.toKey == toKey)
36913778 if isDup
36923779 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
36933780 else do
@@ -3823,6 +3910,15 @@ handleDrain = \authorization respond -> do
38233910 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
38243911 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
38253912 atomic do set *drainCompletedAt = 0
3913+ -- Fork a deadline timer: force-complete drain if flush ops hang on slow/dead peers.
3914+ -- Without this, operators waiting on drainCompletedAt > 0 could wait forever.
3915+ fork (do
3916+ sleep drainTimeoutMs
3917+ tTimeout <- now
3918+ completed <- atomic do dc <- *drainCompletedAt; yield dc
3919+ when (completed == 0) (do
3920+ logWarn ("drain: timeout after " ++ show drainTimeoutMs ++ "ms, force-completing")
3921+ atomic do set *drainCompletedAt = tTimeout))
38263922 fork (do
38273923 -- Best-effort flush of pending forwards and acks before stopping
38283924 when (federationEnabled && pendingForwardCount > 0) (do
@@ -3837,7 +3933,10 @@ handleDrain = \authorization respond -> do
38373933 -- Wait for in-flight federation requests to complete (timeout: 2x fetch timeout)
38383934 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))
38393935 t2 <- now
3840− atomic do set *drainCompletedAt = t2)
3936+ -- Only set if deadline timer hasn't already force-completed
3937+ atomic do
3938+ dc <- *drainCompletedAt
3939+ when (dc == 0) (set *drainCompletedAt = t2))
38413940 logInfo ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
38423941 when (pendingForwardCount > 0)
38433942 (logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))
@@ -3871,8 +3970,12 @@ handlePrune = \authorization respond -> do
38713970
38723971handleHealth = \authorization respond -> do
38733972 t <- now
3874− let isAuthed = verifyMaintenanceAuth authorization
38753973 -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
3974+ -- but have their own dedicated rate limiter to prevent DDoS on /health from saturating reads.
3975+ healthLimited <- checkHealthRate t
3976+ if healthLimited then yield (respond {ok: True {}, pruneHealthy: True {}, heartbeatHealthy: True {}, capacityHealthy: True {}, federationHealthy: True {}, snapshotHealthy: True {}, watchdogHealthy: True {}})
3977+ else do
3978+ let isAuthed = verifyMaintenanceAuth authorization
38763979 result <- atomic do
38773980 lastPrune <- *lastPrunedAt
38783981 lastHb <- *lastHeartbeatAt
@@ -4127,7 +4230,7 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
41274230 set *forwards = cappedForwards
41284231 -- Restore forward nonces (dedup by messageId + toKey + originServer, keep fresher on conflict)
41294232 forwardNonces <- *forwardNonces
4130− let validNonces = do n <- forwardNoncesData; where t - n.seenAt < forwardNonceRetention; yield n
4233+ let validNonces = do n <- forwardNoncesData; where n.seenAt <= t && t - n.seenAt < forwardNonceRetention; yield n
41314234 let mergedNonces = mergeByKey (\a b -> a.messageId == b.messageId && a.toKey == b.toKey && a.originServer == b.originServer) (\x -> x.seenAt) forwardNonces validNonces
41324235 let cappedNonces = takeRelation maxForwardNonces mergedNonces
41334236 set *forwardNonces = cappedNonces
@@ -4739,7 +4842,7 @@ main = do
47394842 && verificationTokenExpiry > 0
47404843 && maxAckRetries > 0 && maxPairRatePerPair > 0 && pairRateWindowMs > 0
47414844 && maxFederationPairRatePerPair > 0 && federationPairRateWindowMs > 0
4742− && reconciliationFrequency > 0 && maxForwardNonces > 0
4845+ && maxForwardNonces > 0
47434846 && maxGossipEventsPerServer > 0 && gossipEventRateWindowMs > 0
47444847 && maxGlobalFederationConcurrency > 0
47454848 -- Relationship invariants: these misconfigurations cause subtle bugs