@@ -195,15 +195,24 @@ lightReconciliationInterval = 5
195195federationLatencyWindowMs : Int<Ms>
196196federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
197197
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)
206+
198207-- Server build version: included in /health and /stats responses for deployment verification.
199208-- 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"
201210
202211-- Federation protocol version: included in gossip/forward/ack requests.
203212-- Peers log warnings when they receive requests from older versions.
204213-- Backward compatible: older peers ignore unknown fields (Knot's JSON parser
205214-- drops unknown keys). 0 = pre-versioning peer (field missing or unset).
206−federationProtocolVersion = 13
215+federationProtocolVersion = 14
207216
208217-- Max seenBy list length in gossip events (bounds memory and bandwidth per event)
209218maxSeenByLen = 20
@@ -317,6 +326,10 @@ maxSenderExpirationsPerKey = 100
317326-- so senders can learn their message was delivered (returned in poll response).
318327maxSenderDeliveries = 10000
319328maxSenderDeliveriesPerKey = 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
320333-- Max pubkeys in a batch lookup request
321334maxBatchLookupKeys = 50
322335-- 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
465478type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int, prevCount: Int}
466479-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles
467480type 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}
469482type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
470483type RecipientCount = {pubkey: PubkeyHex, count: Int}
471484type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
@@ -499,6 +512,11 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
499512-- Only for locally-originated messages (senderKey != ""); federated messages have no senderKey.
500513type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
501514
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}
519+
502520-- ============================================
503521-- Persisted State
504522-- ============================================
@@ -777,6 +795,13 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
777795*federationLatencyCount : Int
778796*federationLatencyMax : Int<Ms>
779797
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
804+
780805-- Federation verify callback metrics (exposed in /stats, reset with latency window)
781806*slowVerifyCount : Int
782807*verifyFailCount : Int
@@ -885,6 +910,10 @@ maxPeerLatencyEntries = 100
885910-- Populated during ackPolledMessages and handleAckMessages for locally-originated messages.
886911*senderDeliveries : [SenderDelivery]
887912
913+-- Cancel notifications: messages retracted by the sender, queued for the recipient's next poll.
914+-- Populated during handleCancelMessage, handleFederationCancel, handleFederationBatchCancel.
915+*cancelNotifications : [CancelNotification]
916+
888917-- Pending federation cancels: queued when cancel delivery to a remote server fails.
889918-- Retried during the prune cycle alongside forward retries and ack retries.
890919*pendingFederationCancels : [PendingFederationCancel]
@@ -995,7 +1024,7 @@ route AuthApi where
9951024route ClientApi where
9961025 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
9971026 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}]}
9991028 = Poll
10001029 /messages
10011030 -- Send an E2E encrypted message to a recipient pubkey.
@@ -1057,7 +1086,7 @@ route FederationApi where
10571086 /federation
10581087 -- Receive presence gossip events from a peer server
10591088 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}
10611090 = RecvGossip
10621091 -- Receive a forwarded message from a peer server on behalf of a sender.
10631092 -- NOTE: senderKey is intentionally NOT included in the forward protocol. The sender's
@@ -1066,11 +1095,11 @@ route FederationApi where
10661095 -- federated messages requires the client to include sender identity in the encrypted
10671096 -- payload; the server cannot verify sender identity for federation-forwarded messages.
10681097 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}
10701099 = ForwardMessage
10711100 -- Acknowledge delivery of messages back to the origin server
10721101 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}
10741103 = FederationAck
10751104 -- Cancel (retract) a previously forwarded message. The origin server sends this
10761105 -- when the sender cancels a message via /messages/cancel. The receiving server
@@ -1081,17 +1110,17 @@ route FederationApi where
10811110 -- would go to the wrong server, which wouldn't have the token. This prevents a
10821111 -- malicious relay from cancelling messages it forwarded on behalf of the origin.
10831112 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}
10851114 = FederationCancel
10861115 -- Batch cancel: cancel multiple previously forwarded messages in one request.
10871116 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}
10891118 = FederationBatchCancel
10901119 -- Batch forward: forward multiple messages in one request (reduces HTTP round-trips).
10911120 -- Each message is processed independently; partial success is possible.
10921121 -- Results are returned per-message so the sender knows which forwards succeeded.
10931122 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}
10951124 = FederationBatchForward
10961125
10971126-- Federation callback verification: the receiving server calls back to the sender
@@ -1213,7 +1242,7 @@ route HealthApi where
12131242-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
12141243route StatsApi where
12151244 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}}
12171246 = Stats
12181247 GET /stats/peers headers {authorization: Text}
12191248 -> {peers: [PeerStatus], ok: Bool}
@@ -1350,6 +1379,30 @@ drainDropCounter = \entries key ->
13501379 case headOf (do e <- entries; where e.senderKey == key; yield e) of
13511380 Nothing {} -> {count: 0, remaining: entries}
13521381 Just {value: e} -> {count: e.count, remaining: do x <- entries; where x.senderKey != key; yield x}
1382+
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 {}
1391+ else do
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)
1395+ else updated
1396+
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
1405+
13531406-- NOTE: fold always traverses the full list even after finding the first element.
13541407-- This is O(n) where O(1) suffices, but Knot lacks list destructuring (case xs of
13551408-- (x : rest) -> ...) that would allow short-circuit. Used pervasively on hot paths
@@ -1722,7 +1775,7 @@ generateFederationAuth : Text -> Timestamp -> IO {network} Text
17221775generateFederationAuth = \scope requestTimestamp ->
17231776 if serverFederationPrivateKeyHex != ""
17241777 then do
1725− nonce <- genTokenN 2
1778+ nonce <- genTokenN 4
17261779 let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN ++ "|" ++ nonce)
17271780 let sig = sign (bytesFromHex serverFederationPrivateKeyHex) message
17281781 yield ("Signed " ++ bytesToHex sig ++ ":" ++ nonce)
@@ -2496,26 +2549,20 @@ countByRecipient = \items ->
24962549 let sorted = sortBy (\m -> m.toKey) items in
24972550 countSortedGroups sorted
24982551
2499−-- Linear scan over sorted items: counts consecutive runs of the same toKey.
2500−-- Tail-recursive (TCO-safe in Knot). Each step either increments the current
2501−-- group's count or emits the group and starts a new one.
2502−-- Uses messageId for element removal (unique per bucket) instead of full structural
2503−-- equality, avoiding expensive per-field comparison on Message records with large blobs.
2552+-- Single-pass fold over sorted items: counts consecutive runs of the same toKey.
2553+-- O(n) total: one fold step per item, O(1) prepend per group boundary.
2554+-- Replaces the previous headOf+filter pattern which was O(n) per step = O(n²) total.
2555+-- Uses "union [new] acc" (prepend, O(1)) instead of "union acc [new]" (copy, O(k)).
25042556countSortedGroups = \items ->
2505− case headOf items of
2506− Nothing {} -> []
2507− Just {value: first} ->
2508− let rest = do x <- items; where x.messageId != first.messageId; yield x in
2509− countSortedGroupsInner rest first.toKey 1 []
2510−
2511−countSortedGroupsInner = \items currentKey currentCount acc ->
2512− case headOf items of
2513− Nothing {} -> union acc [{pubkey: currentKey, count: currentCount}]
2514− Just {value: x} ->
2515− let rest = do r <- items; where r.messageId != x.messageId; yield r in
2516− if x.toKey == currentKey
2517− then countSortedGroupsInner rest currentKey (currentCount + 1) acc
2518− else countSortedGroupsInner rest x.toKey 1 (union acc [{pubkey: currentKey, count: currentCount}])
2557+ let result = fold (\acc m ->
2558+ if acc.currentKey == ""
2559+ then {currentKey: m.toKey, currentCount: 1, groups: acc.groups}
2560+ else if m.toKey == acc.currentKey
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
2565+ else union [{pubkey: result.currentKey, count: result.currentCount}] result.groups
25192566
25202567-- Batch-decrement recipient counts: subtract per-key removal counts in a single pass
25212568-- instead of calling adjustRecipientCount in a loop.
@@ -3206,6 +3253,11 @@ fetchFederation = \server request -> do
32063253 set *federationLatencyTotal = total + latency
32073254 set *federationLatencyCount = cnt + 1
32083255 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
32093261 -- Per-peer latency tracking
32103262 peerLats <- *peerLatencies
32113263 let peerEntry = headOf (do p <- peerLats; where p.server == server; yield p)
@@ -3749,6 +3801,9 @@ pruneExpired = \t -> do
37493801 -- Prune old sender delivery receipts (same retention as expirations)
37503802 senderDel <- *senderDeliveries
37513803 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
37523807 -- Reconcile global federation connection counter with actual in-flight state
37533808 -- (self-heals drift from leaked slots where a forked task died without decrementing)
37543809 inFlightRec <- *federationInFlight
@@ -4130,13 +4185,13 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
41304185 if isBlockedPeer fromServer blocked
41314186 then do
41324187 logWarn ("federation: blocked peer " ++ fromServer ++ " rejected")
4133− yield (respond {ok: False {}, error: "blocked"})
4188+ yield (respond {ok: False {}, error: "blocked", retryAfterMs: 0})
41344189 else do
41354190 allowed <- *allowedPeers
41364191 if not (isAllowedByAllowlist fromServer allowed)
41374192 then do
41384193 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})
41404195 else do
41414196 -- Dispatch on auth scheme: "Signed " = Ed25519 signature, "Bearer " = verify callback
41424197 -- New format "Signed <sig>:<nonce>" includes a random nonce to prevent replay of
@@ -4144,7 +4199,7 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
41444199 verified <- if take 7 authorization == "Signed "
41454200 then do
41464201 let payload = drop 7 authorization
4147− if payload == "" || tooLong payload (maxSignatureLen + 13)
4202+ if payload == "" || tooLong payload (maxSignatureLen + 25)
41484203 then yield (False {})
41494204 else
41504205 let parts = splitOnColon payload in
@@ -4153,15 +4208,15 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
41534208 else verifyFederationSignature fromServer payload scope requestTimestamp
41544209 else verifyFederationCallback fromServer authorization scope
41554210 if not verified
4156− then yield (respond {ok: False {}, error: "unauthorized"})
4211+ then yield (respond {ok: False {}, error: "unauthorized", retryAfterMs: 0})
41574212 -- Replay protection: reject requests with timestamps too far from local clock.
41584213 -- requestTimestamp == 0 means the peer doesn't support this field yet (accept).
41594214 else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance)
41604215 then do
41614216 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})
41634218 else if not precondition
4164− then yield (respond {ok: False {}, error: "invalid_request"})
4219+ then yield (respond {ok: False {}, error: "invalid_request", retryAfterMs: 0})
41654220 else do
41664221 -- Replay detection: reject duplicate (server, timestamp) pairs within the skew window
41674222 replayDetected <- if requestTimestamp > 0
@@ -4178,21 +4233,23 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
41784233 if replayDetected
41794234 then do
41804235 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})
41824237 else do
41834238 globalLimited <- checkGlobalRate t
41844239 if globalLimited
41854240 then do
4186− logEvent "warn" "rate_limited_federation_global" ("peer=" ++ fromServer ++ " limit=" ++ show maxGlobalRequestRate)
4187− -- Encode retryAfterMs in error string so peers can make smarter retry decisions.
4188− -- Format: "rate_limited:NNNms" — peers that don't parse it still see "rate_limited:...".
4189− yield (respond {ok: False {}, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
4241+ effGlobal <- effectiveGlobalRequestRate
4242+ logEvent "warn" "rate_limited_federation_global" ("peer=" ++ fromServer ++ " limit=" ++ show effGlobal)
4243+ -- retryAfterMs provides a structured hint so peers can make smarter retry decisions.
4244+ -- Error string still includes the value for backward compatibility with pre-v14 peers.
4245+ yield (respond {ok: False {}, error: "rate_limited", retryAfterMs: globalRateWindowMs})
41904246 else do
41914247 rateLimited <- checkFederationRate fromServer t
41924248 if rateLimited
41934249 then do
4194− logEvent "warn" "rate_limited_federation_per_server" ("peer=" ++ fromServer ++ " limit=" ++ show maxFederationRatePerServer)
4195− yield (respond {ok: False {}, error: "rate_limited:" ++ show federationRateWindowMs ++ "ms"})
4250+ effFed <- effectiveFederationRate
4251+ logEvent "warn" "rate_limited_federation_per_server" ("peer=" ++ fromServer ++ " limit=" ++ show effFed)
4252+ yield (respond {ok: False {}, error: "rate_limited", retryAfterMs: federationRateWindowMs})
41964253 else do
41974254 when clearOnAuth (clearFailure fromServer)
41984255 action
@@ -5011,6 +5068,10 @@ pruneAndRetry = \t -> do
50115068 set *federationLatencyTotal = 0
50125069 set *federationLatencyCount = 0
50135070 set *federationLatencyMax = 0
5071+ set *federationLatencyBucket0 = 0
5072+ set *federationLatencyBucket1 = 0
5073+ set *federationLatencyBucket2 = 0
5074+ set *federationLatencyBucket3 = 0
50145075 set *slowVerifyCount = 0
50155076 set *verifyFailCount = 0
50165077 -- Reset rate-limit rejection counters (windowed with federation latency metrics)
@@ -5492,8 +5553,9 @@ handleChallenge = \pubkey clientIp respond -> do
54925553 set *challengeRates = rateResult.entries
54935554 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
54945555 yield {accepted: True {}, rateLimited: False {}, retryAfterMs: 0, error: ""}
5495− when result.rateLimited
5496− (logEvent "warn" "rate_limited_per_key" ("endpoint=challenge key=" ++ pubkey ++ " limit=" ++ show maxChallengeRatePerKey))
5556+ when result.rateLimited (do
5557+ effChal <- effectiveChallengeRate
5558+ logEvent "warn" "rate_limited_per_key" ("endpoint=challenge key=" ++ pubkey ++ " limit=" ++ show effChal))
54975559 when (not result.accepted && not result.rateLimited)
54985560 (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
54995561 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 ->
55165578 set *challengeRates = rateResult.entries
55175579 yield rateResult.limited
55185580 if authRateLimited then do
5519− logEvent "warn" "rate_limited_per_key" ("endpoint=verify key=" ++ pubkey ++ " limit=" ++ show maxChallengeRatePerKey)
5581+ effChal <- effectiveChallengeRate
5582+ logEvent "warn" "rate_limited_per_key" ("endpoint=verify key=" ++ pubkey ++ " limit=" ++ show effChal)
55205583 yield (respond {token: "", expiresAt: 0, error: "rate_limited:" ++ show challengeRateWindowMs ++ "ms"})
55215584 else do
55225585 authGlobalLimited <- checkAuthGlobalRate t
@@ -5648,10 +5711,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
56485711 -- Cursor staleness: cursor is older than message retention, so messages were missed.
56495712 -- Different from cursorReset (server restart): staleness means client was offline too long.
56505713 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: []})
56525715 else do
56535716 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: []})
56555718 else do
56565719 -- Capture presenceVersion before long-poll wait to skip unchanged lookups
56575720 startPresenceVersion <- *presenceVersion
@@ -5713,10 +5776,20 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
57135776 let deResult = drainDropCounter den result.pubkey
57145777 when (deResult.count > 0) (set *droppedExpirationNotifs = deResult.remaining)
57155778 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 []
5782+ else atomic do
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
5788+ yield capped
57165789 -- Expose server load to clients: 0 = healthy (<=80% capacity), up to 90 = near-full.
57175790 -- Clients can voluntarily reduce poll frequency or batch sends when loadPct > 0.
57185791 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})
57205793
57215794-- Core send logic: dedup, rate limiting, store message, create forwards.
57225795-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
@@ -6163,6 +6236,9 @@ handleLogout = \authorization clientIp respond -> do
61636236-- Called by the receiving server during federation callback verification.
61646237-- Tokens are single-use and expire after verificationTokenExpiry.
61656238handleVerifyToken = \token scope clientIp respond -> do
6239+ -- Reject empty scope: prevents matching against any cached verification entry
6240+ if scope == "" then yield (respond {ok: False {}})
6241+ else do
61666242 t <- now
61676243 globalRateLimited <- checkGlobalRate t
61686244 if globalRateLimited then yield (respond {ok: False {}})
@@ -6236,14 +6312,14 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
62366312 yield (False {})
62376313 if eventRateLimited then do
62386314 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})
62406316 else do
62416317 let validated = validGossipEvents cappedEvents t
62426318 let onlineEntries = validated.online
62436319 let trustedOffline = trustedOfflineEvents validated.offline fromServer
62446320 -- Early return: skip the expensive presence merge when all events are invalid/expired
62456321 if count onlineEntries == 0 && count trustedOffline == 0
6246− then yield (respond {ok: True {}, error: ""})
6322+ then yield (respond {ok: True {}, error: "", retryAfterMs: 0})
62476323 else do
62486324 -- Warn if a peer sends both online and offline for the same key in one batch (suspicious)
62496325 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
63296405 when (thirdPartyHops > 1 && count thirdPartyEvents > 0) (sendBatchedGossip targets thirdPartyEvents thirdPartyHops))
63306406 when (federationEnabled && count newlyOnlineRemote > 0)
63316407 (forwardQueuedMessages newlyOnlineRemote t))
6332− yield (respond {ok: True {}, error: ""}))
6408+ yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
63336409
63346410-- POST /federation/forward: Receive a message forwarded from a peer server.
63356411-- Accepts both direct forwards (fromServer == originServer) and relay forwards
@@ -6361,13 +6437,13 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
63616437 if not originOk
63626438 then do
63636439 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})
63656441 else do
63666442 -- Replay protection: reject forwards with timestamps older than messageMaxAge.
63676443 -- Legitimate queued messages have timestamps within this window. Replayed captures
63686444 -- become stale once dedup state (messages + ackedIds + deliveredIds) is pruned (~30 days).
63696445 if t - timestamp > messageMaxAge
6370− then yield (respond {ok: False {}, error: "stale_timestamp"})
6446+ then yield (respond {ok: False {}, error: "stale_timestamp", retryAfterMs: 0})
63716447 else do
63726448 -- Presence check + message storage in a single atomic block to prevent TOCTOU
63736449 -- race where a key goes offline between presence check and store.
@@ -6443,7 +6519,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
64436519 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}, nonceCutoffRejected: False {}}
64446520 if result.noPresence then do
64456521 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})
64476523 else do
64486524 case result.forward of
64496525 ForwardRejected {} ->
@@ -6452,12 +6528,12 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
64526528 else if result.globalCapReached
64536529 then logWarn ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
64546530 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)
64566532 else logWarn ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
64576533 _ -> yield {}
64586534 yield (case result.forward of
6459− ForwardRejected {} -> respond {ok: False {}, error: if result.nonceCutoffRejected then "nonce_cutoff" else "mailbox_full"}
6460− _ -> respond {ok: True {}, error: ""}))
6535+ ForwardRejected {} -> respond {ok: False {}, error: if result.nonceCutoffRejected then "nonce_cutoff" else "mailbox_full", retryAfterMs: 0}
6536+ _ -> respond {ok: True {}, error: "", retryAfterMs: 0}))
64616537
64626538-- Generate sender delivery receipts for messages confirmed delivered via federation ack.
64636539-- Uses senderKey from the ForwardEntry directly (stored since v12), so receipts are
@@ -6509,7 +6585,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
65096585 let corrId = sanitizeCorrelationId correlationId
65106586 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " ack from " ++ fromServer))
65116587 -- 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}
65136589 -- clearOnAuth=False: defer clearFailure until we confirm the ack matched actual
65146590 -- forward entries. Prevents a peer from resetting its failure/backoff counters by
65156591 -- spamming acks for messageIds we never forwarded to it.
@@ -6528,7 +6604,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
65286604 then do
65296605 logWarn ("federation/ack: rejecting ack from " ++ fromServer ++ " — all " ++ show (count dedupedIds) ++ " messageIds unmatched (no forward entries)")
65306606 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})
65326608 else do
65336609 when (unmatched > 0) (do
65346610 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
65676643 -- Origin servers can drop entries with found=False from their retry queues instead
65686644 -- of retrying indefinitely for messages that were already pruned or never forwarded.
65696645 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}))
65716647
65726648-- POST /federation/cancel: Cancel a previously forwarded message.
65736649-- Called by the origin server when the sender retracts a message via /messages/cancel.
@@ -6584,7 +6660,7 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
65846660 t <- now
65856661 let corrId = sanitizeCorrelationId correlationId
65866662 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}
65886664 withFederationAuth authorization t cancelRespond fromServer ("cancel:" ++ messageId) requestTimestamp (not (isLocalServer fromServer)) (False {}) (do
65896665 result <- atomic do
65906666 let b = pubkeyBucket toKey
@@ -6615,9 +6691,10 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
66156691 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
66166692 else updatedNonces)
66176693 yield {cancelled: True {}}
6618− when result.cancelled
6619− (logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")"))
6620− yield (respond {ok: True {}, error: "", cancelled: result.cancelled}))
6694+ when result.cancelled (do
6695+ logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")")
6696+ queueCancelNotification toKey messageId t)
6697+ yield (respond {ok: True {}, error: "", cancelled: result.cancelled, retryAfterMs: 0}))
66216698
66226699-- POST /federation/cancel-batch: Batch cancel multiple previously forwarded messages.
66236700-- Reuses the cancel logic from handleFederationCancel for each entry.
@@ -6625,14 +6702,14 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
66256702 t <- now
66266703 let corrId = sanitizeCorrelationId correlationId
66276704 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}
66296706 withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (False {}) (do
66306707 let capped = takeRelation maxBatchSize cancels
66316708 results <- cancelBatchInner capped fromServer t []
66326709 let cancelledCount = count (do r <- results; where r.cancelled; yield r)
66336710 when (cancelledCount > 0)
66346711 (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}))
66366713
66376714-- Process cancel entries grouped by bucket: all entries sharing a pubkeyBucket are
66386715-- processed in a single atomic block, reducing bucket reads from O(entries) to O(buckets).
@@ -6665,6 +6742,13 @@ cancelBatchInner = \entries fromServer t acc ->
66656742 tfc <- *totalForwardCount
66666743 set *totalForwardCount = max 0 (tfc - r.fwdRemoved))
66676744 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 {}
6750+ Just {} -> do
6751+ queueCancelNotificationBatch cancelledEntries t
66686752 cancelBatchInner otherBuckets fromServer t (union acc results)
66696753
66706754-- Process all cancel entries for a single bucket within one atomic block.
@@ -6698,7 +6782,7 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp correlation
66986782 t <- now
66996783 let corrId = sanitizeCorrelationId correlationId
67006784 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}
67026786 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (True {}) (do
67036787 let capped = takeRelation maxBatchSize forwards
67046788 -- Pre-deduplicate (messageId, toKey) pairs: a malicious peer could send 100 entries
@@ -6709,7 +6793,7 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp correlation
67096793 let storedCount = count (do r <- results; where r.ok; yield r)
67106794 when (storedCount > 0)
67116795 (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}))
67136797
67146798-- Process forward entries grouped by bucket: all entries sharing a pubkeyBucket(toKey) are
67156799-- 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 ->
69187002 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
69197003 else updatedNonces)
69207004 yield {cancelled: True {}, cancelServers: allCancelServers}
7005+ when cancelResult.cancelled (queueCancelNotification entry.to entry.messageId t)
69217006 let newFedAcc = if cancelResult.cancelled && count cancelResult.cancelServers > 0
69227007 then union fedAcc [{messageId: entry.messageId, toKey: entry.to, servers: cancelResult.cancelServers}]
69237008 else fedAcc
@@ -7194,27 +7279,16 @@ handleHealth = \authorization respond -> do
71947279 if healthLimited then readCachedHealth respond
71957280 else do
71967281 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).
71987283 -- This prevents timing side-channels: an attacker cannot infer server load by measuring
71997284 -- response time differences between cached (fast) and computed (slow) health checks.
72007285 -- Cache is kept fresh by computeAndCacheHealth called from background prune (~30s).
72017286 -- 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.
72027290 if not isAuthed
7203− then do
7204− cached <- atomic do
7205− cOk <- *cachedHealthOk
7206− dp <- *drainPhase
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,
7217− degradedPeers: 0})
7291+ then readCachedHealth respond
72187292 else do
72197293 result <- computeAndCacheHealth t
72207294 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
72297303 draining: result.isDraining,
72307304 degradedPeers: result.degradedPeers})
72317305
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}}
72337307
72347308handleStats = \authorization respond -> do
72357309 t <- now
@@ -7290,6 +7364,10 @@ handleStats = \authorization respond -> do
72907364 flt <- *federationLatencyTotal
72917365 flc <- *federationLatencyCount
72927366 flm <- *federationLatencyMax
7367+ flb0 <- *federationLatencyBucket0
7368+ flb1 <- *federationLatencyBucket1
7369+ flb2 <- *federationLatencyBucket2
7370+ flb3 <- *federationLatencyBucket3
72937371 svc <- *slowVerifyCount
72947372 vfc <- *verifyFailCount
72957373 dca <- *drainCompletedAt
@@ -7308,7 +7386,7 @@ handleStats = \authorization respond -> do
73087386 cslm <- *clientSendLatencyMax
73097387 gss <- *gossipSendSuccess
73107388 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}
73127390 -- Compute derived metrics
73137391 let sessions = core.sc
73147392 let messages = core.tmc
@@ -7334,7 +7412,7 @@ handleStats = \authorization respond -> do
73347412 let sendAvg = if counters.cslc > 0 then counters.cslt / counters.cslc else 0
73357413 yield (respond {ok: True {}, version: serverBuildVersion,
73367414 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}},
73387416 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},
73397417 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}})
73407418 else yield (respond zeroStats)
@@ -7349,6 +7427,7 @@ handleStatsPeers = \authorization respond -> do
73497427 then do
73507428 failures <- *forwardFailures
73517429 peerLats <- *peerLatencies
7430+ gossipRates <- *gossipEventRates
73527431 -- Per-bucket aggregation: compute forward counts without materializing all 16 buckets.
73537432 -- Reuses computeForwardStats which reads one bucket at a time.
73547433 fwdStats <- computeForwardStats 0 t
@@ -7357,9 +7436,11 @@ handleStatsPeers = \authorization respond -> do
73577436 let peerLatFor = \server -> case headOf (do p <- peerLats; where p.server == server; yield p) of
73587437 Nothing {} -> {avg: 0, mx: 0, samples: 0}
73597438 Just {value: p} -> {avg: if p.count > 0 then p.totalMs / p.count else 0, mx: p.maxMs, samples: p.count}
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}
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
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}
73617442 -- 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}
73637444 yield (respond {peers: union peers fwdOnlyPeers, ok: True {}})
73647445 else yield (respond {peers: [], ok: False {}})
73657446
@@ -7737,6 +7818,8 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
77377818 yield {cancelled: True {}, cancelServers: allCancelServers, nonceLruEvicted: nonceResult.lruEvicted, nonceOldestAge: nonceResult.oldestRetainedAge}
77387819 -- Log nonce eviction outside atomic block (println can't run inside STM)
77397820 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)
77407823 -- Send federation batch cancel to remote servers that may have the message.
77417824 -- Uses FederationBatchCancel (single-entry batch) instead of FederationCancel
77427825 -- for code path consistency with handleCancelBatch.
@@ -8414,8 +8497,8 @@ serveOverloaded = \req -> case req of
84148497 RevokeSession {respond} -> respond {ok: False {}, authorized: False {}}
84158498 Logout {respond} -> respond {ok: False {}, authorized: False {}}
84168499 -- 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}
8418− SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded"}
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}
8501+ SendMessage {respond} -> respond {status: MsRejected {}, messageId: "", retryAfterMs: 1000, error: "overloaded", loadPct: 0}
84198502 MessageStatus {respond} -> respond {status: "", authorized: False {}}
84208503 AckMessages {respond} -> respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "overloaded"}
84218504 SendBatch {respond} -> respond {results: [], authorized: False {}, retryAfterMs: 1000}
@@ -8426,12 +8509,12 @@ serveOverloaded = \req -> case req of
84268509 LookupPresence {respond} -> respond {server: "", authorized: False {}, error: "overloaded"}
84278510 BatchLookupPresence {respond} -> respond {results: [], authorized: False {}, error: "overloaded"}
84288511 -- Federation routes
8429− RecvGossip {respond} -> respond {ok: False {}, error: "overloaded:1000ms"}
8430− ForwardMessage {respond} -> respond {ok: False {}, error: "overloaded:1000ms"}
8431− FederationAck {respond} -> respond {ok: False {}, error: "overloaded:1000ms", ackedCount: 0, results: []}
8432− FederationCancel {respond} -> respond {ok: False {}, error: "overloaded:1000ms", cancelled: False {}}
8433− FederationBatchCancel {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}
8434− FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded:1000ms", results: []}
8512+ RecvGossip {respond} -> respond {ok: False {}, error: "overloaded", retryAfterMs: 1000}
8513+ ForwardMessage {respond} -> respond {ok: False {}, error: "overloaded", retryAfterMs: 1000}
8514+ FederationAck {respond} -> respond {ok: False {}, error: "overloaded", ackedCount: 0, results: [], retryAfterMs: 1000}
8515+ FederationCancel {respond} -> respond {ok: False {}, error: "overloaded", cancelled: False {}, retryAfterMs: 1000}
8516+ FederationBatchCancel {respond} -> respond {ok: False {}, error: "overloaded", results: [], retryAfterMs: 1000}
8517+ FederationBatchForward {respond} -> respond {ok: False {}, error: "overloaded", results: [], retryAfterMs: 1000}
84358518 VerifyToken {respond} -> respond {ok: False {}}
84368519 FederationPubkey {respond} -> respond {pubkey: "", ok: False {}}
84378520 -- Stats/admin routes (exempt routes handled by isOverloadExempt; these are fallbacks)
@@ -8828,9 +8911,12 @@ main = do
88288911 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"
88298912 else do
88308913 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
88318917 atomic do
88328918 set *startedAt = t
8833− set *serverGeneration = t
8919+ set *serverGeneration = t * 1000 + genJitter
88348920 -- Empty maintenanceSecret is the recommended deployment path: use /maintenance/rotate-secret
88358921 -- at startup via deployment scripts for zero-knowledge secret management. Log at INFO level
88368922 -- since operators following the recommended path shouldn't see warnings on every startup.