@@ -438,9 +438,6 @@ isQueued = \ds -> case ds of DsQueued {} -> True {}; _ -> False {}
438438nextSeqTs : Timestamp -> Timestamp -> Timestamp
439439nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
440440
441−takeN : [a] -> Int -> [a]
442−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
443−
444441dedupMessageIds : [MessageIdHex] -> [MessageIdHex]
445442dedupMessageIds = \messageIds ->
446443 fold (\acc mid ->
@@ -448,39 +445,11 @@ dedupMessageIds = \messageIds ->
448445 else if any acc (\a -> a == mid) then acc
449446 else union acc [mid]) [] messageIds
450447
451−boundedSmallest = \items n getKey ->
452− (fold (\acc a ->
453− let k = getKey a in
454− if acc.size < n
455− then
456− let newHeap = union acc.heap [a] in
457− let newSize = count newHeap in
458− if newSize > acc.size
459− then {size: newSize, heap: newHeap, heapMax: Just {value: case acc.heapMax of
460− Nothing {} -> k
461− Just {value: v} -> max k v
462− }}
463− else acc
464− else case acc.heapMax of
465− Nothing {} -> acc
466− Just {value: maxVal} ->
467− if k < maxVal
468− then
469− let removed = fold (\r x ->
470− if not r.done && getKey x == maxVal
471− then {done: True {}, items: r.items}
472− else {done: r.done, items: union r.items [x]}) {done: False {}, items: []} acc.heap in
473− let newHeap = union removed.items [a] in
474− let newMax = fold (\best x -> Just {value: case best of
475− Nothing {} -> getKey x
476− Just {value: v} -> max (getKey x) v
477− }) Nothing {} newHeap in
478− {size: acc.size, heap: newHeap, heapMax: newMax}
479− else acc) {size: 0, heap: [], heapMax: Nothing {}} items).heap
480−
481448evictOldestAckedIds = \items n ->
482449 if n <= 0 then items
483− else diff items (boundedSmallest items n (\a -> a.ackedAt))
450+ else do
451+ ii <- items
452+ diff items (takeRelation n ( sortBy (\a -> a.ackedAt) items))
484453
485454safeCompareLoop : Text -> Text -> Int -> Int -> Int
486455safeCompareLoop = \s1 s2 n result ->
@@ -553,7 +522,7 @@ capGossipPresence = \args ->
553522 let updatesOnlyAfterOffline = removeOffline updatesOnlyMerged args.trustedOffline args.t in
554523 let reclaimSlots = maxPresenceEntries - count updatesOnlyAfterOffline in
555524 let newEntries = do e <- args.trustedDeduped; where none args.presence (\p -> samePubkeyServer p e); yield e in
556− let reclaimedEntries = takeN newEntries reclaimSlots in
525+ let reclaimedEntries = takeRelation reclaimSlots newEntries in
557526 {finalPresence: union updatesOnlyAfterOffline reclaimedEntries,
558527 appliedOnline: union
559528 (do e <- updatesOnly; where none args.presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield (pubkeyServer e))
@@ -751,7 +720,7 @@ forwardQueuedMessages = \targets t -> do
751720 where none forwards (\f -> f.messageId == m.messageId && f.toServer == target.server)
752721 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
753722 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp}
754− let capped = takeN candidates (maxForwardsEntries - count forwards)
723+ let capped = takeRelation (maxForwardsEntries - count forwards) candidates
755724 let updatedForwards = union forwards capped
756725 set *forwards = updatedForwards
757726 yield capped
@@ -832,11 +801,10 @@ pollOnce = \token cursor startWakeup -> do
832801 atomic do
833802 wakeup <- *pollWakeup
834803 allMessages <- *messages
835− let matchingMsgs = do
804+ let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (do
836805 m <- allMessages
837806 where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
838− yield m
839− let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
807+ yield m))
840808 if count limitedMsgs > 0
841809 then
842810 let evts = do
@@ -945,15 +913,14 @@ serve = \req -> case req of
945913 seqCounter <- *seqCounter
946914 let seqTs = nextSeqTs t seqCounter
947915 set *seqCounter = seqTs
948− let updatedMessages = union messages
916+ set *messages = union messages
949917 [{messageId: msgId, toKey: to,
950918 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
951− set *messages = updatedMessages
952919 let toPresence = do p <- presence; where p.pubkey == to && p.expiresAt > t; yield p
953920 let localOnline = any toPresence (\p -> isLocalServer p.server)
954921 let remoteServers = collectPeerServers toPresence (\p -> p.server)
955922 forwards <- *forwards
956− let forwardServers = takeN remoteServers (maxForwardsEntries - count forwards)
923+ let forwardServers = takeRelation (maxForwardsEntries - count forwards) remoteServers
957924 let shouldForward = federationEnabled && count forwardServers > 0
958925 let delivery = if localOnline then DsDelivered {}
959926 else if shouldForward then DsFederated {}
@@ -1004,7 +971,7 @@ serve = \req -> case req of
1004971 ackedIds <- *ackedIds
1005972 let combinedAckedIds = union ackedIds ackedFederated
1006973 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
1007− if excessCount <= 0 then [] else boundedSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
974+ if excessCount <= 0 then [] else takeRelation excessCount (sortBy (\a -> a.ackedAt) combinedAckedIds)
1008975 let remainingAckedIds = diff combinedAckedIds evictedEntries
1009976 set *ackedIds = remainingAckedIds
1010977 yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}})