d751898 · skrepka · git.iko.soy

git.iko.soy/skrepka

Implement 19 server improvements: cancel authorization hardening, TOCTOU fixes, prune synchronization, nonce eviction correctness, session refresh contention reduction, Prometheus metrics, latency histograms, federation delivery receipts, request pressure health, dynamic stuck detection, and presence TTL alignment

d751898iko6 months ago

server.knot+365−99

compare with master

unifiedsplit

@@ -137,7 +137,7 @@ sessionExpiry = 3600000 -- 1 hour

137maxSessionLifetime : Int<Ms>

138maxSessionLifetime = 86400000 -- 24 hours; absolute session lifetime (forces re-auth even if sliding window keeps refreshing)

139onlineGossipTtl : 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

141offlineGossipMinTtl : Int<S>

142offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events

143maxFederationConcurrency = 10 -- max concurrent outbound federation requests per cycle

@@ -206,13 +206,13 @@ federationLatencyT3 = 2000 -- slow (degraded or distant peer)

206

207-- Server build version: included in /health and /stats responses for deployment verification.

208-- 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"

210

211-- Federation protocol version: included in gossip/forward/ack requests.

212-- Peers log warnings when they receive requests from older versions.

213-- Backward compatible: older peers ignore unknown fields (Knot's JSON parser

214-- drops unknown keys). 0 = pre-versioning peer (field missing or unset).

215−federationProtocolVersion = 14

215+federationProtocolVersion = 15

216

217-- Max seenBy list length in gossip events (bounds memory and bandwidth per event)

218maxSeenByLen = 20

@@ -753,6 +753,10 @@ type CancelNotification = {toKey: PubkeyHex, messageId: MessageIdHex, cancelledA

753

754*pruneGeneration : Int

755

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+

756*ipRates : [IpRateEntry]

760*ipRates : [IpRateEntry]

757

761

758*lookupRates : [LookupRateEntry]

762*lookupRates : [LookupRateEntry]

@@ -968,6 +972,30 @@ maxPeerLatencyEntries = 100

968*clientSendLatencyCount : Int

972*clientSendLatencyCount : Int

969*clientSendLatencyMax : Int<Ms>

973*clientSendLatencyMax : Int<Ms>

970

974

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+

971-- Last time the background watchdog loop ran (for health check liveness detection)

999-- Last time the background watchdog loop ran (for health check liveness detection)

972*lastWatchdogAt : Timestamp

1000*lastWatchdogAt : Timestamp

973

1001

@@ -984,6 +1012,8 @@ maxPeerLatencyEntries = 100

984*cachedHealthForwardQueue : Bool

1012*cachedHealthForwardQueue : Bool

985-- Cached count of federation peers with active failure records (degraded connectivity)

1013-- Cached count of federation peers with active failure records (degraded connectivity)

986*cachedHealthDegradedPeers : Int

1014*cachedHealthDegradedPeers : Int

1015+-- Cached request pressure health (activeRequests < 90% of maxActiveRequests)

1016+*cachedHealthRequestPressure : Bool

987-- Cached count of destination servers with stale forward queues (updated during prune cycle).

1017-- Cached count of destination servers with stale forward queues (updated during prune cycle).

988-- Used by health checks to avoid scanning all 16 forward buckets on every /health request.

1018-- Used by health checks to avoid scanning all 16 forward buckets on every /health request.

989*cachedForwardStaleCount : Int

1019*cachedForwardStaleCount : Int

@@ -1122,6 +1152,13 @@ route FederationApi where

1122 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}

1152 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}

1123 -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int}

1153 -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int}

1124 = FederationBatchForward

1154 = 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

1125

1162

1126-- Federation callback verification: the receiving server calls back to the sender

1163-- Federation callback verification: the receiving server calls back to the sender

1127-- to confirm a request token is genuine. No shared secret needed.

1164-- to confirm a request token is genuine. No shared secret needed.

@@ -1232,7 +1269,7 @@ route MaintenanceApi where

1232 = RotateSecret

1269 = RotateSecret

1233

1270

1234route HealthApi where

1271route 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

1236

1273

1237-- Returns counts of in-memory collections (requires maintenance secret)

1274-- Returns counts of in-memory collections (requires maintenance secret)

1238-- TODO: If Knot adds raw text response support, add a /metrics endpoint

1275-- TODO: If Knot adds raw text response support, add a /metrics endpoint

@@ -1242,7 +1279,7 @@ route HealthApi where

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

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

1243route StatsApi where

1280route StatsApi where

1244 GET /stats headers {authorization: Text}

1281 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}}

1246 = Stats

1283 = Stats

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

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

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

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

@@ -1267,6 +1304,14 @@ route StatsApi where

1267 -> {ok: Bool, scalePct: Int, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}

1304 -> {ok: Bool, scalePct: Int, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}

1268 = StatsRates

1305 = StatsRates

1269

1306

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+

1270-- Readiness probe: returns false during startup warmup (before first prune completes),

1315-- Readiness probe: returns false during startup warmup (before first prune completes),

1271-- true once the server is fully operational. Unlike /health, this is unauthenticated

1316-- true once the server is fully operational. Unlike /health, this is unauthenticated

1272-- and doesn't check subsystem liveness — it only indicates whether the server has

1317-- and doesn't check subsystem liveness — it only indicates whether the server has

@@ -1274,7 +1319,7 @@ route StatsApi where

