@@ -876,8 +876,8 @@ sendBatchedGossip = \peers events hops -> forEach peers id (\server -> do
876876broadcastOfflineGossip = \t expiredLocal -> do
877877 presence <- *presence
878878 let peers = knownPeerServersFromPresence presence
879− when (federationEnabled && count peers > 0 && count expiredLocal > 0) (do
880− sendBatchedGossip peers (pluck expiredLocal (\p -> mkGossip (GossipOffline {}) p.pubkey t offlineGossipMinTtl)) gossipOriginHops)
879+ when (federationEnabled && count expiredLocal > 0)
880+ (sendBatchedGossip peers (pluck expiredLocal (\p -> mkGossip (GossipOffline {}) p.pubkey t offlineGossipMinTtl)) gossipOriginHops)
881881
882882-- ============================================
883883-- Forward retries and acks
@@ -898,8 +898,9 @@ retryPendingForwards = do
898898 fromServer: serverHostN, requestTimestamp: t2,
899899 authorization: auth, correlationId: "",
900900 expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})
901− let success = case result of Ok r -> r.ok; _ -> False {}
902− case result of Err _ -> recordFailure server; Ok _ -> when success (clearFailure server)
901+ success <- case result of
902+ Err _ -> do recordFailure server; yield False
903+ Ok r -> do when r.ok (clearFailure server); yield r.ok
903904 let isMatch = \x -> x.messageId == f.messageId && x.toServer == server
904905 atomic do
905906 cur <- *forwards
@@ -943,12 +944,11 @@ pruneAndRetry = \t -> do
943944 when federationEnabled (do
944945 fork retryPendingForwards
945946 aids <- *ackedIds
946− fork (sendAcksAndClean (filter aids (\a -> not (isLocalServer a.originServer)))))
947− -- Refresh local presence for active sessions every ~5 minutes
948− lastRefresh <- *lastPresenceRefresh
949− when (federationEnabled && t - lastRefresh >= presenceRefreshInterval) (do
950− atomic do set *lastPresenceRefresh = t
951− refreshLocalPresence t)
947+ fork (sendAcksAndClean (filter aids (\a -> not (isLocalServer a.originServer))))
948+ lastRefresh <- *lastPresenceRefresh
949+ when (t - lastRefresh >= presenceRefreshInterval) (do
950+ atomic do set *lastPresenceRefresh = t
951+ refreshLocalPresence t))
952952 yield {}
953953
954954refreshLocalPresence = \t -> do
@@ -961,7 +961,7 @@ refreshLocalPresence = \t -> do
961961 else p)
962962 set *presence = updated
963963 yield {refreshed: filter updated (\p -> isLocalServer p.server && p.expiresAt == t + sessionExpiry), peers: knownPeerServersFromPresence updated}
964− when (count refreshState.refreshed > 0 && count refreshState.peers > 0)
964+ when (count refreshState.refreshed > 0)
965965 (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> mkGossip (GossipOnline {}) p.pubkey t (clampTtl (sessionExpiry / 1000)))) gossipOriginHops))
966966
967967-- ============================================
@@ -977,14 +977,13 @@ handleChallenge = \pubkey clientIp respond -> do
977977 result <- atomic do
978978 ch <- *challenges
979979 let cappedActive = capPerKey (filter ch (\c -> c.expiresAt > t)) (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt)
980− rates <- *challengeRates
981− let r = checkRate rates pubkey maxChallengeRatePerKey challengeRateWindowMs 5000 t
982980 if count cappedActive >= maxChallenges then yield {ok: False {}, error: "capacity"}
983− else if r.limited then yield {ok: False {}, error: "rate_limited"}
984981 else do
985− set *challengeRates = r.entries
986− set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
987− yield {ok: True {}, error: ""}
982+ limited <- chargeRate challengeRates pubkey maxChallengeRatePerKey challengeRateWindowMs t
983+ if limited then yield {ok: False {}, error: "rate_limited"}
984+ else do
985+ set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
986+ yield {ok: True {}, error: ""}
988987 yield (respond {challenge: if result.ok then challenge else "", ok: result.ok, expiresAt: if result.ok then t + challengeExpiry else 0, retryAfterMs: 0, error: result.error})
989988
990989handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond -> do
@@ -1006,8 +1005,8 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
10061005 ses <- *sessions
10071006 set *sessions = capList (union (if revokeOthers then removeWhere ses (\s -> s.pubkey == pubkey) else capPerKey ses (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt)) [{token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}]) maxSessions (\s -> s.expiresAt)
10081007 pres <- *presence
1009− set *presence = capList (union (removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)) [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1010− newPres <- *presence
1008+ let newPres = capList (union (removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)) [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1009+ set *presence = newPres
10111010 yield {valid: True {}, peers: knownPeerServersFromPresence newPres}
10121011 if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})
10131012 else do
@@ -1042,7 +1041,7 @@ handleLogout = \authorization clientIp respond -> do
10421041 set *presence = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
10431042 yield (knownPeerServersFromPresence pres)
10441043 yield {ok: True {}, pubkey: pubkey, peers: peers}))
1045− when (removed.ok && removed.pubkey != "" && count removed.peers > 0 && federationEnabled)
1044+ when (removed.ok && federationEnabled)
10461045 (fork (sendBatchedGossip removed.peers [mkGossip (GossipOffline {}) removed.pubkey t offlineGossipMinTtl] gossipOriginHops))
10471046 yield (respond {ok: removed.ok, authorized: removed.ok})
10481047
@@ -1058,21 +1057,16 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
10581057 let cursorWasReset = generation != serverGen && cursor > 0
10591058 if dp >= 2 then yield (respond (emptyPollResp 0 (True {}) (False {}) (True {}) serverGen 0 cursorWasReset))
10601059 else if g then yield (respond (emptyPollResp 0 (False {}) (True {}) (False {}) serverGen globalRateWindowMs cursorWasReset))
1061− else case extractToken authorization of
1062− Nothing {} -> yield (respond (emptyPollResp 0 (False {}) (False {}) (False {}) serverGen 0 cursorWasReset))
1063− Just {value: token} ->
1064− pollOnce token (if generation != serverGen then 0 else cursor) cursorWasReset (takeRelation maxWatchKeys watchKeys) clientIp serverGen respond t
1065−
1066−pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -> do
1067− setup <- atomic do
1068− ses <- *sessions
1069− case findFirst ses (\s -> s.expiresAt > t && s.token == token && matchIp clientIp s.ip) of
1070− Nothing {} -> yield Nothing {}
1071− Just {value: s} -> do
1072− let clampedCursor = clamp 0 t cursor
1073− pendingAcks <- if clampedCursor > 0 then ackPolledMessages s.pubkey clampedCursor t else yield []
1074− startWakeup <- *pollWakeup
1075− yield (Just {value: {pubkey: s.pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, pendingAcks: pendingAcks}})
1060+ else pollOnce authorization (if generation != serverGen then 0 else cursor) cursorWasReset (takeRelation maxWatchKeys watchKeys) clientIp serverGen respond t
1061+
1062+pollOnce = \authorization cursor cursorWasReset watchKeys clientIp serverGen respond t -> do
1063+ setup <- atomic (withSessionAuth authorization t clientIp
1064+ (yield Nothing {})
1065+ (\pubkey -> do
1066+ let clampedCursor = clamp 0 t cursor
1067+ pendingAcks <- if clampedCursor > 0 then ackPolledMessages pubkey clampedCursor t else yield []
1068+ startWakeup <- *pollWakeup
1069+ yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, pendingAcks: pendingAcks}})))
10761070 case setup of
10771071 Nothing {} -> yield (respond (emptyPollResp 0 (False {}) (False {}) (False {}) serverGen 0 cursorWasReset))
10781072 Just {value: s} -> do
@@ -1087,34 +1081,30 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -
10871081 else do
10881082 pollResult <- doPoll s t
10891083 atomic (pruneRef activePolls (\p -> p.pubkey != s.pubkey))
1090− when (federationEnabled && count s.pendingAcks > 0)
1091− (fork (sendAcksAndClean s.pendingAcks))
1092− presenceUpdates <- if count watchKeys > 0 && pollResult.authorized
1093− then atomic do
1094− t2 <- now
1095− pres <- *presence
1096− yield (pluck watchKeys (\k -> {pubkey: k, online: count (presenceForKey pres k t2) > 0}))
1097− else yield []
1084+ when federationEnabled (fork (sendAcksAndClean s.pendingAcks))
1085+ presenceUpdates <- atomic do
1086+ t2 <- now
1087+ pres <- *presence
1088+ yield (pluck watchKeys (\k -> {pubkey: k, online: count (presenceForKey pres k t2) > 0}))
10981089 drains <- atomic do
10991090 myExp <- collectAndPrune senderExpirations (\x -> x.senderKey == s.pubkey) (\x -> {messageId: x.messageId, toKey: x.toKey, expiredAt: x.expiredAt})
11001091 myDel <- collectAndPrune senderDeliveries (\x -> x.senderKey == s.pubkey) (\x -> {messageId: x.messageId, toKey: x.toKey, deliveredAt: x.deliveredAt})
11011092 myCancels <- collectAndPrune cancelNotifications (\x -> x.toKey == s.pubkey) (\x -> {messageId: x.messageId, cancelledAt: x.cancelledAt})
11021093 yield {expirations: myExp, deliveries: myDel, cancellations: myCancels}
1103− yield (respond {events: pollResult.events, cursor: pollResult.cursor, authorized: pollResult.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: pollResult.authorized && count pollResult.events == 0, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: pollResult.hasMore, cursorStale: False {}, messageExpirations: drains.expirations, messageDeliveries: drains.deliveries, droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: drains.cancellations})
1094+ yield (respond {events: pollResult.events, cursor: pollResult.cursor, authorized: True {}, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: count pollResult.events == 0, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: pollResult.hasMore, cursorStale: False {}, messageExpirations: drains.expirations, messageDeliveries: drains.deliveries, droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: drains.cancellations})
11041095
11051096doPoll = \s t -> do
11061097 let collectAt = \tnow -> atomic do
11071098 msgs <- *messages
11081099 yield (collectPollEvents msgs s.pubkey s.clampedCursor tnow)
11091100 initial <- collectAt t
1110− r <- if count initial.events > 0 then yield initial
1111− else do
1112− atomic do
1113− wakeup <- *pollWakeup
1114− if wakeup > s.startWakeup then yield {} else retry
1115− t2 <- now
1116− collectAt t2
1117− yield {authorized: True {}, events: r.events, cursor: r.cursor, hasMore: r.hasMore}
1101+ if count initial.events > 0 then yield initial
1102+ else do
1103+ atomic do
1104+ wakeup <- *pollWakeup
1105+ if wakeup > s.startWakeup then yield {} else retry
1106+ t2 <- now
1107+ collectAt t2
11181108
11191109collectPollEvents = \messages pubkey cursor t ->
11201110 let myMsgs = filter messages (\m -> m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor) in
@@ -1150,7 +1140,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
11501140 withSessionAuth authorization t clientIp
11511141 (yield {status: MsUnauthorized {}, remoteServers: []})
11521142 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effMsgId expiresInMs (clamp 0 maxMessagePriority priority) t)
1153− when (federationEnabled && count result.remoteServers > 0)
1143+ when federationEnabled
11541144 (fork (forkForwardsTo effMsgId to encryptedBlob clampedTs result.remoteServers))
11551145 let ok = case result.status of
11561146 MsDelivered {} -> True {}; MsFederated {} -> True {}
@@ -1179,8 +1169,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
11791169 set *pollWakeup = seqTs
11801170 let toPresence = presenceForKey pres to t
11811171 let remoteServers = collectPeers toPresence (\p -> p.server)
1182− forwardsEnqueued <- if federationEnabled && count remoteServers > 0
1183− then do
1172+ forwardsEnqueued <- if federationEnabled then do
11841173 fwds <- *forwards
11851174 let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.messageId == effMsgId && f.toServer == s)) (\s -> {messageId: effMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: expiresAt, priority: msgPriority, senderKey: senderKey})
11861175 appendCapped forwards newFwds maxForwardsEntries (\f -> f.createdAt)
@@ -1216,7 +1205,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
12161205 t <- now
12171206 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}
12181207 gateRated (False {}) authorization clientIp lookupRates maxLookupRatePerKey lookupRateWindowMs t respond mkErr
1219− {server: "", authorized: True {}, error: "rate_limited"} (\senderKey -> do
1208+ {server: "", authorized: True {}, error: "rate_limited"} (\_ -> do
12201209 pres <- *presence
12211210 yield (respond {server: maskedServer pres pubkey t, authorized: True {}, error: ""}))
12221211
@@ -1224,7 +1213,7 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
12241213 t <- now
12251214 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}
12261215 gateRated (False {}) authorization clientIp lookupRates maxLookupRatePerKey lookupRateWindowMs t respond mkErr
1227− {results: [], authorized: True {}, error: "rate_limited"} (\senderKey -> do
1216+ {results: [], authorized: True {}, error: "rate_limited"} (\_ -> do
12281217 pres <- *presence
12291218 let results = do
12301219 k <- takeRelation 50 pubkeys
@@ -1261,7 +1250,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
12611250 toAck <- removeForKey matches
12621251 ackedFed <- recordAcksAndDeliveries toAck t
12631252 yield {ackedCount: count toAck, pendingAcks: ackedFed}
1264− fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
1253+ when federationEnabled (fork (sendAcksAndClean result.pendingAcks))
12651254 yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount, error: ""}))
12661255
12671256handleListMessages = \offset authorization clientIp respond -> do
@@ -1287,7 +1276,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
12871276 let ackedFed = mapWhere toDel (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
12881277 appendCapped ackedIds ackedFed maxAckedIds (\a -> a.ackedAt)
12891278 yield {deletedCount: count toDel, pendingAcks: ackedFed}
1290− fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
1279+ when federationEnabled (fork (sendAcksAndClean result.pendingAcks))
12911280 yield (respond {ok: True {}, authorized: True {}, deletedCount: result.deletedCount, error: ""}))
12921281
12931282handleCancelMessage = \messageId to authorization clientIp respond -> do
@@ -1298,16 +1287,12 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
12981287 let matches = \x -> x.messageId == messageId && x.toKey == to
12991288 result <- atomic do
13001289 msgs <- *messages
1301− case findFirst msgs matches of
1302− Nothing {} -> yield {cancelled: False {}, cancelServers: []}
1303− Just {value: msg} ->
1304− if msg.originServer != "" || msg.senderKey != senderKey
1305− then yield {cancelled: False {}, cancelServers: []}
1306− else do
1307− fwdServers <- applyCancel msgs matches to messageId t
1308− pres <- *presence
1309− yield {cancelled: True {}, cancelServers: unique (union (collectPeers (presenceForKey pres to t) (\p -> p.server)) fwdServers)}
1310− when (result.cancelled && count result.cancelServers > 0 && federationEnabled)
1290+ if any msgs (\m -> matches m && m.originServer == "" && m.senderKey == senderKey) then do
1291+ fwdServers <- applyCancel msgs matches to messageId t
1292+ pres <- *presence
1293+ yield {cancelled: True {}, cancelServers: unique (union (collectPeers (presenceForKey pres to t) (\p -> p.server)) fwdServers)}
1294+ else yield {cancelled: False {}, cancelServers: []}
1295+ when (result.cancelled && federationEnabled)
13111296 (fork (sendCancelToEach messageId to result.cancelServers))
13121297 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
13131298
@@ -1331,14 +1316,10 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
13311316 let matches = \x -> x.messageId == e.messageId && x.toKey == e.to
13321317 r <- atomic do
13331318 msgs <- *messages
1334− case findFirst msgs matches of
1335− Nothing {} -> yield False
1336− Just {value: msg} ->
1337− if msg.originServer != "" || msg.senderKey != senderKey
1338− then yield False
1339− else do
1340− _ <- applyCancel msgs matches e.to e.messageId t
1341− yield True
1319+ if any msgs (\m -> matches m && m.originServer == "" && m.senderKey == senderKey) then do
1320+ _ <- applyCancel msgs matches e.to e.messageId t
1321+ yield True
1322+ else yield False
13421323 yield {messageId: e.messageId, cancelled: r})
13431324 yield (respond {results: results, authorized: True {}, error: ""}))
13441325
@@ -1353,7 +1334,7 @@ handleSendBatch = \messages authorization clientIp respond -> do
13531334 r <- if t - m.timestamp > messageMaxAge
13541335 then yield {status: MsRejected {}, remoteServers: []}
13551336 else atomic do sendMessageCore senderKey m.to m.encryptedBlob clampedTs effMsgId m.expiresInMs clampedPri t
1356− when (federationEnabled && count r.remoteServers > 0)
1337+ when federationEnabled
13571338 (fork (forkForwardsTo effMsgId m.to m.encryptedBlob clampedTs r.remoteServers))
13581339 yield {messageId: effMsgId, status: r.status})
13591340 yield (respond {results: results, authorized: True {}, retryAfterMs: 0}))
@@ -1381,7 +1362,7 @@ handleFederationPubkey = \respond -> do
13811362handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlationId authorization respond -> do
13821363 t <- now
13831364 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (clamp 0 maxGossipHops hopsRemaining >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
1384− valid <- atomic do
1365+ newlyOnline <- atomic do
13851366 pres <- *presence
13861367 let validEvents = do
13871368 e <- events
@@ -1392,9 +1373,8 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
13921373 let online = mapWhere validEvents (\v -> case v.eventType of GossipOnline {} -> True {}; _ -> False {}) (\v -> {pubkey: v.pubkey, server: v.server, expiresAt: v.expiresAt, timestamp: v.timestamp})
13931374 let offline = mapWhere validEvents (\v -> case v.eventType of GossipOffline {} -> True {}; _ -> False {}) (\v -> {pubkey: v.pubkey, server: v.server, timestamp: v.timestamp})
13941375 set *presence = capList (mergePresence (filter pres (\p -> none offline (\o -> o.pubkey == p.pubkey && o.server == p.server && o.timestamp > p.timestamp))) online) maxPresenceEntries (\p -> p.expiresAt)
1395− yield {newlyOnline: mapWhere online (\e -> not (isLocalServer e.server) && none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)) (\e -> {pubkey: e.pubkey, server: e.server})}
1396− when (count valid.newlyOnline > 0)
1397− (fork (forEach valid.newlyOnline (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t)))
1376+ yield (mapWhere online (\e -> not (isLocalServer e.server) && none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)) (\e -> {pubkey: e.pubkey, server: e.server}))
1377+ fork (forEach newlyOnline (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t))
13981378 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
13991379
14001380-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.
@@ -1412,12 +1392,7 @@ storeFederated = \toKey encryptedBlob timestamp originServer messageId expiresAt
14121392 else if count msgs >= maxMessages then yield "global_full"
14131393 else if countWhere msgs (\m -> m.toKey == toKey) >= maxMessagesPerRecipient then yield "mailbox_full"
14141394 else do
1415− limited <- if checkRecvRate then do
1416− rr <- *recvRates
1417− let r = checkRate rr toKey maxRecvRatePerKey recvRateWindowMs 5000 t
1418− set *recvRates = r.entries
1419− yield r.limited
1420− else yield False
1395+ limited <- if checkRecvRate then chargeRate recvRates toKey maxRecvRatePerKey recvRateWindowMs t else yield False
14211396 if limited then yield "rate_limited"
14221397 else do
14231398 seq <- *seqCounter
@@ -1434,9 +1409,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
14341409 then yield (respond {ok: False {}, error: "stale_timestamp", retryAfterMs: 0})
14351410 else do
14361411 result <- atomic (storeFederated toKey encryptedBlob timestamp originServer messageId expiresAt priority (True {}) t)
1437− if result == ""
1438− then yield (respond {ok: True {}, error: "", retryAfterMs: 0})
1439− else yield (respond {ok: False {}, error: result, retryAfterMs: 0}))
1412+ yield (respond {ok: result == "", error: result, retryAfterMs: 0}))
14401413
14411414handleFederationAck = \messageIds originServer fromServer requestTimestamp correlationId authorization respond -> do
14421415 t <- now
@@ -1447,12 +1420,11 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
14471420 fwds <- *forwards
14481421 let isAcked = \f -> f.toServer == fromServer && elem dedupedIds f.messageId
14491422 let removed = filter fwds isAcked
1450− set *forwards = removeWhere fwds isAcked
1451− -- Reclaim origin messages with no remaining forwards
1423+ let remainingFwds = removeWhere fwds isAcked
1424+ set *forwards = remainingFwds
14521425 msgs <- *messages
14531426 let removedIds = pluck removed (\f -> f.messageId)
14541427 pres <- *presence
1455− remainingFwds <- *forwards
14561428 let toReap = filter msgs (\m ->
14571429 elem removedIds m.messageId
14581430 && t - m.receivedAt >= ackGracePeriod
@@ -1469,11 +1441,10 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
14691441 let matches = \x -> x.messageId == messageId && x.toKey == toKey
14701442 cancelled <- atomic do
14711443 msgs <- *messages
1472− case findFirst msgs (\m -> matches m && m.originServer == fromServer) of
1473− Nothing {} -> yield False
1474− Just {} -> do
1475− _ <- applyCancel msgs matches toKey messageId t
1476− yield True
1444+ if any msgs (\m -> matches m && m.originServer == fromServer) then do
1445+ _ <- applyCancel msgs matches toKey messageId t
1446+ yield True
1447+ else yield False
14771448 yield (respond {ok: True {}, error: "", cancelled: cancelled, retryAfterMs: 0}))
14781449
14791450handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId authorization respond -> do
@@ -1484,11 +1455,10 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
14841455 let matches = \x -> x.messageId == e.messageId && x.toKey == e.toKey
14851456 cancelled <- atomic do
14861457 msgs <- *messages
1487− case findFirst msgs (\m -> matches m && m.originServer == fromServer) of
1488− Nothing {} -> yield False
1489− Just {} -> do
1490− _ <- applyCancel msgs matches e.toKey e.messageId t
1491− yield True
1458+ if any msgs (\m -> matches m && m.originServer == fromServer) then do
1459+ _ <- applyCancel msgs matches e.toKey e.messageId t
1460+ yield True
1461+ else yield False
14921462 yield {messageId: e.messageId, cancelled: cancelled})
14931463 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
14941464
@@ -1540,16 +1510,16 @@ handleDrain = \phase authorization respond -> withMaintAuth authorization
15401510 drainResult <- atomic do
15411511 set *drainPhase = effectivePhase
15421512 set *drainCompletedAt = t
1543− if effectivePhase == 1 then yield {drainedCount: 0, peers: [], localKeys: []}
1513+ if effectivePhase == 1 then yield {peers: [], localKeys: []}
15441514 else do
15451515 pres <- *presence
15461516 let localEntries = filter pres (\p -> isLocalServer p.server && p.expiresAt > t)
15471517 set *presence = filter pres (\p -> not (isLocalServer p.server) || p.expiresAt <= t)
15481518 set *pollWakeup = t
1549− yield {drainedCount: count localEntries, peers: knownPeerServersFromPresence pres, localKeys: localEntries}
1550− when (effectivePhase == 2 && count drainResult.localKeys > 0 && count drainResult.peers > 0 && federationEnabled)
1519+ yield {peers: knownPeerServersFromPresence pres, localKeys: localEntries}
1520+ when (effectivePhase == 2 && count drainResult.localKeys > 0 && federationEnabled)
15511521 (fork (sendBatchedGossip drainResult.peers (pluck drainResult.localKeys (\p -> mkGossip (GossipOffline {}) p.pubkey t offlineGossipMinTtl)) gossipOriginHops))
1552− yield (respond {ok: True {}, drained: drainResult.drainedCount, phase: effectivePhase}))
1522+ yield (respond {ok: True {}, drained: count drainResult.localKeys, phase: effectivePhase}))
15531523
15541524handleUndrain = \authorization respond -> withMaintAuth authorization
15551525 (yield (respond {ok: False {}}))
@@ -1659,13 +1629,12 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
16591629
16601630handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
16611631 (yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0}))
1662− (if ip == "" then yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0})
1663− else do
1664− keys <- atomic do
1632+ (do
1633+ keys <- if ip == "" then yield [] else atomic do
16651634 ses <- *sessions
16661635 pruneRef sessions (\s -> s.ip != ip)
16671636 yield (mapWhere ses (\s -> s.ip == ip) (\s -> s.pubkey))
1668− yield (respond {ok: True {}, revokedCount: count keys, affectedKeys: count (unique keys)}))
1637+ yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count (unique keys)}))
16691638
16701639handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization
16711640 (yield (respond {sessions: [], ok: False {}, total: 0}))