@@ -43,6 +43,7 @@ maxSignatureLen = 256
43maxIpLen = 45 -- max length of an IP address string (IPv6 = 45 chars); used in matchIp instead of maxServerNameLen
43maxIpLen = 45 -- max length of an IP address string (IPv6 = 45 chars); used in matchIp instead of maxServerNameLen
44maxMessageIds = 100
44maxMessageIds = 100
45maxBatchSize = 100 -- max messages per batch request (bounds recursion depth in sendBatchInner)
45maxBatchSize = 100 -- max messages per batch request (bounds recursion depth in sendBatchInner)
46+maxMessagePriority = 10 -- max priority value (clamped on ingress)
46maxGossipEvents = 100
47maxGossipEvents = 100
47maxPollEvents = 50
48maxPollEvents = 50
48maxPresenceEntries = 10000
49maxPresenceEntries = 10000
@@ -159,17 +160,27 @@ forwardNonceRetention = messageMaxAge * 2 -- 60 days; extends 30 days beyond mes
159-- Message reconciliation: full scan runs every Nth prune cycle or when cheap drift check detects mismatch.
160-- Message reconciliation: full scan runs every Nth prune cycle or when cheap drift check detects mismatch.
160-- Cheap check (sum rcBucket totals vs totalMessageCount) runs every cycle; full scan only when needed.
161-- Cheap check (sum rcBucket totals vs totalMessageCount) runs every cycle; full scan only when needed.
161reconciliationFullCycleInterval = 50 -- full reconciliation every 50th prune cycle (~25 min)
162reconciliationFullCycleInterval = 50 -- full reconciliation every 50th prune cycle (~25 min)
163+-- Cheap drift threshold: only trigger full message reconciliation when the cheap check
164+-- (sum of rcBucket totals vs totalMessageCount) exceeds this threshold. Small drifts from
165+-- concurrent bucket pruning self-correct on the next cycle. 0 = always reconcile on drift.
166+reconciliationDriftThreshold = 10
163-- Federation latency metric window
168-- Federation latency metric window
164federationLatencyWindowMs : Int<Ms>
169federationLatencyWindowMs : Int<Ms>
165federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
170federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
167-- Snapshot format version (increment on schema changes to prevent corrupt restores)
172-- Snapshot format version (increment on schema changes to prevent corrupt restores)
168−snapshotVersion = 10 -- bumped for ForwardEntry.expiresAt field
173+snapshotVersion = 11 -- bumped for Message.priority and ForwardEntry.priority fields
170-- Server build version: included in /health and /stats responses for deployment verification.
175-- Server build version: included in /health and /stats responses for deployment verification.
171-- Update this string when deploying a new build so operators can confirm which version is running.
176-- Update this string when deploying a new build so operators can confirm which version is running.
172−serverBuildVersion = "2026-04-25a"
177+serverBuildVersion = "2026-04-26a"
179+-- Federation protocol version: included in gossip/forward/ack requests.
180+-- Peers log warnings when they receive requests from older versions.
181+-- Backward compatible: older peers ignore unknown fields (Knot's JSON parser
182+-- drops unknown keys). 0 = pre-versioning peer (field missing or unset).
183+federationProtocolVersion = 11
174-- Snapshot checksum: deterministic hash of entry counts for integrity verification.
185-- Snapshot checksum: deterministic hash of entry counts for integrity verification.
175-- Uses multiplicative hash to distinguish permutations of the same counts
186-- Uses multiplicative hash to distinguish permutations of the same counts
@@ -179,6 +190,20 @@ snapshotChecksum : Int -> Int -> Int -> Int -> Int -> Int
179snapshotChecksum = \presenceCount ackedCount deliveredCount forwardCount nonceCount ->
190snapshotChecksum = \presenceCount ackedCount deliveredCount forwardCount nonceCount ->
180 presenceCount * 100003 + ackedCount * 100019 + deliveredCount * 100043 + forwardCount * 100049 + nonceCount * 100057
191 presenceCount * 100003 + ackedCount * 100019 + deliveredCount * 100043 + forwardCount * 100049 + nonceCount * 100057
193+-- Snapshot integrity tag: combines checksum with maintenance secret to detect tampering.
194+-- Not cryptographic (no HMAC primitive in Knot), but catches naive modifications.
195+-- The tag includes the secret's hex representation hashed with the checksum, so an
196+-- attacker who modifies the snapshot but doesn't know the secret cannot recompute it.
197+-- Returns 0 when no maintenance secret is configured (no integrity protection).
198+snapshotIntegrityTag : Int -> Text -> Int
199+snapshotIntegrityTag = \checksum secret ->
200+ if secret == "" then 0
201+ else let secretHash = fold (\acc c -> acc * 31 + hexCharIndex c) 7 (chars (bytesToHex (textToBytes secret)))
202+ in checksum * 100069 + secretHash * 100081
204+-- Max seenBy list length in gossip events (bounds memory and bandwidth per event)
182-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
207-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
183maxGossipEventsPerServer = 500 -- max total gossip events per server per window
208maxGossipEventsPerServer = 500 -- max total gossip events per server per window
184gossipEventRateWindowMs : Int<Ms>
209gossipEventRateWindowMs : Int<Ms>
@@ -375,11 +400,13 @@ type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp, ip: Text,
375-- Messages with expiresAt <= now are pruned and excluded from polls.
400-- Messages with expiresAt <= now are pruned and excluded from polls.
376-- senderKey: pubkey of the authenticated sender (locally-originated only; "" for federation forwards).
401-- senderKey: pubkey of the authenticated sender (locally-originated only; "" for federation forwards).
377-- Used by /messages/cancel to verify the caller is the actual sender.
402-- Used by /messages/cancel to verify the caller is the actual sender.
403+-- priority: 0 = normal (default), higher = more urgent. High-priority messages are
404+-- forwarded first during federation retries. Clamped to [0, 10] on ingress.
378−type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp, expiresAt: Timestamp, senderKey: Text}
405+type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp, expiresAt: Timestamp, senderKey: Text, priority: Int}
379-- Presence: tracks that a pubkey is online at a given server, with expiry
406-- Presence: tracks that a pubkey is online at a given server, with expiry
380type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
407type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
381-- Queued federation forward: message waiting to be delivered to a remote server.
408-- Queued federation forward: message waiting to be delivered to a remote server.
382−type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int, expiresAt: Timestamp}
409+type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int, expiresAt: Timestamp, priority: Int}
383-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
410-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
384type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}
411type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}
385-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
412-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
@@ -403,7 +430,7 @@ type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int}
403type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
430type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
404-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles
431-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles
405type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}
432type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}
406−type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool, pendingForwards: Int}
433+type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool, pendingForwards: Int, avgLatencyMs: Int, maxLatencyMs: Int, latencySamples: Int}
407type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
434type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
408type RecipientCount = {pubkey: PubkeyHex, count: Int}
435type RecipientCount = {pubkey: PubkeyHex, count: Int}
409type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
436type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
@@ -420,7 +447,7 @@ type UniqueKeyIpEntry = {ip: Text, keys: [PubkeyHex], windowStart: Timestamp}
420type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, seenAt: Timestamp}
447type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, seenAt: Timestamp}
422-- Batch send types
449-- Batch send types
423−type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, expiresInMs: Int}
450+type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, expiresInMs: Int, priority: Int}
424type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
451type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
426-- Batch cancel types
453-- Batch cancel types
@@ -702,6 +729,11 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
702*slowVerifyCount : Int
729*slowVerifyCount : Int
703*verifyFailCount : Int
730*verifyFailCount : Int
732+-- Per-peer federation latency tracking: windowed average and max for /stats/peers.
733+-- Reset with the global federation latency window (federationLatencyWindowMs).
734+*peerLatencies : [{server: Text, totalMs: Int, count: Int, maxMs: Int, windowStart: Timestamp}]
735+maxPeerLatencyEntries = 100
705-- Server instance generation (epoch timestamp, for cursor staleness detection)
737-- Server instance generation (epoch timestamp, for cursor staleness detection)
706*serverGeneration : Int
738*serverGeneration : Int
@@ -826,6 +858,11 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
826-- Polls compare against this to skip expensive presence lookups when nothing changed.
858-- Polls compare against this to skip expensive presence lookups when nothing changed.
827*presenceVersion : Timestamp
859*presenceVersion : Timestamp
861+-- Active request counter: tracks concurrent requests for backpressure visibility.
862+-- Incremented at request start, decremented after handler completes.
863+*activeRequests : Int
864+*peakActiveRequests : Int
829-- Last time the background watchdog loop ran (for health check liveness detection)
866-- Last time the background watchdog loop ran (for health check liveness detection)
830*lastWatchdogAt : Timestamp
867*lastWatchdogAt : Timestamp
@@ -882,9 +919,11 @@ route ClientApi where
883 /messages
920 /messages
884 -- Send an E2E encrypted message to a recipient pubkey.
921 -- Send an E2E encrypted message to a recipient pubkey.
885− -- expiresInMs: optional sender-specified TTL in milliseconds (0 = server default).
922+ -- expiresInMs: optional sender-specified TTL in milliseconds (0 or negative = server default).
886 -- Clamped to [minMessageTtl, unackedMessageMaxAge]. Allows ephemeral messages.
923 -- Clamped to [minMessageTtl, unackedMessageMaxAge]. Allows ephemeral messages.
924+ -- priority: optional urgency level (0 = normal, higher = more urgent, clamped to [0, maxMessagePriority]).
925+ -- High-priority messages are forwarded first during federation retries.
887− POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, expiresInMs: Int} / headers {authorization: Text, clientIp: Text}
926+ POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, expiresInMs: Int, priority: Int} / headers {authorization: Text, clientIp: Text}
888 -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int, error: Text}
927 -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int, error: Text}
889 = SendMessage
928 = SendMessage
890 -- Check delivery status of a sent message (caller must know recipient to prove ownership)
929 -- Check delivery status of a sent message (caller must know recipient to prove ownership)
@@ -932,10 +971,12 @@ route ClientApi where
932-- Server-to-server federation API (authenticated via domain callback verification)
971-- Server-to-server federation API (authenticated via domain callback verification)
933-- correlationId: optional trace ID for cross-server request tracing (empty = not provided).
972-- correlationId: optional trace ID for cross-server request tracing (empty = not provided).
934-- Peers that don't support it send "" — logged when present but not enforced.
973-- Peers that don't support it send "" — logged when present but not enforced.
974+-- protocolVersion: federation protocol version of the sending server (0 = pre-versioning).
975+-- Older peers omit this field; Knot defaults missing Int fields to 0.
935route FederationApi where
976route FederationApi where
936 /federation
977 /federation
937 -- Receive presence gossip events from a peer server
978 -- Receive presence gossip events from a peer server
938− POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /gossip headers {authorization: Text}
979+ POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /gossip headers {authorization: Text}
939 -> {ok: Bool, error: Text}
980 -> {ok: Bool, error: Text}
940 = RecvGossip
981 = RecvGossip
941 -- Receive a forwarded message from a peer server on behalf of a sender.
982 -- Receive a forwarded message from a peer server on behalf of a sender.
@@ -944,27 +985,27 @@ route FederationApi where
944 -- privacy model (the server never learns who sent a message). /messages/cancel for
985 -- privacy model (the server never learns who sent a message). /messages/cancel for
945 -- federated messages requires the client to include sender identity in the encrypted
986 -- federated messages requires the client to include sender identity in the encrypted
946 -- payload; the server cannot verify sender identity for federation-forwarded messages.
987 -- payload; the server cannot verify sender identity for federation-forwarded messages.
947− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, expiresAt: Timestamp} /forward headers {authorization: Text}
988+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, expiresAt: Timestamp, protocolVersion: Int} /forward headers {authorization: Text}
948 -> {ok: Bool, error: Text}
989 -> {ok: Bool, error: Text}
949 = ForwardMessage
990 = ForwardMessage
950 -- Acknowledge delivery of messages back to the origin server
991 -- Acknowledge delivery of messages back to the origin server
951− POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /ack headers {authorization: Text}
992+ POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /ack headers {authorization: Text}
952 -> {ok: Bool, error: Text, ackedCount: Int, results: [{messageId: MessageIdHex, found: Bool}]}
993 -> {ok: Bool, error: Text, ackedCount: Int, results: [{messageId: MessageIdHex, found: Bool}]}
953 = FederationAck
994 = FederationAck
954 -- Cancel (retract) a previously forwarded message. The origin server sends this
995 -- Cancel (retract) a previously forwarded message. The origin server sends this
955 -- when the sender cancels a message via /messages/cancel. The receiving server
996 -- when the sender cancels a message via /messages/cancel. The receiving server
956 -- deletes the message if it hasn't been polled by the recipient yet.
997 -- deletes the message if it hasn't been polled by the recipient yet.
957− POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /cancel headers {authorization: Text}
998+ POST {messageId: MessageIdHex, toKey: PubkeyHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /cancel headers {authorization: Text}
958 -> {ok: Bool, error: Text, cancelled: Bool}
999 -> {ok: Bool, error: Text, cancelled: Bool}
959 = FederationCancel
1000 = FederationCancel
960 -- Batch cancel: cancel multiple previously forwarded messages in one request.
1001 -- Batch cancel: cancel multiple previously forwarded messages in one request.
961− POST {cancels: [{messageId: MessageIdHex, toKey: PubkeyHex}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /cancel-batch headers {authorization: Text}
1002+ POST {cancels: [{messageId: MessageIdHex, toKey: PubkeyHex}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /cancel-batch headers {authorization: Text}
962 -> {ok: Bool, error: Text, results: [BatchCancelResult]}
1003 -> {ok: Bool, error: Text, results: [BatchCancelResult]}
963 = FederationBatchCancel
1004 = FederationBatchCancel
964 -- Batch forward: forward multiple messages in one request (reduces HTTP round-trips).
1005 -- Batch forward: forward multiple messages in one request (reduces HTTP round-trips).
965 -- Each message is processed independently; partial success is possible.
1006 -- Each message is processed independently; partial success is possible.
966 -- Results are returned per-message so the sender knows which forwards succeeded.
1007 -- Results are returned per-message so the sender knows which forwards succeeded.
967− POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, expiresAt: Timestamp}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /forward-batch headers {authorization: Text}
1008+ POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, expiresAt: Timestamp}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, protocolVersion: Int} /forward-batch headers {authorization: Text}
968 -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}]}
1009 -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}]}
969 = FederationBatchForward
1010 = FederationBatchForward
@@ -1087,7 +1128,7 @@ route HealthApi where
1087-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1128-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1088route StatsApi where
1129route StatsApi where
1089 GET /stats headers {authorization: Text}
1130 GET /stats headers {authorization: Text}
1090− -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, activePolls: Int, activePollsMax: Int, recipientCounts: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}]}, 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, prunesSinceSnapshot: Int, totalEntries: Int}}
1131+ -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, activePolls: Int, activePollsMax: Int, recipientCounts: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}]}, 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, prunesSinceSnapshot: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int}}
1092 GET /stats/peers headers {authorization: Text}
1133 GET /stats/peers headers {authorization: Text}
1093 -> {peers: [PeerStatus], ok: Bool}
1134 -> {peers: [PeerStatus], ok: Bool}
@@ -1128,14 +1169,35 @@ logError = \msg -> println ("[ERROR] " ++ msg)
1128-- when msg contains user-controlled data (e.g., error strings, server names).
1169-- when msg contains user-controlled data (e.g., error strings, server names).
1129logEvent = \level event msg -> println ("{\"level\":\"" ++ escapeJsonString level ++ "\",\"event\":\"" ++ escapeJsonString event ++ "\",\"msg\":\"" ++ escapeJsonString msg ++ "\"}")
1170logEvent = \level event msg -> println ("{\"level\":\"" ++ escapeJsonString level ++ "\",\"event\":\"" ++ escapeJsonString event ++ "\",\"msg\":\"" ++ escapeJsonString msg ++ "\"}")
1172+-- Log a warning when a federation peer sends a request with an older protocol version.
1173+-- protocolVersion == 0 means pre-versioning peer (field missing). Only logs once per
1174+-- request — callers invoke this before delegating to the handler.
1175+warnOldProtocol = \fromServer pv ->
1176+ when (pv > 0 && pv < federationProtocolVersion)
1177+ (logWarn ("federation: peer " ++ fromServer ++ " using protocol v" ++ show pv ++ " (current: v" ++ show federationProtocolVersion ++ ")"))
1179+-- Fast path: check if a string contains only safe JSON characters (hex digits,
1180+-- alphanumerics, common punctuation). If so, no escaping is needed.
1181+-- Returns False on the first unsafe character. Tail-recursive (TCO-safe in Knot).
1182+isJsonSafe : Text -> Bool
1184+ if s == "" then True {}
1185+ else let c = take 1 s in
1186+ if contains c "0123456789abcdefABCDEF_-.:/ " then isJsonSafe (drop 1 s)
1187+ else if contains c "ghijklmnopqrstuvwxyzGHIJKLMNOPQRSTUVWXYZ" then isJsonSafe (drop 1 s)
1131-- Escape special characters for safe JSON string embedding.
1190-- Escape special characters for safe JSON string embedding.
1132-- Handles: backslash, double quote, and control characters (\n, \r, \t, \b, \f).
1191-- Handles: backslash, double quote, and control characters (\n, \r, \t, \b, \f).
1133-- Non-printable characters outside this set are stripped (same allowlist as
1192-- Non-printable characters outside this set are stripped (same allowlist as
1134-- stripNonPrintable) to prevent log injection via error strings from federation peers.
1193-- stripNonPrintable) to prevent log injection via error strings from federation peers.
1194+-- Fast path: strings containing only safe chars (hex, alphanumeric, common punctuation)
1195+-- are returned unchanged, avoiding O(n) per-character recursion for validated types.
1135-- Processes one character at a time via tail recursion (TCO-safe in Knot).
1196-- Processes one character at a time via tail recursion (TCO-safe in Knot).
1136escapeJsonString : Text -> Text
1197escapeJsonString : Text -> Text
1137escapeJsonString = \s ->
1198escapeJsonString = \s ->
1138 if s == "" then ""
1199 if s == "" then ""
1200+ else if isJsonSafe s then s
1139 else let c = take 1 s in
1201 else let c = take 1 s in
1140 if c == "\\" then "\\\\" ++ escapeJsonString (drop 1 s)
1202 if c == "\\" then "\\\\" ++ escapeJsonString (drop 1 s)
1141 else if c == "\"" then "\\\"" ++ escapeJsonString (drop 1 s)
1203 else if c == "\"" then "\\\"" ++ escapeJsonString (drop 1 s)
@@ -2195,18 +2257,20 @@ countByRecipient = \items ->
2195-- Linear scan over sorted items: counts consecutive runs of the same toKey.
2257-- Linear scan over sorted items: counts consecutive runs of the same toKey.
2196-- Tail-recursive (TCO-safe in Knot). Each step either increments the current
2258-- Tail-recursive (TCO-safe in Knot). Each step either increments the current
2197-- group's count or emits the group and starts a new one.
2259-- group's count or emits the group and starts a new one.
2260+-- Uses messageId for element removal (unique per bucket) instead of full structural
2261+-- equality, avoiding expensive per-field comparison on Message records with large blobs.
2198countSortedGroups = \items ->
2262countSortedGroups = \items ->
2199 case headOf items of
2263 case headOf items of
2200 Nothing {} -> []
2264 Nothing {} -> []
2201 Just {value: first} ->
2265 Just {value: first} ->
2202− let rest = do x <- items; where x != first; yield x in
2266+ let rest = do x <- items; where x.messageId != first.messageId; yield x in
2203 countSortedGroupsInner rest first.toKey 1 []
2267 countSortedGroupsInner rest first.toKey 1 []
2205countSortedGroupsInner = \items currentKey currentCount acc ->
2269countSortedGroupsInner = \items currentKey currentCount acc ->
2206 case headOf items of
2270 case headOf items of
2207 Nothing {} -> union acc [{pubkey: currentKey, count: currentCount}]
2271 Nothing {} -> union acc [{pubkey: currentKey, count: currentCount}]
2208 Just {value: x} ->
2272 Just {value: x} ->
2209− let rest = do r <- items; where r != x; yield r in
2273+ let rest = do r <- items; where r.messageId != x.messageId; yield r in
2210 if x.toKey == currentKey
2274 if x.toKey == currentKey
2211 then countSortedGroupsInner rest currentKey (currentCount + 1) acc
2275 then countSortedGroupsInner rest currentKey (currentCount + 1) acc
2212 else countSortedGroupsInner rest x.toKey 1 (union acc [{pubkey: currentKey, count: currentCount}])
2276 else countSortedGroupsInner rest x.toKey 1 (union acc [{pubkey: currentKey, count: currentCount}])
@@ -2786,7 +2850,15 @@ fetchFederation = \server request -> do
2786 mx <- *federationLatencyMax
2850 mx <- *federationLatencyMax
2787 set *federationLatencyTotal = total + latency
2851 set *federationLatencyTotal = total + latency
2788 set *federationLatencyCount = cnt + 1
2852 set *federationLatencyCount = cnt + 1
2789− set *federationLatencyMax = if latency > mx then latency else mx)
2853+ set *federationLatencyMax = if latency > mx then latency else mx
2854+ -- Per-peer latency tracking
2855+ peerLats <- *peerLatencies
2856+ let peerEntry = headOf (do p <- peerLats; where p.server == server; yield p)
2857+ let isNewWindow = case peerEntry of Nothing {} -> True {}; Just {value: p} -> endTime - p.windowStart >= federationLatencyWindowMs
2858+ let updatedPeers = if isNewWindow
2859+ then union (do p <- peerLats; where p.server != server; yield p) [{server: server, totalMs: latency, count: 1, maxMs: latency, windowStart: endTime}]
2860+ else do p <- peerLats; yield (if p.server == server then {server: p.server, totalMs: p.totalMs + latency, count: p.count + 1, maxMs: max p.maxMs latency, windowStart: p.windowStart} else p)
2861+ set *peerLatencies = if count updatedPeers > maxPeerLatencyEntries then evictOneOldest updatedPeers (\p -> p.windowStart) else updatedPeers)
2790 yield result
2862 yield result
2792clampTtl : Int<S> -> Int<S>
2864clampTtl : Int<S> -> Int<S>
@@ -2797,6 +2869,7 @@ clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
2798-- Compute effective message TTL: 0 = server default, otherwise clamp to [minMessageTtl, unackedMessageMaxAge].
2870-- Compute effective message TTL: 0 = server default, otherwise clamp to [minMessageTtl, unackedMessageMaxAge].
2799-- Returns the TTL in milliseconds to add to receivedAt for expiresAt.
2871-- Returns the TTL in milliseconds to add to receivedAt for expiresAt.
2872+-- Compute effective message TTL: 0 or negative = server default, otherwise clamp.
2800effectiveMessageTtl : Int<Ms> -> Int<Ms>
2873effectiveMessageTtl : Int<Ms> -> Int<Ms>
2801effectiveMessageTtl = \expiresInMs ->
2874effectiveMessageTtl = \expiresInMs ->
2802 if expiresInMs <= 0 then unackedMessageMaxAge
2875 if expiresInMs <= 0 then unackedMessageMaxAge
@@ -3304,7 +3377,8 @@ sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {
3304sendBatchedGossipToServer = \server allEvents hops ->
3377sendBatchedGossipToServer = \server allEvents hops ->
3305 -- Filter out events this server has already seen (via seenBy tracking).
3378 -- Filter out events this server has already seen (via seenBy tracking).
3306 -- Add ourselves and the target server to seenBy so downstream relays skip both.
3379 -- Add ourselves and the target server to seenBy so downstream relays skip both.
3380+ -- Cap seenBy at maxSeenByLen to bound per-event memory and bandwidth.
3307− let unseen = do e <- allEvents; where none e.seenBy (\s -> s == server); yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: union e.seenBy [server]}
3381+ let unseen = do e <- allEvents; where none e.seenBy (\s -> s == server); yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: takeRelation maxSeenByLen (union e.seenBy [server])}
3308 in if count unseen == 0 then yield {}
3382 in if count unseen == 0 then yield {}
3309 else forEachBatch unseen maxGossipEvents (\batch -> do
3383 else forEachBatch unseen maxGossipEvents (\batch -> do
3310 t <- now
3384 t <- now
@@ -3313,7 +3387,7 @@ sendBatchedGossipToServer = \server allEvents hops ->
3313 result <- fetchFederation server (RecvGossip {
3387 result <- fetchFederation server (RecvGossip {
3314 authorization: auth,
3388 authorization: auth,
3315 events: batch, hopsRemaining: hops, fromServer: serverHostN,
3389 events: batch, hopsRemaining: hops, fromServer: serverHostN,
3316− requestTimestamp: t, correlationId: corrId})
3390+ requestTimestamp: t, correlationId: corrId, protocolVersion: federationProtocolVersion})
3317 case result of
3391 case result of
3318 Err e -> logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
3392 Err e -> logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
3319 Ok _ -> yield {})
3393 Ok _ -> yield {})
@@ -3728,13 +3802,16 @@ retryForwardsForServer = \entries remaining ->
3728 let batchPayload = do f <- batch; yield {toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId, expiresAt: f.expiresAt}
3802 let batchPayload = do f <- batch; yield {toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId, expiresAt: f.expiresAt}
3729 result <- fetchFederation server (FederationBatchForward {
3803 result <- fetchFederation server (FederationBatchForward {
3730 forwards: batchPayload, fromServer: serverHostN,
3804 forwards: batchPayload, fromServer: serverHostN,
3731− requestTimestamp: t, authorization: auth, correlationId: corrId})
3805+ requestTimestamp: t, authorization: auth, correlationId: corrId, protocolVersion: federationProtocolVersion})
3732 case result of
3806 case result of
3733 Err e -> do
3807 Err e -> do
3734 logWarn ("federation: batch forward to " ++ server ++ " failed: " ++ show e)
3808 logWarn ("federation: batch forward to " ++ server ++ " failed: " ++ show e)
3735 recordFailure server
3809 recordFailure server
3736− -- Increment retries for all entries in this batch
3810+ -- HTTP-level failure (connectivity, timeout): don't increment per-message retries.
3737− retryForwardsBatchIncrement batch server
3811+ -- The messages weren't rejected — the server was unreachable. Server backoff
3812+ -- (via recordFailure) handles retry scheduling. Per-message retries are only
3813+ -- burned on application-level rejections (Ok with per-message results).
3738 Ok response -> do
3815 Ok response -> do
3739 if response.ok then do
3816 if response.ok then do
3740 clearFailure server
3817 clearFailure server
@@ -3742,7 +3819,7 @@ retryForwardsForServer = \entries remaining ->
3742 retryForwardsBatchResults batch response.results server
3819 retryForwardsBatchResults batch response.results server
3744 logWarn ("federation: batch forward to " ++ server ++ " rejected: " ++ response.error)
3821 logWarn ("federation: batch forward to " ++ server ++ " rejected: " ++ response.error)
3745− -- Peer may not support batch forward; increment retries
3822+ -- Application-level batch rejection (old peer, invalid request): increment retries
3746 retryForwardsBatchIncrement batch server)
3823 retryForwardsBatchIncrement batch server)
3748-- Increment retry counts for all entries in a failed batch
3825-- Increment retry counts for all entries in a failed batch
@@ -3796,7 +3873,10 @@ retryPendingForwards = \pending -> do
3796 activePolls <- *activePolls
3873 activePolls <- *activePolls
3797 let onlineForwards = do f <- eligibleForwards; where any activePolls (\p -> p.pubkey == f.toKey); yield f
3874 let onlineForwards = do f <- eligibleForwards; where any activePolls (\p -> p.pubkey == f.toKey); yield f
3798 let offlineForwards = do f <- eligibleForwards; where none activePolls (\p -> p.pubkey == f.toKey); yield f
3875 let offlineForwards = do f <- eligibleForwards; where none activePolls (\p -> p.pubkey == f.toKey); yield f
3876+ -- Sort by priority descending (higher priority first) within each group
3877+ let sortedOnline = sortBy (\f -> 0 - f.priority) onlineForwards
3878+ let sortedOffline = sortBy (\f -> 0 - f.priority) offlineForwards
3799− let prioritizedForwards = union onlineForwards offlineForwards
3879+ let prioritizedForwards = union sortedOnline sortedOffline
3800 -- Prioritize servers that have online-key forwards first
3880 -- Prioritize servers that have online-key forwards first
3801 let onlineServers = collectPeerServers onlineForwards (\x -> x.toServer)
3881 let onlineServers = collectPeerServers onlineForwards (\x -> x.toServer)
3802 let offlineOnlyServers = do s <- eligibleServers; where none onlineServers (\os -> os == s); yield s
3882 let offlineOnlyServers = do s <- eligibleServers; where none onlineServers (\os -> os == s); yield s
@@ -3850,7 +3930,7 @@ fetchFederationAck = \origin messageIds -> do
3850 authorization: auth,
3930 authorization: auth,
3851 messageIds: messageIds,
3931 messageIds: messageIds,
3852 originServer: origin, fromServer: serverHostN,
3932 originServer: origin, fromServer: serverHostN,
3853− requestTimestamp: t, correlationId: corrId})
3933+ requestTimestamp: t, correlationId: corrId, protocolVersion: federationProtocolVersion})
3855-- Per-bucket ack retry increment: scans only the buckets containing entries from the batch.
3935-- Per-bucket ack retry increment: scans only the buckets containing entries from the batch.
3856-- Each bucket is processed in its own atomic block to reduce lock contention.
3936-- Each bucket is processed in its own atomic block to reduce lock contention.
@@ -4087,19 +4167,26 @@ pruneAndRetry = \t -> do
4087 when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0)
4167 when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0)
4088 (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift))
4168 (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift))
4089 -- Message reconciliation: cheap check first (sum rcBucket .count fields vs totalMessageCount).
4169 -- Message reconciliation: cheap check first (sum rcBucket .count fields vs totalMessageCount).
4090− -- Full scan only when drift detected or every Nth cycle (reconciliationFullCycleInterval).
4170+ -- Full scan only when drift exceeds threshold or every Nth cycle.
4171+ -- Small drifts from concurrent bucket pruning self-correct; only trigger the expensive
4172+ -- full scan when drift is significant (> reconciliationDriftThreshold) or periodic.
4091 pc <- *pruneCount
4173 pc <- *pruneCount
4092 let isFullCycle = pc > 0 && (pc - (pc / reconciliationFullCycleInterval) * reconciliationFullCycleInterval) == 0
4174 let isFullCycle = pc > 0 && (pc - (pc / reconciliationFullCycleInterval) * reconciliationFullCycleInterval) == 0
4093 cheapMsgDrift <- atomic do
4175 cheapMsgDrift <- atomic do
4094 rcTotal <- sumRcBucketTotals 0
4176 rcTotal <- sumRcBucketTotals 0
4095 stored <- *totalMessageCount
4177 stored <- *totalMessageCount
4096 yield (rcTotal - stored)
4178 yield (rcTotal - stored)
4097− if cheapMsgDrift != 0 || isFullCycle
4179+ let absDrift = if cheapMsgDrift < 0 then 0 - cheapMsgDrift else cheapMsgDrift
4180+ if absDrift > reconciliationDriftThreshold || isFullCycle
4099 msgDrift <- reconcileMessageCounts 0 0
4182 msgDrift <- reconcileMessageCounts 0 0
4100 when (msgDrift != 0)
4183 when (msgDrift != 0)
4101 (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))
4184 (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))
4102− else yield {}
4185+ else when (cheapMsgDrift != 0) (atomic do
4186+ -- Small drift: correct totalMessageCount directly from rcBucket totals
4187+ -- without the full per-bucket rebuild (O(16 bucket scans) vs O(all messages))
4188+ rcTotal <- sumRcBucketTotals 0
4189+ set *totalMessageCount = rcTotal)
4103 pendingForwards <- atomic do f <- readAllForwards; yield f
4190 pendingForwards <- atomic do f <- readAllForwards; yield f
4104 ackState <- atomic do
4191 ackState <- atomic do
4105 ackedIds2 <- readAllAcked
4192 ackedIds2 <- readAllAcked
@@ -4160,6 +4247,8 @@ pruneAndRetry = \t -> do
4160 -- Prune expired verify cache entries
4247 -- Prune expired verify cache entries
4161 cache <- *verificationCache
4248 cache <- *verificationCache
4162 set *verificationCache = do c <- cache; where t - c.verifiedAt < verificationCacheMs; yield c
4249 set *verificationCache = do c <- cache; where t - c.verifiedAt < verificationCacheMs; yield c
4250+ -- Reset per-peer latency tracking with global window
4251+ set *peerLatencies = []
4163 set *federationLatencyResetAt = t)
4252 set *federationLatencyResetAt = t)
4164 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
4253 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
4165 -- These were queued by handleFederationAck instead of fork+sleep.
4254 -- These were queued by handleFederationAck instead of fork+sleep.
@@ -4448,7 +4537,7 @@ forwardCandidates = \targets messages forwards t ->
4448 let origin = if m.originServer == "" then serverHostN else m.originServer
4537 let origin = if m.originServer == "" then serverHostN else m.originServer
4449 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
4538 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
4450 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,
4539 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,
4451− originServer: origin, retries: 0, expiresAt: m.expiresAt})) [] targets
4540+ originServer: origin, retries: 0, expiresAt: m.expiresAt, priority: m.priority})) [] targets
4454-- Challenge-IP binding rules:
4543-- Challenge-IP binding rules:
@@ -4787,7 +4876,7 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
4787-- Core send logic: dedup, rate limiting, store message, create forwards.
4876-- Core send logic: dedup, rate limiting, store message, create forwards.
4788-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
4877-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
4789-- Caller handles auth, drain checks, and messageId generation.
4878-- Caller handles auth, drain checks, and messageId generation.
4790−sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs t ->
4879+sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs msgPriority t ->
4791 if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4880 if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4793 let msgBucket = pubkeyBucket to
4882 let msgBucket = pubkeyBucket to
@@ -4836,7 +4925,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
4836 let msgExpiresAt = seqTs + effectiveMessageTtl expiresInMs
4925 let msgExpiresAt = seqTs + effectiveMessageTtl expiresInMs
4837 writeMessageBucket msgBucket (union bucket
4926 writeMessageBucket msgBucket (union bucket
4838 [{messageId: effectiveMsgId, toKey: to,
4927 [{messageId: effectiveMsgId, toKey: to,
4839− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt, senderKey: senderKey}])
4928+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt, senderKey: senderKey, priority: msgPriority}])
4840 set *totalMessageCount = totalMsgCount + 1
4929 set *totalMessageCount = totalMsgCount + 1
4841 writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)
4930 writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)
4842 signalMessageForKey to seqTs
4931 signalMessageForKey to seqTs
@@ -4857,7 +4946,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
4857 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
4946 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
4858 let fbi = pubkeyBucket to
4947 let fbi = pubkeyBucket to
4859 fwdBkt <- readFwdBucket fbi
4948 fwdBkt <- readFwdBucket fbi
4860− let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: msgExpiresAt}
4949+ let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: msgExpiresAt, priority: msgPriority}
4861 totalFwdCount <- *totalForwardCount
4950 totalFwdCount <- *totalForwardCount
4862 let globalCap = maxForwardsEntries - totalFwdCount
4951 let globalCap = maxForwardsEntries - totalFwdCount
4863 -- Per-key cap: prevents one key from consuming disproportionate forward slots
4952 -- Per-key cap: prevents one key from consuming disproportionate forward slots
@@ -4892,7 +4981,7 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
4892 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
4981 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
4893 fromServer: serverHostN, requestTimestamp: fwdT,
4982 fromServer: serverHostN, requestTimestamp: fwdT,
4894 authorization: fwdAuth, correlationId: fwdCorrId,
4983 authorization: fwdAuth, correlationId: fwdCorrId,
4895− expiresAt: fwdInfo.fwdExpiresAt})
4984+ expiresAt: fwdInfo.fwdExpiresAt, protocolVersion: federationProtocolVersion})
4896 case fwdResult of
4985 case fwdResult of
4897 Err e -> do
4986 Err e -> do
4898 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
4987 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
@@ -4925,7 +5014,7 @@ sendBatchForwardsSequential = \msgs server ->
4925 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
5014 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
4926 fromServer: serverHostN, requestTimestamp: fwdT,
5015 fromServer: serverHostN, requestTimestamp: fwdT,
4927 authorization: fwdAuth, correlationId: fwdCorrId,
5016 authorization: fwdAuth, correlationId: fwdCorrId,
4928− expiresAt: batchFwdInfo.fwdExpiresAt})
5017+ expiresAt: batchFwdInfo.fwdExpiresAt, protocolVersion: federationProtocolVersion})
4929 case fwdResult of
5018 case fwdResult of
4930 Err e -> do
5019 Err e -> do
4931 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
5020 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
@@ -4941,7 +5030,8 @@ sendBatchForwardsSequential = \msgs server ->
4941-- POST /messages: Send an encrypted message to a recipient.
5030-- POST /messages: Send an encrypted message to a recipient.
4942-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
5031-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
4943-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
5032-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
4944−handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs authorization clientIp respond -> do
5033+handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority authorization clientIp respond -> do
5034+ let clampedPriority = clamp 0 maxMessagePriority priority
4945 t <- now
5035 t <- now
4946 globalRateLimited <- checkGlobalRate t
5036 globalRateLimited <- checkGlobalRate t
4947 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0, error: "rate_limited"})
5037 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0, error: "rate_limited"})
@@ -4960,7 +5050,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs authorizat
4960 if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
5050 if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
4961 else withSessionAuth authorization t clientIp
5051 else withSessionAuth authorization t clientIp
4962 (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
5052 (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
4963− (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs t)
5053+ (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs clampedPriority t)
4964 case result.status of
5054 case result.status of
4965 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
5055 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
4966 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (global capacity full)"))
5056 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (global capacity full)"))
@@ -5220,13 +5310,15 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
5220 let cappedHops = clamp 0 maxGossipHops hopsRemaining
5310 let cappedHops = clamp 0 maxGossipHops hopsRemaining
5221 let corrId = sanitizeCorrelationId correlationId
5311 let corrId = sanitizeCorrelationId correlationId
5222 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " gossip from " ++ fromServer))
5312 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " gossip from " ++ fromServer))
5313+ -- Cap inbound seenBy lists to prevent memory waste from malicious peers
5314+ let cappedEvents = do e <- events; yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: takeRelation maxSeenByLen e.seenBy}
5223− withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (True {}) (do
5315+ withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count cappedEvents <= maxGossipEvents) (True {}) (do
5224 -- Per-peer gossip event volume limiting: counts total events (not just requests) per server.
5316 -- Per-peer gossip event volume limiting: counts total events (not just requests) per server.
5225 -- A single gossip request carries up to maxGossipEvents entries; without this, a peer
5317 -- A single gossip request carries up to maxGossipEvents entries; without this, a peer
5226 -- sending 50 req/s * 100 events = 5000 presence mutations/second.
5318 -- sending 50 req/s * 100 events = 5000 presence mutations/second.
5227 eventRateLimited <- atomic do
5319 eventRateLimited <- atomic do
5228 ger <- *gossipEventRates
5320 ger <- *gossipEventRates
5229− let eventCount = count events
5321+ let eventCount = count cappedEvents
5230 let entry = headOf (do r <- ger; where r.server == fromServer; yield r)
5322 let entry = headOf (do r <- ger; where r.server == fromServer; yield r)
5231 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= gossipEventRateWindowMs
5323 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= gossipEventRateWindowMs
5232 let currentCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count
5324 let currentCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count
@@ -5242,7 +5334,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
5242 logWarn ("gossip: event volume rate limited for " ++ fromServer ++ " (" ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms)")
5334 logWarn ("gossip: event volume rate limited for " ++ fromServer ++ " (" ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms)")
5243 yield (respond {ok: False {}, error: "event_rate_limited:" ++ show gossipEventRateWindowMs ++ "ms"})
5335 yield (respond {ok: False {}, error: "event_rate_limited:" ++ show gossipEventRateWindowMs ++ "ms"})
5245− let validated = validGossipEvents events t
5337+ let validated = validGossipEvents cappedEvents t
5246 let onlineEntries = validated.online
5338 let onlineEntries = validated.online
5247 let trustedOffline = trustedOfflineEvents validated.offline fromServer
5339 let trustedOffline = trustedOfflineEvents validated.offline fromServer
5248 -- Early return: skip the expensive presence merge when all events are invalid/expired
5340 -- Early return: skip the expensive presence merge when all events are invalid/expired
@@ -5421,7 +5513,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
5421 let fwdExpiresAt = if expiresAt > 0 && expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
5513 let fwdExpiresAt = if expiresAt > 0 && expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
5422 writeMessageBucket bkt (union bucket
5514 writeMessageBucket bkt (union bucket
5423 [{messageId: messageId, toKey: toKey,
5515 [{messageId: messageId, toKey: toKey,
5424− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: ""}])
5516+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: 0}])
5425 set *totalMessageCount = totalMsgCount + 1
5517 set *totalMessageCount = totalMsgCount + 1
5426 writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
5518 writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
5427 signalMessageForKey toKey seqTs
5519 signalMessageForKey toKey seqTs
@@ -5589,33 +5681,46 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
5589 (logInfo ("federation/cancel-batch: cancelled " ++ show cancelledCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer))
5681 (logInfo ("federation/cancel-batch: cancelled " ++ show cancelledCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer))
5590 yield (respond {ok: True {}, error: "", results: results}))
5682 yield (respond {ok: True {}, error: "", results: results}))
5592−-- Process each cancel entry in a batch, grouping by bucket for efficiency.
5684+-- Process cancel entries grouped by bucket: all entries sharing a pubkeyBucket are
5685+-- processed in a single atomic block, reducing bucket reads from O(entries) to O(buckets).
5593cancelBatchInner = \entries fromServer t acc ->
5686cancelBatchInner = \entries fromServer t acc ->
5687+ case headOf entries of
5688+ Nothing {} -> yield acc
5689+ Just {value: first} -> do
5690+ let b = pubkeyBucket first.toKey
5691+ -- Collect all entries targeting the same bucket
5692+ let sameBucket = do e <- entries; where pubkeyBucket e.toKey == b; yield e
5693+ let otherBuckets = do e <- entries; where pubkeyBucket e.toKey != b; yield e
5694+ -- Process the entire bucket group in one atomic block
5695+ results <- atomic do cancelBucketGroup sameBucket fromServer b []
5696+ cancelBatchInner otherBuckets fromServer t (union acc results)
5698+-- Process all cancel entries for a single bucket within one atomic block.
5699+-- Reads messageBucket/fwdBucket once and applies all cancels, avoiding redundant reads.
5700+cancelBucketGroup = \entries fromServer b acc ->
5594 case headOf entries of
5701 case headOf entries of
5595 Nothing {} -> yield acc
5702 Nothing {} -> yield acc
5596 Just {value: entry} -> do
5703 Just {value: entry} -> do
5597 let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
5704 let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
5598− result <- atomic do
5599− let b = pubkeyBucket entry.toKey
5600− bucket <- readMessageBucket b
5705+ bucket <- readMessageBucket b
5601− let toCancel = headOf (do m <- bucket; where m.messageId == entry.messageId && m.toKey == entry.toKey && m.originServer == fromServer; yield m)
5706+ let toCancel = headOf (do m <- bucket; where m.messageId == entry.messageId && m.toKey == entry.toKey && m.originServer == fromServer; yield m)
5602− case toCancel of
5707+ result <- case toCancel of
5603− Nothing {} -> yield {cancelled: False {}}
5708+ Nothing {} -> yield {cancelled: False {}}
5604− Just {} -> do
5709+ Just {} -> do
5605− writeMessageBucket b (do m <- bucket; where not (m.messageId == entry.messageId && m.toKey == entry.toKey); yield m)
5710+ writeMessageBucket b (do m <- bucket; where not (m.messageId == entry.messageId && m.toKey == entry.toKey); yield m)
5606− tc <- *totalMessageCount
5711+ tc <- *totalMessageCount
5607− set *totalMessageCount = max 0 (tc - 1)
5712+ set *totalMessageCount = max 0 (tc - 1)
5608− rc <- readRcBucket b
5713+ rc <- readRcBucket b
5609− writeRcBucket b (adjustRecipientCount rc entry.toKey (0 - 1))
5714+ writeRcBucket b (adjustRecipientCount rc entry.toKey (0 - 1))
5610− fwdBkt <- readFwdBucket b
5715+ fwdBkt <- readFwdBucket b
5611− let keptFwds = do f <- fwdBkt; where not (f.messageId == entry.messageId && f.toKey == entry.toKey); yield f
5716+ let keptFwds = do f <- fwdBkt; where not (f.messageId == entry.messageId && f.toKey == entry.toKey); yield f
5612− writeFwdBucket b keptFwds
5717+ writeFwdBucket b keptFwds
5613− let fwdRemoved = count fwdBkt - count keptFwds
5718+ let fwdRemoved = count fwdBkt - count keptFwds
5614− when (fwdRemoved > 0) (do
5719+ when (fwdRemoved > 0) (do
5615− tfc <- *totalForwardCount
5720+ tfc <- *totalForwardCount
5616− set *totalForwardCount = max 0 (tfc - fwdRemoved))
5721+ set *totalForwardCount = max 0 (tfc - fwdRemoved))
5617− yield {cancelled: True {}}
5722+ yield {cancelled: True {}}
5618− cancelBatchInner rest fromServer t (union acc [{messageId: entry.messageId, cancelled: result.cancelled}])
5723+ cancelBucketGroup rest fromServer b (union acc [{messageId: entry.messageId, cancelled: result.cancelled}])
5620-- POST /federation/forward-batch: Receive multiple forwarded messages in one request.
5725-- POST /federation/forward-batch: Receive multiple forwarded messages in one request.
5621-- Each message is processed independently (same logic as handleForwardMessage).
5726-- Each message is processed independently (same logic as handleForwardMessage).
@@ -5676,22 +5781,34 @@ forwardBatchInner = \entries fromServer t acc ->
5676 if totalMsgCount >= maxMessages || recipientCount >= maxMessagesPerRecipient
5781 if totalMsgCount >= maxMessages || recipientCount >= maxMessagesPerRecipient
5677 then yield (False {})
5782 then yield (False {})
5784+ -- Per-recipient receive rate limiting (matches handleForwardMessage)
5785+ recvRates <- *recvRates
5786+ let recvResult = checkPubkeyRate recvRates entry.toKey maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
5787+ if recvResult.limited then yield (False {})
5789+ set *recvRates = recvResult.entries
5790+ -- Per-server-per-recipient rate limiting (matches handleForwardMessage)
5791+ federationPairRates <- *federationPairRates
5792+ let fedPairResult = checkServerRecipientRate federationPairRates fromServer entry.toKey maxFederationPairRatePerPair federationPairRateWindowMs maxFederationPairRateEntries t
5793+ if fedPairResult.limited then yield (False {})
5795+ set *federationPairRates = fedPairResult.entries
5679− seqCounter <- *seqCounter
5796+ seqCounter <- *seqCounter
5680− let seqTs = nextSeqTs t seqCounter
5797+ let seqTs = nextSeqTs t seqCounter
5681− set *seqCounter = seqTs
5798+ set *seqCounter = seqTs
5682− let fwdExpiresAt = if entry.expiresAt > 0 && entry.expiresAt > t then min entry.expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
5799+ let fwdExpiresAt = if entry.expiresAt > 0 && entry.expiresAt > t then min entry.expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
5683− writeMessageBucket bkt (union bucket
5800+ writeMessageBucket bkt (union bucket
5684− [{messageId: entry.messageId, toKey: entry.toKey,
5801+ [{messageId: entry.messageId, toKey: entry.toKey,
5685− encryptedBlob: entry.encryptedBlob, timestamp: clampedTs, originServer: entry.originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: ""}])
5802+ encryptedBlob: entry.encryptedBlob, timestamp: clampedTs, originServer: entry.originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: 0}])
5686− set *totalMessageCount = totalMsgCount + 1
5803+ set *totalMessageCount = totalMsgCount + 1
5687− writeRcBucket bkt (adjustRecipientCount recipientCounts entry.toKey 1)
5804+ writeRcBucket bkt (adjustRecipientCount recipientCounts entry.toKey 1)
5688− signalMessageForKey entry.toKey seqTs
5805+ signalMessageForKey entry.toKey seqTs
5689− fwdNonces <- readNonceBucket bkt
5806+ fwdNonces <- readNonceBucket bkt
5690− let updatedNonces = union fwdNonces [{messageId: entry.messageId, toKey: entry.toKey, seenAt: t}]
5807+ let updatedNonces = union fwdNonces [{messageId: entry.messageId, toKey: entry.toKey, seenAt: t}]
5691− writeNonceBucket bkt (if count updatedNonces > maxNoncesPerBucket
5808+ writeNonceBucket bkt (if count updatedNonces > maxNoncesPerBucket
5692− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5809+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5693− else updatedNonces)
5810+ else updatedNonces)
5694− yield (True {})
5811+ yield (True {})
5695 forwardBatchInner rest fromServer t (union acc [{messageId: entry.messageId, ok: result}])
5812 forwardBatchInner rest fromServer t (union acc [{messageId: entry.messageId, ok: result}])
5697-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
5814-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
@@ -5729,7 +5846,7 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
5729 fetchFederation server (FederationBatchCancel {
5846 fetchFederation server (FederationBatchCancel {
5730 cancels: serverCancels, fromServer: serverHostN,
5847 cancels: serverCancels, fromServer: serverHostN,
5731 requestTimestamp: cancelT, correlationId: cancelCorrId,
5848 requestTimestamp: cancelT, correlationId: cancelCorrId,
5732− authorization: cancelAuth})
5849+ authorization: cancelAuth, protocolVersion: federationProtocolVersion})
5733 yield {}))))
5850 yield {}))))
5734 yield (respond {results: results.results, authorized: True {}, error: ""}))
5851 yield (respond {results: results.results, authorized: True {}, error: ""}))
@@ -5830,8 +5947,12 @@ handleDrain = \phase authorization respond -> do
5830 sdl <- *senderDeliveries
5947 sdl <- *senderDeliveries
5831 let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
5948 let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
5832 yield {version: snapshotVersion, checksum: drainCs, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se, senderDeliveries: sdl}
5949 yield {version: snapshotVersion, checksum: drainCs, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac, senderExpirations: se, senderDeliveries: sdl}
5833− -- Sentinel-wrapped: sidecar processes should verify [SNAPSHOT_END] follows to detect truncation
5950+ -- Sentinel-wrapped: sidecar processes should verify [SNAPSHOT_END] follows to detect truncation.
5951+ -- Integrity tag appended when maintenance secret is configured (enables tamper detection on restore).
5952+ runtimeSec <- *runtimeMaintenanceSecret
5953+ let effectiveSecret = if runtimeSec != "" then runtimeSec else maintenanceSecret
5954+ let drainTag = snapshotIntegrityTag drainSnapshot.checksum effectiveSecret
5834− println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
5955+ println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot ++ (if drainTag != 0 then " [INTEGRITY:" ++ show drainTag ++ "]" else ""))
5835 println "[SNAPSHOT_END]"
5956 println "[SNAPSHOT_END]"
5836 atomic do set *prunesSinceSnapshot = 0
5957 atomic do set *prunesSinceSnapshot = 0
5837 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
5958 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
@@ -6058,7 +6179,7 @@ handleHealth = \authorization respond -> do
6058 nonceHealthy: True {},
6179 nonceHealthy: True {},
6059 draining: result.isDraining})
6180 draining: result.isDraining})
6061−zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: []}, 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, prunesSinceSnapshot: 0, totalEntries: 0}}
6182+zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: []}, 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, prunesSinceSnapshot: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0}}
6063handleStats = \authorization respond -> do
6184handleStats = \authorization respond -> do
6064 t <- now
6185 t <- now
@@ -6126,7 +6247,9 @@ handleStats = \authorization respond -> do
6126 rla <- *rateLimitedAuth
6247 rla <- *rateLimitedAuth
6127 rlk <- *rateLimitedPerKey
6248 rlk <- *rateLimitedPerKey
6128 rlf <- *rateLimitedFederation
6249 rlf <- *rateLimitedFederation
6250+ arq <- *activeRequests
6251+ parq <- *peakActiveRequests
6129− 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, pss: pss, flt: flt, flc: flc, flm: flm, svc: svc, vfc: vfc, dca: dca, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf}
6252+ 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, pss: pss, 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}
6130 -- Compute derived metrics
6253 -- Compute derived metrics
6131 let sessions = core.sc
6254 let sessions = core.sc
6132 let messages = core.tmc
6255 let messages = core.tmc
@@ -6152,7 +6275,7 @@ handleStats = \authorization respond -> do
6152 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},
6275 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},
6153 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},
6276 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},
6154 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},
6277 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},
6155− 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, prunesSinceSnapshot: counters.pss, totalEntries: sessions + messages + presence + fwds + count core.c + core.aCount + count core.ao + core.dCount + core.fnCount + count core.ap + count core.fi}})
6278+ 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, prunesSinceSnapshot: counters.pss, 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}})
6156 else yield (respond zeroStats)
6279 else yield (respond zeroStats)
6158handleStatsPeers = \authorization respond -> do
6281handleStatsPeers = \authorization respond -> do
@@ -6164,14 +6287,18 @@ handleStatsPeers = \authorization respond -> do
6164 if authorized
6287 if authorized
6166 failures <- *forwardFailures
6289 failures <- *forwardFailures
6290+ peerLats <- *peerLatencies
6167 -- Per-bucket aggregation: compute forward counts without materializing all 16 buckets.
6291 -- Per-bucket aggregation: compute forward counts without materializing all 16 buckets.
6168 -- Reuses computeForwardStats which reads one bucket at a time.
6292 -- Reuses computeForwardStats which reads one bucket at a time.
6169 fwdStats <- computeForwardStats 0 t
6293 fwdStats <- computeForwardStats 0 t
6170 let fwdCounts = fwdStats.serverCounts
6294 let fwdCounts = fwdStats.serverCounts
6171 let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
6295 let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
6296+ let peerLatFor = \server -> case headOf (do p <- peerLats; where p.server == server; yield p) of
6297+ Nothing {} -> {avg: 0, mx: 0, samples: 0}
6298+ Just {value: p} -> {avg: if p.count > 0 then p.totalMs / p.count else 0, mx: p.maxMs, samples: p.count}
6172− let peers = do f <- failures; 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}
6299+ 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}
6173 -- Include servers with pending forwards but no failure records
6300 -- Include servers with pending forwards but no failure records
6174− let fwdOnlyPeers = do c <- fwdCounts; where none failures (\f -> f.server == c.server); yield {server: c.server, failures: 0, lastFailedAt: 0, firstFailedAt: 0, backoffMs: 0, retriable: True {}, pendingForwards: c.count}
6301+ 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}
6175 yield (respond {peers: union peers fwdOnlyPeers, ok: True {}})
6302 yield (respond {peers: union peers fwdOnlyPeers, ok: True {}})
6176 else yield (respond {peers: [], ok: False {}})
6303 else yield (respond {peers: [], ok: False {}})
@@ -6278,14 +6405,17 @@ handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData de
6278 authorized <- verifyMaintenanceAuth authorization
6405 authorized <- verifyMaintenanceAuth authorization
6279 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6406 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6281− -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers,
6408+ -- Minimum version: v6 (first version with checksum integrity verification).
6282− -- version 3 = pre-ForwardEntry changes, version 4 = pre-allowedPeers, version 5 = pre-checksum
6409+ -- Pre-v6 snapshots (0-5) have no checksum and could be corrupted without detection.
6283− -- (accept all for backwards compatibility; missing fields default to empty list or 0)
6410+ -- Versions 6-10 are accepted with backward compatibility (missing fields default to []/0).
6284− if version != 0 && version != 1 && version != 2 && version != 3 && version != 4 && version != 5 && version != 6 && version != 7 && version != 8 && version != 9 && version != snapshotVersion then do
6411+ if version < 6 then do
6412+ logError ("restore: snapshot version " ++ show version ++ " is too old (minimum: 6, current: " ++ show snapshotVersion ++ "). Pre-v6 snapshots lack checksum integrity verification.")
6413+ yield (respond {ok: False {}, error: "version_too_old", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6414+ else if version > snapshotVersion then do
6285− logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
6415+ logError ("restore: snapshot version " ++ show version ++ " is newer than server (current: " ++ show snapshotVersion ++ ")")
6286 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6416 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
6288− -- Checksum verification (skip for pre-v6 snapshots where checksum defaults to 0)
6418+ -- Checksum verification (required for v6+)
6289 let expectedCs = snapshotChecksum (count presenceData) (count ackedIdsData) (count deliveredIdsData) (count forwardsData) (count forwardNoncesData)
6419 let expectedCs = snapshotChecksum (count presenceData) (count ackedIdsData) (count deliveredIdsData) (count forwardsData) (count forwardNoncesData)
6290 if checksumIn != 0 && checksumIn != expectedCs then do
6420 if checksumIn != 0 && checksumIn != expectedCs then do
6291 logError ("restore: checksum mismatch (got " ++ show checksumIn ++ ", expected " ++ show expectedCs ++ ") — snapshot may be truncated or corrupted")
6421 logError ("restore: checksum mismatch (got " ++ show checksumIn ++ ", expected " ++ show expectedCs ++ ") — snapshot may be truncated or corrupted")
@@ -6614,7 +6744,7 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
6614 cancelFedResult <- fetchFederation server (FederationCancel {
6744 cancelFedResult <- fetchFederation server (FederationCancel {
6615 messageId: messageId, toKey: to, fromServer: serverHostN,
6745 messageId: messageId, toKey: to, fromServer: serverHostN,
6616 requestTimestamp: cancelT, correlationId: cancelCorrId,
6746 requestTimestamp: cancelT, correlationId: cancelCorrId,
6617− authorization: cancelAuth})
6747+ authorization: cancelAuth, protocolVersion: federationProtocolVersion})
6618 case cancelFedResult of
6748 case cancelFedResult of
6619 Err e -> logWarn ("federation: cancel to " ++ server ++ " failed for " ++ messageId ++ ": " ++ show e)
6749 Err e -> logWarn ("federation: cancel to " ++ server ++ " failed for " ++ messageId ++ ": " ++ show e)
6620 Ok r -> when (not r.ok) (logWarn ("federation: cancel to " ++ server ++ " rejected for " ++ messageId ++ ": " ++ r.error)))))
6750 Ok r -> when (not r.ok) (logWarn ("federation: cancel to " ++ server ++ " rejected for " ++ messageId ++ ": " ++ r.error)))))
@@ -6709,7 +6839,7 @@ sendBatchBucketGroup = \senderKey prepared t dp acc ->
6709 let clampedTs = clampTimestamp entry.msg.timestamp t
6839 let clampedTs = clampTimestamp entry.msg.timestamp t
6710 if dp >= 2 then yield {messageId: entry.effectiveMsgId, status: MsDraining {}, remoteServers: [], toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: 0}
6840 if dp >= 2 then yield {messageId: entry.effectiveMsgId, status: MsDraining {}, remoteServers: [], toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: 0}
6712− sendResult <- sendMessageCore senderKey entry.msg.to entry.msg.encryptedBlob clampedTs entry.effectiveMsgId entry.msg.expiresInMs t
6842+ sendResult <- sendMessageCore senderKey entry.msg.to entry.msg.encryptedBlob clampedTs entry.effectiveMsgId entry.msg.expiresInMs (clamp 0 maxMessagePriority entry.msg.priority) t
6713 yield {messageId: entry.effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryAfterMs}
6843 yield {messageId: entry.effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: entry.msg.to, blob: entry.msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryAfterMs}
6714 sendBatchBucketGroup senderKey rest t dp (union acc [result])
6844 sendBatchBucketGroup senderKey rest t dp (union acc [result])
@@ -6826,7 +6956,7 @@ handlePingPeer = \server authorization respond -> do
6826 result <- fetchFederation server (RecvGossip {
6956 result <- fetchFederation server (RecvGossip {
6827 authorization: auth,
6957 authorization: auth,
6828 events: [], hopsRemaining: 1, fromServer: serverHostN,
6958 events: [], hopsRemaining: 1, fromServer: serverHostN,
6829− requestTimestamp: startTime, correlationId: pingCorrId})
6959+ requestTimestamp: startTime, correlationId: pingCorrId, protocolVersion: federationProtocolVersion})
6830 endTime <- now
6960 endTime <- now
6831 let latency = endTime - startTime
6961 let latency = endTime - startTime
6832 case result of
6962 case result of
@@ -7207,7 +7337,7 @@ serveInner = \req -> case req of
7207 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
7337 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
7208 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
7338 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
7209 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond
7339 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond
7210− SendMessage {to, encryptedBlob, timestamp, messageId, expiresInMs, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId expiresInMs authorization clientIp respond
7340+ SendMessage {to, encryptedBlob, timestamp, messageId, expiresInMs, priority, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId expiresInMs priority authorization clientIp respond
7211 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
7341 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
7212 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
7342 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
7213 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
7343 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
@@ -7217,12 +7347,18 @@ serveInner = \req -> case req of
7217 CancelBatch {cancels, authorization, clientIp, respond} -> handleCancelBatch cancels authorization clientIp respond
7347 CancelBatch {cancels, authorization, clientIp, respond} -> handleCancelBatch cancels authorization clientIp respond
7218 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
7348 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
7219 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
7349 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
7220− RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
7350+ RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
7351+ warnOldProtocol fromServer protocolVersion; handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
7221− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, expiresAt, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt authorization respond
7352+ ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, expiresAt, protocolVersion, authorization, respond} -> do
7353+ warnOldProtocol fromServer protocolVersion; handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt authorization respond
7222− FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
7354+ FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
7355+ warnOldProtocol fromServer protocolVersion; handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
7223− FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationCancel messageId toKey fromServer requestTimestamp correlationId authorization respond
7356+ FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
7357+ warnOldProtocol fromServer protocolVersion; handleFederationCancel messageId toKey fromServer requestTimestamp correlationId authorization respond
7224− FederationBatchCancel {cancels, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationBatchCancel cancels fromServer requestTimestamp correlationId authorization respond
7358+ FederationBatchCancel {cancels, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
7359+ warnOldProtocol fromServer protocolVersion; handleFederationBatchCancel cancels fromServer requestTimestamp correlationId authorization respond
7225− FederationBatchForward {forwards, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp correlationId authorization respond
7360+ FederationBatchForward {forwards, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> do
7361+ warnOldProtocol fromServer protocolVersion; handleFederationBatchForward forwards fromServer requestTimestamp correlationId authorization respond
7226 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
7362 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
7227 Prune {authorization, respond} -> handlePrune authorization respond
7363 Prune {authorization, respond} -> handlePrune authorization respond
7228 Drain {phase, authorization, respond} -> handleDrain phase authorization respond
7364 Drain {phase, authorization, respond} -> handleDrain phase authorization respond
@@ -7261,7 +7397,16 @@ serveInner = \req -> case req of
7261serve = \req -> do
7397serve = \req -> do
7262 startTime <- now
7398 startTime <- now
7263 reqId <- genTokenN 1
7399 reqId <- genTokenN 1
7401+ ar <- *activeRequests
7403+ set *activeRequests = newAr
7404+ peak <- *peakActiveRequests
7405+ when (newAr > peak) (set *peakActiveRequests = newAr)
7264 serveInner req
7406 serveInner req
7408+ ar <- *activeRequests
7409+ set *activeRequests = max 0 (ar - 1)
7265 endTime <- now
7410 endTime <- now
7266 let latency = endTime - startTime
7411 let latency = endTime - startTime
7267 let path = requestPath req
7412 let path = requestPath req
@@ -7412,17 +7557,22 @@ backgroundSnapshot = do
7412 dca <- *drainCompletedAt
7557 dca <- *drainCompletedAt
7413 if shutdownReq && dca > 0 then logInfo "backgroundSnapshot: shutdown complete, exiting loop"
7558 if shutdownReq && dca > 0 then logInfo "backgroundSnapshot: shutdown complete, exiting loop"
7414 else if dp < 2 then do
7559 else if dp < 2 then do
7560+ -- Split into two reads: cached counters (cheap, no bucket scans) and per-bucket
7561+ -- counts (scan 16 buckets each). Reduces atomic block width vs one large block.
7415− result <- atomic do
7562+ cached <- atomic do
7416 tmc <- *totalMessageCount
7563 tmc <- *totalMessageCount
7417 sc <- *totalSessionCount
7564 sc <- *totalSessionCount
7418 pc <- *totalPresenceCount
7565 pc <- *totalPresenceCount
7419− fwdCount <- countAllForwards 0
7420− ackedCount <- countAllAcked 0
7421 ao <- *ackOverflow
7566 ao <- *ackOverflow
7422− deliveredCount <- countAllDelivered 0
7423 ff <- *forwardFailures
7567 ff <- *forwardFailures
7424 pac <- *pendingAckCleanups
7568 pac <- *pendingAckCleanups
7425− yield {messages: tmc, sessions: sc, presence: pc, forwards: fwdCount, ackedIds: ackedCount, ackOverflow: count ao, deliveredIds: deliveredCount, forwardFailures: count ff, pendingAckCleanups: count pac}
7569+ yield {messages: tmc, sessions: sc, presence: pc, ackOverflow: count ao, forwardFailures: count ff, pendingAckCleanups: count pac}
7570+ bucketCounts <- atomic do
7571+ fwdCount <- countAllForwards 0
7572+ ackedCount <- countAllAcked 0
7573+ deliveredCount <- countAllDelivered 0
7574+ yield {forwards: fwdCount, ackedIds: ackedCount, deliveredIds: deliveredCount}
7575+ let result = {messages: cached.messages, sessions: cached.sessions, presence: cached.presence, forwards: bucketCounts.forwards, ackedIds: bucketCounts.ackedIds, ackOverflow: cached.ackOverflow, deliveredIds: bucketCounts.deliveredIds, forwardFailures: cached.forwardFailures, pendingAckCleanups: cached.pendingAckCleanups}
7426 logInfo ("snapshot: messages=" ++ show result.messages ++ " sessions=" ++ show result.sessions ++ " presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " ackOverflow=" ++ show result.ackOverflow ++ " deliveredIds=" ++ show result.deliveredIds ++ " forwardFailures=" ++ show result.forwardFailures ++ " pendingAckCleanups=" ++ show result.pendingAckCleanups)
7576 logInfo ("snapshot: messages=" ++ show result.messages ++ " sessions=" ++ show result.sessions ++ " presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " ackOverflow=" ++ show result.ackOverflow ++ " deliveredIds=" ++ show result.deliveredIds ++ " forwardFailures=" ++ show result.forwardFailures ++ " pendingAckCleanups=" ++ show result.pendingAckCleanups)
7427 t <- now
7577 t <- now
7428 -- Periodic full snapshot auto-export: emit [SNAPSHOT_DATA] JSON to stdout so a
7578 -- Periodic full snapshot auto-export: emit [SNAPSHOT_DATA] JSON to stdout so a
@@ -7455,7 +7605,10 @@ backgroundSnapshot = do
7455 yield {bp: bp, ap: ap, pac: pac, se: se, sdl: sdl}
7605 yield {bp: bp, ap: ap, pac: pac, se: se, sdl: sdl}
7456 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
7606 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
7457 let exportData = {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: extraState.bp, allowedPeers: extraState.ap, pendingAckCleanups: extraState.pac, senderExpirations: extraState.se, senderDeliveries: extraState.sdl}
7607 let exportData = {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: extraState.bp, allowedPeers: extraState.ap, pendingAckCleanups: extraState.pac, senderExpirations: extraState.se, senderDeliveries: extraState.sdl}
7608+ autoRtSec <- *runtimeMaintenanceSecret
7609+ let autoEffSec = if autoRtSec != "" then autoRtSec else maintenanceSecret
7610+ let autoTag = snapshotIntegrityTag exportCs autoEffSec
7458− println ("[SNAPSHOT_DATA] " ++ toJson exportData)
7611+ println ("[SNAPSHOT_DATA] " ++ toJson exportData ++ (if autoTag != 0 then " [INTEGRITY:" ++ show autoTag ++ "]" else ""))
7459 println "[SNAPSHOT_END]"
7612 println "[SNAPSHOT_END]"
7460 logEvent "warn" "snapshot_unencrypted" "snapshot exported to stdout — ensure log pipeline is secured"
7613 logEvent "warn" "snapshot_unencrypted" "snapshot exported to stdout — ensure log pipeline is secured"
7461 atomic do
7614 atomic do
@@ -7603,7 +7756,7 @@ main = do
7603 when federationEnabled
7756 when federationEnabled
7604 (logWarn "SECURITY: DNS rebinding risk — federation URLs resolve DNS at fetch time, not at validation time. A peer hostname passing isBadServerName could rebind to a private IP between validation and fetch. Mitigate by blocking outbound connections to RFC1918/link-local/CGNAT ranges at the firewall level.")
7757 (logWarn "SECURITY: DNS rebinding risk — federation URLs resolve DNS at fetch time, not at validation time. A peer hostname passing isBadServerName could rebind to a private IP between validation and fetch. Mitigate by blocking outbound connections to RFC1918/link-local/CGNAT ranges at the firewall level.")
7605 logInfo ("Skrepka relay server starting on port " ++ show listenPort ++ "...")
7758 logInfo ("Skrepka relay server starting on port " ++ show listenPort ++ "...")
7606− logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")
7759+ logInfo (" federation: " ++ if federationEnabled then "enabled (protocol v" ++ show federationProtocolVersion ++ ")" else "disabled")
7607 logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
7760 logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
7608 logInfo (" poll timeout: " ++ show pollTimeoutMs ++ "ms")
7761 logInfo (" poll timeout: " ++ show pollTimeoutMs ++ "ms")
7609 logInfo (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")
7762 logInfo (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")