3a92fc8 · skrepka · git.iko.soy

git.iko.soy/skrepka

Implement 9 server optimizations: consolidate forward bucket analysis, optimize batch recipient operations, add ackGracePeriod runtime config, and track nonce eviction metrics

3a92fc8iko6 months ago

server.knot+114−56

compare with master

unifiedsplit

@@ -969,6 +969,7 @@ maxPeerLatencyEntries = 100

969*configOverrideMaxMessagesPerRecipient : Int

970*configOverrideMaxGossipHops : Int

971*configOverridePollTimeoutMs : Int

972+*configOverrideAckGracePeriod : Int

972

973

973-- Rate-limit rejection counters: track how many requests were rejected by each rate limiter.

974-- Rate-limit rejection counters: track how many requests were rejected by each rate limiter.

974-- Exposed via /stats for operators to tune rate limits without parsing logs.

975-- Exposed via /stats for operators to tune rate limits without parsing logs.

@@ -1055,6 +1056,11 @@ clientLatencyT3 = 1000

1055-- Avoids 3 STM reads per rate check on the hot path. 100 = no scaling (default).

1056-- Avoids 3 STM reads per rate check on the hot path. 100 = no scaling (default).

1056*cachedAdaptiveScale : Int

1057*cachedAdaptiveScale : Int

1057

1058

1059+-- Nonce eviction counter: tracks total LRU nonce evictions within the current metrics window.

1060+-- Exposed in /stats capacity for operators to monitor replay-window health.

1061+-- Reset with federation latency window (same cadence as other windowed metrics).

1062+*nonceEvictionCount : Int

1063+

1058-- Gossip send metrics (windowed, reset with federation latency metrics)

1064-- Gossip send metrics (windowed, reset with federation latency metrics)

1059*gossipSendSuccess : Int

1065*gossipSendSuccess : Int

1060*gossipSendFail : Int

1066*gossipSendFail : Int

@@ -1302,10 +1308,10 @@ route MaintenanceApi where

1302 = RotateSecret

1308 = RotateSecret

1303 -- Read current effective operational config (compile-time defaults + runtime overrides)

1309 -- Read current effective operational config (compile-time defaults + runtime overrides)

1304 GET /maintenance/config headers {authorization: Text}

1310 GET /maintenance/config headers {authorization: Text}

1305− -> {ok: Bool, maxMessagesPerRecipient: Int, maxGossipHops: Int, pollTimeoutMs: Int}

1311+ -> {ok: Bool, maxMessagesPerRecipient: Int, maxGossipHops: Int, pollTimeoutMs: Int, ackGracePeriod: Int}

1306 = GetConfig

1312 = GetConfig

1307 -- Update operational config at runtime (0 = reset to compile-time default)

1313 -- Update operational config at runtime (0 = reset to compile-time default)

1308− POST {maxMessagesPerRecipient: Int, maxGossipHops: Int, pollTimeoutMs: Int} /maintenance/config headers {authorization: Text}

1314+ POST {maxMessagesPerRecipient: Int, maxGossipHops: Int, pollTimeoutMs: Int, ackGracePeriod: Int} /maintenance/config headers {authorization: Text}

1309 -> {ok: Bool}

1315 -> {ok: Bool}

1310 = SetConfig

1316 = SetConfig

1311

1317

@@ -1320,7 +1326,7 @@ route HealthApi where

1320-- BREAKING CHANGE from flat format — existing scrapers must update field paths.

1326-- BREAKING CHANGE from flat format — existing scrapers must update field paths.

1321route StatsApi where

1327route StatsApi where

1322 GET /stats headers {authorization: Text}

1328 GET /stats headers {authorization: Text}

1323− -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, activePolls: Int, activePollsMax: Int, recipientCounts: Int, messageBucketMax: Int, messageBucketMin: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}], pendingCancels: Int, gossipSuccess: Int, gossipFail: Int, latencyBuckets: {lt100: Int, lt500: Int, lt2000: Int, gte2000: Int}}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, senderExpirations: Int, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int, clientCancels: Int, federationCancels: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, pollLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int, sendLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, authAvgMs: Int, authMaxMs: Int, authCount: Int}}

1329+ -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, nonceEvictions: Int, activePolls: Int, activePollsMax: Int, recipientCounts: Int, messageBucketMax: Int, messageBucketMin: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}], pendingCancels: Int, gossipSuccess: Int, gossipFail: Int, latencyBuckets: {lt100: Int, lt500: Int, lt2000: Int, gte2000: Int}}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, senderExpirations: Int, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int, clientCancels: Int, federationCancels: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, pollLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int, sendLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, authAvgMs: Int, authMaxMs: Int, authCount: Int}}

1324 = Stats

1330 = Stats

1325 GET /stats/peers headers {authorization: Text}

1331 GET /stats/peers headers {authorization: Text}

1326 -> {peers: [PeerStatus], ok: Bool}

1332 -> {peers: [PeerStatus], ok: Bool}

@@ -2606,18 +2612,22 @@ reconcileMessageCountsPerBucket = \b totalDrift ->

2606 else do

2612 else do

2607 bucketDrift <- atomic do

2613 bucketDrift <- atomic do

2608 bucket <- readMessageBucket b

2614 bucket <- readMessageBucket b

2609− let counts = countByRecipient bucket

2610− let actual = fold (\a c -> a + c.count) 0 counts

2611− -- Read current rcBucket to compute drift vs what's tracked

2612 oldCounts <- readRcBucket b

2615 oldCounts <- readRcBucket b

2613 let tracked = fold (\a c -> a + c.count) 0 oldCounts

2616 let tracked = fold (\a c -> a + c.count) 0 oldCounts

2617+ -- Skip the O(n log n) countByRecipient sort when the bucket total matches.

2618+ -- Per-key imbalances (key A +1, key B -1) are missed but caught by

2619+ -- the periodic full cycle (every reconciliationFullCycleInterval prune cycles).

2620+ if count bucket == tracked then yield 0

2621+ else do

2622+ let counts = countByRecipient bucket

2623+ let actual = fold (\a c -> a + c.count) 0 counts

2614− writeRcBucket b counts

2624+ writeRcBucket b counts

2615− -- Apply per-bucket drift to totalMessageCount atomically

2625+ -- Apply per-bucket drift to totalMessageCount atomically

2616− let drift = actual - tracked

2626+ let drift = actual - tracked

2617− when (drift != 0) (do

2627+ when (drift != 0) (do

2618− stored <- *totalMessageCount

2628+ stored <- *totalMessageCount

2619− set *totalMessageCount = max 0 (stored + drift))

2629+ set *totalMessageCount = max 0 (stored + drift))

2620− yield drift

2630+ yield drift

2621 reconcileMessageCountsPerBucket (b + 1) (totalDrift + bucketDrift)

