a6dc1b0 · skrepka · git.iko.soy

git.iko.soy/skrepka

Implement 11 server.knot improvements: forward nonce for local messages, separate revoke rate limiter, drain timeout, reconcile message counters every cycle, uppercase hex support, forward partitioning optimization, health rate limiting, and other security/reliability hardening

a6dc1b0iko6 months ago

server.knot+135−32

compare with master

unifiedsplit

@@ -61,6 +61,11 @@ maxSendRatePerKey = 20 -- max messages sent per key per window

61sendRateWindowMs : Int<Ms>

62sendRateWindowMs = 1000

63maxSendRateEntries = 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

64-- Per-recipient receive rate limiting

69-- Per-recipient receive rate limiting

65maxRecvRatePerKey = 10 -- max messages received by a key per window (across all senders)

70maxRecvRatePerKey = 10 -- max messages received by a key per window (across all senders)

66recvRateWindowMs : Int<Ms>

71recvRateWindowMs : Int<Ms>

@@ -131,8 +136,7 @@ maxRecentFederationRequests = 1000

131maxForwardNonces = 100000

136maxForwardNonces = 100000

132forwardNonceRetention : Int<Ms>

137forwardNonceRetention : Int<Ms>

133forwardNonceRetention = messageMaxAge * 2 -- 60 days; extends 30 days beyond message dedup

138forwardNonceRetention = messageMaxAge * 2 -- 60 days; extends 30 days beyond message dedup

134−-- Counter reconciliation: every N prune cycles, recount from truth to self-heal drift

139+-- (reconciliationFrequency removed: message counters now reconcile every prune cycle)

135−reconciliationFrequency = 10

136

140

137-- Federation latency metric window

141-- Federation latency metric window

138federationLatencyWindowMs : Int<Ms>

142federationLatencyWindowMs : Int<Ms>

@@ -151,6 +155,9 @@ maxGlobalFederationConcurrency = 30

151-- Peer blocklist: max blocked servers tracked at runtime (managed via /maintenance/block, /maintenance/unblock)

155-- Peer blocklist: max blocked servers tracked at runtime (managed via /maintenance/block, /maintenance/unblock)

152maxBlockedPeers = 100

156maxBlockedPeers = 100

153

157

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

154-- Background task intervals

161-- Background task intervals

155backgroundPruneInterval : Int<Ms>

162backgroundPruneInterval : Int<Ms>

156backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks

163backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks

@@ -212,6 +219,10 @@ maxPendingAckCleanups = 10000

212-- Per-key deferred ack cleanup cap (prevents one noisy key from dominating the queue)

219-- Per-key deferred ack cleanup cap (prevents one noisy key from dominating the queue)

213maxPendingAckCleanupsPerKey = 100

220maxPendingAckCleanupsPerKey = 100

214

221

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

215-- Health check thresholds (background task freshness)

226-- Health check thresholds (background task freshness)

216healthPruneMaxAge : Int<Ms>

227healthPruneMaxAge : Int<Ms>

217healthPruneMaxAge = backgroundPruneInterval * 3

228healthPruneMaxAge = backgroundPruneInterval * 3

@@ -428,6 +439,9 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}

428

439

429*challengeRates : [ChallengeRateEntry]

440*challengeRates : [ChallengeRateEntry]

430

441

442+-- Separate revoke rate limiter (prevents challenge floods from exhausting revoke budget)

443+*revokeRates : [ChallengeRateEntry]

444+

431-- Heartbeat signal: incremented periodically to wake all blocking long-polls

445-- Heartbeat signal: incremented periodically to wake all blocking long-polls

432*pollWakeup : Timestamp

446*pollWakeup : Timestamp

433

447

@@ -467,6 +481,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}

467

481

468*globalRateWindowStart : Timestamp

482*globalRateWindowStart : Timestamp

469

483

484+-- Health endpoint rate limiting state (separate from global rate to avoid starving real requests)

