29c98d0 · skrepka · git.iko.soy

git.iko.soy/skrepka

Implement 5 server improvements: health drain status for load balancers, skip empty bucket prune forks, per-server forward queue depth in stats, poll cursor reset flag, and federation acks on admin message purge

29c98d0iko6 months ago

server.knot+64−24

compare with master

unifiedsplit

@@ -335,7 +335,7 @@ type IpRateEntry = {ip: Text, windowStart: Timestamp, count: Int}

335type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}

336-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles

337type 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}

339type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}

340type RecipientCount = {pubkey: PubkeyHex, count: Int}

341type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}

@@ -723,7 +723,7 @@ route AuthApi where

723route ClientApi where

724 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp

725 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}

727 = Poll

728 /messages

729 -- Send an E2E encrypted message to a recipient pubkey

@@ -838,7 +838,7 @@ route MaintenanceApi where

838 = SetRateLimits

839

840route 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

842

843-- Returns counts of in-memory collections (requires maintenance secret)

844-- TODO: If Knot adds raw text response support, add a /metrics endpoint

@@ -1670,10 +1670,13 @@ pruneMessageBucket = \b messageCutoff -> atomic do

1670-- deltas and apply them in a single final atomic block.

1671pruneMessageBucketsConcurrent = \b messageCutoff ->

1672 if b > 15 then yield {}

1673+ else do

1674+ bucket <- readMessageBucket b

1675+ if count bucket == 0 then pruneMessageBucketsConcurrent (b + 1) messageCutoff

1673− else if b < 15 then do

1676+ else if b < 15 then do

1674− fork (pruneMessageBucket b messageCutoff)

1677+ fork (pruneMessageBucket b messageCutoff)

1675− pruneMessageBucketsConcurrent (b + 1) messageCutoff

1678+ pruneMessageBucketsConcurrent (b + 1) messageCutoff

1676− else pruneMessageBucket b messageCutoff

1679+ else pruneMessageBucket b messageCutoff

1677

1680

1678-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16

1681-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16

1679-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one

1682-- 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 ->

3783handlePoll = \cursor generation watchKeys authorization clientIp respond -> do

3786handlePoll = \cursor generation watchKeys authorization clientIp respond -> do

3784 t <- now

3787 t <- now

3785 serverGen <- *serverGeneration

3788 serverGen <- *serverGeneration

3789+ let generationMismatch = generation != serverGen

3786− let effectiveCursor = if generation == serverGen then cursor else 0

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

3787 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys

3794 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys

3788 draining <- *draining

3795 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})

3790 else do

3797 else do

3791 globalRateLimited <- checkGlobalRate t

3798 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})

3793 else do

3800 else do

3794 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups

3801 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups

3795 startPresenceVersion <- *presenceVersion

3802 startPresenceVersion <- *presenceVersion

@@ -3812,7 +3819,7 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do

3812 sendAcksAndClean result.pendingAcks

3819 sendAcksAndClean result.pendingAcks

3813 retryEvictedAcks result.evictedEntries result.pendingAcks))

3820 retryEvictedAcks result.evictedEntries result.pendingAcks))

3814 let isHeartbeat = result.authorized && count result.events == 0

3821 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})

3816

3823

3817-- Core send logic: dedup, rate limiting, store message, create forwards.

3824-- Core send logic: dedup, rate limiting, store message, create forwards.

3818-- Called inside an atomic block by both handleSendMessage and sendBatchInner.

3825-- Called inside an atomic block by both handleSendMessage and sendBatchInner.

@@ -4589,7 +4596,8 @@ handleHealth = \authorization respond -> do

4589 cF <- *cachedHealthFederation

4596 cF <- *cachedHealthFederation

4590 cS <- *cachedHealthSnapshot

4597 cS <- *cachedHealthSnapshot

4591 cW <- *cachedHealthWatchdog

4598 cW <- *cachedHealthWatchdog

4599+ isDraining <- *draining

4592− yield {ok: cOk, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW}

4600+ yield {ok: cOk && not isDraining, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, draining: isDraining}

4593 yield (respond cached)

4601 yield (respond cached)

4594 else do

4602 else do

4595 let isAuthed = verifyMaintenanceAuth authorization

4603 let isAuthed = verifyMaintenanceAuth authorization

@@ -4607,6 +4615,7 @@ handleHealth = \authorization respond -> do

4607 forwardFailures <- *forwardFailures

4615 forwardFailures <- *forwardFailures

4608 pruneRun <- *pruneRunning

4616 pruneRun <- *pruneRunning

4609 lastSnap <- *lastSnapshotExportAt

4617 lastSnap <- *lastSnapshotExportAt

4618+ isDraining <- *draining

4610 let warmup = t - sa < healthPruneMaxAge

4619 let warmup = t - sa < healthPruneMaxAge

4611 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2

4620 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2

4612 -- Capacity pressure: flag when any capped collection exceeds 90% full

4621 -- Capacity pressure: flag when any capped collection exceeds 90% full

@@ -4631,8 +4640,10 @@ handleHealth = \authorization respond -> do

4631 capacityOk: capacityOk,

4640 capacityOk: capacityOk,

4632 federationOk: fedOk,

4641 federationOk: fedOk,

4633 snapshotOk: snapOk,

4642 snapshotOk: snapOk,

4634− watchdogOk: wdOk}

4643+ watchdogOk: wdOk,

4644+ isDraining: isDraining}

