9f1d050 · skrepka · git.iko.soy

git.iko.soy/skrepka

Simplify code and improve type safety: type timeout as Int<Ms>, factor capPerKeyAndGlobal, optimize takeN from O(n²) to O(n), remove redundant checks and filters, simplify joins and conditionals

9f1d050iko6 months ago

server.knot+23−24

compare with master

unifiedsplit

@@ -110,7 +110,7 @@ route AuthApi where

110 = VerifyAuth

111

112route ClientApi where

113− POST {cursor: Timestamp, timeout: Int} /poll headers {authorization: Text}

113+ POST {cursor: Timestamp, timeout: Int<Ms>} /poll headers {authorization: Text}

114 -> {events: [PollEvent], cursor: Timestamp, authorized: Bool}

115 = Poll

116 /messages

@@ -395,9 +395,9 @@ clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp

395

396nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1

397

398−isFederatedOrigin = \originServer -> not (originServer == "") && not (originServer == serverHostN)

398+isFederatedOrigin = \originServer -> not (originServer == "" || originServer == serverHostN)

399

400−takeN = \items n -> fold (\acc e -> if count acc >= n then acc else union acc [e]) [] items

400+takeN = \items n -> (fold (\acc e -> if acc.n >= n then acc else {n: acc.n + 1, items: union acc.items [e]}) {n: 0, items: []} items).items

401

402dedupMessageIds = \messageIds ->

403 fold (\acc mid ->

@@ -471,6 +471,9 @@ capGlobal = \items maxCount getKey ->

471 then evictOneOldest items getKey

472 else items

473

474+capPerKeyAndGlobal = \items matchKey maxPerKey getExpiry t maxGlobal ->

475+ capGlobal (capPerKey items matchKey maxPerKey getExpiry t) maxGlobal getExpiry

476+

474collectPeerServers = \items getServer ->

477collectPeerServers = \items getServer ->

475 fold (\acc item ->

478 fold (\acc item ->

476 let s = getServer item in

479 let s = getServer item in

@@ -587,12 +590,11 @@ sendBatchedGossip = \peerServers allEvents hops ->

587

590

588broadcastOfflineGossip = \t peerServers expiredLocal ->

591broadcastOfflineGossip = \t peerServers expiredLocal ->

589 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do

592 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do

590− let expiredEntries = mergeByKey (\a b -> a.pubkey == b.pubkey) [] (do

591− p <- expiredLocal

592− let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)

593− yield {pubkey: p.pubkey, ttl: ttl, expiresAt: p.expiresAt})

594 let gossipEvents = do

593 let gossipEvents = do

595− e <- expiredEntries

594+ e <- mergeByKey (\a b -> a.pubkey == b.pubkey) [] (do

595+ p <- expiredLocal

596+ let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)

597+ yield {pubkey: p.pubkey, ttl: ttl, expiresAt: p.expiresAt})

596 yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}

598 yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}

597 sendBatchedGossip peerServers gossipEvents 2)

599 sendBatchedGossip peerServers gossipEvents 2)

598

600

@@ -687,11 +689,12 @@ pruneAndRetry = \t -> do

687 let pendingAcks = do a <- ackedIds; where isFederatedOrigin a.originServer; yield a

689 let pendingAcks = do a <- ackedIds; where isFederatedOrigin a.originServer; yield a

688 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks}

690 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks}

689 broadcastOfflineGossip t result.peerServers result.expiredLocal

691 broadcastOfflineGossip t result.peerServers result.expiredLocal

692+ when federationEnabled (do

690− when (federationEnabled && count result.pendingForwards > 0)

693+ when (count result.pendingForwards > 0)

691− (retryPendingForwards result.pendingForwards)

694+ (retryPendingForwards result.pendingForwards)

692− when (federationEnabled && count result.pendingAcks > 0) (do

695+ when (count result.pendingAcks > 0) (do

693− let ackOriginServers = collectPeerServers result.pendingAcks (\a -> a.originServer)

696+ let ackOriginServers = collectPeerServers result.pendingAcks (\a -> a.originServer)

694− sendAcksAndClean result.pendingAcks ackOriginServers)

697+ sendAcksAndClean result.pendingAcks ackOriginServers))