2631 reconcileMessageCountsPerBucket (b + 1) (totalDrift + bucketDrift)

2622

2632

2623-- ============================================

2633-- ============================================

@@ -2630,6 +2640,10 @@ reconcileMessageCountsPerBucket = \b totalDrift ->

2630-- TODO: If Knot adds a map/dictionary type, replace [RecipientCount] with a

2640-- TODO: If Knot adds a map/dictionary type, replace [RecipientCount] with a

2631-- Map<PubkeyHex, Int> for O(1) lookups.

2641-- Map<PubkeyHex, Int> for O(1) lookups.

2632

2642

2643+-- O(n/16) per lookup (one sharded bucket). fold continues past the match, but each

2644+-- post-match step is a single comparison + branch (no allocation). headOf would also

2645+-- traverse the full list (its fold doesn't short-circuit either) while adding intermediate

2646+-- list allocation from the filter, so fold is the better choice within Knot's constraints.

2633getRecipientCount : PubkeyHex -> [RecipientCount] -> Int

2647getRecipientCount : PubkeyHex -> [RecipientCount] -> Int

2634getRecipientCount = \pubkey counts ->

2648getRecipientCount = \pubkey counts ->

2635 fold (\acc c -> if c.pubkey == pubkey then c.count else acc) 0 counts

2649 fold (\acc c -> if c.pubkey == pubkey then c.count else acc) 0 counts

@@ -2672,11 +2686,15 @@ countSortedGroups = \items ->

2672 in if result.currentKey == "" then result.groups

2686 in if result.currentKey == "" then result.groups

2673 else union [{pubkey: result.currentKey, count: result.currentCount}] result.groups

2687 else union [{pubkey: result.currentKey, count: result.currentCount}] result.groups

2674

2688

2675−-- Batch-decrement recipient counts: subtract per-key removal counts in a single pass

2689+-- Batch-decrement recipient counts: fold over removedCounts applying adjustRecipientCount.

2676−-- instead of calling adjustRecipientCount in a loop.

2690+-- Fast path: returns counts unchanged when removedCounts is empty (common in prune cycles

2691+-- with no expired messages). When non-empty, each adjustRecipientCount is a single pass

2692+-- over counts (O(n*m) total, same as before, but avoids the getRecipientCount lookup

2693+-- overhead for entries in counts that have no matching removal).

2677batchDecrementRecipientCounts : [RecipientCount] -> [RecipientCount] -> [RecipientCount]

2694batchDecrementRecipientCounts : [RecipientCount] -> [RecipientCount] -> [RecipientCount]

2678batchDecrementRecipientCounts = \counts removedCounts ->

2695batchDecrementRecipientCounts = \counts removedCounts ->

2679− do c <- counts; let adj = c.count - getRecipientCount c.pubkey removedCounts; where adj > 0; yield {pubkey: c.pubkey, count: adj}

2696+ if count removedCounts == 0 then counts

2697+ else fold (\acc rc -> adjustRecipientCount acc rc.pubkey (0 - rc.count)) counts removedCounts

2680

2698

2681-- Recipient count bucket helpers (sharded by pubkeyBucket, same as messages)

2699-- Recipient count bucket helpers (sharded by pubkeyBucket, same as messages)

2682readRcBucket : Int -> IO [RecipientCount]

2700readRcBucket : Int -> IO [RecipientCount]

@@ -2970,7 +2988,9 @@ updateForwardNonce = \nb messageId toKey t -> do

2970 -- evicting recent nonces would tighten the cutoff for bucket 12 (with older retained nonces).

2988 -- evicting recent nonces would tighten the cutoff for bucket 12 (with older retained nonces).

2971 when (lruEvicted > 0) (do

2989 when (lruEvicted > 0) (do

2972 let bucketMin = case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> n.seenAt

2990 let bucketMin = case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> n.seenAt

2973− writeNonceCutoff nb bucketMin)

2991+ writeNonceCutoff nb bucketMin

2992+ nec <- *nonceEvictionCount

2993+ set *nonceEvictionCount = nec + lruEvicted)

2974 yield {lruEvicted: lruEvicted, oldestRetainedAge: oldestRetainedAge}

2994 yield {lruEvicted: lruEvicted, oldestRetainedAge: oldestRetainedAge}

2975

2995

2976-- Log nonce LRU eviction if any occurred. Call outside atomic blocks.

2996-- Log nonce LRU eviction if any occurred. Call outside atomic blocks.

@@ -4192,6 +4212,11 @@ effectivePollTimeoutMs = do

4192 o <- *configOverridePollTimeoutMs

4212 o <- *configOverridePollTimeoutMs

4193 yield (if o > 0 then o else pollTimeoutMs)

4213 yield (if o > 0 then o else pollTimeoutMs)

4194

4214

4215+effectiveAckGracePeriod : IO Int<Ms>

4216+effectiveAckGracePeriod = do

4217+ o <- *configOverrideAckGracePeriod

4218+ yield (if o > 0 then o else ackGracePeriod)

4219+

4195checkFederationRate = \server t -> atomic do

4220checkFederationRate = \server t -> atomic do

4196 federationRates <- *federationRates

4221 federationRates <- *federationRates

4197 fedRate <- effectiveFederationRate

4222 fedRate <- effectiveFederationRate

@@ -4560,7 +4585,8 @@ ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of

4560 where not (any fwdBkt (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey))

4585 where not (any fwdBkt (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey))

4561 yield rf

4586 yield rf

4562 -- Phase 2: single pass over messages — remove those matching any eligible entry past grace period

4587 -- Phase 2: single pass over messages — remove those matching any eligible entry past grace period

4588+ effGrace <- effectiveAckGracePeriod

4563− let shouldRemove = \m -> any eligible (\rf -> m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod)

4589+ let shouldRemove = \m -> any eligible (\rf -> m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= effGrace)

4564 let cleanMsgs = do m <- msgs; where not (shouldRemove m); yield m

4590 let cleanMsgs = do m <- msgs; where not (shouldRemove m); yield m

4565 let removedMsgs = do m <- msgs; where shouldRemove m; yield m

4591 let removedMsgs = do m <- msgs; where shouldRemove m; yield m

4566 let removedCount = count removedMsgs

4592 let removedCount = count removedMsgs

@@ -5335,6 +5361,8 @@ pruneAndRetry = \t -> do

5335 -- Reset cancel metrics (windowed with other metrics)

5361 -- Reset cancel metrics (windowed with other metrics)

5336 set *clientCancelCount = 0

5362 set *clientCancelCount = 0

5337 set *federationCancelCount = 0

5363 set *federationCancelCount = 0

5364+ -- Reset nonce eviction counter (windowed with other metrics)

5365+ set *nonceEvictionCount = 0

5338 set *federationLatencyResetAt = t)

5366 set *federationLatencyResetAt = t)

5339 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.

5367 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.

