@@ -312,26 +312,11 @@ route Api = AuthApi | ClientApi | FederationApi
312-- Everything is intentionally O(n) over flat lists; performance is sacrificed
312-- Everything is intentionally O(n) over flat lists; performance is sacrificed
313-- for clarity per the file header.
313-- for clarity per the file header.
315−-- True when every element matches.
316−findFirst : [a] -> (a -> Bool) -> Maybe a
317−findFirst = \xs pred -> headOf (filter pred xs)
318-- Run `action` only when `cond` is True; otherwise no-op.
315-- 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 {}
322−min = \a b -> if a < b then a else b
324−max = \a b -> if a > b then a else b
325-- Confine `x` to the closed interval [lo, hi].
316-- Confine `x` to the closed interval [lo, hi].
326−clamp : a -> a -> a -> a
317+clamp : Ord a => a -> a -> a -> a
327clamp = \lo hi x -> max lo (min hi x)
318clamp = \lo hi x -> max lo (min hi x)
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}
335-- Cap to maxN by dropping the smallest-key entries (LRU-style). Always sorts.
320-- Cap to maxN by dropping the smallest-key entries (LRU-style). Always sorts.
336-- For session/challenge/rate caps, `getKey` is typically `expiresAt` or
321-- For session/challenge/rate caps, `getKey` is typically `expiresAt` or
337-- `windowStart` so the oldest entries are evicted first.
322-- `windowStart` so the oldest entries are evicted first.
@@ -347,7 +332,7 @@ capPerKey = \items pred maxN getKey -> union (filter (\x -> not pred x) items) (
348-- Sequence an IO action over each item, in order. Recurses through the list
333-- Sequence an IO action over each item, in order. Recurses through the list
349-- one element at a time; this is the closest equivalent of `mapM_` here.
334-- 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 {}
351forEach = \items action -> case items of
336forEach = \items action -> case items of
352 [] -> yield {}
337 [] -> yield {}
353 Cons x rest -> do
338 Cons x rest -> do
@@ -570,7 +555,7 @@ retryBackoff = \failures -> min retryBackoffMax (retryBackoffBase * max 1 failur
570-- True when enough time has passed since the last failure for `server` to be
555-- True when enough time has passed since the last failure for `server` to be
571-- worth retrying (or no failure recorded at all).
556-- worth retrying (or no failure recorded at all).
572shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
557shouldRetryServer : 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
575-- Bump the failure counter for `server`. Preserves `firstFailedAt` across
560-- Bump the failure counter for `server`. Preserves `firstFailedAt` across
576-- subsequent failures.
561-- subsequent failures.
@@ -655,7 +640,7 @@ pruneExpired = \t -> atomic do
656-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.
641-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.
657-- Records or clears federation-failure counters per call.
642-- Records or clears federation-failure counters per call.
658−sendBatchedGossip : [Text] -> [GossipEventType] -> IO {} {}
643+sendBatchedGossip : [Text] -> [GossipEventType] -> IO {rw *forwardFailures, network, clock} {}
659sendBatchedGossip = \peers events -> forEach peers (\server -> do
644sendBatchedGossip = \peers events -> forEach peers (\server -> do
661 failures <- *forwardFailures
646 failures <- *forwardFailures
@@ -667,7 +652,7 @@ sendBatchedGossip = \peers events -> forEach peers (\server -> do
668-- Tell known peers that the listed local pubkeys just went offline. No-op when
653-- Tell known peers that the listed local pubkeys just went offline. No-op when
669-- federation is disabled or the list is empty.
654-- 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} {}
671broadcastOfflineGossip = \expiredLocal -> do
656broadcastOfflineGossip = \expiredLocal -> do
672 presence <- *presence
657 presence <- *presence
673 when (federationEnabled && count expiredLocal > 0) (sendBatchedGossip (knownPeerServersFromPresence presence) (map (\p -> GossipOffline {
658 when (federationEnabled && count expiredLocal > 0) (sendBatchedGossip (knownPeerServersFromPresence presence) (map (\p -> GossipOffline {
@@ -686,7 +671,7 @@ broadcastOfflineGossip = \expiredLocal -> do
686-- Walk pending forwards grouped by destination, retry each (up to maxBatchSize
671-- Walk pending forwards grouped by destination, retry each (up to maxBatchSize
687-- per peer) using `/federation/forward`. Drops successful entries; for failures
672-- per peer) using `/federation/forward`. Drops successful entries; for failures
688-- bumps the retry count and gives up after `maxForwardRetries`.
673-- 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} {}
690retryPendingForwards = do
675retryPendingForwards = do
692 fwds <- *forwards
677 fwds <- *forwards
@@ -713,13 +698,8 @@ forwardQueuedToServer : PubkeyHex -> ServerName -> IO {rw *forwards, r *messages
713forwardQueuedToServer = \pubkey server -> atomic (do
698forwardQueuedToServer = \pubkey server -> atomic (do
714 msgs <- *messages
699 msgs <- *messages
715 fwds <- *forwards
700 fwds <- *forwards
716− let candidates = filter (\m -> m.toKey == pubkey && not (any (\f -> f.encryptedBlob == m.encryptedBlob && f.toServer == server) fwds)) msgs
701+ let candidates = filter (\m -> m.toKey == pubkey && not any (\f -> f.encryptedBlob == m.encryptedBlob && f.toServer == server) fwds) msgs
717− let newFwds = map (\m -> {
722− }) (take maxForwardsPerServer candidates)
702+ let newFwds = map (\m -> {toServer: server, m.toKey, m.encryptedBlob, retries: 0}) (take maxForwardsPerServer candidates)
723 *forwards = union fwds newFwds)
703 *forwards = union fwds newFwds)
725-- ============================================
705-- ============================================
@@ -783,7 +763,7 @@ handleChallenge = \pubkey clientIp -> do
783-- On success, mints a session token, records local presence (`pubkey` is now
763-- On success, mints a session token, records local presence (`pubkey` is now
784-- live on this server), and gossips an `online` event to known peers. If
764-- live on this server), and gossips an `online` event to known peers. If
785-- `revokeOthers` is set, all other sessions for the same pubkey are dropped.
765-- `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})
787handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
767handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
789 preErr <- authPreCheck clientIp t
769 preErr <- authPreCheck clientIp t
@@ -833,7 +813,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
834-- POST /poll: top-level dispatch. Validates the global rate gate, then hands
814-- POST /poll: top-level dispatch. Validates the global rate gate, then hands
835-- off to `pollOnce` for the actual long-poll body.
815-- 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]})
837handlePoll = \cursor authorization clientIp -> do
817handlePoll = \cursor authorization clientIp -> do
839 g <- checkGlobalRate t
819 g <- checkGlobalRate t
@@ -845,7 +825,7 @@ handlePoll = \cursor authorization clientIp -> do
845-- (`ackPolledMessages`), refreshes the caller's local presence row so peers
825-- (`ackPolledMessages`), refreshes the caller's local presence row so peers
846-- don't time us out mid-session, then runs `doPoll` to collect events (or
826-- don't time us out mid-session, then runs `doPoll` to collect events (or
847-- block until wakeup) and assembles the response.
827-- 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]})
849pollOnce = \authorization cursor clientIp t -> do
829pollOnce = \authorization cursor clientIp t -> do
850 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
830 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
851 let clampedCursor = clamp 0 t cursor
831 let clampedCursor = clamp 0 t cursor
@@ -853,7 +833,7 @@ pollOnce = \authorization cursor clientIp t -> do
853 startWakeup <- *pollWakeup
833 startWakeup <- *pollWakeup
854 pres <- *presence
834 pres <- *presence
855 let mine = \p -> p.pubkey == pubkey && isLocalServer p.server
835 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
857 refreshPeers <- if stale
837 refreshPeers <- if stale
859 let kept = filter (\p -> not mine p) pres
839 let kept = filter (\p -> not mine p) pres
@@ -988,12 +968,7 @@ commitSendMessage = \to encryptedBlob remoteServers t -> do
988 *pollWakeup = seqTs
968 *pollWakeup = seqTs
989 when federationEnabled (do
969 when federationEnabled (do
990 fwds <- *forwards
970 fwds <- *forwards
991− let newFwds = map (\s -> {
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)
997 *forwards = union fwds newFwds)
972 *forwards = union fwds newFwds)
999-- Fire one /federation/forward call per pre-staged server. On success the
974-- Fire one /federation/forward call per pre-staged server. On success the
@@ -1002,7 +977,7 @@ commitSendMessage = \to encryptedBlob remoteServers t -> do
1002-- `*messages` row is dropped too: the origin's job is done and the message
977-- `*messages` row is dropped too: the origin's job is done and the message
1003-- now lives on the receiving peer's mailbox. On failure the forwards row
978-- now lives on the receiving peer's mailbox. On failure the forwards row
1004-- stays so `retryPendingForwards` can re-attempt with backoff.
979-- 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} {}
1006forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
981forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
1007 fwds <- *forwards
982 fwds <- *forwards
1008 let same = \x -> x.encryptedBlob == encryptedBlob && x.toServer == server
983 let same = \x -> x.encryptedBlob == encryptedBlob && x.toServer == server
@@ -1039,7 +1014,7 @@ forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
1039-- pubkeys with a `GossipOnline` event (so Online wins on collision). For keys
1014-- pubkeys with a `GossipOnline` event (so Online wins on collision). For keys
1040-- that transition offline → online, flushes any locally-buffered messages via
1015-- that transition offline → online, flushes any locally-buffered messages via
1041-- `forwardQueuedToServer`.
1016-- `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 {})
1043handleRecvGossip = \events fromServer -> do
1018handleRecvGossip = \events fromServer -> do
1044 t <- now
1019 t <- now
1045 if not isRemote fromServer || isBadServerName fromServer || count events > maxGossipEvents
1020 if not isRemote fromServer || isBadServerName fromServer || count events > maxGossipEvents
@@ -1062,7 +1037,7 @@ handleRecvGossip = \events fromServer -> do
1062 let kept = filter (\p -> p.server != fromServer || not elem p.pubkey touchedPubkeys) pres
1037 let kept = filter (\p -> p.server != fromServer || not elem p.pubkey touchedPubkeys) pres
1063 let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys
1038 let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys
1064 *presence = union kept added
1039 *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))
1066 fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer))
1041 fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer))
1067 yield (Ok {value: {}})
1042 yield (Ok {value: {}})
@@ -1078,7 +1053,7 @@ storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWa
1078storeFederated = \toKey encryptedBlob t -> do
1053storeFederated = \toKey encryptedBlob t -> do
1079 pres <- *presence
1054 pres <- *presence
1080 msgs <- *messages
1055 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
1082 then yield "no_presence"
1057 then yield "no_presence"
1084 seq <- *seqCounter
1059 seq <- *seqCounter
@@ -1128,7 +1103,7 @@ api = serve Api where
1128-- Endless tick: prune immediately on first call, then every
1103-- Endless tick: prune immediately on first call, then every
1129-- `backgroundPruneInterval`. The startup prune clears stale rows persisted
1104-- `backgroundPruneInterval`. The startup prune clears stale rows persisted
1130-- by the previous process before the next interval elapses.
1105-- 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
1132backgroundPrune = do
1107backgroundPrune = do
1133 t <- now
1108 t <- now
1134 pruneAndRetry t
1109 pruneAndRetry t