19a7d60 · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Simplify server: flatten nested conditions, unify rate helpers, use pruneRef

19a7d60iko6 months ago

server.knot+79−106

compare with master

unifiedsplit

@@ -440,7 +440,7 @@ route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | Federati

440-- ============================================

441

442any : [a] -> (a -> Bool) -> Bool

443−any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs

443+any = \xs pred -> count (do x <- xs; where pred x; yield x) > 0

444none : [a] -> (a -> Bool) -> Bool

445none = \xs pred -> not (any xs pred)

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

@@ -461,17 +461,15 @@ pruneRef = \ref keep -> do

461-- Add `server` to the named list ref if absent. Returns the new total.

462addServerToList = \listRef server -> atomic do

463 current <- *listRef

464− if any current (\x -> x == server) then yield (count current)

464+ let updated = if any current (\x -> x == server) then current else union current [server]

465− else do

465+ set *listRef = updated

466− set *listRef = union current [server]

467− yield (count current + 1)

466+ yield (count updated)

468

467

469-- Remove `server` from the named list ref. Returns whether it was present.

468-- Remove `server` from the named list ref. Returns whether it was present.

470removeServerFromList = \listRef server -> atomic do

469removeServerFromList = \listRef server -> atomic do

471 current <- *listRef

470 current <- *listRef

471+ pruneRef listRef (\x -> x != server)

472− let had = any current (\x -> x == server)

472+ yield (any current (\x -> x == server))

473− set *listRef = do x <- current; where x != server; yield x

474− yield had

475

473

476headOf = \items -> fold (\acc x -> case acc of

474headOf = \items -> fold (\acc x -> case acc of

477 Nothing {} -> Just {value: x}

475 Nothing {} -> Just {value: x}

@@ -669,17 +667,16 @@ verifyFederationAuth = \authorization fromServer scope requestTimestamp -> do

669withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do

667withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do

670 blocked <- *blockedPeers

668 blocked <- *blockedPeers

671 allowed <- *allowedPeers

669 allowed <- *allowedPeers

672− let denyReason =

670+ let preErr =

673 if isBlockedPeer fromServer blocked then "blocked"

671 if isBlockedPeer fromServer blocked then "blocked"

674 else if not (isAllowedByAllowlist fromServer allowed) then "not_allowed"

672 else if not (isAllowedByAllowlist fromServer allowed) then "not_allowed"

673+ else if not precondition then "invalid_request"

674+ else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance) then "clock_skew"

675 else ""

676− if denyReason != "" then yield (respond {ok: False {}, error: denyReason, retryAfterMs: 0})

676+ if preErr != "" then yield (respond {ok: False {}, error: preErr, retryAfterMs: 0})

677 else do

678 verified <- verifyFederationAuth authorization fromServer scope requestTimestamp

679 if not verified then yield (respond {ok: False {}, error: "unauthorized", retryAfterMs: 0})

680− else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance)

681− then yield (respond {ok: False {}, error: "clock_skew", retryAfterMs: 0})

682− else if not precondition then yield (respond {ok: False {}, error: "invalid_request", retryAfterMs: 0})

683 else do

680 else do

684 limited <- checkServerRateGate fromServer t

681 limited <- checkServerRateGate fromServer t

685 if limited then yield (respond {ok: False {}, error: "rate_limited", retryAfterMs: federationRateWindowMs})

682 if limited then yield (respond {ok: False {}, error: "rate_limited", retryAfterMs: federationRateWindowMs})

@@ -701,29 +698,20 @@ checkRate = \entries key maxRate windowMs maxEntries t ->

701 let windowStart = case cur of Just {value: r} -> r.windowStart; _ -> t in

698 let windowStart = case cur of Just {value: r} -> r.windowStart; _ -> t in

702 {limited: False {}, entries: capList (union others [{key: key, windowStart: windowStart, count: nextCount}]) maxEntries (\r -> r.windowStart)}

699 {limited: False {}, entries: capList (union others [{key: key, windowStart: windowStart, count: nextCount}]) maxEntries (\r -> r.windowStart)}

703

700

704−checkGlobalRate = \t -> atomic do

701+checkWindowRate = \windowRef countRef windowMs maxRate t -> atomic do

705− ws <- *globalRateWindowStart

702+ ws <- *windowRef

706− c <- *globalRateCount

703+ c <- *countRef

707− if t - ws >= globalRateWindowMs then do

704+ if t - ws >= windowMs then do

708− set *globalRateWindowStart = t

705+ set *windowRef = t

709− set *globalRateCount = 1

706+ set *countRef = 1

710 yield False

