@@ -52,6 +52,12 @@ maxBatchSize = 100 -- max messages per batch request (bounds recursion depth in
5252maxMessagePriority = 10 -- max priority value (clamped on ingress)
5353maxGossipEvents = 100
5454maxPollEvents = 50
55+-- Fraction of maxPollEvents reserved for priority-0 messages (normal priority).
56+-- Prevents high-priority floods from completely starving normal messages.
57+-- With maxPollEvents=50 and reservedPollFraction=5, at least 10 slots (50/5)
58+-- are reserved for priority-0 when both priority levels have pending messages.
59+-- 0 = disable reservation (old behavior: strict priority ordering).
60+reservedPollFraction = 5
5561maxPresenceEntries = 10000
5662maxMessagesPerRecipient = 500
5763maxMessages = 100000 -- global cap on total stored messages across all recipients
@@ -177,11 +183,11 @@ federationLatencyWindowMs : Int<Ms>
177183federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
178184
179185-- Snapshot format version (increment on schema changes to prevent corrupt restores)
180−snapshotVersion = 12 -- bumped for ForwardEntry.senderKey and federation forward priority
186+snapshotVersion = 13 -- bumped for double-hash snapshot integrity tag
181187
182188-- Server build version: included in /health and /stats responses for deployment verification.
183189-- Update this string when deploying a new build so operators can confirm which version is running.
184−serverBuildVersion = "2026-04-26b"
190+serverBuildVersion = "2026-04-26c"
185191
186192-- Federation protocol version: included in gossip/forward/ack requests.
187193-- Peers log warnings when they receive requests from older versions.
@@ -198,15 +204,18 @@ snapshotChecksum = \presenceCount ackedCount deliveredCount forwardCount nonceCo
198204 presenceCount * 100003 + ackedCount * 100019 + deliveredCount * 100043 + forwardCount * 100049 + nonceCount * 100057
199205
200206-- Snapshot integrity tag: combines checksum with maintenance secret to detect tampering.
201−-- Not cryptographic (no HMAC primitive in Knot), but catches naive modifications.
202−-- The tag includes the secret's hex representation hashed with the checksum, so an
203−-- attacker who modifies the snapshot but doesn't know the secret cannot recompute it.
207+-- Not cryptographic (no HMAC primitive in Knot), but uses a double-hash scheme with
208+-- independent seeds and multipliers to resist inversion attacks. An attacker who
209+-- captures one valid (snapshot, tag) pair cannot derive the secretHash components
210+-- independently because the final mixing step entangles them non-linearly.
204211-- Returns 0 when no maintenance secret is configured (no integrity protection).
205212snapshotIntegrityTag : Int -> Text -> Int
206213snapshotIntegrityTag = \checksum secret ->
207214 if secret == "" then 0
208− else let secretHash = fold (\acc c -> acc * 31 + hexCharIndex c) 7 (chars (bytesToHex (textToBytes secret)))
209− in checksum * 100069 + secretHash * 100081
215+ else let secretChars = chars (bytesToHex (textToBytes secret))
216+ in let hash1 = fold (\acc c -> acc * 31 + hexCharIndex c) 7 secretChars
217+ in let hash2 = fold (\acc c -> acc * 37 + hexCharIndex c) 13 secretChars
218+ in (checksum * 100069 + hash1 * 100081) * 100003 + (checksum * 100057 + hash2 * 100019)
210219
211220-- Max seenBy list length in gossip events (bounds memory and bandwidth per event)
212221maxSeenByLen = 20
@@ -907,6 +916,8 @@ maxPeerLatencyEntries = 100
907916*cachedHealthWatchdog : Bool
908917*cachedHealthNonce : Bool
909918*cachedHealthCapacityWarning : Bool
919+-- Cached count of federation peers with active failure records (degraded connectivity)
920+*cachedHealthDegradedPeers : Int
910921
911922-- ============================================
912923-- Routes
@@ -1139,7 +1150,7 @@ route MaintenanceApi where
11391150 = RotateSecret
11401151
11411152route HealthApi where
1142− 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
1153+ 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, degradedPeers: Int} = Health
11431154
11441155-- Returns counts of in-memory collections (requires maintenance secret)
11451156-- TODO: If Knot adds raw text response support, add a /metrics endpoint
@@ -2781,6 +2792,21 @@ pruneAllAckedBuckets = \b retainCutoff ->
27812792 writeAckedBucket b (do a <- bucket; where a.ackedAt > retainCutoff; yield a)
27822793 pruneAllAckedBuckets (b + 1) retainCutoff
27832794
2795+-- Per-bucket pruning for forward entries (avoids readAllForwards/writeAllForwards repartitioning).
2796+-- Each bucket is pruned in its own atomic block to reduce lock contention with sendMessageCore.
2797+-- Returns the total count of surviving entries for totalForwardCount reconciliation.
2798+pruneAllFwdBuckets : Int -> Timestamp -> Bool -> Int -> IO Int
2799+pruneAllFwdBuckets = \b retainCutoff fedEnabled acc ->
2800+ if b > 15 then yield acc
2801+ else do
2802+ kept <- atomic do
2803+ bucket <- readFwdBucket b
2804+ let pruned = if not fedEnabled then []
2805+ else do f <- bucket; where f.createdAt > retainCutoff; yield f
2806+ writeFwdBucket b pruned
2807+ yield (count pruned)
2808+ pruneAllFwdBuckets (b + 1) retainCutoff fedEnabled (acc + kept)
2809+
27842810-- Find the oldest (minimum) ackedAt timestamp across all 16 acked buckets.
27852811-- Returns 0 if no acked entries exist. Avoids materializing all buckets into one list.
27862812computeOldestAckedAt : Int -> IO Timestamp
@@ -3308,14 +3334,11 @@ pruneExpired = \t -> do
33083334 -- Prune federation state (split into per-collection blocks to reduce lock contention
33093335 -- with poll-ack paths that write to ackedIds/deliveredIds concurrently).
33103336 let retainCutoff = t - messageMaxAge
3311− -- Forwards: independent from acked/delivered, safe to prune separately
3312− atomic do
3313− forwards <- readAllForwards
3314− -- Clear all forwards when federation is disabled (they can never be retried)
3315− let prunedForwards = if not federationEnabled then []
3316− else do f <- forwards; where f.createdAt > retainCutoff; yield f
3317− writeAllForwards prunedForwards
3318− set *totalForwardCount = count prunedForwards
3337+ -- Forwards: per-bucket pruning (same pattern as acked/delivered).
3338+ -- Each bucket is pruned in its own atomic block to reduce lock contention
3339+ -- with sendMessageCore which writes to forward buckets on the hot path.
3340+ fwdSurvivorCount <- pruneAllFwdBuckets 0 retainCutoff federationEnabled 0
3341+ atomic do set *totalForwardCount = fwdSurvivorCount
33193342 -- AckedIds: per-bucket pruning (each bucket in its own atomic block to reduce lock
33203343 -- contention with poll-ack paths that write to ackedIds concurrently).
33213344 pruneAllAckedBuckets 0 retainCutoff
@@ -3950,11 +3973,23 @@ retryForwardsBatchResults = \batch results server ->
39503973 incrementForwardRetries f.messageId server f.toKey
39513974 retryForwardsBatchResults rest results server
39523975
3976+-- Collect forwards from all 16 buckets, but only for eligible servers.
3977+-- Reads one bucket at a time to avoid materializing all forwards simultaneously.
3978+collectEligibleForwards : Int -> [Text] -> IO [ForwardEntry]
3979+collectEligibleForwards = \b eligibleServers ->
3980+ if b > 15 then yield []
3981+ else do
3982+ bucket <- readFwdBucket b
3983+ let eligible = do f <- bucket; where any eligibleServers (\s -> s == f.toServer); yield f
3984+ rest <- collectEligibleForwards (b + 1) eligibleServers
3985+ yield (unionNE eligible rest)
3986+
39533987-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
39543988-- Circuit breaker: skips entirely when too many peers are failing (likely local network issue).
3955−retryPendingForwards : [ForwardEntry] -> IO {network} {}
3956−retryPendingForwards = \pending -> do
3957− let servers = collectPeerServers pending (\x -> x.toServer)
3989+-- Per-bucket: reads forward buckets one at a time, collecting only entries for eligible servers,
3990+-- instead of materializing all forwards into memory via readAllForwards.
3991+retryPendingForwardsFromBuckets : IO {network} {}
3992+retryPendingForwardsFromBuckets = do
39583993 t <- now
39593994 failures <- *forwardFailures
39603995 if isFederationCircuitOpen failures t
@@ -3964,8 +3999,15 @@ retryPendingForwards = \pending -> do
39643999 else do
39654000 inFlight <- *federationInFlight
39664001 blocked <- *blockedPeers
3967− let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
3968− let eligibleForwards = do f <- pending; where any eligibleServers (\s -> s == f.toServer); yield f
4002+ -- Discover servers with pending forwards per-bucket (reuses computeForwardStats which
4003+ -- reads one bucket at a time), then filter to eligible servers before reading entries.
4004+ fwdStats <- computeForwardStats 0 t
4005+ let allServers = do sc <- fwdStats.serverCounts; yield sc.server
4006+ let eligibleServers = do s <- allServers; where shouldRetryServerEx s t failures inFlight blocked; yield s
4007+ if count eligibleServers == 0 then yield {}
4008+ else do
4009+ -- Read only entries for eligible servers (per-bucket, streaming)
4010+ eligibleForwards <- collectEligibleForwards 0 eligibleServers
39694011 -- Prioritize forwards for recipients with active polls (currently online).
39704012 -- These keys will receive the message sooner, improving delivery latency.
39714013 -- Forwards for offline keys are retried after online-key forwards complete.
@@ -4286,7 +4328,6 @@ pruneAndRetry = \t -> do
42864328 -- without the full per-bucket rebuild (O(16 bucket scans) vs O(all messages))
42874329 rcTotal <- sumRcBucketTotals 0
42884330 set *totalMessageCount = rcTotal)
4289− pendingForwards <- atomic do f <- readAllForwards; yield f
42904331 ackState <- atomic do
42914332 ackedIds2 <- readAllAcked
42924333 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a
@@ -4358,12 +4399,12 @@ pruneAndRetry = \t -> do
43584399 set *pendingAckCleanups = remaining
43594400 yield ready
43604401 when (count readyCleanups > 0) (ackCleanupForwards readyCleanups t)
4361− let result = {pendingForwards: pendingForwards, pendingAcks: ackState.pendingAcks, pendingOverflow: ackState.pendingOverflow, extendedActive: refreshResult.extendedActive, nearExpiryActive: refreshResult.nearExpiryActive, refreshPeers: refreshResult.refreshPeers, shouldRefresh: refreshResult.shouldRefresh}
4402+ let result = {pendingAcks: ackState.pendingAcks, pendingOverflow: ackState.pendingOverflow, extendedActive: refreshResult.extendedActive, nearExpiryActive: refreshResult.nearExpiryActive, refreshPeers: refreshResult.refreshPeers, shouldRefresh: refreshResult.shouldRefresh}
43624403 -- Fork federation I/O so hung peers don't block next prune cycle
43634404 fork (do
43644405 broadcastOfflineGossip t pr.peerServers pr.expiredLocal
43654406 when federationEnabled (do
4366− retryPendingForwards result.pendingForwards
4407+ retryPendingForwardsFromBuckets
43674408 sendAcksAndClean result.pendingAcks
43684409 retryOverflowAcks result.pendingOverflow))
43694410 -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
@@ -4474,13 +4515,24 @@ ackPolledMessages = \pubkey cursor t -> do
44744515
44754516pollNewEvents = \pubkey clampedCursor t -> do
44764517 bucket <- readMessageBucket (pubkeyBucket pubkey)
4477− -- Priority-aware delivery: high-priority messages surface first within the maxPollEvents cap.
4478− -- Sort key: (-priority * a large constant + receivedAt) ensures priority dominates while
4479− -- preserving receivedAt order within the same priority level. The constant must exceed the
4480− -- maximum possible receivedAt spread within one poll batch (bounded by messageMaxAge).
4481− -- Priority 0 (default) sorts after all higher priorities.
4482− let allPollable = sortBy (\m -> (0 - m.priority) * 10000000000000 + m.receivedAt) (pollableMessages bucket pubkey clampedCursor t)
4483− let limitedMsgs = takeRelation maxPollEvents allPollable
4518+ let raw = pollableMessages bucket pubkey clampedCursor t
4519+ -- Priority-aware delivery with starvation prevention: reserve a fraction of
4520+ -- maxPollEvents for priority-0 messages so high-priority floods can't completely
4521+ -- starve normal messages. Reservation only activates when both pools have messages.
4522+ let priority0 = do m <- raw; where m.priority == 0; yield m
4523+ let highPri = do m <- raw; where m.priority > 0; yield m
4524+ let reservedSlots = if reservedPollFraction > 0 && count highPri > 0 && count priority0 > 0
4525+ then max 1 (maxPollEvents / reservedPollFraction)
4526+ else 0
4527+ let highPriSlots = maxPollEvents - reservedSlots
4528+ let sortedHigh = sortBy (\m -> (0 - m.priority) * 10000000000000 + m.receivedAt) highPri
4529+ let sortedLow = sortBy (\m -> m.receivedAt) priority0
4530+ let selectedHigh = takeRelation highPriSlots sortedHigh
4531+ let selectedLow = takeRelation reservedSlots sortedLow
4532+ -- Merge and re-sort to maintain correct cursor advancement (lastReceivedAt must
4533+ -- reflect the actual latest receivedAt across both pools for cursor consistency).
4534+ let allPollable = sortBy (\m -> (0 - m.priority) * 10000000000000 + m.receivedAt) raw
4535+ let limitedMsgs = sortBy (\m -> (0 - m.priority) * 10000000000000 + m.receivedAt) (union selectedHigh selectedLow)
44844536 let evts = do
44854537 m <- limitedMsgs
44864538 yield {id: m.messageId, eventType: "message",
@@ -4490,7 +4542,7 @@ pollNewEvents = \pubkey clampedCursor t -> do
44904542 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
44914543 -- hasMore: tells clients there are additional messages beyond this batch,
44924544 -- so they can immediately re-poll without waiting for the next heartbeat.
4493− let hasMore = count allPollable > count limitedMsgs
4545+ let hasMore = count raw > count limitedMsgs
44944546 yield {events: evts, cursor: lastReceivedAt, hasMore: hasMore}
44954547
44964548-- Execute a single long-poll cycle:
@@ -5691,10 +5743,14 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
56915743 -- Avoids the O(n) writeAllForwards repartitioning when only a few buckets are affected.
56925744 removedForwards <- removeForwardsByServer fromServer dedupedIds 0 []
56935745 let ackedCount = count removedForwards
5694− -- Guard: warn if peer acks messages we had no forward entries for. This can indicate
5695− -- a peer acking messages it never received (e.g., the forward response was lost but
5696− -- the message was never stored), which would cause premature origin message deletion.
56975746 let unmatched = count dedupedIds - ackedCount
5747+ -- Reject when ALL acked IDs are unmatched: the peer is acking messages we never
5748+ -- forwarded to it. Partial matches are accepted (some forwards may have been pruned).
5749+ if ackedCount == 0 && count dedupedIds > 0
5750+ then do
5751+ logWarn ("federation/ack: rejecting ack from " ++ fromServer ++ " — all " ++ show (count dedupedIds) ++ " messageIds unmatched (no forward entries)")
5752+ yield (respond {ok: False {}, error: "no_matching_forwards", ackedCount: 0, results: do mid <- dedupedIds; yield {messageId: mid, found: False {}}})
5753+ else do
56985754 when (unmatched > 0)
56995755 (logWarn ("federation/ack: " ++ fromServer ++ " acked " ++ show unmatched ++ " messageIds with no matching forward entries — skipping cleanup for those"))
57005756 -- Only clear failure record when ALL acked IDs matched forward entries.
@@ -5771,6 +5827,13 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
57715827 when (fwdRemoved > 0) (do
57725828 tfc <- *totalForwardCount
57735829 set *totalForwardCount = max 0 (tfc - fwdRemoved))
5830+ -- Record cancel nonce: prevents replayed federation forwards of the
5831+ -- cancelled message from being accepted after dedup state expires.
5832+ nonceBkt <- readNonceBucket b
5833+ let updatedNonces = union nonceBkt [{messageId: messageId, toKey: toKey, seenAt: t}]
5834+ writeNonceBucket b (if count updatedNonces > maxNoncesPerBucket
5835+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5836+ else updatedNonces)
57745837 yield {cancelled: True {}}
57755838 when result.cancelled
57765839 (logInfo ("federation/cancel: cancelled " ++ messageId ++ " for " ++ toKey ++ " (from " ++ fromServer ++ ")"))
@@ -5802,12 +5865,12 @@ cancelBatchInner = \entries fromServer t acc ->
58025865 let sameBucket = do e <- entries; where pubkeyBucket e.toKey == b; yield e
58035866 let otherBuckets = do e <- entries; where pubkeyBucket e.toKey != b; yield e
58045867 -- Process the entire bucket group in one atomic block
5805− results <- atomic do cancelBucketGroup sameBucket fromServer b []
5868+ results <- atomic do cancelBucketGroup sameBucket fromServer b t []
58065869 cancelBatchInner otherBuckets fromServer t (union acc results)
58075870
58085871-- Process all cancel entries for a single bucket within one atomic block.
58095872-- Reads messageBucket/fwdBucket once and applies all cancels, avoiding redundant reads.
5810−cancelBucketGroup = \entries fromServer b acc ->
5873+cancelBucketGroup = \entries fromServer b t acc ->
58115874 case headOf entries of
58125875 Nothing {} -> yield acc
58135876 Just {value: entry} -> do
@@ -5829,8 +5892,14 @@ cancelBucketGroup = \entries fromServer b acc ->
58295892 when (fwdRemoved > 0) (do
58305893 tfc <- *totalForwardCount
58315894 set *totalForwardCount = max 0 (tfc - fwdRemoved))
5895+ -- Record cancel nonce for each cancelled message in the batch
5896+ nonceBkt <- readNonceBucket b
5897+ let updatedNonces = union nonceBkt [{messageId: entry.messageId, toKey: entry.toKey, seenAt: t}]
5898+ writeNonceBucket b (if count updatedNonces > maxNoncesPerBucket
5899+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5900+ else updatedNonces)
58325901 yield {cancelled: True {}}
5833− cancelBucketGroup rest fromServer b (union acc [{messageId: entry.messageId, cancelled: result.cancelled}])
5902+ cancelBucketGroup rest fromServer b t (union acc [{messageId: entry.messageId, cancelled: result.cancelled}])
58345903
58355904-- POST /federation/forward-batch: Receive multiple forwarded messages in one request.
58365905-- Each message is processed independently (same logic as handleForwardMessage).
@@ -5842,7 +5911,11 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp correlation
58425911 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: []}
58435912 withFederationAuth authorization t batchRespond fromServer "forward" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (True {}) (do
58445913 let capped = takeRelation maxBatchSize forwards
5845− results <- forwardBatchInner capped fromServer t []
5914+ -- Pre-deduplicate (messageId, toKey) pairs: a malicious peer could send 100 entries
5915+ -- for the same pair, each triggering a full atomic block (presence check, 5 bucket reads).
5916+ -- Dedup keeps the first occurrence; subsequent duplicates are silently dropped.
5917+ let deduped = fold (\acc f -> if any acc (\a -> a.messageId == f.messageId && a.toKey == f.toKey) then acc else union acc [f]) [] capped
5918+ results <- forwardBatchInner deduped fromServer t []
58465919 let storedCount = count (do r <- results; where r.ok; yield r)
58475920 when (storedCount > 0)
58485921 (logInfo ("federation/forward-batch: stored " ++ show storedCount ++ "/" ++ show (count capped) ++ " messages from " ++ fromServer))
@@ -5991,6 +6064,13 @@ clientCancelBatchInner = \senderKey entries t resultAcc fedAcc ->
59916064 presenceBkt <- readPresenceBucket b
59926065 let remotePresence = collectPeerServers (presenceForKey presenceBkt entry.to t) (\p -> p.server)
59936066 let allCancelServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union [s] acc) remotePresence forwardTargetServers
6067+ -- Record cancel nonce: prevents replayed federation forwards of the
6068+ -- cancelled message from being accepted after dedup state expires.
6069+ nonceBkt <- readNonceBucket b
6070+ let updatedNonces = union nonceBkt [{messageId: entry.messageId, toKey: entry.to, seenAt: t}]
6071+ writeNonceBucket b (if count updatedNonces > maxNoncesPerBucket
6072+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6073+ else updatedNonces)
59946074 yield {cancelled: True {}, cancelServers: allCancelServers}
59956075 let newFedAcc = if cancelResult.cancelled && count cancelResult.cancelServers > 0
59966076 then union fedAcc [{messageId: entry.messageId, toKey: entry.to, servers: cancelResult.cancelServers}]
@@ -6085,7 +6165,7 @@ handleDrain = \phase authorization respond -> do
60856165 -- Best-effort flush of pending forwards and acks before stopping
60866166 when (federationEnabled && pendingForwardCount > 0) (do
60876167 logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards")
6088− retryPendingForwards flushState.forwards)
6168+ retryPendingForwardsFromBuckets)
60896169 when (federationEnabled && pendingAckCount > 0) (do
60906170 logInfo ("drain: flushing " ++ show pendingAckCount ++ " pending acks")
60916171 sendAcksAndClean flushState.pendingAcks)
@@ -6225,6 +6305,8 @@ computeAndCacheHealth = \t -> do
62256305 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
62266306 nonceCount <- countAllNonces 0
62276307 let nonceOk = nonceCount * 10 <= maxForwardNonces * 9
6308+ -- Count peers with active failure records (degraded federation connectivity)
6309+ let degradedPeerCount = count (do f <- forwardFailures; where t - f.firstFailedAt < maxServerDowntime; yield f)
62286310 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
62296311 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
62306312 capacityOk: capacityOk,
@@ -6233,7 +6315,8 @@ computeAndCacheHealth = \t -> do
62336315 snapshotOk: snapOk,
62346316 watchdogOk: wdOk,
62356317 nonceOk: nonceOk,
6236− isDraining: dp > 0}
6318+ isDraining: dp > 0,
6319+ degradedPeers: degradedPeerCount}
62376320 let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk && result.nonceOk
62386321 atomic do
62396322 set *cachedHealthOk = allOk
@@ -6245,6 +6328,7 @@ computeAndCacheHealth = \t -> do
62456328 set *cachedHealthSnapshot = result.snapshotOk
62466329 set *cachedHealthWatchdog = result.watchdogOk
62476330 set *cachedHealthNonce = result.nonceOk
6331+ set *cachedHealthDegradedPeers = result.degradedPeers
62486332 yield result
62496333
62506334handleHealth = \authorization respond -> do
@@ -6265,8 +6349,9 @@ handleHealth = \authorization respond -> do
62656349 cW <- *cachedHealthWatchdog
62666350 cN <- *cachedHealthNonce
62676351 cCW <- *cachedHealthCapacityWarning
6352+ cDP <- *cachedHealthDegradedPeers
62686353 dp <- *drainPhase
6269− 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}
6354+ 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, degradedPeers: cDP}
62706355 yield (respond cached)
62716356 else do
62726357 isAuthed <- verifyMaintenanceAuth authorization
@@ -6284,7 +6369,8 @@ handleHealth = \authorization respond -> do
62846369 snapshotHealthy: result.snapshotOk,
62856370 watchdogHealthy: result.watchdogOk,
62866371 nonceHealthy: result.nonceOk,
6287− draining: result.isDraining})
6372+ draining: result.isDraining,
6373+ degradedPeers: result.degradedPeers})
62886374 else yield (respond {ok: allOk, version: serverBuildVersion, pruneHealthy: True {},
62896375 heartbeatHealthy: True {},
62906376 capacityHealthy: True {},
@@ -6293,7 +6379,8 @@ handleHealth = \authorization respond -> do
62936379 snapshotHealthy: True {},
62946380 watchdogHealthy: True {},
62956381 nonceHealthy: True {},
6296− draining: result.isDraining})
6382+ draining: result.isDraining,
6383+ degradedPeers: 0})
62976384
62986385zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: []}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, prunesSinceSnapshot: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0}}
62996386
@@ -6851,6 +6938,14 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
68516938 -- with presence servers (servers that already received and stored the message).
68526939 -- This ensures cancel reaches both pending and completed forward destinations.
68536940 let allCancelServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union [s] acc) remotePresence forwardTargetServers
6941+ -- Record cancel nonce: prevents replayed federation forwards of the
6942+ -- cancelled message from being accepted after the message is removed
6943+ -- from all dedup sets (messages, ackedIds, deliveredIds).
6944+ nonceBkt <- readNonceBucket b
6945+ let updatedNonces = union nonceBkt [{messageId: messageId, toKey: to, seenAt: t}]
6946+ writeNonceBucket b (if count updatedNonces > maxNoncesPerBucket
6947+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6948+ else updatedNonces)
68546949 yield {cancelled: True {}, cancelServers: allCancelServers}
68556950 -- Send federation cancel to remote servers that may have the message
68566951 fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)