6af6f79 · skrepka · git.iko.soy

git.iko.soy/skrepka

Remove elem and unique helpers; simplify mutation syntax and gossip presence handling

6af6f79iko5 months ago

server.knot+55−87

@@ -246,10 +246,7 @@ impl FromJSON GossipEventType where

246 parseJson t = (\s -> if s == "online" then GossipOnline {} else if s == "offline" then GossipOffline {} else GossipUnknown {}) (parseJson t : Text)

247

248-- One presence change exchanged between federated servers.

249−type GossipEvent = {

250− eventType: GossipEventType,

249+type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex}

251− pubkey: PubkeyHex

252−}

253

250

254-- Outcome classification returned to the sender of /messages and per-entry in

251-- Outcome classification returned to the sender of /messages and per-entry in

255-- /messages/batch. The first four are success-ish (see `isSendSuccess`); the

252-- /messages/batch. The first four are success-ish (see `isSendSuccess`); the

@@ -325,11 +322,7 @@ type Session = {

325type Message = {toKey: PubkeyHex, encryptedBlob: BlobHex, receivedAt: Timestamp}

322type Message = {toKey: PubkeyHex, encryptedBlob: BlobHex, receivedAt: Timestamp}

326-- Last-known location of a pubkey, either local (`server == serverHostN`) or

323-- Last-known location of a pubkey, either local (`server == serverHostN`) or

327-- learned from a peer via gossip. Multiple entries per pubkey are allowed.

324-- learned from a peer via gossip. Multiple entries per pubkey are allowed.

328−type PresenceEntry = {

329− pubkey: PubkeyHex,

330− server: AnyServerName,

325+type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp}

331− expiresAt: Timestamp

332−}

333-- A message queued for cross-server delivery to `toServer`. `retries` is

326-- A message queued for cross-server delivery to `toServer`. `retries` is

334-- incremented by `retryPendingForwards` until `maxForwardRetries` is hit.

327-- incremented by `retryPendingForwards` until `maxForwardRetries` is hit.

335-- Identity is derived from `encryptedBlob` (see `Message`).

328-- Identity is derived from `encryptedBlob` (see `Message`).

@@ -538,8 +531,6 @@ none : [a] -> (a -> Bool) -> Bool

538none = \xs pred -> not any xs pred

531none = \xs pred -> not any xs pred

539all : [a] -> (a -> Bool) -> Bool

532all : [a] -> (a -> Bool) -> Bool

540all = \xs pred -> none xs (\x -> not pred x)

533all = \xs pred -> none xs (\x -> not pred x)

541−elem : [a] -> a -> Bool

542−elem = \xs x -> any xs (\y -> y == x)

543findFirst : [a] -> (a -> Bool) -> Maybe a

534findFirst : [a] -> (a -> Bool) -> Maybe a

544findFirst = \xs pred -> headOf (filter xs pred)

535findFirst = \xs pred -> headOf (filter xs pred)

545when : Bool -> IO {} {} -> IO {} {}

536when : Bool -> IO {} {} -> IO {} {}

@@ -552,11 +543,6 @@ clamp : a -> a -> a -> a

552clamp = \lo hi x -> max lo (min hi x)

543clamp = \lo hi x -> max lo (min hi x)

553abs : Int -> Int

544abs : Int -> Int

554abs = \x -> if x < 0 then 0 - x else x

545abs = \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−

560-- Pair each element with its 0-based index.

546-- Pair each element with its 0-based index.

561withIndex : [a] -> [{i: Int, v: a}]

547withIndex : [a] -> [{i: Int, v: a}]

562withIndex = fold (\acc x -> union acc [{i: count acc, v: x}]) []

548withIndex = fold (\acc x -> union acc [{i: count acc, v: x}]) []

@@ -757,7 +743,7 @@ gateClientWrite = \auth ip t respond mkErr rateResp onAuth ->

757 rates <- *clientWriteRates

743 rates <- *clientWriteRates