485+*healthRateCount : Int

486+*healthRateWindowStart : Timestamp

487+

470-- Background task cycle counters (exposed via /stats for health monitoring)

488-- Background task cycle counters (exposed via /stats for health monitoring)

471*pruneCount : Int

489*pruneCount : Int

472

490

@@ -2043,6 +2061,8 @@ pruneExpired = \t -> do

2043 set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r

2061 set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r

2044 challengeRates <- *challengeRates

2062 challengeRates <- *challengeRates

2045 set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r

2063 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

2046 recvRates <- *recvRates

2066 recvRates <- *recvRates

2047 set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r

2067 set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r

2048 pairRates <- *pairRates

2068 pairRates <- *pairRates

@@ -2052,6 +2072,11 @@ pruneExpired = \t -> do

2052 when (t - authWS >= authGlobalRateWindowMs) (do

2072 when (t - authWS >= authGlobalRateWindowMs) (do

2053 set *authGlobalRateWindowStart = t

2073 set *authGlobalRateWindowStart = t

2054 set *authGlobalRateCount = 0)

2074 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)

2055 atomic do

2080 atomic do

2056 federationRates <- *federationRates

2081 federationRates <- *federationRates

2057 set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs; yield r

2082 set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs; yield r

@@ -2200,6 +2225,21 @@ checkAuthGlobalRate = \t -> atomic do

2200 set *authGlobalRateCount = rateCount + 1

2225 set *authGlobalRateCount = rateCount + 1

2201 yield (False {})

2226 yield (False {})

2202

2227

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+

2203-- IP-based rate limiting; skips if clientIp header is empty (no reverse proxy)

2243-- IP-based rate limiting; skips if clientIp header is empty (no reverse proxy)

2204checkClientIpRate = \ip t ->

2244checkClientIpRate = \ip t ->

2205 if ip == "" then yield (False {})

2245 if ip == "" then yield (False {})

@@ -2419,9 +2459,8 @@ retryPendingForwards = \pending -> do

2419 inFlight <- *federationInFlight

2459 inFlight <- *federationInFlight

2420 blocked <- *blockedPeers

2460 blocked <- *blockedPeers

2421 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s

2461 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s

2422− sendToEach eligibleServers (\server ->

2462+ let eligibleForwards = do f <- pending; where any eligibleServers (\s -> s == f.toServer); yield f

2423− retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)

2463+ sendForwardsPartitioned eligibleForwards eligibleServers maxForwardRetriesPerServer

2424− yield {}

2425

2464

2426-- When a remote key comes online, create forward entries for queued messages and send them

2465-- When a remote key comes online, create forward entries for queued messages and send them

2427forwardQueuedMessages = \targets t -> do

2466forwardQueuedMessages = \targets t -> do

@@ -2448,8 +2487,8 @@ forwardQueuedMessages = \targets t -> do

2448 inFlight <- *federationInFlight

2487 inFlight <- *federationInFlight

2449 blocked <- *blockedPeers

2488 blocked <- *blockedPeers

2450 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s

2489 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s

2451− sendToEach eligibleServers (\server ->

2490+ let eligibleForwards = do f <- newForwards; where any eligibleServers (\s -> s == f.toServer); yield f

2452− retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)

2491+ sendForwardsPartitioned eligibleForwards eligibleServers maxForwardRetriesPerServer

2453

2492

2454fetchFederationAck = \origin messageIds -> do

2493fetchFederationAck = \origin messageIds -> do

2455 t <- now

2494 t <- now

@@ -2624,9 +2663,10 @@ waitForInFlightDrain = \deadline -> do

2624pruneAndRetry = \t -> do

2663pruneAndRetry = \t -> do

2625 atomic do set *lastPrunedAt = t

2664 atomic do set *lastPrunedAt = t

2626 pr <- pruneExpired t

2665 pr <- pruneExpired t