707 yield False

711− else if c >= maxGlobalRequestRate then yield True

708+ else if c >= maxRate then yield True

712 else do

709 else do

713− set *globalRateCount = c + 1

710+ set *countRef = c + 1

714 yield False

711 yield False

715

712

716−checkAuthGlobalRate = \t -> atomic do

713+checkGlobalRate = \t -> checkWindowRate globalRateWindowStart globalRateCount globalRateWindowMs maxGlobalRequestRate t

717− ws <- *authGlobalRateWindowStart

714+checkAuthGlobalRate = \t -> checkWindowRate authGlobalRateWindowStart authGlobalRateCount authGlobalRateWindowMs maxAuthGlobalRate t

718− c <- *authGlobalRateCount

719− if t - ws >= authGlobalRateWindowMs then do

720− set *authGlobalRateWindowStart = t

721− set *authGlobalRateCount = 1

722− yield False

723− else if c >= maxAuthGlobalRate then yield True

724− else do

725− set *authGlobalRateCount = c + 1

726− yield False

727

715

728checkIpRateGate = \ip t ->

716checkIpRateGate = \ip t ->

729 if ip == "" then yield False

717 if ip == "" then yield False

@@ -740,13 +728,12 @@ checkServerRateGate = \server t ->

740

728

741checkPreAuth = \clientIp t -> do

729checkPreAuth = \clientIp t -> do

742 dp <- *drainPhase

730 dp <- *drainPhase

731+ g <- checkGlobalRate t

732+ i <- checkIpRateGate clientIp t

743− if dp >= 1 then yield "draining"

733+ yield (if dp >= 1 then "draining"

744− else do

745− g <- checkGlobalRate t

746− if g then yield "global_rate_limited"

734+ else if g then "global_rate_limited"

747− else do

748− i <- checkIpRateGate clientIp t

749− yield (if i then "ip_rate_limited" else "")

735+ else if i then "ip_rate_limited"

736+ else "")

750

737

751-- ============================================

738-- ============================================

752-- Federation failure tracking (simple, no circuit breaker)

739-- Federation failure tracking (simple, no circuit breaker)

@@ -766,14 +753,12 @@ recordFailure = \server -> do

766 atomic do

753 atomic do

767 failures <- *forwardFailures

754 failures <- *forwardFailures

768 let others = do f <- failures; where f.server != server; yield f

755 let others = do f <- failures; where f.server != server; yield f

769− let prior = headOf (do f <- failures; where f.server == server; yield f)

756+ let new = case headOf (do f <- failures; where f.server == server; yield f) of

770− let firstAt = case prior of Just {value: f} -> f.firstFailedAt; _ -> t

757+ Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}

771− let nextN = case prior of Just {value: f} -> min (f.failures + 1) 10; _ -> 1

758+ Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}

772− set *forwardFailures = union others [{server: server, failedAt: t, failures: nextN, firstFailedAt: firstAt}]

759+ set *forwardFailures = union others [new]

773

760

774−clearFailure = \server -> atomic do

761+clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))

775− failures <- *forwardFailures

776− set *forwardFailures = do f <- failures; where f.server != server; yield f

777

762

778-- ============================================

763-- ============================================

779-- Common utility queries

764-- Common utility queries

@@ -1135,9 +1120,7 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -

1135 then yield (respond {events: [], cursor: cursor, authorized: True {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: pollTimeoutMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: []})

1120 then yield (respond {events: [], cursor: cursor, authorized: True {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: pollTimeoutMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: []})

1136 else do

1121 else do

1137 pollResult <- doPoll s t

1122 pollResult <- doPoll s t

1138− atomic do

1139− aps <- *activePolls

1123+ atomic (pruneRef activePolls (\p -> p.pubkey != s.pubkey))

1140− set *activePolls = do p <- aps; where p.pubkey != s.pubkey; yield p

1141 when (federationEnabled && count s.pendingAcks > 0)

1124 when (federationEnabled && count s.pendingAcks > 0)

1142 (fork (sendAcksAndClean s.pendingAcks))

1125 (fork (sendAcksAndClean s.pendingAcks))

1143 presenceUpdates <- if count watchKeys > 0 && pollResult.authorized

1126 presenceUpdates <- if count watchKeys > 0 && pollResult.authorized

@@ -1231,58 +1214,50 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a

1231 MsQueued {} -> True {}; MsDuplicate {} -> True {}; _ -> False {}

1214 MsQueued {} -> True {}; MsDuplicate {} -> True {}; _ -> False {}

1232 yield (respond {status: result.status, messageId: if ok then effMsgId else "", retryAfterMs: 0, error: if ok then "" else "rejected", loadPct: 0})

1215 yield (respond {status: result.status, messageId: if ok then effMsgId else "", retryAfterMs: 0, error: if ok then "" else "rejected", loadPct: 0})

1233

1216

1234−sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msgPriority t ->

1217+sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msgPriority t -> do

1218+ msgs <- *messages

1219+ aids <- *ackedIds

1220+ dids <- *deliveredIds

1221+ let isDup = (effMsgId != "" && any msgs (\m -> m.messageId == effMsgId && m.toKey == to))

1222+ || hasMessage effMsgId to aids

1223+ || hasMessage effMsgId to dids

1224+ sendRates <- *sendRates

1225+ let sr = checkRate sendRates senderKey maxSendRatePerKey sendRateWindowMs 5000 t

1226+ set *sendRates = sr.entries

1227+ pairRates <- *pairRates

1228+ let pr = checkRate pairRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs 5000 t

1229+ set *pairRates = pr.entries

1230+ recvRates <- *recvRates

1231+ let rr = checkRate recvRates to maxRecvRatePerKey recvRateWindowMs 5000 t

1232+ set *recvRates = rr.entries

1233+ let recipientCount = count (do m <- msgs; where m.toKey == to; yield m)

1235 if senderKey == to then yield {status: MsRejected {}, remoteServers: []}

1234 if senderKey == to then yield {status: MsRejected {}, remoteServers: []}

1235+ else if isDup then yield {status: MsDuplicate {}, remoteServers: []}

1236+ else if sr.limited || pr.limited || rr.limited then yield {status: MsRateLimited {}, remoteServers: []}

1237+ else if count msgs >= maxMessages then yield {status: MsRejected {}, remoteServers: []}

1238+ else if recipientCount >= maxMessagesPerRecipient then yield {status: MsQueued {}, remoteServers: []}

1236 else do

1239 else do

1237− msgs <- *messages

1238− aids <- *ackedIds

1239− dids <- *deliveredIds

1240− let isDup = (effMsgId != "" && any msgs (\m -> m.messageId == effMsgId && m.toKey == to))

1241− || hasMessage effMsgId to aids

1242− || hasMessage effMsgId to dids

1243− if isDup then yield {status: MsDuplicate {}, remoteServers: []}

1244− else do

1245− sendRates <- *sendRates

1246− let sr = checkRate sendRates senderKey maxSendRatePerKey sendRateWindowMs 5000 t

1247− if sr.limited then yield {status: MsRateLimited {}, remoteServers: []}

1248− else do

1249− set *sendRates = sr.entries

1250− pairRates <- *pairRates

1251− let pr = checkRate pairRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs 5000 t

1252− set *pairRates = pr.entries

1253− if pr.limited then yield {status: MsRateLimited {}, remoteServers: []}

1254− else do

1255− recvRates <- *recvRates

1256− let rr = checkRate recvRates to maxRecvRatePerKey recvRateWindowMs 5000 t

1257− if rr.limited then yield {status: MsRateLimited {}, remoteServers: []}

1258− else do

1259− set *recvRates = rr.entries

1260− if count msgs >= maxMessages then yield {status: MsRejected {}, remoteServers: []}

1261− else

1262− let recipientCount = count (do m <- msgs; where m.toKey == to; yield m) in

1263− if recipientCount >= maxMessagesPerRecipient then yield {status: MsQueued {}, remoteServers: []}

1264− else do

1265− pres <- *presence

1240+ pres <- *presence

1266− seq <- *seqCounter

1241+ seq <- *seqCounter

1267− let seqTs = nextSeqTs t seq

1242+ let seqTs = nextSeqTs t seq

1268− set *seqCounter = seqTs

1243+ set *seqCounter = seqTs

1269− let expiresAt = seqTs + effectiveMessageTtl expiresInMs

1244+ let expiresAt = seqTs + effectiveMessageTtl expiresInMs

1270− set *messages = union msgs [{messageId: effMsgId, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: expiresAt, senderKey: senderKey, priority: msgPriority}]

1245+ set *messages = union msgs [{messageId: effMsgId, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: expiresAt, senderKey: senderKey, priority: msgPriority}]

1271− set *pollWakeup = seqTs

1246+ set *pollWakeup = seqTs

1272− let toPresence = presenceForKey pres to t

1247+ let toPresence = presenceForKey pres to t

1273− let localOnline = any toPresence (\p -> isLocalServer p.server)

1248+ let localOnline = any toPresence (\p -> isLocalServer p.server)

1274− let remoteServers = collectPeers toPresence (\p -> p.server)

1249+ let remoteServers = collectPeers toPresence (\p -> p.server)

1275− forwardsEnqueued <- if federationEnabled && count remoteServers > 0

1250+ forwardsEnqueued <- if federationEnabled && count remoteServers > 0

1276− then do

1251+ then do

1277− fwds <- *forwards

1252+ fwds <- *forwards

1278− let newFwds = do s <- remoteServers; where none fwds (\f -> f.messageId == effMsgId && f.toServer == s); yield {messageId: effMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: expiresAt, priority: msgPriority, senderKey: senderKey}

1253+ let newFwds = do s <- remoteServers; where none fwds (\f -> f.messageId == effMsgId && f.toServer == s); yield {messageId: effMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: expiresAt, priority: msgPriority, senderKey: senderKey}

1279− set *forwards = capList (union fwds newFwds) maxForwardsEntries (\f -> f.createdAt)

1254+ set *forwards = capList (union fwds newFwds) maxForwardsEntries (\f -> f.createdAt)

1280− yield (count newFwds > 0)

1255+ yield (count newFwds > 0)

1281− else yield False

1256+ else yield False

1282− let status = if localOnline then MsDelivered {}

1257+ let status = if localOnline then MsDelivered {}

1283− else if forwardsEnqueued then MsFederated {}

1258+ else if forwardsEnqueued then MsFederated {}

1284− else MsQueued {}

1259+ else MsQueued {}

1285− yield {status: status, remoteServers: remoteServers}

1260+ yield {status: status, remoteServers: remoteServers}

1286

1261

1287forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->

1262forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->

1288 forEach servers (\s -> s) (\server -> do

1263 forEach servers (\s -> s) (\server -> do

@@ -1894,9 +1869,8 @@ handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth autho

1894 revoked <- atomic do

1869 revoked <- atomic do

1895 ses <- *sessions

1870 ses <- *sessions

1896 let mine = do s <- ses; where s.pubkey == pubkey; yield s

1871 let mine = do s <- ses; where s.pubkey == pubkey; yield s

1897− set *sessions = do s <- ses; where s.pubkey != pubkey; yield s

1872+ pruneRef sessions (\s -> s.pubkey != pubkey)

1898− pres <- *presence

1899− set *presence = do p <- pres; where p.pubkey != pubkey || not (isLocalServer p.server); yield p

1873+ pruneRef presence (\p -> p.pubkey != pubkey || not (isLocalServer p.server))

1900 yield (count mine)

1874 yield (count mine)

1901 yield (respond {ok: True {}, revokedCount: revoked}))

1875 yield (respond {ok: True {}, revokedCount: revoked}))

1902

1876

@@ -1907,9 +1881,8 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author

1907 purged <- atomic do

1881 purged <- atomic do

1908 msgs <- *messages

1882 msgs <- *messages

1909 let mine = do m <- msgs; where m.toKey == pubkey; yield m

1883 let mine = do m <- msgs; where m.toKey == pubkey; yield m

1910− set *messages = do m <- msgs; where m.toKey != pubkey; yield m

1884+ pruneRef messages (\m -> m.toKey != pubkey)

1911− fwds <- *forwards

1912− set *forwards = do f <- fwds; where f.toKey != pubkey; yield f

1885+ pruneRef forwards (\f -> f.toKey != pubkey)

1913 let federated = do m <- mine; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}

1886 let federated = do m <- mine; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}

1914 aids <- *ackedIds

1887 aids <- *ackedIds

1915 set *ackedIds = capList (union aids federated) maxAckedIds (\a -> a.ackedAt)

1888 set *ackedIds = capList (union aids federated) maxAckedIds (\a -> a.ackedAt)

@@ -1923,7 +1896,7 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization

1923 result <- atomic do

1896 result <- atomic do

1924 ses <- *sessions

1897 ses <- *sessions

1925 let mine = do s <- ses; where s.ip == ip; yield s

1898 let mine = do s <- ses; where s.ip == ip; yield s

1926− set *sessions = do s <- ses; where s.ip != ip; yield s

1899+ pruneRef sessions (\s -> s.ip != ip)

1927 let keys = unique (do s <- mine; yield s.pubkey)

1900 let keys = unique (do s <- mine; yield s.pubkey)

1928 yield {revoked: count mine, keys: count keys}

1901 yield {revoked: count mine, keys: count keys}

1929 yield (respond {ok: True {}, revokedCount: result.revoked, affectedKeys: result.keys}))

1902 yield (respond {ok: True {}, revokedCount: result.revoked, affectedKeys: result.keys}))