4645+ -- Draining servers report ok=False so load balancers stop routing traffic to them

4635− let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk

4646+ let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk

4636 -- Cache health result for rate-limited responses

4647 -- Cache health result for rate-limited responses

4637 atomic do

4648 atomic do

4638 set *cachedHealthOk = allOk

4649 set *cachedHealthOk = allOk

@@ -4644,19 +4655,22 @@ handleHealth = \authorization respond -> do

4644 set *cachedHealthWatchdog = result.watchdogOk

4655 set *cachedHealthWatchdog = result.watchdogOk

4645 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,

4656 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,

4646 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.

4657 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.

4658+ -- Draining status is always visible (load balancers need it for routing decisions).

4647 if isAuthed

4659 if isAuthed

4648 then yield (respond {ok: allOk, pruneHealthy: result.pruneOk,

4660 then yield (respond {ok: allOk, pruneHealthy: result.pruneOk,

4649 heartbeatHealthy: result.heartbeatOk,

4661 heartbeatHealthy: result.heartbeatOk,

4650 capacityHealthy: result.capacityOk,

4662 capacityHealthy: result.capacityOk,

4651 federationHealthy: result.federationOk,

4663 federationHealthy: result.federationOk,

4652 snapshotHealthy: result.snapshotOk,

4664 snapshotHealthy: result.snapshotOk,

4653− watchdogHealthy: result.watchdogOk})

4665+ watchdogHealthy: result.watchdogOk,

4666+ draining: result.isDraining})

4654 else yield (respond {ok: allOk, pruneHealthy: True {},

4667 else yield (respond {ok: allOk, pruneHealthy: True {},

4655 heartbeatHealthy: True {},

4668 heartbeatHealthy: True {},

4656 capacityHealthy: True {},

4669 capacityHealthy: True {},

4657 federationHealthy: True {},

4670 federationHealthy: True {},

4658 snapshotHealthy: True {},

4671 snapshotHealthy: True {},

4659− watchdogHealthy: True {}})

4672+ watchdogHealthy: True {},

4673+ draining: result.isDraining})

4660

4674

4661zeroStats = {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 {}}}

4675zeroStats = {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 {}}}

4662

4676

@@ -4742,8 +4756,16 @@ handleStatsPeers = \authorization respond -> do

4742 if authorized

4756 if authorized

4743 then do

4757 then do

4744 failures <- *forwardFailures

4758 failures <- *forwardFailures

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

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}

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}

4746− yield (respond {peers: peers, ok: True {}})

4768+ yield (respond {peers: union peers fwdOnlyPeers, ok: True {}})

4747 else yield (respond {peers: [], ok: False {}})

4769 else yield (respond {peers: [], ok: False {}})

4748

4770

4749handleStatsKey = \pubkey authorization respond -> do

4771handleStatsKey = \pubkey authorization respond -> do

@@ -5155,12 +5177,14 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do

5155-- POST /maintenance/messages/purge: Purge all pending messages for a specific pubkey.

5177-- POST /maintenance/messages/purge: Purge all pending messages for a specific pubkey.

5156-- Use for inbox-bombing abuse response: clears the recipient's mailbox without

5178-- Use for inbox-bombing abuse response: clears the recipient's mailbox without

5157-- requiring their session token. Also cancels pending forwards for purged messages.

5179-- 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.

5158handleAdminPurgeMessages = \pubkey authorization respond -> do

5182handleAdminPurgeMessages = \pubkey authorization respond -> do

5159 let authorized = verifyMaintenanceAuth authorization

5183 let authorized = verifyMaintenanceAuth authorization

5160 if not authorized then yield (respond {ok: False {}, purgedCount: 0})

5184 if not authorized then yield (respond {ok: False {}, purgedCount: 0})

5161 else do

5185 else do

5162 t <- now

5186 t <- now

5163− purgedCount <- atomic do

5187+ purgeResult <- atomic do

5164 let b = pubkeyBucket pubkey

5188 let b = pubkeyBucket pubkey

5165 bucket <- readMessageBucket b

5189 bucket <- readMessageBucket b

5166 let toDelete = do m <- bucket; where m.toKey == pubkey; yield m

5190 let toDelete = do m <- bucket; where m.toKey == pubkey; yield m

@@ -5176,11 +5200,27 @@ handleAdminPurgeMessages = \pubkey authorization respond -> do

5176 let pfb = pubkeyBucket pubkey

5200 let pfb = pubkeyBucket pubkey

5177 fwdBkt <- readFwdBucket pfb

5201 fwdBkt <- readFwdBucket pfb

5178 writeFwdBucket pfb (do f <- fwdBkt; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)

5202 writeFwdBucket pfb (do f <- fwdBkt; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)

5179− yield removed

5203+ -- Create ack entries for federated messages so origin servers learn about the purge

5180− else yield 0

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: []}

5181− when (purgedCount > 0)

5215+ when (purgeResult.purgedCount > 0)

5182− (logInfo ("admin: purged " ++ show purgedCount ++ " messages for " ++ pubkey))

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))

5183− yield (respond {ok: True {}, purgedCount: purgedCount})

5223+ yield (respond {ok: True {}, purgedCount: purgeResult.purgedCount})

5184

5224

5185-- GET /maintenance/rate-limits: Return current effective rate limits.

5225-- GET /maintenance/rate-limits: Return current effective rate limits.

5186handleGetRateLimits = \authorization respond -> do

5226handleGetRateLimits = \authorization respond -> do