@@ -195,15 +195,24 @@ lightReconciliationInterval = 5
195federationLatencyWindowMs : Int<Ms>
195federationLatencyWindowMs : Int<Ms>
196federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
196federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
198+-- Federation latency histogram thresholds (ms). Counts how many calls fell into each bucket:
199+-- [0, t1), [t1, t2), [t2, t3), [t3, +inf). Operators use these to approximate P95/P99.
200+federationLatencyT1 : Int<Ms>
201+federationLatencyT1 = 100 -- fast (local network / same datacenter)
202+federationLatencyT2 : Int<Ms>
203+federationLatencyT2 = 500 -- normal (cross-region)
204+federationLatencyT3 : Int<Ms>
205+federationLatencyT3 = 2000 -- slow (degraded or distant peer)
198-- Server build version: included in /health and /stats responses for deployment verification.
207-- Server build version: included in /health and /stats responses for deployment verification.
199-- Update this string when deploying a new build so operators can confirm which version is running.
208-- Update this string when deploying a new build so operators can confirm which version is running.
200−serverBuildVersion = "2026-04-26e"
209+serverBuildVersion = "2026-04-26f"
202-- Federation protocol version: included in gossip/forward/ack requests.
211-- Federation protocol version: included in gossip/forward/ack requests.
203-- Peers log warnings when they receive requests from older versions.
212-- Peers log warnings when they receive requests from older versions.
204-- Backward compatible: older peers ignore unknown fields (Knot's JSON parser
213-- Backward compatible: older peers ignore unknown fields (Knot's JSON parser
205-- drops unknown keys). 0 = pre-versioning peer (field missing or unset).
214-- drops unknown keys). 0 = pre-versioning peer (field missing or unset).
206−federationProtocolVersion = 13
215+federationProtocolVersion = 14
208-- Max seenBy list length in gossip events (bounds memory and bandwidth per event)
217-- Max seenBy list length in gossip events (bounds memory and bandwidth per event)
209maxSeenByLen = 20
218maxSeenByLen = 20
@@ -317,6 +326,10 @@ maxSenderExpirationsPerKey = 100
317-- so senders can learn their message was delivered (returned in poll response).
326-- so senders can learn their message was delivered (returned in poll response).
318maxSenderDeliveries = 10000
327maxSenderDeliveries = 10000
319maxSenderDeliveriesPerKey = 100
328maxSenderDeliveriesPerKey = 100
329+-- Cancel notifications: track when a message is retracted so the recipient can remove it from their UI.
330+-- Populated when /messages/cancel, /federation/cancel, or /federation/cancel-batch succeeds.
331+maxCancelNotifications = 10000
332+maxCancelNotificationsPerKey = 100
320-- Max pubkeys in a batch lookup request
333-- Max pubkeys in a batch lookup request
321maxBatchLookupKeys = 50
334maxBatchLookupKeys = 50
322-- Max per-sender drop counter entries (bounds memory for per-sender receipt/expiration drop tracking)
335-- Max per-sender drop counter entries (bounds memory for per-sender receipt/expiration drop tracking)
@@ -465,7 +478,7 @@ type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int, prevCount: Int
465type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}
478type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}
466-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles
479-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles
467type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}
480type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}
468−type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool, pendingForwards: Int, avgLatencyMs: Int, maxLatencyMs: Int, latencySamples: Int}
481+type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool, pendingForwards: Int, avgLatencyMs: Int, maxLatencyMs: Int, latencySamples: Int, gossipEventsPerMin: Int}
469type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
482type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
470type RecipientCount = {pubkey: PubkeyHex, count: Int}
483type RecipientCount = {pubkey: PubkeyHex, count: Int}
471type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
484type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
@@ -499,6 +512,11 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
499-- Only for locally-originated messages (senderKey != ""); federated messages have no senderKey.
512-- Only for locally-originated messages (senderKey != ""); federated messages have no senderKey.
500type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
513type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
515+-- Cancel notification: tracks when a message was retracted by the sender. Delivered to
516+-- the recipient via their next poll response so clients can remove the message from the UI.
517+-- Keyed by toKey (recipient). cancelledAt is when the cancel was processed on this server.
518+type CancelNotification = {toKey: PubkeyHex, messageId: MessageIdHex, cancelledAt: Timestamp}
502-- ============================================
520-- ============================================
503-- Persisted State
521-- Persisted State
504-- ============================================
522-- ============================================
@@ -777,6 +795,13 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
777*federationLatencyCount : Int
795*federationLatencyCount : Int
778*federationLatencyMax : Int<Ms>
796*federationLatencyMax : Int<Ms>
798+-- Federation latency histogram: counts calls in each bucket for percentile approximation.
799+-- Buckets: [0,T1), [T1,T2), [T2,T3), [T3,+inf). Reset with latency window.
800+*federationLatencyBucket0 : Int
801+*federationLatencyBucket1 : Int
802+*federationLatencyBucket2 : Int
803+*federationLatencyBucket3 : Int
780-- Federation verify callback metrics (exposed in /stats, reset with latency window)
805-- Federation verify callback metrics (exposed in /stats, reset with latency window)
781*slowVerifyCount : Int
806*slowVerifyCount : Int
782*verifyFailCount : Int
807*verifyFailCount : Int
@@ -885,6 +910,10 @@ maxPeerLatencyEntries = 100
885-- Populated during ackPolledMessages and handleAckMessages for locally-originated messages.
910-- Populated during ackPolledMessages and handleAckMessages for locally-originated messages.
886*senderDeliveries : [SenderDelivery]
911*senderDeliveries : [SenderDelivery]
913+-- Cancel notifications: messages retracted by the sender, queued for the recipient's next poll.
914+-- Populated during handleCancelMessage, handleFederationCancel, handleFederationBatchCancel.
915+*cancelNotifications : [CancelNotification]
888-- Pending federation cancels: queued when cancel delivery to a remote server fails.
917-- Pending federation cancels: queued when cancel delivery to a remote server fails.
889-- Retried during the prune cycle alongside forward retries and ack retries.
918-- Retried during the prune cycle alongside forward retries and ack retries.
890*pendingFederationCancels : [PendingFederationCancel]
919*pendingFederationCancels : [PendingFederationCancel]
@@ -995,7 +1024,7 @@ route AuthApi where
995route ClientApi where
1024route ClientApi where
996 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
1025 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
997 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
1026 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
998− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}], droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int, loadPct: Int}
1027+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}], droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int, loadPct: Int, messageCancellations: [{messageId: MessageIdHex, cancelledAt: Timestamp}]}
1000 /messages
1029 /messages
1001 -- Send an E2E encrypted message to a recipient pubkey.
1030 -- Send an E2E encrypted message to a recipient pubkey.
@@ -1057,7 +1086,7 @@ route FederationApi where
1057 /federation
1086 /federation
1058 -- Receive presence gossip events from a peer server
1087 -- Receive presence gossip events from a peer server
1059 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /gossip headers {authorization: Text}
1088 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /gossip headers {authorization: Text}
1060− -> {ok: Bool, error: Text}
1089+ -> {ok: Bool, error: Text, retryAfterMs: Int}
1061 = RecvGossip
1090 = RecvGossip
1062 -- Receive a forwarded message from a peer server on behalf of a sender.
1091 -- Receive a forwarded message from a peer server on behalf of a sender.
1063 -- NOTE: senderKey is intentionally NOT included in the forward protocol. The sender's
1092 -- NOTE: senderKey is intentionally NOT included in the forward protocol. The sender's
@@ -1066,11 +1095,11 @@ route FederationApi where
1066 -- federated messages requires the client to include sender identity in the encrypted
1095 -- federated messages requires the client to include sender identity in the encrypted
1067 -- payload; the server cannot verify sender identity for federation-forwarded messages.
1096 -- payload; the server cannot verify sender identity for federation-forwarded messages.
1068 POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, expiresAt: Timestamp, priority: Int, protocolVersion: Int} /forward headers {authorization: Text}
1097 POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, expiresAt: Timestamp, priority: Int, protocolVersion: Int} /forward headers {authorization: Text}
1069− -> {ok: Bool, error: Text}
1098+ -> {ok: Bool, error: Text, retryAfterMs: Int}
1070 = ForwardMessage
1099 = ForwardMessage
1071 -- Acknowledge delivery of messages back to the origin server
1100 -- Acknowledge delivery of messages back to the origin server
1072 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /ack headers {authorization: Text}
1101 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /ack headers {authorization: Text}
1073− -> {ok: Bool, error: Text, ackedCount: Int, results: [{messageId: MessageIdHex, found: Bool}]}
1102+ -> {ok: Bool, error: Text, ackedCount: Int, results: [{messageId: MessageIdHex, found: Bool}], retryAfterMs: Int}
1074 = FederationAck
1103 = FederationAck
1075 -- Cancel (retract) a previously forwarded message. The origin server sends this
1104 -- Cancel (retract) a previously forwarded message. The origin server sends this
1076 -- when the sender cancels a message via /messages/cancel. The receiving server
1105 -- when the sender cancels a message via /messages/cancel. The receiving server
@@ -1081,17 +1110,17 @@ route FederationApi where
1081 -- would go to the wrong server, which wouldn't have the token. This prevents a
1110 -- would go to the wrong server, which wouldn't have the token. This prevents a
1082 -- malicious relay from cancelling messages it forwarded on behalf of the origin.
1111 -- malicious relay from cancelling messages it forwarded on behalf of the origin.
1083 POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /cancel headers {authorization: Text}
1112 POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /cancel headers {authorization: Text}
1084− -> {ok: Bool, error: Text, cancelled: Bool}
1113+ -> {ok: Bool, error: Text, cancelled: Bool, retryAfterMs: Int}
1085 = FederationCancel
1114 = FederationCancel
1086 -- Batch cancel: cancel multiple previously forwarded messages in one request.
1115 -- Batch cancel: cancel multiple previously forwarded messages in one request.
1087 POST {cancels: [{messageId: MessageIdHex, toKey: PubkeyHex}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /cancel-batch headers {authorization: Text}
1116 POST {cancels: [{messageId: MessageIdHex, toKey: PubkeyHex}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /cancel-batch headers {authorization: Text}
1088− -> {ok: Bool, error: Text, results: [BatchCancelResult]}
1117+ -> {ok: Bool, error: Text, results: [BatchCancelResult], retryAfterMs: Int}
1089 = FederationBatchCancel
1118 = FederationBatchCancel
1090 -- Batch forward: forward multiple messages in one request (reduces HTTP round-trips).
1119 -- Batch forward: forward multiple messages in one request (reduces HTTP round-trips).
1091 -- Each message is processed independently; partial success is possible.
1120 -- Each message is processed independently; partial success is possible.
1092 -- Results are returned per-message so the sender knows which forwards succeeded.
1121 -- Results are returned per-message so the sender knows which forwards succeeded.
1093 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}
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}
1094− -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}]}
1123+ -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int}
1095 = FederationBatchForward
1124 = FederationBatchForward
1097-- Federation callback verification: the receiving server calls back to the sender
1126-- Federation callback verification: the receiving server calls back to the sender
@@ -1213,7 +1242,7 @@ route HealthApi where
1213-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1242-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1214route StatsApi where
1243route StatsApi where
1215 GET /stats headers {authorization: Text}
1244 GET /stats headers {authorization: Text}
1216− -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, activePolls: Int, activePollsMax: Int, recipientCounts: Int, messageBucketMax: Int, messageBucketMin: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}], pendingCancels: Int, gossipSuccess: Int, gossipFail: Int}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, senderExpirations: Int, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int}}
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}}
1218 GET /stats/peers headers {authorization: Text}
1247 GET /stats/peers headers {authorization: Text}
1219 -> {peers: [PeerStatus], ok: Bool}
1248 -> {peers: [PeerStatus], ok: Bool}
@@ -1350,6 +1379,30 @@ drainDropCounter = \entries key ->
1350 case headOf (do e <- entries; where e.senderKey == key; yield e) of
1379 case headOf (do e <- entries; where e.senderKey == key; yield e) of
1351 Nothing {} -> {count: 0, remaining: entries}
1380 Nothing {} -> {count: 0, remaining: entries}
1352 Just {value: e} -> {count: e.count, remaining: do x <- entries; where x.senderKey != key; yield x}
1381 Just {value: e} -> {count: e.count, remaining: do x <- entries; where x.senderKey != key; yield x}
1383+-- Queue a cancel notification for a recipient. Called when a message is cancelled
1384+-- so the recipient's next poll delivers the retraction. Per-key cap prevents one
1385+-- targeted recipient from consuming the whole notification queue.
1386+queueCancelNotification : PubkeyHex -> MessageIdHex -> Timestamp -> IO ()
1387+queueCancelNotification = \toKey messageId t -> atomic do
1388+ cn <- *cancelNotifications
1389+ let forKey = count (do c <- cn; where c.toKey == toKey; yield c)
1390+ if forKey >= maxCancelNotificationsPerKey then yield {}
1392+ let updated = union cn [{toKey: toKey, messageId: messageId, cancelledAt: t}]
1393+ set *cancelNotifications = if count updated > maxCancelNotifications
1394+ then evictOneOldest updated (\c -> c.cancelledAt)
1397+-- Batch version: queue cancel notifications for multiple entries at once.
1398+queueCancelNotificationBatch = \entries t ->
1399+ case headOf entries of
1400+ Nothing {} -> yield {}
1401+ Just {value: e} -> do
1402+ let rest = do x <- entries; where x.messageId != e.messageId || x.toKey != e.toKey; yield x
1403+ queueCancelNotification e.toKey e.messageId t
1404+ queueCancelNotificationBatch rest t
1353-- NOTE: fold always traverses the full list even after finding the first element.
1406-- NOTE: fold always traverses the full list even after finding the first element.
1354-- This is O(n) where O(1) suffices, but Knot lacks list destructuring (case xs of
1407-- This is O(n) where O(1) suffices, but Knot lacks list destructuring (case xs of
1355-- (x : rest) -> ...) that would allow short-circuit. Used pervasively on hot paths
1408-- (x : rest) -> ...) that would allow short-circuit. Used pervasively on hot paths
@@ -1722,7 +1775,7 @@ generateFederationAuth : Text -> Timestamp -> IO {network} Text
1722generateFederationAuth = \scope requestTimestamp ->
1775generateFederationAuth = \scope requestTimestamp ->
1723 if serverFederationPrivateKeyHex != ""
1776 if serverFederationPrivateKeyHex != ""
1725− nonce <- genTokenN 2
1778+ nonce <- genTokenN 4
1726 let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN ++ "|" ++ nonce)
1779 let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN ++ "|" ++ nonce)
1727 let sig = sign (bytesFromHex serverFederationPrivateKeyHex) message
1780 let sig = sign (bytesFromHex serverFederationPrivateKeyHex) message
1728 yield ("Signed " ++ bytesToHex sig ++ ":" ++ nonce)
1781 yield ("Signed " ++ bytesToHex sig ++ ":" ++ nonce)
@@ -2496,26 +2549,20 @@ countByRecipient = \items ->
2496 let sorted = sortBy (\m -> m.toKey) items in
2549 let sorted = sortBy (\m -> m.toKey) items in
2497 countSortedGroups sorted
2550 countSortedGroups sorted
2499−-- Linear scan over sorted items: counts consecutive runs of the same toKey.
2552+-- Single-pass fold over sorted items: counts consecutive runs of the same toKey.
2500−-- Tail-recursive (TCO-safe in Knot). Each step either increments the current
2553+-- O(n) total: one fold step per item, O(1) prepend per group boundary.
2501−-- group's count or emits the group and starts a new one.
2554+-- Replaces the previous headOf+filter pattern which was O(n) per step = O(n²) total.
2502−-- Uses messageId for element removal (unique per bucket) instead of full structural
2555+-- Uses "union [new] acc" (prepend, O(1)) instead of "union acc [new]" (copy, O(k)).
2503−-- equality, avoiding expensive per-field comparison on Message records with large blobs.
2504countSortedGroups = \items ->
2556countSortedGroups = \items ->
2505− case headOf items of
2557+ let result = fold (\acc m ->
2506− Nothing {} -> []
2558+ if acc.currentKey == ""
2507− Just {value: first} ->
2559+ then {currentKey: m.toKey, currentCount: 1, groups: acc.groups}
2508− let rest = do x <- items; where x.messageId != first.messageId; yield x in
2509− countSortedGroupsInner rest first.toKey 1 []
2511−countSortedGroupsInner = \items currentKey currentCount acc ->
2512− case headOf items of
2513− Nothing {} -> union acc [{pubkey: currentKey, count: currentCount}]
2515− let rest = do r <- items; where r.messageId != x.messageId; yield r in
2516− if x.toKey == currentKey
2560+ else if m.toKey == acc.currentKey
2517− then countSortedGroupsInner rest currentKey (currentCount + 1) acc
2561+ then {currentKey: acc.currentKey, currentCount: acc.currentCount + 1, groups: acc.groups}
2562+ else {currentKey: m.toKey, currentCount: 1, groups: union [{pubkey: acc.currentKey, count: acc.currentCount}] acc.groups})
2563+ {currentKey: "", currentCount: 0, groups: []} items
2564+ in if result.currentKey == "" then result.groups
2518− else countSortedGroupsInner rest x.toKey 1 (union acc [{pubkey: currentKey, count: currentCount}])
2565+ else union [{pubkey: result.currentKey, count: result.currentCount}] result.groups
2520-- Batch-decrement recipient counts: subtract per-key removal counts in a single pass
2567-- Batch-decrement recipient counts: subtract per-key removal counts in a single pass
2521-- instead of calling adjustRecipientCount in a loop.
2568-- instead of calling adjustRecipientCount in a loop.
@@ -3206,6 +3253,11 @@ fetchFederation = \server request -> do
3206 set *federationLatencyTotal = total + latency
3253 set *federationLatencyTotal = total + latency
3207 set *federationLatencyCount = cnt + 1
3254 set *federationLatencyCount = cnt + 1
3208 set *federationLatencyMax = if latency > mx then latency else mx
3255 set *federationLatencyMax = if latency > mx then latency else mx
3256+ -- Histogram bucket increment
3257+ if latency < federationLatencyT1 then do b <- *federationLatencyBucket0; set *federationLatencyBucket0 = b + 1
3258+ else if latency < federationLatencyT2 then do b <- *federationLatencyBucket1; set *federationLatencyBucket1 = b + 1
3259+ else if latency < federationLatencyT3 then do b <- *federationLatencyBucket2; set *federationLatencyBucket2 = b + 1
3260+ else do b <- *federationLatencyBucket3; set *federationLatencyBucket3 = b + 1
3209 -- Per-peer latency tracking
3261 -- Per-peer latency tracking
3210 peerLats <- *peerLatencies
3262 peerLats <- *peerLatencies
3211 let peerEntry = headOf (do p <- peerLats; where p.server == server; yield p)
3263 let peerEntry = headOf (do p <- peerLats; where p.server == server; yield p)
@@ -3749,6 +3801,9 @@ pruneExpired = \t -> do
3749 -- Prune old sender delivery receipts (same retention as expirations)
3801 -- Prune old sender delivery receipts (same retention as expirations)
3750 senderDel <- *senderDeliveries
3802 senderDel <- *senderDeliveries
3751 set *senderDeliveries = do d <- senderDel; where t - d.deliveredAt < unackedMessageMaxAge; yield d
3803 set *senderDeliveries = do d <- senderDel; where t - d.deliveredAt < unackedMessageMaxAge; yield d
3804+ -- Prune old cancel notifications (same retention as delivery receipts)
3805+ cancelNotifs <- *cancelNotifications
3806+ set *cancelNotifications = do c <- cancelNotifs; where t - c.cancelledAt < unackedMessageMaxAge; yield c
3752 -- Reconcile global federation connection counter with actual in-flight state
3807 -- Reconcile global federation connection counter with actual in-flight state
3753 -- (self-heals drift from leaked slots where a forked task died without decrementing)
3808 -- (self-heals drift from leaked slots where a forked task died without decrementing)
3754 inFlightRec <- *federationInFlight
3809 inFlightRec <- *federationInFlight
@@ -4130,13 +4185,13 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
4130 if isBlockedPeer fromServer blocked
4185 if isBlockedPeer fromServer blocked
4132 logWarn ("federation: blocked peer " ++ fromServer ++ " rejected")
4187 logWarn ("federation: blocked peer " ++ fromServer ++ " rejected")
4133− yield (respond {ok: False {}, error: "blocked"})
4188+ yield (respond {ok: False {}, error: "blocked", retryAfterMs: 0})
4135 allowed <- *allowedPeers
4190 allowed <- *allowedPeers
4136 if not (isAllowedByAllowlist fromServer allowed)
4191 if not (isAllowedByAllowlist fromServer allowed)
4138 logWarn ("federation: peer " ++ fromServer ++ " not in allowlist, rejected")
4193 logWarn ("federation: peer " ++ fromServer ++ " not in allowlist, rejected")
4139− yield (respond {ok: False {}, error: "not_allowed"})
4194+ yield (respond {ok: False {}, error: "not_allowed", retryAfterMs: 0})
4141 -- Dispatch on auth scheme: "Signed " = Ed25519 signature, "Bearer " = verify callback
4196 -- Dispatch on auth scheme: "Signed " = Ed25519 signature, "Bearer " = verify callback
4142 -- New format "Signed <sig>:<nonce>" includes a random nonce to prevent replay of
4197 -- New format "Signed <sig>:<nonce>" includes a random nonce to prevent replay of
@@ -4144,7 +4199,7 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
4144 verified <- if take 7 authorization == "Signed "
4199 verified <- if take 7 authorization == "Signed "
4146 let payload = drop 7 authorization
4201 let payload = drop 7 authorization
4147− if payload == "" || tooLong payload (maxSignatureLen + 13)
4202+ if payload == "" || tooLong payload (maxSignatureLen + 25)
4148 then yield (False {})
4203 then yield (False {})
4150 let parts = splitOnColon payload in
4205 let parts = splitOnColon payload in
@@ -4153,15 +4208,15 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
4153 else verifyFederationSignature fromServer payload scope requestTimestamp
4208 else verifyFederationSignature fromServer payload scope requestTimestamp
4154 else verifyFederationCallback fromServer authorization scope
4209 else verifyFederationCallback fromServer authorization scope
4155 if not verified
4210 if not verified
4156− then yield (respond {ok: False {}, error: "unauthorized"})
4211+ then yield (respond {ok: False {}, error: "unauthorized", retryAfterMs: 0})
4157 -- Replay protection: reject requests with timestamps too far from local clock.
4212 -- Replay protection: reject requests with timestamps too far from local clock.
4158 -- requestTimestamp == 0 means the peer doesn't support this field yet (accept).
4213 -- requestTimestamp == 0 means the peer doesn't support this field yet (accept).
4159 else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance)
4214 else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance)
4161 logWarn ("federation: clock skew " ++ show (t - requestTimestamp) ++ "ms from " ++ fromServer ++ ", rejecting (tolerance: " ++ show clockSkewTolerance ++ "ms)")
4216 logWarn ("federation: clock skew " ++ show (t - requestTimestamp) ++ "ms from " ++ fromServer ++ ", rejecting (tolerance: " ++ show clockSkewTolerance ++ "ms)")
4162− yield (respond {ok: False {}, error: "clock_skew"})
4217+ yield (respond {ok: False {}, error: "clock_skew", retryAfterMs: 0})
4163 else if not precondition
4218 else if not precondition
4164− then yield (respond {ok: False {}, error: "invalid_request"})
4219+ then yield (respond {ok: False {}, error: "invalid_request", retryAfterMs: 0})
4166 -- Replay detection: reject duplicate (server, timestamp) pairs within the skew window
4221 -- Replay detection: reject duplicate (server, timestamp) pairs within the skew window
4167 replayDetected <- if requestTimestamp > 0
4222 replayDetected <- if requestTimestamp > 0
@@ -4178,21 +4233,23 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
4178 if replayDetected
4233 if replayDetected
4180 logWarn ("federation: replay detected from " ++ fromServer ++ " (timestamp: " ++ show requestTimestamp ++ ")")
4235 logWarn ("federation: replay detected from " ++ fromServer ++ " (timestamp: " ++ show requestTimestamp ++ ")")
4181− yield (respond {ok: False {}, error: "replay_detected"})
4236+ yield (respond {ok: False {}, error: "replay_detected", retryAfterMs: 0})
4183 globalLimited <- checkGlobalRate t
4238 globalLimited <- checkGlobalRate t
4184 if globalLimited
4239 if globalLimited
4241+ effGlobal <- effectiveGlobalRequestRate
4186− logEvent "warn" "rate_limited_federation_global" ("peer=" ++ fromServer ++ " limit=" ++ show maxGlobalRequestRate)
4242+ logEvent "warn" "rate_limited_federation_global" ("peer=" ++ fromServer ++ " limit=" ++ show effGlobal)
4187− -- Encode retryAfterMs in error string so peers can make smarter retry decisions.
4243+ -- retryAfterMs provides a structured hint so peers can make smarter retry decisions.
4188− -- Format: "rate_limited:NNNms" — peers that don't parse it still see "rate_limited:...".
4244+ -- Error string still includes the value for backward compatibility with pre-v14 peers.
4189− yield (respond {ok: False {}, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
4245+ yield (respond {ok: False {}, error: "rate_limited", retryAfterMs: globalRateWindowMs})
4191 rateLimited <- checkFederationRate fromServer t
4247 rateLimited <- checkFederationRate fromServer t
4192 if rateLimited
4248 if rateLimited
4250+ effFed <- effectiveFederationRate
4194− logEvent "warn" "rate_limited_federation_per_server" ("peer=" ++ fromServer ++ " limit=" ++ show maxFederationRatePerServer)
4251+ logEvent "warn" "rate_limited_federation_per_server" ("peer=" ++ fromServer ++ " limit=" ++ show effFed)
4195− yield (respond {ok: False {}, error: "rate_limited:" ++ show federationRateWindowMs ++ "ms"})
4252+ yield (respond {ok: False {}, error: "rate_limited", retryAfterMs: federationRateWindowMs})
4197 when clearOnAuth (clearFailure fromServer)
4254 when clearOnAuth (clearFailure fromServer)
@@ -5011,6 +5068,10 @@ pruneAndRetry = \t -> do
5011 set *federationLatencyTotal = 0
5068 set *federationLatencyTotal = 0
5012 set *federationLatencyCount = 0
5069 set *federationLatencyCount = 0
5013 set *federationLatencyMax = 0
5070 set *federationLatencyMax = 0
5071+ set *federationLatencyBucket0 = 0
5072+ set *federationLatencyBucket1 = 0
5073+ set *federationLatencyBucket2 = 0
5074+ set *federationLatencyBucket3 = 0
5014 set *slowVerifyCount = 0
5075 set *slowVerifyCount = 0
5015 set *verifyFailCount = 0
5076 set *verifyFailCount = 0
5016 -- Reset rate-limit rejection counters (windowed with federation latency metrics)
5077 -- Reset rate-limit rejection counters (windowed with federation latency metrics)
@@ -5492,8 +5553,9 @@ handleChallenge = \pubkey clientIp respond -> do
5492 set *challengeRates = rateResult.entries
5553 set *challengeRates = rateResult.entries
5493 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
5554 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
5494 yield {accepted: True {}, rateLimited: False {}, retryAfterMs: 0, error: ""}
5555 yield {accepted: True {}, rateLimited: False {}, retryAfterMs: 0, error: ""}
5495− when result.rateLimited
5556+ when result.rateLimited (do
5557+ effChal <- effectiveChallengeRate
5496− (logEvent "warn" "rate_limited_per_key" ("endpoint=challenge key=" ++ pubkey ++ " limit=" ++ show maxChallengeRatePerKey))
5558+ logEvent "warn" "rate_limited_per_key" ("endpoint=challenge key=" ++ pubkey ++ " limit=" ++ show effChal))
5497 when (not result.accepted && not result.rateLimited)
5559 when (not result.accepted && not result.rateLimited)
5498 (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
5560 (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
5499 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0, retryAfterMs: result.retryAfterMs, error: result.error})
5561 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0, retryAfterMs: result.retryAfterMs, error: result.error})
@@ -5516,7 +5578,8 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
5516 set *challengeRates = rateResult.entries
5578 set *challengeRates = rateResult.entries
5517 yield rateResult.limited
5579 yield rateResult.limited
5518 if authRateLimited then do
5580 if authRateLimited then do
5581+ effChal <- effectiveChallengeRate
5519− logEvent "warn" "rate_limited_per_key" ("endpoint=verify key=" ++ pubkey ++ " limit=" ++ show maxChallengeRatePerKey)
5582+ logEvent "warn" "rate_limited_per_key" ("endpoint=verify key=" ++ pubkey ++ " limit=" ++ show effChal)
5520 yield (respond {token: "", expiresAt: 0, error: "rate_limited:" ++ show challengeRateWindowMs ++ "ms"})
5583 yield (respond {token: "", expiresAt: 0, error: "rate_limited:" ++ show challengeRateWindowMs ++ "ms"})
5522 authGlobalLimited <- checkAuthGlobalRate t
5585 authGlobalLimited <- checkAuthGlobalRate t
@@ -5648,10 +5711,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
5648 -- Cursor staleness: cursor is older than message retention, so messages were missed.
5711 -- Cursor staleness: cursor is older than message retention, so messages were missed.
5649 -- Different from cursorReset (server restart): staleness means client was offline too long.
5712 -- Different from cursorReset (server restart): staleness means client was offline too long.
5650 let cursorStale = effectiveCursor > 0 && t - effectiveCursor > unackedMessageMaxAge
5713 let cursorStale = effectiveCursor > 0 && t - effectiveCursor > unackedMessageMaxAge
5651− if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0})
5714+ if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: []})
5653 globalRateLimited <- checkGlobalRate t
5716 globalRateLimited <- checkGlobalRate t
5654− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0})
5717+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: []})
5656 -- Capture presenceVersion before long-poll wait to skip unchanged lookups
5719 -- Capture presenceVersion before long-poll wait to skip unchanged lookups
5657 startPresenceVersion <- *presenceVersion
5720 startPresenceVersion <- *presenceVersion
@@ -5713,10 +5776,20 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
5713 let deResult = drainDropCounter den result.pubkey
5776 let deResult = drainDropCounter den result.pubkey
5714 when (deResult.count > 0) (set *droppedExpirationNotifs = deResult.remaining)
5777 when (deResult.count > 0) (set *droppedExpirationNotifs = deResult.remaining)
5715 yield {dr: drResult.count, de: deResult.count}
5778 yield {dr: drResult.count, de: deResult.count}
5779+ -- Deliver cancel notifications: messages retracted by the sender, delivered to the
5780+ -- recipient so clients can remove them from the UI. Keyed by toKey (recipient pubkey).
5781+ cancellations <- if not result.authorized || result.pubkey == "" then yield []
5783+ cn <- *cancelNotifications
5784+ let forKey = do c <- cn; where c.toKey == result.pubkey; yield {messageId: c.messageId, cancelledAt: c.cancelledAt}
5785+ let remaining = do c <- cn; where c.toKey != result.pubkey; yield c
5786+ set *cancelNotifications = remaining
5787+ let capped = takeRelation maxCancelNotificationsPerKey forKey
5716 -- Expose server load to clients: 0 = healthy (<=80% capacity), up to 90 = near-full.
5789 -- Expose server load to clients: 0 = healthy (<=80% capacity), up to 90 = near-full.
5717 -- Clients can voluntarily reduce poll frequency or batch sends when loadPct > 0.
5790 -- Clients can voluntarily reduce poll frequency or batch sends when loadPct > 0.
5718 loadPct <- do scale <- *cachedAdaptiveScale; yield (if scale >= 100 then 0 else 100 - scale)
5791 loadPct <- do scale <- *cachedAdaptiveScale; yield (if scale >= 100 then 0 else 100 - scale)
5719− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, hasMore: result.hasMore, cursorStale: cursorStale, messageExpirations: expirations, messageDeliveries: deliveries, droppedDeliveryReceipts: drops.dr, droppedExpirationNotifs: drops.de, loadPct: loadPct})
5792+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, hasMore: result.hasMore, cursorStale: cursorStale, messageExpirations: expirations, messageDeliveries: deliveries, droppedDeliveryReceipts: drops.dr, droppedExpirationNotifs: drops.de, loadPct: loadPct, messageCancellations: cancellations})
5721-- Core send logic: dedup, rate limiting, store message, create forwards.
5794-- Core send logic: dedup, rate limiting, store message, create forwards.
5722-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
5795-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
@@ -6163,6 +6236,9 @@ handleLogout = \authorization clientIp respond -> do
6163-- Called by the receiving server during federation callback verification.
6236-- Called by the receiving server during federation callback verification.
6164-- Tokens are single-use and expire after verificationTokenExpiry.
6237-- Tokens are single-use and expire after verificationTokenExpiry.
6165handleVerifyToken = \token scope clientIp respond -> do
6238handleVerifyToken = \token scope clientIp respond -> do
6239+ -- Reject empty scope: prevents matching against any cached verification entry
6240+ if scope == "" then yield (respond {ok: False {}})
6166 t <- now
6242 t <- now
6167 globalRateLimited <- checkGlobalRate t
6243 globalRateLimited <- checkGlobalRate t
6168 if globalRateLimited then yield (respond {ok: False {}})
6244 if globalRateLimited then yield (respond {ok: False {}})
@@ -6236,14 +6312,14 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
6236 yield (False {})
6312 yield (False {})
6237 if eventRateLimited then do
6313 if eventRateLimited then do
6238 logEvent "warn" "rate_limited_gossip_volume" ("peer=" ++ fromServer ++ " limit=" ++ show maxGossipEventsPerServer)
6314 logEvent "warn" "rate_limited_gossip_volume" ("peer=" ++ fromServer ++ " limit=" ++ show maxGossipEventsPerServer)
6239− yield (respond {ok: False {}, error: "event_rate_limited:" ++ show gossipEventRateWindowMs ++ "ms"})
6315+ yield (respond {ok: False {}, error: "event_rate_limited", retryAfterMs: gossipEventRateWindowMs})
6241 let validated = validGossipEvents cappedEvents t
6317 let validated = validGossipEvents cappedEvents t
6242 let onlineEntries = validated.online
6318 let onlineEntries = validated.online
6243 let trustedOffline = trustedOfflineEvents validated.offline fromServer
6319 let trustedOffline = trustedOfflineEvents validated.offline fromServer
6244 -- Early return: skip the expensive presence merge when all events are invalid/expired
6320 -- Early return: skip the expensive presence merge when all events are invalid/expired
6245 if count onlineEntries == 0 && count trustedOffline == 0
6321 if count onlineEntries == 0 && count trustedOffline == 0
6246− then yield (respond {ok: True {}, error: ""})
6322+ then yield (respond {ok: True {}, error: "", retryAfterMs: 0})
6248 -- Warn if a peer sends both online and offline for the same key in one batch (suspicious)
6324 -- Warn if a peer sends both online and offline for the same key in one batch (suspicious)
6249 let conflictingKeys = do on <- onlineEntries; off <- trustedOffline; where on.pubkey == off.pubkey && on.server == off.server; yield on.pubkey
6325 let conflictingKeys = do on <- onlineEntries; off <- trustedOffline; where on.pubkey == off.pubkey && on.server == off.server; yield on.pubkey
@@ -6329,7 +6405,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
6329 when (thirdPartyHops > 1 && count thirdPartyEvents > 0) (sendBatchedGossip targets thirdPartyEvents thirdPartyHops))
6405 when (thirdPartyHops > 1 && count thirdPartyEvents > 0) (sendBatchedGossip targets thirdPartyEvents thirdPartyHops))
6330 when (federationEnabled && count newlyOnlineRemote > 0)
6406 when (federationEnabled && count newlyOnlineRemote > 0)
6331 (forwardQueuedMessages newlyOnlineRemote t))
6407 (forwardQueuedMessages newlyOnlineRemote t))
6332− yield (respond {ok: True {}, error: ""}))
6408+ yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
6334-- POST /federation/forward: Receive a message forwarded from a peer server.
6410-- POST /federation/forward: Receive a message forwarded from a peer server.
6335-- Accepts both direct forwards (fromServer == originServer) and relay forwards
6411-- Accepts both direct forwards (fromServer == originServer) and relay forwards
@@ -6361,13 +6437,13 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
6361 if not originOk
6437 if not originOk
6363 logWarn ("forward: rejected relay with untrusted origin " ++ originServer ++ " from " ++ fromServer)
6439 logWarn ("forward: rejected relay with untrusted origin " ++ originServer ++ " from " ++ fromServer)
6364− yield (respond {ok: False {}, error: "unknown_origin"})
6440+ yield (respond {ok: False {}, error: "unknown_origin", retryAfterMs: 0})
6366 -- Replay protection: reject forwards with timestamps older than messageMaxAge.
6442 -- Replay protection: reject forwards with timestamps older than messageMaxAge.
6367 -- Legitimate queued messages have timestamps within this window. Replayed captures
6443 -- Legitimate queued messages have timestamps within this window. Replayed captures
6368 -- become stale once dedup state (messages + ackedIds + deliveredIds) is pruned (~30 days).
6444 -- become stale once dedup state (messages + ackedIds + deliveredIds) is pruned (~30 days).
6369 if t - timestamp > messageMaxAge
6445 if t - timestamp > messageMaxAge
6370− then yield (respond {ok: False {}, error: "stale_timestamp"})
6446+ then yield (respond {ok: False {}, error: "stale_timestamp", retryAfterMs: 0})
6372 -- Presence check + message storage in a single atomic block to prevent TOCTOU
6448 -- Presence check + message storage in a single atomic block to prevent TOCTOU
6373 -- race where a key goes offline between presence check and store.
6449 -- race where a key goes offline between presence check and store.
@@ -6443,7 +6519,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
6443 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: False {}}
6519 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: False {}}
6444 if result.noPresence then do
6520 if result.noPresence then do
6445 logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
6521 logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
6446− yield (respond {ok: False {}, error: "no_presence"})
6522+ yield (respond {ok: False {}, error: "no_presence", retryAfterMs: 0})
6448 case result.forward of
6524 case result.forward of
6449 ForwardRejected {} ->
6525 ForwardRejected {} ->
@@ -6452,12 +6528,12 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
6452 else if result.globalCapReached
6528 else if result.globalCapReached
6453 then logWarn ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
6529 then logWarn ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
6454 else if result.recvRateLimited
6530 else if result.recvRateLimited
6455− then logEvent "warn" "rate_limited_recv" ("key=" ++ toKey ++ " limit=" ++ show maxRecvRatePerKey)
6531+ then do effRecv <- effectiveRecvRate; logEvent "warn" "rate_limited_recv" ("key=" ++ toKey ++ " limit=" ++ show effRecv)
6456 else logWarn ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
6532 else logWarn ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
6457 _ -> yield {}
6533 _ -> yield {}
6458 yield (case result.forward of
6534 yield (case result.forward of
6459− ForwardRejected {} -> respond {ok: False {}, error: if result.nonceCutoffRejected then "nonce_cutoff" else "mailbox_full"}
6535+ ForwardRejected {} -> respond {ok: False {}, error: if result.nonceCutoffRejected then "nonce_cutoff" else "mailbox_full", retryAfterMs: 0}
6460− _ -> respond {ok: True {}, error: ""}))
6536+ _ -> respond {ok: True {}, error: "", retryAfterMs: 0}))
6462-- Generate sender delivery receipts for messages confirmed delivered via federation ack.
6538-- Generate sender delivery receipts for messages confirmed delivered via federation ack.
6463-- Uses senderKey from the ForwardEntry directly (stored since v12), so receipts are
6539-- Uses senderKey from the ForwardEntry directly (stored since v12), so receipts are
@@ -6509,7 +6585,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
6509 let corrId = sanitizeCorrelationId correlationId
6585 let corrId = sanitizeCorrelationId correlationId
6510 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " ack from " ++ fromServer))
6586 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " ack from " ++ fromServer))
6511 -- Wrap respond to add ackedCount default for withFederationAuth error paths
6587 -- Wrap respond to add ackedCount default for withFederationAuth error paths
6512− let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0, results: []}
6588+ let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0, results: [], retryAfterMs: r.retryAfterMs}
6513 -- clearOnAuth=False: defer clearFailure until we confirm the ack matched actual
6589 -- clearOnAuth=False: defer clearFailure until we confirm the ack matched actual
6514 -- forward entries. Prevents a peer from resetting its failure/backoff counters by
6590 -- forward entries. Prevents a peer from resetting its failure/backoff counters by
6515 -- spamming acks for messageIds we never forwarded to it.
6591 -- spamming acks for messageIds we never forwarded to it.
@@ -6528,7 +6604,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
6529 logWarn ("federation/ack: rejecting ack from " ++ fromServer ++ " — all " ++ show (count dedupedIds) ++ " messageIds unmatched (no forward entries)")
6605 logWarn ("federation/ack: rejecting ack from " ++ fromServer ++ " — all " ++ show (count dedupedIds) ++ " messageIds unmatched (no forward entries)")
6530 logEvent "warn" "ack_all_unmatched" (fromServer ++ " " ++ show (count dedupedIds))
6606 logEvent "warn" "ack_all_unmatched" (fromServer ++ " " ++ show (count dedupedIds))
6531− yield (respond {ok: False {}, error: "no_matching_forwards", ackedCount: 0, results: do mid <- dedupedIds; yield {messageId: mid, found: False {}}})
6607+ yield (respond {ok: False {}, error: "no_matching_forwards", ackedCount: 0, results: do mid <- dedupedIds; yield {messageId: mid, found: False {}}, retryAfterMs: 0})
6533 when (unmatched > 0) (do
6609 when (unmatched > 0) (do
6534 logWarn ("federation/ack: " ++ fromServer ++ " acked " ++ show unmatched ++ " messageIds with no matching forward entries — skipping cleanup for those")
6610 logWarn ("federation/ack: " ++ fromServer ++ " acked " ++ show unmatched ++ " messageIds with no matching forward entries — skipping cleanup for those")
@@ -6567,7 +6643,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
6567 -- Origin servers can drop entries with found=False from their retry queues instead
6643 -- Origin servers can drop entries with found=False from their retry queues instead
6568 -- of retrying indefinitely for messages that were already pruned or never forwarded.
6644 -- of retrying indefinitely for messages that were already pruned or never forwarded.
6569 let perMessageResults = do mid <- dedupedIds; yield {messageId: mid, found: any removedForwards (\f -> f.messageId == mid)}
6645 let perMessageResults = do mid <- dedupedIds; yield {messageId: mid, found: any removedForwards (\f -> f.messageId == mid)}
6570− yield (respond {ok: True {}, error: "", ackedCount: ackedCount, results: perMessageResults}))
6646+ yield (respond {ok: True {}, error: "", ackedCount: ackedCount, results: perMessageResults, retryAfterMs: 0}))
6572-- POST /federation/cancel: Cancel a previously forwarded message.
6648-- POST /federation/cancel: Cancel a previously forwarded message.
6573-- Called by the origin server when the sender retracts a message via /messages/cancel.
6649-- Called by the origin server when the sender retracts a message via /messages/cancel.
@@ -6584,7 +6660,7 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
6584 t <- now
6660 t <- now
6585 let corrId = sanitizeCorrelationId correlationId
6661 let corrId = sanitizeCorrelationId correlationId
6586 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel " ++ messageId ++ " from " ++ fromServer))
6662 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel " ++ messageId ++ " from " ++ fromServer))
6587− let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}}
6663+ let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}, retryAfterMs: r.retryAfterMs}
6588 withFederationAuth authorization t cancelRespond fromServer ("cancel:" ++ messageId) requestTimestamp (not (isLocalServer fromServer)) (False {}) (do
6664 withFederationAuth authorization t cancelRespond fromServer ("cancel:" ++ messageId) requestTimestamp (not (isLocalServer fromServer)) (False {}) (do
6589 result <- atomic do
6665 result <- atomic do
6590 let b = pubkeyBucket toKey
6666 let b = pubkeyBucket toKey
@@ -6615,9 +6691,10 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
6615 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6691 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6616 else updatedNonces)
6692 else updatedNonces)
6617 yield {cancelled: True {}}
6693 yield {cancelled: True {}}
6618− when result.cancelled
6694+ when result.cancelled (do
6619− (logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")"))
6695+ logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")")
6696+ queueCancelNotification toKey messageId t)
6620− yield (respond {ok: True {}, error: "", cancelled: result.cancelled}))
6697+ yield (respond {ok: True {}, error: "", cancelled: result.cancelled, retryAfterMs: 0}))
6622-- POST /federation/cancel-batch: Batch cancel multiple previously forwarded messages.
6699-- POST /federation/cancel-batch: Batch cancel multiple previously forwarded messages.
6623-- Reuses the cancel logic from handleFederationCancel for each entry.
6700-- Reuses the cancel logic from handleFederationCancel for each entry.
@@ -6625,14 +6702,14 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
6625 t <- now
6702 t <- now
6626 let corrId = sanitizeCorrelationId correlationId
6703 let corrId = sanitizeCorrelationId correlationId
6627 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel-batch (" ++ show (count cancels) ++ " entries) from " ++ fromServer))
6704 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " cancel-batch (" ++ show (count cancels) ++ " entries) from " ++ fromServer))
6628− let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: []}
6705+ let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
6629 withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (False {}) (do
6706 withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (False {}) (do
6630 let capped = takeRelation maxBatchSize cancels
6707 let capped = takeRelation maxBatchSize cancels
6631 results <- cancelBatchInner capped fromServer t []
6708 results <- cancelBatchInner capped fromServer t []
6632 let cancelledCount = count (do r <- results; where r.cancelled; yield r)
6709 let cancelledCount = count (do r <- results; where r.cancelled; yield r)
6633 when (cancelledCount > 0)
6710 when (cancelledCount > 0)
6634 (logInfo ("federation/cancel-batch: cancelled " ++ show cancelledCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer))
6711 (logInfo ("federation/cancel-batch: cancelled " ++ show cancelledCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer))
6635− yield (respond {ok: True {}, error: "", results: results}))
6712+ yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
6637-- Process cancel entries grouped by bucket: all entries sharing a pubkeyBucket are
6714-- Process cancel entries grouped by bucket: all entries sharing a pubkeyBucket are
6638-- processed in a single atomic block, reducing bucket reads from O(entries) to O(buckets).
6715-- processed in a single atomic block, reducing bucket reads from O(entries) to O(buckets).
@@ -6665,6 +6742,13 @@ cancelBatchInner = \entries fromServer t acc ->
6665 tfc <- *totalForwardCount
6742 tfc <- *totalForwardCount
6666 set *totalForwardCount = max 0 (tfc - r.fwdRemoved))
6743 set *totalForwardCount = max 0 (tfc - r.fwdRemoved))
6667 yield r.acc
6744 yield r.acc
6745+ -- Queue cancel notifications for successfully cancelled messages in this bucket group
6746+ let cancelled = do r <- results; where r.cancelled; yield r
6747+ let cancelledEntries = do c <- cancelled; e <- sameBucket; where e.messageId == c.messageId; yield e
6748+ case headOf cancelledEntries of
6749+ Nothing {} -> yield {}
6751+ queueCancelNotificationBatch cancelledEntries t
6668 cancelBatchInner otherBuckets fromServer t (union acc results)
6752 cancelBatchInner otherBuckets fromServer t (union acc results)
6670-- Process all cancel entries for a single bucket within one atomic block.
6754-- Process all cancel entries for a single bucket within one atomic block.
@@ -6698,7 +6782,7 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp correlation
6698 t <- now
6782 t <- now
6699 let corrId = sanitizeCorrelationId correlationId
6783 let corrId = sanitizeCorrelationId correlationId
6700 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward-batch (" ++ show (count forwards) ++ " entries) from " ++ fromServer))
6784 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward-batch (" ++ show (count forwards) ++ " entries) from " ++ fromServer))
6701− let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: []}
6785+ let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
6702 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (True {}) (do
6786 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (True {}) (do
6703 let capped = takeRelation maxBatchSize forwards
6787 let capped = takeRelation maxBatchSize forwards
6704 -- Pre-deduplicate (messageId, toKey) pairs: a malicious peer could send 100 entries
6788 -- Pre-deduplicate (messageId, toKey) pairs: a malicious peer could send 100 entries
@@ -6709,7 +6793,7 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp correlation
6709 let storedCount = count (do r <- results; where r.ok; yield r)
6793 let storedCount = count (do r <- results; where r.ok; yield r)
6710 when (storedCount > 0)
6794 when (storedCount > 0)
6711 (logInfo ("federation/forward-batch: stored " ++ show storedCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer))
6795 (logInfo ("federation/forward-batch: stored " ++ show storedCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer))
6712− yield (respond {ok: True {}, error: "", results: results}))
6796+ yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
6714-- Process forward entries grouped by bucket: all entries sharing a pubkeyBucket(toKey) are
6798-- Process forward entries grouped by bucket: all entries sharing a pubkeyBucket(toKey) are
6715-- processed in a single atomic block, reducing bucket reads from O(entries) to O(buckets).
6799-- processed in a single atomic block, reducing bucket reads from O(entries) to O(buckets).
@@ -6918,6 +7002,7 @@ clientCancelBatchInner = \senderKey entries t resultAcc fedAcc ->
6918 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
7002 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6919 else updatedNonces)
7003 else updatedNonces)
6920 yield {cancelled: True {}, cancelServers: allCancelServers}
7004 yield {cancelled: True {}, cancelServers: allCancelServers}
7005+ when cancelResult.cancelled (queueCancelNotification entry.to entry.messageId t)
6921 let newFedAcc = if cancelResult.cancelled && count cancelResult.cancelServers > 0
7006 let newFedAcc = if cancelResult.cancelled && count cancelResult.cancelServers > 0
6922 then union fedAcc [{messageId: entry.messageId, toKey: entry.to, servers: cancelResult.cancelServers}]
7007 then union fedAcc [{messageId: entry.messageId, toKey: entry.to, servers: cancelResult.cancelServers}]
6923 else fedAcc
7008 else fedAcc
@@ -7194,27 +7279,16 @@ handleHealth = \authorization respond -> do
7194 if healthLimited then readCachedHealth respond
7279 if healthLimited then readCachedHealth respond
7196 isAuthed <- verifyMaintenanceAuth authorization
7281 isAuthed <- verifyMaintenanceAuth authorization
7197− -- Unauthenticated callers always get cached results (constant-time, no live computation).
7282+ -- Unauthenticated callers get cached results (constant-time, no live computation).
7198 -- This prevents timing side-channels: an attacker cannot infer server load by measuring
7283 -- This prevents timing side-channels: an attacker cannot infer server load by measuring
7199 -- response time differences between cached (fast) and computed (slow) health checks.
7284 -- response time differences between cached (fast) and computed (slow) health checks.
7200 -- Cache is kept fresh by computeAndCacheHealth called from background prune (~30s).
7285 -- Cache is kept fresh by computeAndCacheHealth called from background prune (~30s).
7201 -- Only authenticated callers trigger live computation + cache refresh.
7286 -- Only authenticated callers trigger live computation + cache refresh.
7287+ -- Subsystem booleans are returned from cache so load balancers and k8s probes can
7288+ -- detect degradation without requiring authentication. The cached reads are all
7289+ -- constant-time (single STM reads) so no timing information is leaked.
7202 if not isAuthed
7290 if not isAuthed
7203− then do
7291+ then readCachedHealth respond
7204− cached <- atomic do
7205− cOk <- *cachedHealthOk
7207− yield {ok: cOk && dp == 0, draining: dp > 0}
7208− yield (respond {ok: cached.ok, version: serverBuildVersion, pruneHealthy: True {},
7209− heartbeatHealthy: True {},
7210− capacityHealthy: True {},
7211− capacityWarning: False {},
7212− federationHealthy: True {},
7213− watchdogHealthy: True {},
7214− nonceHealthy: True {},
7215− forwardQueueHealthy: True {},
7216− draining: cached.draining,
7219 result <- computeAndCacheHealth t
7293 result <- computeAndCacheHealth t
7220 let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk && result.forwardQueueOk
7294 let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.watchdogOk && result.nonceOk && result.forwardQueueOk
@@ -7229,7 +7303,7 @@ handleHealth = \authorization respond -> do
7229 draining: result.isDraining,
7303 draining: result.isDraining,
7230 degradedPeers: result.degradedPeers})
7304 degradedPeers: result.degradedPeers})
7232−zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0, messageBucketMax: 0, messageBucketMin: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: [], pendingCancels: 0, gossipSuccess: 0, gossipFail: 0}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0}}
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}}
7234handleStats = \authorization respond -> do
7308handleStats = \authorization respond -> do
7235 t <- now
7309 t <- now
@@ -7290,6 +7364,10 @@ handleStats = \authorization respond -> do
7290 flt <- *federationLatencyTotal
7364 flt <- *federationLatencyTotal
7291 flc <- *federationLatencyCount
7365 flc <- *federationLatencyCount
7292 flm <- *federationLatencyMax
7366 flm <- *federationLatencyMax
7367+ flb0 <- *federationLatencyBucket0
7368+ flb1 <- *federationLatencyBucket1
7369+ flb2 <- *federationLatencyBucket2
7370+ flb3 <- *federationLatencyBucket3
7293 svc <- *slowVerifyCount
7371 svc <- *slowVerifyCount
7294 vfc <- *verifyFailCount
7372 vfc <- *verifyFailCount
7295 dca <- *drainCompletedAt
7373 dca <- *drainCompletedAt
@@ -7308,7 +7386,7 @@ handleStats = \authorization respond -> do
7308 cslm <- *clientSendLatencyMax
7386 cslm <- *clientSendLatencyMax
7309 gss <- *gossipSendSuccess
7387 gss <- *gossipSendSuccess
7310 gsf <- *gossipSendFail
7388 gsf <- *gossipSendFail
7311− yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, svc: svc, vfc: vfc, dca: dca, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf, arq: arq, parq: parq, cplt: cplt, cplc: cplc, cplm: cplm, cslt: cslt, cslc: cslc, cslm: cslm, gss: gss, gsf: gsf}
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}
7312 -- Compute derived metrics
7390 -- Compute derived metrics
7313 let sessions = core.sc
7391 let sessions = core.sc
7314 let messages = core.tmc
7392 let messages = core.tmc
@@ -7334,7 +7412,7 @@ handleStats = \authorization respond -> do
7334 let sendAvg = if counters.cslc > 0 then counters.cslt / counters.cslc else 0
7412 let sendAvg = if counters.cslc > 0 then counters.cslt / counters.cslc else 0
7335 yield (respond {ok: True {}, version: serverBuildVersion,
7413 yield (respond {ok: True {}, version: serverBuildVersion,
7336 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},
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},
7337− 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},
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}},
7338 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},
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},
7339 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}})
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}})
7340 else yield (respond zeroStats)
7418 else yield (respond zeroStats)
@@ -7349,6 +7427,7 @@ handleStatsPeers = \authorization respond -> do
7350 failures <- *forwardFailures
7428 failures <- *forwardFailures
7351 peerLats <- *peerLatencies
7429 peerLats <- *peerLatencies
7430+ gossipRates <- *gossipEventRates
7352 -- Per-bucket aggregation: compute forward counts without materializing all 16 buckets.
7431 -- Per-bucket aggregation: compute forward counts without materializing all 16 buckets.
7353 -- Reuses computeForwardStats which reads one bucket at a time.
7432 -- Reuses computeForwardStats which reads one bucket at a time.
7354 fwdStats <- computeForwardStats 0 t
7433 fwdStats <- computeForwardStats 0 t
@@ -7357,9 +7436,11 @@ handleStatsPeers = \authorization respond -> do
7357 let peerLatFor = \server -> case headOf (do p <- peerLats; where p.server == server; yield p) of
7436 let peerLatFor = \server -> case headOf (do p <- peerLats; where p.server == server; yield p) of
7358 Nothing {} -> {avg: 0, mx: 0, samples: 0}
7437 Nothing {} -> {avg: 0, mx: 0, samples: 0}
7359 Just {value: p} -> {avg: if p.count > 0 then p.totalMs / p.count else 0, mx: p.maxMs, samples: p.count}
7438 Just {value: p} -> {avg: if p.count > 0 then p.totalMs / p.count else 0, mx: p.maxMs, samples: p.count}
7439+ -- Per-peer gossip event volume: events per minute from gossipEventRates (1-minute sliding window)
7440+ let gossipFor = \server -> fold (\acc r -> if r.server == server then r.count + r.prevCount else acc) 0 gossipRates
7360− let peers = do f <- failures; let lat = peerLatFor f.server in yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoffFor f.failures f.server, retriable: shouldRetryServer f.server t failures, pendingForwards: fwdCountFor f.server, avgLatencyMs: lat.avg, maxLatencyMs: lat.mx, latencySamples: lat.samples}
7441+ let peers = do f <- failures; let lat = peerLatFor f.server in yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoffFor f.failures f.server, retriable: shouldRetryServer f.server t failures, pendingForwards: fwdCountFor f.server, avgLatencyMs: lat.avg, maxLatencyMs: lat.mx, latencySamples: lat.samples, gossipEventsPerMin: gossipFor f.server}
7361 -- Include servers with pending forwards but no failure records
7442 -- Include servers with pending forwards but no failure records
7362− let fwdOnlyPeers = do c <- fwdCounts; where none failures (\f -> f.server == c.server); let lat = peerLatFor c.server in yield {server: c.server, failures: 0, lastFailedAt: 0, firstFailedAt: 0, backoffMs: 0, retriable: True {}, pendingForwards: c.count, avgLatencyMs: lat.avg, maxLatencyMs: lat.mx, latencySamples: lat.samples}
7443+ let fwdOnlyPeers = do c <- fwdCounts; where none failures (\f -> f.server == c.server); let lat = peerLatFor c.server in yield {server: c.server, failures: 0, lastFailedAt: 0, firstFailedAt: 0, backoffMs: 0, retriable: True {}, pendingForwards: c.count, avgLatencyMs: lat.avg, maxLatencyMs: lat.mx, latencySamples: lat.samples, gossipEventsPerMin: gossipFor c.server}
7363 yield (respond {peers: union peers fwdOnlyPeers, ok: True {}})
7444 yield (respond {peers: union peers fwdOnlyPeers, ok: True {}})
7364 else yield (respond {peers: [], ok: False {}})
7445 else yield (respond {peers: [], ok: False {}})
@@ -7737,6 +7818,8 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
7737 yield {cancelled: True {}, cancelServers: allCancelServers, nonceLruEvicted: nonceResult.lruEvicted, nonceOldestAge: nonceResult.oldestRetainedAge}
7818 yield {cancelled: True {}, cancelServers: allCancelServers, nonceLruEvicted: nonceResult.lruEvicted, nonceOldestAge: nonceResult.oldestRetainedAge}
7738 -- Log nonce eviction outside atomic block (println can't run inside STM)
7819 -- Log nonce eviction outside atomic block (println can't run inside STM)
7739 logNonceEviction (pubkeyBucket to) cancelResult.nonceLruEvicted cancelResult.nonceOldestAge
7820 logNonceEviction (pubkeyBucket to) cancelResult.nonceLruEvicted cancelResult.nonceOldestAge
7821+ -- Notify recipient that this message was retracted (delivered via next poll)
7822+ when cancelResult.cancelled (queueCancelNotification to messageId t)
7740 -- Send federation batch cancel to remote servers that may have the message.
7823 -- Send federation batch cancel to remote servers that may have the message.
7741 -- Uses FederationBatchCancel (single-entry batch) instead of FederationCancel
7824 -- Uses FederationBatchCancel (single-entry batch) instead of FederationCancel
7742 -- for code path consistency with handleCancelBatch.
7825 -- for code path consistency with handleCancelBatch.
@@ -8414,8 +8497,8 @@ serveOverloaded = \req -> case req of
8414 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}
8497 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}
8415 Logout {respond} -> respond {ok: False {}, authorized: False {}}
8498 Logout {respond} -> respond {ok: False {}, authorized: False {}}
8416 -- Client routes
8499 -- Client routes
8417− Poll {respond} -> respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: 0, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 1000, cursorReset: False {}, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0}
8500+ Poll {respond} -> respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: 0, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 1000, cursorReset: False {}, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0}
8418− SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded"}
8501+ SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded", loadPct: 0}
8419 MessageStatus {respond} -> respond {status: "", authorized: False {}}
8502 MessageStatus {respond} -> respond {status: "", authorized: False {}}
8420 AckMessages {respond} -> respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "overloaded"}
8503 AckMessages {respond} -> respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "overloaded"}
8421 SendBatch {respond} -> respond {results: [], authorized: False {}, retryAfterMs: 1000}
8504 SendBatch {respond} -> respond {results: [], authorized: False {}, retryAfterMs: 1000}
@@ -8426,12 +8509,12 @@ serveOverloaded = \req -> case req of
8426 LookupPresence {respond} -> respond {server: "", authorized: False {}, error: "overloaded"}
8509 LookupPresence {respond} -> respond {server: "", authorized: False {}, error: "overloaded"}
8427 BatchLookupPresence {respond} -> respond {results: [], authorized: False {}, error: "overloaded"}
8510 BatchLookupPresence {respond} -> respond {results: [], authorized: False {}, error: "overloaded"}
8428 -- Federation routes
8511 -- Federation routes
8429− RecvGossip {respond} -> respond {ok: False {}, error: "overloaded:1000ms"}
8512+ RecvGossip {respond} -> respond {ok: False {}, error: "overloaded", retryAfterMs: 1000}
8430− ForwardMessage {respond} -> respond {ok: False {}, error: "overloaded:1000ms"}
8513+ ForwardMessage {respond} -> respond {ok: False {}, error: "overloaded", retryAfterMs: 1000}
8431− FederationAck {respond} -> respond {ok: False {}, error: "overloaded:1000ms", ackedCount: 0, results: []}
8514+ FederationAck {respond} -> respond {ok: False {}, error: "overloaded", ackedCount: 0, results: [], retryAfterMs: 1000}
8432− FederationCancel {respond} -> respond {ok: False {}, error: "overloaded:1000ms", cancelled: False {}}
8515+ FederationCancel {respond} -> respond {ok: False {}, error: "overloaded", cancelled: False {}, retryAfterMs: 1000}
8433− FederationBatchCancel {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}
8516+ FederationBatchCancel {respond} -> respond {ok: False {}, error: "overloaded", results: [], retryAfterMs: 1000}
8434− FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}
8517+ FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded", results: [], retryAfterMs: 1000}
8435 VerifyToken {respond} -> respond {ok: False {}}
8518 VerifyToken {respond} -> respond {ok: False {}}
8436 FederationPubkey {respond} -> respond {pubkey: "", ok: False {}}
8519 FederationPubkey {respond} -> respond {pubkey: "", ok: False {}}
8437 -- Stats/admin routes (exempt routes handled by isOverloadExempt; these are fallbacks)
8520 -- Stats/admin routes (exempt routes handled by isOverloadExempt; these are fallbacks)
@@ -8828,9 +8911,12 @@ main = do
8828 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"
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"
8830 t <- now
8913 t <- now
8914+ -- Mix random bits into generation to prevent sub-millisecond restart collisions.
8915+ -- Clients use generation to detect server restarts; same generation = stale cursor.
8916+ genJitter <- randomInt 1000000
8831 atomic do
8917 atomic do
8832 set *startedAt = t
8918 set *startedAt = t
8833− set *serverGeneration = t
8919+ set *serverGeneration = t * 1000 + genJitter
8834 -- Empty maintenanceSecret is the recommended deployment path: use /maintenance/rotate-secret
8920 -- Empty maintenanceSecret is the recommended deployment path: use /maintenance/rotate-secret
8835 -- at startup via deployment scripts for zero-knowledge secret management. Log at INFO level
8921 -- at startup via deployment scripts for zero-knowledge secret management. Log at INFO level
8836 -- since operators following the recommended path shouldn't see warnings on every startup.
8922 -- since operators following the recommended path shouldn't see warnings on every startup.