5340 -- These were queued by handleFederationAck instead of fork+sleep.

5368 -- These were queued by handleFederationAck instead of fork+sleep.

@@ -6041,8 +6069,16 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do

6041-- Core send logic: dedup, rate limiting, store message, create forwards.

6069-- Core send logic: dedup, rate limiting, store message, create forwards.

6042-- Called inside an atomic block by both handleSendMessage and sendBatchInner.

6070-- Called inside an atomic block by both handleSendMessage and sendBatchInner.

6043-- Caller handles auth, drain checks, and messageId generation.

6071-- Caller handles auth, drain checks, and messageId generation.

6072+-- This function intentionally runs as a single monolithic atomic block:

6073+-- - Splitting into phases (check → store) introduces TOCTOU gaps that require

6074+-- re-checking dedup in Phase 2, adding overhead without reducing contention.

6075+-- - Rate limits committed in Phase 1 would be wasted if Phase 2 denies (dedup).

6076+-- - sendBatchBucketGroup depends on consistent state across messages in the same

6077+-- bucket group (each call sees the previous call's writes).

6078+-- - 16-way bucket sharding already limits contention to 1/16 of concurrent senders;

6079+-- further splitting yields marginal benefit.

6044sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs msgPriority t ->

6080sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs msgPriority t ->

6045− if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}

6081+ if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0, nonceBucket: 0, nonceEvicted: 0, nonceOldestAge: 0}

6046 else do

6082 else do

6047 let msgBucket = pubkeyBucket to

6083 let msgBucket = pubkeyBucket to

6048 bucket <- readMessageBucket msgBucket

6084 bucket <- readMessageBucket msgBucket

@@ -6053,28 +6089,28 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn

6053 -- expire but before the nonce expires.

6089 -- expire but before the nonce expires.

6054 nonceBkt <- readNonceBucket msgBucket

6090 nonceBkt <- readNonceBucket msgBucket

6055 let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt || any nonceBkt (\n -> n.messageId == effectiveMsgId && n.toKey == to)

6091 let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt || any nonceBkt (\n -> n.messageId == effectiveMsgId && n.toKey == to)

6056− if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}

6092+ if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0, nonceBucket: 0, nonceEvicted: 0, nonceOldestAge: 0}

6057 else do

6093 else do

6058 sendRates <- *sendRates

6094 sendRates <- *sendRates

6059 sendRate <- effectiveSendRate

6095 sendRate <- effectiveSendRate

6060 let rateResult = checkPubkeyRate sendRates senderKey sendRate sendRateWindowMs maxSendRateEntries t

6096 let rateResult = checkPubkeyRate sendRates senderKey sendRate sendRateWindowMs maxSendRateEntries t

6061− if rateResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: rateResult.retryAfterMs}

6097+ if rateResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: rateResult.retryAfterMs, nonceBucket: 0, nonceEvicted: 0, nonceOldestAge: 0}

6062 else do

6098 else do

6063 set *sendRates = rateResult.entries

6099 set *sendRates = rateResult.entries

6064 pairRates <- *pairRates

6100 pairRates <- *pairRates

6065 pRate <- effectivePairRate

6101 pRate <- effectivePairRate

6066 let pairResult = checkPairRate pairRates senderKey to pRate pairRateWindowMs maxPairRateEntries t

6102 let pairResult = checkPairRate pairRates senderKey to pRate pairRateWindowMs maxPairRateEntries t

6067 set *pairRates = pairResult.entries

6103 set *pairRates = pairResult.entries

6068− if pairResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: pairResult.retryAfterMs}

6104+ if pairResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: pairResult.retryAfterMs, nonceBucket: 0, nonceEvicted: 0, nonceOldestAge: 0}

6069 else do

6105 else do

6070 recvRates <- *recvRates

6106 recvRates <- *recvRates

6071 recvRate <- effectiveRecvRate

6107 recvRate <- effectiveRecvRate

6072 let recvResult = checkPubkeyRate recvRates to recvRate recvRateWindowMs maxRecvRateEntries t

6108 let recvResult = checkPubkeyRate recvRates to recvRate recvRateWindowMs maxRecvRateEntries t

6073− if recvResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: recvResult.retryAfterMs}

6109+ if recvResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: recvResult.retryAfterMs, nonceBucket: 0, nonceEvicted: 0, nonceOldestAge: 0}

6074 else do

6110 else do

6075 set *recvRates = recvResult.entries

6111 set *recvRates = recvResult.entries

6076 totalMsgCount <- *totalMessageCount

6112 totalMsgCount <- *totalMessageCount

6077− if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}

6113+ if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0, nonceBucket: 0, nonceEvicted: 0, nonceOldestAge: 0}

6078 else do

6114 else do

6079 recipientCounts <- readRcBucket msgBucket

6115 recipientCounts <- readRcBucket msgBucket

6080 let recipientCount = getRecipientCount to recipientCounts

6116 let recipientCount = getRecipientCount to recipientCounts

@@ -6082,7 +6118,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn

6082 -- an attacker from probing mailbox fullness for arbitrary pubkeys. Global

6118 -- an attacker from probing mailbox fullness for arbitrary pubkeys. Global

6083 -- capacity still returns MsRejected since it's not per-recipient.

6119 -- capacity still returns MsRejected since it's not per-recipient.

6084 effMaxPerRecipient <- effectiveMaxMessagesPerRecipient

6120 effMaxPerRecipient <- effectiveMaxMessagesPerRecipient

6085− if recipientCount >= effMaxPerRecipient then yield {status: MsQueued {}, remoteServers: [], retryAfterMs: 0}

6121+ if recipientCount >= effMaxPerRecipient then yield {status: MsQueued {}, remoteServers: [], retryAfterMs: 0, nonceBucket: 0, nonceEvicted: 0, nonceOldestAge: 0}

6086 else do

6122 else do

6087 recipientPresence <- readPresenceBucket (pubkeyBucket to)

6123 recipientPresence <- readPresenceBucket (pubkeyBucket to)

6088 seqCounter <- *seqCounter

6124 seqCounter <- *seqCounter

@@ -6099,25 +6135,30 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn

6099 -- if the message expires from messages/ackedIds/deliveredIds and a federation

6135 -- if the message expires from messages/ackedIds/deliveredIds and a federation

6100 -- peer forwards it back. Without this, the nonce check in handleForwardMessage

6136 -- peer forwards it back. Without this, the nonce check in handleForwardMessage

6101 -- would miss locally-originated messages entirely.

6137 -- would miss locally-originated messages entirely.

6102− recordForwardNonce effectiveMsgId to t

6138+ -- Inlined: calls updateForwardNonce directly instead of recordForwardNonce

6139+ -- to avoid nested atomic block. Eviction info is returned for deferred logging.

6140+ nonceResult <- updateForwardNonce msgBucket effectiveMsgId to t