1274route ReadyApi where

1319route ReadyApi where

1275 GET /ready -> {ready: Bool} = Ready

1320 GET /ready -> {ready: Bool} = Ready

1276

1321

1278

1323

1279-- ============================================

1324-- ============================================

1280-- Helpers

1325-- Helpers

@@ -1775,7 +1820,7 @@ generateFederationAuth : Text -> Timestamp -> IO {network} Text

1775generateFederationAuth = \scope requestTimestamp ->

1820generateFederationAuth = \scope requestTimestamp ->

1776 if serverFederationPrivateKeyHex != ""

1821 if serverFederationPrivateKeyHex != ""

1777 then do

1822 then do

1778− nonce <- genTokenN 4

1823+ nonce <- genTokenN tokenChunks

1779 let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN ++ "|" ++ nonce)

1824 let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN ++ "|" ++ nonce)

1780 let sig = sign (bytesFromHex serverFederationPrivateKeyHex) message

1825 let sig = sign (bytesFromHex serverFederationPrivateKeyHex) message

1781 yield ("Signed " ++ bytesToHex sig ++ ":" ++ nonce)

1826 yield ("Signed " ++ bytesToHex sig ++ ":" ++ nonce)

@@ -2460,22 +2505,33 @@ pruneMessageBucket = \b t -> atomic do

2460 set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1 t) den perKeyDroppedEntries))

2505 set *droppedExpirationNotifs = fold (\acc e -> incrementDropCounter acc e.senderKey 1 t) den perKeyDroppedEntries))

2461 else yield {}

2506 else yield {}

2462

2507

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.

2464-- Each bucket's atomic block only contends on *totalMessageCount (a single Int);

2511-- Each bucket's atomic block only contends on *totalMessageCount (a single Int);

2465-- rcBuckets and message buckets are per-bucket and independent.

2512-- rcBuckets and message buckets are per-bucket and independent.

2466−-- STM contention on the shared counter is acceptable because: (1) per-bucket prune

2513+*pendingBucketPrunes : Int

2467−-- is fast (list filter + counter decrement), (2) prune runs every ~30s, not on the

2514+

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.

2470pruneMessageBucketsConcurrent = \b t ->

2515pruneMessageBucketsConcurrent = \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 {}

2472 else do

2522 else do

2473 bucket <- readMessageBucket b

2523 bucket <- readMessageBucket b

2474 if count bucket == 0 then pruneMessageBucketsConcurrent (b + 1) t

2524 if count bucket == 0 then pruneMessageBucketsConcurrent (b + 1) t

2475− else if b < 15 then do

2525+ else do

2526+ atomic do

2527+ p <- *pendingBucketPrunes

2528+ set *pendingBucketPrunes = p + 1

2529+ fork (do

2476− fork (pruneMessageBucket b t)

2530+ pruneMessageBucket b t

2531+ atomic do

2532+ p <- *pendingBucketPrunes

2533+ set *pendingBucketPrunes = max 0 (p - 1))

2477 pruneMessageBucketsConcurrent (b + 1) t

2534 pruneMessageBucketsConcurrent (b + 1) t

2478− else pruneMessageBucket b t

2479

2535

2480-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16

2536-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16

2481-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one

2537-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one

@@ -2486,31 +2542,33 @@ pruneMessageBucketsConcurrent = \b t ->

2486-- commutative with concurrent sendMessageCore: if sendMessageCore increments by N between our

2542-- commutative with concurrent sendMessageCore: if sendMessageCore increments by N between our

2487-- snapshot and final adjustment, the live counter becomes (snapshot + N + drift) = (actual + N),

2543-- snapshot and final adjustment, the live counter becomes (snapshot + N + drift) = (actual + N),

2488-- which is correct. The old approach overwrote with the absolute recount, losing concurrent N.

2544-- which is correct. The old approach overwrote with the absolute recount, losing concurrent N.

2489−-- Delta-based reconciliation: snapshots totalMessageCount before scanning, then applies

2545+-- Per-bucket atomic reconciliation: each bucket is reconciled independently, computing

2490−-- the computed drift atomically. This preserves concurrent sendMessageCore increments:

2546+-- its own delta and applying it to totalMessageCount atomically. This is safe with

2491−-- if sendMessageCore adds N messages between snapshot and final write, the live counter

2547+-- concurrent prune decrements: if a forked pruneMessageBucket modifies totalMessageCount

2492−-- becomes (snapshot + N) + (actual - snapshot) = actual + N, which is correct.

2548+-- between our bucket N and bucket N+1, each bucket's delta is applied independently and

2493−-- Old approach: set totalMessageCount = actual (overwrites concurrent N).

2549+-- the shared counter converges. The old snapshot-then-delta approach could overcorrect

2550+-- when concurrent decrements occurred between snapshot and final write.

2494reconcileMessageCountsDelta : IO Int

2551reconcileMessageCountsDelta : IO Int

2495−reconcileMessageCountsDelta = do

2552+reconcileMessageCountsDelta = reconcileMessageCountsPerBucket 0 0

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

2503

2553

2504−reconcileMessageCounts = \b totalAcc ->

2554+reconcileMessageCountsPerBucket = \b totalDrift ->

2505− if b > 15 then do

2555+ if b > 15 then yield totalDrift

2506− yield totalAcc

2507 else do

2556 else do

2508− bucketTotal <- atomic do

2557+ bucketDrift <- atomic do