758 let r = checkRate rates pk maxClientWriteRatePerKey clientWriteRateWindowMs t

744 let r = checkRate rates pk maxClientWriteRatePerKey clientWriteRateWindowMs t

759 -- TODO: make this no be a `replace`

745 -- TODO: make this no be a `replace`

760− replace *clientWriteRates = r.entries

746+ *clientWriteRates = r.entries

761 yield r.limited

747 yield r.limited

762 if lim then yield (respond rateResp) else onAuth pk)

748 if lim then yield (respond rateResp) else onAuth pk)

763

749

@@ -770,7 +756,7 @@ gateLookup = \auth ip t respond mkErr rateResp onAuth ->

770 rates <- *lookupRates

756 rates <- *lookupRates

771 let r = checkRate rates pk maxLookupRatePerKey lookupRateWindowMs t

757 let r = checkRate rates pk maxLookupRatePerKey lookupRateWindowMs t

772 -- TODO: make this no be a `replace`

758 -- TODO: make this no be a `replace`

773− replace *lookupRates = r.entries

759+ *lookupRates = r.entries

774 yield r.limited

760 yield r.limited

775 if lim then yield (respond rateResp) else onAuth pk)

761 if lim then yield (respond rateResp) else onAuth pk)

776

762

@@ -832,9 +818,9 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp

832 blocked <- *blockedPeers

818 blocked <- *blockedPeers

833 allowed <- *allowedPeers

819 allowed <- *allowedPeers

834 let err = \e r -> yield (respond {ok: False {}, error: e, retryAfterMs: r})

820 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

836 then "blocked"

822 then "blocked"

837− else if count allowed > 0 && not elem allowed fromServer

823+ else if count allowed > 0 && not elem fromServer allowed

838 then "not_allowed"

824 then "not_allowed"

839 else if not precondition

825 else if not precondition

840 then "invalid_request"

826 then "invalid_request"

@@ -909,7 +895,7 @@ then yield (False {})

909else atomic (do

895else atomic (do

910 rates <- *ipRates

896 rates <- *ipRates

911 let r = checkRate rates ip maxIpRatePerIp ipRateWindowMs t

897 let r = checkRate rates ip maxIpRatePerIp ipRateWindowMs t

912− replace *ipRates = r.entries

898+ *ipRates = r.entries

913 yield r.limited)

899 yield r.limited)

914

900

915-- Per-peer federation rate gate; charged inside `withFederationAuth`.

901-- Per-peer federation rate gate; charged inside `withFederationAuth`.

@@ -917,7 +903,7 @@ checkServerRateGate : Text -> Int -> IO {rw *federationRates} Bool

917checkServerRateGate = \server t -> atomic (do

903checkServerRateGate = \server t -> atomic (do

918 rates <- *federationRates

904 rates <- *federationRates

919 let r = checkRate rates server maxFederationRatePerServer federationRateWindowMs t

905 let r = checkRate rates server maxFederationRatePerServer federationRateWindowMs t

920− replace *federationRates = r.entries

906+ *federationRates = r.entries

921 yield r.limited)

907 yield r.limited)

922

908

923-- Pre-auth gate: drain + global + ip + auth-global. Returns "" when ok, else an error code.

909-- 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)

986

972

987-- Drop duplicates and cap to `maxMessageIds` (defense against bloated requests).

973-- Drop duplicates and cap to `maxMessageIds` (defense against bloated requests).

988dedupMessageIds : [MessageIdHex] -> [MessageIdHex]

974dedupMessageIds : [MessageIdHex] -> [MessageIdHex]

989−dedupMessageIds = \ids -> takeRelation maxMessageIds (unique ids)

975+dedupMessageIds = \ids -> takeRelation maxMessageIds ids

990

976