6103 let toPresence = presenceForKey recipientPresence to t

6141 let toPresence = presenceForKey recipientPresence to t

6104 let localOnline = hasLocalPresence toPresence

6142 let localOnline = hasLocalPresence toPresence

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

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

6106 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do

6144 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do

6107 let fbi = pubkeyBucket to

6145 let fbi = pubkeyBucket to

6108 fwdBkt <- readFwdBucket fbi

6146 fwdBkt <- readFwdBucket fbi

6109− -- Per-server cap: prevents one unreachable peer from consuming the entire forward queue

6147+ -- Single-pass forward bucket analysis: computes per-server dedup/count,

6148+ -- per-key count, and per-sender-per-key count in one fold over fwdBkt.

6149+ -- Replaces 4 separate scans (2 per server + 2 global) with a single traversal.

6150+ let fwdStats = fold (\acc f ->

6151+ let toKeyMatch = f.toKey == to in

6152+ {keyCount: if toKeyMatch then acc.keyCount + 1 else acc.keyCount,

6153+ senderKeyCount: if senderKey != "" && toKeyMatch && f.senderKey == senderKey then acc.senderKeyCount + 1 else acc.senderKeyCount,

6154+ perServer: do ps <- acc.perServer; yield (if f.toServer == ps.server then {server: ps.server, count: ps.count + 1, hasDup: ps.hasDup || f.messageId == effectiveMsgId} else ps)})

6155+ {keyCount: 0, senderKeyCount: 0, perServer: do s <- remoteServers; yield {server: s, count: 0, hasDup: False {}}}

6156+ fwdBkt

6110− let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s) && count (do f <- fwdBkt; where f.toServer == s; yield f) < maxForwardsPerServer; yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: msgExpiresAt, priority: msgPriority, senderKey: senderKey}

6157+ let newFwds = do ps <- fwdStats.perServer; where not ps.hasDup && ps.count < maxForwardsPerServer; yield {messageId: effectiveMsgId, toServer: ps.server, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: msgExpiresAt, priority: msgPriority, senderKey: senderKey}

6111 totalFwdCount <- *totalForwardCount

6158 totalFwdCount <- *totalForwardCount

6112 let globalCap = maxForwardsEntries - totalFwdCount

6159 let globalCap = maxForwardsEntries - totalFwdCount

6113− -- Per-key cap: prevents one key from consuming disproportionate forward slots

6114− let keyFwdCount = count (do f <- fwdBkt; where f.toKey == to; yield f)

6115− let keyCap = maxForwardsPerKey - keyFwdCount

6160+ let keyCap = maxForwardsPerKey - fwdStats.keyCount

6116− -- Per-sender-per-recipient cap: prevents a single sender from monopolizing

6117− -- a recipient's forward queue, ensuring at least (maxForwardsPerKey /

6118− -- maxForwardsPerSenderPerKey) distinct senders can queue forwards.

6119− let senderKeyFwdCount = if senderKey != "" then count (do f <- fwdBkt; where f.senderKey == senderKey && f.toKey == to; yield f) else 0

6120− let senderKeyCap = if senderKey != "" && senderKeyFwdCount >= maxForwardsPerSenderPerKey then 0 else min globalCap keyCap

6161+ let senderKeyCap = if senderKey != "" && fwdStats.senderKeyCount >= maxForwardsPerSenderPerKey then 0 else min globalCap keyCap

6121 let cappedFwds = takeRelation senderKeyCap newFwds

6162 let cappedFwds = takeRelation senderKeyCap newFwds

6122 writeFwdBucket fbi (union fwdBkt cappedFwds)

6163 writeFwdBucket fbi (union fwdBkt cappedFwds)

6123 set *totalForwardCount = totalFwdCount + count cappedFwds

6164 set *totalForwardCount = totalFwdCount + count cappedFwds

@@ -6127,7 +6168,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn

6127 let status = if localOnline then MsDelivered {}

6168 let status = if localOnline then MsDelivered {}

6128 else if federationEnabled && forwardsEnqueued then MsFederated {}

6169 else if federationEnabled && forwardsEnqueued then MsFederated {}

6129 else MsQueued {}

6170 else MsQueued {}

6130− yield {status: status, remoteServers: remoteServers, retryAfterMs: 0}

6171+ yield {status: status, remoteServers: remoteServers, retryAfterMs: 0, nonceBucket: msgBucket, nonceEvicted: nonceResult.lruEvicted, nonceOldestAge: nonceResult.oldestRetainedAge}

6131

6172

6132-- Fork federation forwards for messages that need to be delivered to remote servers.

6173-- Fork federation forwards for messages that need to be delivered to remote servers.

6133forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->

6174forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->

@@ -6271,8 +6312,10 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a

6271 let clampedTs = clampTimestamp timestamp t

6312 let clampedTs = clampTimestamp timestamp t

6272 result <- atomic do

6313 result <- atomic do

6273 withSessionAuth authorization t clientIp

6314 withSessionAuth authorization t clientIp

6274− (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})

6315+ (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0, nonceBucket: 0, nonceEvicted: 0, nonceOldestAge: 0})

6275 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs clampedPriority t)

6316 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs clampedPriority t)

6317+ -- Deferred nonce eviction logging (outside atomic block to avoid holding the lock during IO)

6318+ logNonceEviction result.nonceBucket result.nonceEvicted result.nonceOldestAge

6276 case result.status of

6319 case result.status of

6277 MsRateLimited {} -> logEvent "warn" "rate_limited_send" ("messageId=" ++ effectiveMsgId)

6320 MsRateLimited {} -> logEvent "warn" "rate_limited_send" ("messageId=" ++ effectiveMsgId)

6278 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (global capacity full)"))

6321 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (global capacity full)"))

@@ -6880,7 +6923,8 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre

6880 when (ackedCount > 0) (do

6923 when (ackedCount > 0) (do

6881 capWarning <- atomic do

6924 capWarning <- atomic do

6882 cleanups <- *pendingAckCleanups

6925 cleanups <- *pendingAckCleanups

6926+ effGrace <- effectiveAckGracePeriod

6883− let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}

6927+ let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + effGrace}

6884 -- Per-key cap: limit cleanups per recipient to prevent one noisy key from dominating the queue

6928 -- Per-key cap: limit cleanups per recipient to prevent one noisy key from dominating the queue

6885 let perKeyCapped = fold (\acc c ->

6929 let perKeyCapped = fold (\acc c ->

6886 let existingForKey = count (do x <- acc; where x.toKey == c.toKey; yield x)

6930 let existingForKey = count (do x <- acc; where x.toKey == c.toKey; yield x)

@@ -7657,7 +7701,7 @@ handleHealth = \authorization respond -> do

7657 draining: result.isDraining,

7701 draining: result.isDraining,

7658 degradedPeers: result.degradedPeers})

