be70b1b · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Simplify server: extract gateRated, applyCancel, handleListPeers helpers; inline zipPairs

be70b1biko6 months ago

server.knot+44−51

compare with master

unifiedsplit

@@ -615,6 +615,12 @@ gateAuth = \checkDrain auth ip t respond mkErr onAuth -> do

615 if g then yield (respond (mkErr "rate_limited"))

616 else withSessionAuth auth t ip (yield (respond (mkErr "unauthorized"))) onAuth

617

618+-- gateAuth + per-key rate gate. `rateResp` is the response value when the per-key gate trips.

619+gateRated = \checkDrain auth ip rateRef maxRate windowMs t respond mkErr rateResp onAuth ->

620+ gateAuth checkDrain auth ip t respond mkErr (\pk -> do

621+ lim <- checkPubkeyRateGate rateRef pk maxRate windowMs t

622+ if lim then yield (respond rateResp) else onAuth pk)

623+

618authenticateSession : [Session] -> Text -> Timestamp -> AuthResult

624authenticateSession : [Session] -> Text -> Timestamp -> AuthResult

619authenticateSession = \sessions token t ->

625authenticateSession = \sessions token t ->

620 case findFirst sessions (\s -> s.expiresAt > t && s.token == token) of

626 case findFirst sessions (\s -> s.expiresAt > t && s.token == token) of

@@ -865,6 +871,16 @@ recordAcksAndDeliveries = \toAck t -> do

865 appendCapped senderDeliveries localDel 10000 (\d -> d.deliveredAt)

871 appendCapped senderDeliveries localDel 10000 (\d -> d.deliveredAt)

866 yield ackedFed

872 yield ackedFed

867

873

874+-- Common cancel block: drop matching messages + forwards, record cancel notification.

875+-- Returns the federated forward servers that had the message (for caller follow-up).

876+applyCancel = \msgs matches to mid t -> do

877+ set *messages = removeWhere msgs matches

878+ fwds <- *forwards

879+ let fwdServers = collectPeers (filter fwds matches) (\f -> f.toServer)

880+ pruneRef forwards (\f -> not (matches f))

881+ appendCapped cancelNotifications [{toKey: to, messageId: mid, cancelledAt: t}] 10000 (\c -> c.cancelledAt)

882+ yield fwdServers

883+

868sendBatchedGossip = \peers events hops -> forEach peers id (\server -> do

884sendBatchedGossip = \peers events hops -> forEach peers id (\server -> do

869 t <- now

885 t <- now

870 failures <- *forwardFailures

886 failures <- *forwardFailures

@@ -1253,20 +1269,16 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->

1253handleLookupPresence = \pubkey authorization clientIp respond -> do

1269handleLookupPresence = \pubkey authorization clientIp respond -> do

1254 t <- now

1270 t <- now

1255 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}

1271 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}

1256− gateAuth (False {}) authorization clientIp t respond mkErr (\senderKey -> do

1257− lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t

1272+ gateRated (False {}) authorization clientIp lookupRates maxLookupRatePerKey lookupRateWindowMs t respond mkErr

1258− if lim then yield (respond {server: "", authorized: True {}, error: "rate_limited"})

1273+ {server: "", authorized: True {}, error: "rate_limited"} (\senderKey -> do

1259− else do

1260 pres <- *presence

1274 pres <- *presence

1261 yield (respond {server: maskedServer pres pubkey t, authorized: True {}, error: ""}))

1275 yield (respond {server: maskedServer pres pubkey t, authorized: True {}, error: ""}))

1262

1276

1263handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do

1277handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do

1264 t <- now

1278 t <- now

1265 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}

1279 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}

1266− gateAuth (False {}) authorization clientIp t respond mkErr (\senderKey -> do

1267− lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t

1280+ gateRated (False {}) authorization clientIp lookupRates maxLookupRatePerKey lookupRateWindowMs t respond mkErr

1268− if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})

