@@ -185,6 +185,10 @@ reconciliationFullCycleInterval = 50 -- full reconciliation every 50th prune cyc
185-- (sum of rcBucket totals vs totalMessageCount) exceeds this threshold. Small drifts from
185-- (sum of rcBucket totals vs totalMessageCount) exceeds this threshold. Small drifts from
186-- concurrent bucket pruning self-correct on the next cycle. 0 = always reconcile on drift.
186-- concurrent bucket pruning self-correct on the next cycle. 0 = always reconcile on drift.
187reconciliationDriftThreshold = 10
187reconciliationDriftThreshold = 10
188+-- Lightweight reconciliation interval: presence/session/forward counter reconciliation
189+-- runs every Nth prune cycle. Between cycles, cached counters are trusted. Message
190+-- reconciliation has its own interval (reconciliationFullCycleInterval).
191+lightReconciliationInterval = 5
189-- Federation latency metric window
193-- Federation latency metric window
190federationLatencyWindowMs : Int<Ms>
194federationLatencyWindowMs : Int<Ms>
@@ -937,6 +941,14 @@ maxPeerLatencyEntries = 100
937-- Used by health checks to avoid scanning all 16 forward buckets on every /health request.
941-- Used by health checks to avoid scanning all 16 forward buckets on every /health request.
938*cachedForwardStaleCount : Int
942*cachedForwardStaleCount : Int
944+-- Cached adaptive rate scale factor (0-100%). Updated every prune cycle.
945+-- Avoids 3 STM reads per rate check on the hot path. 100 = no scaling (default).
946+*cachedAdaptiveScale : Int
948+-- Gossip send metrics (windowed, reset with federation latency metrics)
949+*gossipSendSuccess : Int
950+*gossipSendFail : Int
940-- ============================================
952-- ============================================
942-- ============================================
954-- ============================================
@@ -1183,7 +1195,7 @@ route HealthApi where
1183-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1195-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1184route StatsApi where
1196route StatsApi where
1185 GET /stats headers {authorization: Text}
1197 GET /stats headers {authorization: Text}
1186− -> {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}, 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}, 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, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int}}
1198+ -> {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}, 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, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int}}
1188 GET /stats/peers headers {authorization: Text}
1200 GET /stats/peers headers {authorization: Text}
1189 -> {peers: [PeerStatus], ok: Bool}
1201 -> {peers: [PeerStatus], ok: Bool}
@@ -1671,16 +1683,31 @@ isBlockedPeer = \server blocked -> any blocked (\b -> b == server)
1671isAllowedByAllowlist : Text -> [Text] -> Bool
1683isAllowedByAllowlist : Text -> [Text] -> Bool
1672isAllowedByAllowlist = \server allowed -> count allowed == 0 || any allowed (\a -> a == server)
1684isAllowedByAllowlist = \server allowed -> count allowed == 0 || any allowed (\a -> a == server)
1686+-- Split text at the first colon. Returns {before, after}. Used to parse "Signed <sig>:<nonce>".
1687+-- Safe: signature hex and nonce are both [0-9a-f]+ (no colons), so exactly one colon separates them.
1688+splitOnColon : Text -> {before: Text, after: Text}
1689+splitOnColon = \s -> splitOnColonInner s 0
1691+splitOnColonInner : Text -> Int -> {before: Text, after: Text}
1692+splitOnColonInner = \s pos ->
1693+ if drop pos s == "" then {before: s, after: ""}
1694+ else if take 1 (drop pos s) == ":" then {before: take pos s, after: drop (pos + 1) s}
1695+ else splitOnColonInner s (pos + 1)
1674-- Generate a single-use token for outbound federation auth.
1697-- Generate a single-use token for outbound federation auth.
1675-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
1698-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
1676-- issued for one operation cannot authorize a different one.
1699-- issued for one operation cannot authorize a different one.
1700+-- When signing is enabled, includes a random nonce in the signed message to prevent
1701+-- replay attacks (two requests with the same timestamp+scope produce different signatures).
1702+-- Format: "Signed <sig>:<nonce>" — backward compatible (old format has no colon).
1677generateFederationAuth : Text -> Timestamp -> IO {network} Text
1703generateFederationAuth : Text -> Timestamp -> IO {network} Text
1678generateFederationAuth = \scope requestTimestamp ->
1704generateFederationAuth = \scope requestTimestamp ->
1679 if serverFederationPrivateKeyHex != ""
1705 if serverFederationPrivateKeyHex != ""
1707+ nonce <- genTokenN 2
1681− let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN)
1708+ let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN ++ "|" ++ nonce)
1682 let sig = sign (bytesFromHex serverFederationPrivateKeyHex) message
1709 let sig = sign (bytesFromHex serverFederationPrivateKeyHex) message
1683− yield ("Signed " ++ bytesToHex sig)
1710+ yield ("Signed " ++ bytesToHex sig ++ ":" ++ nonce)
1685 token <- genToken
1712 token <- genToken
1686 t <- now
1713 t <- now
@@ -2701,6 +2728,37 @@ pruneNonceBucketRange = \b endB t ->
2701 when (count pruned < count bucket) (writeNonceBucket b pruned)
2728 when (count pruned < count bucket) (writeNonceBucket b pruned)
2702 pruneNonceBucketRange (b + 1) endB t
2729 pruneNonceBucketRange (b + 1) endB t
2731+-- Record a forward nonce and handle capacity eviction with LRU logging.
2732+-- Usable inside or outside atomic blocks (reads/writes nonce bucket refs).
2733+-- Returns eviction info for deferred logging when called inside atomic.
2734+updateForwardNonce : Int -> MessageIdHex -> PubkeyHex -> Timestamp -> IO {lruEvicted: Int, oldestRetainedAge: Int}
2735+updateForwardNonce = \nb messageId toKey t -> do
2736+ fwdNonces <- readNonceBucket nb
2737+ let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, seenAt: t}]
2738+ let evictedNonces = if count updatedNonces > maxNoncesPerBucket
2739+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
2741+ let lruEvicted = count updatedNonces - count evictedNonces
2742+ let oldestRetainedAge = if lruEvicted > 0
2743+ then case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> t - n.seenAt
2745+ writeNonceBucket nb evictedNonces
2746+ yield {lruEvicted: lruEvicted, oldestRetainedAge: oldestRetainedAge}
2748+-- Log nonce LRU eviction if any occurred. Call outside atomic blocks.
2749+logNonceEviction : Int -> Int -> Int -> IO ()
2750+logNonceEviction = \nb lruEvicted oldestRetainedAge ->
2751+ when (lruEvicted > 0) (do
2752+ let level = if oldestRetainedAge < messageMaxAge then "error" else "warn"
2753+ logKV level "nonce_lru_eviction" ["bucket=" ++ show nb, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket, "oldestRetainedMs=" ++ show oldestRetainedAge])
2755+-- Full record-and-log helper for use outside atomic blocks.
2756+recordForwardNonce : MessageIdHex -> PubkeyHex -> Timestamp -> IO ()
2757+recordForwardNonce = \messageId toKey t -> do
2758+ let nb = pubkeyBucket toKey
2759+ result <- updateForwardNonce nb messageId toKey t
2760+ logNonceEviction nb result.lruEvicted result.oldestRetainedAge
2704-- ============================================
2762-- ============================================
2705-- Forward Bucket Helpers
2763-- Forward Bucket Helpers
2706-- ============================================
2764-- ============================================
@@ -2772,6 +2830,19 @@ countAllForwards = \b ->
2772 rest <- countAllForwards (b + 1)
2830 rest <- countAllForwards (b + 1)
2773 yield (count bucket + rest)
2831 yield (count bucket + rest)
2833+-- Compute min and max message counts across all 16 buckets (for stats bucket skew detection).
2834+-- Returns {bucketMin, bucketMax}. O(16) bucket reads.
2835+messageBucketMinMax : IO {bucketMin: Int, bucketMax: Int}
2836+messageBucketMinMax = messageBucketMinMaxInner 0 999999999 0
2838+messageBucketMinMaxInner : Int -> Int -> Int -> IO {bucketMin: Int, bucketMax: Int}
2839+messageBucketMinMaxInner = \b currentMin currentMax ->
2840+ if b > 15 then yield {bucketMin: if currentMin == 999999999 then 0 else currentMin, bucketMax: currentMax}
2842+ bucket <- readMessageBucket b
2843+ let c = count bucket
2844+ messageBucketMinMaxInner (b + 1) (min currentMin c) (max currentMax c)
2775-- Compute forward stats per-bucket without materializing all forwards into one list.
2846-- Compute forward stats per-bucket without materializing all forwards into one list.
2776-- Returns total count, oldest age, unique server count, and max per-server queue depth.
2847-- Returns total count, oldest age, unique server count, and max per-server queue depth.
2777computeForwardStats : Int -> Timestamp -> IO {totalCount: Int, oldestAge: Int, serverCount: Int, maxDepth: Int, serverCounts: [{server: Text, count: Int}]}
2848computeForwardStats : Int -> Timestamp -> IO {totalCount: Int, oldestAge: Int, serverCount: Int, maxDepth: Int, serverCounts: [{server: Text, count: Int}]}
@@ -3429,6 +3500,13 @@ hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey
3429-- inflate effective presence lifetime by up to clockSkewTolerance ms. Clamping to now removes
3500-- inflate effective presence lifetime by up to clockSkewTolerance ms. Clamping to now removes
3430-- this inflation: the peer's declared TTL is honored, but the expiry clock starts from now,
3501-- this inflation: the peer's declared TTL is honored, but the expiry clock starts from now,
3431-- not from a future timestamp.
3502-- not from a future timestamp.
3503+-- Filter seenBy entries to only known peers + ourselves. Unknown entries are dropped
3504+-- to prevent gossip suppression attacks (a malicious peer injecting arbitrary server
3505+-- names into seenBy to block propagation to those servers).
3506+filterSeenBy : [Text] -> [Text] -> [Text]
3507+filterSeenBy = \seenBy knownPeers ->
3508+ do s <- seenBy; where s == serverHostN || any knownPeers (\p -> p == s); yield s
3432validGossipEvents : [GossipEvent] -> Timestamp -> {online: [ValidGossipEntry], offline: [ValidGossipEntry]}
3510validGossipEvents : [GossipEvent] -> Timestamp -> {online: [ValidGossipEntry], offline: [ValidGossipEntry]}
3433validGossipEvents = \events t ->
3511validGossipEvents = \events t ->
3434 let validate = \e ->
3512 let validate = \e ->
@@ -3517,15 +3595,9 @@ pruneExpired = \t -> do
3517 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
3595 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
3518 -- Prune sessions and count in a single pass (avoids redundant 16-bucket traversal).
3596 -- Prune sessions and count in a single pass (avoids redundant 16-bucket traversal).
3519 -- pruneAndCountAllSessions applies the filter to each bucket and sums the survivors.
3597 -- pruneAndCountAllSessions applies the filter to each bucket and sums the survivors.
3520− oldSessionCount <- *totalSessionCount
3521 sessionCount <- pruneAndCountAllSessions (\bucket -> do x <- bucket; where x.expiresAt > t && t - x.createdAt < maxSessionLifetime; yield x)
3598 sessionCount <- pruneAndCountAllSessions (\bucket -> do x <- bucket; where x.expiresAt > t && t - x.createdAt < maxSessionLifetime; yield x)
3522 atomic do
3599 atomic do
3523 set *totalSessionCount = sessionCount
3600 set *totalSessionCount = sessionCount
3524− -- Bump presenceVersion when sessions are pruned: if a pubkey's last session expired,
3525− -- its presence will be pruned in a subsequent cycle. Bumping here ensures polls
3526− -- detect the transition sooner (within one prune interval) rather than waiting for
3527− -- the presence entry's separate TTL to expire.
3528− when (sessionCount < oldSessionCount) (set *presenceVersion = t)
3529 -- Prune messages across all 16 buckets concurrently. Each bucket prune is atomic
3601 -- Prune messages across all 16 buckets concurrently. Each bucket prune is atomic
3530 -- and touches only its own message bucket + rcBucket. The only shared write is
3602 -- and touches only its own message bucket + rcBucket. The only shared write is
3531 -- *totalMessageCount (a single Int), so STM contention is minimal.
3603 -- *totalMessageCount (a single Int), so STM contention is minimal.
@@ -3681,12 +3753,21 @@ forEachBatch = \items batchSize process ->
3681 forEachBatch split.rest batchSize process
3753 forEachBatch split.rest batchSize process
3683sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {}
3755sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {}
3684−sendBatchedGossipToServer = \server allEvents hops ->
3756+sendBatchedGossipToServer = \server allEvents hops -> do
3757+ -- Check backoff before sending: skip servers that recently failed (same backoff as forwards).
3758+ -- Without this, gossip to a dead peer generates unbounded log spam every refresh cycle.
3760+ failures <- *forwardFailures
3761+ inFlight <- *federationInFlight
3762+ blocked <- *blockedPeers
3763+ if not (shouldRetryServerEx server t0 failures inFlight blocked) then yield {}
3685 -- Filter out events this server has already seen (via seenBy tracking).
3765 -- Filter out events this server has already seen (via seenBy tracking).
3686 -- Add ourselves and the target server to seenBy so downstream relays skip both.
3766 -- Add ourselves and the target server to seenBy so downstream relays skip both.
3687− -- Cap seenBy at maxSeenByLen to bound per-event memory and bandwidth.
3767+ -- Cap seenBy at maxSeenByLen and filter to known peers (defense-in-depth against spoofed seenBy).
3768+ knownPeersForSeen <- *knownPeerServers
3688− let unseen = do e <- allEvents; where none e.seenBy (\s -> s == server); yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: takeRelation maxSeenByLen (union e.seenBy [server])}
3769+ let unseen = do e <- allEvents; where none e.seenBy (\s -> s == server); yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: filterSeenBy (takeRelation maxSeenByLen (union e.seenBy [server])) knownPeersForSeen}
3689− in if count unseen == 0 then yield {}
3770+ if count unseen == 0 then yield {}
3690 else forEachBatch unseen maxGossipEvents (\batch -> do
3771 else forEachBatch unseen maxGossipEvents (\batch -> do
3691 t <- now
3772 t <- now
3692 auth <- generateFederationAuth "gossip" t
3773 auth <- generateFederationAuth "gossip" t
@@ -3696,8 +3777,13 @@ sendBatchedGossipToServer = \server allEvents hops ->
3696 events: batch, hopsRemaining: hops, fromServer: serverHostN,
3777 events: batch, hopsRemaining: hops, fromServer: serverHostN,
3697 requestTimestamp: t, correlationId: corrId, protocolVersion: federationProtocolVersion})
3778 requestTimestamp: t, correlationId: corrId, protocolVersion: federationProtocolVersion})
3698 case result of
3779 case result of
3699− Err e -> logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
3781+ logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
3782+ recordFailure server
3783+ atomic do gs <- *gossipSendFail; set *gossipSendFail = gs + 1
3700− Ok _ -> yield {})
3784+ Ok _ -> do
3785+ clearFailure server
3786+ atomic do gs <- *gossipSendSuccess; set *gossipSendSuccess = gs + 1)
3702-- Sends gossip events to each peer server. Per-server seenBy filtering happens inside
3788-- Sends gossip events to each peer server. Per-server seenBy filtering happens inside
3703-- sendBatchedGossipToServer: if all events for a server are already in its seenBy list,
3789-- sendBatchedGossipToServer: if all events for a server are already in its seenBy list,
@@ -3764,14 +3850,14 @@ adaptiveRateScale = \msgCount sessCount presCount ->
3765-- Apply adaptive scaling to a base rate limit. Reads cached counters (O(1), no bucket scans)
3851-- Apply adaptive scaling to a base rate limit. Reads cached counters (O(1), no bucket scans)
3766-- and scales the rate proportionally. Returns at least 1 to prevent division-by-zero in callers.
3852-- and scales the rate proportionally. Returns at least 1 to prevent division-by-zero in callers.
3853+-- Apply adaptive scaling to a base rate limit. Reads cached scale factor (O(1), updated
3854+-- every prune cycle) instead of 3 counter reads per call. Returns at least 1 to prevent
3855+-- division-by-zero in callers.
3767adaptiveRate : Int -> IO Int
3856adaptiveRate : Int -> IO Int
3768adaptiveRate = \baseRate -> do
3857adaptiveRate = \baseRate -> do
3769− scale <- atomic do
3858+ scale <- *cachedAdaptiveScale
3770− tmc <- *totalMessageCount
3859+ let effectiveScale = if scale <= 0 then 100 else scale
3771− sc <- *totalSessionCount
3772− pc <- *totalPresenceCount
3773− yield (adaptiveRateScale tmc sc pc)
3774− yield (max 1 (baseRate * scale / 100))
3860+ yield (max 1 (baseRate * effectiveScale / 100))
3776-- Effective rate limit readers: return runtime override if set (> 0), else compile-time default.
3862-- Effective rate limit readers: return runtime override if set (> 0), else compile-time default.
3777-- All per-key and global rates are scaled by adaptiveRate under high utilization.
3863-- All per-key and global rates are scaled by adaptiveRate under high utilization.
@@ -3985,6 +4071,7 @@ fetchAndCachePeerKey = \server t -> do
3985-- Verify an inbound Ed25519 signed federation request.
4071-- Verify an inbound Ed25519 signed federation request.
3986-- Reconstructs the signed message from (scope, requestTimestamp, fromServer),
4072-- Reconstructs the signed message from (scope, requestTimestamp, fromServer),
3987-- fetches the peer's public key (cached), and verifies the signature.
4073-- fetches the peer's public key (cached), and verifies the signature.
4074+-- Verify signature without nonce (backward compatible with old peers).
3988verifyFederationSignature : Text -> Text -> Text -> Timestamp -> IO {network} Bool
4075verifyFederationSignature : Text -> Text -> Text -> Timestamp -> IO {network} Bool
3989verifyFederationSignature = \fromServer signatureHex scope requestTimestamp -> do
4076verifyFederationSignature = \fromServer signatureHex scope requestTimestamp -> do
3990 t <- now
4077 t <- now
@@ -3994,6 +4081,16 @@ verifyFederationSignature = \fromServer signatureHex scope requestTimestamp -> d
3994 let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer)
4081 let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer)
3995 yield (verify (bytesFromHex pubkeyHex) message (bytesFromHex signatureHex))
4082 yield (verify (bytesFromHex pubkeyHex) message (bytesFromHex signatureHex))
4084+-- Verify signature with nonce (new format: prevents replay of same-millisecond requests).
4085+verifyFederationSignatureWithNonce : Text -> Text -> Text -> Text -> Timestamp -> IO {network} Bool
4086+verifyFederationSignatureWithNonce = \fromServer signatureHex nonce scope requestTimestamp -> do
4088+ pubkeyHex <- fetchPeerPublicKey fromServer t
4089+ if pubkeyHex == "" then yield (False {})
4091+ let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer ++ "|" ++ nonce)
4092+ yield (verify (bytesFromHex pubkeyHex) message (bytesFromHex signatureHex))
3997-- Federation auth wrapper. Intentionally does NOT call checkPreAuth because:
4094-- Federation auth wrapper. Intentionally does NOT call checkPreAuth because:
3998-- (a) Soft drain (phase 1) blocks new client auth but must not block federation —
4095-- (a) Soft drain (phase 1) blocks new client auth but must not block federation —
3999-- peers need to deliver messages and acks to existing recipients during drain.
4096-- peers need to deliver messages and acks to existing recipients during drain.
@@ -4016,12 +4113,18 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
4016 yield (respond {ok: False {}, error: "not_allowed"})
4113 yield (respond {ok: False {}, error: "not_allowed"})
4018 -- Dispatch on auth scheme: "Signed " = Ed25519 signature, "Bearer " = verify callback
4115 -- Dispatch on auth scheme: "Signed " = Ed25519 signature, "Bearer " = verify callback
4116+ -- New format "Signed <sig>:<nonce>" includes a random nonce to prevent replay of
4117+ -- same-millisecond requests. Old format "Signed <sig>" (no colon) is backward compatible.
4019 verified <- if take 7 authorization == "Signed "
4118 verified <- if take 7 authorization == "Signed "
4021− let signatureHex = drop 7 authorization
4120+ let payload = drop 7 authorization
4022− if signatureHex == "" || tooLong signatureHex maxSignatureLen
4121+ if payload == "" || tooLong payload (maxSignatureLen + 13)
4023 then yield (False {})
4122 then yield (False {})
4124+ let parts = splitOnColon payload in
4125+ if parts.after != ""
4126+ then verifyFederationSignatureWithNonce fromServer parts.before parts.after scope requestTimestamp
4024− else verifyFederationSignature fromServer signatureHex scope requestTimestamp
4127+ else verifyFederationSignature fromServer payload scope requestTimestamp
4025 else verifyFederationCallback fromServer authorization scope
4128 else verifyFederationCallback fromServer authorization scope
4026 if not verified
4129 if not verified
4027 then yield (respond {ok: False {}, error: "unauthorized"})
4130 then yield (respond {ok: False {}, error: "unauthorized"})
@@ -4752,7 +4855,12 @@ waitForInFlightDrain = \deadline -> do
4752 if activeCount <= 0 then yield {}
4855 if activeCount <= 0 then yield {}
4753 else retry
4856 else retry
4754 timedOut <- *drainTimedOut
4857 timedOut <- *drainTimedOut
4755− when timedOut (logWarn "drain: timed out waiting for in-flight federation tasks")
4859+ logWarn "drain: timed out waiting for in-flight federation tasks"
4861+ inFlight <- *federationInFlight
4862+ let hungSlots = do r <- inFlight; where r.count > 0; yield r
4863+ when (count hungSlots > 0) (logStaleInFlight hungSlots t2))
4757-- Full maintenance cycle: prune expired state, retry pending forwards and acks,
4865-- Full maintenance cycle: prune expired state, retry pending forwards and acks,
4758-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
4866-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
@@ -4773,6 +4881,8 @@ pruneAndRetry = \t -> do
4773 sc <- *totalSessionCount
4881 sc <- *totalSessionCount
4774 pc <- *totalPresenceCount
4882 pc <- *totalPresenceCount
4775 fc <- *totalForwardCount
4883 fc <- *totalForwardCount
4884+ -- Update cached adaptive rate scale (avoids 3 reads per rate check on hot path)
4885+ set *cachedAdaptiveScale = adaptiveRateScale tmc sc pc
4776 yield {messages: tmc, sessions: sc, presence: pc, forwards: fc}
4886 yield {messages: tmc, sessions: sc, presence: pc, forwards: fc}
4777 when (capacityCounts.messages * 4 >= maxMessages * 3)
4887 when (capacityCounts.messages * 4 >= maxMessages * 3)
4778 (logEvent "warn" "capacity_warning" ("messages at " ++ show (capacityCounts.messages * 100 / maxMessages) ++ "% (" ++ show capacityCounts.messages ++ "/" ++ show maxMessages ++ ")"))
4888 (logEvent "warn" "capacity_warning" ("messages at " ++ show (capacityCounts.messages * 100 / maxMessages) ++ "% (" ++ show capacityCounts.messages ++ "/" ++ show maxMessages ++ ")"))
@@ -4782,32 +4892,33 @@ pruneAndRetry = \t -> do
4782 (logEvent "warn" "capacity_warning" ("presence at " ++ show (capacityCounts.presence * 100 / maxPresenceEntries) ++ "% (" ++ show capacityCounts.presence ++ "/" ++ show maxPresenceEntries ++ ")"))
4892 (logEvent "warn" "capacity_warning" ("presence at " ++ show (capacityCounts.presence * 100 / maxPresenceEntries) ++ "% (" ++ show capacityCounts.presence ++ "/" ++ show maxPresenceEntries ++ ")"))
4783 when (capacityCounts.forwards * 4 >= maxForwardsEntries * 3)
4893 when (capacityCounts.forwards * 4 >= maxForwardsEntries * 3)
4784 (logEvent "warn" "capacity_warning" ("forwards at " ++ show (capacityCounts.forwards * 100 / maxForwardsEntries) ++ "% (" ++ show capacityCounts.forwards ++ "/" ++ show maxForwardsEntries ++ ")"))
4894 (logEvent "warn" "capacity_warning" ("forwards at " ++ show (capacityCounts.forwards * 100 / maxForwardsEntries) ++ "% (" ++ show capacityCounts.forwards ++ "/" ++ show maxForwardsEntries ++ ")"))
4785− -- Counter reconciliation: reconcile all counters every prune cycle.
4895+ -- Counter reconciliation: presence/session/forward counters reconciled every Nth cycle
4786− -- Presence and session counters are cheap (just counting list lengths across buckets).
4896+ -- (lightReconciliationInterval). Between cycles, cached counters are trusted — drift
4787− -- Message reconciliation is heavier (reads all messages, rebuilds per-recipient counts)
4897+ -- from concurrent operations is typically small and self-corrects.
4788− -- but runs every cycle to minimize the drift window from concurrent bucket pruning.
4898+ pc <- *pruneCount
4899+ let isLightReconcCycle = pc > 0 && (pc - (pc / lightReconciliationInterval) * lightReconciliationInterval) == 0
4900+ when isLightReconcCycle (do
4789− presDrift <- atomic do
4901+ presDrift <- atomic do
4790− actual <- countAllPresence
4902+ actual <- countAllPresence
4791− stored <- *totalPresenceCount
4903+ stored <- *totalPresenceCount
4792− set *totalPresenceCount = actual
4904+ set *totalPresenceCount = actual
4793− yield (actual - stored)
4905+ yield (actual - stored)
4794− sessDrift <- atomic do
4906+ sessDrift <- atomic do
4795− actual <- countAllSessions
4907+ actual <- countAllSessions
4796− stored <- *totalSessionCount
4908+ stored <- *totalSessionCount
4797− set *totalSessionCount = actual
4909+ set *totalSessionCount = actual
4798− yield (actual - stored)
4910+ yield (actual - stored)
4799− fwdDrift <- atomic do
4911+ fwdDrift <- atomic do
4800− actual <- countAllForwards 0
4912+ actual <- countAllForwards 0
4801− stored <- *totalForwardCount
4913+ stored <- *totalForwardCount
4802− set *totalForwardCount = actual
4914+ set *totalForwardCount = actual
4803− yield (actual - stored)
4915+ yield (actual - stored)
4804− when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0)
4916+ when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0)
4805− (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift))
4917+ (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift)))
4806 -- Message reconciliation: cheap check first (sum rcBucket .count fields vs totalMessageCount).
4918 -- Message reconciliation: cheap check first (sum rcBucket .count fields vs totalMessageCount).
4807 -- Full scan only when drift exceeds threshold or every Nth cycle.
4919 -- Full scan only when drift exceeds threshold or every Nth cycle.
4808 -- Small drifts from concurrent bucket pruning self-correct; only trigger the expensive
4920 -- Small drifts from concurrent bucket pruning self-correct; only trigger the expensive
4809 -- full scan when drift is significant (> reconciliationDriftThreshold) or periodic.
4921 -- full scan when drift is significant (> reconciliationDriftThreshold) or periodic.
4811 let isFullCycle = pc > 0 && (pc - (pc / reconciliationFullCycleInterval) * reconciliationFullCycleInterval) == 0
4922 let isFullCycle = pc > 0 && (pc - (pc / reconciliationFullCycleInterval) * reconciliationFullCycleInterval) == 0
4812 cheapMsgDrift <- atomic do
4923 cheapMsgDrift <- atomic do
4813 rcTotal <- sumRcBucketTotals 0
4924 rcTotal <- sumRcBucketTotals 0
@@ -4885,6 +4996,9 @@ pruneAndRetry = \t -> do
4885 set *verificationCache = do c <- cache; where t - c.verifiedAt < verificationCacheMaxMs; yield c
4996 set *verificationCache = do c <- cache; where t - c.verifiedAt < verificationCacheMaxMs; yield c
4886 -- Reset per-peer latency tracking with global window
4997 -- Reset per-peer latency tracking with global window
4887 set *peerLatencies = []
4998 set *peerLatencies = []
4999+ -- Reset gossip send metrics (windowed with federation metrics)
5000+ set *gossipSendSuccess = 0
5001+ set *gossipSendFail = 0
4888 -- Reset client operation latency metrics (windowed with federation metrics)
5002 -- Reset client operation latency metrics (windowed with federation metrics)
4889 set *clientPollLatencyTotal = 0
5003 set *clientPollLatencyTotal = 0
4890 set *clientPollLatencyCount = 0
5004 set *clientPollLatencyCount = 0
@@ -5050,7 +5164,6 @@ pollNewEvents = \pubkey clampedCursor t -> do
5050 let selectedLow = takeRelation reservedSlots sortedLow
5164 let selectedLow = takeRelation reservedSlots sortedLow
5051 -- Merge and re-sort to maintain correct cursor advancement (lastReceivedAt must
5165 -- Merge and re-sort to maintain correct cursor advancement (lastReceivedAt must
5052 -- reflect the actual latest receivedAt across both pools for cursor consistency).
5166 -- reflect the actual latest receivedAt across both pools for cursor consistency).
5053− let allPollable = sortBy (\m -> (0 - m.priority) * 10000000000000 + m.receivedAt) raw
5054 let limitedMsgs = sortBy (\m -> (0 - m.priority) * 10000000000000 + m.receivedAt) (union selectedHigh selectedLow)
5167 let limitedMsgs = sortBy (\m -> (0 - m.priority) * 10000000000000 + m.receivedAt) (union selectedHigh selectedLow)
5055 let evts = do
5168 let evts = do
5056 m <- limitedMsgs
5169 m <- limitedMsgs
@@ -5627,25 +5740,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
5627 -- if the message expires from messages/ackedIds/deliveredIds and a federation
5740 -- if the message expires from messages/ackedIds/deliveredIds and a federation
5628 -- peer forwards it back. Without this, the nonce check in handleForwardMessage
5741 -- peer forwards it back. Without this, the nonce check in handleForwardMessage
5629 -- would miss locally-originated messages entirely.
5742 -- would miss locally-originated messages entirely.
5630− -- Sharded: writes only to the bucket for this recipient's pubkey.
5743+ recordForwardNonce effectiveMsgId to t
5631− let nb = pubkeyBucket to
5632− fwdNonces <- readNonceBucket nb
5633− let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, seenAt: t}]
5634− let evictedNonces = if count updatedNonces > maxNoncesPerBucket
5635− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5637− -- Log LRU nonce eviction: evicted nonces create a bounded replay window
5638− -- where messages with those nonces could be re-accepted until natural expiry.
5639− -- oldestRetainedAge: age of the oldest surviving nonce (new dedup coverage floor).
5640− -- If > messageMaxAge, evicted nonces were in the extended-protection zone only
5641− -- (message/acked/delivered dedup still covers the gap). If <= messageMaxAge,
5642− -- some evicted nonces were still within primary dedup coverage (lower risk).
5643− let lruEvicted = count updatedNonces - count evictedNonces
5644− when (lruEvicted > 0) (do
5645− let oldestRetainedAge = case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> t - n.seenAt
5646− let level = if oldestRetainedAge < messageMaxAge then "error" else "warn"
5647− logKV level "nonce_lru_eviction" ["bucket=" ++ show nb, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket, "oldestRetainedMs=" ++ show oldestRetainedAge])
5648− writeNonceBucket nb evictedNonces
5649 let toPresence = presenceForKey recipientPresence to t
5744 let toPresence = presenceForKey recipientPresence to t
5650 let localOnline = hasLocalPresence toPresence
5745 let localOnline = hasLocalPresence toPresence
5651 let remoteServers = collectPeerServers toPresence (\p -> p.server)
5746 let remoteServers = collectPeerServers toPresence (\p -> p.server)
@@ -6044,16 +6139,13 @@ handleVerifyToken = \token scope clientIp respond -> do
6045 found <- atomic do
6140 found <- atomic do
6046 pending <- *pendingVerifications
6141 pending <- *pendingVerifications
6047− -- Pre-filter by scope, expiry, and first hex char before running expensive safeCompare.
6142+ -- Filter by scope and expiry only — no token prefix optimization.
6048− -- Reduces scan from O(n) safeCompare calls to O(n/16) by filtering on token prefix.
6143+ -- The previous first-char pre-filter leaked one hex digit via timing side-channel.
6049− -- The prefix filter is not constant-time, but single-use tokens with 24 bytes of randomness
6144+ -- safeCompare must run on all candidates to prevent information leakage.
6050− -- make first-char leakage irrelevant (16^47 remaining possibilities in 30s window).
6145+ -- Bounded by maxPendingVerifications and further filtered by scope+expiry.
6051− let firstChar = take 1 token
6052− let isCandidate = \p -> p.scope == scope && t - p.createdAt < verificationTokenExpiry && take 1 p.token == firstChar
6146+ let isCandidate = \p -> p.scope == scope && t - p.createdAt < verificationTokenExpiry
6053 let candidates = do p <- pending; where isCandidate p; yield p
6147 let candidates = do p <- pending; where isCandidate p; yield p
6054 let match = any candidates (\p -> safeCompare p.token token)
6148 let match = any candidates (\p -> safeCompare p.token token)
6055− -- Optimized removal: run safeCompare only on candidates (O(n/16)) instead of
6056− -- all pending (O(n)). Non-candidates are kept without constant-time comparison.
6057 when match (set *pendingVerifications = union
6149 when match (set *pendingVerifications = union
6058 (do p <- pending; where not (isCandidate p); yield p)
6150 (do p <- pending; where not (isCandidate p); yield p)
6059 (do p <- candidates; where not (safeCompare p.token token); yield p))
6151 (do p <- candidates; where not (safeCompare p.token token); yield p))
@@ -6073,8 +6165,9 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
6073 let cappedHops = clamp 0 maxGossipHops hopsRemaining
6165 let cappedHops = clamp 0 maxGossipHops hopsRemaining
6074 let corrId = sanitizeCorrelationId correlationId
6166 let corrId = sanitizeCorrelationId correlationId
6075 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " gossip from " ++ fromServer))
6167 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " gossip from " ++ fromServer))
6076− -- Cap inbound seenBy lists to prevent memory waste from malicious peers
6168+ -- Cap inbound seenBy lists and filter to known peers to prevent gossip suppression attacks
6169+ knownPeersForSeenBy <- *knownPeerServers
6077− let cappedEvents = do e <- events; yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: takeRelation maxSeenByLen e.seenBy}
6170+ let cappedEvents = do e <- events; yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: filterSeenBy (takeRelation maxSeenByLen e.seenBy) knownPeersForSeenBy}
6078 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count cappedEvents <= maxGossipEvents) (True {}) (do
6171 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count cappedEvents <= maxGossipEvents) (True {}) (do
6079 -- Per-peer gossip event volume limiting: counts total events (not just requests) per server.
6172 -- Per-peer gossip event volume limiting: counts total events (not just requests) per server.
6080 -- A single gossip request carries up to maxGossipEvents entries; without this, a peer
6173 -- A single gossip request carries up to maxGossipEvents entries; without this, a peer
@@ -6287,18 +6380,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
6287 writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
6380 writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
6288 signalMessageForKey toKey seqTs
6381 signalMessageForKey toKey seqTs
6289 -- Record forward nonce for replay protection beyond message/ack dedup lifetime.
6382 -- Record forward nonce for replay protection beyond message/ack dedup lifetime.
6290− -- Sharded: writes only to the bucket for this recipient's pubkey.
6383+ recordForwardNonce messageId toKey t
6291− fwdNonces <- readNonceBucket bkt
6292− let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, seenAt: t}]
6293− let evictedNonces = if count updatedNonces > maxNoncesPerBucket
6294− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6296− let lruEvicted = count updatedNonces - count evictedNonces
6297− when (lruEvicted > 0) (do
6298− let oldestRetainedAge = case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> t - n.seenAt
6299− let level = if oldestRetainedAge < messageMaxAge then "error" else "warn"
6300− logKV level "nonce_lru_eviction" ["bucket=" ++ show bkt, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket, "oldestRetainedMs=" ++ show oldestRetainedAge])
6301− writeNonceBucket bkt evictedNonces
6302 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
6384 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
6303 if result.noPresence then do
6385 if result.noPresence then do
6304 logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
6386 logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
@@ -7067,7 +7149,7 @@ handleHealth = \authorization respond -> do
7067 draining: result.isDraining,
7149 draining: result.isDraining,
7068 degradedPeers: 0})
7150 degradedPeers: 0})
7070−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}, 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}, 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, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0}}
7152+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}, 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, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0}}
7072handleStats = \authorization respond -> do
7154handleStats = \authorization respond -> do
7073 t <- now
7155 t <- now
@@ -7082,6 +7164,7 @@ handleStats = \authorization respond -> do
7082 -- Forward stats computed per-bucket to avoid materializing all 16 buckets into one list.
7164 -- Forward stats computed per-bucket to avoid materializing all 16 buckets into one list.
7083 -- Each bucket is read independently; counts and per-server stats are aggregated.
7165 -- Each bucket is read independently; counts and per-server stats are aggregated.
7084 fwdStats <- computeForwardStats 0 t
7166 fwdStats <- computeForwardStats 0 t
7167+ bucketStats <- messageBucketMinMax
7085 core <- atomic do
7168 core <- atomic do
7086 sc <- *totalSessionCount
7169 sc <- *totalSessionCount
7087 pc <- *totalPresenceCount
7170 pc <- *totalPresenceCount
@@ -7143,7 +7226,9 @@ handleStats = \authorization respond -> do
7143 cslt <- *clientSendLatencyTotal
7226 cslt <- *clientSendLatencyTotal
7144 cslc <- *clientSendLatencyCount
7227 cslc <- *clientSendLatencyCount
7145 cslm <- *clientSendLatencyMax
7228 cslm <- *clientSendLatencyMax
7229+ gss <- *gossipSendSuccess
7230+ gsf <- *gossipSendFail
7146− 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, 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}
7231+ 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, 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}
7147 -- Compute derived metrics
7232 -- Compute derived metrics
7148 let sessions = core.sc
7233 let sessions = core.sc
7149 let messages = core.tmc
7234 let messages = core.tmc
@@ -7168,8 +7253,8 @@ handleStats = \authorization respond -> do
7168 let pollAvg = if counters.cplc > 0 then counters.cplt / counters.cplc else 0
7253 let pollAvg = if counters.cplc > 0 then counters.cplt / counters.cplc else 0
7169 let sendAvg = if counters.cslc > 0 then counters.cslt / counters.cslc else 0
7254 let sendAvg = if counters.cslc > 0 then counters.cslt / counters.cslc else 0
7170 yield (respond {ok: True {}, version: serverBuildVersion,
7255 yield (respond {ok: True {}, version: serverBuildVersion,
7171− 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},
7256+ 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},
7172− 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},
7257+ 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},
7173 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},
7258 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},
7174 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, sendAvgMs: sendAvg, sendMaxMs: counters.cslm, sendCount: counters.cslc}})
7259 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, sendAvgMs: sendAvg, sendMaxMs: counters.cslm, sendCount: counters.cslc}})
7175 else yield (respond zeroStats)
7260 else yield (respond zeroStats)
@@ -7539,13 +7624,13 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
7539 bucket <- readMessageBucket b
7624 bucket <- readMessageBucket b
7540 let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == to; yield m)
7625 let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == to; yield m)
7541 case toCancel of
7626 case toCancel of
7542− Nothing {} -> yield {cancelled: False {}, cancelServers: []}
7627+ Nothing {} -> yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}
7543 Just {value: msg} ->
7628 Just {value: msg} ->
7544 -- Only allow cancelling locally-originated messages (originServer == "")
7629 -- Only allow cancelling locally-originated messages (originServer == "")
7545 -- where the caller is the actual sender. Federation forwards have no
7630 -- where the caller is the actual sender. Federation forwards have no
7546 -- senderKey and should not be cancellable via this endpoint.
7631 -- senderKey and should not be cancellable via this endpoint.
7547− if msg.originServer != "" then yield {cancelled: False {}, cancelServers: []}
7632+ if msg.originServer != "" then yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}
7548− else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, cancelServers: []}
7633+ else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}
7550 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)
7635 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)
7551 tc <- *totalMessageCount
7636 tc <- *totalMessageCount
@@ -7568,12 +7653,10 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
7568 -- Record cancel nonce: prevents replayed federation forwards of the
7653 -- Record cancel nonce: prevents replayed federation forwards of the
7569 -- cancelled message from being accepted after the message is removed
7654 -- cancelled message from being accepted after the message is removed
7570 -- from all dedup sets (messages, ackedIds, deliveredIds).
7655 -- from all dedup sets (messages, ackedIds, deliveredIds).
7571− nonceBkt <- readNonceBucket b
7656+ nonceResult <- updateForwardNonce b messageId to t
7572− let updatedNonces = union nonceBkt [{messageId: messageId, toKey: to, seenAt: t}]
7573− writeNonceBucket b (if count updatedNonces > maxNoncesPerBucket
7574− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
7575− else updatedNonces)
7576− yield {cancelled: True {}, cancelServers: allCancelServers}
7657+ yield {cancelled: True {}, cancelServers: allCancelServers, nonceLruEvicted: nonceResult.lruEvicted, nonceOldestAge: nonceResult.oldestRetainedAge}
7658+ -- Log nonce eviction outside atomic block (println can't run inside STM)
7659+ logNonceEviction (pubkeyBucket to) cancelResult.nonceLruEvicted cancelResult.nonceOldestAge
7577 -- Send federation batch cancel to remote servers that may have the message.
7660 -- Send federation batch cancel to remote servers that may have the message.
7578 -- Uses FederationBatchCancel (single-entry batch) instead of FederationCancel
7661 -- Uses FederationBatchCancel (single-entry batch) instead of FederationCancel
7579 -- for code path consistency with handleCancelBatch.
7662 -- for code path consistency with handleCancelBatch.