@@ -251,8 +251,8 @@ type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex}
251-- /messages signals failure via the per-entry `error` field on each result:
251-- /messages signals failure via the per-entry `error` field on each result:
252-- empty = success (delivered, federated, queued, or silent duplicate),
252-- empty = success (delivered, federated, queued, or silent duplicate),
253-- non-empty = "rate_limited" | "self_send". A non-empty top-level `error`
253-- non-empty = "rate_limited" | "self_send". A non-empty top-level `error`
254−-- ("draining" | "rate_limited" | "unauthorized") means the request was
254+-- ("rate_limited" | "unauthorized") means the request was rejected before
255−-- rejected before any entry was processed and `results` is empty.
255+-- any entry was processed and `results` is empty.
256-- The sender's UI treats every success the same and learns "they got it"
256-- The sender's UI treats every success the same and learns "they got it"
257-- only via the recipient's E2E ack.
257-- only via the recipient's E2E ack.
@@ -323,7 +323,7 @@ type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
323-- Mutable refs (`*x`) holding the entire server state in memory. Every list
323-- Mutable refs (`*x`) holding the entire server state in memory. Every list
324-- scan is O(n); scaling is the first thing the production version replaces.
324-- scan is O(n); scaling is the first thing the production version replaces.
325-- Grouped here: auth (challenges/sessions), mailbox (messages/forwards/acks),
325-- 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.
328*challenges : [Challenge]
328*challenges : [Challenge]
329*sessions : [Session]
329*sessions : [Session]
@@ -356,9 +356,6 @@ type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
356*startedAt : Timestamp
356*startedAt : Timestamp
357*serverGeneration : Int
357*serverGeneration : Int
360−*drainCompletedAt : Timestamp
362*runtimeMaintenanceSecret : Text
359*runtimeMaintenanceSecret : Text
364*federationPrivateKeyHex : Text
361*federationPrivateKeyHex : Text
@@ -378,7 +375,7 @@ type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
378-- ClientApi -- /poll, send/batch/list/ack/delete, presence lookup
375-- ClientApi -- /poll, send/batch/list/ack/delete, presence lookup
379-- FederationApi -- inter-server gossip, forward, ack, delivery receipts
376-- FederationApi -- inter-server gossip, forward, ack, delivery receipts
380-- FederationPubkey -- exposes this server's federation Ed25519 public key
377-- 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)
382-- HealthApi/StatsApi/MetricsApi/ReadyApi -- observability
379-- HealthApi/StatsApi/MetricsApi/ReadyApi -- observability
383-- Response field shapes are kept stable for backwards compatibility even when
380-- Response field shapes are kept stable for backwards compatibility even when
384-- the underlying value is no longer computed (zero-filled).
381-- the underlying value is no longer computed (zero-filled).
@@ -390,7 +387,7 @@ route AuthApi where
390 POST /auth/logout headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool} = Logout
387 POST /auth/logout headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool} = Logout
392route ClientApi where
389route 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
394 POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, clientIp: Text} -> {results: [{messageId: MessageIdHex, error: Text}], error: Text, retryAfterMs: Int} = SendMessages
391 POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, clientIp: Text} -> {results: [{messageId: MessageIdHex, error: Text}], error: Text, retryAfterMs: Int} = SendMessages
395 POST {to: PubkeyHex, messageId: MessageIdHex} /messages/status headers {authorization: Text, clientIp: Text} -> {status: Text, authorized: Bool} = MessageStatus
392 POST {to: PubkeyHex, messageId: MessageIdHex} /messages/status headers {authorization: Text, clientIp: Text} -> {status: Text, authorized: Bool} = MessageStatus
396 POST {messageIds: [MessageIdHex]} /messages/ack headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, ackedCount: Int, error: Text} = AckMessages
393 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
410route MaintenanceApi where
407route MaintenanceApi where
411 POST /maintenance/prune headers {authorization: Text} -> {ok: Bool} = Prune
408 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
415 POST {server: ServerName} /maintenance/block headers {authorization: Text} -> {ok: Bool, blocked: Int} = BlockPeer
409 POST {server: ServerName} /maintenance/block headers {authorization: Text} -> {ok: Bool, blocked: Int} = BlockPeer
416 POST {server: ServerName} /maintenance/unblock headers {authorization: Text} -> {ok: Bool, unblocked: Bool} = UnblockPeer
410 POST {server: ServerName} /maintenance/unblock headers {authorization: Text} -> {ok: Bool, unblocked: Bool} = UnblockPeer
417 GET /maintenance/blocked headers {authorization: Text} -> {peers: [Text], ok: Bool} = ListBlocked
411 GET /maintenance/blocked headers {authorization: Text} -> {peers: [Text], ok: Bool} = ListBlocked
@@ -432,7 +426,7 @@ route MaintenanceApi where
432 GET /maintenance/config headers {authorization: Text} -> {ok: Bool, pollTimeoutMs: Int, ackGracePeriod: Int} = GetConfig
426 GET /maintenance/config headers {authorization: Text} -> {ok: Bool, pollTimeoutMs: Int, ackGracePeriod: Int} = GetConfig
434route HealthApi where
428route 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
437-- Per-peer health snapshot for /stats/peers. Latency/gossip-rate fields are
431-- Per-peer health snapshot for /stats/peers. Latency/gossip-rate fields are
438-- zero-filled in this simplified server.
432-- zero-filled in this simplified server.
@@ -455,7 +449,7 @@ type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
455type RecipientCount = {pubkey: PubkeyHex, count: Int}
449type RecipientCount = {pubkey: PubkeyHex, count: Int}
457route StatsApi where
451route 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
459 GET /stats/peers headers {authorization: Text} -> {peers: [PeerStatus], ok: Bool} = StatsPeers
453 GET /stats/peers headers {authorization: Text} -> {peers: [PeerStatus], ok: Bool} = StatsPeers
460 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
454 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
461 POST {limit: Int} /stats/top-recipients headers {authorization: Text} -> {recipients: [RecipientCount], ok: Bool, total: Int} = TopRecipients
455 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]
541capPerKey = \items pred maxN getKey -> union (removeWhere items pred) (capList (filter items pred) (maxN - 1) getKey)
535capPerKey = \items pred maxN getKey -> union (removeWhere items pred) (capList (filter items pred) (maxN - 1) getKey)
543-- Shared empty/error shape for /poll. Only a few fields ever vary.
537-- 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}
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}
545−emptyPollResp = \cursor authorized rateLimited draining gen retryAfterMs cursorWasReset -> {
539+emptyPollResp = \cursor authorized rateLimited gen retryAfterMs cursorWasReset -> {
546 events: [],
540 events: [],
548 authorized,
542 authorized,
549 rateLimited,
543 rateLimited,
551 generation: gen,
544 generation: gen,
552 heartbeat: False {},
545 heartbeat: False {},
553 retryAfterMs,
546 retryAfterMs,
@@ -650,7 +643,7 @@ isBadServerName = \sn -> sn == "" || sn == "localhost" || not all (chars sn) (\c
650-- ============================================
643-- ============================================
651-- Bearer token extraction, session lookup, maintenance secret check, and the
644-- Bearer token extraction, session lookup, maintenance secret check, and the
652-- `gateAuth`/`gateClientWrite`/`gateLookup` combinators that every client
645-- `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).
654-- Federation requests use a separate path (`withFederationAuth`) using
647-- Federation requests use a separate path (`withFederationAuth`) using
655-- Ed25519-signed auth verified via the peer's published public key.
648-- Ed25519-signed auth verified via the peer's published public key.
@@ -679,21 +672,18 @@ withMaintAuth = \authorization onUnauth onAuth -> do
679 authorized <- verifyMaintenanceAuth authorization
672 authorized <- verifyMaintenanceAuth authorization
680 if authorized then onAuth else onUnauth
673 if authorized then onAuth else onUnauth
682−-- Common gate for handlers: optional drain check, global rate, session auth.
675+-- Common gate for handlers: global rate, session auth.
683−-- mkErr is called with one of: "draining", "rate_limited", "unauthorized".
676+-- mkErr is called with one of: "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
677+gateAuth : Text -> Text -> Timestamp -> (resp -> ret) -> (Text -> resp) -> (PubkeyHex -> IO {} ret) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions} ret
685−gateAuth = \checkDrain auth ip t respond mkErr onAuth -> do
678+gateAuth = \auth ip t respond mkErr onAuth -> do
687 g <- checkGlobalRate t
679 g <- checkGlobalRate t
688 let err = \e -> yield (respond (mkErr e))
680 let err = \e -> yield (respond (mkErr e))
689− if checkDrain && dp >= 2
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
693-- gateAuth + per-key rate gate. `rateResp` is the response value when the per-key gate trips.
683-- 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
695gateClientWrite = \auth ip t respond mkErr rateResp onAuth ->
685gateClientWrite = \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
697 lim <- atomic do
687 lim <- atomic do
698 rates <- *clientWriteRates
688 rates <- *clientWriteRates
699 let r = checkRate rates pk maxClientWriteRatePerKey clientWriteRateWindowMs t
689 let r = checkRate rates pk maxClientWriteRatePerKey clientWriteRateWindowMs t
@@ -702,11 +692,10 @@ gateClientWrite = \auth ip t respond mkErr rateResp onAuth ->
702 yield r.limited
692 yield r.limited
703 if lim then yield (respond rateResp) else onAuth pk)
693 if lim then yield (respond rateResp) else onAuth pk)
705−-- Variant of `gateClientWrite` that skips the drain check (lookups still work
706−-- while draining) and uses the per-key lookup-rate window instead.
695+-- Variant of `gateClientWrite` that 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
696+gateLookup : Text -> Text -> Int<Ms> -> (a -> b) -> (Text -> a) -> a -> (Text -> IO {} b) -> IO {rw *globalRateCount, rw *globalRateWindowStart, r *sessions} b
708gateLookup = \auth ip t respond mkErr rateResp onAuth ->
697gateLookup = \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
710 lim <- atomic do
699 lim <- atomic do
711 rates <- *lookupRates
700 rates <- *lookupRates
712 let r = checkRate rates pk maxLookupRatePerKey lookupRateWindowMs t
701 let r = checkRate rates pk maxLookupRatePerKey lookupRateWindowMs t
@@ -804,7 +793,7 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
804-- entries live in a `RateEntry` list ref capped at 5000 with LRU eviction.
793-- entries live in a `RateEntry` list ref capped at 5000 with LRU eviction.
805-- * Global (`checkGlobalRate`/`checkAuthGlobalRate`) — single counter+windowStart
794-- * Global (`checkGlobalRate`/`checkAuthGlobalRate`) — single counter+windowStart
806-- pair, used for the request-wide and auth-wide ceilings.
795-- 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
808-- gate for unauthenticated handlers (challenge/verify).
797-- gate for unauthenticated handlers (challenge/verify).
810-- Returns {limited, entries}. Increments count when the key fits within the window.
799-- Returns {limited, entries}. Increments count when the key fits within the window.
@@ -865,16 +854,13 @@ checkServerRateGate = \server t -> atomic (do
865 *federationRates = r.entries
854 *federationRates = r.entries
866 yield r.limited)
855 yield r.limited)
868−-- Pre-auth gate: drain + global + ip + auth-global. Returns "" when ok, else an error code.
857+-- Pre-auth gate: 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
858+authPreCheck : Text -> Int<Ms> -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates} Text
870authPreCheck = \clientIp t -> do
859authPreCheck = \clientIp t -> do
872 g <- checkGlobalRate t
860 g <- checkGlobalRate t
873 i <- checkIpRateGate clientIp t
861 i <- checkIpRateGate clientIp t
874 a <- checkAuthGlobalRate t
862 a <- checkAuthGlobalRate t
875− yield (if dp >= 1
863+ yield (if g
878 then "global_rate_limited"
864 then "global_rate_limited"
879 else if i then "ip_rate_limited" else if a then "auth_rate_limited" else "")
865 else if i then "ip_rate_limited" else if a then "auth_rate_limited" else "")
@@ -1143,7 +1129,7 @@ refreshLocalPresence = \t -> do
1143-- POST /auth/challenge: issue a fresh signed-nonce challenge for `pubkey`.
1129-- POST /auth/challenge: issue a fresh signed-nonce challenge for `pubkey`.
1144-- Capped per-key (maxChallengesPerKey) and rate-limited; the IP is recorded
1130-- Capped per-key (maxChallengesPerKey) and rate-limited; the IP is recorded
1145-- so /auth/verify can require the same client.
1131-- 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
1147handleChallenge = \pubkey clientIp respond -> do
1133handleChallenge = \pubkey clientIp respond -> do
1148 t <- now
1134 t <- now
1149 preErr <- if requireClientIp && clientIp == ""
1135 preErr <- if requireClientIp && clientIp == ""
@@ -1181,7 +1167,7 @@ handleChallenge = \pubkey clientIp respond -> do
1181-- On success, mints a session token, records local presence (`pubkey` is now
1167-- On success, mints a session token, records local presence (`pubkey` is now
1182-- live on this server), and gossips an `online` event to known peers. If
1168-- live on this server), and gossips an `online` event to known peers. If
1183-- `revokeOthers` is set, all other sessions for the same pubkey are dropped.
1169-- `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
1185handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond -> do
1171handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond -> do
1186 t <- now
1172 t <- now
1187 preErr <- authPreCheck clientIp t
1173 preErr <- authPreCheck clientIp t
@@ -1192,14 +1178,13 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
1192 token <- genToken
1178 token <- genToken
1193 result <- atomic (do
1179 result <- atomic (do
1194 ch <- *challenges
1180 ch <- *challenges
1196 let same = \c -> c.pubkey == pubkey && c.challenge == challenge
1181 let same = \c -> c.pubkey == pubkey && c.challenge == challenge
1197 let sigValid = case bytesFromHex pubkey of
1182 let sigValid = case bytesFromHex pubkey of
1198 Just {value: pk} -> case bytesFromHex signature of
1183 Just {value: pk} -> case bytesFromHex signature of
1199 Just {value: sigBytes} -> verify pk (textToBytes challenge) sigBytes
1184 Just {value: sigBytes} -> verify pk (textToBytes challenge) sigBytes
1200 _ -> False {}
1185 _ -> False {}
1201 _ -> False {}
1186 _ -> 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
1203 if not valid
1188 if not valid
1204 then yield {valid: False {}, peers: []}
1189 then yield {valid: False {}, peers: []}
@@ -1226,10 +1211,10 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
1227-- POST /auth/revoke: caller (any session for the pubkey) revokes a sibling
1212-- POST /auth/revoke: caller (any session for the pubkey) revokes a sibling
1228-- session by token. Returns ok=False if no such sibling existed.
1213-- 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
1230handleRevoke = \targetToken authorization clientIp respond -> do
1215handleRevoke = \targetToken authorization clientIp respond -> do
1231 t <- now
1216 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
1233 result <- atomic (do
1218 result <- atomic (do
1234 ses <- *sessions
1219 ses <- *sessions
1235 let pred = \s -> s.pubkey == caller && s.token == targetToken
1220 let pred = \s -> s.pubkey == caller && s.token == targetToken
@@ -1276,27 +1261,24 @@ handleLogout = \authorization clientIp respond -> do
1276-- plus piggybacked sender-side expirations/deliveries.
1261-- plus piggybacked sender-side expirations/deliveries.
1277-- Per-pubkey concurrency is capped via `activePolls` to bound long-poll holds.
1262-- Per-pubkey concurrency is capped via `activePolls` to bound long-poll holds.
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
1280-- generation mismatch (server restart) to force a cursor reset, then hands
1265-- generation mismatch (server restart) to force a cursor reset, then hands
1281-- off to `pollOnce` for the actual long-poll body.
1266-- 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
1283handlePoll = \cursor generation authorization clientIp respond -> do
1268handlePoll = \cursor generation authorization clientIp respond -> do
1284 t <- now
1269 t <- now
1285 serverGen <- *serverGeneration
1270 serverGen <- *serverGeneration
1287 g <- checkGlobalRate t
1271 g <- checkGlobalRate t
1288 let cursorWasReset = generation != serverGen && cursor > 0
1272 let cursorWasReset = generation != serverGen && cursor > 0
1289− let er = \auth rate drain retryRate -> yield (respond (emptyPollResp 0 auth rate drain serverGen retryRate cursorWasReset))
1273+ let er = \auth rate retryRate -> yield (respond (emptyPollResp 0 auth rate serverGen retryRate cursorWasReset))
1290− if dp >= 2
1274+ if g
1291− then er (True {}) (False {}) (True {}) 0
1293− then er (False {}) (True {}) (False {}) globalRateWindowMs
1275+ then er (False {}) (True {}) globalRateWindowMs
1294 else pollOnce authorization (if cursorWasReset then 0 else cursor) cursorWasReset clientIp serverGen respond t
1276 else pollOnce authorization (if cursorWasReset then 0 else cursor) cursorWasReset clientIp serverGen respond t
1296-- Body of a single poll. Treats the cursor advance as an implicit ack
1278-- Body of a single poll. Treats the cursor advance as an implicit ack
1297-- (`ackPolledMessages`), acquires a per-pubkey poll slot, runs `doPoll` to
1279-- (`ackPolledMessages`), acquires a per-pubkey poll slot, runs `doPoll` to
1298-- collect events (or block until wakeup), then assembles the response.
1280-- 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
1300pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t -> do
1282pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t -> do
1301 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
1283 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
1302 let clampedCursor = clamp 0 t cursor
1284 let clampedCursor = clamp 0 t cursor
@@ -1304,7 +1286,7 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
1304 startWakeup <- *pollWakeup
1286 startWakeup <- *pollWakeup
1305 yield (Just {value: {pubkey, clampedCursor, startWakeup}}))
1287 yield (Just {value: {pubkey, clampedCursor, startWakeup}}))
1306 case setup of
1288 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))
1308 Just {value: s} -> do
1290 Just {value: s} -> do
1309 acquired <- atomic (do
1291 acquired <- atomic (do
1310 aps <- *activePolls
1292 aps <- *activePolls
@@ -1314,7 +1296,7 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
1314 *activePolls = union aps [{s.pubkey, startedAt: t}]
1296 *activePolls = union aps [{s.pubkey, startedAt: t}]
1315 yield (True {}))
1297 yield (True {}))
1316 if not acquired
1298 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))
1319 pollResult <- doPoll s t
1301 pollResult <- doPoll s t
1320 atomic (do aps <- *activePolls; *activePolls = filter aps (\p -> p.pubkey != s.pubkey))
1302 atomic (do aps <- *activePolls; *activePolls = filter aps (\p -> p.pubkey != s.pubkey))
@@ -1323,7 +1305,6 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
1323 pollResult.cursor,
1305 pollResult.cursor,
1324 authorized: True {},
1306 authorized: True {},
1325 rateLimited: False {},
1307 rateLimited: False {},
1326− draining: False {},
1327 generation: serverGen,
1308 generation: serverGen,
1328 heartbeat: count pollResult.events == 0,
1309 heartbeat: count pollResult.events == 0,
1329 retryAfterMs: 0,
1310 retryAfterMs: 0,
@@ -1371,22 +1352,22 @@ ackPolledMessages = \pubkey cursor -> do
1371-- Accept a message for a recipient pubkey. Placement (live local session,
1352-- Accept a message for a recipient pubkey. Placement (live local session,
1372-- federated forward, or held-for-presence) is internal; the response only
1353-- federated forward, or held-for-presence) is internal; the response only
1373-- says whether the message was accepted (`error: ""`) or refused (`error:
1354-- says whether the message was accepted (`error: ""`) or refused (`error:
1374−-- "draining" | "rate_limited" | "unauthorized" | "self_send"`). Duplicates
1355+-- "rate_limited" | "unauthorized" | "self_send"`). Duplicates are silently
1375−-- are silently treated as success so retries are idempotent.
1356+-- treated as success so retries are idempotent.
1376−-- Send is gated by drain phase, global rate, and the per-sender / per-pair /
1357+-- Send is gated by global rate and the per-sender / per-pair / per-recipient
1377−-- per-recipient rate limits charged together inside `sendMessageCore`.
1358+-- rate limits charged together inside `sendMessageCore`.
1378-- Federated forwards are kicked off in a background fork after the atomic.
1359-- Federated forwards are kicked off in a background fork after the atomic.
1380-- POST /messages: send a list of ciphertexts. Each entry runs through
1361-- POST /messages: send a list of ciphertexts. Each entry runs through
1381-- `sendMessageCore` separately so per-entry rate limits/errors don't abort
1362-- `sendMessageCore` separately so per-entry rate limits/errors don't abort
1382-- the rest; federated forwards are forked per entry. Top-level `error`
1363-- the rest; federated forwards are forked per entry. Top-level `error`
1383−-- carries global gate failures ("draining" | "rate_limited" | "unauthorized")
1364+-- carries global gate failures ("rate_limited" | "unauthorized") and
1384−-- and short-circuits with empty `results`.
1365+-- 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
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
1386handleSendMessages = \messages authorization clientIp respond -> do
1367handleSendMessages = \messages authorization clientIp respond -> do
1387 t <- now
1368 t <- now
1388 let mkErr = \e -> {results: [], error: e, retryAfterMs: 0}
1369 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
1390 results <- mapEach (takeRelation maxBatchSize messages) (\m -> do
1371 results <- mapEach (takeRelation maxBatchSize messages) (\m -> do
1391 let effMsgId = messageIdOf m.encryptedBlob
1372 let effMsgId = messageIdOf m.encryptedBlob
1392 r <- atomic (do sendMessageCore senderKey m.to m.encryptedBlob t)
1373 r <- atomic (do sendMessageCore senderKey m.to m.encryptedBlob t)
@@ -1494,7 +1475,7 @@ forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
1494-- GET /lookup/{pubkey}: return our local server name if the key is here,
1475-- GET /lookup/{pubkey}: return our local server name if the key is here,
1495-- "remote" if known on a peer, "" otherwise. Peer hostnames are deliberately
1476-- "remote" if known on a peer, "" otherwise. Peer hostnames are deliberately
1496-- masked (see section header).
1477-- 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
1498handleLookupPresence = \pubkey authorization clientIp respond -> do
1479handleLookupPresence = \pubkey authorization clientIp respond -> do
1499 t <- now
1480 t <- now
1500 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then e else ""}
1481 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then e else ""}
@@ -1508,7 +1489,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
1509-- POST /lookup/batch: same masking as `handleLookupPresence`, but for up to
1490-- POST /lookup/batch: same masking as `handleLookupPresence`, but for up to
1510-- 50 pubkeys in one round-trip.
1491-- 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
1512handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
1493handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
1513 t <- now
1494 t <- now
1514 let mkErr = \e -> {
1495 let mkErr = \e -> {
@@ -1528,11 +1509,11 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
1528-- POST /messages/status: classify a message as
1509-- POST /messages/status: classify a message as
1529-- pending (in mailbox), forwarding (en-route to a peer),
1510-- pending (in mailbox), forwarding (en-route to a peer),
1530-- delivered (acked or terminal), or unknown.
1511-- 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
1532handleMessageStatus = \to messageId authorization clientIp respond -> do
1513handleMessageStatus = \to messageId authorization clientIp respond -> do
1533 t <- now
1514 t <- now
1534 let mkErr = \_ -> {status: "", authorized: False {}}
1515 let mkErr = \_ -> {status: "", authorized: False {}}
1535− gateAuth (False {}) authorization clientIp t respond mkErr (\_ -> do
1516+ gateAuth authorization clientIp t respond mkErr (\_ -> do
1536 msgs <- *messages
1517 msgs <- *messages
1537 fwds <- *forwards
1518 fwds <- *forwards
1538 dids <- *deliveredIds
1519 dids <- *deliveredIds
@@ -1552,7 +1533,7 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
1553-- POST /messages/ack: explicit ack by message id (alternative to the implicit
1534-- POST /messages/ack: explicit ack by message id (alternative to the implicit
1554-- cursor-based ack in /poll). Drops mailbox+forward rows.
1535-- 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
1556handleAckMessages = \messageIds authorization clientIp respond -> do
1537handleAckMessages = \messageIds authorization clientIp respond -> do
1557 t <- now
1538 t <- now
1558 let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}
1539 let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}
@@ -1569,7 +1550,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
1570-- POST /messages/list: paginated peek at the caller's mailbox without acking.
1551-- POST /messages/list: paginated peek at the caller's mailbox without acking.
1571-- Returns metadata only (no ciphertext) plus the total count.
1552-- 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
1573handleListMessages = \offset authorization clientIp respond -> do
1554handleListMessages = \offset authorization clientIp respond -> do
1574 t <- now
1555 t <- now
1575 let mkErr = \e -> {
1556 let mkErr = \e -> {
@@ -1578,7 +1559,7 @@ handleListMessages = \offset authorization clientIp respond -> do
1578 count: 0,
1559 count: 0,
1579 error: if e == "rate_limited" then e else ""
1560 error: if e == "rate_limited" then e else ""
1581− gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do
1562+ gateAuth authorization clientIp t respond mkErr (\pubkey -> do
1582 lim <- atomic (do
1563 lim <- atomic (do
1583 rates <- *lookupRates
1564 rates <- *lookupRates
1584 let r = checkRate rates pubkey maxLookupRatePerKey lookupRateWindowMs t
1565 let r = checkRate rates pubkey maxLookupRatePerKey lookupRateWindowMs t
@@ -1598,14 +1579,14 @@ handleListMessages = \offset authorization clientIp respond -> do
1599-- POST /messages/delete: same effect as ack from the recipient's side, with
1580-- POST /messages/delete: same effect as ack from the recipient's side, with
1600-- a "delete" semantic for the response.
1581-- 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
1602handleDeleteMessages = \messageIds authorization clientIp respond -> do
1583handleDeleteMessages = \messageIds authorization clientIp respond -> do
1603 t <- now
1584 t <- now
1604 let mkErr = \e -> {
1585 let mkErr = \e -> {
1605 ok: False {},
1586 ok: False {},
1606 authorized: False {},
1587 authorized: False {},
1607 deletedCount: 0,
1588 deletedCount: 0,
1608− error: if e == "draining" then "" else e
1589+ error: e
1610 gateClientWrite authorization clientIp t respond mkErr {
1591 gateClientWrite authorization clientIp t respond mkErr {
1611 ok: False {},
1592 ok: False {},
@@ -1727,50 +1708,9 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp authorizati
1727-- Maintenance handlers
1708-- Maintenance handlers
1728-- ============================================
1709-- ============================================
1729-- Operator endpoints, all gated by `runtimeMaintenanceSecret` (Bearer header).
1710-- 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
1711+-- Admin endpoints can revoke sessions by pubkey or IP, purge a recipient's
1732−-- pubkey or IP, purge a recipient's mailbox, list aggregated session activity,
1712+-- mailbox, list aggregated session activity, block/allow peers, rotate the
1733−-- block/allow peers, rotate the maintenance secret, and read live config.
1713+-- maintenance secret, and read live config.
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 {
1744− let effectivePhase = if phase == 1 then 1 else 2
1746− drainResult <- atomic (do
1747− *drainPhase = effectivePhase
1748− *drainCompletedAt = t
1749− if effectivePhase == 1
1750− then yield {peers: [], localKeys: []}
1753− let isLocalLive = \p -> isLocalServer p.server && p.expiresAt > t
1754− *presence = removeWhere pres isLocalLive
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}))
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 {}}))
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 {
1775-- POST /maintenance/prune: trigger an out-of-band `pruneAndRetry` tick.
1715-- POST /maintenance/prune: trigger an out-of-band `pruneAndRetry` tick.
1776-- Useful when the operator wants to force expiry/forward retries without
1716-- Useful when the operator wants to force expiry/forward retries without
@@ -1792,18 +1732,17 @@ handleReconcile = \authorization respond -> do
1792-- GET /health: aggregate readiness flags. During an initial warmup window
1732-- GET /health: aggregate readiness flags. During an initial warmup window
1793-- (3 prune intervals after startup) prune/heartbeat checks always pass to
1733-- (3 prune intervals after startup) prune/heartbeat checks always pass to
1794-- avoid false negatives before the background loops have ticked.
1734-- 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
1796handleHealth = \authorization respond -> do
1736handleHealth = \authorization respond -> do
1797 t <- now
1737 t <- now
1798 lastPrune <- *lastPrunedAt
1738 lastPrune <- *lastPrunedAt
1799 lastHb <- *lastHeartbeatAt
1739 lastHb <- *lastHeartbeatAt
1800 sa <- *startedAt
1740 sa <- *startedAt
1802 let warmup = t - sa < backgroundPruneInterval * 3
1741 let warmup = t - sa < backgroundPruneInterval * 3
1803 let pruneOk = warmup || t - lastPrune < backgroundPruneInterval * 3
1742 let pruneOk = warmup || t - lastPrune < backgroundPruneInterval * 3
1804 let heartbeatOk = warmup || t - lastHb < pollTimeoutMs * 3
1743 let heartbeatOk = warmup || t - lastHb < pollTimeoutMs * 3
1805 yield (respond {
1744 yield (respond {
1806− ok: dp == 0 && pruneOk && heartbeatOk,
1745+ ok: pruneOk && heartbeatOk,
1807 version: serverBuildVersion,
1746 version: serverBuildVersion,
1808 pruneHealthy: pruneOk,
1747 pruneHealthy: pruneOk,
1809 heartbeatHealthy: heartbeatOk,
1748 heartbeatHealthy: heartbeatOk,
@@ -1814,7 +1753,6 @@ handleHealth = \authorization respond -> do
1814 nonceHealthy: True {},
1753 nonceHealthy: True {},
1815 forwardQueueHealthy: True {},
1754 forwardQueueHealthy: True {},
1816 requestPressureHealthy: True {},
1755 requestPressureHealthy: True {},
1818 degradedPeers: 0
1756 degradedPeers: 0
@@ -2053,7 +1991,7 @@ handleGetConfig = \authorization respond -> do
2053-- GET /stats: capacity gauges + minimal federation/timing snapshot. Most of
1991-- GET /stats: capacity gauges + minimal federation/timing snapshot. Most of
2054-- the historical metric fields are zero-filled (see `zeroStats`) — only the
1992-- the historical metric fields are zero-filled (see `zeroStats`) — only the
2055-- counts that come straight from the state refs are populated.
1993-- 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
2057handleStats = \authorization respond -> withMaintAuth authorization (yield (respond zeroStats)) (do
1995handleStats = \authorization respond -> withMaintAuth authorization (yield (respond zeroStats)) (do
2058 t <- now
1996 t <- now
2059 sa <- *startedAt
1997 sa <- *startedAt
@@ -2069,7 +2007,6 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
2069 hc <- *heartbeatCount
2007 hc <- *heartbeatCount
2070 lp <- *lastPrunedAt
2008 lp <- *lastPrunedAt
2071 lh <- *lastHeartbeatAt
2009 lh <- *lastHeartbeatAt
2073 let sCount = count sessions
2010 let sCount = count sessions
2074 let mCount = count msgs
2011 let mCount = count msgs
2075 let pCount = count pres
2012 let pCount = count pres
@@ -2111,7 +2048,6 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
2111 oldestAckAge: 0,
2048 oldestAckAge: 0,
2112 recentGossip: 0,
2049 recentGossip: 0,
2113 blockedPeers: count bp,
2050 blockedPeers: count bp,
2114− drainCompletedAt: 0,
2115 forwardServers: 0,
2051 forwardServers: 0,
2116 maxForwardQueueDepth: 0,
2052 maxForwardQueueDepth: 0,
2117 topFailingPeers: [],
2053 topFailingPeers: [],
@@ -2146,8 +2082,6 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
2146 uptime: t - sa,
2082 uptime: t - sa,
2147 lastPruneAt: lp,
2083 lastPruneAt: lp,
2148 lastHeartbeatAt: lh,
2084 lastHeartbeatAt: lh,
2151 totalEntries: sCount + mCount + pCount + fCount,
2085 totalEntries: sCount + mCount + pCount + fCount,
2152 activeRequests: 0,
2086 activeRequests: 0,
2153 peakActiveRequests: 0,
2087 peakActiveRequests: 0,
@@ -2168,7 +2102,7 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
2168-- All-zero /stats payload returned to unauthorized callers. The shape is kept
2102-- All-zero /stats payload returned to unauthorized callers. The shape is kept
2169-- in lockstep with the response in `handleStats` so older operator tooling
2103-- in lockstep with the response in `handleStats` so older operator tooling
2170-- never sees a missing field.
2104-- 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}
2172zeroStats = {
2106zeroStats = {
2173 ok: False {},
2107 ok: False {},
2174 version: serverBuildVersion,
2108 version: serverBuildVersion,
@@ -2206,7 +2140,6 @@ zeroStats = {
2206 oldestAckAge: 0,
2140 oldestAckAge: 0,
2207 recentGossip: 0,
2141 recentGossip: 0,
2208 blockedPeers: 0,
2142 blockedPeers: 0,
2209− drainCompletedAt: 0,
2210 forwardServers: 0,
2143 forwardServers: 0,
2211 maxForwardQueueDepth: 0,
2144 maxForwardQueueDepth: 0,
2212 topFailingPeers: [],
2145 topFailingPeers: [],
@@ -2241,8 +2174,6 @@ zeroStats = {
2241 uptime: 0,
2174 uptime: 0,
2242 lastPruneAt: 0,
2175 lastPruneAt: 0,
2243 lastHeartbeatAt: 0,
2176 lastHeartbeatAt: 0,
2244− draining: False {},
2246 totalEntries: 0,
2177 totalEntries: 0,
2247 activeRequests: 0,
2178 activeRequests: 0,
2248 peakActiveRequests: 0,
2179 peakActiveRequests: 0,
@@ -2375,7 +2306,7 @@ handleStatsRates = \authorization respond -> do
2375-- GET /metrics: Prometheus-style text exposition of the most useful gauges.
2306-- GET /metrics: Prometheus-style text exposition of the most useful gauges.
2376-- The `authorization` header is accepted but not validated here — operators
2307-- The `authorization` header is accepted but not validated here — operators
2377-- relying on auth must front this endpoint with a reverse proxy that enforces it.
2308-- 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
2379handleMetrics = \authorization respond -> do
2310handleMetrics = \authorization respond -> do
2380 t <- now
2311 t <- now
2381 msgs <- *messages
2312 msgs <- *messages
@@ -2384,7 +2315,6 @@ handleMetrics = \authorization respond -> do
2384 fwds <- *forwards
2315 fwds <- *forwards
2385 aps <- *activePolls
2316 aps <- *activePolls
2386 sa <- *startedAt
2317 sa <- *startedAt
2388 yield (respond {
2318 yield (respond {
2389 metrics: fold (\acc r -> acc ++ r.k ++ " " ++ show r.v ++ "\n") "" [
2319 metrics: fold (\acc r -> acc ++ r.k ++ " " ++ show r.v ++ "\n") "" [
2390 {k: "skrepka_messages_total", v: count msgs},
2320 {k: "skrepka_messages_total", v: count msgs},
@@ -2392,8 +2322,7 @@ handleMetrics = \authorization respond -> do
2392 {k: "skrepka_presence_total", v: count pres},
2322 {k: "skrepka_presence_total", v: count pres},
2393 {k: "skrepka_forwards_total", v: count fwds},
2323 {k: "skrepka_forwards_total", v: count fwds},
2394 {k: "skrepka_active_polls", v: count aps},
2324 {k: "skrepka_active_polls", v: count aps},
2395− {k: "skrepka_uptime_seconds", v: (t - sa) / 1000},
2325+ {k: "skrepka_uptime_seconds", v: (t - sa) / 1000}
2396− {k: "skrepka_draining", v: if dp > 0 then 1 else 0}
2398 ok: True {}
2327 ok: True {}
@@ -2425,9 +2354,6 @@ serve = \req -> case req of
2425 FederationBatchForward {forwards, fromServer, requestTimestamp, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp authorization respond
2354 FederationBatchForward {forwards, fromServer, requestTimestamp, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp authorization respond
2426 FederationPubkey {respond} -> handleFederationPubkey respond
2355 FederationPubkey {respond} -> handleFederationPubkey respond
2427 Prune {authorization, respond} -> handlePrune authorization respond
2356 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
2431 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
2357 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
2432 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
2358 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
2433 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
2359 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
@@ -2459,21 +2385,19 @@ serve = \req -> case req of
2459-- ============================================
2385-- ============================================
2460-- Background loops
2386-- Background loops
2461-- ============================================
2387-- ============================================
2462−-- `backgroundPrune` runs `pruneAndRetry` every `backgroundPruneInterval` ms
2388+-- `backgroundPrune` runs `pruneAndRetry` every `backgroundPruneInterval` ms.
2463−-- (skipped while drain phase >= 2). `pollHeartbeat` bumps `pollWakeup` every
2389+-- `pollHeartbeat` bumps `pollWakeup` every `pollTimeoutMs` so long-poll
2464−-- `pollTimeoutMs` so long-poll holders unblock and return an empty heartbeat
2390+-- holders unblock and return an empty heartbeat response, keeping connections
2465−-- response, keeping connections fresh.
2391+-- fresh.
2467−-- Endless tick: every `backgroundPruneInterval` run `pruneAndRetry` (skipped
2393+-- Endless tick: every `backgroundPruneInterval` run `pruneAndRetry` and bump
2468−-- when drain phase >= 2) and bump `pruneCount` for /stats and /health.
2394+-- `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
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
2470backgroundPrune = do
2396backgroundPrune = do
2471 sleep backgroundPruneInterval
2397 sleep backgroundPruneInterval
2474− t <- now
2398+ t <- now
2475− atomic (do pc <- *pruneCount; *pruneCount = pc + 1)
2399+ atomic (do pc <- *pruneCount; *pruneCount = pc + 1)
2476− pruneAndRetry t)
2400+ pruneAndRetry t
2477 backgroundPrune
2401 backgroundPrune
2479-- Endless tick that bumps `pollWakeup` every `pollTimeoutMs` so any blocked
2403-- Endless tick that bumps `pollWakeup` every `pollTimeoutMs` so any blocked