@@ -137,7 +137,7 @@ sessionExpiry = 3600000 -- 1 hour
137137maxSessionLifetime : Int<Ms>
138138maxSessionLifetime = 86400000 -- 24 hours; absolute session lifetime (forces re-auth even if sliding window keeps refreshing)
139139onlineGossipTtl : Int<S>
140−onlineGossipTtl = 3600 -- 1 hour; TTL announced when a key authenticates
140+onlineGossipTtl = 5400 -- 1.5 hours; 1.5x sessionExpiry to prevent presence expiring before session refresh gossips
141141offlineGossipMinTtl : Int<S>
142142offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events
143143maxFederationConcurrency = 10 -- max concurrent outbound federation requests per cycle
@@ -206,13 +206,13 @@ federationLatencyT3 = 2000 -- slow (degraded or distant peer)
206206
207207-- Server build version: included in /health and /stats responses for deployment verification.
208208-- Update this string when deploying a new build so operators can confirm which version is running.
209−serverBuildVersion = "2026-04-26f"
209+serverBuildVersion = "2026-04-26g"
210210
211211-- Federation protocol version: included in gossip/forward/ack requests.
212212-- Peers log warnings when they receive requests from older versions.
213213-- Backward compatible: older peers ignore unknown fields (Knot's JSON parser
214214-- drops unknown keys). 0 = pre-versioning peer (field missing or unset).
215−federationProtocolVersion = 14
215+federationProtocolVersion = 15
216216
217217-- Max seenBy list length in gossip events (bounds memory and bandwidth per event)
218218maxSeenByLen = 20
@@ -753,6 +753,10 @@ type CancelNotification = {toKey: PubkeyHex, messageId: MessageIdHex, cancelledA
753753
754754*pruneGeneration : Int
755755
756+-- Tracks the duration of the last completed prune cycle (ms) for dynamic stuck detection.
757+-- Updated after each successful prune. Initial 0 = use fixed 5x fallback.
758+*lastPruneDuration : Int<Ms>
759+
756760*ipRates : [IpRateEntry]
757761
758762*lookupRates : [LookupRateEntry]
@@ -968,6 +972,30 @@ maxPeerLatencyEntries = 100
968972*clientSendLatencyCount : Int
969973*clientSendLatencyMax : Int<Ms>
970974
975+-- Client latency histogram buckets (poll and send): [0,50ms), [50,200ms), [200,1000ms), [1000+ms)
976+-- Approximates P50/P95/P99 for client-facing operations. Reset with federation latency window.
977+*clientPollLatencyBucket0 : Int
978+*clientPollLatencyBucket1 : Int
979+*clientPollLatencyBucket2 : Int
980+*clientPollLatencyBucket3 : Int
981+*clientSendLatencyBucket0 : Int
982+*clientSendLatencyBucket1 : Int
983+*clientSendLatencyBucket2 : Int
984+*clientSendLatencyBucket3 : Int
985+
986+-- Client latency histogram thresholds (ms)
987+clientLatencyT1 : Int<Ms>
988+clientLatencyT1 = 50
989+clientLatencyT2 : Int<Ms>
990+clientLatencyT2 = 200
991+clientLatencyT3 : Int<Ms>
992+clientLatencyT3 = 1000
993+
994+-- Auth (challenge + verify) latency tracking (for /stats, reset with latency window)
995+*clientAuthLatencyTotal : Int<Ms>
996+*clientAuthLatencyCount : Int
997+*clientAuthLatencyMax : Int<Ms>
998+
971999-- Last time the background watchdog loop ran (for health check liveness detection)
9721000*lastWatchdogAt : Timestamp
9731001
@@ -984,6 +1012,8 @@ maxPeerLatencyEntries = 100
9841012*cachedHealthForwardQueue : Bool
9851013-- Cached count of federation peers with active failure records (degraded connectivity)
9861014*cachedHealthDegradedPeers : Int
1015+-- Cached request pressure health (activeRequests < 90% of maxActiveRequests)
1016+*cachedHealthRequestPressure : Bool
9871017-- Cached count of destination servers with stale forward queues (updated during prune cycle).
9881018-- Used by health checks to avoid scanning all 16 forward buckets on every /health request.
9891019*cachedForwardStaleCount : Int
@@ -1122,6 +1152,13 @@ route FederationApi where
11221152 POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, expiresAt: Timestamp, priority: Int}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /forward-batch headers {authorization: Text}
11231153 -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int}
11241154 = FederationBatchForward
1155+ -- Delivery receipt: destination server notifies origin that a federated message was
1156+ -- polled/acked by the recipient. Origin server records it as a sender delivery receipt.
1157+ -- Best-effort: failures are not retried (ack path already confirms delivery at the
1158+ -- protocol level; receipts are a UX enhancement for sender-side delivery indicators).
1159+ POST {messageIds: [MessageIdHex], toKeys: [PubkeyHex], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /delivery-receipt headers {authorization: Text}
1160+ -> {ok: Bool, error: Text, receivedCount: Int, retryAfterMs: Int}
1161+ = FederationDeliveryReceipt
11251162
11261163-- Federation callback verification: the receiving server calls back to the sender
11271164-- to confirm a request token is genuine. No shared secret needed.
@@ -1232,7 +1269,7 @@ route MaintenanceApi where
12321269 = RotateSecret
12331270
12341271route HealthApi where
1235− GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, forwardQueueHealthy: Bool, draining: Bool, degradedPeers: Int} = Health
1272+ GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, forwardQueueHealthy: Bool, requestPressureHealthy: Bool, draining: Bool, degradedPeers: Int} = Health
12361273
12371274-- Returns counts of in-memory collections (requires maintenance secret)
12381275-- TODO: If Knot adds raw text response support, add a /metrics endpoint
@@ -1242,7 +1279,7 @@ route HealthApi where
12421279-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
12431280route StatsApi where
12441281 GET /stats headers {authorization: Text}
1245− -> {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, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int}}
1282+ -> {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}}
12461283 = Stats
12471284 GET /stats/peers headers {authorization: Text}
12481285 -> {peers: [PeerStatus], ok: Bool}
@@ -1267,6 +1304,14 @@ route StatsApi where
12671304 -> {ok: Bool, scalePct: Int, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}
12681305 = StatsRates
12691306
1307+-- Prometheus-compatible metrics endpoint. Returns key metrics in a JSON field containing
1308+-- Prometheus exposition format text. A lightweight scraper/adapter extracts the "metrics"
1309+-- field and serves it as text/plain to Prometheus. This is the best we can do since Knot
1310+-- routes always return JSON (no raw text response support).
1311+-- Rate-limited via health rate limiter (same as /ready and /federation/pubkey).
1312+route MetricsApi where
1313+ GET /metrics headers {authorization: Text} -> {metrics: Text, ok: Bool} = Metrics
1314+
12701315-- Readiness probe: returns false during startup warmup (before first prune completes),
12711316-- true once the server is fully operational. Unlike /health, this is unauthenticated
12721317-- and doesn't check subsystem liveness — it only indicates whether the server has
@@ -1274,7 +1319,7 @@ route StatsApi where
12741319route ReadyApi where
12751320 GET /ready -> {ready: Bool} = Ready
12761321
1277−route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | FederationPubkeyApi | MaintenanceApi | HealthApi | StatsApi | ReadyApi
1322+route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | FederationPubkeyApi | MaintenanceApi | HealthApi | StatsApi | MetricsApi | ReadyApi
12781323
12791324-- ============================================
12801325-- Helpers
@@ -1775,7 +1820,7 @@ generateFederationAuth : Text -> Timestamp -> IO {network} Text
17751820generateFederationAuth = \scope requestTimestamp ->
17761821 if serverFederationPrivateKeyHex != ""
17771822 then do
1778− nonce <- genTokenN 4
1823+ nonce <- genTokenN tokenChunks
17791824 let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN ++ "|" ++ nonce)
17801825 let sig = sign (bytesFromHex serverFederationPrivateKeyHex) message
17811826 yield ("Signed " ++ bytesToHex sig ++ ":" ++ nonce)
@@ -2460,22 +2505,33 @@ pruneMessageBucket = \b t -> atomic do
24602505 set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1 t) den perKeyDroppedEntries))
24612506 else yield {}
24622507
2463−-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
2508+-- Prune all 16 message buckets with concurrency: fork non-empty buckets, then wait
2509+-- for all forks to complete via a counter-based barrier before returning. This ensures
2510+-- forward retries (which run after pruning) see fully pruned message state.
24642511-- Each bucket's atomic block only contends on *totalMessageCount (a single Int);
24652512-- rcBuckets and message buckets are per-bucket and independent.
2466−-- STM contention on the shared counter is acceptable because: (1) per-bucket prune
2467−-- is fast (list filter + counter decrement), (2) prune runs every ~30s, not on the
2468−-- hot path, (3) Knot's fork is fire-and-forget so we can't collect per-bucket
2469−-- deltas and apply them in a single final atomic block.
2513+*pendingBucketPrunes : Int
2514+
24702515pruneMessageBucketsConcurrent = \b t ->
2471− if b > 15 then yield {}
2516+ if b > 15 then do
2517+ -- Wait for all forked bucket prunes to complete via STM retry barrier.
2518+ -- Each fork decrements *pendingBucketPrunes; when it reaches 0, all are done.
2519+ atomic do
2520+ pending <- *pendingBucketPrunes
2521+ if pending > 0 then retry else yield {}
24722522 else do
24732523 bucket <- readMessageBucket b
24742524 if count bucket == 0 then pruneMessageBucketsConcurrent (b + 1) t
2475− else if b < 15 then do
2476− fork (pruneMessageBucket b t)
2525+ else do
2526+ atomic do
2527+ p <- *pendingBucketPrunes
2528+ set *pendingBucketPrunes = p + 1
2529+ fork (do
2530+ pruneMessageBucket b t
2531+ atomic do
2532+ p <- *pendingBucketPrunes
2533+ set *pendingBucketPrunes = max 0 (p - 1))
24772534 pruneMessageBucketsConcurrent (b + 1) t
2478− else pruneMessageBucket b t
24792535
24802536-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16
24812537-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one
@@ -2486,31 +2542,33 @@ pruneMessageBucketsConcurrent = \b t ->
24862542-- commutative with concurrent sendMessageCore: if sendMessageCore increments by N between our
24872543-- snapshot and final adjustment, the live counter becomes (snapshot + N + drift) = (actual + N),
24882544-- which is correct. The old approach overwrote with the absolute recount, losing concurrent N.
2489−-- Delta-based reconciliation: snapshots totalMessageCount before scanning, then applies
2490−-- the computed drift atomically. This preserves concurrent sendMessageCore increments:
2491−-- if sendMessageCore adds N messages between snapshot and final write, the live counter
2492−-- becomes (snapshot + N) + (actual - snapshot) = actual + N, which is correct.
2493−-- Old approach: set totalMessageCount = actual (overwrites concurrent N).
2545+-- Per-bucket atomic reconciliation: each bucket is reconciled independently, computing
2546+-- its own delta and applying it to totalMessageCount atomically. This is safe with
2547+-- concurrent prune decrements: if a forked pruneMessageBucket modifies totalMessageCount
2548+-- between our bucket N and bucket N+1, each bucket's delta is applied independently and
2549+-- the shared counter converges. The old snapshot-then-delta approach could overcorrect
2550+-- when concurrent decrements occurred between snapshot and final write.
24942551reconcileMessageCountsDelta : IO Int
2495−reconcileMessageCountsDelta = do
2496− snapshot <- *totalMessageCount
2497− actual <- reconcileMessageCounts 0 0
2498− atomic do
2499− stored <- *totalMessageCount
2500− let drift = actual - snapshot
2501− set *totalMessageCount = max 0 (stored + drift)
2502− yield drift
2552+reconcileMessageCountsDelta = reconcileMessageCountsPerBucket 0 0
25032553
2504−reconcileMessageCounts = \b totalAcc ->
2505− if b > 15 then do
2506− yield totalAcc
2554+reconcileMessageCountsPerBucket = \b totalDrift ->
2555+ if b > 15 then yield totalDrift
25072556 else do
2508− bucketTotal <- atomic do
2557+ bucketDrift <- atomic do
25092558 bucket <- readMessageBucket b
25102559 let counts = countByRecipient bucket
2560+ let actual = fold (\a c -> a + c.count) 0 counts
2561+ -- Read current rcBucket to compute drift vs what's tracked
2562+ oldCounts <- readRcBucket b
2563+ let tracked = fold (\a c -> a + c.count) 0 oldCounts
25112564 writeRcBucket b counts
2512− yield (fold (\a c -> a + c.count) 0 counts)
2513− reconcileMessageCounts (b + 1) (totalAcc + bucketTotal)
2565+ -- Apply per-bucket drift to totalMessageCount atomically
2566+ let drift = actual - tracked
2567+ when (drift != 0) (do
2568+ stored <- *totalMessageCount
2569+ set *totalMessageCount = max 0 (stored + drift))
2570+ yield drift
2571+ reconcileMessageCountsPerBucket (b + 1) (totalDrift + bucketDrift)
25142572
25152573-- ============================================
25162574-- Recipient Count Helpers
@@ -2808,16 +2866,29 @@ updateForwardNonce = \nb messageId toKey t -> do
28082866 then case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> t - n.seenAt
28092867 else 0
28102868 writeNonceBucket nb evictedNonces
2811− -- Update nonce eviction cutoff: the seenAt of the oldest retained nonce.
2812− -- Forwards with timestamps older than this cannot be verified for dedup.
2869+ -- Update nonce eviction cutoff: recompute the global minimum oldest-retained-seenAt
2870+ -- across all 16 buckets. A per-bucket-only approach incorrectly ratchets the cutoff
2871+ -- upward — a bucket with newer evictions would mask older retained nonces in other buckets,
2872+ -- causing legitimate forwards with older timestamps to be rejected.
28132873 when (lruEvicted > 0) (do
2814− let oldestRetainedSeenAt = case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> t; Just {value: n} -> n.seenAt
2815− cutoff <- *nonceEvictionCutoff
2816− -- Only advance the cutoff (never regress): a bucket with older retained nonces
2817− -- shouldn't lower the cutoff set by a bucket with newer evictions.
2818− when (oldestRetainedSeenAt > cutoff) (set *nonceEvictionCutoff = oldestRetainedSeenAt))
2874+ globalMin <- computeGlobalNonceCutoff 0 0
2875+ set *nonceEvictionCutoff = globalMin)
28192876 yield {lruEvicted: lruEvicted, oldestRetainedAge: oldestRetainedAge}
28202877
2878+-- Compute the global nonce eviction cutoff: the minimum oldest-retained-seenAt across
2879+-- all 16 nonce buckets. Returns 0 if no bucket has undergone eviction (all nonces retained).
2880+-- O(16 * bucket_size) scan, called only after LRU eviction (not on every insert).
2881+computeGlobalNonceCutoff : Int -> Timestamp -> IO Timestamp
2882+computeGlobalNonceCutoff = \b acc ->
2883+ if b > 15 then yield acc
2884+ else do
2885+ bucket <- readNonceBucket b
2886+ case minBy bucket (\n -> n.seenAt) of
2887+ Nothing {} -> computeGlobalNonceCutoff (b + 1) acc
2888+ Just {value: n} ->
2889+ let bucketMin = n.seenAt in
2890+ computeGlobalNonceCutoff (b + 1) (if acc == 0 then bucketMin else min acc bucketMin)
2891+
28212892-- Log nonce LRU eviction if any occurred. Call outside atomic blocks.
28222893logNonceEviction : Int -> Int -> Int -> IO ()
28232894logNonceEviction = \nb lruEvicted oldestRetainedAge ->
@@ -3432,12 +3503,21 @@ extractOldest = \items getKey ->
34323503 else {candidate: acc.candidate, items: union acc.items [x]}
34333504 ) {candidate: Nothing {}, items: []} items
34343505
3435−-- Remove the single item with the smallest key value (LRU-style eviction)
3506+-- Remove the single item with the smallest key value (LRU-style eviction).
3507+-- Single-pass: first fold finds the min key, second fold filters it out.
3508+-- Two passes over the data but O(1) intermediate state (just the min key),
3509+-- vs extractOldest which builds a full items list during the fold.
34363510evictOneOldest = \items getKey ->
3437− let r = extractOldest items getKey in
3438− case r.candidate of
3511+ let minResult = fold (\acc x -> case acc of
3512+ Nothing {} -> Just {value: getKey x}
3513+ Just {value: v} -> if getKey x < v then Just {value: getKey x} else acc) Nothing {} items
3514+ in case minResult of
34393515 Nothing {} -> items
3440− Just {} -> r.items
3516+ Just {value: minKey} ->
3517+ let r = fold (\acc x ->
3518+ if not acc.done && getKey x == minKey then {items: acc.items, done: True {}}
3519+ else {items: union acc.items [x], done: acc.done}) {items: [], done: False {}} items
3520+ in r.items
34413521
34423522-- Remove the first item matching a predicate. O(n) single pass.
34433523removeFirstMatch = \items pred ->
@@ -3512,6 +3592,7 @@ removeOneForKey = \items pubkey ->
35123592
35133593-- Extract unique remote (non-local) server names from a list of items.
35143594-- O(n*k) where k = unique servers; acceptable since k is typically small (<20).
3595+-- Knot lacks efficient set/map types, so fold-with-any is the best available approach.
35153596-- NOTE: `union [s] acc` prepends (O(1) per step) vs `union acc [s]` which copies
35163597-- acc (O(|acc|) per step). Since this is a set (order doesn't matter), prepend is
35173598-- preferred for large collections. All set-accumulating folds use this pattern.
@@ -4272,12 +4353,12 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
42724353 -- still stay alive as long as the client is active within any sessionExpiry window.
42734354 -- Uses pubkey pre-filter to find the matching session's expiry without safeCompare,
42744355 -- since pubkey was already verified by authenticateSession above.
4275− let needsRefresh = any sessions (\s -> s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 2)
4356+ let needsRefresh = any sessions (\s -> s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 3)
42764357 when needsRefresh
42774358 -- Pubkey pre-filter on safeCompare: only runs constant-time comparison for
42784359 -- sessions matching the authenticated pubkey (typically 1-2 per bucket),
42794360 -- skipping the expensive safeCompare for all unrelated sessions (~600/bucket).
4280− (writeSessionBucket bucket (do s <- sessions; yield (if s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 2 && safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip, createdAt: s.createdAt} else s)))
4361+ (writeSessionBucket bucket (do s <- sessions; yield (if s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 3 && safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip, createdAt: s.createdAt} else s)))
42814362 onAuth pubkey
42824363
42834364-- Check if a pubkey has no remaining sessions and, if so, clean up local presence.
@@ -5095,6 +5176,19 @@ pruneAndRetry = \t -> do
50955176 set *clientSendLatencyTotal = 0
50965177 set *clientSendLatencyCount = 0
50975178 set *clientSendLatencyMax = 0
5179+ -- Reset client latency histogram buckets
5180+ set *clientPollLatencyBucket0 = 0
5181+ set *clientPollLatencyBucket1 = 0
5182+ set *clientPollLatencyBucket2 = 0
5183+ set *clientPollLatencyBucket3 = 0
5184+ set *clientSendLatencyBucket0 = 0
5185+ set *clientSendLatencyBucket1 = 0
5186+ set *clientSendLatencyBucket2 = 0
5187+ set *clientSendLatencyBucket3 = 0
5188+ -- Reset auth latency metrics
5189+ set *clientAuthLatencyTotal = 0
5190+ set *clientAuthLatencyCount = 0
5191+ set *clientAuthLatencyMax = 0
50985192 set *federationLatencyResetAt = t)
50995193 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
51005194 -- These were queued by handleFederationAck instead of fork+sleep.
@@ -5288,9 +5382,9 @@ pollOnce = \token cursor clientIp -> do
52885382 -- Avoids rewriting the session bucket on every poll (reduces write contention
52895383 -- on the hottest path). Uses pubkey pre-filter so safeCompare only runs on
52905384 -- sessions matching the authenticated pubkey (~1-2 per bucket vs ~600).
5291− let needsRefresh = any sessions (\s -> s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 2)
5385+ let needsRefresh = any sessions (\s -> s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 3)
52925386 when needsRefresh
5293− (writeSessionBucket bucket (do s <- sessions; yield (if s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 2 && safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip, createdAt: s.createdAt} else s)))
5387+ (writeSessionBucket bucket (do s <- sessions; yield (if s.pubkey == pubkey && s.expiresAt - t < sessionExpiry / 3 && safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip, createdAt: s.createdAt} else s)))
52945388 seqCounter <- *seqCounter
52955389 let clampedCursor = clamp 0 (max seqCounter t) cursor
52965390 ackResult <- if clampedCursor > 0
@@ -5600,38 +5694,32 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
56005694
56015695verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do
56025696 let expiry = t + sessionExpiry
5603− validation <- atomic do
5697+ -- Generate token before atomic block (IO can't run inside STM).
5698+ -- Token is discarded if challenge validation fails — no session is created.
5699+ token <- genToken
5700+ -- Single atomic block for challenge verification + session creation.
5701+ -- Eliminates the TOCTOU gap where a challenge could be consumed without
5702+ -- creating a session (e.g., drain phase changes between two atomic blocks).
5703+ authResult <- atomic do
56045704 challenges <- *challenges
56055705 let hasMatch = challengeMatch challenges pubkey challenge clientIp t
56065706 -- Only consume the challenge on full success (match AND valid signature).
5607− -- This prevents an attacker who intercepts the challenge string from burning it
5608− -- by submitting a bad signature from the same IP. Rate limiter still constrains
5609− -- brute-force attempts; not consuming on IP mismatch prevents remote attackers
5610− -- from burning challenges.
5611− -- verify() is inside the atomic block to prevent a TOCTOU gap between challenge
5612− -- validation and consumption. It's pure (no IO) so safe in STM.
5707+ -- verify() is pure (no IO) so safe in STM.
56135708 if not hasMatch
5614− then yield {valid: False {}}
5709+ then yield {valid: False {}, result: Nothing {}}
56155710 else do
56165711 let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
56175712 if not sigValid
5618− then yield {valid: False {}}
5713+ then yield {valid: False {}, result: Nothing {}}
56195714 else do
56205715 set *challenges = challengesExcluding challenges pubkey challenge
5621− yield {valid: True {}}
5622− if validation.valid
5623− then do
5624− token <- genToken
5625− let newBucket = tokenBucket token
5626− authResult <- atomic do
5627− dp <- *drainPhase
5628− if dp >= 1 then yield (Nothing {})
5629− else do
5630− tpc <- *totalPresenceCount
5631− knownPeers <- *knownPeerServers
5632− -- Count per-key sessions without materializing all sessions (O(sessions) but no union)
5633− sessionsForKey <- countSessionsForKey pubkey
5634− tsc <- *totalSessionCount
5716+ dp <- *drainPhase
5717+ if dp >= 1 then yield {valid: True {}, result: Nothing {}}
5718+ else do
5719+ tpc <- *totalPresenceCount
5720+ knownPeers <- *knownPeerServers
5721+ sessionsForKey <- countSessionsForKey pubkey
5722+ tsc <- *totalSessionCount
56355723 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
56365724 -- Update presence: replace local entry for this pubkey, evict oldest if over capacity
56375725 let pb = pubkeyBucket pubkey
@@ -5652,7 +5740,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
56525740 currentBucket <- readSessionBucket nb
56535741 writeSessionBucket nb (union currentBucket [newSession])
56545742 set *totalSessionCount = tsc + 1
5655− yield (Just {value: {peerServers: peerServers, revokedCount: 0}})
5743+ yield {valid: True {}, result: Just {value: {peerServers: peerServers, revokedCount: 0}}}
56565744 else do
56575745 -- Slow path: revocation or capacity pressure.
56585746 -- Bucket-by-bucket approach: revoke sessions for this pubkey across all
@@ -5684,16 +5772,17 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
56845772 -- Final count: estimatedTotal + 1 (new session) - globalEviction (0 or 1)
56855773 let globalEvicted = if estimatedTotal >= maxSessions then 1 else 0
56865774 set *totalSessionCount = estimatedTotal - globalEvicted + 1
5687− yield (Just {value: {peerServers: peerServers, revokedCount: revokedCount}})
5688− case authResult of
5689− Nothing {} -> yield (respond {token: "", expiresAt: 0, error: "draining"})
5690− Just {value: result} -> do
5775+ yield {valid: True {}, result: Just {value: {peerServers: peerServers, revokedCount: revokedCount}}}
5776+ if not authResult.valid
5777+ then yield (respond {token: "", expiresAt: 0, error: "invalid"})
5778+ else case authResult.result of
5779+ Nothing {} -> yield (respond {token: "", expiresAt: 0, error: "draining"})
5780+ Just {value: result} -> do
56915781 when (revokeOthers && result.revokedCount > 0)
56925782 (logInfo ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
56935783 fork (when federationEnabled
56945784 (sendBatchedGossip result.peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl, seenBy: [serverHostN]}] gossipOriginHops))
56955785 yield (respond {token: token, expiresAt: expiry, error: ""})
5696− else yield (respond {token: "", expiresAt: 0, error: "invalid"})
56975786
56985787-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor
56995788-- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.
@@ -6199,9 +6288,8 @@ handleRevoke = \targetToken authorization clientIp respond -> do
61996288-- This gives faster offline detection than waiting for session expiry (1 hour).
62006289handleLogout = \authorization clientIp respond -> do
62016290 t <- now
6202− dp <- *drainPhase
6203− if dp >= 2 then yield (respond {ok: False {}, authorized: False {}})
6204− else do
6291+ -- Logout is allowed during all drain phases (including full drain): it reduces load
6292+ -- and the offline gossip is skipped during drain by gossipOfflineIfLastSession.
62056293 globalRateLimited <- checkGlobalRate t
62066294 if globalRateLimited then yield (respond {ok: False {}, authorized: False {}})
62076295 else do
@@ -6918,6 +7006,74 @@ forwardBucketGroupInner = \entries fromServer b t acc presenceBkt bucket ackedBk
69187006 else updatedNonces
69197007 forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt newBucket ackedBkt delivBkt newNonceBkt newRc (msgStored + 1)
69207008
7009+-- POST /federation/delivery-receipt: Destination server notifies origin that federated
7010+-- messages were polled/acked by the recipient. The origin server records delivery receipts
7011+-- so the sender can see "delivered" status via their next poll response.
7012+-- Best-effort: uses federation rate limiting but failures are not retried. The federation
7013+-- ack path already confirms protocol-level delivery; receipts are a UX enhancement.
7014+handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp correlationId authorization respond -> do
7015+ t <- now
7016+ let corrId = sanitizeCorrelationId correlationId
7017+ when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " delivery-receipt from " ++ fromServer))
7018+ let receiptRespond = \r -> respond {ok: r.ok, error: r.error, receivedCount: 0, retryAfterMs: r.retryAfterMs}
7019+ withFederationAuth authorization t receiptRespond fromServer "delivery-receipt" requestTimestamp (not (isLocalServer fromServer) && count messageIds <= maxMessageIds) (False {}) (do
7020+ -- Pair messageIds with toKeys (parallel arrays). Reject if lengths differ.
7021+ if count messageIds != count toKeys
7022+ then yield (respond {ok: False {}, error: "mismatched_lengths", receivedCount: 0, retryAfterMs: 0})
7023+ else do
7024+ -- Look up senderKey for each message from the local message store.
7025+ -- Only locally-originated messages (originServer == "") have senderKey.
7026+ let capped = takeRelation maxMessageIds messageIds
7027+ receivedCount <- atomic do
7028+ deliveries <- fold (\acc mid ->
7029+ -- Find the matching toKey by position (parallel arrays)
7030+ let idx = count acc.checked in
7031+ let toKey = fold (\tacc tk -> if tacc.i == idx then {i: tacc.i + 1, key: tk} else {i: tacc.i + 1, key: tacc.key}) {i: 0, key: ""} toKeys in
7032+ if toKey.key == ""
7033+ then {checked: union acc.checked [mid], receipts: acc.receipts}
7034+ else do
7035+ let b = pubkeyBucket toKey.key
7036+ -- Check if we have the original message with senderKey
7037+ -- (message may have been pruned, but senderKey was already captured in forward entries by fedDeliveryReceipts)
7038+ {checked: union acc.checked [mid], receipts: union acc.receipts [{senderKey: "", messageId: mid, toKey: toKey.key, deliveredAt: t}]})
7039+ {checked: [], receipts: []} capped
7040+ -- For federation delivery receipts, we need to look up the senderKey from message store
7041+ -- Per-bucket: group by pubkeyBucket and look up senderKey from messages
7042+ let withSender = fold (\acc r ->
7043+ acc) [] deliveries.receipts
7044+ -- Simpler approach: record receipts with empty senderKey — the sender can match by messageId
7045+ -- (they know which messages they sent). This avoids the expensive per-message lookup.
7046+ -- Actually, we need senderKey to route receipts to the right sender's poll response.
7047+ -- Fall back to scanning message buckets for the senderKey.
7048+ yield (count deliveries.receipts)
7049+ -- Look up senderKey per-bucket for the receipt messages
7050+ receiptsByBucket capped toKeys t 0
7051+ yield (respond {ok: True {}, error: "", receivedCount: receivedCount, retryAfterMs: 0}))
7052+
7053+-- Look up senderKey from messages and record delivery receipts per-bucket.
7054+receiptsByBucket = \messageIds toKeys t idx ->
7055+ if idx >= count messageIds then yield {}
7056+ else do
7057+ -- Extract messageId and toKey at current index
7058+ let mid = fold (\acc m -> if acc.i == idx then {i: acc.i + 1, val: m} else {i: acc.i + 1, val: acc.val}) {i: 0, val: ""} messageIds
7059+ let tk = fold (\acc k -> if acc.i == idx then {i: acc.i + 1, val: k} else {i: acc.i + 1, val: acc.val}) {i: 0, val: ""} toKeys
7060+ when (mid.val != "" && tk.val != "") (do
7061+ let b = pubkeyBucket tk.val
7062+ atomic do
7063+ bucket <- readMessageBucket b
7064+ let msg = headOf (do m <- bucket; where m.messageId == mid.val && m.toKey == tk.val && m.senderKey != "" && m.originServer == ""; yield m)
7065+ case msg of
7066+ Nothing {} -> yield {}
7067+ Just {value: m} -> do
7068+ sd <- *senderDeliveries
7069+ let isDup = any sd (\d -> d.messageId == mid.val && d.toKey == tk.val)
7070+ when (not isDup) (do
7071+ let combined = union sd [{senderKey: m.senderKey, messageId: mid.val, toKey: tk.val, deliveredAt: t}]
7072+ set *senderDeliveries = if count combined > maxSenderDeliveries
7073+ then evictOneOldest combined (\d -> d.deliveredAt)
7074+ else combined))
7075+ receiptsByBucket messageIds toKeys t (idx + 1)
7076+
69217077-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
69227078-- Auth checked once; each cancel processed independently. Partial success possible.
69237079handleCancelBatch = \cancels authorization clientIp respond -> do
@@ -6980,7 +7136,7 @@ clientCancelBatchInner = \senderKey entries t resultAcc fedAcc ->
69807136 Nothing {} -> yield {cancelled: False {}, cancelServers: []}
69817137 Just {value: msg} ->
69827138 if msg.originServer != "" then yield {cancelled: False {}, cancelServers: []}
6983− else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, cancelServers: []}
7139+ else if msg.senderKey == "" || msg.senderKey != senderKey then yield {cancelled: False {}, cancelServers: []}
69847140 else do
69857141 writeMessageBucket b (do m <- bucket; where not (m.messageId == entry.messageId && m.toKey == entry.to); yield m)
69867142 tc <- *totalMessageCount
@@ -7201,7 +7357,9 @@ computeAndCacheHealth = \t -> do
72017357 pruneRun <- *pruneRunning
72027358 dp <- *drainPhase
72037359 let warmup = t - sa < healthPruneMaxAge
7204− let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 5
7360+ lastDur <- *lastPruneDuration
7361+ let pruneStuckThreshold = if lastDur > 0 then clamp (backgroundPruneInterval * 3) (backgroundPruneInterval * 10) (lastDur * 3) else backgroundPruneInterval * 5
7362+ let pruneStuck = pruneRun > 0 && t - pruneRun > pruneStuckThreshold
72057363 let cancelCount = count pendingCancels
72067364 let capacityOk = tmc * 10 <= maxMessages * 9
72077365 && presenceCount * 10 <= maxPresenceEntries * 9
@@ -7229,6 +7387,9 @@ computeAndCacheHealth = \t -> do
72297387 -- take >24 minutes (healthForwardQueueMaxAge) to appear, so they're always caught.
72307388 fwdStaleCount <- *cachedForwardStaleCount
72317389 let fwdQueueOk = fwdStaleCount < healthForwardQueueMinStaleServers
7390+ -- Request pressure: healthy when active requests < 90% of max (or backpressure disabled)
7391+ arq <- *activeRequests
7392+ let requestPressureOk = maxActiveRequests <= 0 || arq * 10 < maxActiveRequests * 9
72327393 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
72337394 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
72347395 capacityOk: capacityOk,
@@ -7237,9 +7398,10 @@ computeAndCacheHealth = \t -> do
72377398 watchdogOk: wdOk,
72387399 nonceOk: nonceOk,
72397400 forwardQueueOk: fwdQueueOk,
7401+ requestPressureOk: requestPressureOk,
72407402 isDraining: dp > 0,
72417403 degradedPeers: degradedPeerCount}
7242− let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk && result.forwardQueueOk
7404+ let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk && result.forwardQueueOk && result.requestPressureOk
72437405 atomic do
72447406 set *cachedHealthOk = allOk
72457407 set *cachedHealthPrune = result.pruneOk
@@ -7250,6 +7412,7 @@ computeAndCacheHealth = \t -> do
72507412 set *cachedHealthWatchdog = result.watchdogOk
72517413 set *cachedHealthNonce = result.nonceOk
72527414 set *cachedHealthForwardQueue = result.forwardQueueOk
7415+ set *cachedHealthRequestPressure = result.requestPressureOk
72537416 set *cachedHealthDegradedPeers = result.degradedPeers
72547417 yield result
72557418
@@ -7265,10 +7428,11 @@ readCachedHealth = \respond -> do
72657428 cW <- *cachedHealthWatchdog
72667429 cN <- *cachedHealthNonce
72677430 cFQ <- *cachedHealthForwardQueue
7431+ cRP <- *cachedHealthRequestPressure
72687432 cCW <- *cachedHealthCapacityWarning
72697433 cDP <- *cachedHealthDegradedPeers
72707434 dp <- *drainPhase
7271− yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, capacityWarning: cCW, federationHealthy: cF, watchdogHealthy: cW, nonceHealthy: cN, forwardQueueHealthy: cFQ, draining: dp > 0, degradedPeers: cDP}
7435+ yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, capacityWarning: cCW, federationHealthy: cF, watchdogHealthy: cW, nonceHealthy: cN, forwardQueueHealthy: cFQ, requestPressureHealthy: cRP, draining: dp > 0, degradedPeers: cDP}
72727436 yield (respond cached)
72737437
72747438handleHealth = \authorization respond -> do
@@ -7291,7 +7455,7 @@ handleHealth = \authorization respond -> do
72917455 then readCachedHealth respond
72927456 else do
72937457 result <- computeAndCacheHealth t
7294− let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk && result.forwardQueueOk
7458+ let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk && result.forwardQueueOk && result.requestPressureOk
72957459 yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: result.pruneOk,
72967460 heartbeatHealthy: result.heartbeatOk,
72977461 capacityHealthy: result.capacityOk,
@@ -7300,10 +7464,11 @@ handleHealth = \authorization respond -> do
73007464 watchdogHealthy: result.watchdogOk,
73017465 nonceHealthy: result.nonceOk,
73027466 forwardQueueHealthy: result.forwardQueueOk,
7467+ requestPressureHealthy: result.requestPressureOk,
73037468 draining: result.isDraining,
73047469 degradedPeers: result.degradedPeers})
73057470
7306−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, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0}}
7471+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}}
73077472
73087473handleStats = \authorization respond -> do
73097474 t <- now
@@ -7386,7 +7551,18 @@ handleStats = \authorization respond -> do
73867551 cslm <- *clientSendLatencyMax
73877552 gss <- *gossipSendSuccess
73887553 gsf <- *gossipSendFail
7389− 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}
7554+ cplb0 <- *clientPollLatencyBucket0
7555+ cplb1 <- *clientPollLatencyBucket1
7556+ cplb2 <- *clientPollLatencyBucket2
7557+ cplb3 <- *clientPollLatencyBucket3
7558+ cslb0 <- *clientSendLatencyBucket0
7559+ cslb1 <- *clientSendLatencyBucket1
7560+ cslb2 <- *clientSendLatencyBucket2
7561+ cslb3 <- *clientSendLatencyBucket3
7562+ calt <- *clientAuthLatencyTotal
7563+ calc <- *clientAuthLatencyCount
7564+ calm <- *clientAuthLatencyMax
7565+ 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}
73907566 -- Compute derived metrics
73917567 let sessions = core.sc
73927568 let messages = core.tmc
@@ -7414,7 +7590,7 @@ handleStats = \authorization respond -> do
74147590 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},
74157591 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}},
74167592 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},
7417− 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}})
7593+ 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}})
74187594 else yield (respond zeroStats)
74197595
74207596handleStatsPeers = \authorization respond -> do
@@ -7565,6 +7741,52 @@ handleReady = \respond -> do
75657741 let isReady = lastPrune > 0 && lastPrune >= sa
75667742 yield (respond {ready: isReady})
75677743
7744+-- GET /metrics: Prometheus-format metrics in a JSON wrapper.
7745+-- Authenticated callers get full metrics; unauthenticated get basic capacity only.
7746+handleMetrics = \authorization respond -> do
7747+ t <- now
7748+ healthLimited <- checkHealthRate t
7749+ if healthLimited then yield (respond {metrics: "", ok: False {}})
7750+ else do
7751+ isAuthed <- verifyMaintenanceAuth authorization
7752+ result <- atomic do
7753+ tmc <- *totalMessageCount
7754+ sc <- *totalSessionCount
7755+ pc <- *totalPresenceCount
7756+ fc <- *totalForwardCount
7757+ ap <- *activePolls
7758+ arq <- *activeRequests
7759+ parq <- *peakActiveRequests
7760+ sa <- *startedAt
7761+ dp <- *drainPhase
7762+ yield {tmc: tmc, sc: sc, pc: pc, fc: fc, ap: ap, arq: arq, parq: parq, sa: sa, dp: dp}
7763+ -- Basic capacity metrics (always available)
7764+ let base = "# HELP skrepka_messages_total Current message count\n# TYPE skrepka_messages_total gauge\nskrepka_messages_total " ++ show result.tmc ++ "\n# HELP skrepka_messages_max Maximum message capacity\n# TYPE skrepka_messages_max gauge\nskrepka_messages_max " ++ show maxMessages ++ "\n# HELP skrepka_sessions_total Current session count\n# TYPE skrepka_sessions_total gauge\nskrepka_sessions_total " ++ show result.sc ++ "\n# HELP skrepka_sessions_max Maximum session capacity\n# TYPE skrepka_sessions_max gauge\nskrepka_sessions_max " ++ show maxSessions ++ "\n# HELP skrepka_presence_total Current presence count\n# TYPE skrepka_presence_total gauge\nskrepka_presence_total " ++ show result.pc ++ "\n# HELP skrepka_presence_max Maximum presence capacity\n# TYPE skrepka_presence_max gauge\nskrepka_presence_max " ++ show maxPresenceEntries ++ "\n# HELP skrepka_forwards_total Current forward queue size\n# TYPE skrepka_forwards_total gauge\nskrepka_forwards_total " ++ show result.fc ++ "\n# HELP skrepka_active_polls Current active long-poll connections\n# TYPE skrepka_active_polls gauge\nskrepka_active_polls " ++ show (count result.ap) ++ "\n# HELP skrepka_active_requests Current active HTTP requests\n# TYPE skrepka_active_requests gauge\nskrepka_active_requests " ++ show result.arq ++ "\n# HELP skrepka_peak_active_requests Peak concurrent requests since last reset\n# TYPE skrepka_peak_active_requests gauge\nskrepka_peak_active_requests " ++ show result.parq ++ "\n# HELP skrepka_uptime_seconds Server uptime in seconds\n# TYPE skrepka_uptime_seconds gauge\nskrepka_uptime_seconds " ++ show ((t - result.sa) / 1000) ++ "\n# HELP skrepka_draining Whether the server is draining\n# TYPE skrepka_draining gauge\nskrepka_draining " ++ if result.dp > 0 then "1" else "0" ++ "\n"
7765+ if not isAuthed then yield (respond {metrics: base, ok: True {}})
7766+ else do
7767+ -- Extended metrics for authenticated callers
7768+ ext <- atomic do
7769+ flc <- *federationLatencyCount
7770+ flt <- *federationLatencyTotal
7771+ flm <- *federationLatencyMax
7772+ cplc <- *clientPollLatencyCount
7773+ cplt <- *clientPollLatencyTotal
7774+ cslc <- *clientSendLatencyCount
7775+ cslt <- *clientSendLatencyTotal
7776+ rlg <- *rateLimitedGlobal
7777+ rli <- *rateLimitedIp
7778+ rla <- *rateLimitedAuth
7779+ rlk <- *rateLimitedPerKey
7780+ rlf <- *rateLimitedFederation
7781+ gss <- *gossipSendSuccess
7782+ gsf <- *gossipSendFail
7783+ 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}
7784+ let fedAvg = if ext.flc > 0 then ext.flt / ext.flc else 0
7785+ let pollAvg = if ext.cplc > 0 then ext.cplt / ext.cplc else 0
7786+ let sendAvg = if ext.cslc > 0 then ext.cslt / ext.cslc else 0
7787+ 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"
7788+ yield (respond {metrics: base ++ extended, ok: True {}})
7789+
75687790-- POST /messages/ack: Explicitly acknowledge received messages by ID.
75697791-- Alternative to the implicit cursor-based ack in poll. Benefits:
75707792-- 1. Reduces the window for duplicate delivery (ack immediately without waiting for next poll)
@@ -7788,10 +8010,13 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
77888010 Nothing {} -> yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}
77898011 Just {value: msg} ->
77908012 -- Only allow cancelling locally-originated messages (originServer == "")
7791− -- where the caller is the actual sender. Federation forwards have no
7792− -- senderKey and should not be cancellable via this endpoint.
8013+ -- where the caller's pubkey matches the stored senderKey. Federation
8014+ -- forwards have no senderKey and cannot be cancelled via this endpoint.
8015+ -- Strict check: senderKey must be non-empty and match. MessageId knowledge
8016+ -- alone is insufficient — prevents cancel if messageId leaks via logs,
8017+ -- federation peers, or recipient sharing.
77938018 if msg.originServer != "" then yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}
7794− else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}
8019+ else if msg.senderKey == "" || msg.senderKey != senderKey then yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}
77958020 else do
77968021 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)
77978022 tc <- *totalMessageCount
@@ -8350,6 +8575,7 @@ requestPath = \req -> case req of
83508575 FederationCancel {} -> "POST /federation/cancel"
83518576 FederationBatchCancel {} -> "POST /federation/cancel-batch"
83528577 FederationBatchForward {} -> "POST /federation/forward-batch"
8578+ FederationDeliveryReceipt {} -> "POST /federation/delivery-receipt"
83538579 VerifyToken {} -> "POST /federation/verify"
83548580 FederationPubkey {} -> "GET /federation/pubkey"
83558581 Prune {} -> "POST /maintenance/prune"
@@ -8380,6 +8606,7 @@ requestPath = \req -> case req of
83808606 TopRecipients {} -> "POST /stats/top-recipients"
83818607 StatsForwards {} -> "GET /stats/forwards"
83828608 StatsRates {} -> "GET /stats/rates"
8609+ Metrics {} -> "GET /metrics"
83838610 Ready {} -> "GET /ready"
83848611
83858612-- Extract client IP for access logging (empty for federation/maintenance/stats routes).
@@ -8436,6 +8663,8 @@ serveInner = \req -> case req of
84368663 warnOldProtocol fromServer protocolVersion; handleFederationBatchCancel cancels fromServer requestTimestamp correlationId authorization respond
84378664 FederationBatchForward {forwards, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
84388665 warnOldProtocol fromServer protocolVersion; handleFederationBatchForward forwards fromServer requestTimestamp correlationId authorization respond
8666+ FederationDeliveryReceipt {messageIds, toKeys, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
8667+ warnOldProtocol fromServer protocolVersion; handleFederationDeliveryReceipt messageIds toKeys fromServer requestTimestamp correlationId authorization respond
84398668 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
84408669 FederationPubkey {respond} -> handleFederationPubkey respond
84418670 Prune {authorization, respond} -> handlePrune authorization respond
@@ -8466,6 +8695,7 @@ serveInner = \req -> case req of
84668695 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond
84678696 StatsForwards {authorization, respond} -> handleStatsForwards authorization respond
84688697 StatsRates {authorization, respond} -> handleStatsRates authorization respond
8698+ Metrics {authorization, respond} -> handleMetrics authorization respond
84698699 Ready {respond} -> handleReady respond
84708700
84718701-- Backpressure: check if the server is overloaded (too many concurrent requests).
@@ -8483,6 +8713,7 @@ isOverloadExempt = \req -> case req of
84838713 StatsPeers {} -> True {}
84848714 StatsKey {} -> True {}
84858715 StatsRates {} -> True {}
8716+ Metrics {} -> True {}
84868717 _ -> False {}
84878718
84888719-- Reject overloaded requests with appropriate empty/error responses per route type.
@@ -8515,6 +8746,7 @@ serveOverloaded = \req -> case req of
85158746 FederationCancel {respond} -> respond {ok: False {}, error: "overloaded", cancelled: False {}, retryAfterMs: 1000}
85168747 FederationBatchCancel {respond} -> respond {ok: False {}, error: "overloaded", results: [], retryAfterMs: 1000}
85178748 FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded", results: [], retryAfterMs: 1000}
8749+ FederationDeliveryReceipt {respond} -> respond {ok: False {}, error: "overloaded", receivedCount: 0, retryAfterMs: 1000}
85188750 VerifyToken {respond} -> respond {ok: False {}}
85198751 FederationPubkey {respond} -> respond {pubkey: "", ok: False {}}
85208752 -- Stats/admin routes (exempt routes handled by isOverloadExempt; these are fallbacks)
@@ -8537,6 +8769,7 @@ serveOverloaded = \req -> case req of
85378769 UnallowPeer {respond} -> respond {ok: False {}, removed: False {}}
85388770 ListAllowed {respond} -> respond {peers: [], ok: False {}}
85398771 RotateSecret {respond} -> respond {ok: False {}, error: "overloaded"}
8772+ Metrics {respond} -> respond {metrics: "", ok: False {}}
85408773 -- Exempt routes handled by isOverloadExempt; this branch is unreachable
85418774 _ -> yield {}
85428775
@@ -8579,6 +8812,11 @@ serve = \req -> do
85798812 set *clientPollLatencyTotal = total + latency
85808813 set *clientPollLatencyCount = cnt + 1
85818814 set *clientPollLatencyMax = if latency > mx then latency else mx
8815+ -- Histogram bucket increment
8816+ if latency < clientLatencyT1 then do b <- *clientPollLatencyBucket0; set *clientPollLatencyBucket0 = b + 1
8817+ else if latency < clientLatencyT2 then do b <- *clientPollLatencyBucket1; set *clientPollLatencyBucket1 = b + 1
8818+ else if latency < clientLatencyT3 then do b <- *clientPollLatencyBucket2; set *clientPollLatencyBucket2 = b + 1
8819+ else do b <- *clientPollLatencyBucket3; set *clientPollLatencyBucket3 = b + 1
85828820 SendMessage {} -> atomic do
85838821 total <- *clientSendLatencyTotal
85848822 cnt <- *clientSendLatencyCount
@@ -8586,6 +8824,10 @@ serve = \req -> do
85868824 set *clientSendLatencyTotal = total + latency
85878825 set *clientSendLatencyCount = cnt + 1
85888826 set *clientSendLatencyMax = if latency > mx then latency else mx
8827+ if latency < clientLatencyT1 then do b <- *clientSendLatencyBucket0; set *clientSendLatencyBucket0 = b + 1
8828+ else if latency < clientLatencyT2 then do b <- *clientSendLatencyBucket1; set *clientSendLatencyBucket1 = b + 1
8829+ else if latency < clientLatencyT3 then do b <- *clientSendLatencyBucket2; set *clientSendLatencyBucket2 = b + 1
8830+ else do b <- *clientSendLatencyBucket3; set *clientSendLatencyBucket3 = b + 1
85898831 SendBatch {} -> atomic do
85908832 total <- *clientSendLatencyTotal
85918833 cnt <- *clientSendLatencyCount
@@ -8593,6 +8835,24 @@ serve = \req -> do
85938835 set *clientSendLatencyTotal = total + latency
85948836 set *clientSendLatencyCount = cnt + 1
85958837 set *clientSendLatencyMax = if latency > mx then latency else mx
8838+ if latency < clientLatencyT1 then do b <- *clientSendLatencyBucket0; set *clientSendLatencyBucket0 = b + 1
8839+ else if latency < clientLatencyT2 then do b <- *clientSendLatencyBucket1; set *clientSendLatencyBucket1 = b + 1
8840+ else if latency < clientLatencyT3 then do b <- *clientSendLatencyBucket2; set *clientSendLatencyBucket2 = b + 1
8841+ else do b <- *clientSendLatencyBucket3; set *clientSendLatencyBucket3 = b + 1
8842+ ReqChallenge {} -> atomic do
8843+ total <- *clientAuthLatencyTotal
8844+ cnt <- *clientAuthLatencyCount
8845+ mx <- *clientAuthLatencyMax
8846+ set *clientAuthLatencyTotal = total + latency
8847+ set *clientAuthLatencyCount = cnt + 1
8848+ set *clientAuthLatencyMax = if latency > mx then latency else mx
8849+ VerifyAuth {} -> atomic do
8850+ total <- *clientAuthLatencyTotal
8851+ cnt <- *clientAuthLatencyCount
8852+ mx <- *clientAuthLatencyMax
8853+ set *clientAuthLatencyTotal = total + latency
8854+ set *clientAuthLatencyCount = cnt + 1
8855+ set *clientAuthLatencyMax = if latency > mx then latency else mx
85968856 _ -> yield {}
85978857 let path = requestPath req
85988858 let ip = requestClientIp req
@@ -8672,10 +8932,12 @@ backgroundPrune = do
86728932 set *lastHeartbeatAt = t)
86738933 status <- atomic do
86748934 running <- *pruneRunning
8675− -- 5x threshold: allows heavy prune cycles (reconciliation after restore, large nonce
8676− -- pruning) to complete without triggering false stuck-recovery. Original 2x was too
8677− -- tight and could cause spurious fork resets during legitimate long-running prune cycles.
8678− if running > 0 && t - running < backgroundPruneInterval * 5
8935+ -- Dynamic stuck threshold: 3x the last known cycle duration, with a floor of 3x prune interval
8936+ -- and ceiling of 10x prune interval. Adapts to heavy cycles (reconciliation, large nonce pruning)
8937+ -- without the fixed 5x being too tight for fast cycles or too loose for slow ones.
8938+ lastDuration <- *lastPruneDuration
8939+ let dynamicThreshold = if lastDuration > 0 then clamp (backgroundPruneInterval * 3) (backgroundPruneInterval * 10) (lastDuration * 3) else backgroundPruneInterval * 5
8940+ if running > 0 && t - running < dynamicThreshold
86798941 then yield {canRun: False {}, recovered: False {}, gen: 0, stuckSince: 0}
86808942 else do
86818943 set *pruneRunning = t
@@ -8714,9 +8976,12 @@ backgroundPrune = do
87148976 pruneAndRetry t
87158977 -- Only clear running flag if this is still the latest generation
87168978 -- (prevents a stuck old fork from clearing the flag after a recovery cycle started)
8979+ endT <- now
87178980 atomic do
87188981 currentGen <- *pruneGeneration
8719− when (currentGen == status.gen) (set *pruneRunning = 0)
8982+ when (currentGen == status.gen) (do
8983+ set *pruneRunning = 0
8984+ set *lastPruneDuration = endT - t)
87208985 af <- *activePruneForks
87218986 set *activePruneForks = max 0 (af - 1))
87228987 else logWarn "backgroundPrune: skipping (previous cycle still running)"
@@ -8905,10 +9170,11 @@ main = do
89059170 && forwardNonceRetention >= messageMaxAge
89069171 && maxGlobalFederationConcurrency >= maxFederationConcurrency
89079172 && maxGossipEventsPerServer > maxGossipEvents
9173+ && onlineGossipTtl * 1000 > sessionExpiry
89089174 if not configOk
89099175 then logError "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
89109176 else if not configRelationsOk
8911− then logError "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, sessionExpiry <= maxSessionLifetime, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, maxForwardsPerSenderPerKey <= maxForwardsPerKey, per-key/per-server limits <= global limits), server not started"
9177+ then logError "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, sessionExpiry <= maxSessionLifetime, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, onlineGossipTtl*1000 > sessionExpiry, maxForwardsPerSenderPerKey <= maxForwardsPerKey, per-key/per-server limits <= global limits), server not started"
89129178 else do
89139179 t <- now
89149180 -- Mix random bits into generation to prevent sub-millisecond restart collisions.