991-- All unexpired presence rows for a given pubkey (may include both local and

977-- All unexpired presence rows for a given pubkey (may include both local and

992-- remote entries — the recipient may live on multiple servers simultaneously).

978-- 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 &

1009-- Project items through `getServer`, drop our own hostname, and dedupe.

995-- Project items through `getServer`, drop our own hostname, and dedupe.

1010-- Used to derive peer recipient lists from forwards/presence/etc.

996-- Used to derive peer recipient lists from forwards/presence/etc.

1011collectPeers : [a] -> (a -> Text) -> [Text]

997collectPeers : [a] -> (a -> Text) -> [Text]

1012−collectPeers = \items getServer -> unique (filter (pluck items getServer) isRemote)

998+collectPeers = \items getServer -> filter (pluck items getServer) isRemote

1013

999

1014-- All distinct remote hostnames currently appearing in `presence` (the gossip

1000-- All distinct remote hostnames currently appearing in `presence` (the gossip

1015-- broadcast set we know about, less ourselves).

1001-- broadcast set we know about, less ourselves).

@@ -1057,10 +1043,7 @@ pruneExpired = \t -> atomic do

1057

1043

1058-- Build a gossip event with the originating pubkey's status.

1044-- Build a gossip event with the originating pubkey's status.

1059mkGossip : GossipEventType -> PubkeyHex -> GossipEvent

1045mkGossip : GossipEventType -> PubkeyHex -> GossipEvent

1060−mkGossip = \etype pk -> {

1046+mkGossip = \etype pk -> {eventType: etype, pubkey: pk}

1061− eventType: etype,

1062− pubkey: pk

1063−}

1064

1047

1065-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.

1048-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.

1066-- Records or clears federation-failure counters per call.

1049-- Records or clears federation-failure counters per call.

@@ -1106,7 +1089,7 @@ retryPendingForwards = do

1106 fwds <- *forwards

1089 fwds <- *forwards

1107 failures <- *forwardFailures

1090 failures <- *forwardFailures

1108 blocked <- *blockedPeers

1091 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

1110 t2 <- now

1093 t2 <- now

1111 let messageId = messageIdOf f.encryptedBlob

1094 let messageId = messageIdOf f.encryptedBlob

1112 auth <- generateFederationAuth ("forward:" ++ messageId) t2

1095 auth <- generateFederationAuth ("forward:" ++ messageId) t2

@@ -1178,15 +1161,12 @@ refreshLocalPresence = \t -> do

1178 refreshState <- atomic (do

1161 refreshState <- atomic (do

1179 sessions <- *sessions

1162 sessions <- *sessions

1180 presence <- *presence

1163 presence <- *presence

1181− let sessionKeys = pluck sessions (\s -> s.pubkey)

1164+ let sessionKeys = map (\s -> s.pubkey) sessions

1182− let needsRefresh = \p -> isLocalServer p.server && elem sessionKeys p.pubkey && p.expiresAt - t < sessionExpiry / 2

1165+ let needsRefresh = \p -> isLocalServer p.server && elem p.pubkey sessionKeys && p.expiresAt - t < sessionExpiry / 2

1183− let refreshed = mapWhere presence needsRefresh (\p -> {

1166+ let kept = filter presence (\p -> not needsRefresh p)

1184− p.pubkey,

1185− p.server,

1186− expiresAt: t + sessionExpiry

1167+ let refreshed = map (\p -> {p.pubkey, p.server, expiresAt: t + sessionExpiry}) (filter presence needsRefresh)

1187− })

1188− let updated = union (removeWhere presence needsRefresh) refreshed

1168+ let updated = union kept refreshed

1189− replace *presence = updated

1169+ *presence = updated

1190 yield {refreshed, peers: knownPeerServersFromPresence updated})

1170 yield {refreshed, peers: knownPeerServersFromPresence updated})

1191 when (count refreshState.refreshed > 0) (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> mkGossip (GossipOnline {}) p.pubkey))))

1171 when (count refreshState.refreshed > 0) (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> mkGossip (GossipOnline {}) p.pubkey))))

1192

1172

@@ -1225,11 +1205,11 @@ handleChallenge = \pubkey clientIp respond -> do

1225 else do

1205 else do

1226 chRates <- *challengeRates

