@@ -47,9 +47,6 @@ gossipOriginHops = 2
4747
4848backgroundPruneInterval : Int<Ms>
4949backgroundPruneInterval = 30000 -- 30 seconds
50−pollSleepInterval : Int<Ms>
51−pollSleepInterval = 500 -- 500ms between poll attempts
52−maxPollWaitAttempts = 60 -- 60 * 500ms = 30 second max long-poll
5350federationEnabled = federationSecret != ""
5451federationAuth = "Bearer " ++ federationSecret
5552federationUrl = \server -> federationScheme ++ server
@@ -103,14 +100,13 @@ type OptionalMessageId = Text where \s -> s == "" || isValidMessageId s
103100isValidServerName : Text -> Bool
104101isValidServerName = \s -> not (tooLong s maxServerNameLen) && s == normalizeServerName s && not (isBadServerName s)
105102type ServerName = Text where isValidServerName
106−data MessageOrigin = LocalOrigin {} | FederatedOrigin {server: ServerName}
107103type AnyServerName = Text where \s -> not (tooLong s maxServerNameLen) && s == normalizeServerName s
108104
109105type Timestamp = Int<Ms>
110106
111107type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp}
112108type 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}
114110type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
115111type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
116112type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
@@ -608,17 +604,18 @@ pruneExpired = \t -> do
608604 where p.expiresAt > t
609605 yield p
610606 set *presence = presenceAfter
607+ let messageCutoff = t - messageMaxAge
611608 set *messages = do
612609 m <- *messages
613− where m.receivedAt >= t - messageMaxAge
610+ where m.receivedAt > messageCutoff
614611 yield m
615612 set *forwards = do
616613 f <- *forwards
617− where f.createdAt >= t - messageMaxAge
614+ where f.createdAt > messageCutoff
618615 yield f
619616 set *ackedIds = do
620617 a <- *ackedIds
621− where a.ackedAt >= t - messageMaxAge
618+ where a.ackedAt > messageCutoff
622619 yield a
623620 yield {expiredLocal: (do
624621 p <- presenceBefore
@@ -708,10 +705,11 @@ withSessionAuth = \authorization t onUnauth onAuth -> do
708705
709706removeForward = \mid srv -> atomic do
710707 forwards <- *forwards
711− set *forwards = do
708+ let updatedForwards = do
712709 x <- forwards
713710 where not (x.messageId == mid && x.toServer == srv)
714711 yield x
712+ set *forwards = updatedForwards
715713
716714retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
717715retryForwardsForServer = \entries remaining ->
@@ -750,7 +748,8 @@ forwardQueuedMessages = \targets t -> do
750748 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
751749 toKey: m.toKey, fromKey: m.fromKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp}
752750 let capped = takeN candidates (maxForwardsEntries - count forwards)
753− set *forwards = union forwards capped
751+ let updatedForwards = union forwards capped
752+ set *forwards = updatedForwards
754753 yield capped
755754 let servers = collectPeerServers newForwards (\x -> x.toServer)
756755 sendToEach servers (\server ->
@@ -768,10 +767,11 @@ sendBatchedAcksToServer = \serverAcks server ->
768767 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
769768 whenOk result (atomic do
770769 ackedIds <- *ackedIds
771− set *ackedIds = do
770+ let updatedAckedIds = do
772771 a <- ackedIds
773772 where not (a.originServer == server && hasMessage a.messageId a.toKey batch)
774− yield a))
773+ yield a
774+ set *ackedIds = updatedAckedIds))
775775
776776sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
777777sendAcksAndClean = \pendingAcks ->
@@ -825,13 +825,14 @@ pollOnce = \token cursor -> do
825825 Authorized {pubkey} -> do
826826 seqCounter <- *seqCounter
827827 let clampedCursor = clamp 0 (max seqCounter t) cursor
828− atomic do
829− matchingMsgs <- do
830− m <- *messages
831− where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
832− yield m
833− let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
834− if count limitedMsgs == 0 then retry
828+ allMessages <- *messages
829+ let matchingMsgs = do
830+ m <- allMessages
831+ where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
832+ yield m
833+ let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
834+ if count limitedMsgs == 0
835+ then yield (PollOk {events: [], cursor: clampedCursor})
835836 else
836837 let evts = do
837838 m <- limitedMsgs
@@ -855,9 +856,10 @@ serve = \req -> case req of
855856 pr <- maybePrune t
856857 challenges <- *challenges
857858 let capped = capPerKeyAndGlobal challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t maxChallenges
858− set *challenges = union
859+ let updatedChallenges = union
859860 capped
860861 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
862+ set *challenges = updatedChallenges
861863 yield pr
862864 forkBroadcastOffline t pruneResult
863865 yield (respond {challenge: challenge})
@@ -876,12 +878,14 @@ serve = \req -> case req of
876878 if not sigValid
877879 then yield {pr: pr, verify: VerifyFailed {}}
878880 else do
879− set *challenges = do
881+ let remainingChallenges = do
880882 c <- challenges
881883 where c.pubkey != pubkey || c.challenge != challenge
882884 yield c
885+ set *challenges = remainingChallenges
883886 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
885889 let presenceWithout = do
886890 p <- presence
887891 where p.pubkey != pubkey || not (isLocalServer p.server)
@@ -932,9 +936,10 @@ serve = \req -> case req of
932936 seqCounter <- *seqCounter
933937 let seqTs = nextSeqTs t seqCounter
934938 set *seqCounter = seqTs
935− set *messages = union messages
939+ let updatedMessages = union messages
936940 [{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
938943 let toPresence = do p <- presence; where p.pubkey == to && p.expiresAt > t; yield p
939944 let localOnline = any toPresence (\p -> isLocalServer p.server)
940945 let remoteServers = collectPeerServers toPresence (\p -> p.server)
@@ -944,9 +949,10 @@ serve = \req -> case req of
944949 let delivery = if localOnline then DsDelivered {}
945950 else if shouldForward then DsFederated {}
946951 else DsQueued {}
947− set *forwards = if shouldForward
952+ let updatedForwards = if shouldForward
948953 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
949954 else forwards
955+ set *forwards = updatedForwards
950956 yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers, fromKey: fromKey}})
951957 case result.send of
952958 SendDone {delivery, remoteServers, forwardServers, fromKey} ->
@@ -980,17 +986,18 @@ serve = \req -> case req of
980986 m <- messages
981987 where m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid)
982988 yield m
983− set *messages = diff messages ackedMessages
989+ let remainingMessages = diff messages ackedMessages
990+ set *messages = remainingMessages
984991 let ackedFederated = do
985992 m <- ackedMessages
986− case m.origin of
987− FederatedOrigin {server} -> yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: server}
988− LocalOrigin {} -> []
993+ where m.originServer != ""
994+ yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
989995 ackedIds <- *ackedIds
990996 let combinedAckedIds = union ackedIds ackedFederated
991997 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
992998 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
9941001 yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}})
9951002 fork (do
9961003 case result.ack of
@@ -1075,9 +1082,10 @@ serve = \req -> case req of
10751082 seqCounter <- *seqCounter
10761083 let seqTs = nextSeqTs t seqCounter
10771084 set *seqCounter = seqTs
1078− set *messages = union messages
1085+ let updatedMessages = union messages
10791086 [{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
10811089 yield {pr: pr, forward: ForwardStored {}}
10821090 forkBroadcastOffline t result.pr
10831091 yield (respond {ok: case result.forward of
@@ -1091,10 +1099,11 @@ serve = \req -> case req of
10911099 pruneResult <- atomic do
10921100 pr <- maybePrune t
10931101 forwards <- *forwards
1094− set *forwards = do
1102+ let remainingForwards = do
10951103 f <- forwards
10961104 where not (f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid))
10971105 yield f
1106+ set *forwards = remainingForwards
10981107 yield pr
10991108 forkBroadcastOffline t pruneResult
11001109 yield (respond {ok: True {}}))