695

698

696pollOnce = \token cursor -> do

699pollOnce = \token cursor -> do

697 t <- now

700 t <- now

@@ -747,10 +750,9 @@ serve = \req -> case req of

747 pruneResult <- atomic do

750 pruneResult <- atomic do

748 pr <- maybePrune t

751 pr <- maybePrune t

749 challenges <- *challenges

752 challenges <- *challenges

750− let activeChallenges = capPerKey challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t

753+ let capped = capPerKeyAndGlobal challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t maxChallenges

751− let globalCapped = capGlobal activeChallenges maxChallenges (\c -> c.expiresAt)

752 set *challenges = union

754 set *challenges = union

753− globalCapped

755+ capped

754 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

756 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

755 yield pr

757 yield pr

756 forkBroadcastOffline t pruneResult

758 forkBroadcastOffline t pruneResult

@@ -775,9 +777,8 @@ serve = \req -> case req of

775 c <- challenges

777 c <- challenges

776 where not (c.pubkey == pubkey && c.challenge == challenge)

778 where not (c.pubkey == pubkey && c.challenge == challenge)

777 yield c

779 yield c

778− let activeSessions = capPerKey sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t

780+ let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions

779− let globalSessionsCapped = capGlobal activeSessions maxSessions (\s -> s.expiresAt)

780− set *sessions = union globalSessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]

781+ set *sessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]

781 let presenceWithout = do

782 let presenceWithout = do

782 p <- presence

783 p <- presence

783 where not (p.pubkey == pubkey && p.server == serverHostN)

784 where not (p.pubkey == pubkey && p.server == serverHostN)

@@ -883,8 +884,7 @@ serve = \req -> case req of

883 messages <- *messages

884 messages <- *messages

884 let ackedMessages = do

885 let ackedMessages = do

885 m <- messages

886 m <- messages

886− mid <- dedupedIds

887− where m.messageId == mid && m.toKey == pubkey

887+ where m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid)

888 yield m

889 set *messages = diff messages ackedMessages

890 let ackedFederated = do

@@ -902,8 +902,7 @@ serve = \req -> case req of

902 AckDone {pendingAcks, evictedEntries} ->

903 when federationEnabled (do

904 let originServers = collectPeerServers pendingAcks (\a -> a.originServer)

905− when (count originServers > 0)

906− (sendAcksAndClean pendingAcks originServers)

905+ sendAcksAndClean pendingAcks originServers

907 retryEvictedAcks evictedEntries pendingAcks)

906 retryEvictedAcks evictedEntries pendingAcks)

908 _ -> yield {}

907 _ -> yield {}

909 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)

908 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)

@@ -993,7 +992,7 @@ serve = \req -> case req of

993 let updatesOnlyMerged = mergeByPubkeyServer presence updatesOnly in

992 let updatesOnlyMerged = mergeByPubkeyServer presence updatesOnly in

994 let updatesOnlyAfterOffline = removeOffline updatesOnlyMerged offlineOnlyKeys trustedOffline t in

993 let updatesOnlyAfterOffline = removeOffline updatesOnlyMerged offlineOnlyKeys trustedOffline t in

995 let reclaimSlots = maxPresenceEntries - count updatesOnlyAfterOffline in

994 let reclaimSlots = maxPresenceEntries - count updatesOnlyAfterOffline in

996− let newEntries = do e <- trustedDeduped; where none presence (\p -> samePubkeyServer p e) && none updatesOnlyAfterOffline (\p -> samePubkeyServer p e); yield e in

995+ let newEntries = do e <- trustedDeduped; where none presence (\p -> samePubkeyServer p e); yield e in

997 let reclaimedEntries = takeN newEntries reclaimSlots in

996 let reclaimedEntries = takeN newEntries reclaimSlots in

998 {finalPresence: union updatesOnlyAfterOffline reclaimedEntries,

997 {finalPresence: union updatesOnlyAfterOffline reclaimedEntries,

999 appliedOnline: union

998 appliedOnline: union