@@ -169,7 +169,7 @@ snapshotVersion = 10 -- bumped for ForwardEntry.expiresAt field
169169
170170-- Server build version: included in /health and /stats responses for deployment verification.
171171-- 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"
173173
174174-- Snapshot checksum: deterministic hash of entry counts for integrity verification.
175175-- Uses multiplicative hash to distinguish permutations of the same counts
@@ -324,7 +324,10 @@ impl FromJSON GossipEventType where
324324 else GossipUnknown {}
325325
326326-- Wire format for gossip events sent between servers
327−type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex, server: ServerName, timestamp: Timestamp, ttl: Int<S>}
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).
330+type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex, server: ServerName, timestamp: Timestamp, ttl: Int<S>, seenBy: [Text]}
328331
329332-- A gossip event that passed validation (timestamp not in future, TTL > 0, not expired)
330333type ValidGossipEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Timestamp, timestamp: Timestamp, ttl: Int<S>}
@@ -845,6 +848,7 @@ type SenderDelivery = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHe
845848*cachedHealthSnapshot : Bool
846849*cachedHealthWatchdog : Bool
847850*cachedHealthNonce : Bool
851+*cachedHealthCapacityWarning : Bool
848852
849853-- ============================================
850854-- Routes
@@ -874,7 +878,7 @@ route AuthApi where
874878route ClientApi where
875879 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
876880 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}]}
878882 = Poll
879883 /messages
880884 -- Send an E2E encrypted message to a recipient pubkey.
@@ -945,7 +949,7 @@ route FederationApi where
945949 = ForwardMessage
946950 -- Acknowledge delivery of messages back to the origin server
947951 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}]}
949953 = FederationAck
950954 -- Cancel (retract) a previously forwarded message. The origin server sends this
951955 -- when the sender cancels a message via /messages/cancel. The receiving server
@@ -957,6 +961,12 @@ route FederationApi where
957961 POST {cancels: [{messageId: MessageIdHex, toKey: PubkeyHex}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /cancel-batch headers {authorization: Text}
958962 -> {ok: Bool, error: Text, results: [BatchCancelResult]}
959963 = 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
960970
961971-- Federation callback verification: the receiving server calls back to the sender
962972-- to confirm a request token is genuine. No shared secret needed.
@@ -1067,7 +1077,7 @@ route MaintenanceApi where
10671077 = RotateSecret
10681078
10691079route 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
10711081
10721082-- Returns counts of in-memory collections (requires maintenance secret)
10731083-- TODO: If Knot adds raw text response support, add a /metrics endpoint
@@ -1077,7 +1087,7 @@ route HealthApi where
10771087-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
10781088route StatsApi where
10791089 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}}
10811091 = Stats
10821092 GET /stats/peers headers {authorization: Text}
10831093 -> {peers: [PeerStatus], ok: Bool}
@@ -1118,7 +1128,10 @@ logError = \msg -> println ("[ERROR] " ++ msg)
11181128-- when msg contains user-controlled data (e.g., error strings, server names).
11191129logEvent = \level event msg -> println ("{\"level\":\"" ++ escapeJsonString level ++ "\",\"event\":\"" ++ escapeJsonString event ++ "\",\"msg\":\"" ++ escapeJsonString msg ++ "\"}")
11201130
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.
11221135-- Processes one character at a time via tail recursion (TCO-safe in Knot).
11231136escapeJsonString : Text -> Text
11241137escapeJsonString = \s ->
@@ -1126,7 +1139,11 @@ escapeJsonString = \s ->
11261139 else let c = take 1 s in
11271140 if c == "\\" then "\\\\" ++ escapeJsonString (drop 1 s)
11281141 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)
11301147
11311148any : [a] -> (a -> Bool) -> Bool
11321149any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
@@ -1137,6 +1154,9 @@ whenOk = \result action -> case result of
11371154 Ok _ -> action
11381155 Err _ -> yield {}
11391156clamp = \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
11401160min = \a b -> if a < b then a else b
11411161max = \a b -> if a > b then a else b
11421162headOf = \items -> fold (\acc x -> case acc of
@@ -1606,7 +1626,7 @@ readAllSessions = do
16061626 s13 <- *sessionBucket13
16071627 s14 <- *sessionBucket14
16081628 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)))))))))))))))
16101630
16111631-- Count sessions across all 16 buckets without materializing the full list
16121632countAllSessions : IO Int
@@ -1837,7 +1857,7 @@ readAllPresence = do
18371857 p13 <- *presenceBucket13
18381858 p14 <- *presenceBucket14
18391859 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)))))))))))))))
18411861
18421862countAllPresence : IO Int
18431863countAllPresence = do
@@ -1997,7 +2017,8 @@ writeMessageBucket = \b msgs ->
19972017 else if b == 14 then set *messageBucket14 = msgs
19982018 else set *messageBucket15 = msgs
19992019
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).
20012022readAllMessages : IO [Message]
20022023readAllMessages = do
20032024 m0 <- *messageBucket0
@@ -2016,7 +2037,7 @@ readAllMessages = do
20162037 m13 <- *messageBucket13
20172038 m14 <- *messageBucket14
20182039 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)))))))))))))))
20202041
20212042-- Apply a filter/transform to every message bucket
20222043transformAllMessageBuckets : ([Message] -> [Message]) -> IO ()
@@ -2163,17 +2184,32 @@ adjustRecipientCount = \counts pubkey delta ->
21632184 else if delta > 0 then union counts [{pubkey: pubkey, count: delta}]
21642185 else counts
21652186
2166−-- Count items per toKey in a single pass: fold accumulates {pubkey, count} pairs,
2167−-- incrementing existing entries or inserting new ones.
2168−-- NOTE: O(n*k) where k = unique recipients due to `any acc` check inside fold.
2169−-- Acceptable for per-bucket data (n ≤ maxMessages/16). A map type would make this O(n).
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).
2187+-- Count items per toKey using sort-then-scan: O(n log n) sort + O(n) linear scan.
2188+-- Replaces the previous O(n*k) fold-with-any approach. For a bucket with 6000 messages
2189+-- and 500 unique recipients, this reduces comparisons from ~3M to ~80k.
2190+-- sortBy groups equal toKeys together; countSortedGroups scans consecutive equal keys.
21722191countByRecipient = \items ->
2173− fold (\acc x ->
2174− if any acc (\c -> c.pubkey == x.toKey)
2175− then do c <- acc; yield (if c.pubkey == x.toKey then {pubkey: c.pubkey, count: c.count + 1} else c)
2176− else union [{pubkey: x.toKey, count: 1}] acc) [] items
2192+ let sorted = sortBy (\m -> m.toKey) items in
2193+ countSortedGroups sorted
2194+
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
2200+ Nothing {} -> []
2201+ Just {value: first} ->
2202+ let rest = do x <- items; where x != first; yield x in
2203+ countSortedGroupsInner rest first.toKey 1 []
2204+
2205+countSortedGroupsInner = \items currentKey currentCount acc ->
2206+ case headOf items of
2207+ Nothing {} -> union acc [{pubkey: currentKey, count: currentCount}]
2208+ Just {value: x} ->
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}])
21772213
21782214-- Batch-decrement recipient counts: subtract per-key removal counts in a single pass
21792215-- instead of calling adjustRecipientCount in a loop.
@@ -2327,7 +2363,7 @@ readAllNonces = do
23272363 n13 <- *nonceBucket13
23282364 n14 <- *nonceBucket14
23292365 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)))))))))))))))
23312367
23322368countAllNonces : Int -> IO Int
23332369countAllNonces = \b ->
@@ -2446,7 +2482,7 @@ readAllForwards = do
24462482 f13 <- *fwdBucket13
24472483 f14 <- *fwdBucket14
24482484 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)))))))))))))))
24502486
24512487countAllForwards : Int -> IO Int
24522488countAllForwards = \b ->
@@ -2588,7 +2624,7 @@ readAllAcked = do
25882624 a4 <- *ackedBucket4; a5 <- *ackedBucket5; a6 <- *ackedBucket6; a7 <- *ackedBucket7
25892625 a8 <- *ackedBucket8; a9 <- *ackedBucket9; a10 <- *ackedBucket10; a11 <- *ackedBucket11
25902626 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)))))))))))))))
25922628
25932629writeAllAcked : [AckedEntry] -> IO ()
25942630writeAllAcked = \entries -> do
@@ -2662,7 +2698,7 @@ readAllDelivered = do
26622698 d4 <- *delivBucket4; d5 <- *delivBucket5; d6 <- *delivBucket6; d7 <- *delivBucket7
26632699 d8 <- *delivBucket8; d9 <- *delivBucket9; d10 <- *delivBucket10; d11 <- *delivBucket11
26642700 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)))))))))))))))
26662702
26672703writeAllDelivered : [DeliveredEntry] -> IO ()
26682704writeAllDelivered = \entries -> do
@@ -3266,7 +3302,11 @@ forEachBatch = \items batchSize process ->
32663302
32673303sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {}
32683304sendBatchedGossipToServer = \server allEvents hops ->
3269− forEachBatch allEvents maxGossipEvents (\batch -> do
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 {}
3309+ else forEachBatch unseen maxGossipEvents (\batch -> do
32703310 t <- now
32713311 auth <- generateFederationAuth "gossip" server
32723312 corrId <- genTokenN 1
@@ -3295,10 +3335,13 @@ broadcastOfflineGossip = \t peerServers expiredLocal ->
32953335 sendBatchedGossip peerServers gossipEvents gossipOriginHops)
32963336
32973337
3298−effectiveGossipEvents = \entries appliedKeys eventType -> do
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.
3341+effectiveGossipEvents = \entries appliedKeys eventType seenBy -> do
32993342 e <- entries
33003343 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}
33023345
33033346mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
33043347 let trustedDeduped = trustedGossipEntries deduped presence fromServer t in
@@ -3561,7 +3604,7 @@ gossipOfflineIfLastSession = \r label -> do
35613604 else do
35623605 logInfo (label ++ ": last session for " ++ r.pubkey ++ ", gossiping offline")
35633606 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)))
35653608
35663609removeForward = \mid srv toKey -> atomic do
35673610 let b = pubkeyBucket toKey
@@ -3671,36 +3714,65 @@ incrementForwardRetries = \messageId toServer toKey -> atomic do
36713714 logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
36723715 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))
36733716
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).
36743719retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
36753720retryForwardsForServer = \entries remaining ->
3676− if remaining <= 0
3677− then yield {}
3678− else case headOf entries of
3679− Nothing {} -> yield {}
3680− Just {value: f} -> do
3681− let rest = do x <- entries; where x.messageId != f.messageId; yield x
3682− t <- now
3683− auth <- generateFederationAuth "forward" f.toServer
3684− corrId <- genTokenN 1
3685− result <- fetchFederation f.toServer (ForwardMessage {
3686− toKey: f.toKey, encryptedBlob: f.encryptedBlob,
3687− timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
3688− fromServer: serverHostN, requestTimestamp: t,
3689− authorization: auth, correlationId: corrId, expiresAt: f.expiresAt})
3690− case result of
3691− Err e -> do
3692− logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
3693− recordFailure f.toServer
3694− incrementForwardRetries f.messageId f.toServer f.toKey
3695− -- Network error: stop retrying this server
3696− Ok response -> do
3697− if response.ok then do
3698− removeForward f.messageId f.toServer f.toKey
3699− clearFailure f.toServer
3700− else do
3701− logWarn ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)
3702− incrementForwardRetries f.messageId f.toServer f.toKey
3703− retryForwardsForServer rest (remaining - 1)
3721+ if remaining <= 0 || count entries == 0 then yield {}
3722+ else do
3723+ let server = case headOf entries of Just {value: f} -> f.toServer; Nothing {} -> ""
3724+ forEachBatch entries maxBatchSize (\batch -> do
3725+ t <- now
3726+ auth <- generateFederationAuth "forward" server
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}
3729+ result <- fetchFederation server (FederationBatchForward {
3730+ forwards: batchPayload, fromServer: serverHostN,
3731+ requestTimestamp: t, authorization: auth, correlationId: corrId})
3732+ case result of
3733+ Err e -> do
3734+ logWarn ("federation: batch forward to " ++ server ++ " failed: " ++ show e)
3735+ recordFailure server
3736+ -- Increment retries for all entries in this batch
3737+ retryForwardsBatchIncrement batch server
3738+ Ok response -> do
3739+ if response.ok then do
3740+ clearFailure server
3741+ -- Process per-message results: remove succeeded, increment retries for failed
3742+ retryForwardsBatchResults batch response.results server
3743+ else do
3744+ logWarn ("federation: batch forward to " ++ server ++ " rejected: " ++ response.error)
3745+ -- Peer may not support batch forward; increment retries
3746+ retryForwardsBatchIncrement batch server)
3747+
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
3756+
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)
3764+ case msgResult of
3765+ Just {value: r} ->
3766+ if r.ok then do
3767+ removeForward f.messageId server f.toKey
3768+ retryForwardsBatchResults rest results server
3769+ else do
3770+ incrementForwardRetries f.messageId server f.toKey
3771+ retryForwardsBatchResults rest results server
3772+ Nothing {} -> do
3773+ -- No result for this message (shouldn't happen) — increment retry
3774+ incrementForwardRetries f.messageId server f.toKey
3775+ retryForwardsBatchResults rest results server
37043776
37053777-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
37063778-- Circuit breaker: skips entirely when too many peers are failing (likely local network issue).
@@ -3977,6 +4049,22 @@ pruneAndRetry = \t -> do
39774049 nonceCount <- countAllNonces 0
39784050 when (nonceCount * 10 >= maxForwardNonces * 9)
39794051 (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 ++ ")"))
39804068 -- Counter reconciliation: reconcile all counters every prune cycle.
39814069 -- Presence and session counters are cheap (just counting list lengths across buckets).
39824070 -- Message reconciliation is heavier (reads all messages, rebuilds per-recipient counts)
@@ -4093,7 +4181,7 @@ pruneAndRetry = \t -> do
40934181 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
40944182 -- trigger offline gossip when they expire (avoids wasted federation traffic)
40954183 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]}
40974185 when (count result.refreshPeers > 0 && count refreshEvents > 0)
40984186 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
40994187 -- Update cached health state so /health returns fresh data even when rate-limited.
@@ -4202,7 +4290,8 @@ pollNewEvents = \pubkey clampedCursor t -> do
42024290 -- (via nextSeqTs) and union appends new messages at the end. Kept as a defensive
42034291 -- measure in case bucket ordering assumptions are ever violated (e.g., by a future
42044292 -- 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
42064295 let evts = do
42074296 m <- limitedMsgs
42084297 yield {id: m.messageId, eventType: "message",
@@ -4210,7 +4299,10 @@ pollNewEvents = \pubkey clampedCursor t -> do
42104299 encryptedBlob: m.encryptedBlob,
42114300 timestamp: m.timestamp}}
42124301 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
4213− yield {events: evts, cursor: lastReceivedAt}
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
4305+ yield {events: evts, cursor: lastReceivedAt, hasMore: hasMore}
42144306
42154307-- Execute a single long-poll cycle:
42164308-- 1. Authenticate and ack messages up to cursor (atomic)
@@ -4240,7 +4332,7 @@ pollOnce = \token cursor clientIp -> do
42404332 startSignal <- readMessageSignal pubkey
42414333 yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, startSignal: startSignal, pendingAcks: ackResult.pendingAcks, evictedEntries: ackResult.evictedEntries}})
42424334 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 {}}
42444336 Just {value: s} -> do
42454337 -- Acquire poll slot (limit concurrent polls per key)
42464338 acquired <- atomic do
@@ -4255,7 +4347,7 @@ pollOnce = \token cursor clientIp -> do
42554347 yield (True {})
42564348 if not acquired
42574349 -- 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 {}}
42594351 else do
42604352 pollResult <- do
42614353 -- Phase 1: check for existing messages (atomic read)
@@ -4264,11 +4356,11 @@ pollOnce = \token cursor clientIp -> do
42644356 if count (pollableMessages bucket s.pubkey s.clampedCursor t) > 0
42654357 then do
42664358 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}})
42684360 else yield (Nothing {})
42694361 case initialResult of
42704362 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}
42724364 Nothing {} -> do
42734365 -- Phase 2: wait for signal or heartbeat
42744366 atomic do
@@ -4284,15 +4376,15 @@ pollOnce = \token cursor clientIp -> do
42844376 t2 <- now
42854377 result <- atomic do
42864378 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 {}}
42884380 else do
42894381 sessions2 <- readSessionBucket bucket
42904382 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 {}}
42924384 else do
42934385 r <- pollNewEvents s.pubkey s.clampedCursor t2
4294− yield {authorized: True {}, events: r.events, cursor: r.cursor}
4295− yield {authorized: result.authorized, events: result.events, cursor: result.cursor}
4386+ yield {authorized: True {}, events: r.events, cursor: r.cursor, hasMore: r.hasMore}
4387+ yield {authorized: result.authorized, events: result.events, cursor: result.cursor, hasMore: result.hasMore}
42964388 -- Release poll slot. Safety: the code between acquire and release only contains
42974389 -- atomic reads and STM retry (no IO that can throw), so this path is always reached.
42984390 -- If it ever isn't (e.g., runtime panic), pruneExpired cleans stale entries
@@ -4300,7 +4392,7 @@ pollOnce = \token cursor clientIp -> do
43004392 atomic do
43014393 activePolls <- *activePolls
43024394 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}
43044396
43054397-- ============================================
43064398-- Derived Relations
@@ -4438,7 +4530,7 @@ newPresenceEntries = \entries presence ->
44384530offlineGossipEvents = \expiredLocal -> do
44394531 p <- expiredLocal
44404532 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]}
44424534
44434535-- Filter targets to only those not locally online (avoid forwarding to keys that can poll directly)
44444536remoteOnlyTargets = \targets localOnline ->
@@ -4616,7 +4708,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
46164708 when (revokeOthers && result.revokedCount > 0)
46174709 (logInfo ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
46184710 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))
46204712 yield (respond {token: token, expiresAt: expiry, error: ""})
46214713 else yield (respond {token: "", expiresAt: 0, error: "invalid"})
46224714
@@ -4633,15 +4725,18 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
46334725 let cursorWasReset = generationMismatch && cursor > 0
46344726 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
46354727 dp <- *drainPhase
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: []})
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
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: []})
46374732 else do
46384733 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: []})
46404735 else do
46414736 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
46424737 startPresenceVersion <- *presenceVersion
46434738 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 {}}
46454740 Just {value: token} -> pollOnce token effectiveCursor clientIp
46464741 -- Check presence for watched keys (piggybacks on poll cycle; max latency = pollTimeoutMs).
46474742 -- 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
46874782 yield capped
46884783 -- Poll slot exhaustion: signal clients to back off for one poll cycle
46894784 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})
46914786
46924787-- Core send logic: dedup, rate limiting, store message, create forwards.
46934788-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
@@ -4729,7 +4824,10 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
47294824 else do
47304825 recipientCounts <- readRcBucket msgBucket
47314826 let recipientCount = getRecipientCount to recipientCounts
4732− if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
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.
4830+ if recipientCount >= maxMessagesPerRecipient then yield {status: MsQueued {}, remoteServers: [], retryAfterMs: 0}
47334831 else do
47344832 recipientPresence <- readPresenceBucket (pubkeyBucket to)
47354833 seqCounter <- *seqCounter
@@ -4865,7 +4963,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs authorizat
48654963 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId expiresInMs t)
48664964 case result.status of
48674965 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)"))
48694967 _ -> yield {}
48704968 forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.remoteServers
48714969 yield (case result.status of
@@ -5195,7 +5293,10 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
51955293 yield {gm: gm, preFiltered: count safeDeduped - count capacityDeduped, localClashCount: count localClashes}
51965294 when (gossipResult.localClashCount > 0)
51975295 (logWarn ("gossip: " ++ fromServer ++ " claims " ++ show gossipResult.localClashCount ++ " locally-authenticated key(s) — filtered from merge"))
5198− let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
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]
5299+ let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {}) rebroadcastSeenBy) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}) rebroadcastSeenBy)
51995300 when (not gossipResult.gm.withinCapacity || gossipResult.preFiltered > 0)
52005301 (logWarn ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer ++ " (pre-filtered: " ++ show gossipResult.preFiltered ++ ")"))
52015302 -- Suppress re-broadcast of events we've already gossiped recently (prevents amplification)
@@ -5376,7 +5477,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
53765477 let corrId = sanitizeCorrelationId correlationId
53775478 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " ack from " ++ fromServer))
53785479 -- 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: []}
53805481 -- clearOnAuth=False: defer clearFailure until we confirm the ack matched actual
53815482 -- forward entries. Prevents a peer from resetting its failure/backoff counters by
53825483 -- spamming acks for messageIds we never forwarded to it.
@@ -5423,7 +5524,12 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
54235524 (logWarn ("federation/ack: dropped " ++ show capWarning.perKeyDropped ++ " deferred cleanups (per-key cap " ++ show maxPendingAckCleanupsPerKey ++ " reached) — origin messages may not be reclaimed"))
54245525 when capWarning.globalCapped
54255526 (logWarn ("federation/ack: pendingAckCleanups at global capacity (" ++ show maxPendingAckCleanups ++ ") — oldest entries truncated, origin messages may not be reclaimed")))
5426− yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
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)}
5532+ yield (respond {ok: True {}, error: "", ackedCount: ackedCount, results: perMessageResults}))
54275533
54285534-- POST /federation/cancel: Cancel a previously forwarded message.
54295535-- Called by the origin server when the sender retracts a message via /messages/cancel.
@@ -5511,6 +5617,83 @@ cancelBatchInner = \entries fromServer t acc ->
55115617 yield {cancelled: True {}}
55125618 cancelBatchInner rest fromServer t (union acc [{messageId: entry.messageId, cancelled: result.cancelled}])
55135619
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
5624+ t <- now
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}))
5635+
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 {}}])
5646+ else do
5647+ -- Relay origin verification (same as handleForwardMessage)
5648+ originOk <- if fromServer == entry.originServer then yield (True {})
5649+ else do
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)
5656+ if not originOk
5657+ then forwardBatchInner rest fromServer t (union acc [{messageId: entry.messageId, ok: False {}}])
5658+ else do
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 {})
5665+ else do
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 {})
5672+ else do
5673+ totalMsgCount <- *totalMessageCount
5674+ recipientCounts <- readRcBucket bkt
5675+ let recipientCount = getRecipientCount entry.toKey recipientCounts
5676+ if totalMsgCount >= maxMessages || recipientCount >= maxMessagesPerRecipient
5677+ then yield (False {})
5678+ else do
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)
5694+ yield (True {})
5695+ forwardBatchInner rest fromServer t (union acc [{messageId: entry.messageId, ok: result}])
5696+
55145697-- POST /messages/cancel-batch: Cancel multiple sent messages in one request.
55155698-- Auth checked once; each cancel processed independently. Partial success possible.
55165699handleCancelBatch = \cancels authorization clientIp respond -> do
@@ -5673,7 +5856,7 @@ handleDrain = \phase authorization respond -> do
56735856 logInfo ("drain: flushing " ++ show pendingAckCount ++ " pending acks")
56745857 sendAcksAndClean flushState.pendingAcks)
56755858 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]}
56775860 sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
56785861 -- Wait for in-flight federation requests to complete (timeout: 2x fetch timeout)
56795862 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))
@@ -5789,6 +5972,11 @@ computeAndCacheHealth = \t -> do
57895972 && forwardCount * 10 <= maxForwardsEntries * 9
57905973 && sessionCount * 10 <= maxSessions * 9
57915974 && 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
57925980 let fedOk = not federationEnabled || warmup
57935981 || count forwardFailures == 0
57945982 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
@@ -5803,6 +5991,7 @@ computeAndCacheHealth = \t -> do
58035991 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
58045992 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
58055993 capacityOk: capacityOk,
5994+ capacityWarning: capacityWarn,
58065995 federationOk: fedOk,
58075996 snapshotOk: snapOk,
58085997 watchdogOk: wdOk,
@@ -5814,6 +6003,7 @@ computeAndCacheHealth = \t -> do
58146003 set *cachedHealthPrune = result.pruneOk
58156004 set *cachedHealthHeartbeat = result.heartbeatOk
58166005 set *cachedHealthCapacity = result.capacityOk
6006+ set *cachedHealthCapacityWarning = result.capacityWarning
58176007 set *cachedHealthFederation = result.federationOk
58186008 set *cachedHealthSnapshot = result.snapshotOk
58196009 set *cachedHealthWatchdog = result.watchdogOk
@@ -5837,8 +6027,9 @@ handleHealth = \authorization respond -> do
58376027 cS <- *cachedHealthSnapshot
58386028 cW <- *cachedHealthWatchdog
58396029 cN <- *cachedHealthNonce
6030+ cCW <- *cachedHealthCapacityWarning
58406031 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}
58426033 yield (respond cached)
58436034 else do
58446035 isAuthed <- verifyMaintenanceAuth authorization
@@ -5851,6 +6042,7 @@ handleHealth = \authorization respond -> do
58516042 then yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: result.pruneOk,
58526043 heartbeatHealthy: result.heartbeatOk,
58536044 capacityHealthy: result.capacityOk,
6045+ capacityWarning: result.capacityWarning,
58546046 federationHealthy: result.federationOk,
58556047 snapshotHealthy: result.snapshotOk,
58566048 watchdogHealthy: result.watchdogOk,
@@ -5859,13 +6051,14 @@ handleHealth = \authorization respond -> do
58596051 else yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: True {},
58606052 heartbeatHealthy: True {},
58616053 capacityHealthy: True {},
6054+ capacityWarning: False {},
58626055 federationHealthy: True {},
58636056 snapshotHealthy: True {},
58646057 watchdogHealthy: True {},
58656058 nonceHealthy: True {},
58666059 draining: result.isDraining})
58676060
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}}
58696062
58706063handleStats = \authorization respond -> do
58716064 t <- now
@@ -5952,9 +6145,12 @@ handleStats = \authorization respond -> do
59526145 let oldestAckAge = if oldestAckTs == 0 then 0 else t - oldestAckTs
59536146 let fwdServerCount = fwdStats.serverCount
59546147 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}
59556151 yield (respond {ok: True {}, version: serverBuildVersion,
59566152 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},
59586154 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},
59596155 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}})
59606156 else yield (respond zeroStats)
@@ -6678,7 +6874,7 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do
66786874 when (result.revokedCount > 0)
66796875 (logInfo ("admin: revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
66806876 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))
66826878 yield (respond {ok: True {}, revokedCount: result.revokedCount})
66836879
66846880-- POST /maintenance/messages/purge: Purge all pending messages for a specific pubkey.
@@ -6747,7 +6943,7 @@ handleAdminRevokeIp = \ip authorization respond -> do
67476943 when (count offlines > 0 && federationEnabled) (do
67486944 peerServers <- *knownPeerServers
67496945 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]}
67516947 sendBatchedGossip peerServers gossipEvents gossipOriginHops)))
67526948 when (result.revokedCount > 0)
67536949 (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
69157111 FederationAck {} -> "POST /federation/ack"
69167112 FederationCancel {} -> "POST /federation/cancel"
69177113 FederationBatchCancel {} -> "POST /federation/cancel-batch"
7114+ FederationBatchForward {} -> "POST /federation/forward-batch"
69187115 VerifyToken {} -> "POST /federation/verify"
69197116 Prune {} -> "POST /maintenance/prune"
69207117 Drain {} -> "POST /maintenance/drain"
@@ -6969,6 +7166,7 @@ requestClientIp = \req -> case req of
69697166 FederationAck {} -> ""
69707167 FederationCancel {} -> ""
69717168 FederationBatchCancel {} -> ""
7169+ FederationBatchForward {} -> ""
69727170 Prune {} -> ""
69737171 Drain {} -> ""
69747172 Shutdown {} -> ""
@@ -7024,6 +7222,7 @@ serveInner = \req -> case req of
70247222 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
70257223 FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationCancel messageId toKey fromServer requestTimestamp correlationId authorization respond
70267224 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
70277226 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
70287227 Prune {authorization, respond} -> handlePrune authorization respond
70297228 Drain {phase, authorization, respond} -> handleDrain phase authorization respond