1281+ {results: [], authorized: True {}, error: "rate_limited"} (\senderKey -> do

1269− else do

1270 pres <- *presence

1282 pres <- *presence

1271 let results = do

1283 let results = do

1272 k <- takeRelation 50 pubkeys

1284 k <- takeRelation 50 pubkeys

@@ -1295,10 +1307,8 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do

1295handleAckMessages = \messageIds authorization clientIp respond -> do

1307handleAckMessages = \messageIds authorization clientIp respond -> do

1296 t <- now

1308 t <- now

1297 let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}

1309 let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}

1298− gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do

1299− lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t

1310+ gateRated (True {}) authorization clientIp clientWriteRates maxClientWriteRatePerKey clientWriteRateWindowMs t respond mkErr

1300− if lim then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})

1311+ {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"} (\pubkey -> do

1301− else do

1302 let dedupedIds = dedupMessageIds messageIds

1312 let dedupedIds = dedupMessageIds messageIds

1303 let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId

1313 let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId

1304 result <- atomic do

1314 result <- atomic do

@@ -1314,10 +1324,8 @@ handleAckMessages = \messageIds authorization clientIp respond -> do

1314handleListMessages = \offset authorization clientIp respond -> do

1324handleListMessages = \offset authorization clientIp respond -> do

1315 t <- now

1325 t <- now

1316 let mkErr = \e -> {messages: [], authorized: False {}, count: 0, error: if e == "rate_limited" then "rate_limited" else ""}

1326 let mkErr = \e -> {messages: [], authorized: False {}, count: 0, error: if e == "rate_limited" then "rate_limited" else ""}

1317− gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do

1318− lim <- checkPubkeyRateGate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs t

1327+ gateRated (True {}) authorization clientIp lookupRates maxLookupRatePerKey lookupRateWindowMs t respond mkErr

1319− if lim then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})

1328+ {messages: [], authorized: True {}, count: 0, error: "rate_limited"} (\pubkey -> do

1320− else do

1321 msgs <- *messages

1329 msgs <- *messages

1322 let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))

1330 let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))

1323 let total = count myMsgs

1331 let total = count myMsgs

@@ -1329,10 +1337,8 @@ handleListMessages = \offset authorization clientIp respond -> do

1329handleDeleteMessages = \messageIds authorization clientIp respond -> do

1337handleDeleteMessages = \messageIds authorization clientIp respond -> do

1330 t <- now

1338 t <- now

1331 let mkErr = \e -> {ok: False {}, authorized: False {}, deletedCount: 0, error: if e == "draining" then "" else e}

1339 let mkErr = \e -> {ok: False {}, authorized: False {}, deletedCount: 0, error: if e == "draining" then "" else e}

1332− gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do

1333− lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t

1340+ gateRated (True {}) authorization clientIp clientWriteRates maxClientWriteRatePerKey clientWriteRateWindowMs t respond mkErr

1334− if lim then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})

