f9e7ad5 · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Extract inline queries into named helpers; make pendingRemoteAcks a & derived relation

f9e7ad5iko6 months ago

server.knot+85−66

compare with master

unifiedsplit

@@ -517,15 +517,15 @@ capGossipPresence = \args ->

517 if count args.afterOffline <= maxPresenceEntries

518 then {finalPresence: args.afterOffline, appliedOnline: args.appliedOnline, withinCapacity: True {}}

519 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

521 let updatesOnlyMerged = mergeByPubkeyServer args.presence updatesOnly in

522 let updatesOnlyAfterOffline = removeOffline updatesOnlyMerged args.trustedOffline args.t in

523 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

525 let reclaimedEntries = takeRelation reclaimSlots newEntries in

526 {finalPresence: union updatesOnlyAfterOffline reclaimedEntries,

527 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)

529 (do e <- reclaimedEntries; yield (pubkeyServer e)),

530 withinCapacity: False {}}

531

@@ -562,34 +562,21 @@ sendToEach = \servers action ->

562 sendToEach rest action

563

564pruneExpired = \t -> do

565+ challenges <- *challenges

565− set *challenges = do

566+ set *challenges = activeByExpiry challenges t

566− c <- *challenges

567+ sessions <- *sessions

567− where c.expiresAt > t

568− yield c

569− set *sessions = do

568+ set *sessions = activeByExpiry sessions t

570− s <- *sessions

571− where s.expiresAt > t

572− yield s

573 presenceBefore <- *presence

569 presenceBefore <- *presence

574 let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)

570 let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)

575− let presenceAfter = do

571+ let presenceAfter = activeByExpiry presenceBefore t

576− p <- presenceBefore

577− where p.expiresAt > t

578− yield p

579 set *presence = presenceAfter

572 set *presence = presenceAfter

580 let messageCutoff = t - messageMaxAge

573 let messageCutoff = t - messageMaxAge

581− set *messages = do

574+ messages <- *messages

582− m <- *messages

583− where m.receivedAt > messageCutoff

575+ set *messages = do m <- messages; where m.receivedAt > messageCutoff; yield m

584− yield m

576+ forwards <- *forwards

585− set *forwards = do

586− f <- *forwards

587− where f.createdAt > messageCutoff

577+ set *forwards = do f <- forwards; where f.createdAt > messageCutoff; yield f

588− yield f

578+ ackedIds <- *ackedIds

589− set *ackedIds = do

590− a <- *ackedIds

591− where a.ackedAt > messageCutoff

579+ set *ackedIds = do a <- ackedIds; where a.ackedAt > messageCutoff; yield a

592− yield a

593 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}

580 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}

594

581

595maybePrune = \t -> do

582maybePrune = \t -> do

@@ -627,10 +614,7 @@ sendBatchedGossip = \peerServers allEvents hops ->

627

614

628broadcastOfflineGossip = \t peerServers expiredLocal ->

615broadcastOfflineGossip = \t peerServers expiredLocal ->

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

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

630− let gossipEvents = mergeByKey (\a b -> a.pubkey == b.pubkey) (\x -> x.timestamp) [] (do

617+ let gossipEvents = mergeByKey (\a b -> a.pubkey == b.pubkey) (\x -> x.timestamp) [] (offlineGossipEvents expiredLocal)

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})

634 sendBatchedGossip peerServers gossipEvents gossipOriginHops)

618 sendBatchedGossip peerServers gossipEvents gossipOriginHops)

635

619

636forkBroadcastOffline = \t result ->

620forkBroadcastOffline = \t result ->

@@ -642,8 +626,8 @@ effectiveGossipEvents = \entries appliedKeys eventType -> do

642 yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}

626 yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}

643

627

644mergeGossipPresence = \presence deduped trustedOffline fromServer t ->

628mergeGossipPresence = \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

629+ let trustedDeduped = trustedGossipEntries deduped presence fromServer t in

646− let appliedOnline = do e <- trustedDeduped; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield (pubkeyServer e) in

630+ let appliedOnline = appliedOnlineKeys trustedDeduped presence in

647 let merged = mergeByPubkeyServer presence trustedDeduped in

631 let merged = mergeByPubkeyServer presence trustedDeduped in

648 let afterOffline = removeOffline merged trustedOffline t in

632 let afterOffline = removeOffline merged trustedOffline t in

649 let appliedOffline = do p <- diff merged afterOffline; yield (pubkeyServer p) in

633 let appliedOffline = do p <- diff merged afterOffline; yield (pubkeyServer p) in