2627− -- Counter reconciliation: always reconcile presence and session counters (cheap: just

2666+ -- Counter reconciliation: reconcile all counters every prune cycle.

2628− -- counting list lengths across buckets). Message reconciliation stays periodic because

2667+ -- Presence and session counters are cheap (just counting list lengths across buckets).

2629− -- it requires reading all messages and rebuilding per-recipient counts.

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.

2630 presDrift <- atomic do

2670 presDrift <- atomic do

2631 actual <- countAllPresence

2671 actual <- countAllPresence

2632 stored <- *totalPresenceCount

2672 stored <- *totalPresenceCount

@@ -2640,11 +2680,9 @@ pruneAndRetry = \t -> do

2640 when (presDrift != 0 || sessDrift != 0)

2680 when (presDrift != 0 || sessDrift != 0)

2641 (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))

2681 (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))

2642 -- Message reconciliation: per-bucket reads reduce lock contention vs reading all 16 at once

2682 -- 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 []

2683+ msgDrift <- reconcileMessageCounts 0 0

2646− when (msgDrift != 0)

2684+ when (msgDrift != 0)

2647− (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift)))

2685+ (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))

2648 pendingForwards <- atomic do f <- *forwards; yield f

2686 pendingForwards <- atomic do f <- *forwards; yield f

2649 ackState <- atomic do

2687 ackState <- atomic do

2650 ackedIds2 <- *ackedIds

2688 ackedIds2 <- *ackedIds

@@ -2705,14 +2743,17 @@ pruneAndRetry = \t -> do

2705 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))

2743 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))

2706

2744

2707-- Map a single hex character to its integer value (0-15).

2745-- 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

2746+-- Handles both lowercase and uppercase hex to prevent hot-bucket skew from

2747+-- federation peers or clients that send uppercase pubkeys.

2709−-- primitives. Clamped to 0-15 by callers (pubkeyBucket, tokenBucket) as a safety net.

2748+-- Clamped to 0-15 by callers (pubkeyBucket, tokenBucket) as a safety net.

2710hexCharIndex : Text -> Int

2749hexCharIndex : Text -> Int

2711hexCharIndex = \c ->

2750hexCharIndex = \c ->

2712 if c == "0" then 0 else if c == "1" then 1 else if c == "2" then 2 else if c == "3" then 3

2751 if c == "0" then 0 else if c == "1" then 1 else if c == "2" then 2 else if c == "3" then 3

2713 else if c == "4" then 4 else if c == "5" then 5 else if c == "6" then 6 else if c == "7" then 7

2752 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

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

2715− else if c == "c" then 12 else if c == "d" then 13 else if c == "e" then 14 else 15

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

2716

2757

2717-- Map pubkey to one of 16 signal buckets based on first hex digit.

2758-- Map pubkey to one of 16 signal buckets based on first hex digit.

2718-- This partitions the signal space so storing a message for one pubkey

2759-- 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

2963

3004

2964forwardsForServer = \forwards server -> do f <- forwards; where f.toServer == server; yield f

3005forwardsForServer = \forwards server -> do f <- forwards; where f.toServer == server; yield f

2965

3006

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+

2966-- Only trust offline events that claim to be from the sending server itself

3020-- Only trust offline events that claim to be from the sending server itself

2967trustedOfflineEvents = \events fromServer ->

3021trustedOfflineEvents = \events fromServer ->

2968 do e <- events; where e.server == fromServer; yield e

3022 do e <- events; where e.server == fromServer; yield e

@@ -3247,6 +3301,16 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->

3247 set *totalMessageCount = totalMsgCount + 1

3301 set *totalMessageCount = totalMsgCount + 1

3248 writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)

3302 writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)

3249 signalMessageForKey to seqTs

3303 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

3250 let toPresence = presenceForKey recipientPresence to t

3314 let toPresence = presenceForKey recipientPresence to t

3251 let localOnline = hasLocalPresence toPresence

3315 let localOnline = hasLocalPresence toPresence

