@@ -47,9 +47,6 @@ gossipOriginHops = 2
48backgroundPruneInterval : Int<Ms>
48backgroundPruneInterval : Int<Ms>
49backgroundPruneInterval = 30000 -- 30 seconds
49backgroundPruneInterval = 30000 -- 30 seconds
50−pollSleepInterval : Int<Ms>
51−pollSleepInterval = 500 -- 500ms between poll attempts
52−maxPollWaitAttempts = 60 -- 60 * 500ms = 30 second max long-poll
53federationEnabled = federationSecret != ""
50federationEnabled = federationSecret != ""
54federationAuth = "Bearer " ++ federationSecret
51federationAuth = "Bearer " ++ federationSecret
55federationUrl = \server -> federationScheme ++ server
52federationUrl = \server -> federationScheme ++ server
@@ -103,14 +100,13 @@ type OptionalMessageId = Text where \s -> s == "" || isValidMessageId s
103isValidServerName : Text -> Bool
100isValidServerName : Text -> Bool
104isValidServerName = \s -> not (tooLong s maxServerNameLen) && s == normalizeServerName s && not (isBadServerName s)
101isValidServerName = \s -> not (tooLong s maxServerNameLen) && s == normalizeServerName s && not (isBadServerName s)
105type ServerName = Text where isValidServerName
102type ServerName = Text where isValidServerName
106−data MessageOrigin = LocalOrigin {} | FederatedOrigin {server: ServerName}
107type AnyServerName = Text where \s -> not (tooLong s maxServerNameLen) && s == normalizeServerName s
103type AnyServerName = Text where \s -> not (tooLong s maxServerNameLen) && s == normalizeServerName s
109type Timestamp = Int<Ms>
105type Timestamp = Int<Ms>
111type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp}
107type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp}
112type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp}
108type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp}
113−type Message = {messageId: MessageIdHex, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, origin: MessageOrigin, receivedAt: Timestamp}
109+type Message = {messageId: MessageIdHex, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp}
114type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
110type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
115type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
111type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
116type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
112type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
@@ -608,17 +604,18 @@ pruneExpired = \t -> do
608 where p.expiresAt > t
604 where p.expiresAt > t
610 set *presence = presenceAfter
606 set *presence = presenceAfter
607+ let messageCutoff = t - messageMaxAge
611 set *messages = do
608 set *messages = do
612 m <- *messages
609 m <- *messages
613− where m.receivedAt >= t - messageMaxAge
610+ where m.receivedAt > messageCutoff
615 set *forwards = do
612 set *forwards = do
616 f <- *forwards
613 f <- *forwards
617− where f.createdAt >= t - messageMaxAge
614+ where f.createdAt > messageCutoff
619 set *ackedIds = do
616 set *ackedIds = do
620 a <- *ackedIds
617 a <- *ackedIds
621− where a.ackedAt >= t - messageMaxAge
618+ where a.ackedAt > messageCutoff
623 yield {expiredLocal: (do
620 yield {expiredLocal: (do
624 p <- presenceBefore
621 p <- presenceBefore
@@ -708,10 +705,11 @@ withSessionAuth = \authorization t onUnauth onAuth -> do
709removeForward = \mid srv -> atomic do
706removeForward = \mid srv -> atomic do
710 forwards <- *forwards
707 forwards <- *forwards
711− set *forwards = do
708+ let updatedForwards = do
712 x <- forwards
709 x <- forwards
713 where not (x.messageId == mid && x.toServer == srv)
710 where not (x.messageId == mid && x.toServer == srv)
712+ set *forwards = updatedForwards
716retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
714retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
717retryForwardsForServer = \entries remaining ->
715retryForwardsForServer = \entries remaining ->
@@ -750,7 +748,8 @@ forwardQueuedMessages = \targets t -> do
750 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
748 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
751 toKey: m.toKey, fromKey: m.fromKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp}
749 toKey: m.toKey, fromKey: m.fromKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp}
752 let capped = takeN candidates (maxForwardsEntries - count forwards)
750 let capped = takeN candidates (maxForwardsEntries - count forwards)
753− set *forwards = union forwards capped
751+ let updatedForwards = union forwards capped
752+ set *forwards = updatedForwards
754 yield capped
753 yield capped
755 let servers = collectPeerServers newForwards (\x -> x.toServer)
754 let servers = collectPeerServers newForwards (\x -> x.toServer)
756 sendToEach servers (\server ->
755 sendToEach servers (\server ->
@@ -768,10 +767,11 @@ sendBatchedAcksToServer = \serverAcks server ->
768 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
767 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
769 whenOk result (atomic do
768 whenOk result (atomic do
770 ackedIds <- *ackedIds
769 ackedIds <- *ackedIds
771− set *ackedIds = do
770+ let updatedAckedIds = do
772 a <- ackedIds
771 a <- ackedIds
773 where not (a.originServer == server && hasMessage a.messageId a.toKey batch)
772 where not (a.originServer == server && hasMessage a.messageId a.toKey batch)
774− yield a))
773+ yield a
774+ set *ackedIds = updatedAckedIds))
776sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
776sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
777sendAcksAndClean = \pendingAcks ->
777sendAcksAndClean = \pendingAcks ->
@@ -825,13 +825,14 @@ pollOnce = \token cursor -> do
825 Authorized {pubkey} -> do
825 Authorized {pubkey} -> do
826 seqCounter <- *seqCounter
826 seqCounter <- *seqCounter
827 let clampedCursor = clamp 0 (max seqCounter t) cursor
827 let clampedCursor = clamp 0 (max seqCounter t) cursor
830− m <- *messages
828+ allMessages <- *messages
829+ let matchingMsgs = do
831− where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
831+ where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
833− let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
833+ let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
834− if count limitedMsgs == 0 then retry
834+ if count limitedMsgs == 0
835+ then yield (PollOk {events: [], cursor: clampedCursor})
836 let evts = do
837 let evts = do
837 m <- limitedMsgs
838 m <- limitedMsgs
@@ -855,9 +856,10 @@ serve = \req -> case req of
855 pr <- maybePrune t
856 pr <- maybePrune t
856 challenges <- *challenges
857 challenges <- *challenges
857 let capped = capPerKeyAndGlobal challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t maxChallenges
858 let capped = capPerKeyAndGlobal challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t maxChallenges
858− set *challenges = union
859+ let updatedChallenges = union
860 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
861 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
862+ set *challenges = updatedChallenges
862 forkBroadcastOffline t pruneResult
864 forkBroadcastOffline t pruneResult
863 yield (respond {challenge: challenge})
865 yield (respond {challenge: challenge})
@@ -876,12 +878,14 @@ serve = \req -> case req of
876 if not sigValid
878 if not sigValid
877 then yield {pr: pr, verify: VerifyFailed {}}
879 then yield {pr: pr, verify: VerifyFailed {}}
879− set *challenges = do
881+ let remainingChallenges = do
880 c <- challenges
882 c <- challenges
881 where c.pubkey != pubkey || c.challenge != challenge
883 where c.pubkey != pubkey || c.challenge != challenge
885+ set *challenges = remainingChallenges
883 let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
886 let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
884− set *sessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
887+ let updatedSessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
888+ set *sessions = updatedSessions
885 let presenceWithout = do
889 let presenceWithout = do
886 p <- presence
890 p <- presence
887 where p.pubkey != pubkey || not (isLocalServer p.server)
891 where p.pubkey != pubkey || not (isLocalServer p.server)
@@ -932,9 +936,10 @@ serve = \req -> case req of
932 seqCounter <- *seqCounter
936 seqCounter <- *seqCounter
933 let seqTs = nextSeqTs t seqCounter
937 let seqTs = nextSeqTs t seqCounter
934 set *seqCounter = seqTs
938 set *seqCounter = seqTs
935− set *messages = union messages
939+ let updatedMessages = union messages
936 [{messageId: msgId, toKey: to, fromKey: fromKey,
940 [{messageId: msgId, toKey: to, fromKey: fromKey,
937− encryptedBlob: encryptedBlob, timestamp: clampedTs, origin: LocalOrigin {}, receivedAt: seqTs}]
941+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
942+ set *messages = updatedMessages
938 let toPresence = do p <- presence; where p.pubkey == to && p.expiresAt > t; yield p
943 let toPresence = do p <- presence; where p.pubkey == to && p.expiresAt > t; yield p
939 let localOnline = any toPresence (\p -> isLocalServer p.server)
944 let localOnline = any toPresence (\p -> isLocalServer p.server)
940 let remoteServers = collectPeerServers toPresence (\p -> p.server)
945 let remoteServers = collectPeerServers toPresence (\p -> p.server)
@@ -944,9 +949,10 @@ serve = \req -> case req of
944 let delivery = if localOnline then DsDelivered {}
949 let delivery = if localOnline then DsDelivered {}
945 else if shouldForward then DsFederated {}
950 else if shouldForward then DsFederated {}
946 else DsQueued {}
951 else DsQueued {}
947− set *forwards = if shouldForward
952+ let updatedForwards = if shouldForward
948 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
953 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
949 else forwards
954 else forwards
955+ set *forwards = updatedForwards
950 yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers, fromKey: fromKey}})
956 yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers, fromKey: fromKey}})
951 case result.send of
957 case result.send of
952 SendDone {delivery, remoteServers, forwardServers, fromKey} ->
958 SendDone {delivery, remoteServers, forwardServers, fromKey} ->
@@ -980,17 +986,18 @@ serve = \req -> case req of
980 m <- messages
986 m <- messages
981 where m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid)
987 where m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid)
983− set *messages = diff messages ackedMessages
989+ let remainingMessages = diff messages ackedMessages
990+ set *messages = remainingMessages
984 let ackedFederated = do
991 let ackedFederated = do
985 m <- ackedMessages
992 m <- ackedMessages
986− case m.origin of
993+ where m.originServer != ""
987− FederatedOrigin {server} -> yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: server}
994+ yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
988− LocalOrigin {} -> []
989 ackedIds <- *ackedIds
995 ackedIds <- *ackedIds
990 let combinedAckedIds = union ackedIds ackedFederated
996 let combinedAckedIds = union ackedIds ackedFederated
991 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
997 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
992 if excessCount <= 0 then [] else boundedSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
998 if excessCount <= 0 then [] else boundedSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
993− set *ackedIds = diff combinedAckedIds evictedEntries
999+ let remainingAckedIds = diff combinedAckedIds evictedEntries
1000+ set *ackedIds = remainingAckedIds
994 yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}})
1001 yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}})
995 fork (do
1002 fork (do
996 case result.ack of
1003 case result.ack of
@@ -1075,9 +1082,10 @@ serve = \req -> case req of
1075 seqCounter <- *seqCounter
1082 seqCounter <- *seqCounter
1076 let seqTs = nextSeqTs t seqCounter
1083 let seqTs = nextSeqTs t seqCounter
1077 set *seqCounter = seqTs
1084 set *seqCounter = seqTs
1078− set *messages = union messages
1085+ let updatedMessages = union messages
1079 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1086 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1080− encryptedBlob: encryptedBlob, timestamp: clampedTs, origin: FederatedOrigin {server: originServer}, receivedAt: seqTs}]
1087+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1088+ set *messages = updatedMessages
1081 yield {pr: pr, forward: ForwardStored {}}
1089 yield {pr: pr, forward: ForwardStored {}}
1082 forkBroadcastOffline t result.pr
1090 forkBroadcastOffline t result.pr
1083 yield (respond {ok: case result.forward of
1091 yield (respond {ok: case result.forward of
@@ -1091,10 +1099,11 @@ serve = \req -> case req of
1091 pruneResult <- atomic do
1099 pruneResult <- atomic do
1092 pr <- maybePrune t
1100 pr <- maybePrune t
1093 forwards <- *forwards
1101 forwards <- *forwards
1094− set *forwards = do
1102+ let remainingForwards = do
1095 f <- forwards
1103 f <- forwards
1096 where not (f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid))
1104 where not (f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid))
1106+ set *forwards = remainingForwards
1098 yield pr
1107 yield pr
1099 forkBroadcastOffline t pruneResult
1108 forkBroadcastOffline t pruneResult
1100 yield (respond {ok: True {}}))
1109 yield (respond {ok: True {}}))