7702 degradedPeers: result.degradedPeers})

7659

7703

7660−zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0, messageBucketMax: 0, messageBucketMin: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: [], pendingCancels: 0, gossipSuccess: 0, gossipFail: 0, latencyBuckets: {lt100: 0, lt500: 0, lt2000: 0, gte2000: 0}}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0, clientCancels: 0, federationCancels: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, pollLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0, sendLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, authAvgMs: 0, authMaxMs: 0, authCount: 0}}

7704+zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, nonceEvictions: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0, messageBucketMax: 0, messageBucketMin: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: [], pendingCancels: 0, gossipSuccess: 0, gossipFail: 0, latencyBuckets: {lt100: 0, lt500: 0, lt2000: 0, gte2000: 0}}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0, clientCancels: 0, federationCancels: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, pollLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0, sendLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, authAvgMs: 0, authMaxMs: 0, authCount: 0}}

7661

7705

7662handleStats = \authorization respond -> do

7706handleStats = \authorization respond -> do

7663 t <- now

7707 t <- now

@@ -7753,7 +7797,8 @@ handleStats = \authorization respond -> do

7753 calm <- *clientAuthLatencyMax

7797 calm <- *clientAuthLatencyMax

7754 ccc <- *clientCancelCount

7798 ccc <- *clientCancelCount

7755 fccn <- *federationCancelCount

7799 fccn <- *federationCancelCount

7800+ nec <- *nonceEvictionCount

7756− yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, flb0: flb0, flb1: flb1, flb2: flb2, flb3: flb3, svc: svc, vfc: vfc, dca: dca, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf, arq: arq, parq: parq, cplt: cplt, cplc: cplc, cplm: cplm, cslt: cslt, cslc: cslc, cslm: cslm, gss: gss, gsf: gsf, cplb0: cplb0, cplb1: cplb1, cplb2: cplb2, cplb3: cplb3, cslb0: cslb0, cslb1: cslb1, cslb2: cslb2, cslb3: cslb3, calt: calt, calc: calc, calm: calm, ccc: ccc, fccn: fccn}

7801+ yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, flb0: flb0, flb1: flb1, flb2: flb2, flb3: flb3, svc: svc, vfc: vfc, dca: dca, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf, arq: arq, parq: parq, cplt: cplt, cplc: cplc, cplm: cplm, cslt: cslt, cslc: cslc, cslm: cslm, gss: gss, gsf: gsf, cplb0: cplb0, cplb1: cplb1, cplb2: cplb2, cplb3: cplb3, cslb0: cslb0, cslb1: cslb1, cslb2: cslb2, cslb3: cslb3, calt: calt, calc: calc, calm: calm, ccc: ccc, fccn: fccn, nec: nec}

7757 -- Compute derived metrics

7802 -- Compute derived metrics

7758 let sessions = core.sc

7803 let sessions = core.sc

7759 let messages = core.tmc

7804 let messages = core.tmc

@@ -7778,7 +7823,7 @@ handleStats = \authorization respond -> do

7778 let pollAvg = if counters.cplc > 0 then counters.cplt / counters.cplc else 0

7823 let pollAvg = if counters.cplc > 0 then counters.cplt / counters.cplc else 0

7779 let sendAvg = if counters.cslc > 0 then counters.cslt / counters.cslc else 0

7824 let sendAvg = if counters.cslc > 0 then counters.cslt / counters.cslc else 0

7780 yield (respond {ok: True {}, version: serverBuildVersion,

7825 yield (respond {ok: True {}, version: serverBuildVersion,

7781− capacity: {sessions: sessions, sessionsMax: maxSessions, sessionsPct: sessPct, messages: messages, messagesMax: maxMessages, messagesPct: msgPct, totalMessageCount: messages, presence: presence, presenceMax: maxPresenceEntries, presencePct: presPct, forwards: fwds, forwardsMax: maxForwardsEntries, forwardsPct: fwdPct, challenges: count core.c, challengesMax: maxChallenges, ackedIds: core.aCount, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: core.dCount, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc, messageBucketMax: bucketStats.bucketMax, messageBucketMin: bucketStats.bucketMin},

7826+ capacity: {sessions: sessions, sessionsMax: maxSessions, sessionsPct: sessPct, messages: messages, messagesMax: maxMessages, messagesPct: msgPct, totalMessageCount: messages, presence: presence, presenceMax: maxPresenceEntries, presencePct: presPct, forwards: fwds, forwardsMax: maxForwardsEntries, forwardsPct: fwdPct, challenges: count core.c, challengesMax: maxChallenges, ackedIds: core.aCount, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: core.dCount, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, nonceEvictions: counters.nec, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc, messageBucketMax: bucketStats.bucketMax, messageBucketMin: bucketStats.bucketMin},

7782 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, slowVerifies: counters.svc, verifyFailures: counters.vfc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca, forwardServers: fwdServerCount, maxForwardQueueDepth: maxFwdDepth, topFailingPeers: topFailingPeers, pendingCancels: count core.pfc, gossipSuccess: counters.gss, gossipFail: counters.gsf, latencyBuckets: {lt100: counters.flb0, lt500: counters.flb1, lt2000: counters.flb2, gte2000: counters.flb3}},

7827 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, slowVerifies: counters.svc, verifyFailures: counters.vfc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca, forwardServers: fwdServerCount, maxForwardQueueDepth: maxFwdDepth, topFailingPeers: topFailingPeers, pendingCancels: count core.pfc, gossipSuccess: counters.gss, gossipFail: counters.gsf, latencyBuckets: {lt100: counters.flb0, lt500: counters.flb1, lt2000: counters.flb2, gte2000: counters.flb3}},

7783 rates: {send: count counters.sr, challenge: count counters.cr, federation: count counters.fr, federationPair: count counters.fpr, global: counters.grc, ip: count counters.ir, lookup: count counters.lr, recv: count counters.rr, pair: count counters.prr, gossipEvent: count core.ger, verifyIp: count core.vir, uniqueKeyIp: count core.ukir, pendingAckCleanups: count core.pac, senderExpirations: count core.se, senderDeliveries: count core.sdl, rejectedGlobal: counters.rlg, rejectedIp: counters.rli, rejectedAuth: counters.rla, rejectedPerKey: counters.rlk, rejectedFederation: counters.rlf, clientCancels: counters.ccc, federationCancels: counters.fccn},

7828 rates: {send: count counters.sr, challenge: count counters.cr, federation: count counters.fr, federationPair: count counters.fpr, global: counters.grc, ip: count counters.ir, lookup: count counters.lr, recv: count counters.rr, pair: count counters.prr, gossipEvent: count core.ger, verifyIp: count core.vir, uniqueKeyIp: count core.ukir, pendingAckCleanups: count core.pac, senderExpirations: count core.se, senderDeliveries: count core.sdl, rejectedGlobal: counters.rlg, rejectedIp: counters.rli, rejectedAuth: counters.rla, rejectedPerKey: counters.rlk, rejectedFederation: counters.rlf, clientCancels: counters.ccc, federationCancels: counters.fccn},

7784 timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0, drainPhase: counters.dr, totalEntries: sessions + messages + presence + fwds + count core.c + core.aCount + count core.ao + core.dCount + core.fnCount + count core.ap + count core.fi, activeRequests: counters.arq, peakActiveRequests: counters.parq, pollAvgMs: pollAvg, pollMaxMs: counters.cplm, pollCount: counters.cplc, pollLatencyBuckets: {lt50: counters.cplb0, lt200: counters.cplb1, lt1000: counters.cplb2, gte1000: counters.cplb3}, sendAvgMs: sendAvg, sendMaxMs: counters.cslm, sendCount: counters.cslc, sendLatencyBuckets: {lt50: counters.cslb0, lt200: counters.cslb1, lt1000: counters.cslb2, gte1000: counters.cslb3}, authAvgMs: if counters.calc > 0 then counters.calt / counters.calc else 0, authMaxMs: counters.calm, authCount: counters.calc}})

