@@ -335,7 +335,7 @@ type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int}
335335type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
336336-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles
337337type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}
338−type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool}
338+type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool, pendingForwards: Int}
339339type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
340340type RecipientCount = {pubkey: PubkeyHex, count: Int}
341341type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
@@ -723,7 +723,7 @@ route AuthApi where
723723route ClientApi where
724724 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
725725 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
726− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int}
726+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool}
727727 = Poll
728728 /messages
729729 -- Send an E2E encrypted message to a recipient pubkey
@@ -838,7 +838,7 @@ route MaintenanceApi where
838838 = SetRateLimits
839839
840840route HealthApi where
841− GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool} = Health
841+ GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool, draining: Bool} = Health
842842
843843-- Returns counts of in-memory collections (requires maintenance secret)
844844-- TODO: If Knot adds raw text response support, add a /metrics endpoint
@@ -1670,10 +1670,13 @@ pruneMessageBucket = \b messageCutoff -> atomic do
16701670-- deltas and apply them in a single final atomic block.
16711671pruneMessageBucketsConcurrent = \b messageCutoff ->
16721672 if b > 15 then yield {}
1673− else if b < 15 then do
1674− fork (pruneMessageBucket b messageCutoff)
1675− pruneMessageBucketsConcurrent (b + 1) messageCutoff
1676− else pruneMessageBucket b messageCutoff
1673+ else do
1674+ bucket <- readMessageBucket b
1675+ if count bucket == 0 then pruneMessageBucketsConcurrent (b + 1) messageCutoff
1676+ else if b < 15 then do
1677+ fork (pruneMessageBucket b messageCutoff)
1678+ pruneMessageBucketsConcurrent (b + 1) messageCutoff
1679+ else pruneMessageBucket b messageCutoff
16771680
16781681-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16
16791682-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one
@@ -3783,13 +3786,17 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
37833786handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
37843787 t <- now
37853788 serverGen <- *serverGeneration
3786− let effectiveCursor = if generation == serverGen then cursor else 0
3789+ let generationMismatch = generation != serverGen
3790+ let effectiveCursor = if generationMismatch then 0 else cursor
3791+ -- Cursor reset: generation changed (server restarted) and client had a non-zero cursor.
3792+ -- Tells the client their cursor was invalidated so they can handle re-delivery.
3793+ let cursorWasReset = generationMismatch && cursor > 0
37873794 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
37883795 draining <- *draining
3789− if draining then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0})
3796+ if draining then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset})
37903797 else do
37913798 globalRateLimited <- checkGlobalRate t
3792− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs})
3799+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset})
37933800 else do
37943801 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
37953802 startPresenceVersion <- *presenceVersion
@@ -3812,7 +3819,7 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
38123819 sendAcksAndClean result.pendingAcks
38133820 retryEvictedAcks result.evictedEntries result.pendingAcks))
38143821 let isHeartbeat = result.authorized && count result.events == 0
3815− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0})
3822+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0, cursorReset: cursorWasReset})
38163823
38173824-- Core send logic: dedup, rate limiting, store message, create forwards.
38183825-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
@@ -4589,7 +4596,8 @@ handleHealth = \authorization respond -> do
45894596 cF <- *cachedHealthFederation
45904597 cS <- *cachedHealthSnapshot
45914598 cW <- *cachedHealthWatchdog
4592− yield {ok: cOk, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW}
4599+ isDraining <- *draining
4600+ yield {ok: cOk && not isDraining, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, draining: isDraining}
45934601 yield (respond cached)
45944602 else do
45954603 let isAuthed = verifyMaintenanceAuth authorization
@@ -4607,6 +4615,7 @@ handleHealth = \authorization respond -> do
46074615 forwardFailures <- *forwardFailures
46084616 pruneRun <- *pruneRunning
46094617 lastSnap <- *lastSnapshotExportAt
4618+ isDraining <- *draining
46104619 let warmup = t - sa < healthPruneMaxAge
46114620 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
46124621 -- Capacity pressure: flag when any capped collection exceeds 90% full
@@ -4631,8 +4640,10 @@ handleHealth = \authorization respond -> do
46314640 capacityOk: capacityOk,
46324641 federationOk: fedOk,
46334642 snapshotOk: snapOk,
4634− watchdogOk: wdOk}
4635− let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
4643+ watchdogOk: wdOk,
4644+ isDraining: isDraining}
4645+ -- Draining servers report ok=False so load balancers stop routing traffic to them
4646+ let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
46364647 -- Cache health result for rate-limited responses
46374648 atomic do
46384649 set *cachedHealthOk = allOk
@@ -4644,19 +4655,22 @@ handleHealth = \authorization respond -> do
46444655 set *cachedHealthWatchdog = result.watchdogOk
46454656 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
46464657 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
4658+ -- Draining status is always visible (load balancers need it for routing decisions).
46474659 if isAuthed
46484660 then yield (respond {ok: allOk, pruneHealthy: result.pruneOk,
46494661 heartbeatHealthy: result.heartbeatOk,
46504662 capacityHealthy: result.capacityOk,
46514663 federationHealthy: result.federationOk,
46524664 snapshotHealthy: result.snapshotOk,
4653− watchdogHealthy: result.watchdogOk})
4665+ watchdogHealthy: result.watchdogOk,
4666+ draining: result.isDraining})
46544667 else yield (respond {ok: allOk, pruneHealthy: True {},
46554668 heartbeatHealthy: True {},
46564669 capacityHealthy: True {},
46574670 federationHealthy: True {},
46584671 snapshotHealthy: True {},
4659− watchdogHealthy: True {}})
4672+ watchdogHealthy: True {},
4673+ draining: result.isDraining})
46604674
46614675zeroStats = {ok: False {}, 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, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 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}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}}}
46624676
@@ -4742,8 +4756,16 @@ handleStatsPeers = \authorization respond -> do
47424756 if authorized
47434757 then do
47444758 failures <- *forwardFailures
4745− let peers = do f <- failures; yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoffFor f.failures f.server, retriable: shouldRetryServer f.server t failures}
4746− yield (respond {peers: peers, ok: True {}})
4759+ forwards <- readAllForwards
4760+ -- Count pending forwards per server in a single pass
4761+ let fwdCounts = fold (\acc f ->
4762+ if any acc (\a -> a.server == f.toServer) then do a <- acc; yield (if a.server == f.toServer then {server: a.server, count: a.count + 1} else a)
4763+ else union acc [{server: f.toServer, count: 1}]) [] forwards
4764+ let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
4765+ let peers = do f <- failures; yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoffFor f.failures f.server, retriable: shouldRetryServer f.server t failures, pendingForwards: fwdCountFor f.server}
4766+ -- Include servers with pending forwards but no failure records
4767+ let fwdOnlyPeers = do c <- fwdCounts; where none failures (\f -> f.server == c.server); yield {server: c.server, failures: 0, lastFailedAt: 0, firstFailedAt: 0, backoffMs: 0, retriable: True {}, pendingForwards: c.count}
4768+ yield (respond {peers: union peers fwdOnlyPeers, ok: True {}})
47474769 else yield (respond {peers: [], ok: False {}})
47484770
47494771handleStatsKey = \pubkey authorization respond -> do
@@ -5155,12 +5177,14 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do
51555177-- POST /maintenance/messages/purge: Purge all pending messages for a specific pubkey.
51565178-- Use for inbox-bombing abuse response: clears the recipient's mailbox without
51575179-- requiring their session token. Also cancels pending forwards for purged messages.
5180+-- Federated messages (originServer != "") generate ack entries so origin servers
5181+-- can reclaim their forward entries instead of retrying until prune.
51585182handleAdminPurgeMessages = \pubkey authorization respond -> do
51595183 let authorized = verifyMaintenanceAuth authorization
51605184 if not authorized then yield (respond {ok: False {}, purgedCount: 0})
51615185 else do
51625186 t <- now
5163− purgedCount <- atomic do
5187+ purgeResult <- atomic do
51645188 let b = pubkeyBucket pubkey
51655189 bucket <- readMessageBucket b
51665190 let toDelete = do m <- bucket; where m.toKey == pubkey; yield m
@@ -5176,11 +5200,27 @@ handleAdminPurgeMessages = \pubkey authorization respond -> do
51765200 let pfb = pubkeyBucket pubkey
51775201 fwdBkt <- readFwdBucket pfb
51785202 writeFwdBucket pfb (do f <- fwdBkt; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
5179− yield removed
5180− else yield 0
5181− when (purgedCount > 0)
5182− (logInfo ("admin: purged " ++ show purgedCount ++ " messages for " ++ pubkey))
5183− yield (respond {ok: True {}, purgedCount: purgedCount})
5203+ -- Create ack entries for federated messages so origin servers learn about the purge
5204+ let federatedPurged = do m <- toDelete; where m.originServer != ""; yield m
5205+ let ackedFederated = federatedAckEntries federatedPurged t
5206+ if count ackedFederated > 0 then do
5207+ ackedBkt <- readAckedBucket b
5208+ let combinedAcked = union ackedBkt ackedFederated
5209+ let excessCount = max 0 (count combinedAcked - maxAckedIds)
5210+ let eviction = evictSmallest combinedAcked excessCount (\a -> a.ackedAt)
5211+ writeAckedBucket b eviction.kept
5212+ yield {purgedCount: removed, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
5213+ else yield {purgedCount: removed, pendingAcks: [], evictedEntries: []}
5214+ else yield {purgedCount: 0, pendingAcks: [], evictedEntries: []}
5215+ when (purgeResult.purgedCount > 0)
5216+ (logInfo ("admin: purged " ++ show purgeResult.purgedCount ++ " messages for " ++ pubkey))
5217+ when (count purgeResult.pendingAcks > 0)
5218+ (logInfo ("admin: sending federation acks for " ++ show (count purgeResult.pendingAcks) ++ " purged federated messages"))
5219+ -- Send federation acks in the background (same pattern as handleAckMessages)
5220+ fork (when federationEnabled (do
5221+ sendAcksAndClean purgeResult.pendingAcks
5222+ retryEvictedAcks purgeResult.evictedEntries purgeResult.pendingAcks))
5223+ yield (respond {ok: True {}, purgedCount: purgeResult.purgedCount})
51845224
51855225-- GET /maintenance/rate-limits: Return current effective rate limits.
51865226handleGetRateLimits = \authorization respond -> do