@@ -7,82 +7,125 @@
77-- ============================================
88-- Configuration
99-- ============================================
10+-- Compile-time constants: bind address, federation identity, capacity ceilings,
11+-- per-key/window rate limits, expiry/backoff/TTL durations, and protocol version.
12+-- All values are plain literals; nothing is read from disk or env at runtime.
1013
14+serverHost : Text
1115serverHost = "localhost"
16+listenPort : Int
1217listenPort = 8080
18+federationScheme : Text
1319federationScheme = "https://"
20+serverFederationPrivateKeyHex : Text
1421serverFederationPrivateKeyHex = ""
22+serverFederationPublicKeyHex : Text
1523serverFederationPublicKeyHex = ""
1624
1725messageMaxAge : Int<Ms>
1826messageMaxAge = 2592000000
19−maxPresenceTtl : Int<S>
20−maxPresenceTtl = 3600
27+maxPresenceTtl : Int<Ms>
28+maxPresenceTtl = 3600000
29+maxPubkeyLen : Int
2130maxPubkeyLen = 256
31+maxBlobLen : Int
2232maxBlobLen = 65536
33+maxSignatureLen : Int
2334maxSignatureLen = 256
35+maxMessageIds : Int
2436maxMessageIds = 100
37+maxBatchSize : Int
2538maxBatchSize = 100
39+maxMessagePriority : Int
2640maxMessagePriority = 10
41+maxGossipEvents : Int
2742maxGossipEvents = 100
43+maxPollEvents : Int
2844maxPollEvents = 50
45+maxPresenceEntries : Int
2946maxPresenceEntries = 10000
47+maxMessagesPerRecipient : Int
3048maxMessagesPerRecipient = 500
49+maxMessages : Int
3150maxMessages = 100000
3251clockSkewTolerance : Int<Ms>
3352clockSkewTolerance = 5000
53+maxForwardsEntries : Int
3454maxForwardsEntries = 10000
55+maxForwardsPerServer : Int
3556maxForwardsPerServer = 2000
57+maxServerNameLen : Int
3658maxServerNameLen = 128
59+maxMessageIdLen : Int
3760maxMessageIdLen = 128
61+maxTokenLen : Int
3862maxTokenLen = 256
63+maxChallenges : Int
3964maxChallenges = 10000
65+maxSessions : Int
4066maxSessions = 10000
67+maxAckedIds : Int
4168maxAckedIds = 10000
69+maxDeliveredIds : Int
4270maxDeliveredIds = 10000
71+maxChallengesPerKey : Int
4372maxChallengesPerKey = 5
73+maxSessionsPerKey : Int
4474maxSessionsPerKey = 10
4575challengeExpiry : Int<Ms>
4676challengeExpiry = 60000
77+maxChallengeRatePerKey : Int
4778maxChallengeRatePerKey = 5
4879challengeRateWindowMs : Int<Ms>
4980challengeRateWindowMs = 1000
81+maxSendRatePerKey : Int
5082maxSendRatePerKey = 20
5183sendRateWindowMs : Int<Ms>
5284sendRateWindowMs = 1000
85+maxRecvRatePerKey : Int
5386maxRecvRatePerKey = 10
5487recvRateWindowMs : Int<Ms>
5588recvRateWindowMs = 1000
89+maxPairRatePerPair : Int
5690maxPairRatePerPair = 30
5791pairRateWindowMs : Int<Ms>
5892pairRateWindowMs = 60000
93+maxFederationRatePerServer : Int
5994maxFederationRatePerServer = 50
6095federationRateWindowMs : Int<Ms>
6196federationRateWindowMs = 1000
97+maxLookupRatePerKey : Int
6298maxLookupRatePerKey = 5
6399lookupRateWindowMs : Int<Ms>
64100lookupRateWindowMs = 1000
101+maxClientWriteRatePerKey : Int
65102maxClientWriteRatePerKey = 10
66103clientWriteRateWindowMs : Int<Ms>
67104clientWriteRateWindowMs = 1000
105+maxIpRatePerIp : Int
68106maxIpRatePerIp = 10
69107ipRateWindowMs : Int<Ms>
70108ipRateWindowMs = 1000
109+maxGlobalRequestRate : Int
71110maxGlobalRequestRate = 1000
72111globalRateWindowMs : Int<Ms>
73112globalRateWindowMs = 1000
113+maxAuthGlobalRate : Int
74114maxAuthGlobalRate = 200
75115authGlobalRateWindowMs : Int<Ms>
76116authGlobalRateWindowMs = 1000
77117
78118sessionExpiry : Int<Ms>
79119sessionExpiry = 3600000
80−onlineGossipTtl : Int<S>
81−onlineGossipTtl = 5400
82−offlineGossipMinTtl : Int<S>
83−offlineGossipMinTtl = 60
120+onlineGossipTtl : Int<Ms>
121+onlineGossipTtl = 5400000
122+offlineGossipMinTtl : Int<Ms>
123+offlineGossipMinTtl = 60000
124+maxForwardRetries : Int
84125maxForwardRetries = 10
126+maxGossipHops : Int
85127maxGossipHops = 5
128+gossipOriginHops : Int
86129gossipOriginHops = 2
87130
88131retryBackoffBase : Int<Ms>
@@ -92,14 +135,18 @@ retryBackoffMax = 480000
92135
93136verificationTokenExpiry : Int<Ms>
94137verificationTokenExpiry = 30000
138+maxPendingVerifications : Int
95139maxPendingVerifications = 10000
96140
97141backgroundPruneInterval : Int<Ms>
98142backgroundPruneInterval = 30000
99143pollTimeoutMs : Int<Ms>
100144pollTimeoutMs = 25000
145+maxPollsPerKey : Int
101146maxPollsPerKey = 3
147+maxActivePolls : Int
102148maxActivePolls = 5000
149+maxWatchKeys : Int
103150maxWatchKeys = 50
104151presenceRefreshInterval : Int<Ms>
105152presenceRefreshInterval = 300000
@@ -111,16 +158,31 @@ minMessageTtl = 60000
111158ackGracePeriod : Int<Ms>
112159ackGracePeriod = 120000
113160
161+federationProtocolVersion : Int
114162federationProtocolVersion = 15
163+serverBuildVersion : Text
115164serverBuildVersion = "2026-04-27-simple"
116165
166+federationEnabled : Bool
117167federationEnabled = True {}
168+federationUrl : Text -> Text
118169federationUrl = \server -> federationScheme ++ server
170+requireClientIp : Bool
119171requireClientIp = True {}
120172
173+msToS : Int<Ms> -> Int<S>
174+msToS = \ms -> stripUnit ms / 1000
175+
176+sToMs : Int<S> -> Int<Ms>
177+sToMs = \s -> stripUnit s * 1000
178+
121179-- ============================================
122180-- Types
123181-- ============================================
182+-- Wire/state shapes plus refinement aliases for hex/server-name validation.
183+-- Hex types are bounded length + lowercase-hex; server names are lowercased
184+-- and stripped of trailing dots, with a deny-list filter (`isBadServerName`)
185+-- applied where SSRF is a concern.
124186
125187type PollEvent = {id: MessageIdHex, eventType: Text, payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex, expiresAt: Timestamp}}
126188
@@ -139,7 +201,7 @@ impl FromJSON GossipEventType where
139201 else if s == "offline" then GossipOffline {}
140202 else GossipUnknown {}
141203
142−type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex, server: ServerName, timestamp: Timestamp, ttl: Int<S>, seenBy: [Text]}
204+type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex, server: ServerName, timestamp: Timestamp, ttl: Int<Ms>, seenBy: [Text]}
143205
144206data MessageStatus = MsDelivered {} | MsFederated {} | MsQueued {} | MsUnauthorized {} | MsRejected {} | MsRateLimited {} | MsDuplicate {} | MsDraining {}
145207
@@ -197,6 +259,10 @@ type BatchCancelResult = {messageId: MessageIdHex, cancelled: Bool}
197259-- ============================================
198260-- Persisted State (all flat lists, no sharding)
199261-- ============================================
262+-- Mutable refs (`*x`) holding the entire server state in memory. Every list
263+-- scan is O(n); scaling is the first thing the production version replaces.
264+-- Grouped here: auth (challenges/sessions), mailbox (messages/forwards/acks),
265+-- presence, rate-limit windows, drain phase, and bookkeeping counters.
200266
201267*challenges : [Challenge]
202268*sessions : [Session]
@@ -251,6 +317,16 @@ type BatchCancelResult = {messageId: MessageIdHex, cancelled: Bool}
251317-- ============================================
252318-- Routes (response shapes preserved; removed metric fields are zero-filled)
253319-- ============================================
320+-- Public HTTP surface, grouped by audience:
321+-- AuthApi -- challenge/verify/revoke/logout (client identity)
322+-- ClientApi -- /poll, send/batch/list/ack/delete/cancel, presence lookup
323+-- FederationApi -- inter-server gossip, forward, ack, cancel, delivery receipts
324+-- FederationVerify -- callback used by peers to verify Bearer tokens this server issued
325+-- FederationPubkey -- exposes this server's federation Ed25519 public key
326+-- MaintenanceApi -- operator endpoints (drain, prune, peer allow/block, admin)
327+-- HealthApi/StatsApi/MetricsApi/ReadyApi -- observability
328+-- Response field shapes are kept stable for backwards compatibility even when
329+-- the underlying value is no longer computed (zero-filled).
254330
255331route AuthApi where
256332 /auth
@@ -439,11 +515,20 @@ route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | Federati
439515-- ============================================
440516-- Generic helpers
441517-- ============================================
518+-- List/predicate combinators (filter/any/none/findFirst/...), token generation,
519+-- and small text/server-name utilities used throughout the rest of the file.
520+-- Everything is intentionally O(n) over flat lists; performance is sacrificed
521+-- for clarity per the file header.
442522
523+filter : (a b) -> (b -> Bool) -> (a b)
443524filter = \xs pred -> do x <- xs; where pred x; yield x
525+removeWhere : (a b) -> (b -> Bool) -> (a b)
444526removeWhere = \xs pred -> filter xs (\x -> not (pred x))
527+pluck : (a b) -> (b -> c) -> (a c)
445528pluck = \xs f -> do x <- xs; yield (f x)
529+mapWhere : (a b) -> (b -> Bool) -> (b -> c) -> (a c)
446530mapWhere = \xs pred -> pluck (filter xs pred)
531+countWhere : (a b) -> (b -> Bool) -> Int
447532countWhere = \xs pred -> count (filter xs pred)
448533any : [a] -> (a -> Bool) -> Bool
449534any = \xs pred -> countWhere xs pred > 0
@@ -451,58 +536,42 @@ none : [a] -> (a -> Bool) -> Bool
451536none = \xs pred -> not (any xs pred)
452537all : [a] -> (a -> Bool) -> Bool
453538all = \xs pred -> none xs (\x -> not (pred x))
539+elem : [a] -> a -> Bool
454540elem = \xs x -> any xs (\y -> y == x)
541+findFirst : [a] -> (a -> Bool) -> Maybe a
455542findFirst = \xs pred -> headOf (filter xs pred)
543+when : Bool -> IO {} {} -> IO {} {}
456544when = \cond action -> if cond then action else yield {}
545+min : a -> a -> a
457546min = \a b -> if a < b then a else b
547+max : a -> a -> a
458548max = \a b -> if a > b then a else b
549+clamp : a -> a -> a -> a
459550clamp = \lo hi x -> max lo (min hi x)
551+abs : Int -> Int
460552abs = \x -> if x < 0 then 0 - x else x
461553
462554-- Dedup: keeps first occurrence (O(n^2)).
555+unique : [a] -> [a]
463556unique = fold (\acc x -> if elem acc x then acc else union acc [x]) []
464557
465558-- Pair each element with its 0-based index.
559+withIndex : [a] -> [{i: Int, v: a}]
466560withIndex = fold (\acc x -> union acc [{i: count acc, v: x}]) []
467561
468−-- Replace a ref's contents with the entries that satisfy `keep`. Caller controls atomicity.
469−pruneRef = \ref keep -> do
470− cur <- *ref
471− set *ref = filter cur keep
472−
473−-- Append `items` to a ref, then cap by `getKey` (drops oldest on overflow).
474−appendCapped = \ref items maxN getKey -> do
475− cur <- *ref
476− set *ref = capList (union cur items) maxN getKey
477−
478−-- Add `server` to the named list ref if absent. Returns the new total.
479−addServerToList = \listRef server -> atomic do
480− current <- *listRef
481− let updated = unique (union current [server])
482− set *listRef = updated
483− yield (count updated)
484−
485−-- Remove `server` from the named list ref. Returns whether it was present.
486−removeServerFromList = \listRef server -> atomic do
487− current <- *listRef
488− set *listRef = removeWhere current (\x -> x == server)
489− yield (elem current server)
490−
491−headOf = \items -> fold (\_ x -> Just {value: x}) Nothing {} (take 1 items)
562+headOf : [a] -> Maybe a
563+headOf = \items -> fold (\_ x -> Just {value: x}) Nothing {} (takeRelation 1 items)
492564
493565-- Cap to maxN by dropping the smallest-key entries (LRU-style). Always sorts.
494−capList = \items maxN getKey -> drop (max 0 (count items - maxN)) (sortBy getKey items)
566+capList : [a] -> Int -> (a -> b) -> [a]
567+capList = \items maxN getKey -> dropRelation (max 0 (count items - maxN)) (sortBy getKey items)
495568
496569-- Integer percentage with zero-denominator guard.
570+pct : Int -> Int -> Int
497571pct = \cur maxN -> if maxN > 0 then cur * 100 / maxN else 0
498572
499−-- Read entries matching `pred` (transformed by `extract`) and prune them in one step.
500−collectAndPrune = \ref pred extract -> do
501− items <- *ref
502− set *ref = removeWhere items pred
503− yield (mapWhere items pred extract)
504−
505573-- Common ack/delete tail: pull matching messages, drop them and their forwards.
574+removeForKey : (forall a. {toKey: PubkeyHex, messageId: MessageIdHex | a} -> Bool) -> IO {} [Message]
506575removeForKey = \pred -> do
507576 msgs <- *messages
508577 fwds <- *forwards
@@ -511,17 +580,21 @@ removeForKey = \pred -> do
511580 yield (filter msgs pred)
512581
513582-- Cap pred-matching entries to maxN-1 (caller adds one more).
583+capPerKey : [a] -> (a -> Bool) -> Int -> (a -> b) -> [a]
514584capPerKey = \items pred maxN getKey ->
515585 union (removeWhere items pred) (capList (filter items pred) (maxN - 1) getKey)
516586
517587-- Shared empty/error shape for /poll. Only a few fields ever vary.
588+emptyPollResp : Timestamp -> Bool -> Bool -> Bool -> Int -> Int -> Bool -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}], droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int, loadPct: Int, messageCancellations: [{messageId: MessageIdHex, cancelledAt: Timestamp}]}
518589emptyPollResp = \cursor authorized rateLimited draining gen retryAfterMs cursorWasReset ->
519590 {events: [], cursor: cursor, authorized: authorized, rateLimited: rateLimited, draining: draining, generation: gen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: retryAfterMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: []}
520591
521592-- Shared error shape for /messages send.
593+sendErrResp : MessageStatus -> Text -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int, error: Text, loadPct: Int}
522594sendErrResp = \status err -> {status: status, messageId: "", retryAfterMs: 0, error: err, loadPct: 0}
523595
524596-- True when a send produced (or matched) a deliverable message.
597+isSendSuccess : MessageStatus -> Bool
525598isSendSuccess = \s -> case s of
526599 MsDelivered {} -> True {}
527600 MsFederated {} -> True {}
@@ -530,27 +603,32 @@ isSendSuccess = \s -> case s of
530603 _ -> False {}
531604
532605-- Sequence an IO action over each item, in order.
533−forEach = \items action -> case headOf items of
606+forEach : [a] -> (a -> IO {} {}) -> IO {} {}
607+forEach = \items action -> case single (takeRelation 1 items) of
534608 Nothing {} -> yield {}
535− Just {value: x} -> do action x; forEach (drop 1 items) action
609+ Just {value: x} -> do action x; forEach (dropRelation 1 items) action
536610
537611-- Like forEach, but collects each action's result into a list.
612+mapEach : [a] -> (a -> IO {} b) -> IO {} [b]
538613mapEach = \items action -> case headOf items of
539614 Nothing {} -> yield []
540615 Just {value: x} -> do
541616 r <- action x
542− rest <- mapEach (drop 1 items) action
617+ rest <- mapEach (dropRelation 1 items) action
543618 yield (union [r] rest)
544619
545620-- Hex tokens: one random hex char per recursive call.
621+genTokenN : Int -> IO {} Text
546622genTokenN = \n -> if n <= 0 then yield "" else do
547623 d <- randomInt 16
548624 rest <- genTokenN (n - 1)
549625 yield (take 1 (drop d "0123456789abcdef") ++ rest)
550626
627+genToken : IO {} Text
551628genToken = genTokenN 48
552629
553630-- Use the supplied message id when non-empty, otherwise generate a fresh one.
631+resolveMessageId : OptionalMessageId -> IO {} OptionalMessageId
554632resolveMessageId = \mid -> if mid != "" then yield mid else genToken
555633
556634tooLong : Text -> Int -> Bool
@@ -593,24 +671,33 @@ isBadServerName = \sn ->
593671-- ============================================
594672-- Auth helpers
595673-- ============================================
674+-- Bearer token extraction, session lookup, maintenance secret check, and the
675+-- `gateAuth`/`gateClientWrite`/`gateLookup` combinators that every client
676+-- handler funnels through (drain → global rate → per-key rate → session).
677+-- Federation requests use a separate path (`withFederationAuth`) supporting
678+-- both Ed25519-signed and callback-Bearer auth.
596679
597680-- Plain ip match (sessions with empty ip skip the check).
681+matchIp : Text -> Text -> Bool
598682matchIp = \clientIp sessionIp -> sessionIp == "" || clientIp == sessionIp
599683
684+extractToken : Text -> Maybe Text
600685extractToken = \h ->
601686 if hasPrefix "Bearer " h && drop 7 h != "" then Just {value: drop 7 h} else Nothing {}
602687
603−verifyMaintenanceAuth : Text -> IO Bool
688+verifyMaintenanceAuth : Text -> IO {} Bool
604689verifyMaintenanceAuth = \authorization -> do
605690 secret <- *runtimeMaintenanceSecret
606691 yield (secret != "" && extractToken authorization == Just {value: secret})
607692
693+withMaintAuth : Text -> IO {} a -> IO {} a -> IO {} a
608694withMaintAuth = \authorization onUnauth onAuth -> do
609695 authorized <- verifyMaintenanceAuth authorization
610696 if authorized then onAuth else onUnauth
611697
612698-- Common gate for handlers: optional drain check, global rate, session auth.
613699-- mkErr is called with one of: "draining", "rate_limited", "unauthorized".
700+gateAuth : Bool -> Text -> Text -> Timestamp -> (resp -> ret) -> (Text -> resp) -> (PubkeyHex -> IO {} ret) -> IO {} ret
614701gateAuth = \checkDrain auth ip t respond mkErr onAuth -> do
615702 dp <- *drainPhase
616703 g <- checkGlobalRate t
@@ -620,14 +707,27 @@ gateAuth = \checkDrain auth ip t respond mkErr onAuth -> do
620707 else withSessionAuth auth t ip (err "unauthorized") onAuth
621708
622709-- gateAuth + per-key rate gate. `rateResp` is the response value when the per-key gate trips.
623−gateRated = \rateRef maxRate windowMs checkDrain auth ip t respond mkErr rateResp onAuth ->
624− gateAuth checkDrain auth ip t respond mkErr (\pk -> do
625− lim <- checkPubkeyRateGate rateRef pk maxRate windowMs t
710+gateClientWrite : Text -> Text -> Timestamp -> (resp -> ret) -> (Text -> resp) -> resp -> (PubkeyHex -> IO {} ret) -> IO {} ret
711+gateClientWrite = \auth ip t respond mkErr rateResp onAuth ->
712+ gateAuth (True {}) auth ip t respond mkErr (\pk -> do
713+ lim <- atomic do
714+ rates <- *clientWriteRates
715+ let r = checkRate rates pk maxClientWriteRatePerKey clientWriteRateWindowMs t
716+ set *clientWriteRates = r.entries
717+ yield r.limited
626718 if lim then yield (respond rateResp) else onAuth pk)
627719
628−gateClientWrite = gateRated clientWriteRates maxClientWriteRatePerKey clientWriteRateWindowMs (True {})
629−gateLookup = gateRated lookupRates maxLookupRatePerKey lookupRateWindowMs (False {})
720+gateLookup : Text -> Text -> Timestamp -> (resp -> ret) -> (Text -> resp) -> resp -> (PubkeyHex -> IO {} ret) -> IO {} ret
721+gateLookup = \auth ip t respond mkErr rateResp onAuth ->
722+ gateAuth (False {}) auth ip t respond mkErr (\pk -> do
723+ lim <- atomic do
724+ rates <- *lookupRates
725+ let r = checkRate rates pk maxLookupRatePerKey lookupRateWindowMs t
726+ set *lookupRates = r.entries
727+ yield r.limited
728+ if lim then yield (respond rateResp) else onAuth pk)
630729
730+withSessionAuth : Text -> Int<Ms> -> Text -> a -> (PubkeyHex -> a) -> a
631731withSessionAuth = \authorization t clientIp onUnauth onAuth ->
632732 case extractToken authorization of
633733 Nothing {} -> onUnauth
@@ -646,14 +746,17 @@ generateFederationAuth = \scope requestTimestamp ->
646746 else do
647747 token <- genToken
648748 t <- now
649− atomic (appendCapped pendingVerifications [{token: token, createdAt: t, scope: scope}] maxPendingVerifications (\p -> p.createdAt))
749+ atomic do
750+ cur <- *pendingVerifications
751+ set *pendingVerifications = capList (union cur [{token: token, createdAt: t, scope: scope}]) maxPendingVerifications (\p -> p.createdAt)
650752 yield ("Bearer " ++ token)
651753
652754-- Verify a federation request: signed (Ed25519 over scope|ts|fromServer|nonce) or callback Bearer token.
755+verifyFederationAuth : Text -> Text -> Text -> a -> IO {network} {} Bool
653756verifyFederationAuth = \authorization fromServer scope requestTimestamp -> do
654757 if hasPrefix "Signed " authorization then do
655758 let payload = drop 7 authorization
656− if payload == "" || tooLong payload (maxSignatureLen + 25) then yield False
759+ if payload == "" || tooLong payload (maxSignatureLen + 25) then yield False {}
657760 else do
658761 -- Split payload on first ':' into {sig, nonce}.
659762 let cs = chars payload
@@ -664,14 +767,15 @@ verifyFederationAuth = \authorization fromServer scope requestTimestamp -> do
664767 let nonce = drop (sepIdx + 1) payload
665768 result <- fetch (federationUrl fromServer) (FederationPubkey {})
666769 case result of
667− Ok r -> yield (r.ok && r.pubkey != "" && verify (bytesFromHex r.pubkey) (textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer ++ (if nonce == "" then "" else "|" ++ nonce))) (bytesFromHex sig))
668− _ -> yield False
770+ Ok {value: r} -> yield (r.ok && r.pubkey != "" && verify (bytesFromHex r.pubkey) (textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer ++ (if nonce == "" then "" else "|" ++ nonce))) (bytesFromHex sig))
771+ _ -> yield False {}
669772 else case extractToken authorization of
670773 Just {value: token} -> do
671− result <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
672− case result of Ok r -> yield r.ok; _ -> yield False
673− _ -> yield False
774+ result <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope, clientIp: ""})
775+ case result of Ok {value: r} -> yield r.ok; _ -> yield False {}
776+ _ -> yield False {}
674777
778+withFederationAuth : Text -> Timestamp -> ({ok: Bool, error: Text, retryAfterMs: Int} -> ret) -> Text -> Text -> Timestamp -> Bool -> IO {network} ret -> IO {network} ret
675779withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do
676780 blocked <- *blockedPeers
677781 allowed <- *allowedPeers
@@ -694,8 +798,16 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
694798-- ============================================
695799-- Rate limiting (plain count-per-window, no sliding interpolation)
696800-- ============================================
801+-- Two flavors of fixed-window counters:
802+-- * Per-key (`checkRate`) — keyed by pubkey, ip, server, or pair string;
803+-- entries live in a `RateEntry` list ref capped at 5000 with LRU eviction.
804+-- * Global (`checkGlobalRate`/`checkAuthGlobalRate`) — single counter+windowStart
805+-- pair, used for the request-wide and auth-wide ceilings.
806+-- `authPreCheck` composes drain + global + ip + auth-global into a one-shot
807+-- gate for unauthenticated handlers (challenge/verify).
697808
698809-- Returns {limited, entries}. Increments count when the key fits within the window.
810+checkRate : [RateEntry] -> Text -> Int -> Int<Ms> -> Timestamp -> {limited: Bool, entries: [RateEntry]}
699811checkRate = \entries key maxRate windowMs t ->
700812 let cur = case findFirst entries (\r -> r.key == key && t - r.windowStart < windowMs) of
701813 Just {value: r} -> {windowStart: r.windowStart, count: r.count + 1}
@@ -704,39 +816,50 @@ checkRate = \entries key maxRate windowMs t ->
704816 then {limited: True {}, entries: entries}
705817 else {limited: False {}, entries: capList (union (removeWhere entries (\x -> x.key == key)) [{key: key, windowStart: cur.windowStart, count: cur.count}]) 5000 (\r -> r.windowStart)}
706818
707−checkWindowRate = \windowRef countRef windowMs maxRate t -> atomic do
708− ws <- *windowRef
709− c <- *countRef
710− if t - ws >= windowMs then do
711− set *windowRef = t
712− set *countRef = 1
713− yield False
714− else if c >= maxRate then yield True
819+checkGlobalRate : Int<Ms> -> IO {} Bool
820+checkGlobalRate = \t -> atomic do
821+ ws <- *globalRateWindowStart
822+ c <- *globalRateCount
823+ if t - ws >= globalRateWindowMs then do
824+ set *globalRateWindowStart = t
825+ set *globalRateCount = 1
826+ yield False {}
827+ else if c >= maxGlobalRequestRate then yield True {}
715828 else do
716− set *countRef = c + 1
717− yield False
718−
719−checkGlobalRate = checkWindowRate globalRateWindowStart globalRateCount globalRateWindowMs maxGlobalRequestRate
720−checkAuthGlobalRate = checkWindowRate authGlobalRateWindowStart authGlobalRateCount authGlobalRateWindowMs maxAuthGlobalRate
829+ set *globalRateCount = c + 1
830+ yield False {}
831+
832+checkAuthGlobalRate : Int<Ms> -> IO {} Bool
833+checkAuthGlobalRate = \t -> atomic do
834+ ws <- *authGlobalRateWindowStart
835+ c <- *authGlobalRateCount
836+ if t - ws >= authGlobalRateWindowMs then do
837+ set *authGlobalRateWindowStart = t
838+ set *authGlobalRateCount = 1
839+ yield False {}
840+ else if c >= maxAuthGlobalRate then yield True {}
841+ else do
842+ set *authGlobalRateCount = c + 1
843+ yield False {}
721844
845+checkIpRateGate : Text -> Timestamp -> IO {} Bool
722846checkIpRateGate = \ip t ->
723− if ip == "" then yield False
724− else checkPubkeyRateGate ipRates ip maxIpRatePerIp ipRateWindowMs t
725−
726−-- Non-atomic version: caller must wrap in `atomic do`.
727−chargeRate = \rateRef key maxRate windowMs t -> do
728− rates <- *rateRef
729− let r = checkRate rates key maxRate windowMs t
730− set *rateRef = r.entries
847+ if ip == "" then yield False {}
848+ else atomic do
849+ rates <- *ipRates
850+ let r = checkRate rates ip maxIpRatePerIp ipRateWindowMs t
851+ set *ipRates = r.entries
852+ yield r.limited
853+
854+checkServerRateGate : Text -> Int -> IO {} Bool
855+checkServerRateGate = \server t -> atomic do
856+ rates <- *federationRates
857+ let r = checkRate rates server maxFederationRatePerServer federationRateWindowMs t
858+ set *federationRates = r.entries
731859 yield r.limited
732860
733−checkPubkeyRateGate = \rateRef pubkey maxRate windowMs t -> atomic
734− (chargeRate rateRef pubkey maxRate windowMs t)
735−
736−checkServerRateGate = \server ->
737− checkPubkeyRateGate federationRates server maxFederationRatePerServer federationRateWindowMs
738−
739861-- Pre-auth gate: drain + global + ip + auth-global. Returns "" when ok, else an error code.
862+authPreCheck : Text -> Timestamp -> IO {} Text
740863authPreCheck = \clientIp t -> do
741864 dp <- *drainPhase
742865 g <- checkGlobalRate t
@@ -751,6 +874,10 @@ authPreCheck = \clientIp t -> do
751874-- ============================================
752875-- Federation failure tracking (simple, no circuit breaker)
753876-- ============================================
877+-- Linear backoff per peer: each consecutive failure multiplies the delay by
878+-- `retryBackoffBase` (capped at `retryBackoffMax`). `shouldRetryServer` is the
879+-- single read-side gate; `recordFailure`/`clearFailure` are the write side.
880+-- There is no half-open state — a single success clears all accumulated failures.
754881
755882retryBackoff : Int -> Int<Ms>
756883retryBackoff = \failures -> min retryBackoffMax (retryBackoffBase * max 1 failures)
@@ -759,6 +886,7 @@ shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
759886shouldRetryServer = \server t failures ->
760887 none failures (\f -> f.server == server && t - f.failedAt < retryBackoff f.failures)
761888
889+recordFailure : Text -> IO {} {}
762890recordFailure = \server -> do
763891 t <- now
764892 atomic do
@@ -770,13 +898,19 @@ recordFailure = \server -> do
770898 set *forwardFailures = union (removeWhere failures same)
771899 [{server: server, failedAt: t, failures: prev.failures, firstFailedAt: prev.firstFailedAt}]
772900
773−clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))
901+clearFailure : Text -> IO {} {}
902+clearFailure = \server -> atomic do
903+ forwardFailures <- *forwardFailures
904+ set *forwardFailures = filter forwardFailures (\f -> f.server != server)
774905
775906-- ============================================
776907-- Common utility queries
777908-- ============================================
909+-- Pure helpers shared by send/poll/federation paths: TTL/timestamp clamping,
910+-- monotonic sequence generation, presence lookup, duplicate detection across
911+-- (messages, ackedIds, deliveredIds), and last-write-wins presence merge.
778912
779−clampTtl : Int<S> -> Int<S>
913+clampTtl : Int<Ms> -> Int<Ms>
780914clampTtl = clamp 1 maxPresenceTtl
781915
782916clampTimestamp : Timestamp -> Timestamp -> Timestamp
@@ -791,10 +925,12 @@ nextSeqTs = \t seqCounter -> max t (seqCounter + 1)
791925dedupMessageIds : [MessageIdHex] -> [MessageIdHex]
792926dedupMessageIds = \ids -> takeRelation maxMessageIds (unique ids)
793927
928+presenceForKey : [PresenceEntry] -> PubkeyHex -> Timestamp -> [PresenceEntry]
794929presenceForKey = \presence pubkey t ->
795930 filter presence (\p -> p.pubkey == pubkey && p.expiresAt > t)
796931
797932-- Resolve a key's server for client lookup, masking remote identities behind "remote".
933+maskedServer : [PresenceEntry] -> PubkeyHex -> Timestamp -> Text
798934maskedServer = \pres pubkey t ->
799935 let here = presenceForKey pres pubkey t in
800936 if any here (\p -> isLocalServer p.server) then serverHostN
@@ -804,6 +940,7 @@ maskedServer = \pres pubkey t ->
804940hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool
805941hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)
806942
943+isDuplicateMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> [{messageId: MessageIdHex, toKey: PubkeyHex | b}] -> [{messageId: MessageIdHex, toKey: PubkeyHex | c}] -> Bool
807944isDuplicateMessage = \mid toKey msgs aids dids ->
808945 hasMessage mid toKey msgs || hasMessage mid toKey aids || hasMessage mid toKey dids
809946
@@ -815,6 +952,7 @@ knownPeerServersFromPresence : [PresenceEntry] -> [Text]
815952knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.server)
816953
817954-- Last-write-wins merge by (pubkey, server), keeping the entry with the larger expiresAt.
955+mergePresence : [PresenceEntry] -> [PresenceEntry] -> [PresenceEntry]
818956mergePresence = \existing incoming ->
819957 fold (\acc e ->
820958 let same = \x -> x.pubkey == e.pubkey && x.server == e.server in
@@ -824,58 +962,87 @@ mergePresence = \existing incoming ->
824962-- ============================================
825963-- Pruning
826964-- ============================================
965+-- Single sweep that drops expired entries from every state ref in one atomic
966+-- block. Returns the locally-expired presence rows whose pubkey has no other
967+-- live local session, so the caller can broadcast offline gossip for them.
827968
969+pruneExpired : Timestamp -> IO {} [PresenceEntry]
828970pruneExpired = \t -> atomic do
829− pruneRef challenges (\x -> x.expiresAt > t)
830− pruneRef sessions (\x -> x.expiresAt > t)
831− pruneRef messages (\m -> m.expiresAt > t)
832− pruneRef forwards (\f -> f.expiresAt > t)
833− pruneRef ackedIds (\a -> t - a.ackedAt < messageMaxAge)
834− pruneRef deliveredIds (\d -> t - d.deliveredAt < messageMaxAge)
835− pruneRef forwardFailures (\f -> t - f.failedAt < 86400000)
836− pruneRef activePolls (\p -> t - p.startedAt < pollTimeoutMs * 3)
837− pruneRef pendingVerifications (\p -> t - p.createdAt < verificationTokenExpiry)
838− pruneRef senderExpirations (\x -> t - x.expiredAt < unackedMessageMaxAge)
839− pruneRef senderDeliveries (\x -> t - x.deliveredAt < unackedMessageMaxAge)
840− pruneRef cancelNotifications (\x -> t - x.cancelledAt < unackedMessageMaxAge)
841− pruneRef pendingFederationCancels (\x -> t - x.createdAt < messageMaxAge)
971+ ch <- *challenges
972+ set *challenges = filter ch (\x -> x.expiresAt > t)
973+ ses <- *sessions
974+ set *sessions = filter ses (\x -> x.expiresAt > t)
975+ msgs <- *messages
976+ set *messages = filter msgs (\m -> m.expiresAt > t)
977+ fwds <- *forwards
978+ set *forwards = filter fwds (\f -> f.expiresAt > t)
979+ aids <- *ackedIds
980+ set *ackedIds = filter aids (\a -> t - a.ackedAt < messageMaxAge)
981+ dids <- *deliveredIds
982+ set *deliveredIds = filter dids (\d -> t - d.deliveredAt < messageMaxAge)
983+ ff <- *forwardFailures
984+ set *forwardFailures = filter ff (\f -> t - f.failedAt < 86400000)
985+ aps <- *activePolls
986+ set *activePolls = filter aps (\p -> t - p.startedAt < pollTimeoutMs * 3)
987+ pv <- *pendingVerifications
988+ set *pendingVerifications = filter pv (\p -> t - p.createdAt < verificationTokenExpiry)
989+ se <- *senderExpirations
990+ set *senderExpirations = filter se (\x -> t - x.expiredAt < unackedMessageMaxAge)
991+ sd <- *senderDeliveries
992+ set *senderDeliveries = filter sd (\x -> t - x.deliveredAt < unackedMessageMaxAge)
993+ cn <- *cancelNotifications
994+ set *cancelNotifications = filter cn (\x -> t - x.cancelledAt < unackedMessageMaxAge)
995+ pfc <- *pendingFederationCancels
996+ set *pendingFederationCancels = filter pfc (\x -> t - x.createdAt < messageMaxAge)
842997 -- presence: remember locally-expired keys for offline gossip
843998 pres <- *presence
844− pruneRef presence (\p -> p.expiresAt > t)
999+ set *presence = filter pres (\p -> p.expiresAt > t)
8451000 let stillLive = \pk -> any pres (\k -> isLocalServer k.server && k.pubkey == pk && k.expiresAt > t)
8461001 yield (filter pres (\p -> p.expiresAt <= t && isLocalServer p.server && not (stillLive p.pubkey)))
8471002
8481003-- ============================================
8491004-- Federation gossip
8501005-- ============================================
1006+-- Construction and outbound dispatch of presence gossip events. Each event
1007+-- carries a `seenBy` list to short-circuit loops on the receiver. Failures
1008+-- from a peer record into `forwardFailures` (shared with forward retries) so
1009+-- both subsystems back off together.
8511010
8521011-- Build a gossip event with this server's identity.
1012+mkGossip : GossipEventType -> PubkeyHex -> Timestamp -> Int<Ms> -> GossipEvent
8531013mkGossip = \etype pk t ttl ->
8541014 {eventType: etype, pubkey: pk, server: serverHostN, timestamp: t, ttl: ttl, seenBy: [serverHostN]}
8551015
8561016-- Build an acked entry for a federated message.
1017+mkAckedEntry : {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text | a} -> Timestamp -> AckedEntry
8571018mkAckedEntry = \m t -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}
8581019
8591020-- Record federation acks and sender deliveries for a batch of just-removed messages.
8601021-- Returns the federated-ack entries (the caller flushes them remotely later).
1022+recordAcksAndDeliveries : [Message] -> Timestamp -> IO {} [AckedEntry]
8611023recordAcksAndDeliveries = \toAck t -> do
8621024 let ackedFed = mapWhere toAck (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
8631025 let myDel = mapWhere toAck (\m -> m.senderKey != "" && m.originServer == "")
8641026 (\m -> {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t})
865− appendCapped ackedIds ackedFed maxAckedIds (\a -> a.ackedAt)
866− appendCapped senderDeliveries myDel 10000 (\d -> d.deliveredAt)
1027+ aids <- *ackedIds
1028+ set *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)
1029+ sds <- *senderDeliveries
1030+ set *senderDeliveries = capList (union sds myDel) 10000 (\d -> d.deliveredAt)
8671031 yield ackedFed
8681032
8691033-- Common cancel block: drop matching messages + forwards, record cancel notification.
8701034-- Returns the federated forward servers that had the message (for caller follow-up).
1035+applyCancel : [a] -> (forall r. {messageId: MessageIdHex, toKey: Text | r} -> Bool) -> Text -> MessageIdHex -> Timestamp -> IO {} [Text]
8711036applyCancel = \msgs matches to mid t -> do
8721037 set *messages = removeWhere msgs matches
8731038 fwds <- *forwards
8741039 set *forwards = removeWhere fwds matches
875− appendCapped cancelNotifications [{toKey: to, messageId: mid, cancelledAt: t}] 10000 (\c -> c.cancelledAt)
1040+ cn <- *cancelNotifications
1041+ set *cancelNotifications = capList (union cn [{toKey: to, messageId: mid, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)
8761042 yield (collectPeers (filter fwds matches) (\f -> f.toServer))
8771043
8781044-- Cancel one message if `ownerPred` holds for it; returns whether a cancel happened.
1045+cancelOne : MessageIdHex -> PubkeyHex -> (Message -> Bool) -> Timestamp -> IO {} Bool
8791046cancelOne = \mid to ownerPred t -> atomic do
8801047 msgs <- *messages
8811048 let matches = \x -> x.messageId == mid && x.toKey == to
@@ -883,6 +1050,7 @@ cancelOne = \mid to ownerPred t -> atomic do
8831050 when owns (do applyCancel msgs matches to mid t; yield {})
8841051 yield owns
8851052
1053+sendBatchedGossip : [Text] -> [GossipEvent] -> IO {network} {}
8861054sendBatchedGossip = \peers events -> forEach peers (\server -> do
8871055 t <- now
8881056 failures <- *forwardFailures
@@ -896,6 +1064,7 @@ sendBatchedGossip = \peers events -> forEach peers (\server -> do
8961064 Err _ -> recordFailure server
8971065 Ok _ -> clearFailure server))
8981066
1067+broadcastOfflineGossip : Timestamp -> [PresenceEntry] -> IO {network} {}
8991068broadcastOfflineGossip = \t expiredLocal -> do
9001069 presence <- *presence
9011070 when (federationEnabled && count expiredLocal > 0)
@@ -904,7 +1073,15 @@ broadcastOfflineGossip = \t expiredLocal -> do
9041073-- ============================================
9051074-- Forward retries and acks
9061075-- ============================================
907−
1076+-- `retryPendingForwards` walks queued cross-server forwards in batches per
1077+-- destination, increments retry counts, and gives up after `maxForwardRetries`.
1078+-- `sendAcksAndClean` flushes federated acks back to origin servers and prunes
1079+-- the local `ackedIds` rows once the remote confirms.
1080+-- `forwardQueuedToServer` is the catch-up path triggered when a remote key
1081+-- comes online: any messages already buffered for that pubkey are enqueued
1082+-- as forwards to the newly-known server.
1083+
1084+retryPendingForwards : IO {network} {}
9081085retryPendingForwards = do
9091086 t <- now
9101087 fwds <- *forwards
@@ -921,8 +1098,8 @@ retryPendingForwards = do
9211098 authorization: auth, correlationId: "",
9221099 expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})
9231100 success <- case result of
924− Err _ -> do recordFailure server; yield False
925− Ok r -> do when r.ok (clearFailure server); yield r.ok
1101+ Err _ -> do recordFailure server; yield False {}
1102+ Ok {value: r} -> do when r.ok (clearFailure server); yield r.ok
9261103 let isMatch = \x -> x.messageId == f.messageId && x.toServer == server
9271104 let isDone = \x -> isMatch x && (success || x.retries + 1 >= maxForwardRetries)
9281105 atomic do
@@ -932,6 +1109,7 @@ retryPendingForwards = do
9321109 then {messageId: x.messageId, toServer: x.toServer, createdAt: x.createdAt, toKey: x.toKey, encryptedBlob: x.encryptedBlob, timestamp: x.timestamp, originServer: x.originServer, retries: x.retries + 1, expiresAt: x.expiresAt, priority: x.priority, senderKey: x.senderKey}
9331110 else x)))
9341111
1112+sendAcksAndClean : [AckedEntry] -> IO {network} {}
9351113sendAcksAndClean = \pendingAcks ->
9361114 forEach (collectPeers pendingAcks (\a -> a.originServer)) (\server -> do
9371115 let mids = mapWhere pendingAcks (\a -> a.originServer == server) (\a -> a.messageId)
@@ -945,9 +1123,12 @@ sendAcksAndClean = \pendingAcks ->
9451123 Err _ -> recordFailure server
9461124 Ok _ -> do
9471125 clearFailure server
948− atomic (pruneRef ackedIds (\a -> not (a.originServer == server && elem mids a.messageId))))
1126+ atomic do
1127+ aids <- *ackedIds
1128+ set *ackedIds = filter aids (\a -> not (a.originServer == server && elem mids a.messageId)))
9491129
9501130-- Forward queued messages to a server when a remote key comes online.
1131+forwardQueuedToServer : PubkeyHex -> Text -> Timestamp -> IO {} {}
9511132forwardQueuedToServer = \pubkey server t -> atomic do
9521133 msgs <- *messages
9531134 fwds <- *forwards
@@ -959,7 +1140,11 @@ forwardQueuedToServer = \pubkey server t -> atomic do
9591140-- ============================================
9601141-- Background loops
9611142-- ============================================
1143+-- Composite tick run from `backgroundPrune` (and on demand via /maintenance/prune):
1144+-- expire stale rows, then opportunistically retry pending forwards and refresh
1145+-- local presence so peers don't see us go offline mid-session.
9621146
1147+pruneAndRetry : Timestamp -> IO {network} {}
9631148pruneAndRetry = \t -> do
9641149 atomic do set *lastPrunedAt = t
9651150 expiredLocal <- pruneExpired t
@@ -973,6 +1158,7 @@ pruneAndRetry = \t -> do
9731158 atomic do set *lastPresenceRefresh = t
9741159 refreshLocalPresence t))
9751160
1161+refreshLocalPresence : Timestamp -> IO {network} {}
9761162refreshLocalPresence = \t -> do
9771163 refreshState <- atomic do
9781164 sessions <- *sessions
@@ -989,7 +1175,21 @@ refreshLocalPresence = \t -> do
9891175-- ============================================
9901176-- Handler: challenge / verify / revoke / logout
9911177-- ============================================
992−
1178+-- Identity is a public key; there are no accounts. The flow is:
1179+-- 1. Client requests a fresh nonce (`handleChallenge`).
1180+-- 2. Client signs the nonce with its private key and submits the signature
1181+-- together with the original challenge (`handleVerifyAuth`).
1182+-- 3. Server returns a Bearer token bound to (pubkey, ip, expiry); the new
1183+-- session also installs a local presence row and gossips an `online`
1184+-- event to known peers.
1185+-- `handleRevoke` lets a session revoke its peers (same pubkey, different
1186+-- token); `handleLogout` invalidates the caller's own token and, if no
1187+-- sibling sessions remain, gossips an `offline` event.
1188+
1189+-- POST /auth/challenge: issue a fresh signed-nonce challenge for `pubkey`.
1190+-- Capped per-key (maxChallengesPerKey) and rate-limited; the IP is recorded
1191+-- so /auth/verify can require the same client.
1192+handleChallenge : PubkeyHex -> Text -> ({challenge: Text, ok: Bool, expiresAt: Timestamp, retryAfterMs: Int, error: Text} -> ret) -> IO {} ret
9931193handleChallenge = \pubkey clientIp respond -> do
9941194 t <- now
9951195 preErr <- if requireClientIp && clientIp == "" then yield "client_ip_required" else authPreCheck clientIp t
@@ -1001,13 +1201,20 @@ handleChallenge = \pubkey clientIp respond -> do
10011201 let cappedActive = capPerKey (filter ch (\c -> c.expiresAt > t)) (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt)
10021202 if count cappedActive >= maxChallenges then yield {ok: False {}, error: "capacity"}
10031203 else do
1004− limited <- chargeRate challengeRates pubkey maxChallengeRatePerKey challengeRateWindowMs t
1005− if limited then yield {ok: False {}, error: "rate_limited"}
1204+ chRates <- *challengeRates
1205+ let chR = checkRate chRates pubkey maxChallengeRatePerKey challengeRateWindowMs t
1206+ set *challengeRates = chR.entries
1207+ if chR.limited then yield {ok: False {}, error: "rate_limited"}
10061208 else do
10071209 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
10081210 yield {ok: True {}, error: ""}
10091211 yield (respond {challenge: if result.ok then challenge else "", ok: result.ok, expiresAt: if result.ok then t + challengeExpiry else 0, retryAfterMs: 0, error: result.error})
10101212
1213+-- POST /auth/verify: validate `signature` over `challenge` for `pubkey`.
1214+-- On success, mints a session token, records local presence (`pubkey` is now
1215+-- live on this server), and gossips an `online` event to known peers. If
1216+-- `revokeOthers` is set, all other sessions for the same pubkey are dropped.
1217+handleVerifyAuth : PubkeyHex -> Text -> SignatureHex -> Bool -> Text -> ({token: Text, expiresAt: Timestamp, error: Text} -> ret) -> IO {network} ret
10111218handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond -> do
10121219 t <- now
10131220 preErr <- authPreCheck clientIp t
@@ -1041,6 +1248,9 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
10411248 (fork (sendBatchedGossip result.peers [mkGossip (GossipOnline {}) pubkey t onlineGossipTtl]))
10421249 yield (respond {token: token, expiresAt: expiry, error: ""})
10431250
1251+-- POST /auth/revoke: caller (any session for the pubkey) revokes a sibling
1252+-- session by token. Returns ok=False if no such sibling existed.
1253+handleRevoke : Text -> Text -> Text -> ({ok: Bool, authorized: Bool} -> ret) -> IO {} ret
10441254handleRevoke = \targetToken authorization clientIp respond -> do
10451255 t <- now
10461256 gateAuth (False {}) authorization clientIp t respond (\_ -> {ok: False {}, authorized: False {}}) (\caller -> do
@@ -1051,6 +1261,9 @@ handleRevoke = \targetToken authorization clientIp respond -> do
10511261 yield (any ses pred)
10521262 yield (respond {ok: result, authorized: True {}}))
10531263
1264+-- POST /auth/logout: drop the caller's own session. If this was the last
1265+-- session for the pubkey, also clear local presence and gossip offline.
1266+handleLogout : Text -> Text -> ({ok: Bool, authorized: Bool} -> ret) -> IO {network} ret
10541267handleLogout = \authorization clientIp respond -> do
10551268 t <- now
10561269 let myToken = case extractToken authorization of Just {value: tk} -> tk; _ -> ""
@@ -1073,18 +1286,35 @@ handleLogout = \authorization clientIp respond -> do
10731286-- ============================================
10741287-- Handler: poll
10751288-- ============================================
1076−
1289+-- Long-poll. The client supplies a cursor (last-seen `receivedAt`) and the
1290+-- server's last-known `generation` (server identity for cursor validity). We:
1291+-- * If the generation mismatches, reset the cursor and signal `cursorReset`.
1292+-- * Ack-and-drop everything <= the cursor (treating poll progress as ack).
1293+-- * Wait on `pollWakeup` (bumped by send/heartbeat) and return new events
1294+-- plus piggybacked metadata (presence updates for `watchKeys`, sender-side
1295+-- expirations/deliveries/cancellations).
1296+-- Per-pubkey concurrency is capped via `activePolls` to bound long-poll holds.
1297+
1298+-- POST /poll: top-level dispatch. Validates drain/global gates, detects a
1299+-- generation mismatch (server restart) to force a cursor reset, then hands
1300+-- off to `pollOnce` for the actual long-poll body.
1301+handlePoll : Timestamp -> Int -> [PubkeyHex] -> Text -> Text -> ({events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}], droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int, loadPct: Int, messageCancellations: [{messageId: MessageIdHex, cancelledAt: Timestamp}]} -> ret) -> IO {network} ret
10771302handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
10781303 t <- now
10791304 serverGen <- *serverGeneration
10801305 dp <- *drainPhase
10811306 g <- checkGlobalRate t
10821307 let cursorWasReset = generation != serverGen && cursor > 0
1083− let er = \auth rate drain retry -> yield (respond (emptyPollResp 0 auth rate drain serverGen retry cursorWasReset))
1308+ let er = \auth rate drain retryRate -> yield (respond (emptyPollResp 0 auth rate drain serverGen retryRate cursorWasReset))
10841309 if dp >= 2 then er (True {}) (False {}) (True {}) 0
10851310 else if g then er (False {}) (True {}) (False {}) globalRateWindowMs
10861311 else pollOnce authorization (if cursorWasReset then 0 else cursor) cursorWasReset (takeRelation maxWatchKeys watchKeys) clientIp serverGen respond t
10871312
1313+-- Body of a single poll. Treats the cursor advance as an implicit ack
1314+-- (`ackPolledMessages`), acquires a per-pubkey poll slot, runs `doPoll` to
1315+-- collect events (or block until wakeup), then assembles the response with
1316+-- piggybacked presence/expiration/delivery/cancellation metadata.
1317+pollOnce : Text -> Timestamp -> Bool -> [PubkeyHex] -> Text -> Int -> ({events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}], messageDeliveries: [{messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}], droppedDeliveryReceipts: Int, droppedExpirationNotifs: Int, loadPct: Int, messageCancellations: [{messageId: MessageIdHex, cancelledAt: Timestamp}]} -> ret) -> Timestamp -> IO {network} ret
10881318pollOnce = \authorization cursor cursorWasReset watchKeys clientIp serverGen respond t -> do
10891319 setup <- atomic (withSessionAuth authorization t clientIp
10901320 (yield Nothing {})
@@ -1098,26 +1328,38 @@ pollOnce = \authorization cursor cursorWasReset watchKeys clientIp serverGen res
10981328 Just {value: s} -> do
10991329 acquired <- atomic do
11001330 aps <- *activePolls
1101− if countWhere aps (\p -> p.pubkey == s.pubkey) >= maxPollsPerKey then yield False
1331+ if countWhere aps (\p -> p.pubkey == s.pubkey) >= maxPollsPerKey then yield False {}
11021332 else do
1103− appendCapped activePolls [{pubkey: s.pubkey, startedAt: t}] maxActivePolls (\p -> p.startedAt)
1104− yield True
1333+ set *activePolls = capList (union aps [{pubkey: s.pubkey, startedAt: t}]) maxActivePolls (\p -> p.startedAt)
1334+ yield True {}
11051335 if not acquired
11061336 then yield (respond (emptyPollResp cursor (True {}) (True {}) (False {}) serverGen pollTimeoutMs cursorWasReset))
11071337 else do
11081338 pollResult <- doPoll s t
1109− atomic (pruneRef activePolls (\p -> p.pubkey != s.pubkey))
1339+ atomic do
1340+ aps <- *activePolls
1341+ set *activePolls = filter aps (\p -> p.pubkey != s.pubkey)
11101342 when federationEnabled (fork (sendAcksAndClean s.pendingAcks))
1343+ t2 <- now
11111344 extras <- atomic do
1112− t2 <- now
11131345 pres <- *presence
11141346 let mine = \x -> x.senderKey == s.pubkey
1115− myExp <- collectAndPrune senderExpirations mine (\x -> {messageId: x.messageId, toKey: x.toKey, expiredAt: x.expiredAt})
1116− myDel <- collectAndPrune senderDeliveries mine (\x -> {messageId: x.messageId, toKey: x.toKey, deliveredAt: x.deliveredAt})
1117− myCancels <- collectAndPrune cancelNotifications (\x -> x.toKey == s.pubkey) (\x -> {messageId: x.messageId, cancelledAt: x.cancelledAt})
1347+ seItems <- *senderExpirations
1348+ set *senderExpirations = removeWhere seItems mine
1349+ let myExp = mapWhere seItems mine (\x -> {messageId: x.messageId, toKey: x.toKey, expiredAt: x.expiredAt})
1350+ sdItems <- *senderDeliveries
1351+ set *senderDeliveries = removeWhere sdItems mine
1352+ let myDel = mapWhere sdItems mine (\x -> {messageId: x.messageId, toKey: x.toKey, deliveredAt: x.deliveredAt})
1353+ cnItems <- *cancelNotifications
1354+ let isMine = \x -> x.toKey == s.pubkey
1355+ set *cancelNotifications = removeWhere cnItems isMine
1356+ let myCancels = mapWhere cnItems isMine (\x -> {messageId: x.messageId, cancelledAt: x.cancelledAt})
11181357 yield {presenceUpdates: pluck watchKeys (\k -> {pubkey: k, online: count (presenceForKey pres k t2) > 0}), expirations: myExp, deliveries: myDel, cancellations: myCancels}
11191358 yield (respond {events: pollResult.events, cursor: pollResult.cursor, authorized: True {}, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: extras.presenceUpdates, heartbeat: count pollResult.events == 0, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: pollResult.hasMore, cursorStale: False {}, messageExpirations: extras.expirations, messageDeliveries: extras.deliveries, droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: extras.cancellations})
11201359
1360+-- One pass: collect events at `t`. If empty, retry-wait on `pollWakeup`
1361+-- (bumped by send/heartbeat) and re-collect once at the new wall time.
1362+doPoll : {pubkey: PubkeyHex, clampedCursor: Timestamp, startWakeup: Timestamp, pendingAcks: [AckedEntry]} -> Timestamp -> IO {} {events: [PollEvent], cursor: Timestamp, hasMore: Bool}
11211363doPoll = \s t -> do
11221364 let collectAt = \tnow -> atomic do
11231365 msgs <- *messages
@@ -1131,6 +1373,9 @@ doPoll = \s t -> do
11311373 t2 <- now
11321374 collectAt t2
11331375
1376+-- Pull this pubkey's unexpired messages newer than `cursor`, sort by
1377+-- receivedAt, take up to `maxPollEvents`, and report `hasMore` if truncated.
1378+collectPollEvents : [Message] -> PubkeyHex -> Timestamp -> Timestamp -> {events: [PollEvent], cursor: Timestamp, hasMore: Bool}
11341379collectPollEvents = \messages pubkey cursor t ->
11351380 let myMsgs = filter messages (\m -> m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor) in
11361381 let limited = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs) in
@@ -1138,19 +1383,36 @@ collectPollEvents = \messages pubkey cursor t ->
11381383 cursor: fold (\acc m -> max acc m.receivedAt) cursor limited,
11391384 hasMore: count myMsgs > count limited}
11401385
1386+-- Treat the supplied cursor as an ack: drop all messages with
1387+-- receivedAt <= cursor, plus their pending forwards, and record federated
1388+-- acks/sender deliveries via `recordAcksAndDeliveries`.
1389+ackPolledMessages : PubkeyHex -> Timestamp -> Timestamp -> IO {} [AckedEntry]
11411390ackPolledMessages = \pubkey cursor t -> do
11421391 msgs <- *messages
11431392 let matches = \m -> m.toKey == pubkey && m.receivedAt <= cursor
11441393 let toAck = filter msgs matches
11451394 let mids = pluck toAck (\m -> m.messageId)
11461395 set *messages = removeWhere msgs matches
1147− pruneRef forwards (\f -> not (f.toKey == pubkey && elem mids f.messageId))
1396+ fwds <- *forwards
1397+ set *forwards = filter fwds (\f -> not (f.toKey == pubkey && elem mids f.messageId))
11481398 recordAcksAndDeliveries toAck t
11491399
11501400-- ============================================
11511401-- Handler: send
11521402-- ============================================
1153−
1403+-- Accept a message for a recipient pubkey and decide its placement:
1404+-- * `MsDelivered` -- recipient has a live local session
1405+-- * `MsFederated` -- recipient lives on a remote peer; queued as a forward
1406+-- * `MsQueued` -- no presence yet; held until presence appears
1407+-- * `MsRejected` / `MsRateLimited` / `MsDuplicate` / `MsDraining` -- refused
1408+-- Send is gated by drain phase, global rate, and the per-sender / per-pair /
1409+-- per-recipient rate limits charged together inside `sendMessageCore`.
1410+-- Federated forwards are kicked off in a background fork after the atomic.
1411+
1412+-- POST /messages: send one ciphertext to `to`. Rejects on draining/rate
1413+-- limit/stale timestamp; otherwise calls `sendMessageCore` inside an atomic
1414+-- block and forks the federation forward (if any) once the local commit lands.
1415+handleSendMessage : PubkeyHex -> BlobHex -> Timestamp -> OptionalMessageId -> Int -> Int -> Text -> Text -> ({status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int, error: Text, loadPct: Int} -> ret) -> IO {network} ret
11541416handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority authorization clientIp respond -> do
11551417 t <- now
11561418 dp <- *drainPhase
@@ -1171,13 +1433,29 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
11711433 let ok = isSendSuccess result.status
11721434 yield (respond {status: result.status, messageId: if ok then effMsgId else "", retryAfterMs: 0, error: if ok then "" else "rejected", loadPct: 0})
11731435
1436+-- Atomic core of /messages and /messages/batch. Charges send/pair/recv rate
1437+-- counters together (so any limit denies the message), enforces caps and
1438+-- self-send/duplicate rejections, allocates a monotonic receivedAt via
1439+-- `seqCounter`, appends the message, and pre-stages forwards for any remote
1440+-- presence rows that already exist for `to`. Returns the chosen status and
1441+-- the list of remote servers the caller should kick forwards to.
1442+sendMessageCore : PubkeyHex -> PubkeyHex -> BlobHex -> Timestamp -> OptionalMessageId -> Int -> Int -> Timestamp -> IO {} {status: MessageStatus, remoteServers: [Text]}
11741443sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msgPriority t -> do
11751444 msgs <- *messages
11761445 aids <- *ackedIds
11771446 dids <- *deliveredIds
1178− srLim <- chargeRate sendRates senderKey maxSendRatePerKey sendRateWindowMs t
1179− prLim <- chargeRate pairRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t
1180− rrLim <- chargeRate recvRates to maxRecvRatePerKey recvRateWindowMs t
1447+ sRates <- *sendRates
1448+ let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t
1449+ set *sendRates = sR.entries
1450+ let srLim = sR.limited
1451+ pRates <- *pairRates
1452+ let pR = checkRate pRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t
1453+ set *pairRates = pR.entries
1454+ let prLim = pR.limited
1455+ rRates <- *recvRates
1456+ let rR = checkRate rRates to maxRecvRatePerKey recvRateWindowMs t
1457+ set *recvRates = rR.entries
1458+ let rrLim = rR.limited
11811459 let reject = \s -> yield {status: s, remoteServers: []}
11821460 if senderKey == to then reject (MsRejected {})
11831461 else if isDuplicateMessage effMsgId to msgs aids dids then reject (MsDuplicate {})
@@ -1197,11 +1475,15 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
11971475 forwardsEnqueued <- if federationEnabled then do
11981476 fwds <- *forwards
11991477 let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.messageId == effMsgId && f.toServer == s)) (\s -> {messageId: effMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: expiresAt, priority: msgPriority, senderKey: senderKey})
1200− appendCapped forwards newFwds maxForwardsEntries (\f -> f.createdAt)
1478+ set *forwards = capList (union fwds newFwds) maxForwardsEntries (\f -> f.createdAt)
12011479 yield (count newFwds > 0)
1202− else yield False
1480+ else yield False {}
12031481 yield {status: if any toPresence (\p -> isLocalServer p.server) then MsDelivered {} else if forwardsEnqueued then MsFederated {} else MsQueued {}, remoteServers: remoteServers}
12041482
1483+-- Fire one /federation/forward call per pre-staged server. On success the
1484+-- corresponding row in `forwards` is pruned (no retry needed); on failure
1485+-- it stays so `retryPendingForwards` can re-attempt with backoff.
1486+forkForwardsTo : OptionalMessageId -> PubkeyHex -> BlobHex -> Timestamp -> [Text] -> IO {network} {}
12051487forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
12061488 forEach servers (\server -> do
12071489 fwds <- *forwards
@@ -1219,14 +1501,23 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
12191501 expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})
12201502 case result of
12211503 Err _ -> recordFailure server
1222− Ok r -> when r.ok (do
1504+ Ok {value: r} -> when r.ok (do
12231505 clearFailure server
1224− atomic (pruneRef forwards (\x -> not (same x)))))
1506+ atomic do
1507+ fwds2 <- *forwards
1508+ set *forwards = filter fwds2 (\x -> not (same x))))
12251509
12261510-- ============================================
12271511-- Handler: lookup
12281512-- ============================================
1229−
1513+-- Presence resolution for clients deciding where to send. Remote presence is
1514+-- masked to the literal string "remote" rather than leaking the peer hostname
1515+-- (the routing decision is the server's, not the sender's).
1516+
1517+-- GET /lookup/{pubkey}: return our local server name if the key is here,
1518+-- "remote" if known on a peer, "" otherwise. Peer hostnames are deliberately
1519+-- masked (see section header).
1520+handleLookupPresence : PubkeyHex -> Text -> Text -> ({server: AnyServerName, authorized: Bool, error: Text} -> ret) -> IO {} ret
12301521handleLookupPresence = \pubkey authorization clientIp respond -> do
12311522 t <- now
12321523 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then e else ""}
@@ -1235,6 +1526,9 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
12351526 pres <- *presence
12361527 yield (respond {server: maskedServer pres pubkey t, authorized: True {}, error: ""}))
12371528
1529+-- POST /lookup/batch: same masking as `handleLookupPresence`, but for up to
1530+-- 50 pubkeys in one round-trip.
1531+handleBatchLookupPresence : [PubkeyHex] -> Text -> Text -> ({results: [{pubkey: PubkeyHex, server: AnyServerName}], authorized: Bool, error: Text} -> ret) -> IO {} ret
12381532handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
12391533 t <- now
12401534 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "rate_limited" then e else ""}
@@ -1244,6 +1538,10 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
12441538 let results = pluck (takeRelation 50 pubkeys) (\k -> {pubkey: k, server: maskedServer pres k t})
12451539 yield (respond {results: results, authorized: True {}, error: ""}))
12461540
1541+-- POST /messages/status: classify a message as
1542+-- pending (in mailbox), forwarding (en-route to a peer),
1543+-- delivered (acked or terminal), or unknown.
1544+handleMessageStatus : PubkeyHex -> MessageIdHex -> Text -> Text -> ({status: Text, authorized: Bool} -> ret) -> IO {} ret
12471545handleMessageStatus = \to messageId authorization clientIp respond -> do
12481546 t <- now
12491547 let mkErr = \_ -> {status: "", authorized: False {}}
@@ -1262,7 +1560,15 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
12621560-- ============================================
12631561-- Handler: ack / list / delete / cancel
12641562-- ============================================
1265−
1563+-- Mailbox mutation by the recipient (ack, list, delete) and by the original
1564+-- sender (cancel, cancel-batch). Cancel propagates over federation when the
1565+-- message has already been forwarded to peers; if the remote rejects, we
1566+-- queue a `pendingFederationCancels` retry.
1567+
1568+-- POST /messages/ack: explicit ack by message id (alternative to the implicit
1569+-- cursor-based ack in /poll). Drops mailbox+forward rows and forks federation
1570+-- ack flush for any messages that originated remotely.
1571+handleAckMessages : [MessageIdHex] -> Text -> Text -> ({ok: Bool, authorized: Bool, ackedCount: Int, error: Text} -> ret) -> IO {network} ret
12661572handleAckMessages = \messageIds authorization clientIp respond -> do
12671573 t <- now
12681574 let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}
@@ -1275,17 +1581,29 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
12751581 when federationEnabled (fork (sendAcksAndClean result.pendingAcks))
12761582 yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount, error: ""}))
12771583
1584+-- POST /messages/list: paginated peek at the caller's mailbox without acking.
1585+-- Returns metadata only (no ciphertext) plus the total count.
1586+handleListMessages : Int -> Text -> Text -> ({messages: [{messageId: MessageIdHex, timestamp: Timestamp, receivedAt: Timestamp}], authorized: Bool, count: Int, error: Text} -> ret) -> IO {} ret
12781587handleListMessages = \offset authorization clientIp respond -> do
12791588 t <- now
12801589 let mkErr = \e -> {messages: [], authorized: False {}, count: 0, error: if e == "rate_limited" then e else ""}
1281− gateRated lookupRates maxLookupRatePerKey lookupRateWindowMs (True {}) authorization clientIp t respond mkErr
1282− {messages: [], authorized: True {}, count: 0, error: "rate_limited"} (\pubkey -> do
1590+ gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do
1591+ lim <- atomic do
1592+ rates <- *lookupRates
1593+ let r = checkRate rates pubkey maxLookupRatePerKey lookupRateWindowMs t
1594+ set *lookupRates = r.entries
1595+ yield r.limited
1596+ if lim then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})
1597+ else do
12831598 msgs <- *messages
12841599 let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))
12851600 let total = count myMsgs
1286− let page = pluck (takeRelation maxPollEvents (drop (clamp 0 total offset) myMsgs)) (\m -> {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt})
1601+ let page = pluck (takeRelation maxPollEvents (dropRelation (clamp 0 total offset) myMsgs)) (\m -> {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt})
12871602 yield (respond {messages: page, authorized: True {}, count: total, error: ""}))
12881603
1604+-- POST /messages/delete: same effect as ack from the recipient's side, with
1605+-- a "delete" semantic for the response.
1606+handleDeleteMessages : [MessageIdHex] -> Text -> Text -> ({ok: Bool, authorized: Bool, deletedCount: Int, error: Text} -> ret) -> IO {network} ret
12891607handleDeleteMessages = \messageIds authorization clientIp respond -> do
12901608 t <- now
12911609 let mkErr = \e -> {ok: False {}, authorized: False {}, deletedCount: 0, error: if e == "draining" then "" else e}
@@ -1294,11 +1612,16 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
12941612 result <- atomic do
12951613 toDel <- removeForKey (\x -> x.toKey == pubkey && elem (dedupMessageIds messageIds) x.messageId)
12961614 let ackedFed = mapWhere toDel (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
1297− appendCapped ackedIds ackedFed maxAckedIds (\a -> a.ackedAt)
1615+ aids <- *ackedIds
1616+ set *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)
12981617 yield {deletedCount: count toDel, pendingAcks: ackedFed}
12991618 when federationEnabled (fork (sendAcksAndClean result.pendingAcks))
13001619 yield (respond {ok: True {}, authorized: True {}, deletedCount: result.deletedCount, error: ""}))
13011620
1621+-- POST /messages/cancel: sender-side recall. Only succeeds while the message
1622+-- is still locally undelivered AND was sent by this caller; if it had already
1623+-- been forwarded to peers, federation cancels are dispatched in a fork.
1624+handleCancelMessage : MessageIdHex -> PubkeyHex -> Text -> Text -> ({ok: Bool, authorized: Bool, cancelled: Bool, error: Text} -> ret) -> IO {network} ret
13021625handleCancelMessage = \messageId to authorization clientIp respond -> do
13031626 t <- now
13041627 let mkErr = \e -> {ok: False {}, authorized: False {}, cancelled: False {}, error: if e == "draining" then "" else e}
@@ -1316,6 +1639,9 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
13161639 (fork (sendCancelToEach messageId to result.cancelServers))
13171640 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
13181641
1642+-- Best-effort federation cancel; on failure the cancel is queued in
1643+-- `pendingFederationCancels` so a later background pass can retry it.
1644+sendCancelToEach : MessageIdHex -> PubkeyHex -> [Text] -> IO {network} {}
13191645sendCancelToEach = \messageId to servers ->
13201646 forEach servers (\server -> do
13211647 t <- now
@@ -1324,9 +1650,13 @@ sendCancelToEach = \messageId to servers ->
13241650 messageId: messageId, toKey: to, fromServer: serverHostN,
13251651 requestTimestamp: t, correlationId: "", authorization: auth,
13261652 protocolVersion: federationProtocolVersion})
1327− when (case result of Err _ -> True {}; Ok r -> not r.ok) (atomic
1328− (appendCapped pendingFederationCancels [{messageId: messageId, toKey: to, toServer: server, createdAt: t, retries: 0}] 10000 (\c -> c.createdAt))))
1653+ when (case result of Err _ -> True {}; Ok {value: r} -> not r.ok) (atomic do
1654+ cur <- *pendingFederationCancels
1655+ set *pendingFederationCancels = capList (union cur [{messageId: messageId, toKey: to, toServer: server, createdAt: t, retries: 0}]) 10000 (\c -> c.createdAt)))
13291656
1657+-- POST /messages/cancel-batch: multi-id sender-side cancel. Same ownership
1658+-- check as `handleCancelMessage`, applied independently per entry.
1659+handleCancelBatch : [BatchCancelEntry] -> Text -> Text -> ({results: [BatchCancelResult], authorized: Bool, error: Text} -> ret) -> IO {} ret
13301660handleCancelBatch = \cancels authorization clientIp respond -> do
13311661 t <- now
13321662 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "draining" then "" else e}
@@ -1337,6 +1667,10 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
13371667 yield {messageId: e.messageId, cancelled: r})
13381668 yield (respond {results: results, authorized: True {}, error: ""}))
13391669
1670+-- POST /messages/batch: multi-recipient send in one call. Each entry runs
1671+-- through `sendMessageCore` separately so per-entry rate limits/errors don't
1672+-- abort the rest; federated forwards are forked per entry.
1673+handleSendBatch : [BatchMessage] -> Text -> Text -> ({results: [BatchResult], authorized: Bool, retryAfterMs: Int} -> ret) -> IO {network} ret
13401674handleSendBatch = \messages authorization clientIp respond -> do
13411675 t <- now
13421676 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
@@ -1355,7 +1689,17 @@ handleSendBatch = \messages authorization clientIp respond -> do
13551689-- ============================================
13561690-- Handler: federation receive
13571691-- ============================================
1358−
1692+-- Inbound endpoints called peer-to-peer. Every handler funnels through
1693+-- `withFederationAuth` which checks: peer block/allow lists, request
1694+-- precondition (e.g. expected origin/server constraints), clock skew vs the
1695+-- request timestamp, signature/Bearer auth, and per-peer rate limit.
1696+-- Receive paths never trust `originServer` for routing; they only use it for
1697+-- ack bookkeeping and SSRF-style filtering.
1698+
1699+-- POST /federation/verify: callback endpoint a peer invokes to validate a
1700+-- Bearer token we issued in `generateFederationAuth`. Single-use: a
1701+-- successful match removes the token from `pendingVerifications`.
1702+handleVerifyToken : Text -> Text -> Text -> ({ok: Bool} -> ret) -> IO {} ret
13591703handleVerifyToken = \token scope clientIp respond -> do
13601704 t <- now
13611705 g <- checkGlobalRate t
@@ -1370,9 +1714,19 @@ handleVerifyToken = \token scope clientIp respond -> do
13701714 yield match
13711715 yield (respond {ok: found})
13721716
1717+-- GET /federation/pubkey: publish the Ed25519 public key peers should use to
1718+-- verify our `Signed ...` federation auth headers. Returns ok=False when the
1719+-- server has no configured federation key (peers fall back to Bearer mode).
1720+handleFederationPubkey : ({pubkey: Text, ok: Bool} -> ret) -> IO {} ret
13731721handleFederationPubkey = \respond -> do
13741722 yield (respond {pubkey: serverFederationPublicKeyHex, ok: serverFederationPublicKeyHex != ""})
13751723
1724+-- POST /federation/gossip: ingest peer presence events. Filters out events
1725+-- whose origin is local (loops) or whose timestamp is in the future beyond
1726+-- skew tolerance, applies last-write-wins merge per (pubkey, server), and
1727+-- — for keys that became newly online — flushes any locally-buffered messages
1728+-- for them via `forwardQueuedToServer` (the catch-up path).
1729+handleRecvGossip : [GossipEvent] -> Int -> ServerName -> Timestamp -> Text -> ({ok: Bool, error: Text, retryAfterMs: Int} -> ret) -> IO {network} ret
13761730handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorization respond -> do
13771731 t <- now
13781732 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (clamp 0 maxGossipHops hopsRemaining >= 1 && isRemote fromServer && count events <= maxGossipEvents) (do
@@ -1396,6 +1750,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
13961750-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.
13971751-- Returns "" on success (including duplicates), otherwise an error code:
13981752-- "no_presence" | "global_full" | "mailbox_full" | "rate_limited" (only when checkRecvRate).
1753+storeFederated : PubkeyHex -> BlobHex -> Timestamp -> ServerName -> MessageIdHex -> Timestamp -> Int -> Bool -> Timestamp -> IO {} Text
13991754storeFederated = \toKey encryptedBlob timestamp originServer messageId expiresAt priority checkRecvRate t -> do
14001755 pres <- *presence
14011756 msgs <- *messages
@@ -1406,7 +1761,12 @@ storeFederated = \toKey encryptedBlob timestamp originServer messageId expiresAt
14061761 else if count msgs >= maxMessages then yield "global_full"
14071762 else if countWhere msgs (\m -> m.toKey == toKey) >= maxMessagesPerRecipient then yield "mailbox_full"
14081763 else do
1409− limited <- if checkRecvRate then chargeRate recvRates toKey maxRecvRatePerKey recvRateWindowMs t else yield False
1764+ limited <- if checkRecvRate then do
1765+ rRates <- *recvRates
1766+ let r = checkRate rRates toKey maxRecvRatePerKey recvRateWindowMs t
1767+ set *recvRates = r.entries
1768+ yield r.limited
1769+ else yield False {}
14101770 if limited then yield "rate_limited"
14111771 else do
14121772 seq <- *seqCounter
@@ -1417,6 +1777,10 @@ storeFederated = \toKey encryptedBlob timestamp originServer messageId expiresAt
14171777 set *pollWakeup = seqTs
14181778 yield ""
14191779
1780+-- POST /federation/forward: receive a single federated message destined for
1781+-- a local key. Delegates the actual store decision to `storeFederated`,
1782+-- which returns "" on success (including the duplicate case).
1783+handleForwardMessage : PubkeyHex -> BlobHex -> Timestamp -> ServerName -> MessageIdHex -> ServerName -> Timestamp -> Timestamp -> Int -> Text -> ({ok: Bool, error: Text, retryAfterMs: Int} -> ret) -> IO {network} ret
14201784handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp expiresAt priority authorization respond -> do
14211785 t <- now
14221786 withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (isRemote originServer && isRemote fromServer) (do
@@ -1426,6 +1790,11 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
14261790 result <- atomic (storeFederated toKey encryptedBlob timestamp originServer messageId expiresAt priority (True {}) t)
14271791 yield (respond {ok: result == "", error: result, retryAfterMs: 0}))
14281792
1793+-- POST /federation/ack: peer reports it has acked messages we forwarded to it.
1794+-- We drop the matching `forwards` rows; if the recipient is no longer locally
1795+-- live and no other forward is still pending for the same id, also reap the
1796+-- backing `messages` row (after `ackGracePeriod` to dampen race with /poll).
1797+handleFederationAck : [MessageIdHex] -> ServerName -> ServerName -> Timestamp -> Text -> ({ok: Bool, error: Text, ackedCount: Int, results: [{messageId: MessageIdHex, found: Bool}], retryAfterMs: Int} -> ret) -> IO {network} ret
14291798handleFederationAck = \messageIds originServer fromServer requestTimestamp authorization respond -> do
14301799 t <- now
14311800 let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0, results: [], retryAfterMs: r.retryAfterMs}
@@ -1449,6 +1818,10 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp autho
14491818 yield {ackedCount: count removed, results: pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})}
14501819 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
14511820
1821+-- POST /federation/cancel: peer asks us to cancel a message it originated
1822+-- here. The owner predicate requires `originServer == fromServer` so peers
1823+-- can only cancel their own messages.
1824+handleFederationCancel : MessageIdHex -> PubkeyHex -> ServerName -> Timestamp -> Text -> ({ok: Bool, error: Text, cancelled: Bool, retryAfterMs: Int} -> ret) -> IO {network} ret
14521825handleFederationCancel = \messageId toKey fromServer requestTimestamp authorization respond -> do
14531826 t <- now
14541827 let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}, retryAfterMs: r.retryAfterMs}
@@ -1456,6 +1829,9 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp authorizat
14561829 cancelled <- cancelOne messageId toKey (\m -> m.originServer == fromServer) t
14571830 yield (respond {ok: True {}, error: "", cancelled: cancelled, retryAfterMs: 0}))
14581831
1832+-- POST /federation/cancel-batch: same ownership constraint as
1833+-- `handleFederationCancel`, applied independently per entry.
1834+handleFederationBatchCancel : [{messageId: MessageIdHex, toKey: PubkeyHex}] -> ServerName -> Timestamp -> Text -> ({ok: Bool, error: Text, results: [BatchCancelResult], retryAfterMs: Int} -> ret) -> IO {network} ret
14591835handleFederationBatchCancel = \cancels fromServer requestTimestamp authorization respond -> do
14601836 t <- now
14611837 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
@@ -1465,18 +1841,26 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp authorization
14651841 yield {messageId: e.messageId, cancelled: cancelled})
14661842 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
14671843
1844+-- POST /federation/forward-batch: bulk variant of `handleForwardMessage`.
1845+-- Skips per-entry recv-rate charging (the per-server federation rate limit
1846+-- already gates the call as a whole).
1847+handleFederationBatchForward : [{toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, expiresAt: Timestamp, priority: Int}] -> ServerName -> Timestamp -> Text -> ({ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} -> ret) -> IO {network} ret
14681848handleFederationBatchForward = \forwards fromServer requestTimestamp authorization respond -> do
14691849 t <- now
14701850 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
14711851 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (isRemote fromServer && count forwards <= maxBatchSize) (do
14721852 results <- mapEach (takeRelation maxBatchSize forwards) (\e -> do
1473− ok <- if isLocalServer e.originServer || t - e.timestamp > messageMaxAge then yield False
1853+ ok <- if isLocalServer e.originServer || t - e.timestamp > messageMaxAge then yield False {}
14741854 else do
14751855 result <- atomic (storeFederated e.toKey e.encryptedBlob e.timestamp e.originServer e.messageId e.expiresAt e.priority (False {}) t)
14761856 yield (result == "")
14771857 yield {messageId: e.messageId, ok: ok})
14781858 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
14791859
1860+-- POST /federation/delivery-receipt: peer confirms delivery of messages whose
1861+-- sender is local to us; we record per-sender delivery rows so that sender's
1862+-- next /poll can include the deliveredAt timestamp.
1863+handleFederationDeliveryReceipt : [MessageIdHex] -> [PubkeyHex] -> ServerName -> Timestamp -> Text -> ({ok: Bool, error: Text, receivedCount: Int, retryAfterMs: Int} -> ret) -> IO {network} ret
14801864handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp authorization respond -> do
14811865 t <- now
14821866 let receiptRespond = \r -> respond {ok: r.ok, error: r.error, receivedCount: 0, retryAfterMs: r.retryAfterMs}
@@ -1491,17 +1875,28 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
14911875 yield {messageId: mp.v, toKey: kp.v}
14921876 atomic do
14931877 msgs <- *messages
1494− appendCapped senderDeliveries (do
1495− p <- pairs
1496− m <- msgs
1497− where m.messageId == p.messageId && m.toKey == p.toKey && m.senderKey != "" && m.originServer == ""
1498− yield {senderKey: m.senderKey, messageId: p.messageId, toKey: p.toKey, deliveredAt: t}) 10000 (\d -> d.deliveredAt)
1878+ sds <- *senderDeliveries
1879+ let newDels = do
1880+ p <- pairs
1881+ m <- msgs
1882+ where m.messageId == p.messageId && m.toKey == p.toKey && m.senderKey != "" && m.originServer == ""
1883+ yield {senderKey: m.senderKey, messageId: p.messageId, toKey: p.toKey, deliveredAt: t}
1884+ set *senderDeliveries = capList (union sds newDels) 10000 (\d -> d.deliveredAt)
14991885 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
15001886
15011887-- ============================================
15021888-- Maintenance handlers
15031889-- ============================================
1504−
1890+-- Operator endpoints, all gated by `runtimeMaintenanceSecret` (Bearer header).
1891+-- Drain is two-phase: phase 1 stops accepting new auth, phase 2 also clears
1892+-- local presence and gossips offline. Admin endpoints can revoke sessions by
1893+-- pubkey or IP, purge a recipient's mailbox, list aggregated session activity,
1894+-- block/allow peers, rotate the maintenance secret, and read live config.
1895+
1896+-- POST /maintenance/drain: phase 1 stops accepting new auth (challenges,
1897+-- verify); phase 2 additionally clears local presence and broadcasts offline
1898+-- gossip for every locally-live key so peers can reroute. Idempotent.
1899+handleDrain : Int -> Text -> ({ok: Bool, drained: Int, phase: Int} -> ret) -> IO {network} ret
15051900handleDrain = \phase authorization respond -> withMaintAuth authorization
15061901 (yield (respond {ok: False {}, drained: 0, phase: 0}))
15071902 (do
@@ -1521,6 +1916,9 @@ handleDrain = \phase authorization respond -> withMaintAuth authorization
15211916 (fork (sendBatchedGossip drainResult.peers (pluck drainResult.localKeys (\p -> mkGossip (GossipOffline {}) p.pubkey t offlineGossipMinTtl))))
15221917 yield (respond {ok: True {}, drained: count drainResult.localKeys, phase: effectivePhase}))
15231918
1919+-- POST /maintenance/undrain: clear drain state and resume normal operation.
1920+-- Does not restore presence — clients must re-authenticate to repopulate.
1921+handleUndrain : Text -> ({ok: Bool} -> ret) -> IO {} ret
15241922handleUndrain = \authorization respond -> withMaintAuth authorization
15251923 (yield (respond {ok: False {}}))
15261924 (do
@@ -1529,19 +1927,33 @@ handleUndrain = \authorization respond -> withMaintAuth authorization
15291927 set *drainCompletedAt = 0
15301928 yield (respond {ok: True {}}))
15311929
1930+-- POST /maintenance/shutdown: convenience wrapper that performs a phase-2
1931+-- drain and reports the count of presence rows cleared.
1932+handleShutdown : Text -> ({ok: Bool, drained: Int} -> ret) -> IO {network} ret
15321933handleShutdown = \authorization respond ->
15331934 handleDrain 2 authorization (\r -> respond {ok: r.ok, drained: r.drained})
15341935
1936+-- POST /maintenance/prune: trigger an out-of-band `pruneAndRetry` tick.
1937+-- Useful when the operator wants to force expiry/forward retries without
1938+-- waiting for the next `backgroundPrune` interval.
1939+handlePrune : Text -> ({ok: Bool} -> ret) -> IO {network} ret
15351940handlePrune = \authorization respond -> do
15361941 t <- now
15371942 authorized <- verifyMaintenanceAuth authorization
15381943 when authorized (fork (pruneAndRetry t))
15391944 yield (respond {ok: authorized})
15401945
1946+-- POST /maintenance/reconcile: stub. Production builds run a state
1947+-- reconciliation pass here; this simplified server reports zero drift.
1948+handleReconcile : Text -> ({ok: Bool, messageDrift: Int, presenceDrift: Int, sessionDrift: Int} -> ret) -> IO {} ret
15411949handleReconcile = \authorization respond -> do
15421950 authorized <- verifyMaintenanceAuth authorization
15431951 yield (respond {ok: authorized, messageDrift: 0, presenceDrift: 0, sessionDrift: 0})
15441952
1953+-- GET /health: aggregate readiness flags. During an initial warmup window
1954+-- (3 prune intervals after startup) prune/heartbeat checks always pass to
1955+-- avoid false negatives before the background loops have ticked.
1956+handleHealth : 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} -> ret) -> IO {} ret
15451957handleHealth = \authorization respond -> do
15461958 t <- now
15471959 lastPrune <- *lastPrunedAt
@@ -1553,36 +1965,80 @@ handleHealth = \authorization respond -> do
15531965 let heartbeatOk = warmup || t - lastHb < pollTimeoutMs * 3
15541966 yield (respond {ok: dp == 0 && pruneOk && heartbeatOk, version: serverBuildVersion, pruneHealthy: pruneOk, heartbeatHealthy: heartbeatOk, capacityHealthy: True {}, capacityWarning: False {}, federationHealthy: True {}, watchdogHealthy: True {}, nonceHealthy: True {}, forwardQueueHealthy: True {}, requestPressureHealthy: True {}, draining: dp > 0, degradedPeers: 0})
15551967
1968+-- GET /ready: signals readiness once the first background prune has run
1969+-- (load-balancer / orchestrator probe).
1970+handleReady : ({ready: Bool} -> ret) -> IO {} ret
15561971handleReady = \respond -> do
15571972 sa <- *startedAt
15581973 lastPrune <- *lastPrunedAt
15591974 yield (respond {ready: lastPrune > 0 && lastPrune >= sa})
15601975
1976+-- POST /maintenance/block: deny-list a peer. Outbound retries skip blocked
1977+-- peers; inbound federation calls from blocked peers fail in `withFederationAuth`.
1978+handleBlockPeer : ServerName -> Text -> ({ok: Bool, blocked: Int} -> ret) -> IO {} ret
15611979handleBlockPeer = \server authorization respond -> withMaintAuth authorization
15621980 (yield (respond {ok: False {}, blocked: 0}))
1563− (do n <- addServerToList blockedPeers server; yield (respond {ok: True {}, blocked: n}))
1564−
1981+ (do
1982+ n <- atomic do
1983+ current <- *blockedPeers
1984+ let updated = unique (union current [server])
1985+ set *blockedPeers = updated
1986+ yield (count updated)
1987+ yield (respond {ok: True {}, blocked: n}))
1988+
1989+-- POST /maintenance/unblock: remove a peer from the deny-list.
1990+handleUnblockPeer : ServerName -> Text -> ({ok: Bool, unblocked: Bool} -> ret) -> IO {} ret
15651991handleUnblockPeer = \server authorization respond -> withMaintAuth authorization
15661992 (yield (respond {ok: False {}, unblocked: False {}}))
1567− (do removed <- removeServerFromList blockedPeers server; yield (respond {ok: True {}, unblocked: removed}))
1568−
1569−handleListPeersResp = \listRef authorization respond -> do
1993+ (do
1994+ removed <- atomic do
1995+ current <- *blockedPeers
1996+ set *blockedPeers = removeWhere current (\x -> x == server)
1997+ yield (elem current server)
1998+ yield (respond {ok: True {}, unblocked: removed}))
1999+
2000+handleListBlocked : Text -> ({peers: [Text], ok: Bool} -> ret) -> IO {} ret
2001+handleListBlocked = \authorization respond -> do
15702002 ok <- verifyMaintenanceAuth authorization
1571− list <- *listRef
2003+ list <- *blockedPeers
15722004 yield (respond {peers: if ok then list else [], ok: ok})
15732005
1574−handleListBlocked = handleListPeersResp blockedPeers
1575−
2006+-- POST /maintenance/allow: when the allowed-peers list is non-empty, only
2007+-- federation calls from listed peers are accepted (whitelist mode). An empty
2008+-- list disables the whitelist.
2009+handleAllowPeer : ServerName -> Text -> ({ok: Bool, allowed: Int} -> ret) -> IO {} ret
15762010handleAllowPeer = \server authorization respond -> withMaintAuth authorization
15772011 (yield (respond {ok: False {}, allowed: 0}))
1578− (do n <- addServerToList allowedPeers server; yield (respond {ok: True {}, allowed: n}))
1579−
2012+ (do
2013+ n <- atomic do
2014+ current <- *allowedPeers
2015+ let updated = unique (union current [server])
2016+ set *allowedPeers = updated
2017+ yield (count updated)
2018+ yield (respond {ok: True {}, allowed: n}))
2019+
2020+-- POST /maintenance/unallow: remove an entry from the allow-list. If the
2021+-- list becomes empty, the whitelist gate disengages.
2022+handleUnallowPeer : ServerName -> Text -> ({ok: Bool, removed: Bool} -> ret) -> IO {} ret
15802023handleUnallowPeer = \server authorization respond -> withMaintAuth authorization
15812024 (yield (respond {ok: False {}, removed: False {}}))
1582− (do removed <- removeServerFromList allowedPeers server; yield (respond {ok: True {}, removed: removed}))
1583−
1584−handleListAllowed = handleListPeersResp allowedPeers
2025+ (do
2026+ removed <- atomic do
2027+ current <- *allowedPeers
2028+ set *allowedPeers = removeWhere current (\x -> x == server)
2029+ yield (elem current server)
2030+ yield (respond {ok: True {}, removed: removed}))
2031+
2032+handleListAllowed : Text -> ({peers: [Text], ok: Bool} -> ret) -> IO {} ret
2033+handleListAllowed = \authorization respond -> do
2034+ ok <- verifyMaintenanceAuth authorization
2035+ list <- *allowedPeers
2036+ yield (respond {peers: if ok then list else [], ok: ok})
15852037
2038+-- POST /maintenance/ping: send a zero-event gossip to a peer and return the
2039+-- round-trip latency. Useful for diagnosing federation reachability without
2040+-- waiting for organic traffic.
2041+handlePingPeer : ServerName -> Text -> ({ok: Bool, latencyMs: Int, error: Text} -> ret) -> IO {network} ret
15862042handlePingPeer = \server authorization respond -> withMaintAuth authorization
15872043 (yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"}))
15882044 (do
@@ -1595,14 +2051,20 @@ handlePingPeer = \server authorization respond -> withMaintAuth authorization
15952051 endTime <- now
15962052 case result of
15972053 Err e -> yield (respond {ok: False {}, latencyMs: endTime - startTime, error: show e})
1598− Ok r -> yield (respond {ok: r.ok, latencyMs: endTime - startTime, error: r.error}))
2054+ Ok {value: r} -> yield (respond {ok: r.ok, latencyMs: endTime - startTime, error: r.error}))
15992055
2056+-- GET /maintenance/sessions/{pubkey}: list a single user's active sessions.
2057+-- Token values are masked ("***") — operators see expiry, ip, createdAt only.
2058+handleAdminListSessions : PubkeyHex -> Text -> ({sessions: [{token: Text, expiresAt: Timestamp, ip: Text, createdAt: Timestamp}], count: Int, ok: Bool} -> ret) -> IO {} ret
16002059handleAdminListSessions = \pubkey authorization respond -> do
16012060 ok <- verifyMaintenanceAuth authorization
16022061 ses <- *sessions
16032062 let mine = if ok then mapWhere ses (\s -> s.pubkey == pubkey) (\s -> {token: "***", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt}) else []
16042063 yield (respond {sessions: mine, count: count mine, ok: ok})
16052064
2065+-- POST /maintenance/sessions/revoke-all: drop every session and local presence
2066+-- row for `pubkey`. Operator-side kick for a compromised key.
2067+handleAdminRevokeSessions : PubkeyHex -> Text -> ({ok: Bool, revokedCount: Int} -> ret) -> IO {} ret
16062068handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth authorization
16072069 (yield (respond {ok: False {}, revokedCount: 0}))
16082070 (do
@@ -1610,10 +2072,15 @@ handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth autho
16102072 ses <- *sessions
16112073 let same = \s -> s.pubkey == pubkey
16122074 set *sessions = removeWhere ses same
1613− pruneRef presence (\p -> p.pubkey != pubkey || isRemote p.server)
2075+ pres <- *presence
2076+ set *presence = filter pres (\p -> p.pubkey != pubkey || isRemote p.server)
16142077 yield (countWhere ses same)
16152078 yield (respond {ok: True {}, revokedCount: revoked}))
16162079
2080+-- POST /maintenance/messages/purge: drop all queued/forwarding messages
2081+-- destined for `pubkey`, recording federation acks for any that originated
2082+-- elsewhere so peers learn we have nothing for the key.
2083+handleAdminPurgeMessages : PubkeyHex -> Text -> ({ok: Bool, purgedCount: Int} -> ret) -> IO {} ret
16172084handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth authorization
16182085 (yield (respond {ok: False {}, purgedCount: 0}))
16192086 (do
@@ -1623,11 +2090,16 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
16232090 let mine = \m -> m.toKey == pubkey
16242091 let purgedMsgs = filter msgs mine
16252092 set *messages = removeWhere msgs mine
1626− pruneRef forwards (\f -> f.toKey != pubkey)
1627− appendCapped ackedIds (mapWhere purgedMsgs (\m -> m.originServer != "") (\m -> mkAckedEntry m t)) maxAckedIds (\a -> a.ackedAt)
2093+ fwds <- *forwards
2094+ set *forwards = filter fwds (\f -> f.toKey != pubkey)
2095+ aids <- *ackedIds
2096+ set *ackedIds = capList (union aids (mapWhere purgedMsgs (\m -> m.originServer != "") (\m -> mkAckedEntry m t))) maxAckedIds (\a -> a.ackedAt)
16282097 yield (count purgedMsgs)
16292098 yield (respond {ok: True {}, purgedCount: purged}))
16302099
2100+-- POST /maintenance/sessions/revoke-ip: kill every session bound to a given
2101+-- client IP, regardless of pubkey. Reports total revoked + distinct keys.
2102+handleAdminRevokeIp : Text -> Text -> ({ok: Bool, revokedCount: Int, affectedKeys: Int} -> ret) -> IO {} ret
16312103handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
16322104 (yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0}))
16332105 (do
@@ -1638,6 +2110,9 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
16382110 yield (mapWhere ses same (\s -> s.pubkey))
16392111 yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count (unique keys)}))
16402112
2113+-- POST /maintenance/sessions/list: paged session inventory aggregated by
2114+-- pubkey (count, distinct ips, latest expiry). Sorted by descending count.
2115+handleAdminListAllSessions : Int -> Text -> ({sessions: [{pubkey: PubkeyHex, count: Int, ips: [Text], latestExpiry: Timestamp}], ok: Bool, total: Int} -> ret) -> IO {} ret
16412116handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization
16422117 (yield (respond {sessions: [], ok: False {}, total: 0}))
16432118 (do
@@ -1654,31 +2129,50 @@ handleAdminListAllSessions = \limit authorization respond -> withMaintAuth autho
16542129 let top = takeRelation cap (sortBy (\g -> 0 - g.count) grouped)
16552130 yield (respond {sessions: top, ok: True {}, total: count grouped}))
16562131
2132+-- GET /maintenance/rate-limits: expose the configured rate ceilings. Returns
2133+-- zeros for unauthenticated callers (response shape stays stable).
2134+handleGetRateLimits : Text -> ({ok: Bool, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int} -> ret) -> IO {} ret
16572135handleGetRateLimits = \authorization respond -> do
16582136 ok <- verifyMaintenanceAuth authorization
16592137 let v = \x -> if ok then x else 0
16602138 yield (respond {ok: ok, globalRequest: v maxGlobalRequestRate, ipRate: v maxIpRatePerIp, sendRate: v maxSendRatePerKey, federationRate: v maxFederationRatePerServer, authGlobal: v maxAuthGlobalRate, recvRate: v maxRecvRatePerKey, challengeRate: v maxChallengeRatePerKey, pairRate: v maxPairRatePerPair, lookupRate: v maxLookupRatePerKey, clientWriteRate: v maxClientWriteRatePerKey})
16612139
2140+-- POST /maintenance/rate-limits: stub. Rate ceilings are compile-time constants
2141+-- in this simplified server, so this just authorizes and returns ok=true.
2142+handleSetRateLimits : Int -> Int -> Int -> Int -> Int -> Int -> Int -> Int -> Int -> Int -> Text -> ({ok: Bool} -> ret) -> IO {} ret
16622143handleSetRateLimits = \globalRequest ipRate sendRate federationRate authGlobal recvRate challengeRate pairRate lookupRate clientWriteRate authorization respond -> do
16632144 authorized <- verifyMaintenanceAuth authorization
16642145 yield (respond {ok: authorized})
16652146
2147+-- POST /maintenance/nonces/flush: stub. Nonce caches were removed; the
2148+-- endpoint is preserved so older operator tooling keeps working.
2149+handleFlushNonces : Text -> ({ok: Bool, flushedCount: Int} -> ret) -> IO {} ret
16662150handleFlushNonces = \authorization respond -> do
16672151 authorized <- verifyMaintenanceAuth authorization
16682152 yield (respond {ok: authorized, flushedCount: 0})
16692153
2154+-- POST /maintenance/rotate-secret: replace the maintenance Bearer secret.
2155+-- The old secret authenticates the rotation; subsequent calls must use the
2156+-- new value.
2157+handleRotateSecret : Text -> Text -> ({ok: Bool, error: Text} -> ret) -> IO {} ret
16702158handleRotateSecret = \newSecret authorization respond -> withMaintAuth authorization
16712159 (yield (respond {ok: False {}, error: "unauthorized"}))
16722160 (do
16732161 atomic do set *runtimeMaintenanceSecret = newSecret
16742162 yield (respond {ok: True {}, error: ""}))
16752163
2164+-- GET /maintenance/config: expose a small set of operationally-relevant
2165+-- compile-time constants (mailbox cap, gossip hops, poll timeout, ack grace).
2166+handleGetConfig : Text -> ({ok: Bool, maxMessagesPerRecipient: Int, maxGossipHops: Int, pollTimeoutMs: Int, ackGracePeriod: Int} -> ret) -> IO {} ret
16762167handleGetConfig = \authorization respond -> do
16772168 ok <- verifyMaintenanceAuth authorization
16782169 let v = \x -> if ok then x else 0
16792170 yield (respond {ok: ok, maxMessagesPerRecipient: v maxMessagesPerRecipient, maxGossipHops: v maxGossipHops, pollTimeoutMs: v pollTimeoutMs, ackGracePeriod: v ackGracePeriod})
16802171
1681−-- /stats: zero-fill all metric fields we no longer compute.
2172+-- GET /stats: capacity gauges + minimal federation/timing snapshot. Most of
2173+-- the historical metric fields are zero-filled (see `zeroStats`) — only the
2174+-- counts that come straight from the state refs are populated.
2175+handleStats : Text -> ({ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, nonceEvictions: Int, activePolls: Int, activePollsMax: 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}], pendingCancels: 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, senderExpirations: Int, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int, clientCancels: Int, federationCancels: 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}} -> ret) -> IO {} ret
16822176handleStats = \authorization respond -> withMaintAuth authorization
16832177 (yield (respond zeroStats))
16842178 (do
@@ -1709,8 +2203,12 @@ handleStats = \authorization respond -> withMaintAuth authorization
17092203 rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0, clientCancels: 0, federationCancels: 0},
17102204 timing: {pruneCount: pc, pruneRunning: 0, heartbeatCount: hc, uptime: t - sa, lastPruneAt: lp, lastHeartbeatAt: lh, draining: dp > 0, drainPhase: dp, totalEntries: sCount + mCount + pCount + fCount, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, pollLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0, sendLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, authAvgMs: 0, authMaxMs: 0, authCount: 0}}))
17112205
2206+zeroStats : {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, nonceEvictions: Int, activePolls: Int, activePollsMax: 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}], pendingCancels: 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, senderExpirations: Int, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int, clientCancels: Int, federationCancels: 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}}
17122207zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, nonceEvictions: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0, messageBucketMax: 0, messageBucketMin: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0, topFailingPeers: [], pendingCancels: 0, gossipSuccess: 0, gossipFail: 0, latencyBuckets: {lt100: 0, lt500: 0, lt2000: 0, gte2000: 0}}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0, clientCancels: 0, federationCancels: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, pollLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0, sendLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0}, authAvgMs: 0, authMaxMs: 0, authCount: 0}}
17132208
2209+-- GET /stats/peers: per-peer failure summary derived from `forwardFailures`,
2210+-- including current backoff and pending forward queue depth.
2211+handleStatsPeers : Text -> ({peers: [PeerStatus], ok: Bool} -> ret) -> IO {} ret
17142212handleStatsPeers = \authorization respond -> withMaintAuth authorization
17152213 (yield (respond {peers: [], ok: False {}}))
17162214 (do
@@ -1720,6 +2218,9 @@ handleStatsPeers = \authorization respond -> withMaintAuth authorization
17202218 yield (respond {peers: pluck failures (\f ->
17212219 {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoff f.failures, retriable: shouldRetryServer f.server t failures, pendingForwards: countWhere fwds (\x -> x.toServer == f.server), avgLatencyMs: 0, maxLatencyMs: 0, latencySamples: 0, gossipEventsPerMin: 0}), ok: True {}}))
17222220
2221+-- GET /stats/key/{pubkey}: per-key footprint — message count, presence rows,
2222+-- session count, queued forwards (with details), and active long-poll count.
2223+handleStatsKey : PubkeyHex -> Text -> ({messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, forwardDetails: [{server: Text, retries: Int, createdAt: Timestamp}], activePolls: Int, ok: Bool} -> ret) -> IO {} ret
17232224handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization
17242225 (yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, forwardDetails: [], activePolls: 0, ok: False {}}))
17252226 (do
@@ -1739,6 +2240,9 @@ handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization
17392240 activePolls: countWhere aps (\x -> x.pubkey == pubkey),
17402241 ok: True {}}))
17412242
2243+-- POST /stats/top-recipients: ranked list of pubkeys by mailbox depth.
2244+-- `limit` defaults to 20, capped at 100, used to spot hot-spots / abuse.
2245+handleTopRecipients : Int -> Text -> ({recipients: [RecipientCount], ok: Bool, total: Int} -> ret) -> IO {} ret
17422246handleTopRecipients = \limit authorization respond -> withMaintAuth authorization
17432247 (yield (respond {recipients: [], ok: False {}, total: 0}))
17442248 (do
@@ -1748,6 +2252,9 @@ handleTopRecipients = \limit authorization respond -> withMaintAuth authorizatio
17482252 let top = takeRelation cap (sortBy (\c -> 0 - c.count) counts)
17492253 yield (respond {recipients: top, ok: True {}, total: count counts}))
17502254
2255+-- GET /stats/forwards: per-destination forward queue summary — count, age of
2256+-- oldest pending entry, and total accumulated retries.
2257+handleStatsForwards : Text -> ({servers: [{server: Text, count: Int, oldestAge: Int, totalRetries: Int}], ok: Bool, total: Int} -> ret) -> IO {} ret
17512258handleStatsForwards = \authorization respond -> withMaintAuth authorization
17522259 (yield (respond {servers: [], ok: False {}, total: 0}))
17532260 (do
@@ -1760,11 +2267,18 @@ handleStatsForwards = \authorization respond -> withMaintAuth authorization
17602267 ok: True {},
17612268 total: count fwds}))
17622269
2270+-- GET /stats/rates: same shape as /maintenance/rate-limits with an extra
2271+-- scalePct field. Kept separate for backwards compatibility with older clients.
2272+handleStatsRates : Text -> ({ok: Bool, scalePct: Int, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int} -> ret) -> IO {} ret
17632273handleStatsRates = \authorization respond -> do
17642274 ok <- verifyMaintenanceAuth authorization
17652275 let v = \x -> if ok then x else 0
17662276 yield (respond {ok: ok, scalePct: v 100, globalRequest: v maxGlobalRequestRate, ipRate: v maxIpRatePerIp, sendRate: v maxSendRatePerKey, federationRate: v maxFederationRatePerServer, authGlobal: v maxAuthGlobalRate, recvRate: v maxRecvRatePerKey, challengeRate: v maxChallengeRatePerKey, pairRate: v maxPairRatePerPair, lookupRate: v maxLookupRatePerKey, clientWriteRate: v maxClientWriteRatePerKey})
17672277
2278+-- GET /metrics: Prometheus-style text exposition of the most useful gauges.
2279+-- The `authorization` header is accepted but not validated here — operators
2280+-- relying on auth must front this endpoint with a reverse proxy that enforces it.
2281+handleMetrics : Text -> ({metrics: Text, ok: Bool} -> ret) -> IO {} ret
17682282handleMetrics = \authorization respond -> do
17692283 t <- now
17702284 msgs <- *messages
@@ -1786,7 +2300,11 @@ handleMetrics = \authorization respond -> do
17862300-- ============================================
17872301-- Dispatch
17882302-- ============================================
2303+-- Single match from generated request constructor (one per route entry above)
2304+-- to the corresponding `handleXxx` function. Adding a new route means adding
2305+-- a `route ... = NewReq` line, a `handleNewReq` definition, and a clause here.
17892306
2307+serve : Api -> IO {network} Response
17902308serve = \req -> case req of
17912309 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
17922310 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
@@ -1847,7 +2365,12 @@ serve = \req -> case req of
18472365-- ============================================
18482366-- Background loops
18492367-- ============================================
2368+-- `backgroundPrune` runs `pruneAndRetry` every `backgroundPruneInterval` ms
2369+-- (skipped while drain phase >= 2). `pollHeartbeat` bumps `pollWakeup` every
2370+-- `pollTimeoutMs` so long-poll holders unblock and return an empty heartbeat
2371+-- response, keeping connections fresh.
18502372
2373+backgroundPrune : IO {network} {}
18512374backgroundPrune = do
18522375 sleep backgroundPruneInterval
18532376 dp <- *drainPhase
@@ -1859,6 +2382,7 @@ backgroundPrune = do
18592382 pruneAndRetry t)
18602383 backgroundPrune
18612384
2385+pollHeartbeat : IO {} {}
18622386pollHeartbeat = do
18632387 sleep pollTimeoutMs
18642388 t <- now
@@ -1872,7 +2396,11 @@ pollHeartbeat = do
18722396-- ============================================
18732397-- Entry Point
18742398-- ============================================
2399+-- Stamp `startedAt` and a fresh `serverGeneration` (used by clients to detect
2400+-- server restarts and reset their poll cursor), spawn the two background
2401+-- loops, and start the listener.
18752402
2403+main : IO {network} {}
18762404main = do
18772405 t <- now
18782406 genJitter <- randomInt 1000000