@@ -868,11 +868,29 @@ maxPeerLatencyEntries = 100
868868*nonceBucket14 : [ForwardNonce]
869869*nonceBucket15 : [ForwardNonce]
870870
871−-- Nonce eviction replay protection: tracks the seenAt of the oldest retained nonce
872−-- after LRU eviction. Forwards with timestamps older than this cutoff cannot be
873−-- verified for dedup (their nonces may have been evicted) and are rejected.
874−-- 0 = no eviction has occurred (all nonces retained), accept all timestamps.
875−*nonceEvictionCutoff : Timestamp
871+-- Nonce eviction replay protection: per-bucket cutoff tracking.
872+-- Each bucket independently tracks the seenAt of its oldest retained nonce after LRU
873+-- eviction. Forwards with timestamps older than the *bucket-specific* cutoff are rejected.
874+-- Previous design used a single global cutoff, which incorrectly rejected valid forwards:
875+-- bucket 3 evicting nonces from 5 min ago would set a cutoff that also rejected forwards
876+-- for bucket 12 (which might have nonces from 30 days ago, all retained).
877+-- 0 = no eviction has occurred in this bucket, accept all timestamps.
878+*nonceEvictionCutoff0 : Timestamp
879+*nonceEvictionCutoff1 : Timestamp
880+*nonceEvictionCutoff2 : Timestamp
881+*nonceEvictionCutoff3 : Timestamp
882+*nonceEvictionCutoff4 : Timestamp
883+*nonceEvictionCutoff5 : Timestamp
884+*nonceEvictionCutoff6 : Timestamp
885+*nonceEvictionCutoff7 : Timestamp
886+*nonceEvictionCutoff8 : Timestamp
887+*nonceEvictionCutoff9 : Timestamp
888+*nonceEvictionCutoff10 : Timestamp
889+*nonceEvictionCutoff11 : Timestamp
890+*nonceEvictionCutoff12 : Timestamp
891+*nonceEvictionCutoff13 : Timestamp
892+*nonceEvictionCutoff14 : Timestamp
893+*nonceEvictionCutoff15 : Timestamp
876894
877895-- Federation verify per-IP rate limiting (tighter than general IP rate)
878896*verifyIpRates : [IpRateEntry]
@@ -946,6 +964,12 @@ maxPeerLatencyEntries = 100
946964*rateOverrideLookupRate : Int
947965*rateOverrideClientWriteRate : Int
948966
967+-- Runtime-configurable operational parameter overrides (0 = use compile-time default).
968+-- Adjusted via /maintenance/config POST endpoint without recompile+restart.
969+*configOverrideMaxMessagesPerRecipient : Int
970+*configOverrideMaxGossipHops : Int
971+*configOverridePollTimeoutMs : Int
972+
949973-- Rate-limit rejection counters: track how many requests were rejected by each rate limiter.
950974-- Exposed via /stats for operators to tune rate limits without parsing logs.
951975-- Reset on each federation latency window reset (same cadence as other metrics).
@@ -961,8 +985,12 @@ maxPeerLatencyEntries = 100
961985
962986-- Active request counter: tracks concurrent requests for backpressure visibility.
963987-- Incremented at request start, decremented after handler completes.
988+-- If serveInner panics (Knot lacks try-finally), the counter leaks. The prune cycle
989+-- reconciles by checking lastRequestCompletedAt: if no request has completed in
990+-- 3x pollTimeoutMs but activeRequests > 0, the counter is reset.
964991*activeRequests : Int
965992*peakActiveRequests : Int
993+*lastRequestCompletedAt : Timestamp
966994
967995-- Client operation latency tracking (for /stats, reset with federation latency window)
968996*clientPollLatencyTotal : Int<Ms>
@@ -1267,6 +1295,14 @@ route MaintenanceApi where
12671295 POST {newSecret: Text} /maintenance/rotate-secret headers {authorization: Text}
12681296 -> {ok: Bool, error: Text}
12691297 = RotateSecret
1298+ -- Read current effective operational config (compile-time defaults + runtime overrides)
1299+ GET /maintenance/config headers {authorization: Text}
1300+ -> {ok: Bool, maxMessagesPerRecipient: Int, maxGossipHops: Int, pollTimeoutMs: Int}
1301+ = GetConfig
1302+ -- Update operational config at runtime (0 = reset to compile-time default)
1303+ POST {maxMessagesPerRecipient: Int, maxGossipHops: Int, pollTimeoutMs: Int} /maintenance/config headers {authorization: Text}
1304+ -> {ok: Bool}
1305+ = SetConfig
12701306
12711307route HealthApi where
12721308 GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, forwardQueueHealthy: Bool, requestPressureHealthy: Bool, draining: Bool, degradedPeers: Int} = Health
@@ -2851,6 +2887,54 @@ pruneNonceBucketRange = \b endB t ->
28512887 when (count pruned < count bucket) (writeNonceBucket b pruned)
28522888 pruneNonceBucketRange (b + 1) endB t
28532889
2890+-- Per-bucket nonce eviction cutoff read/write helpers.
2891+readNonceCutoff : Int -> IO Timestamp
2892+readNonceCutoff = \b ->
2893+ if b == 0 then do s <- *nonceEvictionCutoff0; yield s
2894+ else if b == 1 then do s <- *nonceEvictionCutoff1; yield s
2895+ else if b == 2 then do s <- *nonceEvictionCutoff2; yield s
2896+ else if b == 3 then do s <- *nonceEvictionCutoff3; yield s
2897+ else if b == 4 then do s <- *nonceEvictionCutoff4; yield s
2898+ else if b == 5 then do s <- *nonceEvictionCutoff5; yield s
2899+ else if b == 6 then do s <- *nonceEvictionCutoff6; yield s
2900+ else if b == 7 then do s <- *nonceEvictionCutoff7; yield s
2901+ else if b == 8 then do s <- *nonceEvictionCutoff8; yield s
2902+ else if b == 9 then do s <- *nonceEvictionCutoff9; yield s
2903+ else if b == 10 then do s <- *nonceEvictionCutoff10; yield s
2904+ else if b == 11 then do s <- *nonceEvictionCutoff11; yield s
2905+ else if b == 12 then do s <- *nonceEvictionCutoff12; yield s
2906+ else if b == 13 then do s <- *nonceEvictionCutoff13; yield s
2907+ else if b == 14 then do s <- *nonceEvictionCutoff14; yield s
2908+ else do s <- *nonceEvictionCutoff15; yield s
2909+
2910+writeNonceCutoff : Int -> Timestamp -> IO ()
2911+writeNonceCutoff = \b val ->
2912+ if b == 0 then set *nonceEvictionCutoff0 = val
2913+ else if b == 1 then set *nonceEvictionCutoff1 = val
2914+ else if b == 2 then set *nonceEvictionCutoff2 = val
2915+ else if b == 3 then set *nonceEvictionCutoff3 = val
2916+ else if b == 4 then set *nonceEvictionCutoff4 = val
2917+ else if b == 5 then set *nonceEvictionCutoff5 = val
2918+ else if b == 6 then set *nonceEvictionCutoff6 = val
2919+ else if b == 7 then set *nonceEvictionCutoff7 = val
2920+ else if b == 8 then set *nonceEvictionCutoff8 = val
2921+ else if b == 9 then set *nonceEvictionCutoff9 = val
2922+ else if b == 10 then set *nonceEvictionCutoff10 = val
2923+ else if b == 11 then set *nonceEvictionCutoff11 = val
2924+ else if b == 12 then set *nonceEvictionCutoff12 = val
2925+ else if b == 13 then set *nonceEvictionCutoff13 = val
2926+ else if b == 14 then set *nonceEvictionCutoff14 = val
2927+ else set *nonceEvictionCutoff15 = val
2928+
2929+-- Reset all 16 per-bucket nonce eviction cutoffs to 0 (accept all timestamps).
2930+-- Called by nonce flush to clear stale cutoffs when all nonces are removed.
2931+resetAllNonceCutoffs : Int -> IO ()
2932+resetAllNonceCutoffs = \b ->
2933+ if b > 15 then yield {}
2934+ else do
2935+ writeNonceCutoff b 0
2936+ resetAllNonceCutoffs (b + 1)
2937+
28542938-- Record a forward nonce and handle capacity eviction with LRU logging.
28552939-- Usable inside or outside atomic blocks (reads/writes nonce bucket refs).
28562940-- Returns eviction info for deferred logging when called inside atomic.
@@ -2866,29 +2950,15 @@ updateForwardNonce = \nb messageId toKey t -> do
28662950 then case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> t - n.seenAt
28672951 else 0
28682952 writeNonceBucket nb evictedNonces
2869− -- Update nonce eviction cutoff: recompute the global minimum oldest-retained-seenAt
2870− -- across all 16 buckets. A per-bucket-only approach incorrectly ratchets the cutoff
2871− -- upward — a bucket with newer evictions would mask older retained nonces in other buckets,
2872− -- causing legitimate forwards with older timestamps to be rejected.
2953+ -- Update per-bucket nonce eviction cutoff: set to the oldest retained nonce in this bucket.
2954+ -- Per-bucket cutoffs are correct because each forward is checked against its own bucket's
2955+ -- cutoff (keyed by pubkeyBucket(toKey)). The previous global cutoff was wrong: bucket 3
2956+ -- evicting recent nonces would tighten the cutoff for bucket 12 (with older retained nonces).
28732957 when (lruEvicted > 0) (do
2874− globalMin <- computeGlobalNonceCutoff 0 0
2875− set *nonceEvictionCutoff = globalMin)
2958+ let bucketMin = case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> n.seenAt
2959+ writeNonceCutoff nb bucketMin)
28762960 yield {lruEvicted: lruEvicted, oldestRetainedAge: oldestRetainedAge}
28772961
2878−-- Compute the global nonce eviction cutoff: the minimum oldest-retained-seenAt across
2879−-- all 16 nonce buckets. Returns 0 if no bucket has undergone eviction (all nonces retained).
2880−-- O(16 * bucket_size) scan, called only after LRU eviction (not on every insert).
2881−computeGlobalNonceCutoff : Int -> Timestamp -> IO Timestamp
2882−computeGlobalNonceCutoff = \b acc ->
2883− if b > 15 then yield acc
2884− else do
2885− bucket <- readNonceBucket b
2886− case minBy bucket (\n -> n.seenAt) of
2887− Nothing {} -> computeGlobalNonceCutoff (b + 1) acc
2888− Just {value: n} ->
2889− let bucketMin = n.seenAt in
2890− computeGlobalNonceCutoff (b + 1) (if acc == 0 then bucketMin else min acc bucketMin)
2891−
28922962-- Log nonce LRU eviction if any occurred. Call outside atomic blocks.
28932963logNonceEviction : Int -> Int -> Int -> IO ()
28942964logNonceEviction = \nb lruEvicted oldestRetainedAge ->
@@ -3429,7 +3499,7 @@ checkRateWindow = \entries matchKey makeNew updateCount rotateWindow getCount ge
34293499 then let prevCount = case entry of Nothing {} -> 0; Just {value: r} -> getCount r
34303500 in union (do r <- entries; where not (matchKey r); yield r) [rotateWindow (makeNew t) prevCount]
34313501 else do r <- entries; yield (if matchKey r then updateCount r else r)
3432− in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs; yield r) else newEntries, retryAfterMs: 0}
3502+ in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs * 2; yield r) else newEntries, retryAfterMs: 0}
34333503
34343504-- Rate limiter helpers: wrap common checkRateWindow patterns to reduce boilerplate.
34353505-- Each returns {limited, entries, retryAfterMs} — caller reads/writes the mutable ref.
@@ -3616,12 +3686,23 @@ mergeByKey = \sameKey getFreshness initial items ->
36163686 else union [e]
36173687 (do a <- acc; where not (sameKey a e); yield a)) initial items
36183688
3619−-- Merge presence entries by (pubkey, server), keeping the one with the latest expiresAt
3620−mergeByPubkeyServer = mergeByKey samePubkeyServer (\x -> x.expiresAt)
3689+-- Merge presence entries by (pubkey, server), keeping the one with the latest expiresAt.
3690+-- Deterministic tiebreaker: when two entries for the same (pubkey, server) have equal
3691+-- expiresAt, prefer the one with the lexicographically smaller server name. Without a
3692+-- tiebreaker, the merge result depends on arrival order, causing presence oscillation
3693+-- when two peers gossip the same pubkey simultaneously with the same expiry.
3694+mergeByPubkeyServer = \initial items ->
3695+ fold (\acc e ->
3696+ if any acc (\a -> samePubkeyServer a e && (a.expiresAt > e.expiresAt || (a.expiresAt == e.expiresAt && a.server <= e.server)))
3697+ then acc
3698+ else union [e]
3699+ (do a <- acc; where not (samePubkeyServer a e); yield a)) initial items
36213700
3622−-- Keep entries from setA that are not superseded by a newer-timestamped entry in setB
3701+-- Keep entries from setA that are not superseded by a newer-timestamped entry in setB.
3702+-- Deterministic tiebreaker: when timestamps are equal, the entry with the lexicographically
3703+-- smaller server name wins (consistent with mergeByPubkeyServer).
36233704resolveByTimestamp = \setA setB ->
3624− do e <- setA; where none setB (\o -> samePubkeyServer o e && o.timestamp > e.timestamp); yield e
3705+ do e <- setA; where none setB (\o -> samePubkeyServer o e && (o.timestamp > e.timestamp || (o.timestamp == e.timestamp && o.server < e.server))); yield e
36253706
36263707-- Remove presence entries that have a corresponding offline event with a newer timestamp
36273708removeOffline = \presenceSet trustedOffline t ->
@@ -3793,18 +3874,23 @@ pruneExpired = \t -> do
37933874 set *forwardFailures = do f <- forwardFailures; where (if isKnown f.server then t - f.failedAt < maxServerDowntime else t - f.failedAt < sessionExpiry); yield f
37943875 ackOverflow <- *ackOverflow
37953876 set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a
3796− -- Prune rate limiters (split into smaller atomic blocks to reduce lock contention)
3877+ -- Prune rate limiters (split into smaller atomic blocks to reduce lock contention).
3878+ -- Retention is 2x windowMs because the sliding window algorithm uses prevCount from the
3879+ -- previous window for interpolation. Pruning at 1x windowMs discards prevCount prematurely,
3880+ -- allowing a brief 2x burst at the next window boundary — the exact problem sliding windows
3881+ -- solve. The entry rotates to a new window (resetting count, storing prevCount) at 1x; the
3882+ -- old prevCount is only irrelevant after 2x when the new window itself rotates.
37973883 atomic do
37983884 sendRates <- *sendRates
3799− set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r
3885+ set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs * 2; yield r
38003886 challengeRates <- *challengeRates
3801− set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r
3887+ set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs * 2; yield r
38023888 revokeRates <- *revokeRates
3803− set *revokeRates = do r <- revokeRates; where t - r.windowStart < revokeRateWindowMs; yield r
3889+ set *revokeRates = do r <- revokeRates; where t - r.windowStart < revokeRateWindowMs * 2; yield r
38043890 recvRates <- *recvRates
3805− set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r
3891+ set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs * 2; yield r
38063892 pairRates <- *pairRates
3807− set *pairRates = do r <- pairRates; where t - r.windowStart < pairRateWindowMs; yield r
3893+ set *pairRates = do r <- pairRates; where t - r.windowStart < pairRateWindowMs * 2; yield r
38083894 -- Reset auth-tier global rate window
38093895 authWS <- *authGlobalRateWindowStart
38103896 when (t - authWS >= authGlobalRateWindowMs) (do
@@ -3817,22 +3903,23 @@ pruneExpired = \t -> do
38173903 set *healthRateCount = 0)
38183904 atomic do
38193905 federationRates <- *federationRates
3820− set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs; yield r
3906+ set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs * 2; yield r
38213907 federationPairRates <- *federationPairRates
3822− set *federationPairRates = do r <- federationPairRates; where t - r.windowStart < federationPairRateWindowMs; yield r
3908+ set *federationPairRates = do r <- federationPairRates; where t - r.windowStart < federationPairRateWindowMs * 2; yield r
38233909 ipRates <- *ipRates
3824− set *ipRates = do r <- ipRates; where t - r.windowStart < ipRateWindowMs; yield r
3910+ set *ipRates = do r <- ipRates; where t - r.windowStart < ipRateWindowMs * 2; yield r
38253911 lookupRates <- *lookupRates
3826− set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs; yield r
3912+ set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs * 2; yield r
38273913 clientWriteRates <- *clientWriteRates
3828− set *clientWriteRates = do r <- clientWriteRates; where t - r.windowStart < clientWriteRateWindowMs; yield r
3914+ set *clientWriteRates = do r <- clientWriteRates; where t - r.windowStart < clientWriteRateWindowMs * 2; yield r
38293915 verifyIpRates <- *verifyIpRates
3830− set *verifyIpRates = do r <- verifyIpRates; where t - r.windowStart < verifyRateWindowMs; yield r
3916+ set *verifyIpRates = do r <- verifyIpRates; where t - r.windowStart < verifyRateWindowMs * 2; yield r
38313917 uniqueKeyIpRates <- *uniqueKeyIpRates
3832− set *uniqueKeyIpRates = do r <- uniqueKeyIpRates; where t - r.windowStart < uniqueKeyIpWindowMs; yield r
3918+ set *uniqueKeyIpRates = do r <- uniqueKeyIpRates; where t - r.windowStart < uniqueKeyIpWindowMs * 2; yield r
38333919 atomic do
38343920 activePolls <- *activePolls
3835− set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p
3921+ effPollTimeout <- effectivePollTimeoutMs
3922+ set *activePolls = do p <- activePolls; where t - p.startedAt < effPollTimeout * 3; yield p
38363923 recentGossip <- *recentGossip
38373924 set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
38383925 -- Clean up stale in-flight entries: remove zeroed entries and reset leaked slots
@@ -3867,7 +3954,7 @@ pruneExpired = \t -> do
38673954 pruneNonceBucketsRoundRobin t
38683955 -- Prune gossip event rate entries
38693956 gossipEventRates <- *gossipEventRates
3870− set *gossipEventRates = do r <- gossipEventRates; where t - r.windowStart < gossipEventRateWindowMs; yield r
3957+ set *gossipEventRates = do r <- gossipEventRates; where t - r.windowStart < gossipEventRateWindowMs * 2; yield r
38713958 -- Prune stale deferred ack cleanups (older than messageMaxAge are never useful)
38723959 pendingCleanups <- *pendingAckCleanups
38733960 set *pendingAckCleanups = do c <- pendingCleanups; where t - c.cleanupAfter < messageMaxAge; yield c
@@ -4065,6 +4152,20 @@ effectiveClientWriteRate = do
40654152 o <- *rateOverrideClientWriteRate
40664153 adaptiveRate (if o > 0 then o else maxClientWriteRatePerKey)
40674154
4155+-- Effective operational parameter readers: return runtime override if set (> 0), else compile-time default.
4156+-- These are NOT adaptively scaled (they are capacity limits, not rate limits).
4157+effectiveMaxMessagesPerRecipient = do
4158+ o <- *configOverrideMaxMessagesPerRecipient
4159+ yield (if o > 0 then o else maxMessagesPerRecipient)
4160+
4161+effectiveMaxGossipHops = do
4162+ o <- *configOverrideMaxGossipHops
4163+ yield (if o > 0 then o else maxGossipHops)
4164+
4165+effectivePollTimeoutMs = do
4166+ o <- *configOverridePollTimeoutMs
4167+ yield (if o > 0 then o else pollTimeoutMs)
4168+
40684169checkFederationRate = \server t -> atomic do
40694170 federationRates <- *federationRates
40704171 fedRate <- effectiveFederationRate
@@ -5081,6 +5182,22 @@ pruneAndRetry = \t -> do
50815182 yield (actual - stored)
50825183 when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0)
50835184 (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift)))
5185+ -- Active request counter reconciliation: detect leaked counters from panicked handlers.
5186+ -- Knot lacks try-finally, so if serveInner panics, the activeRequests decrement is skipped.
5187+ -- If no request has completed in 3x pollTimeoutMs (the longest a request should take)
5188+ -- but activeRequests > 0, the counter has leaked. Reset to 0 and log the recovery.
5189+ arqLeak <- atomic do
5190+ arq <- *activeRequests
5191+ if arq > 0 then do
5192+ lastCompleted <- *lastRequestCompletedAt
5193+ if lastCompleted > 0 && t - lastCompleted > pollTimeoutMs * 3
5194+ then do
5195+ set *activeRequests = 0
5196+ yield arq
5197+ else yield 0
5198+ else yield 0
5199+ when (arqLeak > 0)
5200+ (logWarn ("reconciliation: reset leaked activeRequests counter (was " ++ show arqLeak ++ ")"))
50845201 -- Message reconciliation: cheap check first (sum rcBucket .count fields vs totalMessageCount).
50855202 -- Full scan only when drift exceeds threshold or every Nth cycle.
50865203 -- Small drifts from concurrent bucket pruning self-correct; only trigger the expensive
@@ -5853,7 +5970,8 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
58535970 let capped = takeRelation maxSenderDeliveriesPerKey forKey
58545971 yield capped
58555972 -- Poll slot exhaustion: signal clients to back off for one poll cycle
5856− let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
5973+ effPollTimeout <- effectivePollTimeoutMs
5974+ let pollRetryMs = if result.pollSlotFull then effPollTimeout else 0
58575975 -- Read and drain per-sender drop counters so this sender knows if their receipts were lost.
58585976 -- Each sender sees only their own drops (not a global counter conflating all senders).
58595977 drops <- if not result.authorized || result.pubkey == "" then yield {dr: 0, de: 0}
@@ -5923,7 +6041,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
59236041 -- Return MsQueued (not MsRejected) for per-recipient capacity: prevents
59246042 -- an attacker from probing mailbox fullness for arbitrary pubkeys. Global
59256043 -- capacity still returns MsRejected since it's not per-recipient.
5926− if recipientCount >= maxMessagesPerRecipient then yield {status: MsQueued {}, remoteServers: [], retryAfterMs: 0}
6044+ effMaxPerRecipient <- effectiveMaxMessagesPerRecipient
6045+ if recipientCount >= effMaxPerRecipient then yield {status: MsQueued {}, remoteServers: [], retryAfterMs: 0}
59276046 else do
59286047 recipientPresence <- readPresenceBucket (pubkeyBucket to)
59296048 seqCounter <- *seqCounter
@@ -6370,7 +6489,8 @@ handleVerifyToken = \token scope clientIp respond -> do
63706489-- when a remote key comes online.
63716490handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlationId authorization respond -> do
63726491 t <- now
6373− let cappedHops = clamp 0 maxGossipHops hopsRemaining
6492+ effMaxHops <- effectiveMaxGossipHops
6493+ let cappedHops = clamp 0 effMaxHops hopsRemaining
63746494 let corrId = sanitizeCorrelationId correlationId
63756495 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " gossip from " ++ fromServer))
63766496 -- Cap inbound seenBy lists and filter to known peers to prevent gossip suppression attacks
@@ -6396,7 +6516,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
63966516 then let pc = case entry of Nothing {} -> 0; Just {value: r} -> r.count
63976517 in union (do r <- ger; where r.server != fromServer; yield r) [{server: fromServer, windowStart: t, count: eventCount, prevCount: pc}]
63986518 else do r <- ger; yield (if r.server == fromServer then {server: r.server, windowStart: r.windowStart, count: r.count + eventCount, prevCount: r.prevCount} else r)
6399− set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs; yield r) else newEntries
6519+ set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs * 2; yield r) else newEntries
64006520 yield (False {})
64016521 if eventRateLimited then do
64026522 logEvent "warn" "rate_limited_gossip_volume" ("peer=" ++ fromServer ++ " limit=" ++ show maxGossipEventsPerServer)
@@ -6560,10 +6680,12 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
65606680 if isDup
65616681 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: False {}}
65626682 else do
6563− -- Nonce eviction replay protection: if LRU eviction has occurred, we cannot
6564− -- verify dedup for messages whose nonces may have been evicted. Reject forwards
6565− -- with timestamps older than the eviction cutoff as a safety measure.
6566− nonceCutoff <- *nonceEvictionCutoff
6683+ -- Nonce eviction replay protection: if LRU eviction has occurred in this bucket,
6684+ -- we cannot verify dedup for messages whose nonces may have been evicted. Reject
6685+ -- forwards with timestamps older than this bucket's eviction cutoff.
6686+ -- Per-bucket cutoff (not global) ensures eviction in one bucket doesn't reject
6687+ -- valid forwards destined for a different bucket.
6688+ nonceCutoff <- readNonceCutoff bkt
65676689 if nonceCutoff > 0 && timestamp > 0 && timestamp < nonceCutoff
65686690 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: True {}}
65696691 else do
@@ -6573,7 +6695,8 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
65736695 else do
65746696 recipientCounts <- readRcBucket bkt
65756697 let recipientCount = getRecipientCount toKey recipientCounts
6576− if recipientCount >= maxMessagesPerRecipient
6698+ effMaxPerRecipient <- effectiveMaxMessagesPerRecipient
6699+ if recipientCount >= effMaxPerRecipient
65776700 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: False {}}
65786701 else do
65796702 recvRates <- *recvRates
@@ -6966,14 +7089,15 @@ forwardBucketGroupInner = \entries fromServer b t acc presenceBkt bucket ackedBk
69667089 if isDup
69677090 then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
69687091 else do
6969− -- Nonce eviction replay protection (matches handleForwardMessage)
6970− nonceCutoff <- *nonceEvictionCutoff
7092+ -- Nonce eviction replay protection (per-bucket cutoff, matches handleForwardMessage)
7093+ nonceCutoff <- readNonceCutoff b
69717094 if nonceCutoff > 0 && entry.timestamp > 0 && entry.timestamp < nonceCutoff
69727095 then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
69737096 else do
69747097 totalMsgCount <- *totalMessageCount
69757098 let recipientCount = getRecipientCount entry.toKey rc
6976− if totalMsgCount + msgStored >= maxMessages || recipientCount >= maxMessagesPerRecipient
7099+ effMaxPerRecipient <- effectiveMaxMessagesPerRecipient
7100+ if totalMsgCount + msgStored >= maxMessages || recipientCount >= effMaxPerRecipient
69777101 then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
69787102 else do
69797103 -- Per-recipient receive rate limiting (matches handleForwardMessage)
@@ -7021,58 +7145,51 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
70217145 if count messageIds != count toKeys
70227146 then yield (respond {ok: False {}, error: "mismatched_lengths", receivedCount: 0, retryAfterMs: 0})
70237147 else do
7024− -- Look up senderKey for each message from the local message store.
7025− -- Only locally-originated messages (originServer == "") have senderKey.
7148+ -- Zip parallel arrays into pairs upfront (O(n) once), then consume head-first.
7149+ -- Previous code used fold-to-index-N per element, making receiptsByBucket O(n^2).
70267150 let capped = takeRelation maxMessageIds messageIds
7027− receivedCount <- atomic do
7028− deliveries <- fold (\acc mid ->
7029− -- Find the matching toKey by position (parallel arrays)
7030− let idx = count acc.checked in
7031− let toKey = fold (\tacc tk -> if tacc.i == idx then {i: tacc.i + 1, key: tk} else {i: tacc.i + 1, key: tacc.key}) {i: 0, key: ""} toKeys in
7032− if toKey.key == ""
7033− then {checked: union acc.checked [mid], receipts: acc.receipts}
7034− else do
7035− let b = pubkeyBucket toKey.key
7036− -- Check if we have the original message with senderKey
7037− -- (message may have been pruned, but senderKey was already captured in forward entries by fedDeliveryReceipts)
7038− {checked: union acc.checked [mid], receipts: union acc.receipts [{senderKey: "", messageId: mid, toKey: toKey.key, deliveredAt: t}]})
7039− {checked: [], receipts: []} capped
7040− -- For federation delivery receipts, we need to look up the senderKey from message store
7041− -- Per-bucket: group by pubkeyBucket and look up senderKey from messages
7042− let withSender = fold (\acc r ->
7043− acc) [] deliveries.receipts
7044− -- Simpler approach: record receipts with empty senderKey — the sender can match by messageId
7045− -- (they know which messages they sent). This avoids the expensive per-message lookup.
7046− -- Actually, we need senderKey to route receipts to the right sender's poll response.
7047− -- Fall back to scanning message buckets for the senderKey.
7048− yield (count deliveries.receipts)
7049− -- Look up senderKey per-bucket for the receipt messages
7050− receiptsByBucket capped toKeys t 0
7051− yield (respond {ok: True {}, error: "", receivedCount: receivedCount, retryAfterMs: 0}))
7052−
7053−-- Look up senderKey from messages and record delivery receipts per-bucket.
7054−receiptsByBucket = \messageIds toKeys t idx ->
7055− if idx >= count messageIds then yield {}
7056− else do
7057− -- Extract messageId and toKey at current index
7058− let mid = fold (\acc m -> if acc.i == idx then {i: acc.i + 1, val: m} else {i: acc.i + 1, val: acc.val}) {i: 0, val: ""} messageIds
7059− let tk = fold (\acc k -> if acc.i == idx then {i: acc.i + 1, val: k} else {i: acc.i + 1, val: acc.val}) {i: 0, val: ""} toKeys
7060− when (mid.val != "" && tk.val != "") (do
7061− let b = pubkeyBucket tk.val
7151+ let pairs = zipParallelArrays capped toKeys
7152+ -- Look up senderKey per-bucket and record delivery receipts
7153+ receiptsByBucket pairs t
7154+ yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
7155+
7156+-- Zip two parallel arrays into a list of {messageId, toKey} pairs.
7157+-- Consumes both lists head-first in O(n), replacing the O(n^2) fold-to-index pattern.
7158+zipParallelArrays : [MessageIdHex] -> [PubkeyHex] -> [{messageId: MessageIdHex, toKey: PubkeyHex}]
7159+zipParallelArrays = \mids keys ->
7160+ case headOf mids of
7161+ Nothing {} -> []
7162+ Just {value: mid} ->
7163+ case headOf keys of
7164+ Nothing {} -> []
7165+ Just {value: key} ->
7166+ let restMids = do m <- mids; where m != mid; yield m in
7167+ let restKeys = do k <- keys; where k != key; yield k in
7168+ union [{messageId: mid, toKey: key}] (zipParallelArrays restMids restKeys)
7169+
7170+-- Look up senderKey from messages and record delivery receipts.
7171+-- Consumes pre-zipped pairs head-first (O(n) total, was O(n^2) with index-based access).
7172+receiptsByBucket : [{messageId: MessageIdHex, toKey: PubkeyHex}] -> Timestamp -> IO ()
7173+receiptsByBucket = \pairs t ->
7174+ case headOf pairs of
7175+ Nothing {} -> yield {}
7176+ Just {value: pair} -> do
7177+ let rest = do p <- pairs; where p.messageId != pair.messageId || p.toKey != pair.toKey; yield p
7178+ let b = pubkeyBucket pair.toKey
70627179 atomic do
70637180 bucket <- readMessageBucket b
7064− let msg = headOf (do m <- bucket; where m.messageId == mid.val && m.toKey == tk.val && m.senderKey != "" && m.originServer == ""; yield m)
7181+ let msg = headOf (do m <- bucket; where m.messageId == pair.messageId && m.toKey == pair.toKey && m.senderKey != "" && m.originServer == ""; yield m)
70657182 case msg of
70667183 Nothing {} -> yield {}
70677184 Just {value: m} -> do
70687185 sd <- *senderDeliveries
7069− let isDup = any sd (\d -> d.messageId == mid.val && d.toKey == tk.val)
7186+ let isDup = any sd (\d -> d.messageId == pair.messageId && d.toKey == pair.toKey)
70707187 when (not isDup) (do
7071− let combined = union sd [{senderKey: m.senderKey, messageId: mid.val, toKey: tk.val, deliveredAt: t}]
7188+ let combined = union sd [{senderKey: m.senderKey, messageId: pair.messageId, toKey: pair.toKey, deliveredAt: t}]
70727189 set *senderDeliveries = if count combined > maxSenderDeliveries
70737190 then evictOneOldest combined (\d -> d.deliveredAt)
7074− else combined))
7075− receiptsByBucket messageIds toKeys t (idx + 1)
7191+ else combined)
7192+ receiptsByBucket rest t
70767193
70777194-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
70787195-- Auth checked once; each cancel processed independently. Partial success possible.
@@ -8485,6 +8602,8 @@ handleFlushNonces = \authorization respond -> do
84858602 flushedCount <- atomic do
84868603 total <- countAllNonces 0
84878604 writeAllNonces []
8605+ -- Reset all per-bucket eviction cutoffs (no nonces remain, accept all timestamps)
8606+ resetAllNonceCutoffs 0
84888607 yield total
84898608 logInfo ("nonces/flush: flushed " ++ show flushedCount ++ " forward nonces")
84908609 yield (respond {ok: True {}, flushedCount: flushedCount})
@@ -8552,6 +8671,32 @@ handleSetRateLimits = \globalRequest ipRate sendRate federationRate authGlobal r
85528671 logInfo ("rate-limits: updated — global=" ++ show globalRequest ++ " ip=" ++ show ipRate ++ " send=" ++ show sendRate ++ " federation=" ++ show federationRate ++ " auth=" ++ show authGlobal ++ " recv=" ++ show recvRate ++ " challenge=" ++ show challengeRate ++ " pair=" ++ show pairRate ++ " lookup=" ++ show lookupRate ++ " clientWrite=" ++ show clientWriteRate ++ " (0 = default)")
85538672 yield (respond {ok: True {}})
85548673
8674+-- GET /maintenance/config: Return current effective operational parameters.
8675+handleGetConfig = \authorization respond -> do
8676+ authorized <- verifyMaintenanceAuth authorization
8677+ if not authorized then yield (respond {ok: False {}, maxMessagesPerRecipient: 0, maxGossipHops: 0, pollTimeoutMs: 0})
8678+ else do
8679+ result <- atomic do
8680+ mpr <- effectiveMaxMessagesPerRecipient
8681+ mgh <- effectiveMaxGossipHops
8682+ ptm <- effectivePollTimeoutMs
8683+ yield {maxMessagesPerRecipient: mpr, maxGossipHops: mgh, pollTimeoutMs: ptm}
8684+ yield (respond {ok: True {}, maxMessagesPerRecipient: result.maxMessagesPerRecipient, maxGossipHops: result.maxGossipHops, pollTimeoutMs: result.pollTimeoutMs})
8685+
8686+-- POST /maintenance/config: Update operational parameters at runtime. 0 = reset to compile-time default.
8687+-- Changes take effect immediately for new requests. Existing long-polls continue with old values.
8688+handleSetConfig = \newMaxMessagesPerRecipient newMaxGossipHops newPollTimeoutMs authorization respond -> do
8689+ authorized <- verifyMaintenanceAuth authorization
8690+ if not authorized then yield (respond {ok: False {}})
8691+ else do
8692+ logEvent "audit" "set_config" ("maxMessagesPerRecipient=" ++ show newMaxMessagesPerRecipient ++ " maxGossipHops=" ++ show newMaxGossipHops ++ " pollTimeoutMs=" ++ show newPollTimeoutMs)
8693+ atomic do
8694+ set *configOverrideMaxMessagesPerRecipient = max 0 newMaxMessagesPerRecipient
8695+ set *configOverrideMaxGossipHops = max 0 newMaxGossipHops
8696+ set *configOverridePollTimeoutMs = max 0 newPollTimeoutMs
8697+ logInfo ("config: updated — maxMessagesPerRecipient=" ++ show newMaxMessagesPerRecipient ++ " maxGossipHops=" ++ show newMaxGossipHops ++ " pollTimeoutMs=" ++ show newPollTimeoutMs ++ " (0 = default)")
8698+ yield (respond {ok: True {}})
8699+
85558700-- Extract request path for access logging (excludes poll to reduce log volume)
85568701requestPath = \req -> case req of
85578702 ReqChallenge {} -> "POST /auth/challenge"
@@ -8599,6 +8744,8 @@ requestPath = \req -> case req of
85998744 ListAllowed {} -> "GET /maintenance/allowed"
86008745 Reconcile {} -> "POST /maintenance/reconcile"
86018746 RotateSecret {} -> "POST /maintenance/rotate-secret"
8747+ GetConfig {} -> "GET /maintenance/config"
8748+ SetConfig {} -> "POST /maintenance/config"
86028749 Health {} -> "GET /health"
86038750 Stats {} -> "GET /stats"
86048751 StatsPeers {} -> "GET /stats/peers"
@@ -8688,6 +8835,8 @@ serveInner = \req -> case req of
86888835 ListAllowed {authorization, respond} -> handleListAllowed authorization respond
86898836 Reconcile {authorization, respond} -> handleReconcile authorization respond
86908837 RotateSecret {newSecret, authorization, respond} -> handleRotateSecret newSecret authorization respond
8838+ GetConfig {authorization, respond} -> handleGetConfig authorization respond
8839+ SetConfig {maxMessagesPerRecipient, maxGossipHops, pollTimeoutMs, authorization, respond} -> handleSetConfig maxMessagesPerRecipient maxGossipHops pollTimeoutMs authorization respond
86918840 Health {authorization, respond} -> handleHealth authorization respond
86928841 Stats {authorization, respond} -> handleStats authorization respond
86938842 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
@@ -8769,6 +8918,8 @@ serveOverloaded = \req -> case req of
87698918 UnallowPeer {respond} -> respond {ok: False {}, removed: False {}}
87708919 ListAllowed {respond} -> respond {peers: [], ok: False {}}
87718920 RotateSecret {respond} -> respond {ok: False {}, error: "overloaded"}
8921+ GetConfig {respond} -> respond {ok: False {}, maxMessagesPerRecipient: 0, maxGossipHops: 0, pollTimeoutMs: 0}
8922+ SetConfig {respond} -> respond {ok: False {}}
87728923 Metrics {respond} -> respond {metrics: "", ok: False {}}
87738924 -- Exempt routes handled by isOverloadExempt; this branch is unreachable
87748925 _ -> yield {}
@@ -8800,6 +8951,7 @@ serve = \req -> do
88008951 atomic do
88018952 ar <- *activeRequests
88028953 set *activeRequests = max 0 (ar - 1)
8954+ set *lastRequestCompletedAt = startTime
88038955 endTime <- now
88048956 let latency = endTime - startTime
88058957 -- Record client operation latency for /stats monitoring (windowed avg/max).
@@ -8996,12 +9148,13 @@ pollHeartbeat = do
89969148 -- Adaptive poll timeout: scale heartbeat interval based on server load.
89979149 -- adaptiveRateScale returns 10-100 (percentage). At 100% (normal): full pollTimeoutMs.
89989150 -- At 10% (peak load): max(5000, pollTimeoutMs * 10 / 100) = 5000ms.
9151+ basePollTimeout <- effectivePollTimeoutMs
89999152 scale <- atomic do
90009153 tmc <- *totalMessageCount
90019154 sc <- *totalSessionCount
90029155 pc <- *totalPresenceCount
90039156 yield (adaptiveRateScale tmc sc pc)
9004− let effectiveTimeout = max 5000 (pollTimeoutMs * scale / 100)
9157+ let effectiveTimeout = max 5000 (basePollTimeout * scale / 100)
90059158 jitter <- randomInt 2000
90069159 sleep (effectiveTimeout + jitter)
90079160 dp <- *drainPhase