@@ -979,6 +979,11 @@ maxPeerLatencyEntries = 100
979*rateLimitedPerKey : Int
979*rateLimitedPerKey : Int
980*rateLimitedFederation : Int
980*rateLimitedFederation : Int
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
982-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
987-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
983-- Polls compare against this to skip expensive presence lookups when nothing changed.
988-- Polls compare against this to skip expensive presence lookups when nothing changed.
984*presenceVersion : Timestamp
989*presenceVersion : Timestamp
@@ -1315,7 +1320,7 @@ route HealthApi where
1315-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1320-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1316route StatsApi where
1321route StatsApi where
1317 GET /stats headers {authorization: Text}
1322 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}}
1320 GET /stats/peers headers {authorization: Text}
1325 GET /stats/peers headers {authorization: Text}
1321 -> {peers: [PeerStatus], ok: Bool}
1326 -> {peers: [PeerStatus], ok: Bool}
@@ -1411,24 +1416,30 @@ isJsonSafe = \s ->
1411 else False {}
1416 else False {}
1413-- Escape special characters for safe JSON string embedding.
1418-- 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).
1415-- Non-printable characters outside this set are stripped (same allowlist as
1420-- Non-printable characters outside this set are stripped (same allowlist as
1416-- stripNonPrintable) to prevent log injection via error strings from federation peers.
1421-- stripNonPrintable) to prevent log injection via error strings from federation peers.
1417-- Fast path: strings containing only safe chars (hex, alphanumeric, common punctuation)
1422-- Fast path: strings containing only safe chars (hex, alphanumeric, common punctuation)
1418-- are returned unchanged, avoiding O(n) per-character recursion for validated types.
1423-- 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 ++.
1420escapeJsonString : Text -> Text
1426escapeJsonString : Text -> Text
1421escapeJsonString = \s ->
1427escapeJsonString = \s ->
1422 if s == "" then ""
1428 if s == "" then ""
1423 else if isJsonSafe s then s
1429 else if isJsonSafe s then s
1430+ else escapeJsonStringAcc s ""
1432+escapeJsonStringAcc : Text -> Text -> Text
1433+escapeJsonStringAcc = \s acc ->
1434+ if s == "" then acc
1424 else let c = take 1 s in
1435 else let c = take 1 s in
1425− if c == "\\" then "\\\\" ++ escapeJsonString (drop 1 s)
1436+ if c == "\\" then escapeJsonStringAcc (drop 1 s) (acc ++ "\\\\")
1426− else if c == "\"" then "\\\"" ++ escapeJsonString (drop 1 s)
1437+ else if c == "\"" then escapeJsonStringAcc (drop 1 s) (acc ++ "\\\"")
1427− else if c == "\n" then "\\n" ++ escapeJsonString (drop 1 s)
1438+ else if c == "\n" then escapeJsonStringAcc (drop 1 s) (acc ++ "\\n")
1428− else if c == "\r" then "\\r" ++ escapeJsonString (drop 1 s)
1439+ else if c == "\r" then escapeJsonStringAcc (drop 1 s) (acc ++ "\\r")
1429− else if c == "\t" then "\\t" ++ escapeJsonString (drop 1 s)
1440+ else if c == "\t" then escapeJsonStringAcc (drop 1 s) (acc ++ "\\t")
1430− else if contains c printableAscii then c ++ escapeJsonString (drop 1 s)
1441+ else if contains c printableAscii then escapeJsonStringAcc (drop 1 s) (acc ++ c)
1431− else escapeJsonString (drop 1 s)
1442+ else escapeJsonStringAcc (drop 1 s) acc
1433any : [a] -> (a -> Bool) -> Bool
1444any : [a] -> (a -> Bool) -> Bool
1434any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
1445any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
@@ -1548,13 +1559,17 @@ printableAscii = " !\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\
1549-- Strip all non-printable characters from a string (allowlist approach).
1560-- Strip all non-printable characters from a string (allowlist approach).
1550-- Keeps only characters in the printable ASCII range (0x20-0x7E).
1561-- 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 ++.
1552stripNonPrintable : Text -> Text
1564stripNonPrintable : Text -> Text
1553−stripNonPrintable = \s ->
1565+stripNonPrintable = \s -> stripNonPrintableAcc s ""
1567+stripNonPrintableAcc : Text -> Text -> Text
1568+stripNonPrintableAcc = \s acc ->
1554− if s == "" then ""
1569+ if s == "" then acc
1555 else let c = take 1 s in
1570 else let c = take 1 s in
1556− if contains c printableAscii then c ++ stripNonPrintable (drop 1 s)
1571+ if contains c printableAscii then stripNonPrintableAcc (drop 1 s) (acc ++ c)
1557− else stripNonPrintable (drop 1 s)
1572+ else stripNonPrintableAcc (drop 1 s) acc
1559isHexChar : Text -> Bool
1574isHexChar : Text -> Bool
1560isHexChar = \c -> contains c "0123456789abcdef"
1575isHexChar = \c -> contains c "0123456789abcdef"
@@ -5318,6 +5333,9 @@ pruneAndRetry = \t -> do
5318 set *clientAuthLatencyTotal = 0
5333 set *clientAuthLatencyTotal = 0
5319 set *clientAuthLatencyCount = 0
5334 set *clientAuthLatencyCount = 0
5320 set *clientAuthLatencyMax = 0
5335 set *clientAuthLatencyMax = 0
5336+ -- Reset cancel metrics (windowed with other metrics)
5337+ set *clientCancelCount = 0
5338+ set *federationCancelCount = 0
5321 set *federationLatencyResetAt = t)
5339 set *federationLatencyResetAt = t)
5322 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
5340 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
5323 -- These were queued by handleFederationAck instead of fork+sleep.
5341 -- These were queued by handleFederationAck instead of fork+sleep.
@@ -5329,16 +5347,18 @@ pruneAndRetry = \t -> do
5329 yield ready
5347 yield ready
5330 when (count readyCleanups > 0) (ackCleanupForwards readyCleanups t)
5348 when (count readyCleanups > 0) (ackCleanupForwards readyCleanups t)
5331 let result = {pendingAcks: ackState.pendingAcks, pendingOverflow: ackState.pendingOverflow, extendedActive: refreshResult.extendedActive, nearExpiryActive: refreshResult.nearExpiryActive, refreshPeers: refreshResult.refreshPeers, shouldRefresh: refreshResult.shouldRefresh}
5349 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
5350+ -- Fork federation I/O phases independently so a failure in one doesn't block others.
5333− fork (do
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.
5334− broadcastOfflineGossip t pr.peerServers pr.expiredLocal
5353+ fork (broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
5335− when federationEnabled (do
5354+ when federationEnabled (do
5336− retryPendingForwardsFromBuckets
5355+ fork retryPendingForwardsFromBuckets
5337 -- Coalesce regular and overflow acks: merge into a single send pass so acks to
5357 -- Coalesce regular and overflow acks: merge into a single send pass so acks to
5338 -- the same origin server share HTTP round-trips instead of two separate send cycles.
5358 -- the same origin server share HTTP round-trips instead of two separate send cycles.
5339 let coalescedAcks = union result.pendingAcks result.pendingOverflow
5359 let coalescedAcks = union result.pendingAcks result.pendingOverflow
5340− sendAcksAndClean coalescedAcks
5360+ sendAcksAndClean coalescedAcks)
5341− retryPendingCancels))
5361+ fork retryPendingCancels)
5342 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
5362 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
5343 -- trigger offline gossip when they expire (avoids wasted federation traffic)
5363 -- trigger offline gossip when they expire (avoids wasted federation traffic)
5344 when result.shouldRefresh (do
5364 when result.shouldRefresh (do
@@ -5426,7 +5446,12 @@ ackPolledMessages = \pubkey cursor t -> do
5426 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
5446 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
5427 let ackedMsgIds = do m <- toAck; yield m.messageId
5447 let ackedMsgIds = do m <- toAck; yield m.messageId
5428 fwdBkt <- readFwdBucket b
5448 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))
5430 let ackedFederated = federatedAckEntries toAck t
5455 let ackedFederated = federatedAckEntries toAck t
5431 ackedBkt <- readAckedBucket b
5456 ackedBkt <- readAckedBucket b
5432 let combinedAckedIds = union ackedBkt ackedFederated
5457 let combinedAckedIds = union ackedBkt ackedFederated
@@ -5670,6 +5695,8 @@ forwardsForServer = \forwards server -> do f <- forwards; where f.toServer == se
5670-- Partition forwards by server and dispatch sequentially. Each iteration removes
5695-- Partition forwards by server and dispatch sequentially. Each iteration removes
5671-- processed entries from the working list, so subsequent scans are smaller.
5696-- processed entries from the working list, so subsequent scans are smaller.
5672-- Total work: ~n*k/2 vs n*k for the naive sendToEach + forwardsForServer pattern.
5697-- 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.
5673sendForwardsPartitioned = \pending servers maxRetries ->
5700sendForwardsPartitioned = \pending servers maxRetries ->
5674 case headOf servers of
5701 case headOf servers of
5675 Nothing {} -> yield {}
5702 Nothing {} -> yield {}
@@ -5677,6 +5704,8 @@ sendForwardsPartitioned = \pending servers maxRetries ->
5677 let rest = do s <- servers; where s != server; yield s
5704 let rest = do s <- servers; where s != server; yield s
5678 let serverForwards = do f <- pending; where f.toServer == server; yield f
5705 let serverForwards = do f <- pending; where f.toServer == server; yield f
5679 let remaining = do f <- pending; where f.toServer != server; yield f
5706 let remaining = do f <- pending; where f.toServer != server; yield f
5707+ jitter <- randomInt 2000
5680 retryForwardsForServer serverForwards maxRetries
5709 retryForwardsForServer serverForwards maxRetries
5681 sendForwardsPartitioned remaining rest maxRetries
5710 sendForwardsPartitioned remaining rest maxRetries
@@ -6398,10 +6427,15 @@ handleRevoke = \targetToken authorization clientIp respond -> do
6398 let targetBucket = tokenBucket targetToken
6427 let targetBucket = tokenBucket targetToken
6399 result <- atomic do
6428 result <- atomic do
6400 sessions <- readSessionBucket targetBucket
6429 sessions <- readSessionBucket targetBucket
6401− let hasTarget = any sessions (\s -> safeCompare s.token targetToken && s.pubkey == callerPubkey)
6430+ -- Single-pass find-and-remove: avoids calling safeCompare twice per session
6402− if hasTarget
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
6404− writeSessionBucket targetBucket (do s <- sessions; where not (safeCompare s.token targetToken && s.pubkey == callerPubkey); yield s)
6438+ writeSessionBucket targetBucket filtered.kept
6405 sc <- *totalSessionCount
6439 sc <- *totalSessionCount
6406 set *totalSessionCount = max 0 (sc - 1)
6440 set *totalSessionCount = max 0 (sc - 1)
6407 cleanup <- checkLastSessionCleanup callerPubkey t
6441 cleanup <- checkLastSessionCleanup callerPubkey t
@@ -6916,7 +6950,8 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
6916 yield {cancelled: True {}}
6950 yield {cancelled: True {}}
6917 when result.cancelled (do
6951 when result.cancelled (do
6918 logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")")
6952 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)
6920 yield (respond {ok: True {}, error: "", cancelled: result.cancelled, retryAfterMs: 0}))
6955 yield (respond {ok: True {}, error: "", cancelled: result.cancelled, retryAfterMs: 0}))
6922-- POST /federation/cancel-batch: Batch cancel multiple previously forwarded messages.
6957-- POST /federation/cancel-batch: Batch cancel multiple previously forwarded messages.
@@ -6930,8 +6965,9 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
6930 let capped = takeRelation maxBatchSize cancels
6965 let capped = takeRelation maxBatchSize cancels
6931 results <- cancelBatchInner capped fromServer t []
6966 results <- cancelBatchInner capped fromServer t []
6932 let cancelledCount = count (do r <- results; where r.cancelled; yield r)
6967 let cancelledCount = count (do r <- results; where r.cancelled; yield r)
6933− when (cancelledCount > 0)
6968+ when (cancelledCount > 0) (do
6934− (logInfo ("federation/cancel-batch: cancelled " ++ show cancelledCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer))
6969+ logInfo ("federation/cancel-batch: cancelled " ++ show cancelledCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer)
6970+ atomic do fcc <- *federationCancelCount; set *federationCancelCount = fcc + cancelledCount)
6935 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
6971 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
6937-- Process cancel entries grouped by bucket: all entries sharing a pubkeyBucket are
6973-- Process cancel entries grouped by bucket: all entries sharing a pubkeyBucket are
@@ -7071,11 +7107,15 @@ forwardBatchByBucket = \entries fromServer t acc ->
7071 delivBkt <- readDelivBucket b
7107 delivBkt <- readDelivBucket b
7072 nonceBkt <- readNonceBucket b
7108 nonceBkt <- readNonceBucket b
7073 rc <- readRcBucket b
7109 rc <- readRcBucket b
7111+ fpr <- *federationPairRates
7074− r <- forwardBucketGroupInner sameBucket fromServer b t [] presenceBkt bucket ackedBkt delivBkt nonceBkt rc 0
7112+ r <- forwardBucketGroupInner sameBucket fromServer b t [] presenceBkt bucket ackedBkt delivBkt nonceBkt rc 0 rr fpr
7075 -- Write back modified bucket-local state once after processing all entries
7113 -- Write back modified bucket-local state once after processing all entries
7076 writeMessageBucket b r.bucket
7114 writeMessageBucket b r.bucket
7077 writeNonceBucket b r.nonceBkt
7115 writeNonceBucket b r.nonceBkt
7078 writeRcBucket b r.rc
7116 writeRcBucket b r.rc
7117+ set *recvRates = r.recvRates
7118+ set *federationPairRates = r.fedPairRates
7079 -- Batch-adjust totalMessageCount once (not per-entry)
7119 -- Batch-adjust totalMessageCount once (not per-entry)
7080 when (r.msgStored > 0) (do
7120 when (r.msgStored > 0) (do
7081 tc <- *totalMessageCount
7121 tc <- *totalMessageCount
@@ -7085,47 +7125,44 @@ forwardBatchByBucket = \entries fromServer t acc ->
7086-- Process all forward entries for a single bucket within one atomic block.
7126-- Process all forward entries for a single bucket within one atomic block.
7087-- Threads bucket-local state through recursion to avoid redundant per-entry reads.
7127-- Threads bucket-local state through recursion to avoid redundant per-entry reads.
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).
7088−-- Shared state (recvRates, federationPairRates, seqCounter) is read per-entry since
7130+-- seqCounter is still read per-entry since it must monotonically increase.
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 ->
7131+forwardBucketGroupInner = \entries fromServer b t acc presenceBkt bucket ackedBkt delivBkt nonceBkt rc msgStored rr fpr ->
7091 case headOf entries of
7132 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}
7093 Just {value: entry} -> do
7134 Just {value: entry} -> do
7094 let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
7135 let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
7095 let clampedTs = clampTimestamp entry.timestamp t
7136 let clampedTs = clampTimestamp entry.timestamp t
7096 let hasPresence = any presenceBkt (\p -> p.pubkey == entry.toKey && p.expiresAt > t)
7137 let hasPresence = any presenceBkt (\p -> p.pubkey == entry.toKey && p.expiresAt > t)
7097 if not hasPresence
7138 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
7100 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)
7141 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)
7101 if isDup
7142 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
7104 -- Nonce eviction replay protection (per-bucket cutoff, matches handleForwardMessage)
7145 -- Nonce eviction replay protection (per-bucket cutoff, matches handleForwardMessage)
7105 nonceCutoff <- readNonceCutoff b
7146 nonceCutoff <- readNonceCutoff b
7106 if nonceCutoff > 0 && entry.timestamp > 0 && entry.timestamp < nonceCutoff
7147 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
7109 totalMsgCount <- *totalMessageCount
7150 totalMsgCount <- *totalMessageCount
7110 let recipientCount = getRecipientCount entry.toKey rc
7151 let recipientCount = getRecipientCount entry.toKey rc
7111 effMaxPerRecipient <- effectiveMaxMessagesPerRecipient
7152 effMaxPerRecipient <- effectiveMaxMessagesPerRecipient
7112 if totalMsgCount + msgStored >= maxMessages || recipientCount >= effMaxPerRecipient
7153 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
7115 -- Per-recipient receive rate limiting (matches handleForwardMessage)
7156 -- 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
7118 if recvResult.limited
7158 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
7121− set *recvRates = recvResult.entries
7122 -- Per-server-per-recipient rate limiting (matches handleForwardMessage)
7161 -- 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
7125 if fedPairResult.limited
7163 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
7128− set *federationPairRates = fedPairResult.entries
7129 seqCounter <- *seqCounter
7166 seqCounter <- *seqCounter
7130 let seqTs = nextSeqTs t seqCounter
7167 let seqTs = nextSeqTs t seqCounter
7131 set *seqCounter = seqTs
7168 set *seqCounter = seqTs
@@ -7140,7 +7177,7 @@ forwardBucketGroupInner = \entries fromServer b t acc presenceBkt bucket ackedBk
7140 let newNonceBkt = if count updatedNonces > maxNoncesPerBucket
7177 let newNonceBkt = if count updatedNonces > maxNoncesPerBucket
7141 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
7178 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
7142 else updatedNonces
7179 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
7145-- POST /federation/delivery-receipt: Destination server notifies origin that federated
7182-- POST /federation/delivery-receipt: Destination server notifies origin that federated
7146-- messages were polled/acked by the recipient. The origin server records delivery receipts
7183-- messages were polled/acked by the recipient. The origin server records delivery receipts
@@ -7293,7 +7330,11 @@ clientCancelBatchInner = \senderKey entries t resultAcc fedAcc ->
7293 fwdBkt <- readFwdBucket b
7330 fwdBkt <- readFwdBucket b
7294 let cancelledFwds = do f <- fwdBkt; where f.messageId == entry.messageId && f.toKey == entry.to; yield f
7331 let cancelledFwds = do f <- fwdBkt; where f.messageId == entry.messageId && f.toKey == entry.to; yield f
7295 let forwardTargetServers = collectPeerServers cancelledFwds (\f -> f.toServer)
7332 let forwardTargetServers = collectPeerServers cancelledFwds (\f -> f.toServer)
7333+ let fwdRemoved = count cancelledFwds
7296 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == entry.messageId && f.toKey == entry.to); yield f)
7334 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))
7297 presenceBkt <- readPresenceBucket b
7338 presenceBkt <- readPresenceBucket b
7298 let remotePresence = collectPeerServers (presenceForKey presenceBkt entry.to t) (\p -> p.server)
7339 let remotePresence = collectPeerServers (presenceForKey presenceBkt entry.to t) (\p -> p.server)
7299 let allCancelServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union [s] acc) remotePresence forwardTargetServers
7340 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 ->
7305 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
7346 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
7306 else updatedNonces)
7347 else updatedNonces)
7307 yield {cancelled: True {}, cancelServers: allCancelServers}
7348 yield {cancelled: True {}, cancelServers: allCancelServers}
7349+ when cancelResult.cancelled (do
7308− when cancelResult.cancelled (queueCancelNotification entry.to entry.messageId t)
7350+ queueCancelNotification entry.to entry.messageId t
7351+ atomic do cc <- *clientCancelCount; set *clientCancelCount = cc + 1)
7309 let newFedAcc = if cancelResult.cancelled && count cancelResult.cancelServers > 0
7352 let newFedAcc = if cancelResult.cancelled && count cancelResult.cancelServers > 0
7310 then union fedAcc [{messageId: entry.messageId, toKey: entry.to, servers: cancelResult.cancelServers}]
7353 then union fedAcc [{messageId: entry.messageId, toKey: entry.to, servers: cancelResult.cancelServers}]
7311 else fedAcc
7354 else fedAcc
@@ -7615,7 +7658,7 @@ handleHealth = \authorization respond -> do
7615 draining: result.isDraining,
7658 draining: result.isDraining,
7616 degradedPeers: result.degradedPeers})
7659 degradedPeers: result.degradedPeers})
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}}
7620handleStats = \authorization respond -> do
7663handleStats = \authorization respond -> do
7621 t <- now
7664 t <- now
@@ -7709,7 +7752,9 @@ handleStats = \authorization respond -> do
7709 calt <- *clientAuthLatencyTotal
7752 calt <- *clientAuthLatencyTotal
7710 calc <- *clientAuthLatencyCount
7753 calc <- *clientAuthLatencyCount
7711 calm <- *clientAuthLatencyMax
7754 calm <- *clientAuthLatencyMax
7755+ ccc <- *clientCancelCount
7756+ fccn <- *federationCancelCount
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}
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}
7713 -- Compute derived metrics
7758 -- Compute derived metrics
7714 let sessions = core.sc
7759 let sessions = core.sc
7715 let messages = core.tmc
7760 let messages = core.tmc
@@ -7736,7 +7781,7 @@ handleStats = \authorization respond -> do
7736 yield (respond {ok: True {}, version: serverBuildVersion,
7781 yield (respond {ok: True {}, version: serverBuildVersion,
7737 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},
7782 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},
7738 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 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},
7740 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}})
7785 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}})
7741 else yield (respond zeroStats)
7786 else yield (respond zeroStats)
@@ -7927,11 +7972,13 @@ handleMetrics = \authorization respond -> do
7927 rlf <- *rateLimitedFederation
7972 rlf <- *rateLimitedFederation
7928 gss <- *gossipSendSuccess
7973 gss <- *gossipSendSuccess
7929 gsf <- *gossipSendFail
7974 gsf <- *gossipSendFail
7975+ ccc <- *clientCancelCount
7976+ fccn <- *federationCancelCount
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}
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}
7931 let fedAvg = if ext.flc > 0 then ext.flt / ext.flc else 0
7978 let fedAvg = if ext.flc > 0 then ext.flt / ext.flc else 0
7932 let pollAvg = if ext.cplc > 0 then ext.cplt / ext.cplc else 0
7979 let pollAvg = if ext.cplc > 0 then ext.cplt / ext.cplc else 0
7933 let sendAvg = if ext.cslc > 0 then ext.cslt / ext.cslc else 0
7980 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"
7935 yield (respond {metrics: base ++ extended, ok: True {}})
7982 yield (respond {metrics: base ++ extended, ok: True {}})
7937-- POST /messages/ack: Explicitly acknowledge received messages by ID.
7984-- POST /messages/ack: Explicitly acknowledge received messages by ID.
@@ -8082,7 +8129,12 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
8082 let deletedMsgIds = do m <- toDelete; yield m.messageId
8129 let deletedMsgIds = do m <- toDelete; yield m.messageId
8083 let dfb = pubkeyBucket pubkey
8130 let dfb = pubkeyBucket pubkey
8084 fwdBkt <- readFwdBucket dfb
8131 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))
8086 -- Create ack entries for federated messages so origin servers learn about the delete
8138 -- Create ack entries for federated messages so origin servers learn about the delete
8087 let federatedDeleted = do m <- toDelete; where m.originServer != ""; yield m
8139 let federatedDeleted = do m <- toDelete; where m.originServer != ""; yield m
8088 let ackedFederated = federatedAckEntries federatedDeleted t
8140 let ackedFederated = federatedAckEntries federatedDeleted t
@@ -8174,7 +8226,11 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
8174 fwdBkt <- readFwdBucket b
8226 fwdBkt <- readFwdBucket b
8175 let cancelledFwds = do f <- fwdBkt; where f.messageId == messageId && f.toKey == to; yield f
8227 let cancelledFwds = do f <- fwdBkt; where f.messageId == messageId && f.toKey == to; yield f
8176 let forwardTargetServers = collectPeerServers cancelledFwds (\f -> f.toServer)
8228 let forwardTargetServers = collectPeerServers cancelledFwds (\f -> f.toServer)
8229+ let fwdCancelRemoved = count cancelledFwds
8177 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == to); yield f)
8230 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))
8178 -- Collect servers from recipient's presence that may have received the forward
8234 -- Collect servers from recipient's presence that may have received the forward
8179 -- (completed forwards are no longer in fwdBucket, so check presence too)
8235 -- (completed forwards are no longer in fwdBucket, so check presence too)
8180 presenceBkt <- readPresenceBucket b
8236 presenceBkt <- readPresenceBucket b
@@ -8191,7 +8247,9 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
8191 -- Log nonce eviction outside atomic block (println can't run inside STM)
8247 -- Log nonce eviction outside atomic block (println can't run inside STM)
8192 logNonceEviction (pubkeyBucket to) cancelResult.nonceLruEvicted cancelResult.nonceOldestAge
8248 logNonceEviction (pubkeyBucket to) cancelResult.nonceLruEvicted cancelResult.nonceOldestAge
8193 -- Notify recipient that this message was retracted (delivered via next poll)
8249 -- Notify recipient that this message was retracted (delivered via next poll)
8250+ when cancelResult.cancelled (do
8194− when cancelResult.cancelled (queueCancelNotification to messageId t)
8251+ queueCancelNotification to messageId t
8252+ atomic do cc <- *clientCancelCount; set *clientCancelCount = cc + 1)
8195 -- Send federation batch cancel to remote servers that may have the message.
8253 -- Send federation batch cancel to remote servers that may have the message.
8196 -- Uses FederationBatchCancel (single-entry batch) instead of FederationCancel
8254 -- Uses FederationBatchCancel (single-entry batch) instead of FederationCancel
8197 -- for code path consistency with handleCancelBatch.
8255 -- for code path consistency with handleCancelBatch.
@@ -8506,7 +8564,12 @@ handleAdminPurgeMessages = \pubkey authorization respond -> do
8506 let deletedMsgIds = do m <- toDelete; yield m.messageId
8564 let deletedMsgIds = do m <- toDelete; yield m.messageId
8507 let pfb = pubkeyBucket pubkey
8565 let pfb = pubkeyBucket pubkey
8508 fwdBkt <- readFwdBucket pfb
8566 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))
8510 -- Create ack entries for federated messages so origin servers learn about the purge
8573 -- Create ack entries for federated messages so origin servers learn about the purge
8511 let federatedPurged = do m <- toDelete; where m.originServer != ""; yield m
8574 let federatedPurged = do m <- toDelete; where m.originServer != ""; yield m
8512 let ackedFederated = federatedAckEntries federatedPurged t
8575 let ackedFederated = federatedAckEntries federatedPurged t
@@ -8978,11 +9041,11 @@ serve = \req -> do
8978 peak <- *peakActiveRequests
9041 peak <- *peakActiveRequests
8979 when (newAr > peak) (set *peakActiveRequests = newAr)
9042 when (newAr > peak) (set *peakActiveRequests = newAr)
8980 serveInner req
9043 serveInner req
8981 atomic do
9045 atomic do
8982 ar <- *activeRequests
9046 ar <- *activeRequests
8983 set *activeRequests = max 0 (ar - 1)
9047 set *activeRequests = max 0 (ar - 1)
8984− set *lastRequestCompletedAt = startTime
9048+ set *lastRequestCompletedAt = endTime
8986 let latency = endTime - startTime
9049 let latency = endTime - startTime
8987 -- Record client operation latency for /stats monitoring (windowed avg/max).
9050 -- Record client operation latency for /stats monitoring (windowed avg/max).
8988 -- Only tracks poll and send — the two highest-volume client operations.
9051 -- Only tracks poll and send — the two highest-volume client operations.