@@ -979,6 +979,11 @@ maxPeerLatencyEntries = 100
979979*rateLimitedPerKey : Int
980980*rateLimitedFederation : Int
981981
982+-- Cancel metrics: track message cancellation volume for operator visibility.
983+-- Reset with federation latency window (same cadence as other windowed metrics).
984+*clientCancelCount : Int
985+*federationCancelCount : Int
986+
982987-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
983988-- Polls compare against this to skip expensive presence lookups when nothing changed.
984989*presenceVersion : Timestamp
@@ -1315,7 +1320,7 @@ route HealthApi where
13151320-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
13161321route StatsApi where
13171322 GET /stats headers {authorization: Text}
1318− -> {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}, 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}}
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}}
13191324 = Stats
13201325 GET /stats/peers headers {authorization: Text}
13211326 -> {peers: [PeerStatus], ok: Bool}
@@ -1411,24 +1416,30 @@ isJsonSafe = \s ->
14111416 else False {}
14121417
14131418-- Escape special characters for safe JSON string embedding.
1414−-- Handles: backslash, double quote, and control characters (\n, \r, \t, \b, \f).
1419+-- Handles: backslash, double quote, and control characters (\n, \r, \t).
14151420-- Non-printable characters outside this set are stripped (same allowlist as
14161421-- stripNonPrintable) to prevent log injection via error strings from federation peers.
14171422-- Fast path: strings containing only safe chars (hex, alphanumeric, common punctuation)
14181423-- are returned unchanged, avoiding O(n) per-character recursion for validated types.
1419−-- Processes one character at a time via tail recursion (TCO-safe in Knot).
1424+-- Uses an accumulator parameter for tail recursion (TCO-safe in Knot):
1425+-- the recursive call is in tail position, not wrapped by ++.
14201426escapeJsonString : Text -> Text
14211427escapeJsonString = \s ->
14221428 if s == "" then ""
14231429 else if isJsonSafe s then s
1430+ else escapeJsonStringAcc s ""
1431+
1432+escapeJsonStringAcc : Text -> Text -> Text
1433+escapeJsonStringAcc = \s acc ->
1434+ if s == "" then acc
14241435 else let c = take 1 s in
1425− if c == "\\" then "\\\\" ++ escapeJsonString (drop 1 s)
1426− else if c == "\"" then "\\\"" ++ escapeJsonString (drop 1 s)
1427− else if c == "\n" then "\\n" ++ escapeJsonString (drop 1 s)
1428− else if c == "\r" then "\\r" ++ escapeJsonString (drop 1 s)
1429− else if c == "\t" then "\\t" ++ escapeJsonString (drop 1 s)
1430− else if contains c printableAscii then c ++ escapeJsonString (drop 1 s)
1431− else escapeJsonString (drop 1 s)
1436+ if c == "\\" then escapeJsonStringAcc (drop 1 s) (acc ++ "\\\\")
1437+ else if c == "\"" then escapeJsonStringAcc (drop 1 s) (acc ++ "\\\"")
1438+ else if c == "\n" then escapeJsonStringAcc (drop 1 s) (acc ++ "\\n")
1439+ else if c == "\r" then escapeJsonStringAcc (drop 1 s) (acc ++ "\\r")
1440+ else if c == "\t" then escapeJsonStringAcc (drop 1 s) (acc ++ "\\t")
1441+ else if contains c printableAscii then escapeJsonStringAcc (drop 1 s) (acc ++ c)
1442+ else escapeJsonStringAcc (drop 1 s) acc
14321443
14331444any : [a] -> (a -> Bool) -> Bool
14341445any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
@@ -1548,13 +1559,17 @@ printableAscii = " !\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\
15481559
15491560-- Strip all non-printable characters from a string (allowlist approach).
15501561-- Keeps only characters in the printable ASCII range (0x20-0x7E).
1551−-- Tail-recursive: Knot compiles to a loop via TCO.
1562+-- Uses an accumulator parameter for tail recursion (TCO-safe in Knot):
1563+-- the recursive call is in tail position, not wrapped by ++.
15521564stripNonPrintable : Text -> Text
1553−stripNonPrintable = \s ->
1554− if s == "" then ""
1565+stripNonPrintable = \s -> stripNonPrintableAcc s ""
1566+
1567+stripNonPrintableAcc : Text -> Text -> Text
1568+stripNonPrintableAcc = \s acc ->
1569+ if s == "" then acc
15551570 else let c = take 1 s in
1556− if contains c printableAscii then c ++ stripNonPrintable (drop 1 s)
1557− else stripNonPrintable (drop 1 s)
1571+ if contains c printableAscii then stripNonPrintableAcc (drop 1 s) (acc ++ c)
1572+ else stripNonPrintableAcc (drop 1 s) acc
15581573
15591574isHexChar : Text -> Bool
15601575isHexChar = \c -> contains c "0123456789abcdef"
@@ -5318,6 +5333,9 @@ pruneAndRetry = \t -> do
53185333 set *clientAuthLatencyTotal = 0
53195334 set *clientAuthLatencyCount = 0
53205335 set *clientAuthLatencyMax = 0
5336+ -- Reset cancel metrics (windowed with other metrics)
5337+ set *clientCancelCount = 0
5338+ set *federationCancelCount = 0
53215339 set *federationLatencyResetAt = t)
53225340 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
53235341 -- These were queued by handleFederationAck instead of fork+sleep.
@@ -5329,16 +5347,18 @@ pruneAndRetry = \t -> do
53295347 yield ready
53305348 when (count readyCleanups > 0) (ackCleanupForwards readyCleanups t)
53315349 let result = {pendingAcks: ackState.pendingAcks, pendingOverflow: ackState.pendingOverflow, extendedActive: refreshResult.extendedActive, nearExpiryActive: refreshResult.nearExpiryActive, refreshPeers: refreshResult.refreshPeers, shouldRefresh: refreshResult.shouldRefresh}
5332− -- Fork federation I/O so hung peers don't block next prune cycle
5333− fork (do
5334− broadcastOfflineGossip t pr.peerServers pr.expiredLocal
5335− when federationEnabled (do
5336− retryPendingForwardsFromBuckets
5350+ -- Fork federation I/O phases independently so a failure in one doesn't block others.
5351+ -- Previously a single fork ran gossip -> forwards -> acks -> cancels sequentially,
5352+ -- meaning a panic in forward retries would skip ack and cancel retries entirely.
5353+ fork (broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
5354+ when federationEnabled (do
5355+ fork retryPendingForwardsFromBuckets
5356+ fork (do
53375357 -- Coalesce regular and overflow acks: merge into a single send pass so acks to
53385358 -- the same origin server share HTTP round-trips instead of two separate send cycles.
53395359 let coalescedAcks = union result.pendingAcks result.pendingOverflow
5340− sendAcksAndClean coalescedAcks
5341− retryPendingCancels))
5360+ sendAcksAndClean coalescedAcks)
5361+ fork retryPendingCancels)
53425362 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
53435363 -- trigger offline gossip when they expire (avoids wasted federation traffic)
53445364 when result.shouldRefresh (do
@@ -5426,7 +5446,12 @@ ackPolledMessages = \pubkey cursor t -> do
54265446 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
54275447 let ackedMsgIds = do m <- toAck; yield m.messageId
54285448 fwdBkt <- readFwdBucket b
5429− writeFwdBucket b (do f <- fwdBkt; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
5449+ let keptFwds = do f <- fwdBkt; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
5450+ let fwdRemoved = count fwdBkt - count keptFwds
5451+ writeFwdBucket b keptFwds
5452+ when (fwdRemoved > 0) (do
5453+ tfc <- *totalForwardCount
5454+ set *totalForwardCount = max 0 (tfc - fwdRemoved))
54305455 let ackedFederated = federatedAckEntries toAck t
54315456 ackedBkt <- readAckedBucket b
54325457 let combinedAckedIds = union ackedBkt ackedFederated
@@ -5670,6 +5695,8 @@ forwardsForServer = \forwards server -> do f <- forwards; where f.toServer == se
56705695-- Partition forwards by server and dispatch sequentially. Each iteration removes
56715696-- processed entries from the working list, so subsequent scans are smaller.
56725697-- Total work: ~n*k/2 vs n*k for the naive sendToEach + forwardsForServer pattern.
5698+-- Per-server jitter (0-2s) prevents thundering herd when multiple servers restart
5699+-- simultaneously and all retry pending forwards in the same prune window.
56735700sendForwardsPartitioned = \pending servers maxRetries ->
56745701 case headOf servers of
56755702 Nothing {} -> yield {}
@@ -5677,6 +5704,8 @@ sendForwardsPartitioned = \pending servers maxRetries ->
56775704 let rest = do s <- servers; where s != server; yield s
56785705 let serverForwards = do f <- pending; where f.toServer == server; yield f
56795706 let remaining = do f <- pending; where f.toServer != server; yield f
5707+ jitter <- randomInt 2000
5708+ sleep jitter
56805709 retryForwardsForServer serverForwards maxRetries
56815710 sendForwardsPartitioned remaining rest maxRetries
56825711
@@ -6398,10 +6427,15 @@ handleRevoke = \targetToken authorization clientIp respond -> do
63986427 let targetBucket = tokenBucket targetToken
63996428 result <- atomic do
64006429 sessions <- readSessionBucket targetBucket
6401− let hasTarget = any sessions (\s -> safeCompare s.token targetToken && s.pubkey == callerPubkey)
6402− if hasTarget
6430+ -- Single-pass find-and-remove: avoids calling safeCompare twice per session
6431+ -- (once in `any` to check existence, again in `where` to filter).
6432+ let filtered = fold (\acc s ->
6433+ if not acc.found && s.pubkey == callerPubkey && safeCompare s.token targetToken
6434+ then {kept: acc.kept, found: True {}}
6435+ else {kept: union acc.kept [s], found: acc.found}) {kept: [], found: False {}} sessions
6436+ if filtered.found
64036437 then do
6404− writeSessionBucket targetBucket (do s <- sessions; where not (safeCompare s.token targetToken && s.pubkey == callerPubkey); yield s)
6438+ writeSessionBucket targetBucket filtered.kept
64056439 sc <- *totalSessionCount
64066440 set *totalSessionCount = max 0 (sc - 1)
64076441 cleanup <- checkLastSessionCleanup callerPubkey t
@@ -6916,7 +6950,8 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
69166950 yield {cancelled: True {}}
69176951 when result.cancelled (do
69186952 logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")")
6919− queueCancelNotification toKey messageId t)
6953+ queueCancelNotification toKey messageId t
6954+ atomic do fcc <- *federationCancelCount; set *federationCancelCount = fcc + 1)
69206955 yield (respond {ok: True {}, error: "", cancelled: result.cancelled, retryAfterMs: 0}))
69216956
69226957-- POST /federation/cancel-batch: Batch cancel multiple previously forwarded messages.
@@ -6930,8 +6965,9 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
69306965 let capped = takeRelation maxBatchSize cancels
69316966 results <- cancelBatchInner capped fromServer t []
69326967 let cancelledCount = count (do r <- results; where r.cancelled; yield r)
6933− when (cancelledCount > 0)
6934− (logInfo ("federation/cancel-batch: cancelled " ++ show cancelledCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer))
6968+ when (cancelledCount > 0) (do
6969+ logInfo ("federation/cancel-batch: cancelled " ++ show cancelledCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer)
6970+ atomic do fcc <- *federationCancelCount; set *federationCancelCount = fcc + cancelledCount)
69356971 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
69366972
69376973-- Process cancel entries grouped by bucket: all entries sharing a pubkeyBucket are
@@ -7071,11 +7107,15 @@ forwardBatchByBucket = \entries fromServer t acc ->
70717107 delivBkt <- readDelivBucket b
70727108 nonceBkt <- readNonceBucket b
70737109 rc <- readRcBucket b
7074− r <- forwardBucketGroupInner sameBucket fromServer b t [] presenceBkt bucket ackedBkt delivBkt nonceBkt rc 0
7110+ rr <- *recvRates
7111+ fpr <- *federationPairRates
7112+ r <- forwardBucketGroupInner sameBucket fromServer b t [] presenceBkt bucket ackedBkt delivBkt nonceBkt rc 0 rr fpr
70757113 -- Write back modified bucket-local state once after processing all entries
70767114 writeMessageBucket b r.bucket
70777115 writeNonceBucket b r.nonceBkt
70787116 writeRcBucket b r.rc
7117+ set *recvRates = r.recvRates
7118+ set *federationPairRates = r.fedPairRates
70797119 -- Batch-adjust totalMessageCount once (not per-entry)
70807120 when (r.msgStored > 0) (do
70817121 tc <- *totalMessageCount
@@ -7085,47 +7125,44 @@ forwardBatchByBucket = \entries fromServer t acc ->
70857125
70867126-- Process all forward entries for a single bucket within one atomic block.
70877127-- Threads bucket-local state through recursion to avoid redundant per-entry reads.
7088−-- Shared state (recvRates, federationPairRates, seqCounter) is read per-entry since
7089−-- it's modified by each successful store and affects subsequent entries.
7090−forwardBucketGroupInner = \entries fromServer b t acc presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored ->
7128+-- recvRates and federationPairRates are threaded through to avoid re-reading shared
7129+-- refs per entry (was O(entries) reads, now O(1) read + O(1) write per bucket group).
7130+-- seqCounter is still read per-entry since it must monotonically increase.
7131+forwardBucketGroupInner = \entries fromServer b t acc presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored rr fpr ->
70917132 case headOf entries of
7092− Nothing {} -> yield {acc: acc, bucket: bucket, nonceBkt: nonceBkt, rc: rc, msgStored: msgStored}
7133+ Nothing {} -> yield {acc: acc, bucket: bucket, nonceBkt: nonceBkt, rc: rc, msgStored: msgStored, recvRates: rr, fedPairRates: fpr}
70937134 Just {value: entry} -> do
70947135 let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
70957136 let clampedTs = clampTimestamp entry.timestamp t
70967137 let hasPresence = any presenceBkt (\p -> p.pubkey == entry.toKey && p.expiresAt > t)
70977138 if not hasPresence
7098− then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
7139+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored rr fpr
70997140 else do
71007141 let isDup = checkMessageDup bucket entry.messageId entry.toKey || hasMessage entry.messageId entry.toKey ackedBkt || hasMessage entry.messageId entry.toKey delivBkt || any nonceBkt (\n -> n.messageId == entry.messageId && n.toKey == entry.toKey)
71017142 if isDup
7102− then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
7143+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored rr fpr
71037144 else do
71047145 -- Nonce eviction replay protection (per-bucket cutoff, matches handleForwardMessage)
71057146 nonceCutoff <- readNonceCutoff b
71067147 if nonceCutoff > 0 && entry.timestamp > 0 && entry.timestamp < nonceCutoff
7107− then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
7148+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored rr fpr
71087149 else do
71097150 totalMsgCount <- *totalMessageCount
71107151 let recipientCount = getRecipientCount entry.toKey rc
71117152 effMaxPerRecipient <- effectiveMaxMessagesPerRecipient
71127153 if totalMsgCount + msgStored >= maxMessages || recipientCount >= effMaxPerRecipient
7113− then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
7154+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored rr fpr
71147155 else do
71157156 -- Per-recipient receive rate limiting (matches handleForwardMessage)
7116− recvRates <- *recvRates
7117− let recvResult = checkPubkeyRate recvRates entry.toKey maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
7157+ let recvResult = checkPubkeyRate rr entry.toKey maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
71187158 if recvResult.limited
7119− then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
7159+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored rr fpr
71207160 else do
7121− set *recvRates = recvResult.entries
71227161 -- Per-server-per-recipient rate limiting (matches handleForwardMessage)
7123− federationPairRates <- *federationPairRates
7124− let fedPairResult = checkServerRecipientRate federationPairRates fromServer entry.toKey maxFederationPairRatePerPair federationPairRateWindowMs maxFederationPairRateEntries t
7162+ let fedPairResult = checkServerRecipientRate fpr fromServer entry.toKey maxFederationPairRatePerPair federationPairRateWindowMs maxFederationPairRateEntries t
71257163 if fedPairResult.limited
7126− then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored
7164+ then forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: False {}}]) presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored recvResult.entries fpr
71277165 else do
7128− set *federationPairRates = fedPairResult.entries
71297166 seqCounter <- *seqCounter
71307167 let seqTs = nextSeqTs t seqCounter
71317168 set *seqCounter = seqTs
@@ -7140,7 +7177,7 @@ forwardBucketGroupInner = \entries fromServer b t acc presenceBkt bucket ackedBk
71407177 let newNonceBkt = if count updatedNonces > maxNoncesPerBucket
71417178 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
71427179 else updatedNonces
7143− forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt newBucket ackedBkt delivBkt newNonceBkt newRc (msgStored + 1)
7180+ forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt newBucket ackedBkt delivBkt newNonceBkt newRc (msgStored + 1) recvResult.entries fedPairResult.entries
71447181
71457182-- POST /federation/delivery-receipt: Destination server notifies origin that federated
71467183-- messages were polled/acked by the recipient. The origin server records delivery receipts
@@ -7293,7 +7330,11 @@ clientCancelBatchInner = \senderKey entries t resultAcc fedAcc ->
72937330 fwdBkt <- readFwdBucket b
72947331 let cancelledFwds = do f <- fwdBkt; where f.messageId == entry.messageId && f.toKey == entry.to; yield f
72957332 let forwardTargetServers = collectPeerServers cancelledFwds (\f -> f.toServer)
7333+ let fwdRemoved = count cancelledFwds
72967334 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == entry.messageId && f.toKey == entry.to); yield f)
7335+ when (fwdRemoved > 0) (do
7336+ tfc <- *totalForwardCount
7337+ set *totalForwardCount = max 0 (tfc - fwdRemoved))
72977338 presenceBkt <- readPresenceBucket b
72987339 let remotePresence = collectPeerServers (presenceForKey presenceBkt entry.to t) (\p -> p.server)
72997340 let allCancelServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union [s] acc) remotePresence forwardTargetServers
@@ -7305,7 +7346,9 @@ clientCancelBatchInner = \senderKey entries t resultAcc fedAcc ->
73057346 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
73067347 else updatedNonces)
73077348 yield {cancelled: True {}, cancelServers: allCancelServers}
7308− when cancelResult.cancelled (queueCancelNotification entry.to entry.messageId t)
7349+ when cancelResult.cancelled (do
7350+ queueCancelNotification entry.to entry.messageId t
7351+ atomic do cc <- *clientCancelCount; set *clientCancelCount = cc + 1)
73097352 let newFedAcc = if cancelResult.cancelled && count cancelResult.cancelServers > 0
73107353 then union fedAcc [{messageId: entry.messageId, toKey: entry.to, servers: cancelResult.cancelServers}]
73117354 else fedAcc
@@ -7615,7 +7658,7 @@ handleHealth = \authorization respond -> do
76157658 draining: result.isDraining,
76167659 degradedPeers: result.degradedPeers})
76177660
7618−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}, 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+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}}
76197662
76207663handleStats = \authorization respond -> do
76217664 t <- now
@@ -7709,7 +7752,9 @@ handleStats = \authorization respond -> do
77097752 calt <- *clientAuthLatencyTotal
77107753 calc <- *clientAuthLatencyCount
77117754 calm <- *clientAuthLatencyMax
7712− 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}
7755+ ccc <- *clientCancelCount
7756+ fccn <- *federationCancelCount
7757+ 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}
77137758 -- Compute derived metrics
77147759 let sessions = core.sc
77157760 let messages = core.tmc
@@ -7736,7 +7781,7 @@ handleStats = \authorization respond -> do
77367781 yield (respond {ok: True {}, version: serverBuildVersion,
77377782 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},
77387783 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}},
7739− 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},
7784+ 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},
77407785 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}})
77417786 else yield (respond zeroStats)
77427787
@@ -7927,11 +7972,13 @@ handleMetrics = \authorization respond -> do
79277972 rlf <- *rateLimitedFederation
79287973 gss <- *gossipSendSuccess
79297974 gsf <- *gossipSendFail
7930− 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}
7975+ ccc <- *clientCancelCount
7976+ fccn <- *federationCancelCount
7977+ 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}
79317978 let fedAvg = if ext.flc > 0 then ext.flt / ext.flc else 0
79327979 let pollAvg = if ext.cplc > 0 then ext.cplt / ext.cplc else 0
79337980 let sendAvg = if ext.cslc > 0 then ext.cslt / ext.cslc else 0
7934− 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"
7981+ 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"
79357982 yield (respond {metrics: base ++ extended, ok: True {}})
79367983
79377984-- POST /messages/ack: Explicitly acknowledge received messages by ID.
@@ -8082,7 +8129,12 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
80828129 let deletedMsgIds = do m <- toDelete; yield m.messageId
80838130 let dfb = pubkeyBucket pubkey
80848131 fwdBkt <- readFwdBucket dfb
8085− writeFwdBucket dfb (do f <- fwdBkt; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
8132+ let keptDeleteFwds = do f <- fwdBkt; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
8133+ let fwdDeleteRemoved = count fwdBkt - count keptDeleteFwds
8134+ writeFwdBucket dfb keptDeleteFwds
8135+ when (fwdDeleteRemoved > 0) (do
8136+ tfc <- *totalForwardCount
8137+ set *totalForwardCount = max 0 (tfc - fwdDeleteRemoved))
80868138 -- Create ack entries for federated messages so origin servers learn about the delete
80878139 let federatedDeleted = do m <- toDelete; where m.originServer != ""; yield m
80888140 let ackedFederated = federatedAckEntries federatedDeleted t
@@ -8174,7 +8226,11 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
81748226 fwdBkt <- readFwdBucket b
81758227 let cancelledFwds = do f <- fwdBkt; where f.messageId == messageId && f.toKey == to; yield f
81768228 let forwardTargetServers = collectPeerServers cancelledFwds (\f -> f.toServer)
8229+ let fwdCancelRemoved = count cancelledFwds
81778230 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == to); yield f)
8231+ when (fwdCancelRemoved > 0) (do
8232+ tfc <- *totalForwardCount
8233+ set *totalForwardCount = max 0 (tfc - fwdCancelRemoved))
81788234 -- Collect servers from recipient's presence that may have received the forward
81798235 -- (completed forwards are no longer in fwdBucket, so check presence too)
81808236 presenceBkt <- readPresenceBucket b
@@ -8191,7 +8247,9 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
81918247 -- Log nonce eviction outside atomic block (println can't run inside STM)
81928248 logNonceEviction (pubkeyBucket to) cancelResult.nonceLruEvicted cancelResult.nonceOldestAge
81938249 -- Notify recipient that this message was retracted (delivered via next poll)
8194− when cancelResult.cancelled (queueCancelNotification to messageId t)
8250+ when cancelResult.cancelled (do
8251+ queueCancelNotification to messageId t
8252+ atomic do cc <- *clientCancelCount; set *clientCancelCount = cc + 1)
81958253 -- Send federation batch cancel to remote servers that may have the message.
81968254 -- Uses FederationBatchCancel (single-entry batch) instead of FederationCancel
81978255 -- for code path consistency with handleCancelBatch.
@@ -8506,7 +8564,12 @@ handleAdminPurgeMessages = \pubkey authorization respond -> do
85068564 let deletedMsgIds = do m <- toDelete; yield m.messageId
85078565 let pfb = pubkeyBucket pubkey
85088566 fwdBkt <- readFwdBucket pfb
8509− writeFwdBucket pfb (do f <- fwdBkt; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
8567+ let keptPurgeFwds = do f <- fwdBkt; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
8568+ let fwdPurgeRemoved = count fwdBkt - count keptPurgeFwds
8569+ writeFwdBucket pfb keptPurgeFwds
8570+ when (fwdPurgeRemoved > 0) (do
8571+ tfc <- *totalForwardCount
8572+ set *totalForwardCount = max 0 (tfc - fwdPurgeRemoved))
85108573 -- Create ack entries for federated messages so origin servers learn about the purge
85118574 let federatedPurged = do m <- toDelete; where m.originServer != ""; yield m
85128575 let ackedFederated = federatedAckEntries federatedPurged t
@@ -8978,11 +9041,11 @@ serve = \req -> do
89789041 peak <- *peakActiveRequests
89799042 when (newAr > peak) (set *peakActiveRequests = newAr)
89809043 serveInner req
9044+ endTime <- now
89819045 atomic do
89829046 ar <- *activeRequests
89839047 set *activeRequests = max 0 (ar - 1)
8984− set *lastRequestCompletedAt = startTime
8985− endTime <- now
9048+ set *lastRequestCompletedAt = endTime
89869049 let latency = endTime - startTime
89879050 -- Record client operation latency for /stats monitoring (windowed avg/max).
89889051 -- Only tracks poll and send — the two highest-volume client operations.