@@ -517,15 +517,15 @@ capGossipPresence = \args ->
517517 if count args.afterOffline <= maxPresenceEntries
518518 then {finalPresence: args.afterOffline, appliedOnline: args.appliedOnline, withinCapacity: True {}}
519519 else
520− let updatesOnly = do e <- args.trustedDeduped; where any args.presence (\p -> samePubkeyServer p e); yield e in
520+ let updatesOnly = existingPresenceUpdates args.trustedDeduped args.presence in
521521 let updatesOnlyMerged = mergeByPubkeyServer args.presence updatesOnly in
522522 let updatesOnlyAfterOffline = removeOffline updatesOnlyMerged args.trustedOffline args.t in
523523 let reclaimSlots = maxPresenceEntries - count updatesOnlyAfterOffline in
524− let newEntries = do e <- args.trustedDeduped; where none args.presence (\p -> samePubkeyServer p e); yield e in
524+ let newEntries = newPresenceEntries args.trustedDeduped args.presence in
525525 let reclaimedEntries = takeRelation reclaimSlots newEntries in
526526 {finalPresence: union updatesOnlyAfterOffline reclaimedEntries,
527527 appliedOnline: union
528− (do e <- updatesOnly; where none args.presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield (pubkeyServer e))
528+ (appliedOnlineKeys updatesOnly args.presence)
529529 (do e <- reclaimedEntries; yield (pubkeyServer e)),
530530 withinCapacity: False {}}
531531
@@ -562,34 +562,21 @@ sendToEach = \servers action ->
562562 sendToEach rest action
563563
564564pruneExpired = \t -> do
565− set *challenges = do
566− c <- *challenges
567− where c.expiresAt > t
568− yield c
569− set *sessions = do
570− s <- *sessions
571− where s.expiresAt > t
572− yield s
565+ challenges <- *challenges
566+ set *challenges = activeByExpiry challenges t
567+ sessions <- *sessions
568+ set *sessions = activeByExpiry sessions t
573569 presenceBefore <- *presence
574570 let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)
575− let presenceAfter = do
576− p <- presenceBefore
577− where p.expiresAt > t
578− yield p
571+ let presenceAfter = activeByExpiry presenceBefore t
579572 set *presence = presenceAfter
580573 let messageCutoff = t - messageMaxAge
581− set *messages = do
582− m <- *messages
583− where m.receivedAt > messageCutoff
584− yield m
585− set *forwards = do
586− f <- *forwards
587− where f.createdAt > messageCutoff
588− yield f
589− set *ackedIds = do
590− a <- *ackedIds
591− where a.ackedAt > messageCutoff
592− yield a
574+ messages <- *messages
575+ set *messages = do m <- messages; where m.receivedAt > messageCutoff; yield m
576+ forwards <- *forwards
577+ set *forwards = do f <- forwards; where f.createdAt > messageCutoff; yield f
578+ ackedIds <- *ackedIds
579+ set *ackedIds = do a <- ackedIds; where a.ackedAt > messageCutoff; yield a
593580 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
594581
595582maybePrune = \t -> do
@@ -627,10 +614,7 @@ sendBatchedGossip = \peerServers allEvents hops ->
627614
628615broadcastOfflineGossip = \t peerServers expiredLocal ->
629616 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do
630− let gossipEvents = mergeByKey (\a b -> a.pubkey == b.pubkey) (\x -> x.timestamp) [] (do
631− p <- expiredLocal
632− let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
633− yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.expiresAt, ttl: ttl})
617+ let gossipEvents = mergeByKey (\a b -> a.pubkey == b.pubkey) (\x -> x.timestamp) [] (offlineGossipEvents expiredLocal)
634618 sendBatchedGossip peerServers gossipEvents gossipOriginHops)
635619
636620forkBroadcastOffline = \t result ->
@@ -642,8 +626,8 @@ effectiveGossipEvents = \entries appliedKeys eventType -> do
642626 yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
643627
644628mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
645− let trustedDeduped = do e <- deduped; where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t); yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp} in
646− let appliedOnline = do e <- trustedDeduped; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield (pubkeyServer e) in
629+ let trustedDeduped = trustedGossipEntries deduped presence fromServer t in
630+ let appliedOnline = appliedOnlineKeys trustedDeduped presence in
647631 let merged = mergeByPubkeyServer presence trustedDeduped in
648632 let afterOffline = removeOffline merged trustedOffline t in
649633 let appliedOffline = do p <- diff merged afterOffline; yield (pubkeyServer p) in
@@ -696,7 +680,7 @@ retryPendingForwards : [ForwardEntry] -> IO {network} {}
696680retryPendingForwards = \pending -> do
697681 let servers = collectPeerServers pending (\x -> x.toServer)
698682 sendToEach servers (\server ->
699− retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)
683+ retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)
700684 yield {}
701685
702686forwardQueuedMessages = \targets t -> do
@@ -705,7 +689,7 @@ forwardQueuedMessages = \targets t -> do
705689 forwards <- *forwards
706690 presence <- *presence
707691 let localOnline = localOnlineKeys presence t
708− let remoteTargets = do target <- targets; where none localOnline (\pk -> pk == target.pubkey); yield target
692+ let remoteTargets = remoteOnlyTargets targets localOnline
709693 let candidates = forwardCandidates remoteTargets messages forwards t
710694 let capped = takeRelation (maxForwardsEntries - count forwards) candidates
711695 let updatedForwards = union forwards capped
@@ -713,7 +697,7 @@ forwardQueuedMessages = \targets t -> do
713697 yield capped
714698 let servers = collectPeerServers newForwards (\x -> x.toServer)
715699 sendToEach servers (\server ->
716− retryForwardsForServer (do x <- newForwards; where x.toServer == server; yield x) maxForwardRetriesPerServer)
700+ retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)
717701
718702fetchFederationAck = \server messageIds ->
719703 fetch (federationUrl server) (FederationAck {
@@ -727,32 +711,26 @@ sendBatchedAcksToServer = \serverAcks server ->
727711 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
728712 whenOk result (atomic do
729713 ackedIds <- *ackedIds
730− let updatedAckedIds = do
731− a <- ackedIds
732− where not (a.originServer == server && hasMessage a.messageId a.toKey batch)
733− yield a
714+ let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch
734715 set *ackedIds = updatedAckedIds))
735716
736717sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
737718sendAcksAndClean = \pendingAcks ->
738719 let servers = collectPeerServers pendingAcks (\a -> a.originServer) in
739720 sendToEach servers (\server ->
740− sendBatchedAcksToServer (do a <- pendingAcks; where a.originServer == server; yield a) server)
721+ sendBatchedAcksToServer (acksForServer pendingAcks server) server)
741722
742723retryEvictedAcks = \evictedEntries pendingAcks ->
743− let evictedToRetry = do
744− a <- evictedEntries
745− where not (hasMessage a.messageId a.toKey pendingAcks)
746− yield a
724+ let evictedToRetry = retriableEvictedAcks evictedEntries pendingAcks
747725 in do
748726 let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
749727 sendToEach evictedOriginServers (\server -> do
750− fetchResult <- fetchFederationAck server (do a <- evictedToRetry; where a.originServer == server; yield a.messageId)
728+ fetchResult <- fetchFederationAck server (do a <- acksForServer evictedToRetry server; yield a.messageId)
751729 case fetchResult of
752730 Err _ -> do
753731 dropped <- atomic do
754732 ackedIds <- *ackedIds
755− let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a
733+ let reinsert = acksForServer evictedToRetry server
756734 let combined = union ackedIds reinsert
757735 let excess = count combined - maxAckedIds
758736 let afterEviction = if excess <= 0 then combined
@@ -768,8 +746,7 @@ pruneAndRetry = \t -> do
768746 set *lastPrunedAt = t
769747 pr <- pruneExpired t
770748 forwards <- *forwards
771− ackedIds <- *ackedIds
772− let pendingAcks = do a <- ackedIds; where not (isLocalServer a.originServer); yield a
749+ pendingAcks <- &pendingRemoteAcks
773750 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks}
774751 broadcastOfflineGossip t result.peerServers result.expiredLocal
775752 when federationEnabled (do
@@ -846,6 +823,59 @@ forwardCandidates = \targets messages forwards t ->
846823 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
847824 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp}
848825
826+activeByExpiry = \items t -> do x <- items; where x.expiresAt > t; yield x
827+
828+challengeMatch = \challenges pubkey challenge t ->
829+ any challenges (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t)
830+
831+challengesExcluding = \challenges pubkey challenge ->
832+ do c <- challenges; where c.pubkey != pubkey || c.challenge != challenge; yield c
833+
834+hasLocalPresence = \entries -> any entries (\p -> isLocalServer p.server)
835+
836+&pendingRemoteAcks = do
837+ ackedIds <- *ackedIds
838+ yield (do a <- ackedIds; where not (isLocalServer a.originServer); yield a)
839+
840+ackedIdsExcludingBatch = \ackedIds server batch ->
841+ do a <- ackedIds; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a
842+
843+retriableEvictedAcks = \evictedEntries pendingAcks ->
844+ do a <- evictedEntries; where not (hasMessage a.messageId a.toKey pendingAcks); yield a
845+
846+acksForServer = \acks server -> do a <- acks; where a.originServer == server; yield a
847+
848+forwardsForServer = \forwards server -> do f <- forwards; where f.toServer == server; yield f
849+
850+trustedOfflineEvents = \events fromServer ->
851+ do e <- events; where e.server == fromServer; yield e
852+
853+gossipTargets = \peerServers fromServer ->
854+ do s <- peerServers; where s != fromServer; yield s
855+
856+newlyOnlineRemoteKeys = \events appliedKeys ->
857+ do e <- events; where any appliedKeys (\a -> samePubkeyServer a e); where not (isLocalServer e.server); yield {pubkey: e.pubkey, server: e.server}
858+
859+trustedGossipEntries = \deduped presence fromServer t ->
860+ do e <- deduped; where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t); yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp}
861+
862+appliedOnlineKeys = \entries presence ->
863+ do e <- entries; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield (pubkeyServer e)
864+
865+existingPresenceUpdates = \entries presence ->
866+ do e <- entries; where any presence (\p -> samePubkeyServer p e); yield e
867+
868+newPresenceEntries = \entries presence ->
869+ do e <- entries; where none presence (\p -> samePubkeyServer p e); yield e
870+
871+offlineGossipEvents = \expiredLocal -> do
872+ p <- expiredLocal
873+ let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
874+ yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.expiresAt, ttl: ttl}
875+
876+remoteOnlyTargets = \targets localOnline ->
877+ do target <- targets; where none localOnline (\pk -> pk == target.pubkey); yield target
878+
849879-- ============================================
850880-- Request Handler
851881-- ============================================
@@ -876,15 +906,12 @@ serve = \req -> case req of
876906 challenges <- *challenges
877907 sessions <- *sessions
878908 presence <- *presence
879− let hasMatch = any challenges (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t)
909+ let hasMatch = challengeMatch challenges pubkey challenge t
880910 let sigValid = hasMatch && verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
881911 if not sigValid
882912 then yield {pr: pr, verify: VerifyFailed {}}
883913 else do
884− let remainingChallenges = do
885− c <- challenges
886− where c.pubkey != pubkey || c.challenge != challenge
887− yield c
914+ let remainingChallenges = challengesExcluding challenges pubkey challenge
888915 set *challenges = remainingChallenges
889916 let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
890917 let updatedSessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
@@ -941,7 +968,7 @@ serve = \req -> case req of
941968 [{messageId: msgId, toKey: to,
942969 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
943970 let toPresence = presenceForKey presence to t
944− let localOnline = any toPresence (\p -> isLocalServer p.server)
971+ let localOnline = hasLocalPresence toPresence
945972 let remoteServers = collectPeerServers toPresence (\p -> p.server)
946973 forwards <- *forwards
947974 let forwardServers = takeRelation (maxForwardsEntries - count forwards) remoteServers
@@ -1022,7 +1049,7 @@ serve = \req -> case req of
10221049 withFederationAuth authorization t respond (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
10231050 let validated = validGossipEvents events t
10241051 let onlineEntries = validated.online
1025− let trustedOffline = do e <- validated.offline; where e.server == fromServer; yield e
1052+ let trustedOffline = trustedOfflineEvents validated.offline fromServer
10261053 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
10271054 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
10281055 let deduped = mergeByPubkeyServer [] filteredOnline
@@ -1035,18 +1062,10 @@ serve = \req -> case req of
10351062 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
10361063 when (not gossipResult.gm.withinCapacity)
10371064 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
1038− let newlyOnlineRemote = do
1039− e <- filteredOnline
1040− where any gossipResult.gm.appliedKeys (\a -> samePubkeyServer a e)
1041− where not (isLocalServer e.server)
1042− yield {pubkey: e.pubkey, server: e.server}
1065+ let newlyOnlineRemote = newlyOnlineRemoteKeys filteredOnline gossipResult.gm.appliedKeys
10431066 fork (do
10441067 when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (
1045− let targets = do
1046− s <- gossipResult.gm.gossipPeerServers
1047− where s != fromServer
1048− yield s
1049− in sendBatchedGossip targets effectiveEvents (cappedHops - 1))
1068+ sendBatchedGossip (gossipTargets gossipResult.gm.gossipPeerServers fromServer) effectiveEvents (cappedHops - 1))
10501069 broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal
10511070 when (federationEnabled && count newlyOnlineRemote > 0)
10521071 (forwardQueuedMessages newlyOnlineRemote t))