1206 chRates <- *challengeRates

1227 let chR = checkRate chRates pubkey maxChallengeRatePerKey challengeRateWindowMs t

1207 let chR = checkRate chRates pubkey maxChallengeRatePerKey challengeRateWindowMs t

1228− replace *challengeRates = chR.entries

1208+ *challengeRates = chR.entries

1229 if chR.limited

1209 if chR.limited

1230 then yield {ok: False {}, error: "rate_limited"}

1210 then yield {ok: False {}, error: "rate_limited"}

1231 else do

1211 else do

1232− replace *challenges = union cappedActive [

1212+ *challenges = union cappedActive [

1233 {pubkey, challenge, expiresAt: t + challengeExpiry, ip: clientIp}

1213 {pubkey, challenge, expiresAt: t + challengeExpiry, ip: clientIp}

1234 ]

1214 ]

1235 yield {ok: True {}, error: ""})

1215 yield {ok: True {}, error: ""})

@@ -1274,13 +1254,11 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->

1274 then removeWhere ses mine

1254 then removeWhere ses mine

1275 else capPerKey ses mine maxSessionsPerKey (\s -> s.expiresAt)

1255 else capPerKey ses mine maxSessionsPerKey (\s -> s.expiresAt)

1276 let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}

1256 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)

1278 pres <- *presence

1258 pres <- *presence

1279 let prunedPres = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)

1259 let prunedPres = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)

1280− let newPres = union prunedPres [

1281− {pubkey, server: serverHostN, expiresAt: expiry}

1260+ let newPres = union prunedPres [{pubkey, server: serverHostN, expiresAt: expiry}]

1282− ]

1283− replace *presence = newPres

1261+ *presence = newPres

1284 yield {valid: True {}, peers: knownPeerServersFromPresence newPres})

1262 yield {valid: True {}, peers: knownPeerServersFromPresence newPres})

1285 if not result.valid

1263 if not result.valid

1286 then yield (respond {token: "", expiresAt: 0, error: "invalid"})

1264 then yield (respond {token: "", expiresAt: 0, error: "invalid"})

@@ -1429,7 +1407,7 @@ ackPolledMessages = \pubkey cursor -> do

1429 let blobs = pluck toAck (\m -> m.encryptedBlob)

1407 let blobs = pluck toAck (\m -> m.encryptedBlob)

1430 *messages = removeWhere msgs matches

1408 *messages = removeWhere msgs matches

1431 fwds <- *forwards

1409 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))

1433

1411

1434-- ============================================

1412-- ============================================

1435-- Handler: send

1413-- Handler: send

@@ -1482,15 +1460,15 @@ sendMessageCore = \senderKey to encryptedBlob t -> do

1482 dids <- *deliveredIds

1460 dids <- *deliveredIds

1483 sRates <- *sendRates

1461 sRates <- *sendRates

1484 let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t

1462 let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t

1485− replace *sendRates = sR.entries

1463+ *sendRates = sR.entries

1486 let srLim = sR.limited

1464 let srLim = sR.limited

1487 pRates <- *pairRates

1465 pRates <- *pairRates

1488 let pR = checkRate pRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t

1466 let pR = checkRate pRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t

1489− replace *pairRates = pR.entries

1467+ *pairRates = pR.entries

1490 let prLim = pR.limited

1468 let prLim = pR.limited

1491 rRates <- *recvRates

1469 rRates <- *recvRates

1492 let rR = checkRate rRates to maxRecvRatePerKey recvRateWindowMs t

1470 let rR = checkRate rRates to maxRecvRatePerKey recvRateWindowMs t

1493− replace *recvRates = rR.entries

1471+ *recvRates = rR.entries

1494 let rrLim = rR.limited

1472 let rrLim = rR.limited

1495 let reject = \s -> yield {status: s, remoteServers: []}

1473 let reject = \s -> yield {status: s, remoteServers: []}

1496 if senderKey == to

1474 if senderKey == to