2509 bucket <- readMessageBucket b

2558 bucket <- readMessageBucket b

2510 let counts = countByRecipient bucket

2559 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

2511 writeRcBucket b counts

2564 writeRcBucket b counts

2512− yield (fold (\a c -> a + c.count) 0 counts)

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

2513− reconcileMessageCounts (b + 1) (totalAcc + bucketTotal)

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)

2514

2572

2515-- ============================================

2573-- ============================================

2516-- Recipient Count Helpers

2574-- Recipient Count Helpers

@@ -2808,16 +2866,29 @@ updateForwardNonce = \nb messageId toKey t -> do

2808 then case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> t - n.seenAt

2866 then case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> t - n.seenAt

2809 else 0

2867 else 0

2810 writeNonceBucket nb evictedNonces

2868 writeNonceBucket nb evictedNonces

2811− -- Update nonce eviction cutoff: the seenAt of the oldest retained nonce.

2869+ -- Update nonce eviction cutoff: recompute the global minimum oldest-retained-seenAt

2812− -- Forwards with timestamps older than this cannot be verified for dedup.

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.

2813 when (lruEvicted > 0) (do

2873 when (lruEvicted > 0) (do

2814− let oldestRetainedSeenAt = case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> t; Just {value: n} -> n.seenAt

2874+ globalMin <- computeGlobalNonceCutoff 0 0

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))

2875+ set *nonceEvictionCutoff = globalMin)

2819 yield {lruEvicted: lruEvicted, oldestRetainedAge: oldestRetainedAge}

2876 yield {lruEvicted: lruEvicted, oldestRetainedAge: oldestRetainedAge}

2820

2877

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+

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

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

2822logNonceEviction : Int -> Int -> Int -> IO ()

2893logNonceEviction : Int -> Int -> Int -> IO ()

2823logNonceEviction = \nb lruEvicted oldestRetainedAge ->

2894logNonceEviction = \nb lruEvicted oldestRetainedAge ->

@@ -3432,12 +3503,21 @@ extractOldest = \items getKey ->

3432 else {candidate: acc.candidate, items: union acc.items [x]}

3503 else {candidate: acc.candidate, items: union acc.items [x]}

3433 ) {candidate: Nothing {}, items: []} items

3504 ) {candidate: Nothing {}, items: []} items

3434

3505

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.

3436evictOneOldest = \items getKey ->

3510evictOneOldest = \items getKey ->

3437− let r = extractOldest items getKey in

3511+ let minResult = fold (\acc x -> case acc of

3438− case r.candidate 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

3439 Nothing {} -> items

3515 Nothing {} -> 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

3440− Just {} -> r.items

3520+ in r.items

3441

3521

3442-- Remove the first item matching a predicate. O(n) single pass.

3522-- Remove the first item matching a predicate. O(n) single pass.

3443removeFirstMatch = \items pred ->

3523removeFirstMatch = \items pred ->

@@ -3512,6 +3592,7 @@ removeOneForKey = \items pubkey ->

3512

3592

3513-- Extract unique remote (non-local) server names from a list of items.

3593-- Extract unique remote (non-local) server names from a list of items.

3514-- O(n*k) where k = unique servers; acceptable since k is typically small (<20).

3594-- 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.

3515-- NOTE: `union [s] acc` prepends (O(1) per step) vs `union acc [s]` which copies

3596-- NOTE: `union [s] acc` prepends (O(1) per step) vs `union acc [s]` which copies

3516-- acc (O(|acc|) per step). Since this is a set (order doesn't matter), prepend is

3597-- acc (O(|acc|) per step). Since this is a set (order doesn't matter), prepend is

3517-- preferred for large collections. All set-accumulating folds use this pattern.

3598-- preferred for large collections. All set-accumulating folds use this pattern.

@@ -4272,12 +4353,12 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do

4272 -- still stay alive as long as the client is active within any sessionExpiry window.

4353 -- still stay alive as long as the client is active within any sessionExpiry window.

4273 -- Uses pubkey pre-filter to find the matching session's expiry without safeCompare,

4354 -- Uses pubkey pre-filter to find the matching session's expiry without safeCompare,

4274 -- since pubkey was already verified by authenticateSession above.

4355 -- 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)

4276 when needsRefresh

4357 when needsRefresh

4277 -- Pubkey pre-filter on safeCompare: only runs constant-time comparison for

4358 -- Pubkey pre-filter on safeCompare: only runs constant-time comparison for

4278 -- sessions matching the authenticated pubkey (typically 1-2 per bucket),

4359 -- sessions matching the authenticated pubkey (typically 1-2 per bucket),

4279 -- skipping the expensive safeCompare for all unrelated sessions (~600/bucket).

4360 -- 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)))

4281 onAuth pubkey

4362 onAuth pubkey

4282

4363

4283-- Check if a pubkey has no remaining sessions and, if so, clean up local presence.

4364-- Check if a pubkey has no remaining sessions and, if so, clean up local presence.

@@ -5095,6 +5176,19 @@ pruneAndRetry = \t -> do

5095 set *clientSendLatencyTotal = 0

5176 set *clientSendLatencyTotal = 0

5096 set *clientSendLatencyCount = 0

5177 set *clientSendLatencyCount = 0

5097 set *clientSendLatencyMax = 0

5178 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

5098 set *federationLatencyResetAt = t)

5192 set *federationLatencyResetAt = t)

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

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

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

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

