@@ -251,8 +251,8 @@ type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex}
251251-- /messages signals failure via the per-entry `error` field on each result:
252252-- empty = success (delivered, federated, queued, or silent duplicate),
253253-- non-empty = "rate_limited" | "self_send". A non-empty top-level `error`
254−-- ("draining" | "rate_limited" | "unauthorized") means the request was
255−-- rejected before any entry was processed and `results` is empty.
254+-- ("rate_limited" | "unauthorized") means the request was rejected before
255+-- any entry was processed and `results` is empty.
256256-- The sender's UI treats every success the same and learns "they got it"
257257-- only via the recipient's E2E ack.
258258
@@ -323,7 +323,7 @@ type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
323323-- Mutable refs (`*x`) holding the entire server state in memory. Every list
324324-- scan is O(n); scaling is the first thing the production version replaces.
325325-- Grouped here: auth (challenges/sessions), mailbox (messages/forwards/acks),
326−-- presence, rate-limit windows, drain phase, and bookkeeping counters.
326+-- presence, rate-limit windows, and bookkeeping counters.
327327
328328*challenges : [Challenge]
329329*sessions : [Session]
@@ -356,9 +356,6 @@ type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
356356*startedAt : Timestamp
357357*serverGeneration : Int
358358
359−*drainPhase : Int
360−*drainCompletedAt : Timestamp
361−
362359*runtimeMaintenanceSecret : Text
363360
364361*federationPrivateKeyHex : Text
@@ -378,7 +375,7 @@ type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
378375-- ClientApi -- /poll, send/batch/list/ack/delete, presence lookup
379376-- FederationApi -- inter-server gossip, forward, ack, delivery receipts
380377-- FederationPubkey -- exposes this server's federation Ed25519 public key
381−-- MaintenanceApi -- operator endpoints (drain, prune, peer allow/block, admin)
378+-- MaintenanceApi -- operator endpoints (prune, peer allow/block, admin)
382379-- HealthApi/StatsApi/MetricsApi/ReadyApi -- observability
383380-- Response field shapes are kept stable for backwards compatibility even when
384381-- the underlying value is no longer computed (zero-filled).
@@ -390,7 +387,7 @@ route AuthApi where
390387 POST /auth/logout headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool} = Logout
391388
392389route ClientApi where
393− POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int} = Poll
390+ POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int} = Poll
394391 POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, clientIp: Text} -> {results: [{messageId: MessageIdHex, error: Text}], error: Text, retryAfterMs: Int} = SendMessages
395392 POST {to: PubkeyHex, messageId: MessageIdHex} /messages/status headers {authorization: Text, clientIp: Text} -> {status: Text, authorized: Bool} = MessageStatus
396393 POST {messageIds: [MessageIdHex]} /messages/ack headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, ackedCount: Int, error: Text} = AckMessages
@@ -409,9 +406,6 @@ route FederationPubkeyApi where
409406
410407route MaintenanceApi where
411408 POST /maintenance/prune headers {authorization: Text} -> {ok: Bool} = Prune
412− POST {phase: Int} /maintenance/drain headers {authorization: Text} -> {ok: Bool, drained: Int, phase: Int} = Drain
413− POST /maintenance/undrain headers {authorization: Text} -> {ok: Bool} = Undrain
414− POST /maintenance/shutdown headers {authorization: Text} -> {ok: Bool, drained: Int} = Shutdown
415409 POST {server: ServerName} /maintenance/block headers {authorization: Text} -> {ok: Bool, blocked: Int} = BlockPeer
416410 POST {server: ServerName} /maintenance/unblock headers {authorization: Text} -> {ok: Bool, unblocked: Bool} = UnblockPeer
417411 GET /maintenance/blocked headers {authorization: Text} -> {peers: [Text], ok: Bool} = ListBlocked
@@ -432,7 +426,7 @@ route MaintenanceApi where
432426 GET /maintenance/config headers {authorization: Text} -> {ok: Bool, pollTimeoutMs: Int, ackGracePeriod: Int} = GetConfig
433427
434428route HealthApi where
435− GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, forwardQueueHealthy: Bool, requestPressureHealthy: Bool, draining: Bool, degradedPeers: Int} = Health
429+ GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, forwardQueueHealthy: Bool, requestPressureHealthy: Bool, degradedPeers: Int} = Health
436430
437431-- Per-peer health snapshot for /stats/peers. Latency/gossip-rate fields are
438432-- zero-filled in this simplified server.
@@ -455,7 +449,7 @@ type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
455449type RecipientCount = {pubkey: PubkeyHex, count: Int}
456450
457451route StatsApi where
458− GET /stats headers {authorization: Text} -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, totalMessageCount: Int, presence: Int, forwards: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, nonceEvictions: Int, activePolls: Int, recipientCounts: Int, messageBucketMax: Int, messageBucketMin: 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}], gossipSuccess: Int, gossipFail: Int, latencyBuckets: {lt100: Int, lt500: Int, lt2000: Int, gte2000: 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, 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, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, pollLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int, sendLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, authAvgMs: Int, authMaxMs: Int, authCount: Int}} = Stats
452+ GET /stats headers {authorization: Text} -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, totalMessageCount: Int, presence: Int, forwards: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, nonceEvictions: Int, activePolls: Int, recipientCounts: Int, messageBucketMax: Int, messageBucketMin: 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, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}], gossipSuccess: Int, gossipFail: Int, latencyBuckets: {lt100: Int, lt500: Int, lt2000: Int, gte2000: 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, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, pollLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int, sendLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, authAvgMs: Int, authMaxMs: Int, authCount: Int}} = Stats
459453 GET /stats/peers headers {authorization: Text} -> {peers: [PeerStatus], ok: Bool} = StatsPeers
460454 GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text} -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, forwardDetails: [{server: Text, retries: Int}], activePolls: Int, ok: Bool} = StatsKey
461455 POST {limit: Int} /stats/top-recipients headers {authorization: Text} -> {recipients: [RecipientCount], ok: Bool, total: Int} = TopRecipients
@@ -541,13 +535,12 @@ capPerKey : [a] -> (a -> Bool) -> Int -> (a -> b) -> [a]
541535capPerKey = \items pred maxN getKey -> union (removeWhere items pred) (capList (filter items pred) (maxN - 1) getKey)
542536
543537-- Shared empty/error shape for /poll. Only a few fields ever vary.
544−emptyPollResp : Timestamp -> Bool -> Bool -> Bool -> Int -> Int -> Bool -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int}
545−emptyPollResp = \cursor authorized rateLimited draining gen retryAfterMs cursorWasReset -> {
538+emptyPollResp : Timestamp -> Bool -> Bool -> Int -> Int -> Bool -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int}
539+emptyPollResp = \cursor authorized rateLimited gen retryAfterMs cursorWasReset -> {
546540 events: [],
547541 cursor,
548542 authorized,
549543 rateLimited,
550− draining,
551544 generation: gen,
552545 heartbeat: False {},
553546 retryAfterMs,
@@ -650,7 +643,7 @@ isBadServerName = \sn -> sn == "" || sn == "localhost" || not all (chars sn) (\c
650643-- ============================================
651644-- Bearer token extraction, session lookup, maintenance secret check, and the
652645-- `gateAuth`/`gateClientWrite`/`gateLookup` combinators that every client
653−-- handler funnels through (drain → global rate → per-key rate → session).
646+-- handler funnels through (global rate → per-key rate → session).
654647-- Federation requests use a separate path (`withFederationAuth`) using
655648-- Ed25519-signed auth verified via the peer's published public key.
656649
@@ -679,21 +672,18 @@ withMaintAuth = \authorization onUnauth onAuth -> do
679672 authorized <- verifyMaintenanceAuth authorization
680673 if authorized then onAuth else onUnauth
681674
682−-- Common gate for handlers: optional drain check, global rate, session auth.
683−-- mkErr is called with one of: "draining", "rate_limited", "unauthorized".
684−gateAuth : Bool -> Text -> Text -> Timestamp -> (resp -> ret) -> (Text -> resp) -> (PubkeyHex -> IO {} ret) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions} ret
685−gateAuth = \checkDrain auth ip t respond mkErr onAuth -> do
686− dp <- *drainPhase
675+-- Common gate for handlers: global rate, session auth.
676+-- mkErr is called with one of: "rate_limited", "unauthorized".
677+gateAuth : Text -> Text -> Timestamp -> (resp -> ret) -> (Text -> resp) -> (PubkeyHex -> IO {} ret) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions} ret
678+gateAuth = \auth ip t respond mkErr onAuth -> do
687679 g <- checkGlobalRate t
688680 let err = \e -> yield (respond (mkErr e))
689− if checkDrain && dp >= 2
690− then err "draining"
691− else if g then err "rate_limited" else withSessionAuth auth t ip (err "unauthorized") onAuth
681+ if g then err "rate_limited" else withSessionAuth auth t ip (err "unauthorized") onAuth
692682
693683-- gateAuth + per-key rate gate. `rateResp` is the response value when the per-key gate trips.
694−gateClientWrite : Text -> Text -> Int<Ms> -> (a -> b) -> (Text -> a) -> a -> (Text -> IO {} b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions} b
684+gateClientWrite : Text -> Text -> Int<Ms> -> (a -> b) -> (Text -> a) -> a -> (Text -> IO {} b) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions} b
695685gateClientWrite = \auth ip t respond mkErr rateResp onAuth ->
696− gateAuth (True {}) auth ip t respond mkErr (\pk -> do
686+ gateAuth auth ip t respond mkErr (\pk -> do
697687 lim <- atomic do
698688 rates <- *clientWriteRates
699689 let r = checkRate rates pk maxClientWriteRatePerKey clientWriteRateWindowMs t
@@ -702,11 +692,10 @@ gateClientWrite = \auth ip t respond mkErr rateResp onAuth ->
702692 yield r.limited
703693 if lim then yield (respond rateResp) else onAuth pk)
704694
705−-- Variant of `gateClientWrite` that skips the drain check (lookups still work
706−-- while draining) and uses the per-key lookup-rate window instead.
707−gateLookup : Text -> Text -> Int<Ms> -> (a -> b) -> (Text -> a) -> a -> (Text -> IO {} b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions} b
695+-- Variant of `gateClientWrite` that uses the per-key lookup-rate window instead.
696+gateLookup : Text -> Text -> Int<Ms> -> (a -> b) -> (Text -> a) -> a -> (Text -> IO {} b) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions} b
708697gateLookup = \auth ip t respond mkErr rateResp onAuth ->
709− gateAuth (False {}) auth ip t respond mkErr (\pk -> do
698+ gateAuth auth ip t respond mkErr (\pk -> do
710699 lim <- atomic do
711700 rates <- *lookupRates
712701 let r = checkRate rates pk maxLookupRatePerKey lookupRateWindowMs t
@@ -804,7 +793,7 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
804793-- entries live in a `RateEntry` list ref capped at 5000 with LRU eviction.
805794-- * Global (`checkGlobalRate`/`checkAuthGlobalRate`) — single counter+windowStart
806795-- pair, used for the request-wide and auth-wide ceilings.
807−-- `authPreCheck` composes drain + global + ip + auth-global into a one-shot
796+-- `authPreCheck` composes global + ip + auth-global into a one-shot
808797-- gate for unauthenticated handlers (challenge/verify).
809798
810799-- Returns {limited, entries}. Increments count when the key fits within the window.
@@ -865,16 +854,13 @@ checkServerRateGate = \server t -> atomic (do
865854 *federationRates = r.entries
866855 yield r.limited)
867856
868−-- Pre-auth gate: drain + global + ip + auth-global. Returns "" when ok, else an error code.
869−authPreCheck : Text -> Int<Ms> -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates} Text
857+-- Pre-auth gate: global + ip + auth-global. Returns "" when ok, else an error code.
858+authPreCheck : Text -> Int<Ms> -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates} Text
870859authPreCheck = \clientIp t -> do
871− dp <- *drainPhase
872860 g <- checkGlobalRate t
873861 i <- checkIpRateGate clientIp t
874862 a <- checkAuthGlobalRate t
875− yield (if dp >= 1
876− then "draining"
877− else if g
863+ yield (if g
878864 then "global_rate_limited"
879865 else if i then "ip_rate_limited" else if a then "auth_rate_limited" else "")
880866
@@ -1143,7 +1129,7 @@ refreshLocalPresence = \t -> do
11431129-- POST /auth/challenge: issue a fresh signed-nonce challenge for `pubkey`.
11441130-- Capped per-key (maxChallengesPerKey) and rate-limited; the IP is recorded
11451131-- so /auth/verify can require the same client.
1146−handleChallenge : PubkeyHex -> Text -> ({challenge: Text, error: Text, expiresAt: Int, ok: Bool, retryAfterMs: Int} -> a) -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *challengeRates, rw *challenges, r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates, clock, random} a
1132+handleChallenge : PubkeyHex -> Text -> ({challenge: Text, error: Text, expiresAt: Int, ok: Bool, retryAfterMs: Int} -> a) -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *challengeRates, rw *challenges, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates, clock, random} a
11471133handleChallenge = \pubkey clientIp respond -> do
11481134 t <- now
11491135 preErr <- if requireClientIp && clientIp == ""
@@ -1181,7 +1167,7 @@ handleChallenge = \pubkey clientIp respond -> do
11811167-- On success, mints a session token, records local presence (`pubkey` is now
11821168-- live on this server), and gossips an `online` event to known peers. If
11831169-- `revokeOthers` is set, all other sessions for the same pubkey are dropped.
1184−handleVerifyAuth : PubkeyHex -> Text -> Text -> Bool -> Text -> ({error: Text, expiresAt: Int, token: Text} -> a) -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *challenges, r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates, rw *presence, rw *sessions, clock, random} a
1170+handleVerifyAuth : PubkeyHex -> Text -> Text -> Bool -> Text -> ({error: Text, expiresAt: Int, token: Text} -> a) -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *challenges, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates, rw *presence, rw *sessions, clock, random} a
11851171handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond -> do
11861172 t <- now
11871173 preErr <- authPreCheck clientIp t
@@ -1192,14 +1178,13 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
11921178 token <- genToken
11931179 result <- atomic (do
11941180 ch <- *challenges
1195− dp <- *drainPhase
11961181 let same = \c -> c.pubkey == pubkey && c.challenge == challenge
11971182 let sigValid = case bytesFromHex pubkey of
11981183 Just {value: pk} -> case bytesFromHex signature of
11991184 Just {value: sigBytes} -> verify pk (textToBytes challenge) sigBytes
12001185 _ -> False {}
12011186 _ -> False {}
1202− let valid = dp < 1 && any ch (\c -> same c && c.expiresAt > t && matchIp clientIp c.ip) && sigValid
1187+ let valid = any ch (\c -> same c && c.expiresAt > t && matchIp clientIp c.ip) && sigValid
12031188 if not valid
12041189 then yield {valid: False {}, peers: []}
12051190 else do
@@ -1226,10 +1211,10 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
12261211
12271212-- POST /auth/revoke: caller (any session for the pubkey) revokes a sibling
12281213-- session by token. Returns ok=False if no such sibling existed.
1229−handleRevoke : Text -> Text -> Text -> ({authorized: Bool, ok: Bool} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1214+handleRevoke : Text -> Text -> Text -> ({authorized: Bool, ok: Bool} -> a) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
12301215handleRevoke = \targetToken authorization clientIp respond -> do
12311216 t <- now
1232− gateAuth (False {}) authorization clientIp t respond (\_ -> {ok: False {}, authorized: False {}}) (\caller -> do
1217+ gateAuth authorization clientIp t respond (\_ -> {ok: False {}, authorized: False {}}) (\caller -> do
12331218 result <- atomic (do
12341219 ses <- *sessions
12351220 let pred = \s -> s.pubkey == caller && s.token == targetToken
@@ -1276,27 +1261,24 @@ handleLogout = \authorization clientIp respond -> do
12761261-- plus piggybacked sender-side expirations/deliveries.
12771262-- Per-pubkey concurrency is capped via `activePolls` to bound long-poll holds.
12781263
1279−-- POST /poll: top-level dispatch. Validates drain/global gates, detects a
1264+-- POST /poll: top-level dispatch. Validates the global rate gate, detects a
12801265-- generation mismatch (server restart) to force a cursor reset, then hands
12811266-- off to `pollOnce` for the actual long-poll body.
1282−handlePoll : Int -> Int -> Text -> Text -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, draining: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int} -> Response) -> IO {r *drainPhase, r *pollWakeup, r *serverGeneration, r *sessions, rw *activePolls, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, clock} Response
1267+handlePoll : Int -> Int -> Text -> Text -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int} -> Response) -> IO {r *pollWakeup, r *serverGeneration, r *sessions, rw *activePolls, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, clock} Response
12831268handlePoll = \cursor generation authorization clientIp respond -> do
12841269 t <- now
12851270 serverGen <- *serverGeneration
1286− dp <- *drainPhase
12871271 g <- checkGlobalRate t
12881272 let cursorWasReset = generation != serverGen && cursor > 0
1289− let er = \auth rate drain retryRate -> yield (respond (emptyPollResp 0 auth rate drain serverGen retryRate cursorWasReset))
1290− if dp >= 2
1291− then er (True {}) (False {}) (True {}) 0
1292− else if g
1293− then er (False {}) (True {}) (False {}) globalRateWindowMs
1273+ let er = \auth rate retryRate -> yield (respond (emptyPollResp 0 auth rate serverGen retryRate cursorWasReset))
1274+ if g
1275+ then er (False {}) (True {}) globalRateWindowMs
12941276 else pollOnce authorization (if cursorWasReset then 0 else cursor) cursorWasReset clientIp serverGen respond t
12951277
12961278-- Body of a single poll. Treats the cursor advance as an implicit ack
12971279-- (`ackPolledMessages`), acquires a per-pubkey poll slot, runs `doPoll` to
12981280-- collect events (or block until wakeup), then assembles the response.
1299−pollOnce : Text -> Int -> Bool -> Text -> Int -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, draining: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int} -> Response) -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *activePolls, rw *forwards, rw *messages, clock} Response
1281+pollOnce : Text -> Int -> Bool -> Text -> Int -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int} -> Response) -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *activePolls, rw *forwards, rw *messages, clock} Response
13001282pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t -> do
13011283 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
13021284 let clampedCursor = clamp 0 t cursor
@@ -1304,7 +1286,7 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
13041286 startWakeup <- *pollWakeup
13051287 yield (Just {value: {pubkey, clampedCursor, startWakeup}}))
13061288 case setup of
1307− Nothing {} -> yield (respond (emptyPollResp 0 (False {}) (False {}) (False {}) serverGen 0 cursorWasReset))
1289+ Nothing {} -> yield (respond (emptyPollResp 0 (False {}) (False {}) serverGen 0 cursorWasReset))
13081290 Just {value: s} -> do
13091291 acquired <- atomic (do
13101292 aps <- *activePolls
@@ -1314,7 +1296,7 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
13141296 *activePolls = union aps [{s.pubkey, startedAt: t}]
13151297 yield (True {}))
13161298 if not acquired
1317− then yield (respond (emptyPollResp cursor (True {}) (True {}) (False {}) serverGen pollTimeoutMs cursorWasReset))
1299+ then yield (respond (emptyPollResp cursor (True {}) (True {}) serverGen pollTimeoutMs cursorWasReset))
13181300 else do
13191301 pollResult <- doPoll s t
13201302 atomic (do aps <- *activePolls; *activePolls = filter aps (\p -> p.pubkey != s.pubkey))
@@ -1323,7 +1305,6 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
13231305 pollResult.cursor,
13241306 authorized: True {},
13251307 rateLimited: False {},
1326− draining: False {},
13271308 generation: serverGen,
13281309 heartbeat: count pollResult.events == 0,
13291310 retryAfterMs: 0,
@@ -1371,22 +1352,22 @@ ackPolledMessages = \pubkey cursor -> do
13711352-- Accept a message for a recipient pubkey. Placement (live local session,
13721353-- federated forward, or held-for-presence) is internal; the response only
13731354-- says whether the message was accepted (`error: ""`) or refused (`error:
1374−-- "draining" | "rate_limited" | "unauthorized" | "self_send"`). Duplicates
1375−-- are silently treated as success so retries are idempotent.
1376−-- Send is gated by drain phase, global rate, and the per-sender / per-pair /
1377−-- per-recipient rate limits charged together inside `sendMessageCore`.
1355+-- "rate_limited" | "unauthorized" | "self_send"`). Duplicates are silently
1356+-- treated as success so retries are idempotent.
1357+-- Send is gated by global rate and the per-sender / per-pair / per-recipient
1358+-- rate limits charged together inside `sendMessageCore`.
13781359-- Federated forwards are kicked off in a background fork after the atomic.
13791360
13801361-- POST /messages: send a list of ciphertexts. Each entry runs through
13811362-- `sendMessageCore` separately so per-entry rate limits/errors don't abort
13821363-- the rest; federated forwards are forked per entry. Top-level `error`
1383−-- carries global gate failures ("draining" | "rate_limited" | "unauthorized")
1384−-- and short-circuits with empty `results`.
1385−handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> ({error: Text, results: [{error: Text, messageId: MessageIdHex}], retryAfterMs: Int} -> b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} b
1364+-- carries global gate failures ("rate_limited" | "unauthorized") and
1365+-- short-circuits with empty `results`.
1366+handleSendMessages : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> ({error: Text, results: [{error: Text, messageId: MessageIdHex}], retryAfterMs: Int} -> b) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} b
13861367handleSendMessages = \messages authorization clientIp respond -> do
13871368 t <- now
13881369 let mkErr = \e -> {results: [], error: e, retryAfterMs: 0}
1389− gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do
1370+ gateAuth authorization clientIp t respond mkErr (\senderKey -> do
13901371 results <- mapEach (takeRelation maxBatchSize messages) (\m -> do
13911372 let effMsgId = messageIdOf m.encryptedBlob
13921373 r <- atomic (do sendMessageCore senderKey m.to m.encryptedBlob t)
@@ -1494,7 +1475,7 @@ forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
14941475-- GET /lookup/{pubkey}: return our local server name if the key is here,
14951476-- "remote" if known on a peer, "" otherwise. Peer hostnames are deliberately
14961477-- masked (see section header).
1497−handleLookupPresence : PubkeyHex -> Text -> Text -> ({authorized: Bool, error: Text, server: Text} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1478+handleLookupPresence : PubkeyHex -> Text -> Text -> ({authorized: Bool, error: Text, server: Text} -> a) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
14981479handleLookupPresence = \pubkey authorization clientIp respond -> do
14991480 t <- now
15001481 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then e else ""}
@@ -1508,7 +1489,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
15081489
15091490-- POST /lookup/batch: same masking as `handleLookupPresence`, but for up to
15101491-- 50 pubkeys in one round-trip.
1511−handleBatchLookupPresence : [PubkeyHex] -> Text -> Text -> ({authorized: Bool, error: Text, results: [{pubkey: PubkeyHex, server: Text}]} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1492+handleBatchLookupPresence : [PubkeyHex] -> Text -> Text -> ({authorized: Bool, error: Text, results: [{pubkey: PubkeyHex, server: Text}]} -> a) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
15121493handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
15131494 t <- now
15141495 let mkErr = \e -> {
@@ -1528,11 +1509,11 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
15281509-- POST /messages/status: classify a message as
15291510-- pending (in mailbox), forwarding (en-route to a peer),
15301511-- delivered (acked or terminal), or unknown.
1531−handleMessageStatus : PubkeyHex -> MessageIdHex -> Text -> Text -> ({authorized: Bool, status: Text} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1512+handleMessageStatus : PubkeyHex -> MessageIdHex -> Text -> Text -> ({authorized: Bool, status: Text} -> a) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
15321513handleMessageStatus = \to messageId authorization clientIp respond -> do
15331514 t <- now
15341515 let mkErr = \_ -> {status: "", authorized: False {}}
1535− gateAuth (False {}) authorization clientIp t respond mkErr (\_ -> do
1516+ gateAuth authorization clientIp t respond mkErr (\_ -> do
15361517 msgs <- *messages
15371518 fwds <- *forwards
15381519 dids <- *deliveredIds
@@ -1552,7 +1533,7 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
15521533
15531534-- POST /messages/ack: explicit ack by message id (alternative to the implicit
15541535-- cursor-based ack in /poll). Drops mailbox+forward rows.
1555−handleAckMessages : [MessageIdHex] -> Text -> Text -> ({ackedCount: Int, authorized: Bool, error: Text, ok: Bool} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1536+handleAckMessages : [MessageIdHex] -> Text -> Text -> ({ackedCount: Int, authorized: Bool, error: Text, ok: Bool} -> a) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
15561537handleAckMessages = \messageIds authorization clientIp respond -> do
15571538 t <- now
15581539 let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}
@@ -1569,7 +1550,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
15691550
15701551-- POST /messages/list: paginated peek at the caller's mailbox without acking.
15711552-- Returns metadata only (no ciphertext) plus the total count.
1572−handleListMessages : Int -> Text -> Text -> ({authorized: Bool, count: Int, error: Text, messages: [{messageId: MessageIdHex, receivedAt: Int<Ms>}]} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1553+handleListMessages : Int -> Text -> Text -> ({authorized: Bool, count: Int, error: Text, messages: [{messageId: MessageIdHex, receivedAt: Int<Ms>}]} -> a) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
15731554handleListMessages = \offset authorization clientIp respond -> do
15741555 t <- now
15751556 let mkErr = \e -> {
@@ -1578,7 +1559,7 @@ handleListMessages = \offset authorization clientIp respond -> do
15781559 count: 0,
15791560 error: if e == "rate_limited" then e else ""
15801561 }
1581− gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do
1562+ gateAuth authorization clientIp t respond mkErr (\pubkey -> do
15821563 lim <- atomic (do
15831564 rates <- *lookupRates
15841565 let r = checkRate rates pubkey maxLookupRatePerKey lookupRateWindowMs t
@@ -1598,14 +1579,14 @@ handleListMessages = \offset authorization clientIp respond -> do
15981579
15991580-- POST /messages/delete: same effect as ack from the recipient's side, with
16001581-- a "delete" semantic for the response.
1601−handleDeleteMessages : [MessageIdHex] -> Text -> Text -> ({authorized: Bool, deletedCount: Int, error: Text, ok: Bool} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1582+handleDeleteMessages : [MessageIdHex] -> Text -> Text -> ({authorized: Bool, deletedCount: Int, error: Text, ok: Bool} -> a) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
16021583handleDeleteMessages = \messageIds authorization clientIp respond -> do
16031584 t <- now
16041585 let mkErr = \e -> {
16051586 ok: False {},
16061587 authorized: False {},
16071588 deletedCount: 0,
1608− error: if e == "draining" then "" else e
1589+ error: e
16091590 }
16101591 gateClientWrite authorization clientIp t respond mkErr {
16111592 ok: False {},
@@ -1727,50 +1708,9 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp authorizati
17271708-- Maintenance handlers
17281709-- ============================================
17291710-- Operator endpoints, all gated by `runtimeMaintenanceSecret` (Bearer header).
1730−-- Drain is two-phase: phase 1 stops accepting new auth, phase 2 also clears
1731−-- local presence and gossips offline. Admin endpoints can revoke sessions by
1732−-- pubkey or IP, purge a recipient's mailbox, list aggregated session activity,
1733−-- block/allow peers, rotate the maintenance secret, and read live config.
1734−
1735−-- POST /maintenance/drain: phase 1 stops accepting new auth (challenges,
1736−-- verify); phase 2 additionally clears local presence and broadcasts offline
1737−-- gossip for every locally-live key so peers can reroute. Idempotent.
1738−handleDrain : Int -> Text -> ({drained: Int, ok: Bool, phase: Int} -> a) -> IO {rw *drainCompletedAt, rw *drainPhase, rw *pollWakeup, rw *presence, r *runtimeMaintenanceSecret, clock} a
1739−handleDrain = \phase authorization respond -> withMaintAuth authorization (yield (respond {
1740− ok: False {},
1741− drained: 0,
1742− phase: 0
1743−})) (do
1744− let effectivePhase = if phase == 1 then 1 else 2
1745− t <- now
1746− drainResult <- atomic (do
1747− *drainPhase = effectivePhase
1748− *drainCompletedAt = t
1749− if effectivePhase == 1
1750− then yield {peers: [], localKeys: []}
1751− else do
1752− pres <- *presence
1753− let isLocalLive = \p -> isLocalServer p.server && p.expiresAt > t
1754− *presence = removeWhere pres isLocalLive
1755− *pollWakeup = t
1756− yield {peers: knownPeerServersFromPresence pres, localKeys: filter pres isLocalLive})
1757− when (effectivePhase == 2 && count drainResult.localKeys > 0 && federationEnabled) (fork (sendBatchedGossip drainResult.peers (pluck drainResult.localKeys (\p -> mkGossip (GossipOffline {}) p.pubkey))))
1758− yield (respond {ok: True {}, drained: count drainResult.localKeys, phase: effectivePhase}))
1759−
1760−-- POST /maintenance/undrain: clear drain state and resume normal operation.
1761−-- Does not restore presence — clients must re-authenticate to repopulate.
1762−handleUndrain : Text -> ({ok: Bool} -> a) -> IO {rw *drainCompletedAt, rw *drainPhase, r *runtimeMaintenanceSecret} a
1763−handleUndrain = \authorization respond -> withMaintAuth authorization (yield (respond {ok: False {}})) (do
1764− atomic (do *drainPhase = 0; *drainCompletedAt = 0)
1765− yield (respond {ok: True {}}))
1766−
1767−-- POST /maintenance/shutdown: convenience wrapper that performs a phase-2
1768−-- drain and reports the count of presence rows cleared.
1769−handleShutdown : Text -> ({drained: Int, ok: Bool} -> a) -> IO {rw *drainCompletedAt, rw *drainPhase, rw *pollWakeup, rw *presence, r *runtimeMaintenanceSecret, clock} a
1770−handleShutdown = \authorization respond -> handleDrain 2 authorization (\r -> respond {
1771− r.ok,
1772− r.drained
1773−})
1711+-- Admin endpoints can revoke sessions by pubkey or IP, purge a recipient's
1712+-- mailbox, list aggregated session activity, block/allow peers, rotate the
1713+-- maintenance secret, and read live config.
17741714
17751715-- POST /maintenance/prune: trigger an out-of-band `pruneAndRetry` tick.
17761716-- Useful when the operator wants to force expiry/forward retries without
@@ -1792,18 +1732,17 @@ handleReconcile = \authorization respond -> do
17921732-- GET /health: aggregate readiness flags. During an initial warmup window
17931733-- (3 prune intervals after startup) prune/heartbeat checks always pass to
17941734-- avoid false negatives before the background loops have ticked.
1795−handleHealth : a -> ({capacityHealthy: Bool, capacityWarning: Bool, degradedPeers: Int, draining: Bool, federationHealthy: Bool, forwardQueueHealthy: Bool, heartbeatHealthy: Bool, nonceHealthy: Bool, ok: Bool, pruneHealthy: Bool, requestPressureHealthy: Bool, version: Text, watchdogHealthy: Bool} -> b) -> IO {r *drainPhase, r *lastHeartbeatAt, r *lastPrunedAt, r *startedAt, clock} b
1735+handleHealth : a -> ({capacityHealthy: Bool, capacityWarning: Bool, degradedPeers: Int, federationHealthy: Bool, forwardQueueHealthy: Bool, heartbeatHealthy: Bool, nonceHealthy: Bool, ok: Bool, pruneHealthy: Bool, requestPressureHealthy: Bool, version: Text, watchdogHealthy: Bool} -> b) -> IO {r *lastHeartbeatAt, r *lastPrunedAt, r *startedAt, clock} b
17961736handleHealth = \authorization respond -> do
17971737 t <- now
17981738 lastPrune <- *lastPrunedAt
17991739 lastHb <- *lastHeartbeatAt
18001740 sa <- *startedAt
1801− dp <- *drainPhase
18021741 let warmup = t - sa < backgroundPruneInterval * 3
18031742 let pruneOk = warmup || t - lastPrune < backgroundPruneInterval * 3
18041743 let heartbeatOk = warmup || t - lastHb < pollTimeoutMs * 3
18051744 yield (respond {
1806− ok: dp == 0 && pruneOk && heartbeatOk,
1745+ ok: pruneOk && heartbeatOk,
18071746 version: serverBuildVersion,
18081747 pruneHealthy: pruneOk,
18091748 heartbeatHealthy: heartbeatOk,
@@ -1814,7 +1753,6 @@ handleHealth = \authorization respond -> do
18141753 nonceHealthy: True {},
18151754 forwardQueueHealthy: True {},
18161755 requestPressureHealthy: True {},
1817− draining: dp > 0,
18181756 degradedPeers: 0
18191757 })
18201758
@@ -2053,7 +1991,7 @@ handleGetConfig = \authorization respond -> do
20531991-- GET /stats: capacity gauges + minimal federation/timing snapshot. Most of
20541992-- the historical metric fields are zero-filled (see `zeroStats`) — only the
20551993-- counts that come straight from the state refs are populated.
2056−handleStats : Text -> ({capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text} -> a) -> IO {r *activePolls, r *blockedPeers, r *challenges, r *deliveredIds, r *drainPhase, r *forwards, r *heartbeatCount, r *lastHeartbeatAt, r *lastPrunedAt, r *messages, r *presence, r *pruneCount, r *runtimeMaintenanceSecret, r *sessions, r *startedAt, clock} a
1994+handleStats : Text -> ({capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text} -> a) -> IO {r *activePolls, r *blockedPeers, r *challenges, r *deliveredIds, r *forwards, r *heartbeatCount, r *lastHeartbeatAt, r *lastPrunedAt, r *messages, r *presence, r *pruneCount, r *runtimeMaintenanceSecret, r *sessions, r *startedAt, clock} a
20571995handleStats = \authorization respond -> withMaintAuth authorization (yield (respond zeroStats)) (do
20581996 t <- now
20591997 sa <- *startedAt
@@ -2069,7 +2007,6 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
20692007 hc <- *heartbeatCount
20702008 lp <- *lastPrunedAt
20712009 lh <- *lastHeartbeatAt
2072− dp <- *drainPhase
20732010 let sCount = count sessions
20742011 let mCount = count msgs
20752012 let pCount = count pres
@@ -2111,7 +2048,6 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
21112048 oldestAckAge: 0,
21122049 recentGossip: 0,
21132050 blockedPeers: count bp,
2114− drainCompletedAt: 0,
21152051 forwardServers: 0,
21162052 maxForwardQueueDepth: 0,
21172053 topFailingPeers: [],
@@ -2146,8 +2082,6 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
21462082 uptime: t - sa,
21472083 lastPruneAt: lp,
21482084 lastHeartbeatAt: lh,
2149− draining: dp > 0,
2150− drainPhase: dp,
21512085 totalEntries: sCount + mCount + pCount + fCount,
21522086 activeRequests: 0,
21532087 peakActiveRequests: 0,
@@ -2168,7 +2102,7 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
21682102-- All-zero /stats payload returned to unauthorized callers. The shape is kept
21692103-- in lockstep with the response in `handleStats` so older operator tooling
21702104-- never sees a missing field.
2171−zeroStats : {capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text}
2105+zeroStats : {capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text}
21722106zeroStats = {
21732107 ok: False {},
21742108 version: serverBuildVersion,
@@ -2206,7 +2140,6 @@ zeroStats = {
22062140 oldestAckAge: 0,
22072141 recentGossip: 0,
22082142 blockedPeers: 0,
2209− drainCompletedAt: 0,
22102143 forwardServers: 0,
22112144 maxForwardQueueDepth: 0,
22122145 topFailingPeers: [],
@@ -2241,8 +2174,6 @@ zeroStats = {
22412174 uptime: 0,
22422175 lastPruneAt: 0,
22432176 lastHeartbeatAt: 0,
2244− draining: False {},
2245− drainPhase: 0,
22462177 totalEntries: 0,
22472178 activeRequests: 0,
22482179 peakActiveRequests: 0,
@@ -2375,7 +2306,7 @@ handleStatsRates = \authorization respond -> do
23752306-- GET /metrics: Prometheus-style text exposition of the most useful gauges.
23762307-- The `authorization` header is accepted but not validated here — operators
23772308-- relying on auth must front this endpoint with a reverse proxy that enforces it.
2378−handleMetrics : a -> ({metrics: Text, ok: Bool} -> b) -> IO {r *activePolls, r *drainPhase, r *forwards, r *messages, r *presence, r *sessions, r *startedAt, clock} b
2309+handleMetrics : a -> ({metrics: Text, ok: Bool} -> b) -> IO {r *activePolls, r *forwards, r *messages, r *presence, r *sessions, r *startedAt, clock} b
23792310handleMetrics = \authorization respond -> do
23802311 t <- now
23812312 msgs <- *messages
@@ -2384,7 +2315,6 @@ handleMetrics = \authorization respond -> do
23842315 fwds <- *forwards
23852316 aps <- *activePolls
23862317 sa <- *startedAt
2387− dp <- *drainPhase
23882318 yield (respond {
23892319 metrics: fold (\acc r -> acc ++ r.k ++ " " ++ show r.v ++ "\n") "" [
23902320 {k: "skrepka_messages_total", v: count msgs},
@@ -2392,8 +2322,7 @@ handleMetrics = \authorization respond -> do
23922322 {k: "skrepka_presence_total", v: count pres},
23932323 {k: "skrepka_forwards_total", v: count fwds},
23942324 {k: "skrepka_active_polls", v: count aps},
2395− {k: "skrepka_uptime_seconds", v: (t - sa) / 1000},
2396− {k: "skrepka_draining", v: if dp > 0 then 1 else 0}
2325+ {k: "skrepka_uptime_seconds", v: (t - sa) / 1000}
23972326 ],
23982327 ok: True {}
23992328 })
@@ -2425,9 +2354,6 @@ serve = \req -> case req of
24252354 FederationBatchForward {forwards, fromServer, requestTimestamp, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp authorization respond
24262355 FederationPubkey {respond} -> handleFederationPubkey respond
24272356 Prune {authorization, respond} -> handlePrune authorization respond
2428− Drain {phase, authorization, respond} -> handleDrain phase authorization respond
2429− Shutdown {authorization, respond} -> handleShutdown authorization respond
2430− Undrain {authorization, respond} -> handleUndrain authorization respond
24312357 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
24322358 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
24332359 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
@@ -2459,21 +2385,19 @@ serve = \req -> case req of
24592385-- ============================================
24602386-- Background loops
24612387-- ============================================
2462−-- `backgroundPrune` runs `pruneAndRetry` every `backgroundPruneInterval` ms
2463−-- (skipped while drain phase >= 2). `pollHeartbeat` bumps `pollWakeup` every
2464−-- `pollTimeoutMs` so long-poll holders unblock and return an empty heartbeat
2465−-- response, keeping connections fresh.
2466−
2467−-- Endless tick: every `backgroundPruneInterval` run `pruneAndRetry` (skipped
2468−-- when drain phase >= 2) and bump `pruneCount` for /stats and /health.
2469−backgroundPrune : IO {r *drainPhase, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, r *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *pruneCount, rw *sessions, clock} a
2388+-- `backgroundPrune` runs `pruneAndRetry` every `backgroundPruneInterval` ms.
2389+-- `pollHeartbeat` bumps `pollWakeup` every `pollTimeoutMs` so long-poll
2390+-- holders unblock and return an empty heartbeat response, keeping connections
2391+-- fresh.
2392+
2393+-- Endless tick: every `backgroundPruneInterval` run `pruneAndRetry` and bump
2394+-- `pruneCount` for /stats and /health.
2395+backgroundPrune : IO {rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, r *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *pruneCount, rw *sessions, clock} a
24702396backgroundPrune = do
24712397 sleep backgroundPruneInterval
2472− dp <- *drainPhase
2473− when (dp < 2) (do
2474− t <- now
2475− atomic (do pc <- *pruneCount; *pruneCount = pc + 1)
2476− pruneAndRetry t)
2398+ t <- now
2399+ atomic (do pc <- *pruneCount; *pruneCount = pc + 1)
2400+ pruneAndRetry t
24772401 backgroundPrune
24782402
24792403-- Endless tick that bumps `pollWakeup` every `pollTimeoutMs` so any blocked