1341+ {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"} (\pubkey -> do

1335− else do

1336 let dedupedIds = dedupMessageIds messageIds

1342 let dedupedIds = dedupMessageIds messageIds

1337 let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId

1343 let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId

1338 result <- atomic do

1344 result <- atomic do

@@ -1349,10 +1355,8 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do

1349handleCancelMessage = \messageId to authorization clientIp respond -> do

1355handleCancelMessage = \messageId to authorization clientIp respond -> do

1350 t <- now

1356 t <- now

1351 let mkErr = \e -> {ok: False {}, authorized: False {}, cancelled: False {}, error: if e == "draining" then "" else e}

1357 let mkErr = \e -> {ok: False {}, authorized: False {}, cancelled: False {}, error: if e == "draining" then "" else e}

1352− gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do

1353− lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t

1358+ gateRated (True {}) authorization clientIp clientWriteRates maxClientWriteRatePerKey clientWriteRateWindowMs t respond mkErr

1354− if lim then yield (respond {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited"})

1359+ {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited"} (\senderKey -> do

1355− else do

1356 let matches = \x -> x.messageId == messageId && x.toKey == to

1360 let matches = \x -> x.messageId == messageId && x.toKey == to

1357 result <- atomic do

1361 result <- atomic do

1358 msgs <- *messages

1362 msgs <- *messages

@@ -1362,13 +1366,9 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do

1362 if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""

1366 if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""

1363 then yield {cancelled: False {}, cancelServers: []}

1367 then yield {cancelled: False {}, cancelServers: []}

1364 else do

1368 else do

1365− set *messages = removeWhere msgs matches

1369+ fwdServers <- applyCancel msgs matches to messageId t

1366− fwds <- *forwards

1367− let fwdServers = collectPeers (filter fwds matches) (\f -> f.toServer)

1368− pruneRef forwards (\f -> not (matches f))

1369 pres <- *presence

1370 pres <- *presence

1370 let presServers = collectPeers (presenceForKey pres to t) (\p -> p.server)

1371 let presServers = collectPeers (presenceForKey pres to t) (\p -> p.server)

1371− appendCapped cancelNotifications [{toKey: to, messageId: messageId, cancelledAt: t}] 10000 (\c -> c.cancelledAt)

1372 yield {cancelled: True {}, cancelServers: unique (union presServers fwdServers)}

1373 when (result.cancelled && count result.cancelServers > 0 && federationEnabled)

1374 (fork (sendCancelToEach messageId to result.cancelServers))

@@ -1389,10 +1389,8 @@ sendCancelToEach = \messageId to servers ->

1389handleCancelBatch = \cancels authorization clientIp respond -> do

1390 t <- now

1391 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "draining" then "" else e}

1392− gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do

1393− lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t

1392+ gateRated (True {}) authorization clientIp clientWriteRates maxClientWriteRatePerKey clientWriteRateWindowMs t respond mkErr

1394− if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})

1393+ {results: [], authorized: True {}, error: "rate_limited"} (\senderKey -> do

1395− else do

1396 results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do

1394 results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do

1397 let matches = \x -> x.messageId == e.messageId && x.toKey == e.to

1395 let matches = \x -> x.messageId == e.messageId && x.toKey == e.to

1398 r <- atomic do

1396 r <- atomic do

@@ -1403,9 +1401,7 @@ handleCancelBatch = \cancels authorization clientIp respond -> do

1403 if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""

1401 if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""

1404 then yield False

1402 then yield False

1405 else do

1403 else do

1406− set *messages = removeWhere msgs matches

1404+ _ <- applyCancel msgs matches e.to e.messageId t

1407− pruneRef forwards (\f -> not (matches f))

1408− appendCapped cancelNotifications [{toKey: e.to, messageId: e.messageId, cancelledAt: t}] 10000 (\c -> c.cancelledAt)

1409 yield True

1405 yield True

1410 yield {messageId: e.messageId, cancelled: r})

1406 yield {messageId: e.messageId, cancelled: r})

1411 yield (respond {results: results, authorized: True {}, error: ""}))

1407 yield (respond {results: results, authorized: True {}, error: ""}))

@@ -1551,9 +1547,7 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio

1551 case findFirst msgs (\m -> matches m && m.originServer == fromServer) of

1547 case findFirst msgs (\m -> matches m && m.originServer == fromServer) of

1552 Nothing {} -> yield False

1548 Nothing {} -> yield False

1553 Just {} -> do

1549 Just {} -> do

1554− set *messages = removeWhere msgs matches

1550+ _ <- applyCancel msgs matches toKey messageId t

1555− pruneRef forwards (\f -> not (matches f))

1556− appendCapped cancelNotifications [{toKey: toKey, messageId: messageId, cancelledAt: t}] 10000 (\c -> c.cancelledAt)

1557 yield True

1551 yield True

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

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