@@ -5288,9 +5382,9 @@ pollOnce = \token cursor clientIp -> do

5288 -- Avoids rewriting the session bucket on every poll (reduces write contention

5382 -- Avoids rewriting the session bucket on every poll (reduces write contention

5289 -- on the hottest path). Uses pubkey pre-filter so safeCompare only runs on

5383 -- on the hottest path). Uses pubkey pre-filter so safeCompare only runs on

5290 -- sessions matching the authenticated pubkey (~1-2 per bucket vs ~600).

5384 -- 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)

5292 when needsRefresh

5386 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)))

5294 seqCounter <- *seqCounter

5388 seqCounter <- *seqCounter

5295 let clampedCursor = clamp 0 (max seqCounter t) cursor

5389 let clampedCursor = clamp 0 (max seqCounter t) cursor

5296 ackResult <- if clampedCursor > 0

5390 ackResult <- if clampedCursor > 0

@@ -5600,38 +5694,32 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->

5600

5694

5601verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do

5695verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do

5602 let expiry = t + sessionExpiry

5696 let expiry = t + sessionExpiry

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).

5603− validation <- atomic do

5703+ authResult <- atomic do

5604 challenges <- *challenges

5704 challenges <- *challenges

5605 let hasMatch = challengeMatch challenges pubkey challenge clientIp t

5705 let hasMatch = challengeMatch challenges pubkey challenge clientIp t

5606 -- Only consume the challenge on full success (match AND valid signature).

5706 -- 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.

5613 if not hasMatch

5708 if not hasMatch

5614− then yield {valid: False {}}

5709+ then yield {valid: False {}, result: Nothing {}}

5615 else do

5710 else do

5616 let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)

5711 let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)

5617 if not sigValid

5712 if not sigValid

5618− then yield {valid: False {}}

5713+ then yield {valid: False {}, result: Nothing {}}

5619 else do

5714 else do

5620 set *challenges = challengesExcluding challenges pubkey challenge

5715 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

5716+ dp <- *drainPhase

5628− if dp >= 1 then yield (Nothing {})

5717+ if dp >= 1 then yield {valid: True {}, result: Nothing {}}

5629− else do

5718+ else do

5630− tpc <- *totalPresenceCount

5719+ tpc <- *totalPresenceCount

5631− knownPeers <- *knownPeerServers

5720+ knownPeers <- *knownPeerServers

5632− -- Count per-key sessions without materializing all sessions (O(sessions) but no union)

5633− sessionsForKey <- countSessionsForKey pubkey

5721+ sessionsForKey <- countSessionsForKey pubkey

5634− tsc <- *totalSessionCount

5722+ tsc <- *totalSessionCount

5635 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}

5723 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}

5636 -- Update presence: replace local entry for this pubkey, evict oldest if over capacity

5724 -- Update presence: replace local entry for this pubkey, evict oldest if over capacity

5637 let pb = pubkeyBucket pubkey

5725 let pb = pubkeyBucket pubkey

@@ -5652,7 +5740,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->

5652 currentBucket <- readSessionBucket nb

5740 currentBucket <- readSessionBucket nb

5653 writeSessionBucket nb (union currentBucket [newSession])

5741 writeSessionBucket nb (union currentBucket [newSession])

5654 set *totalSessionCount = tsc + 1

5742 set *totalSessionCount = tsc + 1

5655− yield (Just {value: {peerServers: peerServers, revokedCount: 0}})

5743+ yield {valid: True {}, result: Just {value: {peerServers: peerServers, revokedCount: 0}}}

5656 else do

5744 else do

5657 -- Slow path: revocation or capacity pressure.

5745 -- Slow path: revocation or capacity pressure.

5658 -- Bucket-by-bucket approach: revoke sessions for this pubkey across all

5746 -- Bucket-by-bucket approach: revoke sessions for this pubkey across all

@@ -5684,16 +5772,17 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->

5684 -- Final count: estimatedTotal + 1 (new session) - globalEviction (0 or 1)

5772 -- Final count: estimatedTotal + 1 (new session) - globalEviction (0 or 1)

5685 let globalEvicted = if estimatedTotal >= maxSessions then 1 else 0

5773 let globalEvicted = if estimatedTotal >= maxSessions then 1 else 0

5686 set *totalSessionCount = estimatedTotal - globalEvicted + 1

5774 set *totalSessionCount = estimatedTotal - globalEvicted + 1

5687− yield (Just {value: {peerServers: peerServers, revokedCount: revokedCount}})

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"})

5688− case authResult of

5778+ else case authResult.result of

5689− Nothing {} -> yield (respond {token: "", expiresAt: 0, error: "draining"})

5779+ Nothing {} -> yield (respond {token: "", expiresAt: 0, error: "draining"})

5690− Just {value: result} -> do

5780+ Just {value: result} -> do

5691 when (revokeOthers && result.revokedCount > 0)

5781 when (revokeOthers && result.revokedCount > 0)

5692 (logInfo ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))

5782 (logInfo ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))

5693 fork (when federationEnabled

5783 fork (when federationEnabled

5694 (sendBatchedGossip result.peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl, seenBy: [serverHostN]}] gossipOriginHops))

5784 (sendBatchedGossip result.peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl, seenBy: [serverHostN]}] gossipOriginHops))

5695 yield (respond {token: token, expiresAt: expiry, error: ""})

5785 yield (respond {token: token, expiresAt: expiry, error: ""})