3252 let remoteServers = collectPeerServers toPresence (\p -> p.server)

3316 let remoteServers = collectPeerServers toPresence (\p -> p.server)

@@ -3441,14 +3505,14 @@ handleRevoke = \targetToken authorization clientIp respond -> do

3441 if authLimited then yield (respond {ok: False {}, authorized: True {}})

3505 if authLimited then yield (respond {ok: False {}, authorized: True {}})

3442 else do

3506 else do

3443 revokeLimited <- atomic do

3507 revokeLimited <- atomic do

3444− challengeRates <- *challengeRates

3508+ rr <- *revokeRates

3445− let rateResult = checkPubkeyRate challengeRates callerPubkey maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t

3509+ let rateResult = checkPubkeyRate rr callerPubkey maxRevokeRatePerKey revokeRateWindowMs maxRevokeRateEntries t

3446− set *challengeRates = rateResult.entries

3510+ set *revokeRates = rateResult.entries

3447 yield rateResult.limited

3511 yield rateResult.limited

3448 if revokeLimited then yield (respond {ok: False {}, authorized: True {}})

3512 if revokeLimited then yield (respond {ok: False {}, authorized: True {}})

3449 else do

3513 else do

3450 let targetBucket = tokenBucket targetToken

3514 let targetBucket = tokenBucket targetToken

3451− removed <- atomic do

3515+ result <- atomic do

3452 sessions <- readSessionBucket targetBucket

3516 sessions <- readSessionBucket targetBucket

3453 let hasTarget = any sessions (\s -> safeCompare s.token targetToken && s.pubkey == callerPubkey)

3517 let hasTarget = any sessions (\s -> safeCompare s.token targetToken && s.pubkey == callerPubkey)

3454 if hasTarget

3518 if hasTarget

@@ -3456,10 +3520,30 @@ handleRevoke = \targetToken authorization clientIp respond -> do

3456 writeSessionBucket targetBucket (do s <- sessions; where not (safeCompare s.token targetToken && s.pubkey == callerPubkey); yield s)

3520 writeSessionBucket targetBucket (do s <- sessions; where not (safeCompare s.token targetToken && s.pubkey == callerPubkey); yield s)

3457 sc <- *totalSessionCount

3521 sc <- *totalSessionCount

3458 set *totalSessionCount = max 0 (sc - 1)

3522 set *totalSessionCount = max 0 (sc - 1)

3459− yield (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: []}})

3460− else yield (False {})

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)))

3461− when removed (logInfo ("revoke: session revoked for " ++ callerPubkey))

3545+ when (not r.gossipOffline) (logInfo ("revoke: session revoked for " ++ r.pubkey))

3462− yield (respond {ok: removed, authorized: True {}}))

3546+ yield (respond {ok: True {}, authorized: True {}}))

3463

3547

3464-- POST /auth/logout: Voluntary disconnect. Removes the caller's session and, if it was

3548-- POST /auth/logout: Voluntary disconnect. Removes the caller's session and, if it was

3465-- the last active session for this pubkey, clears local presence and gossips offline.

3549-- the last active session for this pubkey, clears local presence and gossips offline.

@@ -3687,7 +3771,10 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

3687 ackedIds <- *ackedIds

3771 ackedIds <- *ackedIds

3688 deliveredIds <- *deliveredIds

3772 deliveredIds <- *deliveredIds

3689 forwardNonces <- *forwardNonces

3773 forwardNonces <- *forwardNonces

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.

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)

3777+ let isDup = checkMessageDup bucket messageId toKey || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds || any forwardNonces (\n -> n.messageId == messageId && n.toKey == toKey)

3691 if isDup

3778 if isDup

3692 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}

3779 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}

3693 else do

3780 else do

@@ -3823,6 +3910,15 @@ handleDrain = \authorization respond -> do

3823 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.

3910 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.

3824 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.

3911 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.