@@ -696,7 +680,7 @@ retryPendingForwards : [ForwardEntry] -> IO {network} {}

696retryPendingForwards = \pending -> do

680retryPendingForwards = \pending -> do

697 let servers = collectPeerServers pending (\x -> x.toServer)

681 let servers = collectPeerServers pending (\x -> x.toServer)

698 sendToEach servers (\server ->

682 sendToEach servers (\server ->

699− retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)

683+ retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)

700 yield {}

684 yield {}

701

685

702forwardQueuedMessages = \targets t -> do

686forwardQueuedMessages = \targets t -> do

@@ -705,7 +689,7 @@ forwardQueuedMessages = \targets t -> do

705 forwards <- *forwards

689 forwards <- *forwards

706 presence <- *presence

690 presence <- *presence

707 let localOnline = localOnlineKeys presence t

691 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

709 let candidates = forwardCandidates remoteTargets messages forwards t

693 let candidates = forwardCandidates remoteTargets messages forwards t

710 let capped = takeRelation (maxForwardsEntries - count forwards) candidates

694 let capped = takeRelation (maxForwardsEntries - count forwards) candidates

711 let updatedForwards = union forwards capped

695 let updatedForwards = union forwards capped

@@ -713,7 +697,7 @@ forwardQueuedMessages = \targets t -> do

713 yield capped

697 yield capped

714 let servers = collectPeerServers newForwards (\x -> x.toServer)

698 let servers = collectPeerServers newForwards (\x -> x.toServer)

715 sendToEach servers (\server ->

699 sendToEach servers (\server ->

716− retryForwardsForServer (do x <- newForwards; where x.toServer == server; yield x) maxForwardRetriesPerServer)

700+ retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)

717

701

718fetchFederationAck = \server messageIds ->

702fetchFederationAck = \server messageIds ->

719 fetch (federationUrl server) (FederationAck {

703 fetch (federationUrl server) (FederationAck {

@@ -727,32 +711,26 @@ sendBatchedAcksToServer = \serverAcks server ->

727 result <- fetchFederationAck server (do a <- batch; yield a.messageId)

711 result <- fetchFederationAck server (do a <- batch; yield a.messageId)

728 whenOk result (atomic do

712 whenOk result (atomic do

729 ackedIds <- *ackedIds

713 ackedIds <- *ackedIds

730− let updatedAckedIds = do

714+ let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch

731− a <- ackedIds

732− where not (a.originServer == server && hasMessage a.messageId a.toKey batch)

733− yield a

734 set *ackedIds = updatedAckedIds))

715 set *ackedIds = updatedAckedIds))

735

716

736sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}

717sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}

737sendAcksAndClean = \pendingAcks ->

718sendAcksAndClean = \pendingAcks ->

738 let servers = collectPeerServers pendingAcks (\a -> a.originServer) in

719 let servers = collectPeerServers pendingAcks (\a -> a.originServer) in

739 sendToEach servers (\server ->

720 sendToEach servers (\server ->

740− sendBatchedAcksToServer (do a <- pendingAcks; where a.originServer == server; yield a) server)

721+ sendBatchedAcksToServer (acksForServer pendingAcks server) server)

741

722

742retryEvictedAcks = \evictedEntries pendingAcks ->

723retryEvictedAcks = \evictedEntries pendingAcks ->

743− let evictedToRetry = do

724+ let evictedToRetry = retriableEvictedAcks evictedEntries pendingAcks

744− a <- evictedEntries

745− where not (hasMessage a.messageId a.toKey pendingAcks)

746− yield a

747 in do

725 in do

748 let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)

726 let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)

749 sendToEach evictedOriginServers (\server -> do

727 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)

751 case fetchResult of

729 case fetchResult of

752 Err _ -> do

730 Err _ -> do

753 dropped <- atomic do

731 dropped <- atomic do

754 ackedIds <- *ackedIds

732 ackedIds <- *ackedIds

755− let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a

733+ let reinsert = acksForServer evictedToRetry server

756 let combined = union ackedIds reinsert

734 let combined = union ackedIds reinsert

757 let excess = count combined - maxAckedIds

735 let excess = count combined - maxAckedIds

758 let afterEviction = if excess <= 0 then combined

736 let afterEviction = if excess <= 0 then combined

@@ -768,8 +746,7 @@ pruneAndRetry = \t -> do

768 set *lastPrunedAt = t

746 set *lastPrunedAt = t

769 pr <- pruneExpired t