5696− else yield (respond {token: "", expiresAt: 0, error: "invalid"})

5697

5786

5698-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor

5787-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor

5699-- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.

5788-- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.

@@ -6199,9 +6288,8 @@ handleRevoke = \targetToken authorization clientIp respond -> do

6199-- This gives faster offline detection than waiting for session expiry (1 hour).

6288-- This gives faster offline detection than waiting for session expiry (1 hour).

6200handleLogout = \authorization clientIp respond -> do

6289handleLogout = \authorization clientIp respond -> do

6201 t <- now

6290 t <- now

6202− dp <- *drainPhase

6291+ -- Logout is allowed during all drain phases (including full drain): it reduces load

6203− if dp >= 2 then yield (respond {ok: False {}, authorized: False {}})

6292+ -- and the offline gossip is skipped during drain by gossipOfflineIfLastSession.

6204− else do

6205 globalRateLimited <- checkGlobalRate t

6293 globalRateLimited <- checkGlobalRate t

6206 if globalRateLimited then yield (respond {ok: False {}, authorized: False {}})

6294 if globalRateLimited then yield (respond {ok: False {}, authorized: False {}})

6207 else do

6295 else do

@@ -6918,6 +7006,74 @@ forwardBucketGroupInner = \entries fromServer b t acc presenceBkt bucket ackedBk

6918 else updatedNonces

7006 else updatedNonces

6919 forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt newBucket ackedBkt delivBkt newNonceBkt newRc (msgStored + 1)

7007 forwardBucketGroupInner rest fromServer b t (union acc [{messageId: entry.messageId, ok: True {}}]) presenceBkt newBucket ackedBkt delivBkt newNonceBkt newRc (msgStored + 1)

6920

7008

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+

6921-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.

7077-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.

6922-- Auth checked once; each cancel processed independently. Partial success possible.

7078-- Auth checked once; each cancel processed independently. Partial success possible.

6923handleCancelBatch = \cancels authorization clientIp respond -> do

7079handleCancelBatch = \cancels authorization clientIp respond -> do

@@ -6980,7 +7136,7 @@ clientCancelBatchInner = \senderKey entries t resultAcc fedAcc ->

6980 Nothing {} -> yield {cancelled: False {}, cancelServers: []}

7136 Nothing {} -> yield {cancelled: False {}, cancelServers: []}

6981 Just {value: msg} ->

7137 Just {value: msg} ->

6982 if msg.originServer != "" then yield {cancelled: False {}, cancelServers: []}

7138 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: []}

6984 else do

7140 else do

6985 writeMessageBucket b (do m <- bucket; where not (m.messageId == entry.messageId && m.toKey == entry.to); yield m)

7141 writeMessageBucket b (do m <- bucket; where not (m.messageId == entry.messageId && m.toKey == entry.to); yield m)

6986 tc <- *totalMessageCount

7142 tc <- *totalMessageCount

@@ -7201,7 +7357,9 @@ computeAndCacheHealth = \t -> do

7201 pruneRun <- *pruneRunning

7357 pruneRun <- *pruneRunning

7202 dp <- *drainPhase

7358 dp <- *drainPhase

7203 let warmup = t - sa < healthPruneMaxAge

7359 let warmup = t - sa < healthPruneMaxAge

7360+ lastDur <- *lastPruneDuration

7361+ let pruneStuckThreshold = if lastDur > 0 then clamp (backgroundPruneInterval * 3) (backgroundPruneInterval * 10) (lastDur * 3) else backgroundPruneInterval * 5

7204− let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 5

7362+ let pruneStuck = pruneRun > 0 && t - pruneRun > pruneStuckThreshold

7205 let cancelCount = count pendingCancels

7363 let cancelCount = count pendingCancels

7206 let capacityOk = tmc * 10 <= maxMessages * 9

7364 let capacityOk = tmc * 10 <= maxMessages * 9

7207 && presenceCount * 10 <= maxPresenceEntries * 9

7365 && presenceCount * 10 <= maxPresenceEntries * 9

@@ -7229,6 +7387,9 @@ computeAndCacheHealth = \t -> do

7229 -- take >24 minutes (healthForwardQueueMaxAge) to appear, so they're always caught.

7387 -- take >24 minutes (healthForwardQueueMaxAge) to appear, so they're always caught.

7230 fwdStaleCount <- *cachedForwardStaleCount

7388 fwdStaleCount <- *cachedForwardStaleCount

7231 let fwdQueueOk = fwdStaleCount < healthForwardQueueMinStaleServers

7389 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

7232 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),

7393 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),

7233 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,

7394 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,

7234 capacityOk: capacityOk,

7395 capacityOk: capacityOk,

@@ -7237,9 +7398,10 @@ computeAndCacheHealth = \t -> do

7237 watchdogOk: wdOk,

7398 watchdogOk: wdOk,

7238 nonceOk: nonceOk,

7399 nonceOk: nonceOk,

7239 forwardQueueOk: fwdQueueOk,

7400 forwardQueueOk: fwdQueueOk,

7401+ requestPressureOk: requestPressureOk,

7240 isDraining: dp > 0,

7402 isDraining: dp > 0,

7241 degradedPeers: degradedPeerCount}

7403 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

7243 atomic do

7405 atomic do

7244 set *cachedHealthOk = allOk

7406 set *cachedHealthOk = allOk

7245 set *cachedHealthPrune = result.pruneOk