@@ -1560,7 +1538,7 @@ forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do

1560 atomic (do

1538 atomic (do

1561 fwds2 <- *forwards

1539 fwds2 <- *forwards

1562 let remaining = filter fwds2 (\x -> not same x)

1540 let remaining = filter fwds2 (\x -> not same x)

1563− replace *forwards = remaining

1541+ *forwards = remaining

1564 pres <- *presence

1542 pres <- *presence

1565 let stillLocal = any pres (\p -> p.pubkey == to && isLocalServer p.server && p.expiresAt > t)

1543 let stillLocal = any pres (\p -> p.pubkey == to && isLocalServer p.server && p.expiresAt > t)

1566 let othersPending = any remaining (\x -> x.encryptedBlob == encryptedBlob)

1544 let othersPending = any remaining (\x -> x.encryptedBlob == encryptedBlob)

@@ -1647,7 +1625,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do

1647 error: "rate_limited"

1625 error: "rate_limited"

1648 } (\pubkey -> do

1626 } (\pubkey -> do

1649 ackedCount <- atomic (do

1627 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))

1651 yield (count toAck))

1629 yield (count toAck))

1652 yield (respond {ok: True {}, authorized: True {}, ackedCount, error: ""}))

1630 yield (respond {ok: True {}, authorized: True {}, ackedCount, error: ""}))

1653

1631

@@ -1666,7 +1644,7 @@ handleListMessages = \offset authorization clientIp respond -> do

1666 lim <- atomic (do

1644 lim <- atomic (do

1667 rates <- *lookupRates

1645 rates <- *lookupRates

1668 let r = checkRate rates pubkey maxLookupRatePerKey lookupRateWindowMs t

1646 let r = checkRate rates pubkey maxLookupRatePerKey lookupRateWindowMs t

1669− replace *lookupRates = r.entries

1647+ *lookupRates = r.entries

1670 yield r.limited)

1648 yield r.limited)

1671 if lim

1649 if lim

1672 then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})

1650 then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})

@@ -1698,7 +1676,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do

1698 error: "rate_limited"

1676 error: "rate_limited"

1699 } (\pubkey -> do

1677 } (\pubkey -> do

1700 deletedCount <- atomic (do

1678 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))

1702 yield (count toDel))

1680 yield (count toDel))

1703 yield (respond {ok: True {}, authorized: True {}, deletedCount, error: ""}))

1681 yield (respond {ok: True {}, authorized: True {}, deletedCount, error: ""}))

1704

1682

@@ -1734,10 +1712,10 @@ handleFederationPubkey = \respond -> do

1734 yield (respond {pubkey: hex, ok: hex != ""})

1712 yield (respond {pubkey: hex, ok: hex != ""})

1735

1713

1736-- POST /federation/gossip: ingest peer presence events. The transport-level

1714-- POST /federation/gossip: ingest peer presence events. The transport-level

1737−-- `fromServer` is the home server for every event. Applies each event in

1715+-- `fromServer` is the home server for every event. Drops every (pubkey,

1738−-- arrival order: `GossipOnline` upserts the (pubkey, fromServer) row with a

1716+-- fromServer) row touched by this batch, then re-adds fresh-TTL rows for

1739−-- fresh receiver-side TTL, `GossipOffline` deletes it. For keys that

1717+-- pubkeys with a `GossipOnline` event (so Online wins on collision). For keys

1740−-- transition offline → online, flushes any locally-buffered messages via

1718+-- that transition offline → online, flushes any locally-buffered messages via

1741-- `forwardQueuedToServer`.

1719-- `forwardQueuedToServer`.