747 pr <- pruneExpired t

770 forwards <- *forwards

748 forwards <- *forwards

771− ackedIds <- *ackedIds

749+ pendingAcks <- &pendingRemoteAcks

772− let pendingAcks = do a <- ackedIds; where not (isLocalServer a.originServer); yield a

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

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

774 broadcastOfflineGossip t result.peerServers result.expiredLocal

751 broadcastOfflineGossip t result.peerServers result.expiredLocal

775 when federationEnabled (do

752 when federationEnabled (do

@@ -846,6 +823,59 @@ forwardCandidates = \targets messages forwards t ->

846 yield {messageId: m.messageId, toServer: target.server, createdAt: t,

823 yield {messageId: m.messageId, toServer: target.server, createdAt: t,

847 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp}

824 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp}

848

825

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+

849-- ============================================

879-- ============================================

850-- Request Handler

880-- Request Handler

851-- ============================================

881-- ============================================

@@ -876,15 +906,12 @@ serve = \req -> case req of

876 challenges <- *challenges

906 challenges <- *challenges

877 sessions <- *sessions

907 sessions <- *sessions

878 presence <- *presence

908 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

880 let sigValid = hasMatch && verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)

910 let sigValid = hasMatch && verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)

881 if not sigValid

911 if not sigValid

882 then yield {pr: pr, verify: VerifyFailed {}}

912 then yield {pr: pr, verify: VerifyFailed {}}

883 else do

913 else do

884− let remainingChallenges = do

914+ let remainingChallenges = challengesExcluding challenges pubkey challenge

885− c <- challenges

886− where c.pubkey != pubkey || c.challenge != challenge

887− yield c

888 set *challenges = remainingChallenges

915 set *challenges = remainingChallenges

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

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

890 let updatedSessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]

917 let updatedSessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]

@@ -941,7 +968,7 @@ serve = \req -> case req of

941 [{messageId: msgId, toKey: to,

968 [{messageId: msgId, toKey: to,

942 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]

969 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]

943 let toPresence = presenceForKey presence to t

970 let toPresence = presenceForKey presence to t

944− let localOnline = any toPresence (\p -> isLocalServer p.server)

971+ let localOnline = hasLocalPresence toPresence

945 let remoteServers = collectPeerServers toPresence (\p -> p.server)

972 let remoteServers = collectPeerServers toPresence (\p -> p.server)

946 forwards <- *forwards

973 forwards <- *forwards

947 let forwardServers = takeRelation (maxForwardsEntries - count forwards) remoteServers

974 let forwardServers = takeRelation (maxForwardsEntries - count forwards) remoteServers

@@ -1022,7 +1049,7 @@ serve = \req -> case req of

1022 withFederationAuth authorization t respond (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do

1049 withFederationAuth authorization t respond (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do

1023 let validated = validGossipEvents events t

1050 let validated = validGossipEvents events t

1024 let onlineEntries = validated.online

1051 let onlineEntries = validated.online

1025− let trustedOffline = do e <- validated.offline; where e.server == fromServer; yield e

1052+ let trustedOffline = trustedOfflineEvents validated.offline fromServer

1026 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline

1053 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline

1027 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries

1054 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries

1028 let deduped = mergeByPubkeyServer [] filteredOnline

1055 let deduped = mergeByPubkeyServer [] filteredOnline

@@ -1035,18 +1062,10 @@ serve = \req -> case req of

1035 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))

1062 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))

1036 when (not gossipResult.gm.withinCapacity)

1063 when (not gossipResult.gm.withinCapacity)

1037 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))

1064 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))

1038− let newlyOnlineRemote = do

1065+ let newlyOnlineRemote = newlyOnlineRemoteKeys filteredOnline gossipResult.gm.appliedKeys

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}

1043 fork (do

1066 fork (do

1044 when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (

1067 when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (

1045− let targets = do

1046− s <- gossipResult.gm.gossipPeerServers

1068+ sendBatchedGossip (gossipTargets gossipResult.gm.gossipPeerServers fromServer) effectiveEvents (cappedHops - 1))

1047− where s != fromServer

1048− yield s

1049− in sendBatchedGossip targets effectiveEvents (cappedHops - 1))

1050 broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal

1069 broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal

1051 when (federationEnabled && count newlyOnlineRemote > 0)

1070 when (federationEnabled && count newlyOnlineRemote > 0)

1052 (forwardQueuedMessages newlyOnlineRemote t))

1071 (forwardQueuedMessages newlyOnlineRemote t))