7407 set *cachedHealthPrune = result.pruneOk

@@ -7250,6 +7412,7 @@ computeAndCacheHealth = \t -> do

7250 set *cachedHealthWatchdog = result.watchdogOk

7412 set *cachedHealthWatchdog = result.watchdogOk

7251 set *cachedHealthNonce = result.nonceOk

7413 set *cachedHealthNonce = result.nonceOk

7252 set *cachedHealthForwardQueue = result.forwardQueueOk

7414 set *cachedHealthForwardQueue = result.forwardQueueOk

7415+ set *cachedHealthRequestPressure = result.requestPressureOk

7253 set *cachedHealthDegradedPeers = result.degradedPeers

7416 set *cachedHealthDegradedPeers = result.degradedPeers

7254 yield result

7417 yield result

7255

7418

@@ -7265,10 +7428,11 @@ readCachedHealth = \respond -> do

7265 cW <- *cachedHealthWatchdog

7428 cW <- *cachedHealthWatchdog

7266 cN <- *cachedHealthNonce

7429 cN <- *cachedHealthNonce

7267 cFQ <- *cachedHealthForwardQueue

7430 cFQ <- *cachedHealthForwardQueue

7431+ cRP <- *cachedHealthRequestPressure

7268 cCW <- *cachedHealthCapacityWarning

7432 cCW <- *cachedHealthCapacityWarning

7269 cDP <- *cachedHealthDegradedPeers

7433 cDP <- *cachedHealthDegradedPeers

7270 dp <- *drainPhase

7434 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}

7272 yield (respond cached)

7436 yield (respond cached)

7273

7437

7274handleHealth = \authorization respond -> do

7438handleHealth = \authorization respond -> do

@@ -7291,7 +7455,7 @@ handleHealth = \authorization respond -> do

7291 then readCachedHealth respond

7455 then readCachedHealth respond

7292 else do

7456 else do

7293 result <- computeAndCacheHealth t

7457 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

7295 yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: result.pruneOk,

7459 yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: result.pruneOk,

7296 heartbeatHealthy: result.heartbeatOk,

7460 heartbeatHealthy: result.heartbeatOk,

7297 capacityHealthy: result.capacityOk,

7461 capacityHealthy: result.capacityOk,

@@ -7300,10 +7464,11 @@ handleHealth = \authorization respond -> do

7300 watchdogHealthy: result.watchdogOk,

7464 watchdogHealthy: result.watchdogOk,

7301 nonceHealthy: result.nonceOk,

7465 nonceHealthy: result.nonceOk,

7302 forwardQueueHealthy: result.forwardQueueOk,

7466 forwardQueueHealthy: result.forwardQueueOk,

7467+ requestPressureHealthy: result.requestPressureOk,

7303 draining: result.isDraining,

7468 draining: result.isDraining,

7304 degradedPeers: result.degradedPeers})

7469 degradedPeers: result.degradedPeers})

7305

7470

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}}

7307

7472

7308handleStats = \authorization respond -> do

7473handleStats = \authorization respond -> do

7309 t <- now

7474 t <- now

@@ -7386,7 +7551,18 @@ handleStats = \authorization respond -> do

7386 cslm <- *clientSendLatencyMax

7551 cslm <- *clientSendLatencyMax

7387 gss <- *gossipSendSuccess

7552 gss <- *gossipSendSuccess

7388 gsf <- *gossipSendFail

7553 gsf <- *gossipSendFail

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

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}

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}

7390 -- Compute derived metrics

7566 -- Compute derived metrics

7391 let sessions = core.sc

7567 let sessions = core.sc

7392 let messages = core.tmc

7568 let messages = core.tmc

@@ -7414,7 +7590,7 @@ handleStats = \authorization respond -> do

7414 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},

7590 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},

7415 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}},

7591 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}},

7416 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},

7592 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}})

7418 else yield (respond zeroStats)

7594 else yield (respond zeroStats)

7419

7595

7420handleStatsPeers = \authorization respond -> do

7596handleStatsPeers = \authorization respond -> do

@@ -7565,6 +7741,52 @@ handleReady = \respond -> do

7565 let isReady = lastPrune > 0 && lastPrune >= sa

7741 let isReady = lastPrune > 0 && lastPrune >= sa

7566 yield (respond {ready: isReady})

7742 yield (respond {ready: isReady})

7567

7743

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+

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

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

7569-- Alternative to the implicit cursor-based ack in poll. Benefits:

7791-- Alternative to the implicit cursor-based ack in poll. Benefits:

7570-- 1. Reduces the window for duplicate delivery (ack immediately without waiting for next poll)

7792-- 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

7788 Nothing {} -> yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}

8010 Nothing {} -> yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}

7789 Just {value: msg} ->

8011 Just {value: msg} ->

7790 -- Only allow cancelling locally-originated messages (originServer == "")

8012 -- Only allow cancelling locally-originated messages (originServer == "")

7791− -- where the caller is the actual sender. Federation forwards have no

8013+ -- where the caller's pubkey matches the stored senderKey. Federation

7792− -- senderKey and should not be cancellable via this endpoint.

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.

7793 if msg.originServer != "" then yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}

8018 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}

7795 else do

8020 else do

7796 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)

8021 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)

7797 tc <- *totalMessageCount

8022 tc <- *totalMessageCount

@@ -8350,6 +8575,7 @@ requestPath = \req -> case req of

