git.iko.soy/skrepka
on master help sign in
Extract helper functions to reduce code duplication in state management and gossip handling
82fcdc7iko
server.knot+62−79
compare with master
unifiedsplit
@@ -427,6 +427,19 @@ evictOneOldest = \items getKey ->
427 then {removed: True {}, items: acc.items}
427 then {removed: True {}, items: acc.items}
428 else {removed: acc.removed, items: union acc.items [x]}) {removed: False {}, items: []} items).items
428 else {removed: acc.removed, items: union acc.items [x]}) {removed: False {}, items: []} items).items
429
429
430+capPerKey = \items matchKey maxPerKey getExpiry t ->
431+ let activeCount = count (do x <- items; where matchKey x; where getExpiry x > t; yield x) in
432+ if activeCount >= maxPerKey
433+ then let forKey = do x <- items; where matchKey x; yield x in
434+ let others = do x <- items; where not (matchKey x); yield x in
435+ union others (evictOneOldest forKey getExpiry)
436+ else items
437+
438+capGlobal = \items maxCount getKey ->
439+ if count items >= maxCount
440+ then evictOneOldest items getKey
441+ else items
442+
430collectPeerServers = \items getServer ->
443collectPeerServers = \items getServer ->
431 fold (\acc item ->
444 fold (\acc item ->
432 let s = getServer item in
445 let s = getServer item in
@@ -434,6 +447,31 @@ collectPeerServers = \items getServer ->
434 else if any acc (\a -> a == s) then acc
447 else if any acc (\a -> a == s) then acc
435 else union acc [s]) [] items
448 else union acc [s]) [] items
436
449
450+samePubkeyServer = \a b -> a.pubkey == b.pubkey && a.server == b.server
451+
452+mergeByPubkeyServer = \initial items ->
453+ fold (\acc e ->
454+ if any acc (\a -> samePubkeyServer a e && a.expiresAt >= e.expiresAt)
455+ then acc
456+ else union
457+ (do a <- acc; where not (samePubkeyServer a e); yield a)
458+ [e]) initial items
459+
460+removeOffline = \presenceSet offlineOnlyKeys trustedOffline t ->
461+ diff presenceSet (do
462+ p <- presenceSet
463+ k <- offlineOnlyKeys
464+ where samePubkeyServer p k
465+ where any trustedOffline (\o -> samePubkeyServer o k && (if o.timestamp > t then t else o.timestamp) > p.timestamp)
466+ yield p)
467+
468+countMessagesFor = \messages toKey t ->
469+ count (do
470+ m <- messages
471+ where m.toKey == toKey
472+ where m.receivedAt >= t - messageMaxAge
473+ yield m)
474+
437validGossipEvents = \events validRequest t eventType ->
475validGossipEvents = \events validRequest t eventType ->
438 do
476 do
439 e <- events
477 e <- events
@@ -688,18 +726,11 @@ serve = \req -> case req of
688 pruneResult <- atomic do
726 pruneResult <- atomic do
689 pr <- maybePrune t
727 pr <- maybePrune t
690 challenges <- *challenges
728 challenges <- *challenges
691− let existingCount = count (do c <- challenges; where c.pubkey == pubkey; where c.expiresAt > t; yield c)
692− let forKey = do c <- challenges; where c.pubkey == pubkey; yield c
693− let others = do c <- challenges; where not (c.pubkey == pubkey); yield c
694− let challengesCapped = if existingCount >= maxChallengesPerKey
729+ let challengesCapped = capPerKey challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t
695− then union others (evictOneOldest forKey (\c -> c.expiresAt))
696− else challenges
697 let withoutExpired = if count challengesCapped >= maxChallenges
730 let withoutExpired = if count challengesCapped >= maxChallenges
698 then do c <- challengesCapped; where c.expiresAt > t; yield c
731 then do c <- challengesCapped; where c.expiresAt > t; yield c
699 else challengesCapped
732 else challengesCapped
700− let globalCapped = if count withoutExpired >= maxChallenges
733+ let globalCapped = capGlobal withoutExpired maxChallenges (\c -> c.expiresAt)
701− then evictOneOldest withoutExpired (\c -> c.expiresAt)
702− else withoutExpired
703 set *challenges = union
734 set *challenges = union
704 globalCapped
735 globalCapped
705 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
736 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
@@ -734,16 +765,9 @@ serve = \req -> case req of
734 yield c
765 yield c
735 Nothing {} -> challenges
766 Nothing {} -> challenges
736 else challenges
767 else challenges
737− let existingForKey = do s <- sessions; where s.pubkey == pubkey; where s.expiresAt > t; yield s
738− let forKey = do s <- sessions; where s.pubkey == pubkey; yield s
739− let others = do s <- sessions; where not (s.pubkey == pubkey); yield s
740− let sessionsCapped = if count existingForKey >= maxSessionsPerKey
768+ let sessionsCapped = capPerKey sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t
741− then union others (evictOneOldest forKey (\s -> s.expiresAt))
742− else sessions
743 let activeSessionsCapped = do s <- sessionsCapped; where s.expiresAt > t; yield s
769 let activeSessionsCapped = do s <- sessionsCapped; where s.expiresAt > t; yield s
744− let globalSessionsCapped = if count activeSessionsCapped >= maxSessions
770+ let globalSessionsCapped = capGlobal activeSessionsCapped maxSessions (\s -> s.expiresAt)
745− then evictOneOldest activeSessionsCapped (\s -> s.expiresAt)
746− else activeSessionsCapped
747 set *sessions = if sigValid
771 set *sessions = if sigValid
748 then union globalSessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]
772 then union globalSessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]
749 else sessions
773 else sessions
@@ -751,9 +775,7 @@ serve = \req -> case req of
751 p <- presence
775 p <- presence
752 where not (p.pubkey == pubkey && p.server == serverHostN)
776 where not (p.pubkey == pubkey && p.server == serverHostN)
753 yield p
777 yield p
754− let presenceEvicted = if count presenceWithout >= maxPresenceEntries
778+ let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
755− then evictOneOldest presenceWithout (\p -> p.expiresAt)
756− else presenceWithout
757 let updatedPresence = if sigValid
779 let updatedPresence = if sigValid
758 then union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
780 then union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
759 else presence
781 else presence
@@ -811,11 +833,7 @@ serve = \req -> case req of
811 let authorized = isJust session
833 let authorized = isJust session
812 seqCounter <- *seqCounter
834 seqCounter <- *seqCounter
813 let seqTs = if t > seqCounter then t else seqCounter + 1
835 let seqTs = if t > seqCounter then t else seqCounter + 1
814− let recipientMsgCount = count (do
836+ let recipientMsgCount = countMessagesFor messages to t
815− m <- messages
816− where m.toKey == to
817− where m.receivedAt >= t - messageMaxAge
818− yield m)
819 let shouldInsert = authorized && recipientMsgCount < maxMessagesPerRecipient
837 let shouldInsert = authorized && recipientMsgCount < maxMessagesPerRecipient
820 set *seqCounter = if shouldInsert then seqTs else seqCounter
838 set *seqCounter = if shouldInsert then seqTs else seqCounter
821 set *messages = if shouldInsert
839 set *messages = if shouldInsert
@@ -990,7 +1008,7 @@ serve = \req -> case req of
990 let offlineEntries = validGossipEvents events validRequest t "offline"
1008 let offlineEntries = validGossipEvents events validRequest t "offline"
991 let filteredOnline = do
1009 let filteredOnline = do
992 e <- onlineEntries
1010 e <- onlineEntries
993− where none offlineEntries (\o -> o.server == fromServer && o.pubkey == e.pubkey && o.server == e.server && o.timestamp > e.timestamp)
1011+ where none offlineEntries (\o -> o.server == fromServer && samePubkeyServer o e && o.timestamp > e.timestamp)
994 yield {pubkey: e.pubkey, server: e.server, expiresAt: e.expiresAt, timestamp: e.timestamp, ttl: e.ttl}
1012 yield {pubkey: e.pubkey, server: e.server, expiresAt: e.expiresAt, timestamp: e.timestamp, ttl: e.ttl}
995 let affectedKeys = union
1013 let affectedKeys = union
996 (do
1014 (do
@@ -999,15 +1017,7 @@ serve = \req -> case req of
999 (do
1017 (do
1000 e <- offlineEntries
1018 e <- offlineEntries
1001 yield {pubkey: e.pubkey, server: e.server})
1019 yield {pubkey: e.pubkey, server: e.server})
1002− let deduped = fold (\acc e ->
1003− if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server && a.expiresAt >= e.expiresAt)
1004− then acc
1005− else union
1006− (do
1007− a <- acc
1008− where not (a.pubkey == e.pubkey && a.server == e.server)
1009− yield a)
1010− [e]) [] filteredOnline
1020+ let deduped = mergeByPubkeyServer [] filteredOnline
1011 gossipResult <- atomic do
1021 gossipResult <- atomic do
1012 pr <- maybePrune t
1022 pr <- maybePrune t
1013 presence <- *presence
1023 presence <- *presence
@@ -1021,59 +1031,36 @@ serve = \req -> case req of
1021 yield e
1031 yield e
1022 let appliedOnline = do
1032 let appliedOnline = do
1023 e <- trustedDeduped
1033 e <- trustedDeduped
1024− where none presence (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)
1034+ where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
1025 yield {pubkey: e.pubkey, server: e.server}
1035 yield {pubkey: e.pubkey, server: e.server}
1026− let merged = fold (\acc e ->
1027− if any acc (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)
1028− then acc
1029− else union
1030− (do
1031− p <- acc
1032− where not (p.pubkey == e.pubkey && p.server == e.server)
1033− yield p)
1034− [e]) presence trustedDeduped
1036+ let merged = mergeByPubkeyServer presence trustedDeduped
1035 let offlineOnlyKeys = do
1037 let offlineOnlyKeys = do
1036 k <- affectedKeys
1038 k <- affectedKeys
1037− where none trustedDeduped (\d -> d.pubkey == k.pubkey && d.server == k.server)
1039+ where none trustedDeduped (\d -> samePubkeyServer d k)
1038 yield k
1040 yield k
1039 let appliedOffline = do
1041 let appliedOffline = do
1040 k <- offlineOnlyKeys
1042 k <- offlineOnlyKeys
1041− where any presence (\p -> p.pubkey == k.pubkey && p.server == k.server && any trustedOffline (\o -> o.pubkey == k.pubkey && o.server == k.server && (if o.timestamp > t then t else o.timestamp) > p.timestamp))
1043+ where any presence (\p -> samePubkeyServer p k && any trustedOffline (\o -> samePubkeyServer o k && (if o.timestamp > t then t else o.timestamp) > p.timestamp))
1042 yield {pubkey: k.pubkey, server: k.server}
1044 yield {pubkey: k.pubkey, server: k.server}
1043− let afterOffline = diff merged (do
1045+ let afterOffline = removeOffline merged offlineOnlyKeys trustedOffline t
1044− p <- merged
1045− k <- offlineOnlyKeys
1046− where p.pubkey == k.pubkey
1047− where p.server == k.server
1048− where any trustedOffline (\o -> o.pubkey == k.pubkey && o.server == k.server && (if o.timestamp > t then t else o.timestamp) > p.timestamp)
1049− yield p)
1050 let withinCapacity = count afterOffline <= maxPresenceEntries
1046 let withinCapacity = count afterOffline <= maxPresenceEntries
1051 let updatesOnlyMerged = if withinCapacity then afterOffline
1047 let updatesOnlyMerged = if withinCapacity then afterOffline
1052 else fold (\acc e ->
1048 else fold (\acc e ->
1053− if any acc (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)
1049+ if any acc (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
1054 then acc
1050 then acc
1055− else if any presence (\p -> p.pubkey == e.pubkey && p.server == e.server)
1051+ else if any presence (\p -> samePubkeyServer p e)
1056 then union
1052 then union
1057− (do
1053+ (do p <- acc; where not (samePubkeyServer p e); yield p)
1058− p <- acc
1059− where not (p.pubkey == e.pubkey && p.server == e.server)
1060− yield p)
1061 [e]
1054 [e]
1062 else acc) presence trustedDeduped
1055 else acc) presence trustedDeduped
1063 let updatesOnlyAfterOffline = if withinCapacity then afterOffline
1056 let updatesOnlyAfterOffline = if withinCapacity then afterOffline
1064− else diff updatesOnlyMerged (do
1065− p <- updatesOnlyMerged
1057+ else removeOffline updatesOnlyMerged offlineOnlyKeys trustedOffline t
1066− k <- offlineOnlyKeys
1067− where p.pubkey == k.pubkey
1068− where p.server == k.server
1069− where any trustedOffline (\o -> o.pubkey == k.pubkey && o.server == k.server && (if o.timestamp > t then t else o.timestamp) > p.timestamp)
1070− yield p)
1071 let reclaimCapacity = if withinCapacity then 0 else maxPresenceEntries - count updatesOnlyAfterOffline
1058 let reclaimCapacity = if withinCapacity then 0 else maxPresenceEntries - count updatesOnlyAfterOffline
1072 let newEntries = if withinCapacity then []
1059 let newEntries = if withinCapacity then []
1073 else do
1060 else do
1074 e <- trustedDeduped
1061 e <- trustedDeduped
1075− where none presence (\p -> p.pubkey == e.pubkey && p.server == e.server)
1062+ where none presence (\p -> samePubkeyServer p e)
1076− where none updatesOnlyAfterOffline (\p -> p.pubkey == e.pubkey && p.server == e.server)
1063+ where none updatesOnlyAfterOffline (\p -> samePubkeyServer p e)
1077 yield e
1064 yield e
1078 let reclaimedEntries = if withinCapacity then []
1065 let reclaimedEntries = if withinCapacity then []
1079 else fold (\acc e ->
1066 else fold (\acc e ->
@@ -1085,8 +1072,8 @@ serve = \req -> case req of
1085 else union
1072 else union
1086 (do
1073 (do
1087 e <- trustedDeduped
1074 e <- trustedDeduped
1088− where any presence (\p -> p.pubkey == e.pubkey && p.server == e.server)
1075+ where any presence (\p -> samePubkeyServer p e)
1089− where none presence (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)
1076+ where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
1090 yield {pubkey: e.pubkey, server: e.server})
1077 yield {pubkey: e.pubkey, server: e.server})
1091 (do
1078 (do
1092 e <- reclaimedEntries
1079 e <- reclaimedEntries
@@ -1099,13 +1086,13 @@ serve = \req -> case req of
1099 yield {appliedKeys: (if withinCapacity then union appliedOnline appliedOffline else union appliedOnlineUpdates appliedOffline), expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1086 yield {appliedKeys: (if withinCapacity then union appliedOnline appliedOffline else union appliedOnlineUpdates appliedOffline), expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1100 let filteredOffline = do
1087 let filteredOffline = do
1101 e <- offlineEntries
1088 e <- offlineEntries
1102− where none onlineEntries (\o -> o.pubkey == e.pubkey && o.server == e.server && o.timestamp > e.timestamp)
1089+ where none onlineEntries (\o -> samePubkeyServer o e && o.timestamp > e.timestamp)
1103 yield e
1090 yield e
1104 let effectiveEvents = do
1091 let effectiveEvents = do
1105 e <- union
1092 e <- union
1106 (do e <- filteredOnline; yield {eventType: "online", pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl})
1093 (do e <- filteredOnline; yield {eventType: "online", pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl})
1107 (do e <- filteredOffline; yield {eventType: "offline", pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl})
1094 (do e <- filteredOffline; yield {eventType: "offline", pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl})
1108− where any gossipResult.appliedKeys (\a -> a.pubkey == e.pubkey && a.server == e.server)
1095+ where any gossipResult.appliedKeys (\a -> samePubkeyServer a e)
1109 yield e
1096 yield e
1110 if not gossipResult.withinCapacity
1097 if not gossipResult.withinCapacity
1111 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)
1098 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)
@@ -1145,11 +1132,7 @@ serve = \req -> case req of
1145 seqCounter <- *seqCounter
1132 seqCounter <- *seqCounter
1146 let seqTs = if t > seqCounter then t else seqCounter + 1
1133 let seqTs = if t > seqCounter then t else seqCounter + 1
1147 let recipientMsgCount = if not (authorized && validOrigin) then 0
1134 let recipientMsgCount = if not (authorized && validOrigin) then 0
1148− else count (do
1135+ else countMessagesFor messages toKey t
1149− m <- messages
1150− where m.toKey == toKey
1151− where m.receivedAt >= t - messageMaxAge
1152− yield m)
1153 let existsInMessages = if not (authorized && validOrigin) then False {}
1136 let existsInMessages = if not (authorized && validOrigin) then False {}
1154 else any messages (\m -> m.messageId == messageId && m.toKey == toKey)
1137 else any messages (\m -> m.messageId == messageId && m.toKey == toKey)
1155 ackedIds <- *ackedIds
1138 ackedIds <- *ackedIds