@@ -110,7 +110,7 @@ route AuthApi where
110110 = VerifyAuth
111111
112112route ClientApi where
113− POST {cursor: Timestamp, timeout: Int} /poll headers {authorization: Text}
113+ POST {cursor: Timestamp, timeout: Int<Ms>} /poll headers {authorization: Text}
114114 -> {events: [PollEvent], cursor: Timestamp, authorized: Bool}
115115 = Poll
116116 /messages
@@ -395,9 +395,9 @@ clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
395395
396396nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
397397
398−isFederatedOrigin = \originServer -> not (originServer == "") && not (originServer == serverHostN)
398+isFederatedOrigin = \originServer -> not (originServer == "" || originServer == serverHostN)
399399
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
401401
402402dedupMessageIds = \messageIds ->
403403 fold (\acc mid ->
@@ -471,6 +471,9 @@ capGlobal = \items maxCount getKey ->
471471 then evictOneOldest items getKey
472472 else items
473473
474+capPerKeyAndGlobal = \items matchKey maxPerKey getExpiry t maxGlobal ->
475+ capGlobal (capPerKey items matchKey maxPerKey getExpiry t) maxGlobal getExpiry
476+
474477collectPeerServers = \items getServer ->
475478 fold (\acc item ->
476479 let s = getServer item in
@@ -587,12 +590,11 @@ sendBatchedGossip = \peerServers allEvents hops ->
587590
588591broadcastOfflineGossip = \t peerServers expiredLocal ->
589592 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})
594593 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})
596598 yield {eventType: "offline", pubkey: e.pubkey, server: serverHostN, timestamp: e.expiresAt, ttl: e.ttl}
597599 sendBatchedGossip peerServers gossipEvents 2)
598600
@@ -687,11 +689,12 @@ pruneAndRetry = \t -> do
687689 let pendingAcks = do a <- ackedIds; where isFederatedOrigin a.originServer; yield a
688690 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks}
689691 broadcastOfflineGossip t result.peerServers result.expiredLocal
690− when (federationEnabled && count result.pendingForwards > 0)
691− (retryPendingForwards result.pendingForwards)
692− when (federationEnabled && count result.pendingAcks > 0) (do
693− let ackOriginServers = collectPeerServers result.pendingAcks (\a -> a.originServer)
694− sendAcksAndClean result.pendingAcks ackOriginServers)
692+ when federationEnabled (do
693+ when (count result.pendingForwards > 0)
694+ (retryPendingForwards result.pendingForwards)
695+ when (count result.pendingAcks > 0) (do
696+ let ackOriginServers = collectPeerServers result.pendingAcks (\a -> a.originServer)
697+ sendAcksAndClean result.pendingAcks ackOriginServers))
695698
696699pollOnce = \token cursor -> do
697700 t <- now
@@ -747,10 +750,9 @@ serve = \req -> case req of
747750 pruneResult <- atomic do
748751 pr <- maybePrune t
749752 challenges <- *challenges
750− let activeChallenges = capPerKey challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t
751− let globalCapped = capGlobal activeChallenges maxChallenges (\c -> c.expiresAt)
753+ let capped = capPerKeyAndGlobal challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t maxChallenges
752754 set *challenges = union
753− globalCapped
755+ capped
754756 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
755757 yield pr
756758 forkBroadcastOffline t pruneResult
@@ -775,9 +777,8 @@ serve = \req -> case req of
775777 c <- challenges
776778 where not (c.pubkey == pubkey && c.challenge == challenge)
777779 yield c
778− let activeSessions = capPerKey sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t
779− let globalSessionsCapped = capGlobal activeSessions maxSessions (\s -> s.expiresAt)
780− set *sessions = union globalSessionsCapped [{token: token, pubkey: pubkey, expiresAt: expiry}]
780+ let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
781+ set *sessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
781782 let presenceWithout = do
782783 p <- presence
783784 where not (p.pubkey == pubkey && p.server == serverHostN)
@@ -883,8 +884,7 @@ serve = \req -> case req of
883884 messages <- *messages
884885 let ackedMessages = do
885886 m <- messages
886− mid <- dedupedIds
887− where m.messageId == mid && m.toKey == pubkey
887+ where m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid)
888888 yield m
889889 set *messages = diff messages ackedMessages
890890 let ackedFederated = do
@@ -902,8 +902,7 @@ serve = \req -> case req of
902902 AckDone {pendingAcks, evictedEntries} ->
903903 when federationEnabled (do
904904 let originServers = collectPeerServers pendingAcks (\a -> a.originServer)
905− when (count originServers > 0)
906− (sendAcksAndClean pendingAcks originServers)
905+ sendAcksAndClean pendingAcks originServers
907906 retryEvictedAcks evictedEntries pendingAcks)
908907 _ -> yield {}
909908 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
@@ -993,7 +992,7 @@ serve = \req -> case req of
993992 let updatesOnlyMerged = mergeByPubkeyServer presence updatesOnly in
994993 let updatesOnlyAfterOffline = removeOffline updatesOnlyMerged offlineOnlyKeys trustedOffline t in
995994 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
997996 let reclaimedEntries = takeN newEntries reclaimSlots in
998997 {finalPresence: union updatesOnlyAfterOffline reclaimedEntries,
999998 appliedOnline: union