8350 FederationCancel {} -> "POST /federation/cancel"

8575 FederationCancel {} -> "POST /federation/cancel"

8351 FederationBatchCancel {} -> "POST /federation/cancel-batch"

8576 FederationBatchCancel {} -> "POST /federation/cancel-batch"

8352 FederationBatchForward {} -> "POST /federation/forward-batch"

8577 FederationBatchForward {} -> "POST /federation/forward-batch"

8578+ FederationDeliveryReceipt {} -> "POST /federation/delivery-receipt"

8353 VerifyToken {} -> "POST /federation/verify"

8579 VerifyToken {} -> "POST /federation/verify"

8354 FederationPubkey {} -> "GET /federation/pubkey"

8580 FederationPubkey {} -> "GET /federation/pubkey"

8355 Prune {} -> "POST /maintenance/prune"

8581 Prune {} -> "POST /maintenance/prune"

@@ -8380,6 +8606,7 @@ requestPath = \req -> case req of

8380 TopRecipients {} -> "POST /stats/top-recipients"

8606 TopRecipients {} -> "POST /stats/top-recipients"

8381 StatsForwards {} -> "GET /stats/forwards"

8607 StatsForwards {} -> "GET /stats/forwards"

8382 StatsRates {} -> "GET /stats/rates"

8608 StatsRates {} -> "GET /stats/rates"

8609+ Metrics {} -> "GET /metrics"

8383 Ready {} -> "GET /ready"

8610 Ready {} -> "GET /ready"

8384

8611

8385-- Extract client IP for access logging (empty for federation/maintenance/stats routes).

8612-- Extract client IP for access logging (empty for federation/maintenance/stats routes).

@@ -8436,6 +8663,8 @@ serveInner = \req -> case req of

8436 warnOldProtocol fromServer protocolVersion; handleFederationBatchCancel cancels fromServer requestTimestamp correlationId authorization respond

8663 warnOldProtocol fromServer protocolVersion; handleFederationBatchCancel cancels fromServer requestTimestamp correlationId authorization respond

8437 FederationBatchForward {forwards, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do

8664 FederationBatchForward {forwards, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do

8438 warnOldProtocol fromServer protocolVersion; handleFederationBatchForward forwards fromServer requestTimestamp correlationId authorization respond

8665 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

8439 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond

8668 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond

8440 FederationPubkey {respond} -> handleFederationPubkey respond

8669 FederationPubkey {respond} -> handleFederationPubkey respond

8441 Prune {authorization, respond} -> handlePrune authorization respond

8670 Prune {authorization, respond} -> handlePrune authorization respond

@@ -8466,6 +8695,7 @@ serveInner = \req -> case req of

8466 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond

8695 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond

8467 StatsForwards {authorization, respond} -> handleStatsForwards authorization respond

8696 StatsForwards {authorization, respond} -> handleStatsForwards authorization respond

8468 StatsRates {authorization, respond} -> handleStatsRates authorization respond

8697 StatsRates {authorization, respond} -> handleStatsRates authorization respond

8698+ Metrics {authorization, respond} -> handleMetrics authorization respond

8469 Ready {respond} -> handleReady respond

8699 Ready {respond} -> handleReady respond

8470

8700

8471-- Backpressure: check if the server is overloaded (too many concurrent requests).

8701-- Backpressure: check if the server is overloaded (too many concurrent requests).

@@ -8483,6 +8713,7 @@ isOverloadExempt = \req -> case req of

8483 StatsPeers {} -> True {}

8713 StatsPeers {} -> True {}

8484 StatsKey {} -> True {}

8714 StatsKey {} -> True {}

8485 StatsRates {} -> True {}

8715 StatsRates {} -> True {}

8716+ Metrics {} -> True {}

8486 _ -> False {}

8717 _ -> False {}

8487

8718

8488-- Reject overloaded requests with appropriate empty/error responses per route type.

8719-- Reject overloaded requests with appropriate empty/error responses per route type.

@@ -8515,6 +8746,7 @@ serveOverloaded = \req -> case req of

8515 FederationCancel {respond} -> respond {ok: False {}, error: "overloaded", cancelled: False {}, retryAfterMs: 1000}

8746 FederationCancel {respond} -> respond {ok: False {}, error: "overloaded", cancelled: False {}, retryAfterMs: 1000}

8516 FederationBatchCancel {respond} -> respond {ok: False {}, error: "overloaded", results: [], retryAfterMs: 1000}

8747 FederationBatchCancel {respond} -> respond {ok: False {}, error: "overloaded", results: [], retryAfterMs: 1000}

8517 FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded", results: [], retryAfterMs: 1000}

8748 FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded", results: [], retryAfterMs: 1000}

8749+ FederationDeliveryReceipt {respond} -> respond {ok: False {}, error: "overloaded", receivedCount: 0, retryAfterMs: 1000}

8518 VerifyToken {respond} -> respond {ok: False {}}

8750 VerifyToken {respond} -> respond {ok: False {}}

8519 FederationPubkey {respond} -> respond {pubkey: "", ok: False {}}

8751 FederationPubkey {respond} -> respond {pubkey: "", ok: False {}}

8520 -- Stats/admin routes (exempt routes handled by isOverloadExempt; these are fallbacks)

8752 -- Stats/admin routes (exempt routes handled by isOverloadExempt; these are fallbacks)

@@ -8537,6 +8769,7 @@ serveOverloaded = \req -> case req of

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

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

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

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

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

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

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

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

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

8541 _ -> yield {}

8774 _ -> yield {}

8542

8775

@@ -8579,6 +8812,11 @@ serve = \req -> do

8579 set *clientPollLatencyTotal = total + latency

8812 set *clientPollLatencyTotal = total + latency

8580 set *clientPollLatencyCount = cnt + 1

8813 set *clientPollLatencyCount = cnt + 1

8581 set *clientPollLatencyMax = if latency > mx then latency else mx

8814 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

8582 SendMessage {} -> atomic do

8820 SendMessage {} -> atomic do

8583 total <- *clientSendLatencyTotal

8821 total <- *clientSendLatencyTotal

8584 cnt <- *clientSendLatencyCount

8822 cnt <- *clientSendLatencyCount

@@ -8586,6 +8824,10 @@ serve = \req -> do

8586 set *clientSendLatencyTotal = total + latency

8824 set *clientSendLatencyTotal = total + latency

8587 set *clientSendLatencyCount = cnt + 1

8825 set *clientSendLatencyCount = cnt + 1

8588 set *clientSendLatencyMax = if latency > mx then latency else mx

8826 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

8589 SendBatch {} -> atomic do

8831 SendBatch {} -> atomic do

8590 total <- *clientSendLatencyTotal

8832 total <- *clientSendLatencyTotal

8591 cnt <- *clientSendLatencyCount

8833 cnt <- *clientSendLatencyCount

@@ -8593,6 +8835,24 @@ serve = \req -> do

8593 set *clientSendLatencyTotal = total + latency

8835 set *clientSendLatencyTotal = total + latency

8594 set *clientSendLatencyCount = cnt + 1

8836 set *clientSendLatencyCount = cnt + 1

8595 set *clientSendLatencyMax = if latency > mx then latency else mx

8837 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

8596 _ -> yield {}

8856 _ -> yield {}

8597 let path = requestPath req

8857 let path = requestPath req

8598 let ip = requestClientIp req

8858 let ip = requestClientIp req

@@ -8672,10 +8932,12 @@ backgroundPrune = do

8672 set *lastHeartbeatAt = t)

8932 set *lastHeartbeatAt = t)