7829 timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0, drainPhase: counters.dr, totalEntries: sessions + messages + presence + fwds + count core.c + core.aCount + count core.ao + core.dCount + core.fnCount + count core.ap + count core.fi, activeRequests: counters.arq, peakActiveRequests: counters.parq, pollAvgMs: pollAvg, pollMaxMs: counters.cplm, pollCount: counters.cplc, pollLatencyBuckets: {lt50: counters.cplb0, lt200: counters.cplb1, lt1000: counters.cplb2, gte1000: counters.cplb3}, sendAvgMs: sendAvg, sendMaxMs: counters.cslm, sendCount: counters.cslc, sendLatencyBuckets: {lt50: counters.cslb0, lt200: counters.cslb1, lt1000: counters.cslb2, gte1000: counters.cslb3}, authAvgMs: if counters.calc > 0 then counters.calt / counters.calc else 0, authMaxMs: counters.calm, authCount: counters.calc}})

@@ -7973,11 +8018,12 @@ handleMetrics = \authorization respond -> do

7973 gsf <- *gossipSendFail

8018 gsf <- *gossipSendFail

7974 ccc <- *clientCancelCount

8019 ccc <- *clientCancelCount

7975 fccn <- *federationCancelCount

8020 fccn <- *federationCancelCount

8021+ nec <- *nonceEvictionCount

7976− yield {flc: flc, flt: flt, flm: flm, cplc: cplc, cplt: cplt, cslc: cslc, cslt: cslt, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf, gss: gss, gsf: gsf, ccc: ccc, fccn: fccn}

8022+ yield {flc: flc, flt: flt, flm: flm, cplc: cplc, cplt: cplt, cslc: cslc, cslt: cslt, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf, gss: gss, gsf: gsf, ccc: ccc, fccn: fccn, nec: nec}

7977 let fedAvg = if ext.flc > 0 then ext.flt / ext.flc else 0

8023 let fedAvg = if ext.flc > 0 then ext.flt / ext.flc else 0

7978 let pollAvg = if ext.cplc > 0 then ext.cplt / ext.cplc else 0

8024 let pollAvg = if ext.cplc > 0 then ext.cplt / ext.cplc else 0

7979 let sendAvg = if ext.cslc > 0 then ext.cslt / ext.cslc else 0

8025 let sendAvg = if ext.cslc > 0 then ext.cslt / ext.cslc else 0

7980− let extended = "# HELP skrepka_federation_latency_avg_ms Average federation call latency\n# TYPE skrepka_federation_latency_avg_ms gauge\nskrepka_federation_latency_avg_ms " ++ show fedAvg ++ "\n# HELP skrepka_federation_latency_max_ms Max federation call latency\n# TYPE skrepka_federation_latency_max_ms gauge\nskrepka_federation_latency_max_ms " ++ show ext.flm ++ "\n# HELP skrepka_federation_calls_total Federation calls in current window\n# TYPE skrepka_federation_calls_total gauge\nskrepka_federation_calls_total " ++ show ext.flc ++ "\n# HELP skrepka_poll_latency_avg_ms Average poll latency\n# TYPE skrepka_poll_latency_avg_ms gauge\nskrepka_poll_latency_avg_ms " ++ show pollAvg ++ "\n# HELP skrepka_send_latency_avg_ms Average send latency\n# TYPE skrepka_send_latency_avg_ms gauge\nskrepka_send_latency_avg_ms " ++ show sendAvg ++ "\n# HELP skrepka_rate_limited_total Rate-limited requests by type\n# TYPE skrepka_rate_limited_total gauge\nskrepka_rate_limited_total{type=\"global\"} " ++ show ext.rlg ++ "\nskrepka_rate_limited_total{type=\"ip\"} " ++ show ext.rli ++ "\nskrepka_rate_limited_total{type=\"auth\"} " ++ show ext.rla ++ "\nskrepka_rate_limited_total{type=\"per_key\"} " ++ show ext.rlk ++ "\nskrepka_rate_limited_total{type=\"federation\"} " ++ show ext.rlf ++ "\n# HELP skrepka_gossip_total Gossip send results\n# TYPE skrepka_gossip_total gauge\nskrepka_gossip_total{result=\"success\"} " ++ show ext.gss ++ "\nskrepka_gossip_total{result=\"fail\"} " ++ show ext.gsf ++ "\n# HELP skrepka_cancels_total Message cancellations by source\n# TYPE skrepka_cancels_total gauge\nskrepka_cancels_total{source=\"client\"} " ++ show ext.ccc ++ "\nskrepka_cancels_total{source=\"federation\"} " ++ show ext.fccn ++ "\n"