1742handleRecvGossip : [{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

1720handleRecvGossip : [{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

1743handleRecvGossip = \events fromServer requestTimestamp authorization respond -> do

1721handleRecvGossip = \events fromServer requestTimestamp authorization respond -> do

@@ -1745,21 +1723,15 @@ handleRecvGossip = \events fromServer requestTimestamp authorization respond ->

1745 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (isRemote fromServer && count events <= maxGossipEvents) (do

1723 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (isRemote fromServer && count events <= maxGossipEvents) (do

1746 newlyOnline <- atomic (do

1724 newlyOnline <- atomic (do

1747 pres <- *presence

1725 pres <- *presence

1748− let final = fold (\state e ->

1726+ let touchedPubkeys = pluck events (\e -> e.pubkey)

1749− (\matches -> case e.eventType of

1727+ let onlinePubkeys = pluck (filter events (\e -> case e.eventType of

1750− GossipOnline {} -> {

1728+ GossipOnline {} -> True {}

1729+ _ -> False {})) (\e -> e.pubkey)

1730+ let kept = filter pres (\p -> p.server != fromServer || not elem p.pubkey touchedPubkeys)

1751− pres: union (removeWhere state.pres matches) [{e.pubkey, server: fromServer, expiresAt: t + onlineGossipTtl}],

1731+ let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys

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

1732+ *presence = union kept added

1761− yield final.newlyOnline)

1733+ yield (filter onlinePubkeys (\pk -> none pres (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t))))

1762− fork (forEach newlyOnline (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t))

1734+ fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer t))

1763 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))

1735 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))

1764

1736

1765-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.

1737-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.

@@ -1785,7 +1757,7 @@ storeFederated = \toKey encryptedBlob checkRecvRate t -> do

1785 then do

1757 then do

1786 rRates <- *recvRates

1758 rRates <- *recvRates

1787 let r = checkRate rRates toKey maxRecvRatePerKey recvRateWindowMs t

1759 let r = checkRate rRates toKey maxRecvRatePerKey recvRateWindowMs t

1788− replace *recvRates = r.entries

1760+ *recvRates = r.entries

1789 yield r.limited

1761 yield r.limited

1790 else yield (False {})

1762 else yield (False {})

1791 if limited

1763 if limited

@@ -1940,8 +1912,8 @@ handleBlockPeer = \server authorization respond -> withMaintAuth authorization (

1940})) (do

1912})) (do

1941 n <- atomic (do

1913 n <- atomic (do

1942 current <- *blockedPeers

1914 current <- *blockedPeers

1943− let updated = unique (union current [server])

1915+ let updated = union current [server]

1944− replace *blockedPeers = updated

1916+ *blockedPeers = updated

1945 yield (count updated))

1917 yield (count updated))

1946 yield (respond {ok: True {}, blocked: n}))

1918 yield (respond {ok: True {}, blocked: n}))

1947

1919

@@ -1954,7 +1926,7 @@ handleUnblockPeer = \server authorization respond -> withMaintAuth authorization

1954 removed <- atomic (do

1926 removed <- atomic (do

1955 current <- *blockedPeers

1927 current <- *blockedPeers

1956 *blockedPeers = removeWhere current (\x -> x == server)

1928 *blockedPeers = removeWhere current (\x -> x == server)

1957− yield (elem current server))

1929+ yield (elem server current))

1958 yield (respond {ok: True {}, unblocked: removed}))

1930 yield (respond {ok: True {}, unblocked: removed}))

1959

1931

1960-- GET /maintenance/blocked: list the deny-listed peers (empty for unauthorized).

1932-- GET /maintenance/blocked: list the deny-listed peers (empty for unauthorized).

@@ -1974,8 +1946,8 @@ handleAllowPeer = \server authorization respond -> withMaintAuth authorization (

1974})) (do

1946})) (do

1975 n <- atomic (do

1947 n <- atomic (do

1976 current <- *allowedPeers

1948 current <- *allowedPeers

1977− let updated = unique (union current [server])

1949+ let updated = union current [server]

1978− replace *allowedPeers = updated

1950+ *allowedPeers = updated

1979 yield (count updated))

1951 yield (count updated))

1980 yield (respond {ok: True {}, allowed: n}))

1952 yield (respond {ok: True {}, allowed: n}))

1981

1953