8673 status <- atomic do

8933 status <- atomic do

8674 running <- *pruneRunning

8934 running <- *pruneRunning

8935+ -- Dynamic stuck threshold: 3x the last known cycle duration, with a floor of 3x prune interval

8675− -- 5x threshold: allows heavy prune cycles (reconciliation after restore, large nonce

8936+ -- and ceiling of 10x prune interval. Adapts to heavy cycles (reconciliation, large nonce pruning)

8676− -- pruning) to complete without triggering false stuck-recovery. Original 2x was too

8937+ -- without the fixed 5x being too tight for fast cycles or too loose for slow ones.

8677− -- tight and could cause spurious fork resets during legitimate long-running prune cycles.

8938+ lastDuration <- *lastPruneDuration

8939+ let dynamicThreshold = if lastDuration > 0 then clamp (backgroundPruneInterval * 3) (backgroundPruneInterval * 10) (lastDuration * 3) else backgroundPruneInterval * 5

8678− if running > 0 && t - running < backgroundPruneInterval * 5

8940+ if running > 0 && t - running < dynamicThreshold

8679 then yield {canRun: False {}, recovered: False {}, gen: 0, stuckSince: 0}

8941 then yield {canRun: False {}, recovered: False {}, gen: 0, stuckSince: 0}

8680 else do

8942 else do

8681 set *pruneRunning = t

8943 set *pruneRunning = t

@@ -8714,9 +8976,12 @@ backgroundPrune = do

8714 pruneAndRetry t

8976 pruneAndRetry t

8715 -- Only clear running flag if this is still the latest generation

8977 -- Only clear running flag if this is still the latest generation

8716 -- (prevents a stuck old fork from clearing the flag after a recovery cycle started)

8978 -- (prevents a stuck old fork from clearing the flag after a recovery cycle started)

8979+ endT <- now

8717 atomic do

8980 atomic do

8718 currentGen <- *pruneGeneration

8981 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)

8720 af <- *activePruneForks

8985 af <- *activePruneForks

8721 set *activePruneForks = max 0 (af - 1))

8986 set *activePruneForks = max 0 (af - 1))

8722 else logWarn "backgroundPrune: skipping (previous cycle still running)"

8987 else logWarn "backgroundPrune: skipping (previous cycle still running)"

@@ -8905,10 +9170,11 @@ main = do

8905 && forwardNonceRetention >= messageMaxAge

9170 && forwardNonceRetention >= messageMaxAge

8906 && maxGlobalFederationConcurrency >= maxFederationConcurrency

9171 && maxGlobalFederationConcurrency >= maxFederationConcurrency

8907 && maxGossipEventsPerServer > maxGossipEvents

9172 && maxGossipEventsPerServer > maxGossipEvents

9173+ && onlineGossipTtl * 1000 > sessionExpiry

8908 if not configOk

9174 if not configOk

8909 then logError "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"

9175 then logError "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"

8910 else if not configRelationsOk

9176 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"

8912 else do

9178 else do

8913 t <- now

9179 t <- now

8914 -- Mix random bits into generation to prevent sub-millisecond restart collisions.

9180 -- Mix random bits into generation to prevent sub-millisecond restart collisions.