8026+ let extended = "# HELP skrepka_federation_latency_avg_ms Average federation call latency\n# TYPE skrepka_federation_latency_avg_ms gauge\nskrepka_federation_latency_avg_ms " ++ show fedAvg ++ "\n# HELP skrepka_federation_latency_max_ms Max federation call latency\n# TYPE skrepka_federation_latency_max_ms gauge\nskrepka_federation_latency_max_ms " ++ show ext.flm ++ "\n# HELP skrepka_federation_calls_total Federation calls in current window\n# TYPE skrepka_federation_calls_total gauge\nskrepka_federation_calls_total " ++ show ext.flc ++ "\n# HELP skrepka_poll_latency_avg_ms Average poll latency\n# TYPE skrepka_poll_latency_avg_ms gauge\nskrepka_poll_latency_avg_ms " ++ show pollAvg ++ "\n# HELP skrepka_send_latency_avg_ms Average send latency\n# TYPE skrepka_send_latency_avg_ms gauge\nskrepka_send_latency_avg_ms " ++ show sendAvg ++ "\n# HELP skrepka_rate_limited_total Rate-limited requests by type\n# TYPE skrepka_rate_limited_total gauge\nskrepka_rate_limited_total{type=\"global\"} " ++ show ext.rlg ++ "\nskrepka_rate_limited_total{type=\"ip\"} " ++ show ext.rli ++ "\nskrepka_rate_limited_total{type=\"auth\"} " ++ show ext.rla ++ "\nskrepka_rate_limited_total{type=\"per_key\"} " ++ show ext.rlk ++ "\nskrepka_rate_limited_total{type=\"federation\"} " ++ show ext.rlf ++ "\n# HELP skrepka_gossip_total Gossip send results\n# TYPE skrepka_gossip_total gauge\nskrepka_gossip_total{result=\"success\"} " ++ show ext.gss ++ "\nskrepka_gossip_total{result=\"fail\"} " ++ show ext.gsf ++ "\n# HELP skrepka_cancels_total Message cancellations by source\n# TYPE skrepka_cancels_total gauge\nskrepka_cancels_total{source=\"client\"} " ++ show ext.ccc ++ "\nskrepka_cancels_total{source=\"federation\"} " ++ show ext.fccn ++ "\n# HELP skrepka_nonce_evictions_total Nonce LRU evictions in current window\n# TYPE skrepka_nonce_evictions_total gauge\nskrepka_nonce_evictions_total " ++ show ext.nec ++ "\n"

7981 yield (respond {metrics: base ++ extended, ok: True {}})

8027 yield (respond {metrics: base ++ extended, ok: True {}})

7982

8028

7983-- POST /messages/ack: Explicitly acknowledge received messages by ID.

8029-- POST /messages/ack: Explicitly acknowledge received messages by ID.

@@ -8328,7 +8374,8 @@ sendBatchInner = \senderKey messages t acc ->

8328 results <- atomic do

8374 results <- atomic do

8329 dp <- *drainPhase

8375 dp <- *drainPhase

8330 sendBatchBucketGroup senderKey prepared t dp []

8376 sendBatchBucketGroup senderKey prepared t dp []

8331− -- Log warnings for rate-limited/rejected messages outside the atomic block

8377+ -- Log nonce evictions and warnings for rate-limited/rejected messages outside the atomic block

8378+ logBatchNonceEvictions results

8332 logBatchResults results

8379 logBatchResults results

8333 sendBatchInner senderKey sameR.rest t (union acc results)

8380 sendBatchInner senderKey sameR.rest t (union acc results)

8334

8381

@@ -8359,15 +8406,24 @@ sendBatchBucketGroup = \senderKey prepared t dp acc ->

8359 let rest = do p <- prepared; where p != entry; yield p

8406 let rest = do p <- prepared; where p != entry; yield p

8360 let clampedPri = clamp 0 maxMessagePriority entry.msg.priority

8407 let clampedPri = clamp 0 maxMessagePriority entry.msg.priority

8361 result <- if entry.stale

8408 result <- if entry.stale

8362− then yield {messageId: entry.effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, retryAfterMs: 0, fwdExpiresAt: 0, fwdPriority: 0}

8409+ then yield {messageId: entry.effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, retryAfterMs: 0, fwdExpiresAt: 0, fwdPriority: 0, nonceBucket: 0, nonceEvicted: 0, nonceOldestAge: 0}

8363 else do

8410 else do

8364 let clampedTs = clampTimestamp entry.msg.timestamp t

8411 let clampedTs = clampTimestamp entry.msg.timestamp t

8365− if dp >= 2 then yield {messageId: entry.effectiveMsgId, status: MsDraining {}, remoteServers: [], toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: 0, fwdExpiresAt: 0, fwdPriority: 0}

8412+ if dp >= 2 then yield {messageId: entry.effectiveMsgId, status: MsDraining {}, remoteServers: [], toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: 0, fwdExpiresAt: 0, fwdPriority: 0, nonceBucket: 0, nonceEvicted: 0, nonceOldestAge: 0}

8366 else do

8413 else do

8367 sendResult <- sendMessageCore senderKey entry.msg.to entry.msg.encryptedBlob clampedTs entry.effectiveMsgId entry.msg.expiresInMs clampedPri t

8414 sendResult <- sendMessageCore senderKey entry.msg.to entry.msg.encryptedBlob clampedTs entry.effectiveMsgId entry.msg.expiresInMs clampedPri t

8368− yield {messageId: entry.effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryAfterMs, fwdExpiresAt: t + effectiveMessageTtl entry.msg.expiresInMs, fwdPriority: clampedPri}

8415+ yield {messageId: entry.effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryAfterMs, fwdExpiresAt: t + effectiveMessageTtl entry.msg.expiresInMs, fwdPriority: clampedPri, nonceBucket: sendResult.nonceBucket, nonceEvicted: sendResult.nonceEvicted, nonceOldestAge: sendResult.nonceOldestAge}

8369 sendBatchBucketGroup senderKey rest t dp (union acc [result])

8416 sendBatchBucketGroup senderKey rest t dp (union acc [result])

8370

8417

8418+-- Log nonce evictions for batch results outside the atomic block

8419+logBatchNonceEvictions = \results ->

8420+ case headOf results of

8421+ Nothing {} -> yield {}

8422+ Just {value: r} -> do

8423+ let rest = do x <- results; where x != r; yield x

8424+ logNonceEviction r.nonceBucket r.nonceEvicted r.nonceOldestAge

8425+ logBatchNonceEvictions rest

8426+

8371-- Log warnings for batch results outside the atomic block

8427-- Log warnings for batch results outside the atomic block

8372logBatchResults = \results ->

8428logBatchResults = \results ->

8373 case headOf results of

8429 case headOf results of

@@ -8766,27 +8822,29 @@ handleSetRateLimits = \globalRequest ipRate sendRate federationRate authGlobal r

8766-- GET /maintenance/config: Return current effective operational parameters.

8822-- GET /maintenance/config: Return current effective operational parameters.

8767handleGetConfig = \authorization respond -> do

8823handleGetConfig = \authorization respond -> do

8768 authorized <- verifyMaintenanceAuth authorization

8824 authorized <- verifyMaintenanceAuth authorization

8769− if not authorized then yield (respond {ok: False {}, maxMessagesPerRecipient: 0, maxGossipHops: 0, pollTimeoutMs: 0})

8825+ if not authorized then yield (respond {ok: False {}, maxMessagesPerRecipient: 0, maxGossipHops: 0, pollTimeoutMs: 0, ackGracePeriod: 0})

8770 else do

8826 else do

8771 result <- atomic do

8827 result <- atomic do

8772 mpr <- effectiveMaxMessagesPerRecipient

8828 mpr <- effectiveMaxMessagesPerRecipient

8773 mgh <- effectiveMaxGossipHops