@@ -1989,7 +1961,7 @@ handleUnallowPeer = \server authorization respond -> withMaintAuth authorization

1989 removed <- atomic (do

1961 removed <- atomic (do

1990 current <- *allowedPeers

1962 current <- *allowedPeers

1991 *allowedPeers = removeWhere current (\x -> x == server)

1963 *allowedPeers = removeWhere current (\x -> x == server)

1992− yield (elem current server))

1964+ yield (elem server current))

1993 yield (respond {ok: True {}, removed}))

1965 yield (respond {ok: True {}, removed}))

1994

1966

1995-- GET /maintenance/allowed: list the allow-listed peers (empty for unauthorized).

1967-- GET /maintenance/allowed: list the allow-listed peers (empty for unauthorized).

@@ -2082,7 +2054,7 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization (

2082 let same = \s -> s.ip == ip

2054 let same = \s -> s.ip == ip

2083 *sessions = removeWhere ses same

2055 *sessions = removeWhere ses same

2084 yield (mapWhere ses same (\s -> s.pubkey)))

2056 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}))

2086

2058

2087-- POST /maintenance/sessions/list: paged session inventory aggregated by

2059-- POST /maintenance/sessions/list: paged session inventory aggregated by

2088-- pubkey (count, distinct ips, latest expiry). Sorted by descending count.

2060-- pubkey (count, distinct ips, latest expiry). Sorted by descending count.

@@ -2095,7 +2067,7 @@ handleAdminListAllSessions = \limit authorization respond -> withMaintAuth autho

2095 t <- now

2067 t <- now

2096 ses <- *sessions

2068 ses <- *sessions

2097 let active = filter ses (\s -> s.expiresAt > t)

2069 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)))

2099 let cap = clamp 1 500 (if limit <= 0 then 100 else limit)

2071 let cap = clamp 1 500 (if limit <= 0 then 100 else limit)

2100 let top = takeRelation cap (sortBy (\g -> 0 - g.count) grouped)

2072 let top = takeRelation cap (sortBy (\g -> 0 - g.count) grouped)

2101 yield (respond {sessions: top, ok: True {}, total: count grouped}))

2073 yield (respond {sessions: top, ok: True {}, total: count grouped}))

@@ -2151,11 +2123,7 @@ handleGetConfig : Text -> ({ackGracePeriod: Int, ok: Bool, pollTimeoutMs: Int} -

2151handleGetConfig = \authorization respond -> do

2123handleGetConfig = \authorization respond -> do

2152 ok <- verifyMaintenanceAuth authorization

2124 ok <- verifyMaintenanceAuth authorization

2153 let v = \x -> if ok then x else 0

2125 let v = \x -> if ok then x else 0

2154− yield (respond {

2155− ok,

2156− pollTimeoutMs: v pollTimeoutMs,

2157− ackGracePeriod: v ackGracePeriod

2126+ yield (respond {ok, pollTimeoutMs: v pollTimeoutMs, ackGracePeriod: v ackGracePeriod})

2158− })

2159

2127

2160-- GET /stats: capacity gauges + minimal federation/timing snapshot. Most of

2128-- GET /stats: capacity gauges + minimal federation/timing snapshot. Most of

2161-- the historical metric fields are zero-filled (see `zeroStats`) — only the

2129-- the historical metric fields are zero-filled (see `zeroStats`) — only the

@@ -2435,7 +2403,7 @@ handleTopRecipients = \limit authorization respond -> withMaintAuth authorizatio

2435 total: 0

2403 total: 0

2436})) (do

2404})) (do

2437 msgs <- *messages

2405 msgs <- *messages

2438− let counts = pluck (unique (pluck msgs (\m -> m.toKey))) (\pk -> {

2406+ let counts = pluck (pluck msgs (\m -> m.toKey)) (\pk -> {

2439 pubkey: pk,

2407 pubkey: pk,

2440 count: countWhere msgs (\m -> m.toKey == pk)

2408 count: countWhere msgs (\m -> m.toKey == pk)

2441 })

2409 })