1559

1553

@@ -1593,7 +1587,11 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp

1593 if count messageIds != count toKeys

1587 if count messageIds != count toKeys

1594 then yield (respond {ok: False {}, error: "mismatched_lengths", receivedCount: 0, retryAfterMs: 0})

1588 then yield (respond {ok: False {}, error: "mismatched_lengths", receivedCount: 0, retryAfterMs: 0})

1595 else do

1589 else do

1596− let pairs = zipPairs messageIds toKeys

1590+ let pairs = do

1591+ mp <- withIndex messageIds

1592+ kp <- withIndex toKeys

1593+ where mp.i == kp.i

1594+ yield {messageId: mp.v, toKey: kp.v}

1597 atomic do

1595 atomic do

1598 msgs <- *messages

1596 msgs <- *messages

1599 let receipts = do

1597 let receipts = do

@@ -1607,10 +1605,6 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp

1607-- Index a list (1-based by virtue of `count` of the partial fold).

1605-- Index a list (1-based by virtue of `count` of the partial fold).

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

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

1609

1607

1610−-- Pair parallel lists by index (O(n*m) cross-filter; trivial code).

1611−zipPairs = \mids keys ->

1612− do mp <- withIndex mids; kp <- withIndex keys; where mp.i == kp.i; yield {messageId: mp.v, toKey: kp.v}

1613−

1614-- ============================================

1608-- ============================================

1615-- Maintenance handlers

1609-- Maintenance handlers

1616-- ============================================

1610-- ============================================

@@ -1682,10 +1676,12 @@ handleUnblockPeer = \server authorization respond -> withMaintAuth authorization

1682 (yield (respond {ok: False {}, unblocked: False {}}))

1676 (yield (respond {ok: False {}, unblocked: False {}}))

1683 (do removed <- removeServerFromList blockedPeers server; yield (respond {ok: True {}, unblocked: removed}))

1677 (do removed <- removeServerFromList blockedPeers server; yield (respond {ok: True {}, unblocked: removed}))

1684

1678

1685−handleListBlocked = \authorization respond -> do

1679+handleListPeers = \listRef authorization respond -> do

1686 ok <- verifyMaintenanceAuth authorization

1680 ok <- verifyMaintenanceAuth authorization

1687− blocked <- *blockedPeers

1681+ list <- *listRef

1688− yield (respond {peers: if ok then blocked else [], ok: ok})

1682+ yield (respond {peers: if ok then list else [], ok: ok})

1683+

1684+handleListBlocked = \authorization respond -> handleListPeers blockedPeers authorization respond

1689

1685

1690handleAllowPeer = \server authorization respond -> withMaintAuth authorization

1686handleAllowPeer = \server authorization respond -> withMaintAuth authorization

1691 (yield (respond {ok: False {}, allowed: 0}))

1687 (yield (respond {ok: False {}, allowed: 0}))

@@ -1695,10 +1691,7 @@ handleUnallowPeer = \server authorization respond -> withMaintAuth authorization

1695 (yield (respond {ok: False {}, removed: False {}}))

1691 (yield (respond {ok: False {}, removed: False {}}))

1696 (do removed <- removeServerFromList allowedPeers server; yield (respond {ok: True {}, removed: removed}))

1692 (do removed <- removeServerFromList allowedPeers server; yield (respond {ok: True {}, removed: removed}))

1697

1693

1698−handleListAllowed = \authorization respond -> do

1694+handleListAllowed = \authorization respond -> handleListPeers allowedPeers authorization respond

1699− ok <- verifyMaintenanceAuth authorization

1700− allowed <- *allowedPeers

1701− yield (respond {peers: if ok then allowed else [], ok: ok})

1702

1695

1703handlePingPeer = \server authorization respond -> withMaintAuth authorization

1696handlePingPeer = \server authorization respond -> withMaintAuth authorization

1704 (yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"}))

1697 (yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"}))