@@ -169,7 +169,7 @@ snapshotVersion = 10 -- bumped for ForwardEntry.expiresAt field
170-- Server build version: included in /health and /stats responses for deployment verification.
170-- 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.
171-- Update this string when deploying a new build so operators can confirm which version is running.
172−serverBuildVersion = "2026-04-25"
172+serverBuildVersion = "2026-04-25a"
174-- Snapshot checksum: deterministic hash of entry counts for integrity verification.
174-- Snapshot checksum: deterministic hash of entry counts for integrity verification.
175-- Uses multiplicative hash to distinguish permutations of the same counts
175-- Uses multiplicative hash to distinguish permutations of the same counts
@@ -324,7 +324,10 @@ impl FromJSON GossipEventType where
324 else GossipUnknown {}
324 else GossipUnknown {}
326-- Wire format for gossip events sent between servers
326-- Wire format for gossip events sent between servers
327+-- seenBy: list of server names that have already received this event.
328+-- Used to suppress redundant re-broadcast: when re-gossiping, skip peers already in seenBy.
329+-- Pre-v11 peers send [] (Knot ignores unknown fields on parse, sends [] for missing lists).
327−type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex, server: ServerName, timestamp: Timestamp, ttl: Int<S>}
330+type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex, server: ServerName, timestamp: Timestamp, ttl: Int<S>, seenBy: [Text]}
329-- A gossip event that passed validation (timestamp not in future, TTL > 0, not expired)
332-- A gossip event that passed validation (timestamp not in future, TTL > 0, not expired)
330type ValidGossipEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Timestamp, timestamp: Timestamp, ttl: Int<S>}
333type ValidGossipEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Timestamp, timestamp: Timestamp, ttl: Int<S>}
@@ -845,6 +848,7 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
845*cachedHealthSnapshot : Bool
848*cachedHealthSnapshot : Bool
846*cachedHealthWatchdog : Bool
849*cachedHealthWatchdog : Bool
847*cachedHealthNonce : Bool
850*cachedHealthNonce : Bool
851+*cachedHealthCapacityWarning : Bool
849-- ============================================
853-- ============================================
@@ -874,7 +878,7 @@ route AuthApi where
874route ClientApi where
878route ClientApi where
875 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
879 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
876 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
880 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
877− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}]}
881+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}]}
879 /messages
883 /messages
880 -- Send an E2E encrypted message to a recipient pubkey.
884 -- Send an E2E encrypted message to a recipient pubkey.
@@ -945,7 +949,7 @@ route FederationApi where
945 = ForwardMessage
949 = ForwardMessage
946 -- Acknowledge delivery of messages back to the origin server
950 -- Acknowledge delivery of messages back to the origin server
947 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /ack headers {authorization: Text}
951 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /ack headers {authorization: Text}
948− -> {ok: Bool, error: Text, ackedCount: Int}
952+ -> {ok: Bool, error: Text, ackedCount: Int, results: [{messageId: MessageIdHex, found: Bool}]}
949 = FederationAck
953 = FederationAck
950 -- Cancel (retract) a previously forwarded message. The origin server sends this
954 -- Cancel (retract) a previously forwarded message. The origin server sends this
951 -- when the sender cancels a message via /messages/cancel. The receiving server
955 -- when the sender cancels a message via /messages/cancel. The receiving server
@@ -957,6 +961,12 @@ route FederationApi where
957 POST {cancels: [{messageId: MessageIdHex, toKey: PubkeyHex}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /cancel-batch headers {authorization: Text}
961 POST {cancels: [{messageId: MessageIdHex, toKey: PubkeyHex}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /cancel-batch headers {authorization: Text}
958 -> {ok: Bool, error: Text, results: [BatchCancelResult]}
962 -> {ok: Bool, error: Text, results: [BatchCancelResult]}
959 = FederationBatchCancel
963 = FederationBatchCancel
964+ -- Batch forward: forward multiple messages in one request (reduces HTTP round-trips).
965+ -- Each message is processed independently; partial success is possible.
966+ -- 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}
968+ -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}]}
969+ = FederationBatchForward
961-- Federation callback verification: the receiving server calls back to the sender
971-- Federation callback verification: the receiving server calls back to the sender
962-- to confirm a request token is genuine. No shared secret needed.
972-- to confirm a request token is genuine. No shared secret needed.
@@ -1067,7 +1077,7 @@ route MaintenanceApi where
1067 = RotateSecret
1077 = RotateSecret
1069route HealthApi where
1079route HealthApi where
1070− GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, draining: Bool} = Health
1080+ GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, draining: Bool} = Health
1072-- Returns counts of in-memory collections (requires maintenance secret)
1082-- Returns counts of in-memory collections (requires maintenance secret)
1073-- TODO: If Knot adds raw text response support, add a /metrics endpoint
1083-- TODO: If Knot adds raw text response support, add a /metrics endpoint
@@ -1077,7 +1087,7 @@ route HealthApi where
1077-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1087-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1078route StatsApi where
1088route StatsApi where
1079 GET /stats headers {authorization: Text}
1089 GET /stats headers {authorization: Text}
1080− -> {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}, 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}}
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}}
1082 GET /stats/peers headers {authorization: Text}
1092 GET /stats/peers headers {authorization: Text}
1083 -> {peers: [PeerStatus], ok: Bool}
1093 -> {peers: [PeerStatus], ok: Bool}
@@ -1118,7 +1128,10 @@ logError = \msg -> println ("[ERROR] " ++ msg)
1118-- when msg contains user-controlled data (e.g., error strings, server names).
1128-- when msg contains user-controlled data (e.g., error strings, server names).
1119logEvent = \level event msg -> println ("{\"level\":\"" ++ escapeJsonString level ++ "\",\"event\":\"" ++ escapeJsonString event ++ "\",\"msg\":\"" ++ escapeJsonString msg ++ "\"}")
1129logEvent = \level event msg -> println ("{\"level\":\"" ++ escapeJsonString level ++ "\",\"event\":\"" ++ escapeJsonString event ++ "\",\"msg\":\"" ++ escapeJsonString msg ++ "\"}")
1121−-- Escape backslashes and double quotes in a string for safe JSON embedding.
1131+-- Escape special characters for safe JSON string embedding.
1132+-- Handles: backslash, double quote, and control characters (\n, \r, \t, \b, \f).
1133+-- Non-printable characters outside this set are stripped (same allowlist as
1134+-- stripNonPrintable) to prevent log injection via error strings from federation peers.
1122-- Processes one character at a time via tail recursion (TCO-safe in Knot).
1135-- Processes one character at a time via tail recursion (TCO-safe in Knot).
1123escapeJsonString : Text -> Text
1136escapeJsonString : Text -> Text
1124escapeJsonString = \s ->
1137escapeJsonString = \s ->
@@ -1126,7 +1139,11 @@ escapeJsonString = \s ->
1126 else let c = take 1 s in
1139 else let c = take 1 s in
1127 if c == "\\" then "\\\\" ++ escapeJsonString (drop 1 s)
1140 if c == "\\" then "\\\\" ++ escapeJsonString (drop 1 s)
1128 else if c == "\"" then "\\\"" ++ escapeJsonString (drop 1 s)
1141 else if c == "\"" then "\\\"" ++ escapeJsonString (drop 1 s)
1129− else c ++ escapeJsonString (drop 1 s)
1142+ else if c == "\n" then "\\n" ++ escapeJsonString (drop 1 s)
1143+ else if c == "\r" then "\\r" ++ escapeJsonString (drop 1 s)
1144+ else if c == "\t" then "\\t" ++ escapeJsonString (drop 1 s)
1145+ else if contains c printableAscii then c ++ escapeJsonString (drop 1 s)
1146+ else escapeJsonString (drop 1 s)
1131any : [a] -> (a -> Bool) -> Bool
1148any : [a] -> (a -> Bool) -> Bool
1132any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
1149any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
@@ -1137,6 +1154,9 @@ whenOk = \result action -> case result of
1137 Ok _ -> action
1154 Ok _ -> action
1138 Err _ -> yield {}
1155 Err _ -> yield {}
1139clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x
1156clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x
1157+-- Union that skips empty lists (avoids allocation overhead on empty buckets).
1158+-- When one side is empty, returns the other without copying.
1159+unionNE = \a b -> if count a == 0 then b else if count b == 0 then a else union a b
1140min = \a b -> if a < b then a else b
1160min = \a b -> if a < b then a else b
1141max = \a b -> if a > b then a else b
1161max = \a b -> if a > b then a else b
1142headOf = \items -> fold (\acc x -> case acc of
1162headOf = \items -> fold (\acc x -> case acc of
@@ -1606,7 +1626,7 @@ readAllSessions = do
1606 s13 <- *sessionBucket13
1626 s13 <- *sessionBucket13
1607 s14 <- *sessionBucket14
1627 s14 <- *sessionBucket14
1608 s15 <- *sessionBucket15
1628 s15 <- *sessionBucket15
1609− yield (union s0 (union s1 (union s2 (union s3 (union s4 (union s5 (union s6 (union s7 (union s8 (union s9 (union s10 (union s11 (union s12 (union s13 (union s14 s15)))))))))))))))
1629+ yield (unionNE s0 (unionNE s1 (unionNE s2 (unionNE s3 (unionNE s4 (unionNE s5 (unionNE s6 (unionNE s7 (unionNE s8 (unionNE s9 (unionNE s10 (unionNE s11 (unionNE s12 (unionNE s13 (unionNE s14 s15)))))))))))))))
1611-- Count sessions across all 16 buckets without materializing the full list
1631-- Count sessions across all 16 buckets without materializing the full list
1612countAllSessions : IO Int
1632countAllSessions : IO Int
@@ -1837,7 +1857,7 @@ readAllPresence = do
1837 p13 <- *presenceBucket13
1857 p13 <- *presenceBucket13
1838 p14 <- *presenceBucket14
1858 p14 <- *presenceBucket14
1839 p15 <- *presenceBucket15
1859 p15 <- *presenceBucket15
1840− yield (union p0 (union p1 (union p2 (union p3 (union p4 (union p5 (union p6 (union p7 (union p8 (union p9 (union p10 (union p11 (union p12 (union p13 (union p14 p15)))))))))))))))
1860+ yield (unionNE p0 (unionNE p1 (unionNE p2 (unionNE p3 (unionNE p4 (unionNE p5 (unionNE p6 (unionNE p7 (unionNE p8 (unionNE p9 (unionNE p10 (unionNE p11 (unionNE p12 (unionNE p13 (unionNE p14 p15)))))))))))))))
1842countAllPresence : IO Int
1862countAllPresence : IO Int
1843countAllPresence = do
1863countAllPresence = do
@@ -1997,7 +2017,8 @@ writeMessageBucket = \b msgs ->
1997 else if b == 14 then set *messageBucket14 = msgs
2017 else if b == 14 then set *messageBucket14 = msgs
1998 else set *messageBucket15 = msgs
2018 else set *messageBucket15 = msgs
2000−-- Read all 16 message buckets (for cold-path operations: federation ack, forward candidates, stats)
2020+-- Read all 16 message buckets (for cold-path operations: federation ack, forward candidates, stats).
2021+-- Uses unionNE to skip empty buckets (avoids allocation overhead on sparse servers).
2001readAllMessages : IO [Message]
2022readAllMessages : IO [Message]
2002readAllMessages = do
2023readAllMessages = do
2003 m0 <- *messageBucket0
2024 m0 <- *messageBucket0
@@ -2016,7 +2037,7 @@ readAllMessages = do
2016 m13 <- *messageBucket13
2037 m13 <- *messageBucket13
2017 m14 <- *messageBucket14
2038 m14 <- *messageBucket14
2018 m15 <- *messageBucket15
2039 m15 <- *messageBucket15
2019− yield (union m0 (union m1 (union m2 (union m3 (union m4 (union m5 (union m6 (union m7 (union m8 (union m9 (union m10 (union m11 (union m12 (union m13 (union m14 m15)))))))))))))))
2040+ yield (unionNE m0 (unionNE m1 (unionNE m2 (unionNE m3 (unionNE m4 (unionNE m5 (unionNE m6 (unionNE m7 (unionNE m8 (unionNE m9 (unionNE m10 (unionNE m11 (unionNE m12 (unionNE m13 (unionNE m14 m15)))))))))))))))
2021-- Apply a filter/transform to every message bucket
2042-- Apply a filter/transform to every message bucket
2022transformAllMessageBuckets : ([Message] -> [Message]) -> IO ()
2043transformAllMessageBuckets : ([Message] -> [Message]) -> IO ()
@@ -2163,17 +2184,32 @@ adjustRecipientCount = \counts pubkey delta ->
2163 else if delta > 0 then union counts [{pubkey: pubkey, count: delta}]
2184 else if delta > 0 then union counts [{pubkey: pubkey, count: delta}]
2164 else counts
2185 else counts
2166−-- Count items per toKey in a single pass: fold accumulates {pubkey, count} pairs,
2187+-- Count items per toKey using sort-then-scan: O(n log n) sort + O(n) linear scan.
2167−-- incrementing existing entries or inserting new ones.
2188+-- Replaces the previous O(n*k) fold-with-any approach. For a bucket with 6000 messages
2168−-- NOTE: O(n*k) where k = unique recipients due to `any acc` check inside fold.
2189+-- and 500 unique recipients, this reduces comparisons from ~3M to ~80k.
2169−-- Acceptable for per-bucket data (n ≤ maxMessages/16). A map type would make this O(n).
2190+-- sortBy groups equal toKeys together; countSortedGroups scans consecutive equal keys.
2170−-- NOTE: union [new] acc prepends (O(1) per step, O(n) total) vs union acc [new] which
2171−-- copies acc (O(|acc|) per step, O(n*k) total). Safe here because this is a set (order irrelevant).
2172countByRecipient = \items ->
2191countByRecipient = \items ->
2173− fold (\acc x ->
2192+ let sorted = sortBy (\m -> m.toKey) items in
2174− if any acc (\c -> c.pubkey == x.toKey)
2193+ countSortedGroups sorted
2175− then do c <- acc; yield (if c.pubkey == x.toKey then {pubkey: c.pubkey, count: c.count + 1} else c)
2194+
2176− else union [{pubkey: x.toKey, count: 1}] acc) [] items
2195+-- 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
2197+-- group's count or emits the group and starts a new one.
2198+countSortedGroups = \items ->
2199+ case headOf items of
2201+ Just {value: first} ->
2202+ let rest = do x <- items; where x != first; yield x in
2203+ countSortedGroupsInner rest first.toKey 1 []
2205+countSortedGroupsInner = \items currentKey currentCount acc ->
2206+ case headOf items of
2207+ Nothing {} -> union acc [{pubkey: currentKey, count: currentCount}]
2209+ let rest = do r <- items; where r != x; yield r in
2210+ if x.toKey == currentKey
2211+ then countSortedGroupsInner rest currentKey (currentCount + 1) acc
2212+ else countSortedGroupsInner rest x.toKey 1 (union acc [{pubkey: currentKey, count: currentCount}])
2178-- Batch-decrement recipient counts: subtract per-key removal counts in a single pass
2214-- Batch-decrement recipient counts: subtract per-key removal counts in a single pass
2179-- instead of calling adjustRecipientCount in a loop.
2215-- instead of calling adjustRecipientCount in a loop.
@@ -2327,7 +2363,7 @@ readAllNonces = do
2327 n13 <- *nonceBucket13
2363 n13 <- *nonceBucket13
2328 n14 <- *nonceBucket14
2364 n14 <- *nonceBucket14
2329 n15 <- *nonceBucket15
2365 n15 <- *nonceBucket15
2330− yield (union n0 (union n1 (union n2 (union n3 (union n4 (union n5 (union n6 (union n7 (union n8 (union n9 (union n10 (union n11 (union n12 (union n13 (union n14 n15)))))))))))))))
2366+ yield (unionNE n0 (unionNE n1 (unionNE n2 (unionNE n3 (unionNE n4 (unionNE n5 (unionNE n6 (unionNE n7 (unionNE n8 (unionNE n9 (unionNE n10 (unionNE n11 (unionNE n12 (unionNE n13 (unionNE n14 n15)))))))))))))))
2332countAllNonces : Int -> IO Int
2368countAllNonces : Int -> IO Int
2333countAllNonces = \b ->
2369countAllNonces = \b ->
@@ -2446,7 +2482,7 @@ readAllForwards = do
2446 f13 <- *fwdBucket13
2482 f13 <- *fwdBucket13
2447 f14 <- *fwdBucket14
2483 f14 <- *fwdBucket14
2448 f15 <- *fwdBucket15
2484 f15 <- *fwdBucket15
2449− yield (union f0 (union f1 (union f2 (union f3 (union f4 (union f5 (union f6 (union f7 (union f8 (union f9 (union f10 (union f11 (union f12 (union f13 (union f14 f15)))))))))))))))
2485+ yield (unionNE f0 (unionNE f1 (unionNE f2 (unionNE f3 (unionNE f4 (unionNE f5 (unionNE f6 (unionNE f7 (unionNE f8 (unionNE f9 (unionNE f10 (unionNE f11 (unionNE f12 (unionNE f13 (unionNE f14 f15)))))))))))))))
2451countAllForwards : Int -> IO Int
2487countAllForwards : Int -> IO Int
2452countAllForwards = \b ->
2488countAllForwards = \b ->
@@ -2588,7 +2624,7 @@ readAllAcked = do
2588 a4 <- *ackedBucket4; a5 <- *ackedBucket5; a6 <- *ackedBucket6; a7 <- *ackedBucket7
2624 a4 <- *ackedBucket4; a5 <- *ackedBucket5; a6 <- *ackedBucket6; a7 <- *ackedBucket7
2589 a8 <- *ackedBucket8; a9 <- *ackedBucket9; a10 <- *ackedBucket10; a11 <- *ackedBucket11
2625 a8 <- *ackedBucket8; a9 <- *ackedBucket9; a10 <- *ackedBucket10; a11 <- *ackedBucket11
2590 a12 <- *ackedBucket12; a13 <- *ackedBucket13; a14 <- *ackedBucket14; a15 <- *ackedBucket15
2626 a12 <- *ackedBucket12; a13 <- *ackedBucket13; a14 <- *ackedBucket14; a15 <- *ackedBucket15
2591− yield (union a0 (union a1 (union a2 (union a3 (union a4 (union a5 (union a6 (union a7 (union a8 (union a9 (union a10 (union a11 (union a12 (union a13 (union a14 a15)))))))))))))))
2627+ yield (unionNE a0 (unionNE a1 (unionNE a2 (unionNE a3 (unionNE a4 (unionNE a5 (unionNE a6 (unionNE a7 (unionNE a8 (unionNE a9 (unionNE a10 (unionNE a11 (unionNE a12 (unionNE a13 (unionNE a14 a15)))))))))))))))
2593writeAllAcked : [AckedEntry] -> IO ()
2629writeAllAcked : [AckedEntry] -> IO ()
2594writeAllAcked = \entries -> do
2630writeAllAcked = \entries -> do
@@ -2662,7 +2698,7 @@ readAllDelivered = do
2662 d4 <- *delivBucket4; d5 <- *delivBucket5; d6 <- *delivBucket6; d7 <- *delivBucket7
2698 d4 <- *delivBucket4; d5 <- *delivBucket5; d6 <- *delivBucket6; d7 <- *delivBucket7
2663 d8 <- *delivBucket8; d9 <- *delivBucket9; d10 <- *delivBucket10; d11 <- *delivBucket11
2699 d8 <- *delivBucket8; d9 <- *delivBucket9; d10 <- *delivBucket10; d11 <- *delivBucket11
2664 d12 <- *delivBucket12; d13 <- *delivBucket13; d14 <- *delivBucket14; d15 <- *delivBucket15
2700 d12 <- *delivBucket12; d13 <- *delivBucket13; d14 <- *delivBucket14; d15 <- *delivBucket15
2665− yield (union d0 (union d1 (union d2 (union d3 (union d4 (union d5 (union d6 (union d7 (union d8 (union d9 (union d10 (union d11 (union d12 (union d13 (union d14 d15)))))))))))))))
2701+ yield (unionNE d0 (unionNE d1 (unionNE d2 (unionNE d3 (unionNE d4 (unionNE d5 (unionNE d6 (unionNE d7 (unionNE d8 (unionNE d9 (unionNE d10 (unionNE d11 (unionNE d12 (unionNE d13 (unionNE d14 d15)))))))))))))))
2667writeAllDelivered : [DeliveredEntry] -> IO ()
2703writeAllDelivered : [DeliveredEntry] -> IO ()
2668writeAllDelivered = \entries -> do
2704writeAllDelivered = \entries -> do
@@ -3266,7 +3302,11 @@ forEachBatch = \items batchSize process ->
3267sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {}
3303sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {}
3268sendBatchedGossipToServer = \server allEvents hops ->
3304sendBatchedGossipToServer = \server allEvents hops ->
3305+ -- 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.
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]}
3308+ in if count unseen == 0 then yield {}
3269− forEachBatch allEvents maxGossipEvents (\batch -> do
3309+ else forEachBatch unseen maxGossipEvents (\batch -> do
3270 t <- now
3310 t <- now
3271 auth <- generateFederationAuth "gossip" server
3311 auth <- generateFederationAuth "gossip" server
3272 corrId <- genTokenN 1
3312 corrId <- genTokenN 1
@@ -3295,10 +3335,13 @@ broadcastOfflineGossip = \t peerServers expiredLocal ->
3295 sendBatchedGossip peerServers gossipEvents gossipOriginHops)
3335 sendBatchedGossip peerServers gossipEvents gossipOriginHops)
3338+-- Build re-broadcast gossip events from validated entries. The seenBy list includes
3339+-- the fromServer (who sent us these events) and ourselves (serverHostN), so downstream
3340+-- peers that already have this gossip via either path won't receive it again.
3298−effectiveGossipEvents = \entries appliedKeys eventType -> do
3341+effectiveGossipEvents = \entries appliedKeys eventType seenBy -> do
3299 e <- entries
3342 e <- entries
3300 where any appliedKeys (\a -> samePubkeyServer a e)
3343 where any appliedKeys (\a -> samePubkeyServer a e)
3301− yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
3344+ yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: seenBy}
3303mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
3346mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
3304 let trustedDeduped = trustedGossipEntries deduped presence fromServer t in
3347 let trustedDeduped = trustedGossipEntries deduped presence fromServer t in
@@ -3561,7 +3604,7 @@ gossipOfflineIfLastSession = \r label -> do
3562 logInfo (label ++ ": last session for " ++ r.pubkey ++ ", gossiping offline")
3605 logInfo (label ++ ": last session for " ++ r.pubkey ++ ", gossiping offline")
3563 when (federationEnabled && count r.peerServers > 0)
3606 when (federationEnabled && count r.peerServers > 0)
3564− (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
3607+ (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]}] gossipOriginHops)))
3566removeForward = \mid srv toKey -> atomic do
3609removeForward = \mid srv toKey -> atomic do
3567 let b = pubkeyBucket toKey
3610 let b = pubkeyBucket toKey
@@ -3671,36 +3714,65 @@ incrementForwardRetries = \messageId toServer toKey -> atomic do
3671 logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
3714 logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
3672 else writeFwdBucket b (do f <- fwdBkt; yield (if f.messageId == messageId && f.toServer == toServer then {messageId: f.messageId, toServer: f.toServer, createdAt: f.createdAt, toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, retries: f.retries + 1, expiresAt: f.expiresAt} else f))
3715 else writeFwdBucket b (do f <- fwdBkt; yield (if f.messageId == messageId && f.toServer == toServer then {messageId: f.messageId, toServer: f.toServer, createdAt: f.createdAt, toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, retries: f.retries + 1, expiresAt: f.expiresAt} else f))
3717+-- Retry forwards for a server using batch forward (1 HTTP round-trip per batch of maxBatchSize).
3718+-- Falls back to single forward if the peer returns an HTTP-level error (old peer without batch support).
3674retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
3719retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
3675retryForwardsForServer = \entries remaining ->
3720retryForwardsForServer = \entries remaining ->
3676− if remaining <= 0
3721+ if remaining <= 0 || count entries == 0 then yield {}
3677− then yield {}
3722+ else do
3678− else case headOf entries of
3723+ let server = case headOf entries of Just {value: f} -> f.toServer; Nothing {} -> ""
3679− Nothing {} -> yield {}
3724+ forEachBatch entries maxBatchSize (\batch -> do
3680− Just {value: f} -> do
3681− let rest = do x <- entries; where x.messageId != f.messageId; yield x
3682− t <- now
3725+ t <- now
3683− auth <- generateFederationAuth "forward" f.toServer
3726+ auth <- generateFederationAuth "forward" server
3684− corrId <- genTokenN 1
3727+ corrId <- genTokenN 1
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}
3685− result <- fetchFederation f.toServer (ForwardMessage {
3729+ result <- fetchFederation server (FederationBatchForward {
3686− toKey: f.toKey, encryptedBlob: f.encryptedBlob,
3687− timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
3688− fromServer: serverHostN, requestTimestamp: t,
3730+ forwards: batchPayload, fromServer: serverHostN,
3689− authorization: auth, correlationId: corrId, expiresAt: f.expiresAt})
3731+ requestTimestamp: t, authorization: auth, correlationId: corrId})
3690− case result of
3732+ case result of
3691− Err e -> do
3733+ Err e -> do
3692− logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
3734+ logWarn ("federation: batch forward to " ++ server ++ " failed: " ++ show e)
3693− recordFailure f.toServer
3735+ recordFailure server
3694− incrementForwardRetries f.messageId f.toServer f.toKey
3736+ -- Increment retries for all entries in this batch
3695− -- Network error: stop retrying this server
3737+ retryForwardsBatchIncrement batch server
3696− Ok response -> do
3738+ Ok response -> do
3697− if response.ok then do
3739+ if response.ok then do
3698− removeForward f.messageId f.toServer f.toKey
3699− clearFailure f.toServer
3740+ clearFailure server
3741+ -- Process per-message results: remove succeeded, increment retries for failed
3742+ retryForwardsBatchResults batch response.results server
3700− else do
3743+ else do
3701− logWarn ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)
3744+ logWarn ("federation: batch forward to " ++ server ++ " rejected: " ++ response.error)
3702− incrementForwardRetries f.messageId f.toServer f.toKey
3745+ -- Peer may not support batch forward; increment retries
3703− retryForwardsForServer rest (remaining - 1)
3746+ retryForwardsBatchIncrement batch server)
3748+-- Increment retry counts for all entries in a failed batch
3749+retryForwardsBatchIncrement = \batch server ->
3750+ case headOf batch of
3751+ Nothing {} -> yield {}
3752+ Just {value: f} -> do
3753+ let rest = do x <- batch; where x.messageId != f.messageId; yield x
3754+ incrementForwardRetries f.messageId server f.toKey
3755+ retryForwardsBatchIncrement rest server
3757+-- Process per-message results from a batch forward response
3758+retryForwardsBatchResults = \batch results server ->
3759+ case headOf batch of
3760+ Nothing {} -> yield {}
3761+ Just {value: f} -> do
3762+ let rest = do x <- batch; where x.messageId != f.messageId; yield x
3763+ let msgResult = headOf (do r <- results; where r.messageId == f.messageId; yield r)
3767+ removeForward f.messageId server f.toKey
3768+ retryForwardsBatchResults rest results server
3770+ incrementForwardRetries f.messageId server f.toKey
3771+ retryForwardsBatchResults rest results server
3773+ -- No result for this message (shouldn't happen) — increment retry
3774+ incrementForwardRetries f.messageId server f.toKey
3775+ retryForwardsBatchResults rest results server
3705-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
3777-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
3706-- Circuit breaker: skips entirely when too many peers are failing (likely local network issue).
3778-- Circuit breaker: skips entirely when too many peers are failing (likely local network issue).
@@ -3977,6 +4049,22 @@ pruneAndRetry = \t -> do
3977 nonceCount <- countAllNonces 0
4049 nonceCount <- countAllNonces 0
3978 when (nonceCount * 10 >= maxForwardNonces * 9)
4050 when (nonceCount * 10 >= maxForwardNonces * 9)
3979 (logWarn ("forward nonces at " ++ show (nonceCount * 100 / maxForwardNonces) ++ "% capacity (" ++ show nonceCount ++ "/" ++ show maxForwardNonces ++ ") — non-expired eviction may create replay windows, consider increasing maxForwardNonces"))
4051 (logWarn ("forward nonces at " ++ show (nonceCount * 100 / maxForwardNonces) ++ "% capacity (" ++ show nonceCount ++ "/" ++ show maxForwardNonces ++ ") — non-expired eviction may create replay windows, consider increasing maxForwardNonces"))
4052+ -- Capacity warnings at 75%: structured log events for monitoring pipelines.
4053+ -- Fires before the 90% health check threshold, giving operators early warning.
4054+ capacityCounts <- atomic do
4055+ tmc <- *totalMessageCount
4056+ sc <- *totalSessionCount
4057+ pc <- *totalPresenceCount
4058+ fc <- *totalForwardCount
4059+ yield {messages: tmc, sessions: sc, presence: pc, forwards: fc}
4060+ when (capacityCounts.messages * 4 >= maxMessages * 3)
4061+ (logEvent "warn" "capacity_warning" ("messages at " ++ show (capacityCounts.messages * 100 / maxMessages) ++ "% (" ++ show capacityCounts.messages ++ "/" ++ show maxMessages ++ ")"))
4062+ when (capacityCounts.sessions * 4 >= maxSessions * 3)
4063+ (logEvent "warn" "capacity_warning" ("sessions at " ++ show (capacityCounts.sessions * 100 / maxSessions) ++ "% (" ++ show capacityCounts.sessions ++ "/" ++ show maxSessions ++ ")"))
4064+ when (capacityCounts.presence * 4 >= maxPresenceEntries * 3)
4065+ (logEvent "warn" "capacity_warning" ("presence at " ++ show (capacityCounts.presence * 100 / maxPresenceEntries) ++ "% (" ++ show capacityCounts.presence ++ "/" ++ show maxPresenceEntries ++ ")"))
4066+ when (capacityCounts.forwards * 4 >= maxForwardsEntries * 3)
4067+ (logEvent "warn" "capacity_warning" ("forwards at " ++ show (capacityCounts.forwards * 100 / maxForwardsEntries) ++ "% (" ++ show capacityCounts.forwards ++ "/" ++ show maxForwardsEntries ++ ")"))
3980 -- Counter reconciliation: reconcile all counters every prune cycle.
4068 -- Counter reconciliation: reconcile all counters every prune cycle.
3981 -- Presence and session counters are cheap (just counting list lengths across buckets).
4069 -- Presence and session counters are cheap (just counting list lengths across buckets).
3982 -- Message reconciliation is heavier (reads all messages, rebuilds per-recipient counts)
4070 -- Message reconciliation is heavier (reads all messages, rebuilds per-recipient counts)
@@ -4093,7 +4181,7 @@ pruneAndRetry = \t -> do
4093 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
4181 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
4094 -- trigger offline gossip when they expire (avoids wasted federation traffic)
4182 -- trigger offline gossip when they expire (avoids wasted federation traffic)
4095 when result.shouldRefresh (do
4183 when result.shouldRefresh (do
4096− let refreshEvents = do p <- result.extendedActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000)}
4184+ let refreshEvents = do p <- result.extendedActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000), seenBy: [serverHostN]}
4097 when (count result.refreshPeers > 0 && count refreshEvents > 0)
4185 when (count result.refreshPeers > 0 && count refreshEvents > 0)
4098 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
4186 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
4099 -- Update cached health state so /health returns fresh data even when rate-limited.
4187 -- Update cached health state so /health returns fresh data even when rate-limited.
@@ -4202,7 +4290,8 @@ pollNewEvents = \pubkey clampedCursor t -> do
4202 -- (via nextSeqTs) and union appends new messages at the end. Kept as a defensive
4290 -- (via nextSeqTs) and union appends new messages at the end. Kept as a defensive
4203 -- measure in case bucket ordering assumptions are ever violated (e.g., by a future
4291 -- measure in case bucket ordering assumptions are ever violated (e.g., by a future
4204 -- bulk-import feature or runtime changes to list representation).
4292 -- bulk-import feature or runtime changes to list representation).
4205− let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages bucket pubkey clampedCursor t))
4293+ let allPollable = sortBy (\m -> m.receivedAt) (pollableMessages bucket pubkey clampedCursor t)
4294+ let limitedMsgs = takeRelation maxPollEvents allPollable
4206 let evts = do
4295 let evts = do
4207 m <- limitedMsgs
4296 m <- limitedMsgs
4208 yield {id: m.messageId, eventType: "message",
4297 yield {id: m.messageId, eventType: "message",
@@ -4210,7 +4299,10 @@ pollNewEvents = \pubkey clampedCursor t -> do
4210 encryptedBlob: m.encryptedBlob,
4299 encryptedBlob: m.encryptedBlob,
4211 timestamp: m.timestamp}}
4300 timestamp: m.timestamp}}
4212 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
4301 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
4302+ -- hasMore: tells clients there are additional messages beyond this batch,
4303+ -- so they can immediately re-poll without waiting for the next heartbeat.
4304+ let hasMore = count allPollable > count limitedMsgs
4213− yield {events: evts, cursor: lastReceivedAt}
4305+ yield {events: evts, cursor: lastReceivedAt, hasMore: hasMore}
4215-- Execute a single long-poll cycle:
4307-- Execute a single long-poll cycle:
4216-- 1. Authenticate and ack messages up to cursor (atomic)
4308-- 1. Authenticate and ack messages up to cursor (atomic)
@@ -4240,7 +4332,7 @@ pollOnce = \token cursor clientIp -> do
4240 startSignal <- readMessageSignal pubkey
4332 startSignal <- readMessageSignal pubkey
4241 yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, startSignal: startSignal, pendingAcks: ackResult.pendingAcks, evictedEntries: ackResult.evictedEntries}})
4333 yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, startSignal: startSignal, pendingAcks: ackResult.pendingAcks, evictedEntries: ackResult.evictedEntries}})
4242 case setup of
4334 case setup of
4243− Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: [], pubkey: "", pollSlotFull: False {}}
4335+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: [], pubkey: "", pollSlotFull: False {}, hasMore: False {}}
4244 Just {value: s} -> do
4336 Just {value: s} -> do
4245 -- Acquire poll slot (limit concurrent polls per key)
4337 -- Acquire poll slot (limit concurrent polls per key)
4246 acquired <- atomic do
4338 acquired <- atomic do
@@ -4255,7 +4347,7 @@ pollOnce = \token cursor clientIp -> do
4255 yield (True {})
4347 yield (True {})
4256 if not acquired
4348 if not acquired
4257 -- Poll slot exhausted: tell client to back off for one poll cycle instead of busy-retrying
4349 -- Poll slot exhausted: tell client to back off for one poll cycle instead of busy-retrying
4258− then yield {authorized: True {}, events: [], cursor: cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries, pubkey: s.pubkey, pollSlotFull: True {}}
4350+ then yield {authorized: True {}, events: [], cursor: cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries, pubkey: s.pubkey, pollSlotFull: True {}, hasMore: False {}}
4260 pollResult <- do
4352 pollResult <- do
4261 -- Phase 1: check for existing messages (atomic read)
4353 -- Phase 1: check for existing messages (atomic read)
@@ -4264,11 +4356,11 @@ pollOnce = \token cursor clientIp -> do
4264 if count (pollableMessages bucket s.pubkey s.clampedCursor t) > 0
4356 if count (pollableMessages bucket s.pubkey s.clampedCursor t) > 0
4266 r <- pollNewEvents s.pubkey s.clampedCursor t
4358 r <- pollNewEvents s.pubkey s.clampedCursor t
4267− yield (Just {value: {events: r.events, cursor: r.cursor}})
4359+ yield (Just {value: {events: r.events, cursor: r.cursor, hasMore: r.hasMore}})
4268 else yield (Nothing {})
4360 else yield (Nothing {})
4269 case initialResult of
4361 case initialResult of
4270 Just {value: result} ->
4362 Just {value: result} ->
4271− yield {authorized: True {}, events: result.events, cursor: result.cursor}
4363+ yield {authorized: True {}, events: result.events, cursor: result.cursor, hasMore: result.hasMore}
4272 Nothing {} -> do
4364 Nothing {} -> do
4273 -- Phase 2: wait for signal or heartbeat
4365 -- Phase 2: wait for signal or heartbeat
4274 atomic do
4366 atomic do
@@ -4284,15 +4376,15 @@ pollOnce = \token cursor clientIp -> do
4284 t2 <- now
4376 t2 <- now
4285 result <- atomic do
4377 result <- atomic do
4286 dp <- *drainPhase
4378 dp <- *drainPhase
4287− if dp >= 2 then yield {authorized: True {}, events: [], cursor: s.clampedCursor}
4379+ if dp >= 2 then yield {authorized: True {}, events: [], cursor: s.clampedCursor, hasMore: False {}}
4289 sessions2 <- readSessionBucket bucket
4381 sessions2 <- readSessionBucket bucket
4290 if not (reauthenticateSession sessions2 token s.pubkey t2)
4382 if not (reauthenticateSession sessions2 token s.pubkey t2)
4291− then yield {authorized: False {}, events: [], cursor: s.clampedCursor}
4383+ then yield {authorized: False {}, events: [], cursor: s.clampedCursor, hasMore: False {}}
4293 r <- pollNewEvents s.pubkey s.clampedCursor t2
4385 r <- pollNewEvents s.pubkey s.clampedCursor t2
4294− yield {authorized: True {}, events: r.events, cursor: r.cursor}
4386+ yield {authorized: True {}, events: r.events, cursor: r.cursor, hasMore: r.hasMore}
4295− yield {authorized: result.authorized, events: result.events, cursor: result.cursor}
4387+ yield {authorized: result.authorized, events: result.events, cursor: result.cursor, hasMore: result.hasMore}
4296 -- Release poll slot. Safety: the code between acquire and release only contains
4388 -- Release poll slot. Safety: the code between acquire and release only contains
4297 -- atomic reads and STM retry (no IO that can throw), so this path is always reached.
4389 -- atomic reads and STM retry (no IO that can throw), so this path is always reached.
4298 -- If it ever isn't (e.g., runtime panic), pruneExpired cleans stale entries
4390 -- If it ever isn't (e.g., runtime panic), pruneExpired cleans stale entries
@@ -4300,7 +4392,7 @@ pollOnce = \token cursor clientIp -> do
4300 atomic do
4392 atomic do
4301 activePolls <- *activePolls
4393 activePolls <- *activePolls
4302 set *activePolls = removeOneForKey activePolls s.pubkey
4394 set *activePolls = removeOneForKey activePolls s.pubkey
4303− yield {authorized: pollResult.authorized, events: pollResult.events, cursor: pollResult.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries, pubkey: s.pubkey, pollSlotFull: False {}}
4395+ yield {authorized: pollResult.authorized, events: pollResult.events, cursor: pollResult.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries, pubkey: s.pubkey, pollSlotFull: False {}, hasMore: pollResult.hasMore}
4305-- ============================================
4397-- ============================================
4306-- Derived Relations
4398-- Derived Relations
@@ -4438,7 +4530,7 @@ newPresenceEntries = \entries presence ->
4438offlineGossipEvents = \expiredLocal -> do
4530offlineGossipEvents = \expiredLocal -> do
4439 p <- expiredLocal
4531 p <- expiredLocal
4440 let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
4532 let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
4441− yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.expiresAt, ttl: ttl}
4533+ yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.expiresAt, ttl: ttl, seenBy: [serverHostN]}
4443-- Filter targets to only those not locally online (avoid forwarding to keys that can poll directly)
4535-- Filter targets to only those not locally online (avoid forwarding to keys that can poll directly)
4444remoteOnlyTargets = \targets localOnline ->
4536remoteOnlyTargets = \targets localOnline ->
@@ -4616,7 +4708,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
4616 when (revokeOthers && result.revokedCount > 0)
4708 when (revokeOthers && result.revokedCount > 0)
4617 (logInfo ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
4709 (logInfo ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
4618 fork (when federationEnabled
4710 fork (when federationEnabled
4619− (sendBatchedGossip result.peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops))
4711+ (sendBatchedGossip result.peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl, seenBy: [serverHostN]}] gossipOriginHops))
4620 yield (respond {token: token, expiresAt: expiry, error: ""})
4712 yield (respond {token: token, expiresAt: expiry, error: ""})
4621 else yield (respond {token: "", expiresAt: 0, error: "invalid"})
4713 else yield (respond {token: "", expiresAt: 0, error: "invalid"})
@@ -4633,15 +4725,18 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
4633 let cursorWasReset = generationMismatch && cursor > 0
4725 let cursorWasReset = generationMismatch && cursor > 0
4634 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
4726 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
4635 dp <- *drainPhase
4727 dp <- *drainPhase
4728+ -- Cursor staleness: cursor is older than message retention, so messages were missed.
4729+ -- Different from cursorReset (server restart): staleness means client was offline too long.
4730+ let cursorStale = effectiveCursor > 0 && t - effectiveCursor > unackedMessageMaxAge
4636− if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, messageExpirations: [], messageDeliveries: []})
4731+ if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: []})
4638 globalRateLimited <- checkGlobalRate t
4733 globalRateLimited <- checkGlobalRate t
4639− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, messageExpirations: [], messageDeliveries: []})
4734+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: []})
4641 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
4736 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
4642 startPresenceVersion <- *presenceVersion
4737 startPresenceVersion <- *presenceVersion
4643 result <- case extractToken authorization of
4738 result <- case extractToken authorization of
4644− Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: [], pubkey: "", pollSlotFull: False {}}
4739+ Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: [], pubkey: "", pollSlotFull: False {}, hasMore: False {}}
4645 Just {value: token} -> pollOnce token effectiveCursor clientIp
4740 Just {value: token} -> pollOnce token effectiveCursor clientIp
4646 -- Check presence for watched keys (piggybacks on poll cycle; max latency = pollTimeoutMs).
4741 -- Check presence for watched keys (piggybacks on poll cycle; max latency = pollTimeoutMs).
4647 -- Skip the expensive bucket reads if presenceVersion hasn't changed since poll start.
4742 -- Skip the expensive bucket reads if presenceVersion hasn't changed since poll start.
@@ -4687,7 +4782,7 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
4687 yield capped
4782 yield capped
4688 -- Poll slot exhaustion: signal clients to back off for one poll cycle
4783 -- Poll slot exhaustion: signal clients to back off for one poll cycle
4689 let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
4784 let pollRetryMs = if result.pollSlotFull then pollTimeoutMs else 0
4690− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, messageExpirations: expirations, messageDeliveries: deliveries})
4785+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: result.pollSlotFull, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: pollRetryMs, cursorReset: cursorWasReset, hasMore: result.hasMore, cursorStale: cursorStale, messageExpirations: expirations, messageDeliveries: deliveries})
4692-- Core send logic: dedup, rate limiting, store message, create forwards.
4787-- Core send logic: dedup, rate limiting, store message, create forwards.
4693-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
4788-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
@@ -4729,7 +4824,10 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
4730 recipientCounts <- readRcBucket msgBucket
4825 recipientCounts <- readRcBucket msgBucket
4731 let recipientCount = getRecipientCount to recipientCounts
4826 let recipientCount = getRecipientCount to recipientCounts
4827+ -- Return MsQueued (not MsRejected) for per-recipient capacity: prevents
4828+ -- an attacker from probing mailbox fullness for arbitrary pubkeys. Global
4829+ -- capacity still returns MsRejected since it's not per-recipient.
4732− if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4830+ if recipientCount >= maxMessagesPerRecipient then yield {status: MsQueued {}, remoteServers: [], retryAfterMs: 0}
4734 recipientPresence <- readPresenceBucket (pubkeyBucket to)
4832 recipientPresence <- readPresenceBucket (pubkeyBucket to)
4735 seqCounter <- *seqCounter
4833 seqCounter <- *seqCounter
@@ -4865,7 +4963,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs authorizat
4865 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs t)
4963 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs t)
4866 case result.status of
4964 case result.status of
4867 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
4965 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
4868− MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))
4966+ MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (global capacity full)"))
4869 _ -> yield {}
4967 _ -> yield {}
4870 forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.remoteServers
4968 forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.remoteServers
4871 yield (case result.status of
4969 yield (case result.status of
@@ -5195,7 +5293,10 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
5195 yield {gm: gm, preFiltered: count safeDeduped - count capacityDeduped, localClashCount: count localClashes}
5293 yield {gm: gm, preFiltered: count safeDeduped - count capacityDeduped, localClashCount: count localClashes}
5196 when (gossipResult.localClashCount > 0)
5294 when (gossipResult.localClashCount > 0)
5197 (logWarn ("gossip: " ++ fromServer ++ " claims " ++ show gossipResult.localClashCount ++ " locally-authenticated key(s) — filtered from merge"))
5295 (logWarn ("gossip: " ++ fromServer ++ " claims " ++ show gossipResult.localClashCount ++ " locally-authenticated key(s) — filtered from merge"))
5296+ -- seenBy for re-broadcast: include fromServer (sender) and ourselves so downstream
5297+ -- peers that share connections with either skip this event.
5298+ let rebroadcastSeenBy = if any [fromServer] (\s -> s == serverHostN) then [serverHostN] else [fromServer, serverHostN]
5198− let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
5299+ let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {}) rebroadcastSeenBy) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}) rebroadcastSeenBy)
5199 when (not gossipResult.gm.withinCapacity || gossipResult.preFiltered > 0)
5300 when (not gossipResult.gm.withinCapacity || gossipResult.preFiltered > 0)
5200 (logWarn ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer ++ " (pre-filtered: " ++ show gossipResult.preFiltered ++ ")"))
5301 (logWarn ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer ++ " (pre-filtered: " ++ show gossipResult.preFiltered ++ ")"))
5201 -- Suppress re-broadcast of events we've already gossiped recently (prevents amplification)
5302 -- Suppress re-broadcast of events we've already gossiped recently (prevents amplification)
@@ -5376,7 +5477,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
5376 let corrId = sanitizeCorrelationId correlationId
5477 let corrId = sanitizeCorrelationId correlationId
5377 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " ack from " ++ fromServer))
5478 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " ack from " ++ fromServer))
5378 -- Wrap respond to add ackedCount default for withFederationAuth error paths
5479 -- Wrap respond to add ackedCount default for withFederationAuth error paths
5379− let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0}
5480+ let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0, results: []}
5380 -- clearOnAuth=False: defer clearFailure until we confirm the ack matched actual
5481 -- clearOnAuth=False: defer clearFailure until we confirm the ack matched actual
5381 -- forward entries. Prevents a peer from resetting its failure/backoff counters by
5482 -- forward entries. Prevents a peer from resetting its failure/backoff counters by
5382 -- spamming acks for messageIds we never forwarded to it.
5483 -- spamming acks for messageIds we never forwarded to it.
@@ -5423,7 +5524,12 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
5423 (logWarn ("federation/ack: dropped " ++ show capWarning.perKeyDropped ++ " deferred cleanups (per-key cap " ++ show maxPendingAckCleanupsPerKey ++ " reached) — origin messages may not be reclaimed"))
5524 (logWarn ("federation/ack: dropped " ++ show capWarning.perKeyDropped ++ " deferred cleanups (per-key cap " ++ show maxPendingAckCleanupsPerKey ++ " reached) — origin messages may not be reclaimed"))
5424 when capWarning.globalCapped
5525 when capWarning.globalCapped
5425 (logWarn ("federation/ack: pendingAckCleanups at global capacity (" ++ show maxPendingAckCleanups ++ ") — oldest entries truncated, origin messages may not be reclaimed")))
5526 (logWarn ("federation/ack: pendingAckCleanups at global capacity (" ++ show maxPendingAckCleanups ++ ") — oldest entries truncated, origin messages may not be reclaimed")))
5527+ -- Per-message ack results: lets the origin server know which messageIds matched
5528+ -- forward entries (found=True) vs which had no matching forward (found=False).
5529+ -- Origin servers can drop entries with found=False from their retry queues instead
5530+ -- of retrying indefinitely for messages that were already pruned or never forwarded.
5531+ let perMessageResults = do mid <- dedupedIds; yield {messageId: mid, found: any removedForwards (\f -> f.messageId == mid)}
5426− yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
5532+ yield (respond {ok: True {}, error: "", ackedCount: ackedCount, results: perMessageResults}))
5428-- POST /federation/cancel: Cancel a previously forwarded message.
5534-- POST /federation/cancel: Cancel a previously forwarded message.
5429-- Called by the origin server when the sender retracts a message via /messages/cancel.
5535-- Called by the origin server when the sender retracts a message via /messages/cancel.
@@ -5511,6 +5617,83 @@ cancelBatchInner = \entries fromServer t acc ->
5511 yield {cancelled: True {}}
5617 yield {cancelled: True {}}
5512 cancelBatchInner rest fromServer t (union acc [{messageId: entry.messageId, cancelled: result.cancelled}])
5618 cancelBatchInner rest fromServer t (union acc [{messageId: entry.messageId, cancelled: result.cancelled}])
5620+-- POST /federation/forward-batch: Receive multiple forwarded messages in one request.
5621+-- Each message is processed independently (same logic as handleForwardMessage).
5622+-- Partial success is possible; per-message results are returned.
5623+handleFederationBatchForward = \forwards fromServer requestTimestamp correlationId authorization respond -> do
5625+ let corrId = sanitizeCorrelationId correlationId
5626+ when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward-batch (" ++ show (count forwards) ++ " entries) from " ++ fromServer))
5627+ let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: []}
5628+ withFederationAuth authorization t batchRespond fromServer "forward" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (True {}) (do
5629+ let capped = takeRelation maxBatchSize forwards
5630+ results <- forwardBatchInner capped fromServer t []
5631+ let storedCount = count (do r <- results; where r.ok; yield r)
5632+ when (storedCount > 0)
5633+ (logInfo ("federation/forward-batch: stored " ++ show storedCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer))
5634+ yield (respond {ok: True {}, error: "", results: results}))
5636+-- Process each forward entry in a batch sequentially. Each entry runs the same
5637+-- validation and storage logic as handleForwardMessage (presence check, dedup,
5638+-- rate limiting, capacity check, nonce recording).
5639+forwardBatchInner = \entries fromServer t acc ->
5640+ case headOf entries of
5641+ Nothing {} -> yield acc
5642+ Just {value: entry} -> do
5643+ let rest = do e <- entries; where e.messageId != entry.messageId || e.toKey != entry.toKey; yield e
5644+ if isLocalServer entry.originServer || t - entry.timestamp > messageMaxAge
5645+ then forwardBatchInner rest fromServer t (union acc [{messageId: entry.messageId, ok: False {}}])
5647+ -- Relay origin verification (same as handleForwardMessage)
5648+ originOk <- if fromServer == entry.originServer then yield (True {})
5650+ knownPeers <- *knownPeerServers
5651+ blocked <- *blockedPeers
5652+ allowed <- *allowedPeers
5653+ yield (any knownPeers (\p -> p == entry.originServer)
5654+ && not (isBlockedPeer entry.originServer blocked)
5655+ && isAllowedByAllowlist entry.originServer allowed)
5657+ then forwardBatchInner rest fromServer t (union acc [{messageId: entry.messageId, ok: False {}}])
5659+ let clampedTs = clampTimestamp entry.timestamp t
5660+ result <- atomic do
5661+ let bkt = pubkeyBucket entry.toKey
5662+ presenceBkt <- readPresenceBucket bkt
5663+ let hasPresence = any presenceBkt (\p -> p.pubkey == entry.toKey && p.expiresAt > t)
5664+ if not hasPresence then yield (False {})
5666+ bucket <- readMessageBucket bkt
5667+ ackedBkt <- readAckedBucket bkt
5668+ delivBkt <- readDelivBucket bkt
5669+ nonceBkt <- readNonceBucket bkt
5670+ let isDup = checkMessageDup bucket entry.messageId entry.toKey || hasMessage entry.messageId entry.toKey ackedBkt || hasMessage entry.messageId entry.toKey delivBkt || any nonceBkt (\n -> n.messageId == entry.messageId && n.toKey == entry.toKey)
5671+ if isDup then yield (True {})
5673+ totalMsgCount <- *totalMessageCount
5674+ recipientCounts <- readRcBucket bkt
5675+ let recipientCount = getRecipientCount entry.toKey recipientCounts
5676+ if totalMsgCount >= maxMessages || recipientCount >= maxMessagesPerRecipient
5677+ then yield (False {})
5679+ seqCounter <- *seqCounter
5680+ let seqTs = nextSeqTs t seqCounter
5681+ set *seqCounter = seqTs
5682+ let fwdExpiresAt = if entry.expiresAt > 0 && entry.expiresAt > t then min entry.expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
5683+ writeMessageBucket bkt (union bucket
5684+ [{messageId: entry.messageId, toKey: entry.toKey,
5685+ encryptedBlob: entry.encryptedBlob, timestamp: clampedTs, originServer: entry.originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: ""}])
5686+ set *totalMessageCount = totalMsgCount + 1
5687+ writeRcBucket bkt (adjustRecipientCount recipientCounts entry.toKey 1)
5688+ signalMessageForKey entry.toKey seqTs
5689+ fwdNonces <- readNonceBucket bkt
5690+ let updatedNonces = union fwdNonces [{messageId: entry.messageId, toKey: entry.toKey, seenAt: t}]
5691+ writeNonceBucket bkt (if count updatedNonces > maxNoncesPerBucket
5692+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5693+ else updatedNonces)
5695+ forwardBatchInner rest fromServer t (union acc [{messageId: entry.messageId, ok: result}])
5514-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
5697-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
5515-- Auth checked once; each cancel processed independently. Partial success possible.
5698-- Auth checked once; each cancel processed independently. Partial success possible.
5516handleCancelBatch = \cancels authorization clientIp respond -> do
5699handleCancelBatch = \cancels authorization clientIp respond -> do
@@ -5673,7 +5856,7 @@ handleDrain = \phase authorization respond -> do
5673 logInfo ("drain: flushing " ++ show pendingAckCount ++ " pending acks")
5856 logInfo ("drain: flushing " ++ show pendingAckCount ++ " pending acks")
5674 sendAcksAndClean flushState.pendingAcks)
5857 sendAcksAndClean flushState.pendingAcks)
5675 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
5858 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
5676− let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
5859+ let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]}
5677 sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
5860 sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
5678 -- Wait for in-flight federation requests to complete (timeout: 2x fetch timeout)
5861 -- Wait for in-flight federation requests to complete (timeout: 2x fetch timeout)
5679 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))
5862 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))
@@ -5789,6 +5972,11 @@ computeAndCacheHealth = \t -> do
5789 && forwardCount * 10 <= maxForwardsEntries * 9
5972 && forwardCount * 10 <= maxForwardsEntries * 9
5790 && sessionCount * 10 <= maxSessions * 9
5973 && sessionCount * 10 <= maxSessions * 9
5791 && count activePolls * 10 <= maxActivePolls * 9
5974 && count activePolls * 10 <= maxActivePolls * 9
5975+ -- 75% capacity early warning (fires before the 90% health check failure)
5976+ let capacityWarn = tmc * 4 >= maxMessages * 3
5977+ || presenceCount * 4 >= maxPresenceEntries * 3
5978+ || forwardCount * 4 >= maxForwardsEntries * 3
5979+ || sessionCount * 4 >= maxSessions * 3
5792 let fedOk = not federationEnabled || warmup
5980 let fedOk = not federationEnabled || warmup
5793 || count forwardFailures == 0
5981 || count forwardFailures == 0
5794 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
5982 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
@@ -5803,6 +5991,7 @@ computeAndCacheHealth = \t -> do
5803 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
5991 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
5804 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
5992 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
5805 capacityOk: capacityOk,
5993 capacityOk: capacityOk,
5994+ capacityWarning: capacityWarn,
5806 federationOk: fedOk,
5995 federationOk: fedOk,
5807 snapshotOk: snapOk,
5996 snapshotOk: snapOk,
5808 watchdogOk: wdOk,
5997 watchdogOk: wdOk,
@@ -5814,6 +6003,7 @@ computeAndCacheHealth = \t -> do
5814 set *cachedHealthPrune = result.pruneOk
6003 set *cachedHealthPrune = result.pruneOk
5815 set *cachedHealthHeartbeat = result.heartbeatOk
6004 set *cachedHealthHeartbeat = result.heartbeatOk
5816 set *cachedHealthCapacity = result.capacityOk
6005 set *cachedHealthCapacity = result.capacityOk
6006+ set *cachedHealthCapacityWarning = result.capacityWarning
5817 set *cachedHealthFederation = result.federationOk
6007 set *cachedHealthFederation = result.federationOk
5818 set *cachedHealthSnapshot = result.snapshotOk
6008 set *cachedHealthSnapshot = result.snapshotOk
5819 set *cachedHealthWatchdog = result.watchdogOk
6009 set *cachedHealthWatchdog = result.watchdogOk
@@ -5837,8 +6027,9 @@ handleHealth = \authorization respond -> do
5837 cS <- *cachedHealthSnapshot
6027 cS <- *cachedHealthSnapshot
5838 cW <- *cachedHealthWatchdog
6028 cW <- *cachedHealthWatchdog
5839 cN <- *cachedHealthNonce
6029 cN <- *cachedHealthNonce
6030+ cCW <- *cachedHealthCapacityWarning
5840 dp <- *drainPhase
6031 dp <- *drainPhase
5841− yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, nonceHealthy: cN, draining: dp > 0}
6032+ yield {ok: cOk && dp == 0, version: serverBuildVersion, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, capacityWarning: cCW, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, nonceHealthy: cN, draining: dp > 0}
5842 yield (respond cached)
6033 yield (respond cached)
5844 isAuthed <- verifyMaintenanceAuth authorization
6035 isAuthed <- verifyMaintenanceAuth authorization
@@ -5851,6 +6042,7 @@ handleHealth = \authorization respond -> do
5851 then yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: result.pruneOk,
6042 then yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: result.pruneOk,
5852 heartbeatHealthy: result.heartbeatOk,
6043 heartbeatHealthy: result.heartbeatOk,
5853 capacityHealthy: result.capacityOk,
6044 capacityHealthy: result.capacityOk,
6045+ capacityWarning: result.capacityWarning,
5854 federationHealthy: result.federationOk,
6046 federationHealthy: result.federationOk,
5855 snapshotHealthy: result.snapshotOk,
6047 snapshotHealthy: result.snapshotOk,
5856 watchdogHealthy: result.watchdogOk,
6048 watchdogHealthy: result.watchdogOk,
@@ -5859,13 +6051,14 @@ handleHealth = \authorization respond -> do
5859 else yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: True {},
6051 else yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: True {},
5860 heartbeatHealthy: True {},
6052 heartbeatHealthy: True {},
5861 capacityHealthy: True {},
6053 capacityHealthy: True {},
6054+ capacityWarning: False {},
5862 federationHealthy: True {},
6055 federationHealthy: True {},
5863 snapshotHealthy: True {},
6056 snapshotHealthy: True {},
5864 watchdogHealthy: True {},
6057 watchdogHealthy: True {},
5865 nonceHealthy: True {},
6058 nonceHealthy: True {},
5866 draining: result.isDraining})
6059 draining: result.isDraining})
5868−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}, 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}}
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}}
5870handleStats = \authorization respond -> do
6063handleStats = \authorization respond -> do
5871 t <- now
6064 t <- now
@@ -5952,9 +6145,12 @@ handleStats = \authorization respond -> do
5952 let oldestAckAge = if oldestAckTs == 0 then 0 else t - oldestAckTs
6145 let oldestAckAge = if oldestAckTs == 0 then 0 else t - oldestAckTs
5953 let fwdServerCount = fwdStats.serverCount
6146 let fwdServerCount = fwdStats.serverCount
5954 let maxFwdDepth = fwdStats.maxDepth
6147 let maxFwdDepth = fwdStats.maxDepth
6148+ -- Top-3 failing peers by failure count (inline in /stats to reduce monitoring queries)
6149+ let topFailing = takeRelation 3 (sortBy (\f -> 0 - f.failures) core.ff)
6150+ let topFailingPeers = do f <- topFailing; yield {server: f.server, failures: f.failures}
5955 yield (respond {ok: True {}, version: serverBuildVersion,
6151 yield (respond {ok: True {}, version: serverBuildVersion,
5956 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},
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},
5957− 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},
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},
5958 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},
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},
5959 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}})
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}})
5960 else yield (respond zeroStats)
6156 else yield (respond zeroStats)
@@ -6678,7 +6874,7 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do
6678 when (result.revokedCount > 0)
6874 when (result.revokedCount > 0)
6679 (logInfo ("admin: revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
6875 (logInfo ("admin: revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
6680 when (result.hadPresence && federationEnabled && count result.peerServers > 0)
6876 when (result.hadPresence && federationEnabled && count result.peerServers > 0)
6681− (fork (sendBatchedGossip result.peerServers [{eventType: GossipOffline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))
6877+ (fork (sendBatchedGossip result.peerServers [{eventType: GossipOffline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]}] gossipOriginHops))
6682 yield (respond {ok: True {}, revokedCount: result.revokedCount})
6878 yield (respond {ok: True {}, revokedCount: result.revokedCount})
6684-- POST /maintenance/messages/purge: Purge all pending messages for a specific pubkey.
6880-- POST /maintenance/messages/purge: Purge all pending messages for a specific pubkey.
@@ -6747,7 +6943,7 @@ handleAdminRevokeIp = \ip authorization respond -> do
6747 when (count offlines > 0 && federationEnabled) (do
6943 when (count offlines > 0 && federationEnabled) (do
6748 peerServers <- *knownPeerServers
6944 peerServers <- *knownPeerServers
6749 when (count peerServers > 0) (do
6945 when (count peerServers > 0) (do
6750− let gossipEvents = do c <- offlines; yield {eventType: GossipOffline {}, pubkey: c.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
6946+ let gossipEvents = do c <- offlines; yield {eventType: GossipOffline {}, pubkey: c.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]}
6751 sendBatchedGossip peerServers gossipEvents gossipOriginHops)))
6947 sendBatchedGossip peerServers gossipEvents gossipOriginHops)))
6752 when (result.revokedCount > 0)
6948 when (result.revokedCount > 0)
6753 (logInfo ("admin: revoked " ++ show result.revokedCount ++ " sessions from IP " ++ ip ++ " (" ++ show (count result.affectedKeys) ++ " keys affected)"))
6949 (logInfo ("admin: revoked " ++ show result.revokedCount ++ " sessions from IP " ++ ip ++ " (" ++ show (count result.affectedKeys) ++ " keys affected)"))
@@ -6915,6 +7111,7 @@ requestPath = \req -> case req of
6915 FederationAck {} -> "POST /federation/ack"
7111 FederationAck {} -> "POST /federation/ack"
6916 FederationCancel {} -> "POST /federation/cancel"
7112 FederationCancel {} -> "POST /federation/cancel"
6917 FederationBatchCancel {} -> "POST /federation/cancel-batch"
7113 FederationBatchCancel {} -> "POST /federation/cancel-batch"
7114+ FederationBatchForward {} -> "POST /federation/forward-batch"
6918 VerifyToken {} -> "POST /federation/verify"
7115 VerifyToken {} -> "POST /federation/verify"
6919 Prune {} -> "POST /maintenance/prune"
7116 Prune {} -> "POST /maintenance/prune"
6920 Drain {} -> "POST /maintenance/drain"
7117 Drain {} -> "POST /maintenance/drain"
@@ -6969,6 +7166,7 @@ requestClientIp = \req -> case req of
6969 FederationAck {} -> ""
7166 FederationAck {} -> ""
6970 FederationCancel {} -> ""
7167 FederationCancel {} -> ""
6971 FederationBatchCancel {} -> ""
7168 FederationBatchCancel {} -> ""
7169+ FederationBatchForward {} -> ""
6972 Prune {} -> ""
7170 Prune {} -> ""
6973 Drain {} -> ""
7171 Drain {} -> ""
6974 Shutdown {} -> ""
7172 Shutdown {} -> ""
@@ -7024,6 +7222,7 @@ serveInner = \req -> case req of
7024 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
7222 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
7025 FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationCancel messageId toKey fromServer requestTimestamp correlationId authorization respond
7223 FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationCancel messageId toKey fromServer requestTimestamp correlationId authorization respond
7026 FederationBatchCancel {cancels, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationBatchCancel cancels fromServer requestTimestamp correlationId authorization respond
7224 FederationBatchCancel {cancels, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationBatchCancel cancels fromServer requestTimestamp correlationId authorization respond
7225+ FederationBatchForward {forwards, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp correlationId authorization respond
7027 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
7226 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
7028 Prune {authorization, respond} -> handlePrune authorization respond
7227 Prune {authorization, respond} -> handlePrune authorization respond
7029 Drain {phase, authorization, respond} -> handleDrain phase authorization respond
7228 Drain {phase, authorization, respond} -> handleDrain phase authorization respond