@@ -206,9 +206,11 @@ federationUrl = \server -> federationScheme ++ server
206206requireClientIp : Bool
207207requireClientIp = True {}
208208
209+-- Drop the Ms unit and convert milliseconds to seconds.
209210msToS : Int<Ms> -> Int<S>
210211msToS = \ms -> stripUnit ms / 1000
211212
213+-- Drop the S unit and convert seconds to milliseconds.
212214sToMs : Int<S> -> Int<Ms>
213215sToMs = \s -> stripUnit s * 1000
214216
@@ -220,12 +222,16 @@ sToMs = \s -> stripUnit s * 1000
220222-- and stripped of trailing dots, with a deny-list filter (`isBadServerName`)
221223-- applied where SSRF is a concern.
222224
225+-- One event in a /poll response. `eventType` is currently always "message"; the
226+-- payload is the cipher blob plus delivery metadata for the recipient.
223227type PollEvent = {
224228 id: MessageIdHex,
225229 eventType: Text,
226230 payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex, expiresAt: Timestamp}
227231}
228232
233+-- Kind of presence event flowing through federation gossip. `GossipUnknown` is
234+-- the parser's fallback for forward-compatibility with future event types.
229235data GossipEventType = GossipOnline {} | GossipOffline {} | GossipUnknown {}
230236
231237impl ToJSON GossipEventType where
@@ -237,6 +243,8 @@ impl ToJSON GossipEventType where
237243impl FromJSON GossipEventType where
238244 parseJson t = (\s -> if s == "online" then GossipOnline {} else if s == "offline" then GossipOffline {} else GossipUnknown {}) (parseJson t : Text)
239245
246+-- One presence change exchanged between federated servers. `seenBy` lists the
247+-- hostnames already aware of the event, used by receivers to break gossip loops.
240248type GossipEvent = {
241249 eventType: GossipEventType,
242250 pubkey: PubkeyHex,
@@ -246,6 +254,9 @@ type GossipEvent = {
246254 seenBy: [Text]
247255}
248256
257+-- Outcome classification returned to the sender of /messages and per-entry in
258+-- /messages/batch. The first four are success-ish (see `isSendSuccess`); the
259+-- rest are refusals.
249260data MessageStatus
250261 = MsDelivered {}
251262 | MsFederated {}
@@ -267,27 +278,42 @@ impl ToJSON MessageStatus where
267278 MsDuplicate {} -> "duplicate"
268279 MsDraining {} -> "draining")
269280
281+-- True when `s` is non-empty lowercase hex of even length, with `maxLen` ceiling.
270282isValidBoundedHex : Int -> Text -> Bool
271283isValidBoundedHex = \maxLen s -> not tooLong s maxLen && isValidHex s
284+-- True when `s` is lowercase hex of exactly `exactLen` characters (even by definition).
272285isValidExactHex : Int -> Text -> Bool
273286isValidExactHex = \exactLen s -> count (chars s) == exactLen && isValidHex s
287+-- Hex-encoded Ed25519 public key (32 bytes → 64 chars exact).
274288type PubkeyHex = Text where isValidExactHex pubkeyLen
289+-- Hex-encoded ciphertext or opaque blob; capped at `maxBlobLen` (~20 MiB).
275290type BlobHex = Text where isValidBoundedHex maxBlobLen
291+-- Hex-encoded Ed25519 signature (64 bytes → 128 chars exact).
276292type SignatureHex = Text where isValidExactHex signatureLen
277293isValidMessageId : Text -> Bool
278294isValidMessageId = isValidBoundedHex maxMessageIdLen
295+-- Hex-encoded message id, length ≤ `maxMessageIdLen`.
279296type MessageIdHex = Text where isValidMessageId
297+-- A `MessageIdHex` or the empty string. Empty means "let the server generate one".
280298type OptionalMessageId = Text where \s -> s == "" || isValidMessageId s
299+-- Normalized hostname (lowercased, no trailing dot, length-bounded).
281300isAnyServerName : Text -> Bool
282301isAnyServerName = \s -> not tooLong s maxServerNameLen && s == normalizeServerName s
302+-- A normalized hostname that also passes the SSRF deny-list (`isBadServerName`).
283303isValidServerName : Text -> Bool
284304isValidServerName = \s -> isAnyServerName s && not isBadServerName s
305+-- Peer hostname suitable for outbound federation calls.
285306type ServerName = Text where isValidServerName
307+-- Normalized hostname with no SSRF filter; used when the value comes from
308+-- presence rows that the client only ever sees masked to "remote".
286309type AnyServerName = Text where isAnyServerName
287310
311+-- Wall-clock time in milliseconds since epoch.
288312type Timestamp = Int<Ms>
289313
314+-- Outstanding nonce issued by /auth/challenge; consumed by /auth/verify.
290315type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp, ip: Text}
316+-- An issued bearer token bound to (pubkey, ip, expiry).
291317type Session = {
292318 token: Text,
293319 pubkey: PubkeyHex,
@@ -295,6 +321,10 @@ type Session = {
295321 ip: Text,
296322 createdAt: Timestamp
297323}
324+-- Mailbox row: a stored message awaiting recipient pickup or federation forward.
325+-- `originServer` is "" for messages submitted directly by a local sender.
326+-- `senderKey` is "" for messages received via federation (we don't see the
327+-- remote sender's pubkey).
298328type Message = {
299329 messageId: MessageIdHex,
300330 toKey: PubkeyHex,
@@ -305,12 +335,16 @@ type Message = {
305335 expiresAt: Timestamp,
306336 senderKey: Text
307337}
338+-- Last-known location of a pubkey, either local (`server == serverHostN`) or
339+-- learned from a peer via gossip. Multiple entries per pubkey are allowed.
308340type PresenceEntry = {
309341 pubkey: PubkeyHex,
310342 server: AnyServerName,
311343 expiresAt: Timestamp,
312344 timestamp: Timestamp
313345}
346+-- A message queued for cross-server delivery to `toServer`. `retries` is
347+-- incremented by `retryPendingForwards` until `maxForwardRetries` is hit.
314348type ForwardEntry = {
315349 messageId: MessageIdHex,
316350 toServer: ServerName,
@@ -323,6 +357,8 @@ type ForwardEntry = {
323357 expiresAt: Timestamp,
324358 senderKey: Text
325359}
360+-- Record that a federated message has been acked locally; flushed back to
361+-- `originServer` by `sendAcksAndClean`.
326362type AckedEntry = {
327363 messageId: MessageIdHex,
328364 toKey: PubkeyHex,
@@ -330,23 +366,33 @@ type AckedEntry = {
330366 originServer: ServerName,
331367 retries: Int
332368}
369+-- Replay-protection record: a message id we already delivered locally.
333370type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
371+-- Running failure counter for a peer; keyed by `server`. Drives `retryBackoff`.
334372type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFailedAt: Timestamp}
373+-- One fixed-window rate counter. `key` identifies the subject (pubkey, ip,
374+-- server, or composite "sender|recipient" string).
335375type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
376+-- A live long-poll slot held by a pubkey; used to cap concurrency per key.
336377type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
337378
379+-- Pending notification to a sender that their queued message expired unacked.
380+-- Drained in the sender's next /poll response.
338381type SenderExpiration = {
339382 senderKey: Text,
340383 messageId: MessageIdHex,
341384 toKey: PubkeyHex,
342385 expiredAt: Timestamp
343386}
387+-- Pending notification to a sender that their message was delivered to the
388+-- recipient. Drained in the sender's next /poll response.
344389type SenderDelivery = {
345390 senderKey: Text,
346391 messageId: MessageIdHex,
347392 toKey: PubkeyHex,
348393 deliveredAt: Timestamp
349394}
395+-- One entry of a /messages/batch request body.
350396type BatchMessage = {
351397 to: PubkeyHex,
352398 encryptedBlob: BlobHex,
@@ -354,6 +400,7 @@ type BatchMessage = {
354400 messageId: OptionalMessageId,
355401 expiresInMs: Int
356402}
403+-- Per-entry result returned by /messages/batch.
357404type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
358405
359406-- ============================================
@@ -480,6 +527,8 @@ route MaintenanceApi where
480527route HealthApi where
481528 GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, forwardQueueHealthy: Bool, requestPressureHealthy: Bool, draining: Bool, degradedPeers: Int} = Health
482529
530+-- Per-peer health snapshot for /stats/peers. Latency/gossip-rate fields are
531+-- zero-filled in this simplified server.
483532type PeerStatus = {
484533 server: Text,
485534 failures: Int,
@@ -493,7 +542,9 @@ type PeerStatus = {
493542 latencySamples: Int,
494543 gossipEventsPerMin: Int
495544}
545+-- Single presence row exposed via /stats/key (server + expiry only).
496546type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
547+-- Mailbox depth ranking entry for /stats/top-recipients.
497548type RecipientCount = {pubkey: PubkeyHex, count: Int}
498549
499550route StatsApi where
@@ -564,6 +615,7 @@ unique = fold (\acc x -> if elem acc x then acc else union acc [x]) []
564615withIndex : [a] -> [{i: Int, v: a}]
565616withIndex = fold (\acc x -> union acc [{i: count acc, v: x}]) []
566617
618+-- First element of a list, or Nothing for an empty list.
567619headOf : [a] -> Maybe a
568620headOf = \items -> fold (\_ x -> Just {value: x}) (Nothing {}) (takeRelation 1 items)
569621
@@ -648,6 +700,7 @@ else do
648700 rest <- genTokenN (n - 1)
649701 yield (take 1 (drop d "0123456789abcdef") ++ rest)
650702
703+-- Standard 48-char hex token used for session bearers and federation nonces.
651704genToken : IO {random} Text
652705genToken = genTokenN 48
653706
@@ -655,30 +708,39 @@ genToken = genTokenN 48
655708resolveMessageId : OptionalMessageId -> IO {random} OptionalMessageId
656709resolveMessageId = \mid -> if mid != "" then yield mid else genToken
657710
711+-- True when `s` exceeds `maxLen` characters.
658712tooLong : Text -> Int -> Bool
659713tooLong = \s maxLen -> count (chars s) > maxLen
660714
715+-- True when `s` is non-empty, even-length, and only lowercase hex digits.
661716isValidHex : Text -> Bool
662717isValidHex = \s -> (\cs -> (\n -> n > 0 && n / 2 * 2 == n && all cs (\c -> contains c "0123456789abcdef")) (count cs)) (chars s)
663718
719+-- Recursively strip every trailing "." from `s` (DNS root form).
664720stripTrailingDot : Text -> Text
665721stripTrailingDot = \s -> if containsSuffix "." s
666722then stripTrailingDot (take (count (chars s) - 1) s)
667723else s
668724
725+-- Canonical form of a hostname: lowercase, no trailing dot.
669726normalizeServerName : Text -> Text
670727normalizeServerName = \s -> toLower (stripTrailingDot s)
671728
729+-- This server's own hostname in canonical form (used everywhere for self-checks).
672730serverHostN : Text
673731serverHostN = normalizeServerName serverHost
732+-- True when `s` refers to this server.
674733isLocalServer : Text -> Bool
675734isLocalServer = \s -> s == serverHostN
735+-- True when `s` refers to any server other than this one.
676736isRemote : Text -> Bool
677737isRemote = \s -> not isLocalServer s
678738
739+-- True when `s` starts with `pre`.
679740hasPrefix : Text -> Text -> Bool
680741hasPrefix = \pre s -> take (count (chars pre)) s == pre
681742
743+-- True when `s` ends with `suffix`.
682744containsSuffix : Text -> Text -> Bool
683745containsSuffix = \suffix s -> drop (max 0 (count (chars s) - count (chars suffix))) s == suffix
684746
@@ -710,16 +772,22 @@ isBadServerName = \sn -> sn == "" || sn == "localhost" || not all (chars sn) (\c
710772matchIp : Text -> Text -> Bool
711773matchIp = \clientIp sessionIp -> sessionIp == "" || clientIp == sessionIp
712774
775+-- Pull the token out of an `Authorization: Bearer <token>` header. Returns
776+-- Nothing if the prefix is missing or the token is empty.
713777extractToken : Text -> Maybe Text
714778extractToken = \h -> if hasPrefix "Bearer " h && drop 7 h != ""
715779then Just {value: drop 7 h}
716780else Nothing {}
717781
782+-- True when the header carries a Bearer token equal to the runtime maintenance
783+-- secret. Plain `==` comparison; not constant-time (see file header).
718784verifyMaintenanceAuth : Text -> IO {r *runtimeMaintenanceSecret} Bool
719785verifyMaintenanceAuth = \authorization -> do
720786 secret <- *runtimeMaintenanceSecret
721787 yield (secret != "" && extractToken authorization == Just {value: secret})
722788
789+-- Branch on maintenance-secret auth: run `onAuth` when the bearer matches,
790+-- `onUnauth` otherwise.
723791withMaintAuth : Text -> IO {| effs} a -> IO {| effs} a -> IO {r *runtimeMaintenanceSecret | effs} a
724792withMaintAuth = \authorization onUnauth onAuth -> do
725793 authorized <- verifyMaintenanceAuth authorization
@@ -748,6 +816,8 @@ gateClientWrite = \auth ip t respond mkErr rateResp onAuth ->
748816 yield r.limited
749817 if lim then yield (respond rateResp) else onAuth pk)
750818
819+-- Variant of `gateClientWrite` that skips the drain check (lookups still work
820+-- while draining) and uses the per-key lookup-rate window instead.
751821gateLookup : Text -> Text -> Int<Ms> -> (a -> b) -> (Text -> a) -> a -> (Text -> IO {} b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions} b
752822gateLookup = \auth ip t respond mkErr rateResp onAuth ->
753823 gateAuth (False {}) auth ip t respond mkErr (\pk -> do
@@ -759,6 +829,8 @@ gateLookup = \auth ip t respond mkErr rateResp onAuth ->
759829 yield r.limited
760830 if lim then yield (respond rateResp) else onAuth pk)
761831
832+-- Resolve a Bearer token to its session pubkey if the session is unexpired and
833+-- the client IP matches. Runs `onAuth pubkey` on success, `onUnauth` otherwise.
762834withSessionAuth : Text -> Int<Ms> -> Text -> IO {| e} a -> (PubkeyHex -> IO {| e} a) -> IO {r *sessions | e} a
763835withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
764836 ses <- *sessions
@@ -768,6 +840,10 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
768840 Just {value: s} -> onAuth s.pubkey
769841 _ -> onUnauth
770842
843+-- Build the `Signed <hexsig>:<nonce>` header for an outbound federation call.
844+-- Signs the canonical "scope|timestamp|fromServer|nonce" string with our
845+-- federation private key (the peer fetches our public key via /federation/pubkey
846+-- to verify). Returns "" if no keypair has been generated yet.
771847generateFederationAuth : Text -> a -> IO {r *federationPrivateKeyHex, random} Text
772848generateFederationAuth = \scope requestTimestamp -> do
773849 hex <- *federationPrivateKeyHex
@@ -802,6 +878,11 @@ verifyFederationAuth = \authorization fromServer scope requestTimestamp -> do
802878 _ -> yield False {}
803879 else yield False {}
804880
881+-- Combined gate for inbound federation calls. Order of checks (each short-
882+-- circuits with an error code in the response):
883+-- 1. peer not block-listed; 2. peer in allow-list (if any); 3. caller-supplied
884+-- `precondition` holds; 4. clock skew within `clockSkewTolerance`;
885+-- 5. Ed25519 signature verifies; 6. per-peer federation rate limit.
805886withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do
806887 blocked <- *blockedPeers
807888 allowed <- *allowedPeers
@@ -842,6 +923,8 @@ checkRate = \entries key maxRate windowMs t -> (\cur -> if cur.count > maxRate t
842923 Just {value: r} -> {r.windowStart, count: r.count + 1}
843924 _ -> {windowStart: t, count: 1})
844925
926+-- Single-counter global rate gate. Resets the counter when the window expires;
927+-- returns True when the limit has been hit (caller responds rate_limited).
845928checkGlobalRate : Timestamp -> IO {rw *globalRateCount, rw *globalRateWindowStart} Bool
846929checkGlobalRate = \t -> atomic (do
847930 ws <- *globalRateWindowStart
@@ -857,6 +940,9 @@ checkGlobalRate = \t -> atomic (do
857940 *globalRateCount = c + 1
858941 yield (False {}))
859942
943+-- Same shape as `checkGlobalRate` but charged only by the auth endpoints
944+-- (challenge + verify). Stops a single attacker from exhausting the global
945+-- counter via auth churn.
860946checkAuthGlobalRate = \t -> atomic (do
861947 ws <- *authGlobalRateWindowStart
862948 c <- *authGlobalRateCount
@@ -871,6 +957,7 @@ checkAuthGlobalRate = \t -> atomic (do
871957 *authGlobalRateCount = c + 1
872958 yield (False {}))
873959
960+-- Per-source-IP rate gate (skipped when the IP is unknown — see `requireClientIp`).
874961checkIpRateGate : Text -> Int<Ms> -> IO {rw *ipRates} Bool
875962checkIpRateGate = \ip t -> if ip == ""
876963then yield (False {})
@@ -880,6 +967,7 @@ else atomic (do
880967 replace *ipRates = r.entries
881968 yield r.limited)
882969
970+-- Per-peer federation rate gate; charged inside `withFederationAuth`.
883971checkServerRateGate : Text -> Int -> IO {rw *federationRates} Bool
884972checkServerRateGate = \server t -> atomic (do
885973 rates <- *federationRates
@@ -908,12 +996,17 @@ authPreCheck = \clientIp t -> do
908996-- single read-side gate; `recordFailure`/`clearFailure` are the write side.
909997-- There is no half-open state — a single success clears all accumulated failures.
910998
999+-- Linear backoff: `retryBackoffBase * failures`, capped at `retryBackoffMax`.
9111000retryBackoff : Int -> Int<Ms>
9121001retryBackoff = \failures -> min retryBackoffMax (retryBackoffBase * max 1 failures)
9131002
1003+-- True when enough time has passed since the last failure for `server` to be
1004+-- worth retrying (or no failure recorded at all).
9141005shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
9151006shouldRetryServer = \server t failures -> none failures (\f -> f.server == server && t - f.failedAt < retryBackoff f.failures)
9161007
1008+-- Bump the failure counter for `server`. Preserves `firstFailedAt` across
1009+-- subsequent failures so health views can show the outage age.
9171010recordFailure : Text -> IO {rw *forwardFailures, clock} {}
9181011recordFailure = \server -> do
9191012 t <- now
@@ -927,6 +1020,8 @@ recordFailure = \server -> do
9271020 {server, failedAt: t, prev.failures, prev.firstFailedAt}
9281021 ])
9291022
1023+-- Drop the failure record for `server` (called after any successful peer call).
1024+-- Single-success recovery — there is no half-open state.
9301025clearFailure : Text -> IO {rw *forwardFailures} {}
9311026clearFailure = \server -> atomic (do
9321027 forwardFailures <- *forwardFailures
@@ -939,23 +1034,33 @@ clearFailure = \server -> atomic (do
9391034-- monotonic sequence generation, presence lookup, duplicate detection across
9401035-- (messages, ackedIds, deliveredIds), and last-write-wins presence merge.
9411036
1037+-- Clamp a client-supplied TTL to (0, maxPresenceTtl].
9421038clampTtl : Int<Ms> -> Int<Ms>
9431039clampTtl = clamp 1 maxPresenceTtl
9441040
1041+-- Clamp a client-supplied message timestamp into [now - messageMaxAge, now].
1042+-- Stops senders from claiming arbitrary future or far-past timestamps.
9451043clampTimestamp : Timestamp -> Timestamp -> Timestamp
9461044clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
9471045
1046+-- Resolve the effective TTL for a stored message: 0/negative means default,
1047+-- otherwise clamp the requested value into [minMessageTtl, unackedMessageMaxAge].
9481048effectiveMessageTtl : Int<Ms> -> Int<Ms>
9491049effectiveMessageTtl = \e -> if e <= 0
9501050then unackedMessageMaxAge
9511051else clamp minMessageTtl unackedMessageMaxAge e
9521052
1053+-- Strictly-monotonic next sequence timestamp: max of wall clock and prior+1.
1054+-- Guarantees /poll cursors order events even when many arrive in the same ms.
9531055nextSeqTs : Timestamp -> Timestamp -> Timestamp
9541056nextSeqTs = \t seqCounter -> max t (seqCounter + 1)
9551057
1058+-- Drop duplicates and cap to `maxMessageIds` (defense against bloated requests).
9561059dedupMessageIds : [MessageIdHex] -> [MessageIdHex]
9571060dedupMessageIds = \ids -> takeRelation maxMessageIds (unique ids)
9581061
1062+-- All unexpired presence rows for a given pubkey (may include both local and
1063+-- remote entries — the recipient may live on multiple servers simultaneously).
9591064presenceForKey : [PresenceEntry] -> PubkeyHex -> Timestamp -> [PresenceEntry]
9601065presenceForKey = \presence pubkey t -> filter presence (\p -> p.pubkey == pubkey && p.expiresAt > t)
9611066
@@ -963,15 +1068,22 @@ presenceForKey = \presence pubkey t -> filter presence (\p -> p.pubkey == pubkey
9631068maskedServer : [PresenceEntry] -> PubkeyHex -> Timestamp -> Text
9641069maskedServer = \pres pubkey t -> (\here -> if any here (\p -> isLocalServer p.server) then serverHostN else if count here > 0 then "remote" else "") (presenceForKey pres pubkey t)
9651070
1071+-- True when any row in `items` has a matching (messageId, toKey) pair.
9661072hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool
9671073hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)
9681074
1075+-- Replay-protection check: True if the (messageId, toKey) is already present in
1076+-- the live mailbox, the federated-acked log, or the delivered log.
9691077isDuplicateMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> [{messageId: MessageIdHex, toKey: PubkeyHex | b}] -> [{messageId: MessageIdHex, toKey: PubkeyHex | c}] -> Bool
9701078isDuplicateMessage = \mid toKey msgs aids dids -> hasMessage mid toKey msgs || hasMessage mid toKey aids || hasMessage mid toKey dids
9711079
1080+-- Project items through `getServer`, drop our own hostname, and dedupe.
1081+-- Used to derive peer recipient lists from forwards/presence/etc.
9721082collectPeers : [a] -> (a -> Text) -> [Text]
9731083collectPeers = \items getServer -> unique (filter (pluck items getServer) isRemote)
9741084
1085+-- All distinct remote hostnames currently appearing in `presence` (the gossip
1086+-- broadcast set we know about, less ourselves).
9751087knownPeerServersFromPresence : [PresenceEntry] -> [Text]
9761088knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.server)
9771089
@@ -986,6 +1098,9 @@ mergePresence = \existing incoming -> fold (\acc e -> (\same -> if any acc (\x -
9861098-- block. Returns the locally-expired presence rows whose pubkey has no other
9871099-- live local session, so the caller can broadcast offline gossip for them.
9881100
1101+-- One sweep over every state ref: drops expired rows in a single atomic block.
1102+-- Returns the locally-expired presence rows whose pubkey has no other live
1103+-- local session, so the caller can broadcast offline gossip for them.
9891104pruneExpired : Int<Ms> -> IO {rw *ackedIds, rw *activePolls, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, rw *senderDeliveries, rw *senderExpirations, rw *sessions} [{expiresAt: Timestamp, pubkey: PubkeyHex, server: AnyServerName, timestamp: Timestamp}]
9901105pruneExpired = \t -> atomic do
9911106 ch <- *challenges
@@ -1054,6 +1169,8 @@ recordAcksAndDeliveries = \toAck t -> do
10541169 *senderDeliveries = capList (union sds myDel) 10000 (\d -> d.deliveredAt)
10551170 yield ackedFed
10561171
1172+-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.
1173+-- Records or clears federation-failure counters per call.
10571174sendBatchedGossip : [Text] -> [GossipEvent] -> IO {} {}
10581175sendBatchedGossip = \peers events -> forEach peers (\server -> do
10591176 t <- now
@@ -1072,6 +1189,8 @@ sendBatchedGossip = \peers events -> forEach peers (\server -> do
10721189 Err _ -> recordFailure server
10731190 Ok _ -> clearFailure server))
10741191
1192+-- Tell known peers that the listed local pubkeys just went offline. No-op when
1193+-- federation is disabled or the list is empty.
10751194broadcastOfflineGossip : Int<Ms> -> [{pubkey: PubkeyHex | a}] -> IO {r *presence} {}
10761195broadcastOfflineGossip = \t expiredLocal -> do
10771196 presence <- *presence
@@ -1088,6 +1207,9 @@ broadcastOfflineGossip = \t expiredLocal -> do
10881207-- comes online: any messages already buffered for that pubkey are enqueued
10891208-- as forwards to the newly-known server.
10901209
1210+-- Walk pending forwards grouped by destination, retry each (up to maxBatchSize
1211+-- per peer) using `/federation/forward`. Drops successful entries; for failures
1212+-- bumps the retry count and gives up after `maxForwardRetries`.
10911213retryPendingForwards : IO {r *blockedPeers, r *forwardFailures, r *forwards, clock} {}
10921214retryPendingForwards = do
10931215 t <- now
@@ -1135,6 +1257,8 @@ retryPendingForwards = do
11351257 }
11361258 else x))))
11371259
1260+-- Flush federated ack receipts back to each origin server in one call, then
1261+-- prune the locally-recorded acked entries the peer confirmed.
11381262sendAcksAndClean : [{messageId: MessageIdHex, originServer: Text | a}] -> IO {} {}
11391263sendAcksAndClean = \pendingAcks -> forEach (collectPeers pendingAcks (\a -> a.originServer)) (\server -> do
11401264 let mids = mapWhere pendingAcks (\a -> a.originServer == server) (\a -> a.messageId)
@@ -1183,6 +1307,9 @@ forwardQueuedToServer = \pubkey server t -> atomic (do
11831307-- expire stale rows, then opportunistically retry pending forwards and refresh
11841308-- local presence so peers don't see us go offline mid-session.
11851309
1310+-- One composite background tick: prune expired state, gossip offline for
1311+-- locally-expired keys, retry pending forwards, flush federation acks, and
1312+-- (occasionally) refresh local presence so peers don't see us drop off.
11861313pruneAndRetry = \t -> do
11871314 *lastPrunedAt = t
11881315 expiredLocal <- pruneExpired t
@@ -1196,6 +1323,9 @@ pruneAndRetry = \t -> do
11961323 *lastPresenceRefresh = t
11971324 refreshLocalPresence t))
11981325
1326+-- Extend the expiry of local presence rows whose backing session is still live
1327+-- but the presence entry is past half its window. Re-broadcasts an `online`
1328+-- event for the refreshed entries so peers don't time them out.
11991329refreshLocalPresence : Int<Ms> -> IO {rw *presence, r *sessions} {}
12001330refreshLocalPresence = \t -> do
12011331 refreshState <- atomic (do
@@ -2103,6 +2233,7 @@ handleUnblockPeer = \server authorization respond -> withMaintAuth authorization
21032233 yield (elem current server))
21042234 yield (respond {ok: True {}, unblocked: removed}))
21052235
2236+-- GET /maintenance/blocked: list the deny-listed peers (empty for unauthorized).
21062237handleListBlocked : Text -> ({ok: Bool, peers: [Text]} -> a) -> IO {r *blockedPeers, r *runtimeMaintenanceSecret} a
21072238handleListBlocked = \authorization respond -> do
21082239 ok <- verifyMaintenanceAuth authorization
@@ -2137,6 +2268,7 @@ handleUnallowPeer = \server authorization respond -> withMaintAuth authorization
21372268 yield (elem current server))
21382269 yield (respond {ok: True {}, removed}))
21392270
2271+-- GET /maintenance/allowed: list the allow-listed peers (empty for unauthorized).
21402272handleListAllowed : Text -> ({ok: Bool, peers: [Text]} -> a) -> IO {r *allowedPeers, r *runtimeMaintenanceSecret} a
21412273handleListAllowed = \authorization respond -> do
21422274 ok <- verifyMaintenanceAuth authorization
@@ -2428,6 +2560,9 @@ handleStats = \authorization respond -> withMaintAuth authorization (yield (resp
24282560 }
24292561 }))
24302562
2563+-- All-zero /stats payload returned to unauthorized callers. The shape is kept
2564+-- in lockstep with the response in `handleStats` so older operator tooling
2565+-- never sees a missing field.
24312566zeroStats : {capacity: {ackOverflow: Int, ackOverflowMax: Int, ackedIds: Int, ackedIdsMax: Int, activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, senderDeliveries: Int, senderExpirations: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text}
24322567zeroStats = {
24332568 ok: False {},
@@ -2672,6 +2807,8 @@ handleMetrics = \authorization respond -> do
26722807-- to the corresponding `handleXxx` function. Adding a new route means adding
26732808-- a `route ... = NewReq` line, a `handleNewReq` definition, and a clause here.
26742809
2810+-- Top-level request dispatcher: maps each generated request constructor to
2811+-- its `handleXxx` function. Every route entry above must have a clause here.
26752812serve = \req -> case req of
26762813 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
26772814 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
@@ -2732,6 +2869,8 @@ serve = \req -> case req of
27322869-- `pollTimeoutMs` so long-poll holders unblock and return an empty heartbeat
27332870-- response, keeping connections fresh.
27342871
2872+-- Endless tick: every `backgroundPruneInterval` run `pruneAndRetry` (skipped
2873+-- when drain phase >= 2) and bump `pruneCount` for /stats and /health.
27352874backgroundPrune : IO {r *drainPhase, rw *ackedIds, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *pruneCount, rw *senderDeliveries, rw *senderExpirations, rw *sessions, clock} a
27362875backgroundPrune = do
27372876 sleep backgroundPruneInterval
@@ -2742,6 +2881,8 @@ backgroundPrune = do
27422881 pruneAndRetry t)
27432882 backgroundPrune
27442883
2884+-- Endless tick that bumps `pollWakeup` every `pollTimeoutMs` so any blocked
2885+-- /poll holders unblock and return an empty heartbeat response.
27452886pollHeartbeat : IO {rw *heartbeatCount, rw *lastHeartbeatAt, rw *pollWakeup, clock} a
27462887pollHeartbeat = do
27472888 sleep pollTimeoutMs
@@ -2756,6 +2897,8 @@ pollHeartbeat = do
27562897-- server restarts and reset their poll cursor), spawn the two background
27572898-- loops, and start the listener.
27582899
2900+-- Entry point: stamp startup metadata, lazily generate the federation keypair
2901+-- on first run, fork the background loops, and start the HTTP listener.
27592902main : IO {| _} {}
27602903main = do
27612904 t <- now