8829 mgh <- effectiveMaxGossipHops

8774 ptm <- effectivePollTimeoutMs

8830 ptm <- effectivePollTimeoutMs

8831+ agp <- effectiveAckGracePeriod

8775− yield {maxMessagesPerRecipient: mpr, maxGossipHops: mgh, pollTimeoutMs: ptm}

8832+ yield {maxMessagesPerRecipient: mpr, maxGossipHops: mgh, pollTimeoutMs: ptm, ackGracePeriod: agp}

8776− yield (respond {ok: True {}, maxMessagesPerRecipient: result.maxMessagesPerRecipient, maxGossipHops: result.maxGossipHops, pollTimeoutMs: result.pollTimeoutMs})

8833+ yield (respond {ok: True {}, maxMessagesPerRecipient: result.maxMessagesPerRecipient, maxGossipHops: result.maxGossipHops, pollTimeoutMs: result.pollTimeoutMs, ackGracePeriod: result.ackGracePeriod})

8777

8834

8778-- POST /maintenance/config: Update operational parameters at runtime. 0 = reset to compile-time default.

8835-- POST /maintenance/config: Update operational parameters at runtime. 0 = reset to compile-time default.

8779-- Changes take effect immediately for new requests. Existing long-polls continue with old values.

8836-- Changes take effect immediately for new requests. Existing long-polls continue with old values.

8780−handleSetConfig = \newMaxMessagesPerRecipient newMaxGossipHops newPollTimeoutMs authorization respond -> do

8837+handleSetConfig = \newMaxMessagesPerRecipient newMaxGossipHops newPollTimeoutMs newAckGracePeriod authorization respond -> do

8781 authorized <- verifyMaintenanceAuth authorization

8838 authorized <- verifyMaintenanceAuth authorization

8782 if not authorized then yield (respond {ok: False {}})

8839 if not authorized then yield (respond {ok: False {}})

8783 else do

8840 else do

8784− logEvent "audit" "set_config" ("maxMessagesPerRecipient=" ++ show newMaxMessagesPerRecipient ++ " maxGossipHops=" ++ show newMaxGossipHops ++ " pollTimeoutMs=" ++ show newPollTimeoutMs)

8841+ logEvent "audit" "set_config" ("maxMessagesPerRecipient=" ++ show newMaxMessagesPerRecipient ++ " maxGossipHops=" ++ show newMaxGossipHops ++ " pollTimeoutMs=" ++ show newPollTimeoutMs ++ " ackGracePeriod=" ++ show newAckGracePeriod)

8785 atomic do

8842 atomic do

8786 set *configOverrideMaxMessagesPerRecipient = max 0 newMaxMessagesPerRecipient

8843 set *configOverrideMaxMessagesPerRecipient = max 0 newMaxMessagesPerRecipient

8787 set *configOverrideMaxGossipHops = max 0 newMaxGossipHops

8844 set *configOverrideMaxGossipHops = max 0 newMaxGossipHops

8788 set *configOverridePollTimeoutMs = max 0 newPollTimeoutMs

8845 set *configOverridePollTimeoutMs = max 0 newPollTimeoutMs

8846+ set *configOverrideAckGracePeriod = max 0 newAckGracePeriod

8789− logInfo ("config: updated — maxMessagesPerRecipient=" ++ show newMaxMessagesPerRecipient ++ " maxGossipHops=" ++ show newMaxGossipHops ++ " pollTimeoutMs=" ++ show newPollTimeoutMs ++ " (0 = default)")

8847+ logInfo ("config: updated — maxMessagesPerRecipient=" ++ show newMaxMessagesPerRecipient ++ " maxGossipHops=" ++ show newMaxGossipHops ++ " pollTimeoutMs=" ++ show newPollTimeoutMs ++ " ackGracePeriod=" ++ show newAckGracePeriod ++ " (0 = default)")

8790 yield (respond {ok: True {}})

8848 yield (respond {ok: True {}})

8791

8849

8792-- Extract request path for access logging (excludes poll to reduce log volume)

8850-- Extract request path for access logging (excludes poll to reduce log volume)

@@ -8928,7 +8986,7 @@ serveInner = \req -> case req of

8928 Reconcile {authorization, respond} -> handleReconcile authorization respond

8986 Reconcile {authorization, respond} -> handleReconcile authorization respond

8929 RotateSecret {newSecret, authorization, respond} -> handleRotateSecret newSecret authorization respond

8987 RotateSecret {newSecret, authorization, respond} -> handleRotateSecret newSecret authorization respond

8930 GetConfig {authorization, respond} -> handleGetConfig authorization respond

8988 GetConfig {authorization, respond} -> handleGetConfig authorization respond

8931− SetConfig {maxMessagesPerRecipient, maxGossipHops, pollTimeoutMs, authorization, respond} -> handleSetConfig maxMessagesPerRecipient maxGossipHops pollTimeoutMs authorization respond

8989+ SetConfig {maxMessagesPerRecipient, maxGossipHops, pollTimeoutMs, ackGracePeriod, authorization, respond} -> handleSetConfig maxMessagesPerRecipient maxGossipHops pollTimeoutMs ackGracePeriod authorization respond

8932 Health {authorization, respond} -> handleHealth authorization respond

8990 Health {authorization, respond} -> handleHealth authorization respond

8933 Stats {authorization, respond} -> handleStats authorization respond

8991 Stats {authorization, respond} -> handleStats authorization respond

8934 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond

8992 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond

@@ -9010,7 +9068,7 @@ serveOverloaded = \req -> case req of

9010 UnallowPeer {respond} -> respond {ok: False {}, removed: False {}}

9068 UnallowPeer {respond} -> respond {ok: False {}, removed: False {}}

9011 ListAllowed {respond} -> respond {peers: [], ok: False {}}

9069 ListAllowed {respond} -> respond {peers: [], ok: False {}}

9012 RotateSecret {respond} -> respond {ok: False {}, error: "overloaded"}

9070 RotateSecret {respond} -> respond {ok: False {}, error: "overloaded"}

9013− GetConfig {respond} -> respond {ok: False {}, maxMessagesPerRecipient: 0, maxGossipHops: 0, pollTimeoutMs: 0}

9071+ GetConfig {respond} -> respond {ok: False {}, maxMessagesPerRecipient: 0, maxGossipHops: 0, pollTimeoutMs: 0, ackGracePeriod: 0}

9014 SetConfig {respond} -> respond {ok: False {}}

9072 SetConfig {respond} -> respond {ok: False {}}

9015 Metrics {respond} -> respond {metrics: "", ok: False {}}

9073 Metrics {respond} -> respond {metrics: "", ok: False {}}

9016 -- Exempt routes handled by isOverloadExempt; this branch is unreachable

9074 -- Exempt routes handled by isOverloadExempt; this branch is unreachable