3825 atomic do set *drainCompletedAt = 0

3912 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))

3826 fork (do

3922 fork (do

3827 -- Best-effort flush of pending forwards and acks before stopping

3923 -- Best-effort flush of pending forwards and acks before stopping

3828 when (federationEnabled && pendingForwardCount > 0) (do

3924 when (federationEnabled && pendingForwardCount > 0) (do

@@ -3837,7 +3933,10 @@ handleDrain = \authorization respond -> do

3837 -- Wait for in-flight federation requests to complete (timeout: 2x fetch timeout)

3933 -- Wait for in-flight federation requests to complete (timeout: 2x fetch timeout)

3838 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))

3934 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))

3839 t2 <- now

3935 t2 <- now

3936+ -- Only set if deadline timer hasn't already force-completed

3937+ atomic do

3938+ dc <- *drainCompletedAt

3840− atomic do set *drainCompletedAt = t2)

3939+ when (dc == 0) (set *drainCompletedAt = t2))

3841 logInfo ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")

3940 logInfo ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")

3842 when (pendingForwardCount > 0)

3941 when (pendingForwardCount > 0)

3843 (logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))

3942 (logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))

@@ -3871,8 +3970,12 @@ handlePrune = \authorization respond -> do

3871

3970

3872handleHealth = \authorization respond -> do

3971handleHealth = \authorization respond -> do

3873 t <- now

3972 t <- now

3874− let isAuthed = verifyMaintenanceAuth authorization

3875 -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)

3973 -- 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

3876 result <- atomic do

3979 result <- atomic do

3877 lastPrune <- *lastPrunedAt

3980 lastPrune <- *lastPrunedAt

3878 lastHb <- *lastHeartbeatAt

3981 lastHb <- *lastHeartbeatAt

@@ -4127,7 +4230,7 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD

4127 set *forwards = cappedForwards

4230 set *forwards = cappedForwards

4128 -- Restore forward nonces (dedup by messageId + toKey + originServer, keep fresher on conflict)

4231 -- Restore forward nonces (dedup by messageId + toKey + originServer, keep fresher on conflict)

4129 forwardNonces <- *forwardNonces

4232 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

4131 let mergedNonces = mergeByKey (\a b -> a.messageId == b.messageId && a.toKey == b.toKey && a.originServer == b.originServer) (\x -> x.seenAt) forwardNonces validNonces

4234 let mergedNonces = mergeByKey (\a b -> a.messageId == b.messageId && a.toKey == b.toKey && a.originServer == b.originServer) (\x -> x.seenAt) forwardNonces validNonces

4132 let cappedNonces = takeRelation maxForwardNonces mergedNonces

4235 let cappedNonces = takeRelation maxForwardNonces mergedNonces

4133 set *forwardNonces = cappedNonces

4236 set *forwardNonces = cappedNonces

@@ -4739,7 +4842,7 @@ main = do

4739 && verificationTokenExpiry > 0

4842 && verificationTokenExpiry > 0

4740 && maxAckRetries > 0 && maxPairRatePerPair > 0 && pairRateWindowMs > 0

4843 && maxAckRetries > 0 && maxPairRatePerPair > 0 && pairRateWindowMs > 0

4741 && maxFederationPairRatePerPair > 0 && federationPairRateWindowMs > 0

4844 && maxFederationPairRatePerPair > 0 && federationPairRateWindowMs > 0

4742− && reconciliationFrequency > 0 && maxForwardNonces > 0

4845+ && maxForwardNonces > 0

4743 && maxGossipEventsPerServer > 0 && gossipEventRateWindowMs > 0

4846 && maxGossipEventsPerServer > 0 && gossipEventRateWindowMs > 0

4744 && maxGlobalFederationConcurrency > 0

4847 && maxGlobalFederationConcurrency > 0

4745 -- Relationship invariants: these misconfigurations cause subtle bugs

4848 -- Relationship invariants: these misconfigurations cause subtle bugs