@@ -312,26 +312,11 @@ route Api = AuthApi | ClientApi | FederationApi
312312-- Everything is intentionally O(n) over flat lists; performance is sacrificed
313313-- for clarity per the file header.
314314
315−-- True when every element matches.
316−findFirst : [a] -> (a -> Bool) -> Maybe a
317−findFirst = \xs pred -> headOf (filter pred xs)
318315-- Run `action` only when `cond` is True; otherwise no-op.
319−when : Bool -> IO {} {} -> IO {} {}
320−when = \cond action -> if cond then action else yield {}
321−min : a -> a -> a
322−min = \a b -> if a < b then a else b
323−max : a -> a -> a
324−max = \a b -> if a > b then a else b
325316-- Confine `x` to the closed interval [lo, hi].
326−clamp : a -> a -> a -> a
317+clamp : Ord a => a -> a -> a -> a
327318clamp = \lo hi x -> max lo (min hi x)
328319
329−-- First element of a list, or Nothing for an empty list.
330−headOf : [a] -> Maybe a
331−headOf = \items -> case items of
332− Cons x _ -> Just {value: x}
333− [] -> Nothing {}
334−
335320-- Cap to maxN by dropping the smallest-key entries (LRU-style). Always sorts.
336321-- For session/challenge/rate caps, `getKey` is typically `expiresAt` or
337322-- `windowStart` so the oldest entries are evicted first.
@@ -347,7 +332,7 @@ capPerKey = \items pred maxN getKey -> union (filter (\x -> not pred x) items) (
347332
348333-- Sequence an IO action over each item, in order. Recurses through the list
349334-- one element at a time; this is the closest equivalent of `mapM_` here.
350−forEach : [a] -> (a -> IO {} {}) -> IO {} {}
335+forEach : [a] -> (a -> IO eff {}) -> IO eff {}
351336forEach = \items action -> case items of
352337 [] -> yield {}
353338 Cons x rest -> do
@@ -570,7 +555,7 @@ retryBackoff = \failures -> min retryBackoffMax (retryBackoffBase * max 1 failur
570555-- True when enough time has passed since the last failure for `server` to be
571556-- worth retrying (or no failure recorded at all).
572557shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
573−shouldRetryServer = \server t failures -> not (any (\f -> f.server == server && t - f.failedAt < retryBackoff f.failures) failures)
558+shouldRetryServer = \server t failures -> not any (\f -> f.server == server && t - f.failedAt < retryBackoff f.failures) failures
574559
575560-- Bump the failure counter for `server`. Preserves `firstFailedAt` across
576561-- subsequent failures.
@@ -655,7 +640,7 @@ pruneExpired = \t -> atomic do
655640
656641-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.
657642-- Records or clears federation-failure counters per call.
658−sendBatchedGossip : [Text] -> [GossipEventType] -> IO {} {}
643+sendBatchedGossip : [Text] -> [GossipEventType] -> IO {rw *forwardFailures, network, clock} {}
659644sendBatchedGossip = \peers events -> forEach peers (\server -> do
660645 t <- now
661646 failures <- *forwardFailures
@@ -667,7 +652,7 @@ sendBatchedGossip = \peers events -> forEach peers (\server -> do
667652
668653-- Tell known peers that the listed local pubkeys just went offline. No-op when
669654-- federation is disabled or the list is empty.
670−broadcastOfflineGossip : [{pubkey: PubkeyHex | a}] -> IO {r *presence} {}
655+broadcastOfflineGossip : [{pubkey: PubkeyHex | a}] -> IO {r *presence, rw *forwardFailures, network, clock} {}
671656broadcastOfflineGossip = \expiredLocal -> do
672657 presence <- *presence
673658 when (federationEnabled && count expiredLocal > 0) (sendBatchedGossip (knownPeerServersFromPresence presence) (map (\p -> GossipOffline {
@@ -686,7 +671,7 @@ broadcastOfflineGossip = \expiredLocal -> do
686671-- Walk pending forwards grouped by destination, retry each (up to maxBatchSize
687672-- per peer) using `/federation/forward`. Drops successful entries; for failures
688673-- bumps the retry count and gives up after `maxForwardRetries`.
689−retryPendingForwards : IO {r *forwardFailures, r *forwards, clock} {}
674+retryPendingForwards : IO {rw *forwardFailures, rw *forwards, network, clock} {}
690675retryPendingForwards = do
691676 t <- now
692677 fwds <- *forwards
@@ -713,13 +698,8 @@ forwardQueuedToServer : PubkeyHex -> ServerName -> IO {rw *forwards, r *messages
713698forwardQueuedToServer = \pubkey server -> atomic (do
714699 msgs <- *messages
715700 fwds <- *forwards
716− let candidates = filter (\m -> m.toKey == pubkey && not (any (\f -> f.encryptedBlob == m.encryptedBlob && f.toServer == server) fwds)) msgs
717− let newFwds = map (\m -> {
718− toServer: server,
719− m.toKey,
720− m.encryptedBlob,
721− retries: 0
722− }) (take maxForwardsPerServer candidates)
701+ let candidates = filter (\m -> m.toKey == pubkey && not any (\f -> f.encryptedBlob == m.encryptedBlob && f.toServer == server) fwds) msgs
702+ let newFwds = map (\m -> {toServer: server, m.toKey, m.encryptedBlob, retries: 0}) (take maxForwardsPerServer candidates)
723703 *forwards = union fwds newFwds)
724704
725705-- ============================================
@@ -783,7 +763,7 @@ handleChallenge = \pubkey clientIp -> do
783763-- On success, mints a session token, records local presence (`pubkey` is now
784764-- live on this server), and gossips an `online` event to known peers. If
785765-- `revokeOthers` is set, all other sessions for the same pubkey are dropped.
786−handleVerifyAuth : PubkeyHex -> Text -> Text -> Bool -> Text -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *challenges, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates, rw *presence, rw *sessions, clock, random} (Result HttpError {expiresAt: Int, token: Text})
766+handleVerifyAuth : PubkeyHex -> Text -> Text -> Bool -> Text -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *challenges, rw *forwardFailures, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates, rw *presence, rw *sessions, network, clock, random} (Result HttpError {expiresAt: Int, token: Text})
787767handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
788768 t <- now
789769 preErr <- authPreCheck clientIp t
@@ -833,7 +813,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
833813
834814-- POST /poll: top-level dispatch. Validates the global rate gate, then hands
835815-- off to `pollOnce` for the actual long-poll body.
836−handlePoll : Int -> Text -> Text -> IO {r *pollWakeup, r *sessions, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, rw *presence, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
816+handlePoll : Int -> Text -> Text -> IO {r *pollWakeup, r *sessions, rw *forwardFailures, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, rw *presence, network, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
837817handlePoll = \cursor authorization clientIp -> do
838818 t <- now
839819 g <- checkGlobalRate t
@@ -845,7 +825,7 @@ handlePoll = \cursor authorization clientIp -> do
845825-- (`ackPolledMessages`), refreshes the caller's local presence row so peers
846826-- don't time us out mid-session, then runs `doPoll` to collect events (or
847827-- block until wakeup) and assembles the response.
848−pollOnce : Text -> Int -> Text -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *forwards, rw *messages, rw *presence, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
828+pollOnce : Text -> Int -> Text -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, network, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent]})
849829pollOnce = \authorization cursor clientIp t -> do
850830 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
851831 let clampedCursor = clamp 0 t cursor
@@ -853,7 +833,7 @@ pollOnce = \authorization cursor clientIp t -> do
853833 startWakeup <- *pollWakeup
854834 pres <- *presence
855835 let mine = \p -> p.pubkey == pubkey && isLocalServer p.server
856− let stale = not (any (\p -> mine p && p.expiresAt - t >= sessionExpiry / 2) pres)
836+ let stale = not any (\p -> mine p && p.expiresAt - t >= sessionExpiry / 2) pres
857837 refreshPeers <- if stale
858838 then do
859839 let kept = filter (\p -> not mine p) pres
@@ -988,12 +968,7 @@ commitSendMessage = \to encryptedBlob remoteServers t -> do
988968 *pollWakeup = seqTs
989969 when federationEnabled (do
990970 fwds <- *forwards
991− let newFwds = map (\s -> {
992− toServer: s,
993− toKey: to,
994− encryptedBlob,
995− retries: 0
996− }) (filter (\s -> not (any (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s) fwds)) remoteServers)
971+ let newFwds = map (\s -> {toServer: s, toKey: to, encryptedBlob, retries: 0}) (filter (\s -> not any (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s) fwds) remoteServers)
997972 *forwards = union fwds newFwds)
998973
999974-- Fire one /federation/forward call per pre-staged server. On success the
@@ -1002,7 +977,7 @@ commitSendMessage = \to encryptedBlob remoteServers t -> do
1002977-- `*messages` row is dropped too: the origin's job is done and the message
1003978-- now lives on the receiving peer's mailbox. On failure the forwards row
1004979-- stays so `retryPendingForwards` can re-attempt with backoff.
1005−forkForwardsTo : PubkeyHex -> BlobHex -> [Text] -> IO {} {}
980+forkForwardsTo : PubkeyHex -> BlobHex -> [Text] -> IO {r *presence, rw *forwardFailures, rw *forwards, rw *messages, network, clock} {}
1006981forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
1007982 fwds <- *forwards
1008983 let same = \x -> x.encryptedBlob == encryptedBlob && x.toServer == server
@@ -1039,7 +1014,7 @@ forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
10391014-- pubkeys with a `GossipOnline` event (so Online wins on collision). For keys
10401015-- that transition offline → online, flushes any locally-buffered messages via
10411016-- `forwardQueuedToServer`.
1042−handleRecvGossip : [GossipEventType] -> Text -> IO {rw *presence, clock} (Result HttpError {})
1017+handleRecvGossip : [GossipEventType] -> Text -> IO {r *messages, rw *forwards, rw *presence, clock} (Result HttpError {})
10431018handleRecvGossip = \events fromServer -> do
10441019 t <- now
10451020 if not isRemote fromServer || isBadServerName fromServer || count events > maxGossipEvents
@@ -1062,7 +1037,7 @@ handleRecvGossip = \events fromServer -> do
10621037 let kept = filter (\p -> p.server != fromServer || not elem p.pubkey touchedPubkeys) pres
10631038 let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys
10641039 *presence = union kept added
1065− yield (filter (\pk -> not (any (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t) pres)) onlinePubkeys))
1040+ yield (filter (\pk -> not any (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t) pres) onlinePubkeys))
10661041 fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer))
10671042 yield (Ok {value: {}})
10681043
@@ -1078,7 +1053,7 @@ storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWa
10781053storeFederated = \toKey encryptedBlob t -> do
10791054 pres <- *presence
10801055 msgs <- *messages
1081− if not (any (\p -> p.pubkey == toKey && isLocalServer p.server && p.expiresAt > t) pres)
1056+ if not any (\p -> p.pubkey == toKey && isLocalServer p.server && p.expiresAt > t) pres
10821057 then yield "no_presence"
10831058 else do
10841059 seq <- *seqCounter
@@ -1128,7 +1103,7 @@ api = serve Api where
11281103-- Endless tick: prune immediately on first call, then every
11291104-- `backgroundPruneInterval`. The startup prune clears stale rows persisted
11301105-- by the previous process before the next interval elapses.
1131−backgroundPrune : IO {rw *challenges, rw *forwardFailures, r *forwards, rw *messages, rw *presence, rw *sessions, clock} a
1106+backgroundPrune : IO {rw *challenges, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, rw *sessions, network, clock} a
11321107backgroundPrune = do
11331108 t <- now
11341109 pruneAndRetry t