@@ -246,10 +246,7 @@ impl FromJSON GossipEventType where
246246 parseJson t = (\s -> if s == "online" then GossipOnline {} else if s == "offline" then GossipOffline {} else GossipUnknown {}) (parseJson t : Text)
247247
248248-- One presence change exchanged between federated servers.
249−type GossipEvent = {
250− eventType: GossipEventType,
251− pubkey: PubkeyHex
252−}
249+type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex}
253250
254251-- Outcome classification returned to the sender of /messages and per-entry in
255252-- /messages/batch. The first four are success-ish (see `isSendSuccess`); the
@@ -325,11 +322,7 @@ type Session = {
325322type Message = {toKey: PubkeyHex, encryptedBlob: BlobHex, receivedAt: Timestamp}
326323-- Last-known location of a pubkey, either local (`server == serverHostN`) or
327324-- learned from a peer via gossip. Multiple entries per pubkey are allowed.
328−type PresenceEntry = {
329− pubkey: PubkeyHex,
330− server: AnyServerName,
331− expiresAt: Timestamp
332−}
325+type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp}
333326-- A message queued for cross-server delivery to `toServer`. `retries` is
334327-- incremented by `retryPendingForwards` until `maxForwardRetries` is hit.
335328-- Identity is derived from `encryptedBlob` (see `Message`).
@@ -538,8 +531,6 @@ none : [a] -> (a -> Bool) -> Bool
538531none = \xs pred -> not any xs pred
539532all : [a] -> (a -> Bool) -> Bool
540533all = \xs pred -> none xs (\x -> not pred x)
541−elem : [a] -> a -> Bool
542−elem = \xs x -> any xs (\y -> y == x)
543534findFirst : [a] -> (a -> Bool) -> Maybe a
544535findFirst = \xs pred -> headOf (filter xs pred)
545536when : Bool -> IO {} {} -> IO {} {}
@@ -552,11 +543,6 @@ clamp : a -> a -> a -> a
552543clamp = \lo hi x -> max lo (min hi x)
553544abs : Int -> Int
554545abs = \x -> if x < 0 then 0 - x else x
555−
556−-- Dedup: keeps first occurrence (O(n^2)).
557−unique : [a] -> [a]
558−unique = fold (\acc x -> if elem acc x then acc else union acc [x]) []
559−
560546-- Pair each element with its 0-based index.
561547withIndex : [a] -> [{i: Int, v: a}]
562548withIndex = fold (\acc x -> union acc [{i: count acc, v: x}]) []
@@ -757,7 +743,7 @@ gateClientWrite = \auth ip t respond mkErr rateResp onAuth ->
757743 rates <- *clientWriteRates
758744 let r = checkRate rates pk maxClientWriteRatePerKey clientWriteRateWindowMs t
759745 -- TODO: make this no be a `replace`
760− replace *clientWriteRates = r.entries
746+ *clientWriteRates = r.entries
761747 yield r.limited
762748 if lim then yield (respond rateResp) else onAuth pk)
763749
@@ -770,7 +756,7 @@ gateLookup = \auth ip t respond mkErr rateResp onAuth ->
770756 rates <- *lookupRates
771757 let r = checkRate rates pk maxLookupRatePerKey lookupRateWindowMs t
772758 -- TODO: make this no be a `replace`
773− replace *lookupRates = r.entries
759+ *lookupRates = r.entries
774760 yield r.limited
775761 if lim then yield (respond rateResp) else onAuth pk)
776762
@@ -832,9 +818,9 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
832818 blocked <- *blockedPeers
833819 allowed <- *allowedPeers
834820 let err = \e r -> yield (respond {ok: False {}, error: e, retryAfterMs: r})
835− let preErr = if elem blocked fromServer
821+ let preErr = if elem fromServer blocked
836822 then "blocked"
837− else if count allowed > 0 && not elem allowed fromServer
823+ else if count allowed > 0 && not elem fromServer allowed
838824 then "not_allowed"
839825 else if not precondition
840826 then "invalid_request"
@@ -909,7 +895,7 @@ then yield (False {})
909895else atomic (do
910896 rates <- *ipRates
911897 let r = checkRate rates ip maxIpRatePerIp ipRateWindowMs t
912− replace *ipRates = r.entries
898+ *ipRates = r.entries
913899 yield r.limited)
914900
915901-- Per-peer federation rate gate; charged inside `withFederationAuth`.
@@ -917,7 +903,7 @@ checkServerRateGate : Text -> Int -> IO {rw *federationRates} Bool
917903checkServerRateGate = \server t -> atomic (do
918904 rates <- *federationRates
919905 let r = checkRate rates server maxFederationRatePerServer federationRateWindowMs t
920− replace *federationRates = r.entries
906+ *federationRates = r.entries
921907 yield r.limited)
922908
923909-- Pre-auth gate: drain + global + ip + auth-global. Returns "" when ok, else an error code.
@@ -986,7 +972,7 @@ nextSeqTs = \t seqCounter -> max t (seqCounter + 1)
986972
987973-- Drop duplicates and cap to `maxMessageIds` (defense against bloated requests).
988974dedupMessageIds : [MessageIdHex] -> [MessageIdHex]
989−dedupMessageIds = \ids -> takeRelation maxMessageIds (unique ids)
975+dedupMessageIds = \ids -> takeRelation maxMessageIds ids
990976
991977-- All unexpired presence rows for a given pubkey (may include both local and
992978-- remote entries — the recipient may live on multiple servers simultaneously).
@@ -1009,7 +995,7 @@ isDuplicateMessage = \blob toKey msgs dids -> any msgs (\m -> m.toKey == toKey &
1009995-- Project items through `getServer`, drop our own hostname, and dedupe.
1010996-- Used to derive peer recipient lists from forwards/presence/etc.
1011997collectPeers : [a] -> (a -> Text) -> [Text]
1012−collectPeers = \items getServer -> unique (filter (pluck items getServer) isRemote)
998+collectPeers = \items getServer -> filter (pluck items getServer) isRemote
1013999
10141000-- All distinct remote hostnames currently appearing in `presence` (the gossip
10151001-- broadcast set we know about, less ourselves).
@@ -1057,10 +1043,7 @@ pruneExpired = \t -> atomic do
10571043
10581044-- Build a gossip event with the originating pubkey's status.
10591045mkGossip : GossipEventType -> PubkeyHex -> GossipEvent
1060−mkGossip = \etype pk -> {
1061− eventType: etype,
1062− pubkey: pk
1063−}
1046+mkGossip = \etype pk -> {eventType: etype, pubkey: pk}
10641047
10651048-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.
10661049-- Records or clears federation-failure counters per call.
@@ -1106,7 +1089,7 @@ retryPendingForwards = do
11061089 fwds <- *forwards
11071090 failures <- *forwardFailures
11081091 blocked <- *blockedPeers
1109− forEach (filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures && not elem blocked s)) (\server -> forEach (takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))) (\f -> do
1092+ forEach (filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures && not elem s blocked)) (\server -> forEach (takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))) (\f -> do
11101093 t2 <- now
11111094 let messageId = messageIdOf f.encryptedBlob
11121095 auth <- generateFederationAuth ("forward:" ++ messageId) t2
@@ -1178,15 +1161,12 @@ refreshLocalPresence = \t -> do
11781161 refreshState <- atomic (do
11791162 sessions <- *sessions
11801163 presence <- *presence
1181− let sessionKeys = pluck sessions (\s -> s.pubkey)
1182− let needsRefresh = \p -> isLocalServer p.server && elem sessionKeys p.pubkey && p.expiresAt - t < sessionExpiry / 2
1183− let refreshed = mapWhere presence needsRefresh (\p -> {
1184− p.pubkey,
1185− p.server,
1186− expiresAt: t + sessionExpiry
1187− })
1188− let updated = union (removeWhere presence needsRefresh) refreshed
1189− replace *presence = updated
1164+ let sessionKeys = map (\s -> s.pubkey) sessions
1165+ let needsRefresh = \p -> isLocalServer p.server && elem p.pubkey sessionKeys && p.expiresAt - t < sessionExpiry / 2
1166+ let kept = filter presence (\p -> not needsRefresh p)
1167+ let refreshed = map (\p -> {p.pubkey, p.server, expiresAt: t + sessionExpiry}) (filter presence needsRefresh)
1168+ let updated = union kept refreshed
1169+ *presence = updated
11901170 yield {refreshed, peers: knownPeerServersFromPresence updated})
11911171 when (count refreshState.refreshed > 0) (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> mkGossip (GossipOnline {}) p.pubkey))))
11921172
@@ -1225,11 +1205,11 @@ handleChallenge = \pubkey clientIp respond -> do
12251205 else do
12261206 chRates <- *challengeRates
12271207 let chR = checkRate chRates pubkey maxChallengeRatePerKey challengeRateWindowMs t
1228− replace *challengeRates = chR.entries
1208+ *challengeRates = chR.entries
12291209 if chR.limited
12301210 then yield {ok: False {}, error: "rate_limited"}
12311211 else do
1232− replace *challenges = union cappedActive [
1212+ *challenges = union cappedActive [
12331213 {pubkey, challenge, expiresAt: t + challengeExpiry, ip: clientIp}
12341214 ]
12351215 yield {ok: True {}, error: ""})
@@ -1274,13 +1254,11 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
12741254 then removeWhere ses mine
12751255 else capPerKey ses mine maxSessionsPerKey (\s -> s.expiresAt)
12761256 let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
1277− replace *sessions = capList (union pruned [newSession]) maxSessions (\s -> s.expiresAt)
1257+ *sessions = capList (union pruned [newSession]) maxSessions (\s -> s.expiresAt)
12781258 pres <- *presence
12791259 let prunedPres = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
1280− let newPres = union prunedPres [
1281− {pubkey, server: serverHostN, expiresAt: expiry}
1282− ]
1283− replace *presence = newPres
1260+ let newPres = union prunedPres [{pubkey, server: serverHostN, expiresAt: expiry}]
1261+ *presence = newPres
12841262 yield {valid: True {}, peers: knownPeerServersFromPresence newPres})
12851263 if not result.valid
12861264 then yield (respond {token: "", expiresAt: 0, error: "invalid"})
@@ -1429,7 +1407,7 @@ ackPolledMessages = \pubkey cursor -> do
14291407 let blobs = pluck toAck (\m -> m.encryptedBlob)
14301408 *messages = removeWhere msgs matches
14311409 fwds <- *forwards
1432− *forwards = filter fwds (\f -> not (f.toKey == pubkey && elem blobs f.encryptedBlob))
1410+ *forwards = filter fwds (\f -> not (f.toKey == pubkey && elem f.encryptedBlob blobs))
14331411
14341412-- ============================================
14351413-- Handler: send
@@ -1482,15 +1460,15 @@ sendMessageCore = \senderKey to encryptedBlob t -> do
14821460 dids <- *deliveredIds
14831461 sRates <- *sendRates
14841462 let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t
1485− replace *sendRates = sR.entries
1463+ *sendRates = sR.entries
14861464 let srLim = sR.limited
14871465 pRates <- *pairRates
14881466 let pR = checkRate pRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t
1489− replace *pairRates = pR.entries
1467+ *pairRates = pR.entries
14901468 let prLim = pR.limited
14911469 rRates <- *recvRates
14921470 let rR = checkRate rRates to maxRecvRatePerKey recvRateWindowMs t
1493− replace *recvRates = rR.entries
1471+ *recvRates = rR.entries
14941472 let rrLim = rR.limited
14951473 let reject = \s -> yield {status: s, remoteServers: []}
14961474 if senderKey == to
@@ -1560,7 +1538,7 @@ forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
15601538 atomic (do
15611539 fwds2 <- *forwards
15621540 let remaining = filter fwds2 (\x -> not same x)
1563− replace *forwards = remaining
1541+ *forwards = remaining
15641542 pres <- *presence
15651543 let stillLocal = any pres (\p -> p.pubkey == to && isLocalServer p.server && p.expiresAt > t)
15661544 let othersPending = any remaining (\x -> x.encryptedBlob == encryptedBlob)
@@ -1647,7 +1625,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
16471625 error: "rate_limited"
16481626 } (\pubkey -> do
16491627 ackedCount <- atomic (do
1650− toAck <- removeForKey (\x -> x.toKey == pubkey && elem (dedupMessageIds messageIds) (messageIdOf x.encryptedBlob))
1628+ toAck <- removeForKey (\x -> x.toKey == pubkey && elem (messageIdOf x.encryptedBlob) (dedupMessageIds messageIds))
16511629 yield (count toAck))
16521630 yield (respond {ok: True {}, authorized: True {}, ackedCount, error: ""}))
16531631
@@ -1666,7 +1644,7 @@ handleListMessages = \offset authorization clientIp respond -> do
16661644 lim <- atomic (do
16671645 rates <- *lookupRates
16681646 let r = checkRate rates pubkey maxLookupRatePerKey lookupRateWindowMs t
1669− replace *lookupRates = r.entries
1647+ *lookupRates = r.entries
16701648 yield r.limited)
16711649 if lim
16721650 then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})
@@ -1698,7 +1676,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
16981676 error: "rate_limited"
16991677 } (\pubkey -> do
17001678 deletedCount <- atomic (do
1701− toDel <- removeForKey (\x -> x.toKey == pubkey && elem (dedupMessageIds messageIds) (messageIdOf x.encryptedBlob))
1679+ toDel <- removeForKey (\x -> x.toKey == pubkey && elem (messageIdOf x.encryptedBlob) (dedupMessageIds messageIds))
17021680 yield (count toDel))
17031681 yield (respond {ok: True {}, authorized: True {}, deletedCount, error: ""}))
17041682
@@ -1734,10 +1712,10 @@ handleFederationPubkey = \respond -> do
17341712 yield (respond {pubkey: hex, ok: hex != ""})
17351713
17361714-- POST /federation/gossip: ingest peer presence events. The transport-level
1737−-- `fromServer` is the home server for every event. Applies each event in
1738−-- arrival order: `GossipOnline` upserts the (pubkey, fromServer) row with a
1739−-- fresh receiver-side TTL, `GossipOffline` deletes it. For keys that
1740−-- transition offline → online, flushes any locally-buffered messages via
1715+-- `fromServer` is the home server for every event. Drops every (pubkey,
1716+-- fromServer) row touched by this batch, then re-adds fresh-TTL rows for
1717+-- pubkeys with a `GossipOnline` event (so Online wins on collision). For keys
1718+-- that transition offline → online, flushes any locally-buffered messages via
17411719-- `forwardQueuedToServer`.
17421720handleRecvGossip : [{eventType: <GossipOffline {} | GossipOnline {} | a>, pubkey: PubkeyHex | b}] -> Text -> Int -> Text -> ({error: Text, ok: Bool, retryAfterMs: Int} -> c) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, rw *presence, clock, network} c
17431721handleRecvGossip = \events fromServer requestTimestamp authorization respond -> do
@@ -1745,21 +1723,15 @@ handleRecvGossip = \events fromServer requestTimestamp authorization respond ->
17451723 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (isRemote fromServer && count events <= maxGossipEvents) (do
17461724 newlyOnline <- atomic (do
17471725 pres <- *presence
1748− let final = fold (\state e ->
1749− (\matches -> case e.eventType of
1750− GossipOnline {} -> {
1751− pres: union (removeWhere state.pres matches) [{e.pubkey, server: fromServer, expiresAt: t + onlineGossipTtl}],
1752− newlyOnline: if any state.pres (\p -> matches p && p.expiresAt > t) then state.newlyOnline else union state.newlyOnline [{e.pubkey, server: fromServer}]
1753− }
1754− GossipOffline {} -> {
1755− pres: removeWhere state.pres matches,
1756− state.newlyOnline
1757− }
1758− _ -> state) (\p -> p.pubkey == e.pubkey && p.server == fromServer)
1759− ) {pres, newlyOnline: []} events
1760− *presence = final.pres
1761− yield final.newlyOnline)
1762− fork (forEach newlyOnline (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t))
1726+ let touchedPubkeys = pluck events (\e -> e.pubkey)
1727+ let onlinePubkeys = pluck (filter events (\e -> case e.eventType of
1728+ GossipOnline {} -> True {}
1729+ _ -> False {})) (\e -> e.pubkey)
1730+ let kept = filter pres (\p -> p.server != fromServer || not elem p.pubkey touchedPubkeys)
1731+ let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys
1732+ *presence = union kept added
1733+ yield (filter onlinePubkeys (\pk -> none pres (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t))))
1734+ fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer t))
17631735 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
17641736
17651737-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.
@@ -1785,7 +1757,7 @@ storeFederated = \toKey encryptedBlob checkRecvRate t -> do
17851757 then do
17861758 rRates <- *recvRates
17871759 let r = checkRate rRates toKey maxRecvRatePerKey recvRateWindowMs t
1788− replace *recvRates = r.entries
1760+ *recvRates = r.entries
17891761 yield r.limited
17901762 else yield (False {})
17911763 if limited
@@ -1940,8 +1912,8 @@ handleBlockPeer = \server authorization respond -> withMaintAuth authorization (
19401912})) (do
19411913 n <- atomic (do
19421914 current <- *blockedPeers
1943− let updated = unique (union current [server])
1944− replace *blockedPeers = updated
1915+ let updated = union current [server]
1916+ *blockedPeers = updated
19451917 yield (count updated))
19461918 yield (respond {ok: True {}, blocked: n}))
19471919
@@ -1954,7 +1926,7 @@ handleUnblockPeer = \server authorization respond -> withMaintAuth authorization
19541926 removed <- atomic (do
19551927 current <- *blockedPeers
19561928 *blockedPeers = removeWhere current (\x -> x == server)
1957− yield (elem current server))
1929+ yield (elem server current))
19581930 yield (respond {ok: True {}, unblocked: removed}))
19591931
19601932-- GET /maintenance/blocked: list the deny-listed peers (empty for unauthorized).
@@ -1974,8 +1946,8 @@ handleAllowPeer = \server authorization respond -> withMaintAuth authorization (
19741946})) (do
19751947 n <- atomic (do
19761948 current <- *allowedPeers
1977− let updated = unique (union current [server])
1978− replace *allowedPeers = updated
1949+ let updated = union current [server]
1950+ *allowedPeers = updated
19791951 yield (count updated))
19801952 yield (respond {ok: True {}, allowed: n}))
19811953
@@ -1989,7 +1961,7 @@ handleUnallowPeer = \server authorization respond -> withMaintAuth authorization
19891961 removed <- atomic (do
19901962 current <- *allowedPeers
19911963 *allowedPeers = removeWhere current (\x -> x == server)
1992− yield (elem current server))
1964+ yield (elem server current))
19931965 yield (respond {ok: True {}, removed}))
19941966
19951967-- GET /maintenance/allowed: list the allow-listed peers (empty for unauthorized).
@@ -2082,7 +2054,7 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization (
20822054 let same = \s -> s.ip == ip
20832055 *sessions = removeWhere ses same
20842056 yield (mapWhere ses same (\s -> s.pubkey)))
2085− yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count (unique keys)}))
2057+ yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count keys}))
20862058
20872059-- POST /maintenance/sessions/list: paged session inventory aggregated by
20882060-- pubkey (count, distinct ips, latest expiry). Sorted by descending count.
@@ -2095,7 +2067,7 @@ handleAdminListAllSessions = \limit authorization respond -> withMaintAuth autho
20952067 t <- now
20962068 ses <- *sessions
20972069 let active = filter ses (\s -> s.expiresAt > t)
2098− let grouped = pluck (unique (pluck active (\s -> s.pubkey))) (\pk -> (\mine -> {pubkey: pk, count: count mine, ips: unique (mapWhere mine (\s -> s.ip != "") (\s -> s.ip)), latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine}) (filter active (\s -> s.pubkey == pk)))
2070+ let grouped = pluck (pluck active (\s -> s.pubkey)) (\pk -> (\mine -> {pubkey: pk, count: count mine, ips: mapWhere mine (\s -> s.ip != "") (\s -> s.ip), latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine}) (filter active (\s -> s.pubkey == pk)))
20992071 let cap = clamp 1 500 (if limit <= 0 then 100 else limit)
21002072 let top = takeRelation cap (sortBy (\g -> 0 - g.count) grouped)
21012073 yield (respond {sessions: top, ok: True {}, total: count grouped}))
@@ -2151,11 +2123,7 @@ handleGetConfig : Text -> ({ackGracePeriod: Int, ok: Bool, pollTimeoutMs: Int} -
21512123handleGetConfig = \authorization respond -> do
21522124 ok <- verifyMaintenanceAuth authorization
21532125 let v = \x -> if ok then x else 0
2154− yield (respond {
2155− ok,
2156− pollTimeoutMs: v pollTimeoutMs,
2157− ackGracePeriod: v ackGracePeriod
2158− })
2126+ yield (respond {ok, pollTimeoutMs: v pollTimeoutMs, ackGracePeriod: v ackGracePeriod})
21592127
21602128-- GET /stats: capacity gauges + minimal federation/timing snapshot. Most of
21612129-- the historical metric fields are zero-filled (see `zeroStats`) — only the
@@ -2435,7 +2403,7 @@ handleTopRecipients = \limit authorization respond -> withMaintAuth authorizatio
24352403 total: 0
24362404})) (do
24372405 msgs <- *messages
2438− let counts = pluck (unique (pluck msgs (\m -> m.toKey))) (\pk -> {
2406+ let counts = pluck (pluck msgs (\m -> m.toKey)) (\pk -> {
24392407 pubkey: pk,
24402408 count: countWhere msgs (\m -> m.toKey == pk)
24412409 })