@@ -35,6 +35,13 @@ maxChallengesPerKey = 5
35maxSessionsPerKey = 10
35maxSessionsPerKey = 10
36challengeExpiry : Int<Ms>
36challengeExpiry : Int<Ms>
37challengeExpiry = 60000 -- 1 minute
37challengeExpiry = 60000 -- 1 minute
38+challengeRateLimit = 50
39+challengeRateWindowMs : Int<Ms>
40+challengeRateWindowMs = 1000
41+maxSendRatePerKey = 20
42+sendRateWindowMs : Int<Ms>
43+sendRateWindowMs = 1000
44+maxSendRateEntries = 1000
38sessionExpiry : Int<Ms>
45sessionExpiry : Int<Ms>
39sessionExpiry = 3600000 -- 1 hour
46sessionExpiry = 3600000 -- 1 hour
40onlineGossipTtl : Int<S>
47onlineGossipTtl : Int<S>
@@ -44,11 +51,18 @@ offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events
44maxForwardRetriesPerServer = 100
51maxForwardRetriesPerServer = 100
45maxGossipHops = 5
52maxGossipHops = 5
46gossipOriginHops = 2
53gossipOriginHops = 2
54+retryBackoffBase : Int<Ms>
55+retryBackoffBase = 30000 -- 30 seconds
56+retryBackoffMax : Int<Ms>
57+retryBackoffMax = 480000 -- 8 minutes
58+maxForwardFailures = 1000
48backgroundPruneInterval : Int<Ms>
60backgroundPruneInterval : Int<Ms>
49backgroundPruneInterval = 30000 -- 30 seconds
61backgroundPruneInterval = 30000 -- 30 seconds
50pollTimeoutMs : Int<Ms>
62pollTimeoutMs : Int<Ms>
51pollTimeoutMs = 25000 -- long-poll heartbeat interval
63pollTimeoutMs = 25000 -- long-poll heartbeat interval
64+presenceRefreshInterval : Int<Ms>
65+presenceRefreshInterval = 300000 -- 5 minutes
52federationEnabled = federationSecret != ""
66federationEnabled = federationSecret != ""
53federationAuth = "Bearer " ++ federationSecret
67federationAuth = "Bearer " ++ federationSecret
54federationUrl = \server -> federationScheme ++ server
68federationUrl = \server -> federationScheme ++ server
@@ -112,6 +126,8 @@ type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHe
112type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
126type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
113type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
127type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
114type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
128type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
129+type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}
130+type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
115data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
131data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
116data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName], forwardServers: [ServerName]} | SendDuplicate {}
132data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName], forwardServers: [ServerName]} | SendDuplicate {}
117data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp, pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
133data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp, pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
@@ -137,12 +153,33 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
138*lastPrunedAt : Timestamp
154*lastPrunedAt : Timestamp
156+*forwardFailures : [ForwardFailure]
158+*sendRates : [SendRateEntry]
160+*lastPresenceRefresh : Timestamp
162+*challengeRateTs : Timestamp
163+*challengeRateCount : Int
140*pollWakeup : Timestamp
165*pollWakeup : Timestamp
142*messageSignal0 : Timestamp
167*messageSignal0 : Timestamp
143*messageSignal1 : Timestamp
168*messageSignal1 : Timestamp
144*messageSignal2 : Timestamp
169*messageSignal2 : Timestamp
145*messageSignal3 : Timestamp
170*messageSignal3 : Timestamp
171+*messageSignal4 : Timestamp
172+*messageSignal5 : Timestamp
173+*messageSignal6 : Timestamp
174+*messageSignal7 : Timestamp
175+*messageSignal8 : Timestamp
176+*messageSignal9 : Timestamp
177+*messageSignal10 : Timestamp
178+*messageSignal11 : Timestamp
179+*messageSignal12 : Timestamp
180+*messageSignal13 : Timestamp
181+*messageSignal14 : Timestamp
182+*messageSignal15 : Timestamp
147-- ============================================
184-- ============================================
@@ -470,9 +507,17 @@ safeCompare = \a b ->
470 safeCompareLoop a b maxTokenLen 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0
507 safeCompareLoop a b maxTokenLen 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0
472evictOneOldest = \items getKey ->
509evictOneOldest = \items getKey ->
473− case minBy items getKey of
510+ let result = fold (\acc x ->
511+ case acc.candidate of
512+ Nothing {} -> {candidate: Just {value: x}, items: acc.items}
514+ if getKey x < getKey c
515+ then {candidate: Just {value: x}, items: union acc.items [c]}
516+ else {candidate: acc.candidate, items: union acc.items [x]}
517+ ) {candidate: Nothing {}, items: []} items
518+ in case result.candidate of
474 Nothing {} -> items
519 Nothing {} -> items
475− Just {value: v} -> diff items [v]
520+ Just {} -> result.items
477capPerKey = \items matchKey maxPerKey getExpiry t ->
522capPerKey = \items matchKey maxPerKey getExpiry t ->
478 let active = do x <- items; where getExpiry x > t; yield x in
523 let active = do x <- items; where getExpiry x > t; yield x in
@@ -585,6 +630,10 @@ pruneExpired = \t -> do
585 set *forwards = do f <- forwards; where f.createdAt > messageCutoff; yield f
630 set *forwards = do f <- forwards; where f.createdAt > messageCutoff; yield f
586 ackedIds <- *ackedIds
631 ackedIds <- *ackedIds
587 set *ackedIds = do a <- ackedIds; where a.ackedAt > messageCutoff; yield a
632 set *ackedIds = do a <- ackedIds; where a.ackedAt > messageCutoff; yield a
633+ sendRates <- *sendRates
634+ set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r
635+ forwardFailures <- *forwardFailures
636+ set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < retryBackoffMax * 2; yield f
588 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
637 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
590maybePrune = \t -> do
639maybePrune = \t -> do
@@ -671,6 +720,34 @@ removeForward = \mid srv -> atomic do
671 forwards <- *forwards
720 forwards <- *forwards
672 set *forwards = forwardsExcluding forwards srv [mid]
721 set *forwards = forwardsExcluding forwards srv [mid]
723+retryBackoff : Int -> Int<Ms>
724+retryBackoff = \failures ->
725+ if failures <= 1 then retryBackoffBase
726+ else if failures <= 2 then retryBackoffBase * 2
727+ else if failures <= 3 then retryBackoffBase * 4
728+ else if failures <= 4 then retryBackoffBase * 8
729+ else retryBackoffMax
731+shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
732+shouldRetryServer = \server t failures ->
733+ case headOf (do f <- failures; where f.server == server; yield f) of
734+ Nothing {} -> True {}
735+ Just {value: f} -> t - f.failedAt >= retryBackoff f.failures
737+recordFailure = \server -> do
740+ failures <- *forwardFailures
741+ let existing = headOf (do f <- failures; where f.server == server; yield f)
742+ let updated = case existing of
743+ Nothing {} -> union failures [{server: server, failedAt: t, failures: 1}]
744+ Just {value: f} -> union (do x <- failures; where x.server != server; yield x) [{server: server, failedAt: t, failures: min (f.failures + 1) 10}]
745+ set *forwardFailures = if count updated > maxForwardFailures then evictOneOldest updated (\x -> x.failedAt) else updated
747+clearFailure = \server -> atomic do
748+ failures <- *forwardFailures
749+ set *forwardFailures = do f <- failures; where f.server != server; yield f
674retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
751retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
675retryForwardsForServer = \entries remaining ->
752retryForwardsForServer = \entries remaining ->
676 if remaining <= 0
753 if remaining <= 0
@@ -684,14 +761,21 @@ retryForwardsForServer = \entries remaining ->
684 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
761 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
685 authorization: federationAuth})
762 authorization: federationAuth})
686 case result of
763 case result of
687− Err _ -> println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId)
765+ println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId)
766+ recordFailure f.toServer
688− Ok _ -> removeForward f.messageId f.toServer
768+ removeForward f.messageId f.toServer
769+ clearFailure f.toServer
689 retryForwardsForServer rest (remaining - 1)
770 retryForwardsForServer rest (remaining - 1)
691retryPendingForwards : [ForwardEntry] -> IO {network} {}
772retryPendingForwards : [ForwardEntry] -> IO {network} {}
692retryPendingForwards = \pending -> do
773retryPendingForwards = \pending -> do
693 let servers = collectPeerServers pending (\x -> x.toServer)
774 let servers = collectPeerServers pending (\x -> x.toServer)
776+ failures <- *forwardFailures
777+ let eligibleServers = do s <- servers; where shouldRetryServer s t failures; yield s
694− sendToEach servers (\server ->
778+ sendToEach eligibleServers (\server ->
695 retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)
779 retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)
@@ -762,33 +846,69 @@ pruneAndRetry = \t -> do
762 forwards <- *forwards
846 forwards <- *forwards
763 ackedIds2 <- *ackedIds
847 ackedIds2 <- *ackedIds
764 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a
848 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a
849+ lastRefresh <- *lastPresenceRefresh
850+ let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval
851+ presence <- *presence
852+ let localActive = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p) else []
853+ let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []
854+ when shouldRefresh (set *lastPresenceRefresh = t)
765− yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks}
855+ yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks, localActive: localActive, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
766 broadcastOfflineGossip t result.peerServers result.expiredLocal
856 broadcastOfflineGossip t result.peerServers result.expiredLocal
767 when federationEnabled (do
857 when federationEnabled (do
768 retryPendingForwards result.pendingForwards
858 retryPendingForwards result.pendingForwards
769 sendAcksAndClean result.pendingAcks)
859 sendAcksAndClean result.pendingAcks)
860+ when result.shouldRefresh (do
861+ let refreshEvents = do p <- result.localActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.timestamp, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000)}
862+ when (count result.refreshPeers > 0 && count refreshEvents > 0)
863+ (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops))
865+hexCharIndex : Text -> Int
867+ if c == "0" then 0 else if c == "1" then 1 else if c == "2" then 2 else if c == "3" then 3
868+ else if c == "4" then 4 else if c == "5" then 5 else if c == "6" then 6 else if c == "7" then 7
869+ else if c == "8" then 8 else if c == "9" then 9 else if c == "a" then 10 else if c == "b" then 11
870+ else if c == "c" then 12 else if c == "d" then 13 else if c == "e" then 14 else 15
771pubkeyBucket : PubkeyHex -> Int
872pubkeyBucket : PubkeyHex -> Int
772−pubkeyBucket = \pubkey ->
873+pubkeyBucket = \pubkey -> hexCharIndex (take 1 pubkey)
773− let c = take 1 pubkey in
774− if contains c "0123" then 0
775− else if contains c "4567" then 1
776− else if contains c "89ab" then 2
779signalMessageForKey = \pubkey ts ->
875signalMessageForKey = \pubkey ts ->
780 let b = pubkeyBucket pubkey in
876 let b = pubkeyBucket pubkey in
781 if b == 0 then set *messageSignal0 = ts
877 if b == 0 then set *messageSignal0 = ts
782 else if b == 1 then set *messageSignal1 = ts
878 else if b == 1 then set *messageSignal1 = ts
783 else if b == 2 then set *messageSignal2 = ts
879 else if b == 2 then set *messageSignal2 = ts
784− else set *messageSignal3 = ts
880+ else if b == 3 then set *messageSignal3 = ts
881+ else if b == 4 then set *messageSignal4 = ts
882+ else if b == 5 then set *messageSignal5 = ts
883+ else if b == 6 then set *messageSignal6 = ts
884+ else if b == 7 then set *messageSignal7 = ts
885+ else if b == 8 then set *messageSignal8 = ts
886+ else if b == 9 then set *messageSignal9 = ts
887+ else if b == 10 then set *messageSignal10 = ts
888+ else if b == 11 then set *messageSignal11 = ts
889+ else if b == 12 then set *messageSignal12 = ts
890+ else if b == 13 then set *messageSignal13 = ts
891+ else if b == 14 then set *messageSignal14 = ts
892+ else set *messageSignal15 = ts
786readMessageSignal = \pubkey ->
894readMessageSignal = \pubkey ->
787 let b = pubkeyBucket pubkey in
895 let b = pubkeyBucket pubkey in
788 if b == 0 then do s <- *messageSignal0; yield s
896 if b == 0 then do s <- *messageSignal0; yield s
789 else if b == 1 then do s <- *messageSignal1; yield s
897 else if b == 1 then do s <- *messageSignal1; yield s
790 else if b == 2 then do s <- *messageSignal2; yield s
898 else if b == 2 then do s <- *messageSignal2; yield s
791− else do s <- *messageSignal3; yield s
899+ else if b == 3 then do s <- *messageSignal3; yield s
900+ else if b == 4 then do s <- *messageSignal4; yield s
901+ else if b == 5 then do s <- *messageSignal5; yield s
902+ else if b == 6 then do s <- *messageSignal6; yield s
903+ else if b == 7 then do s <- *messageSignal7; yield s
904+ else if b == 8 then do s <- *messageSignal8; yield s
905+ else if b == 9 then do s <- *messageSignal9; yield s
906+ else if b == 10 then do s <- *messageSignal10; yield s
907+ else if b == 11 then do s <- *messageSignal11; yield s
908+ else if b == 12 then do s <- *messageSignal12; yield s
909+ else if b == 13 then do s <- *messageSignal13; yield s
910+ else if b == 14 then do s <- *messageSignal14; yield s
911+ else do s <- *messageSignal15; yield s
793pollEventsAndDelete = \pubkey clampedCursor t -> do
913pollEventsAndDelete = \pubkey clampedCursor t -> do
794 messages <- *messages
914 messages <- *messages
@@ -817,30 +937,35 @@ pollEventsAndDelete = \pubkey clampedCursor t -> do
818pollOnce = \token cursor -> do
938pollOnce = \token cursor -> do
820− sessions <- *sessions
941+ sessions <- *sessions
821− let auth = authenticateSession sessions token t
942+ case authenticateSession sessions token t of
823− Unauthorized {} -> yield (PollUnauthorized {})
943+ Unauthorized {} -> yield (Nothing {})
824− Authorized {pubkey} -> do
944+ Authorized {pubkey} -> do
825− seqCounter <- *seqCounter
945+ seqCounter <- *seqCounter
826− let clampedCursor = clamp 0 (max seqCounter t) cursor
946+ let clampedCursor = clamp 0 (max seqCounter t) cursor
827− startWakeup <- *pollWakeup
947+ startWakeup <- *pollWakeup
828− startSignal <- readMessageSignal pubkey
948+ startSignal <- readMessageSignal pubkey
949+ yield (Just {value: {pubkey: pubkey, clampedCursor: clampedCursor, startWakeup: startWakeup, startSignal: startSignal}})
951+ Nothing {} -> yield (PollUnauthorized {})
952+ Just {value: s} -> do
829 -- Phase 1: check for existing messages (no blocking)
953 -- Phase 1: check for existing messages (no blocking)
830 initialMessages <- *messages
954 initialMessages <- *messages
831− let initial = pollableMessages initialMessages pubkey clampedCursor t
955+ let initial = pollableMessages initialMessages s.pubkey s.clampedCursor t
832 if count initial > 0
956 if count initial > 0
833− then atomic do pollEventsAndDelete pubkey clampedCursor t
957+ then atomic do pollEventsAndDelete s.pubkey s.clampedCursor t
835 -- Phase 2: wait for signal or heartbeat (no *messages dependency)
959 -- Phase 2: wait for signal or heartbeat (no *messages dependency)
836 atomic do
960 atomic do
837 wakeup <- *pollWakeup
961 wakeup <- *pollWakeup
838− signal <- readMessageSignal pubkey
962+ signal <- readMessageSignal s.pubkey
839− if wakeup > startWakeup || signal > startSignal
963+ if wakeup > s.startWakeup || signal > s.startSignal
840 then yield {}
964 then yield {}
841 else retry
965 else retry
842 -- Phase 3: read messages after waking, delete atomically
966 -- Phase 3: read messages after waking, delete atomically
843− atomic do pollEventsAndDelete pubkey clampedCursor t
968+ atomic do pollEventsAndDelete s.pubkey s.clampedCursor t2
845-- ============================================
970-- ============================================
846-- Derived Relations
971-- Derived Relations
@@ -874,13 +999,14 @@ forwardsExcluding = \forwards server messageIds ->
874 do f <- forwards; where not (f.toServer == server && any messageIds (\mid -> f.messageId == mid)); yield f
999 do f <- forwards; where not (f.toServer == server && any messageIds (\mid -> f.messageId == mid)); yield f
876forwardCandidates = \targets messages forwards t ->
1001forwardCandidates = \targets messages forwards t ->
877− do
1002+ fold (\acc target ->
878− target <- targets
1003+ let serverFwdIds = do f <- forwards; where f.toServer == target.server; yield f.messageId in
879− m <- messages
1005+ m <- messages
880− where m.toKey == target.pubkey
1006+ where m.toKey == target.pubkey
881− where none forwards (\f -> f.messageId == m.messageId && f.toServer == target.server)
1007+ where none serverFwdIds (\fid -> fid == m.messageId)
882− yield {messageId: m.messageId, toServer: target.server, createdAt: t,
1008+ yield {messageId: m.messageId, toServer: target.server, createdAt: t,
883− toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp}
1009+ toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp})) [] targets
886challengeMatch = \challenges pubkey challenge t ->
1012challengeMatch = \challenges pubkey challenge t ->
@@ -941,14 +1067,25 @@ serve = \req -> case req of
941 t <- now
1067 t <- now
942 result <- atomic do
1068 result <- atomic do
943 pr <- maybePrune t
1069 pr <- maybePrune t
944− challenges <- *challenges
1070+ rateTs <- *challengeRateTs
945− let perKeyCapped = capPerKey challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t
1071+ rateCount <- *challengeRateCount
946− if count perKeyCapped >= maxChallenges
1072+ let newWindow = t - rateTs >= challengeRateWindowMs
1073+ let currentCount = if newWindow then 0 else rateCount
1074+ if currentCount >= challengeRateLimit
947− then yield {pr: pr, accepted: False {}}
1075+ then yield {pr: pr, accepted: False {}, rateLimited: True {}}
1077+ set *challengeRateTs = if newWindow then t else rateTs
1078+ set *challengeRateCount = currentCount + 1
1079+ challenges <- *challenges
1080+ let perKeyCapped = capPerKey challenges (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt) t
1081+ if count perKeyCapped >= maxChallenges
1082+ then yield {pr: pr, accepted: False {}, rateLimited: False {}}
949− set *challenges = union perKeyCapped [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
1084+ set *challenges = union perKeyCapped [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
950− yield {pr: pr, accepted: True {}}
1085+ yield {pr: pr, accepted: True {}, rateLimited: False {}}
1086+ when result.rateLimited
1087+ (println ("challenge: rate limited (" ++ show challengeRateLimit ++ "/s), rejecting for " ++ pubkey))
951− when (not result.accepted)
1088+ when (not result.accepted && not result.rateLimited)
952 (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
1089 (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
953 forkBroadcastOffline t result.pr
1090 forkBroadcastOffline t result.pr
954 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
1091 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
@@ -1015,36 +1152,47 @@ serve = \req -> case req of
1015 pr <- maybePrune t
1152 pr <- maybePrune t
1016 withSessionAuth authorization t
1153 withSessionAuth authorization t
1017 (yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}})
1154 (yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}})
1018− (\_ -> do
1155+ (\senderKey -> do
1156+ sendRates <- *sendRates
1157+ let senderEntry = headOf (do r <- sendRates; where r.pubkey == senderKey; yield r)
1158+ let isNewWindow = case senderEntry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= sendRateWindowMs
1159+ let rateCount = case senderEntry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count
1160+ if rateCount >= maxSendRatePerKey
1161+ then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
1163+ let newRates = if isNewWindow
1164+ then union (do r <- sendRates; where r.pubkey != senderKey; yield r) [{pubkey: senderKey, windowStart: t, count: 1}]
1165+ else do r <- sendRates; yield (if r.pubkey == senderKey then {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1} else r)
1166+ set *sendRates = if count newRates > maxSendRateEntries then (do r <- newRates; where t - r.windowStart < sendRateWindowMs; yield r) else newRates
1019− messages <- *messages
1167+ messages <- *messages
1020− ackedIds <- *ackedIds
1168+ ackedIds <- *ackedIds
1021− if messageId != "" && (hasMessage messageId to messages || hasMessage messageId to ackedIds)
1169+ if messageId != "" && (hasMessage messageId to messages || hasMessage messageId to ackedIds)
1022− then yield {pr: pr, send: SendDuplicate {}}
1170+ then yield {pr: pr, send: SendDuplicate {}}
1023− else if countMessagesFor messages to t >= maxMessagesPerRecipient
1171+ else if countMessagesFor messages to t >= maxMessagesPerRecipient
1024− then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
1172+ then yield {pr: pr, send: SendFailed {status: MsRejected {}}}
1025− else do
1173+ else do
1026− presence <- *presence
1174+ presence <- *presence
1027− seqCounter <- *seqCounter
1175+ seqCounter <- *seqCounter
1028− let seqTs = nextSeqTs t seqCounter
1176+ let seqTs = nextSeqTs t seqCounter
1029− set *seqCounter = seqTs
1177+ set *seqCounter = seqTs
1030− set *messages = union messages
1178+ set *messages = union messages
1031− [{messageId: effectiveMsgId, toKey: to,
1179+ [{messageId: effectiveMsgId, toKey: to,
1032− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
1180+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
1033− signalMessageForKey to seqTs
1181+ signalMessageForKey to seqTs
1034− let toPresence = presenceForKey presence to t
1182+ let toPresence = presenceForKey presence to t
1035− let localOnline = hasLocalPresence toPresence
1183+ let localOnline = hasLocalPresence toPresence
1036− let remoteServers = collectPeerServers toPresence (\p -> p.server)
1184+ let remoteServers = collectPeerServers toPresence (\p -> p.server)
1037− forwards <- *forwards
1185+ forwards <- *forwards
1038− let forwardServers = takeRelation (maxForwardsEntries - count forwards) remoteServers
1186+ let forwardServers = takeRelation (maxForwardsEntries - count forwards) remoteServers
1039− let shouldForward = federationEnabled && count forwardServers > 0
1187+ let shouldForward = federationEnabled && count forwardServers > 0
1040− let delivery = if localOnline then DsDelivered {}
1188+ let delivery = if localOnline then DsDelivered {}
1041− else if shouldForward then DsFederated {}
1189+ else if shouldForward then DsFederated {}
1042− else DsQueued {}
1190+ else DsQueued {}
1043− let updatedForwards = if shouldForward
1191+ let updatedForwards = if shouldForward
1044− then union forwards (do s <- forwardServers; yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs})
1192+ then union forwards (do s <- forwardServers; yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs})
1045− else forwards
1193+ else forwards
1046− set *forwards = updatedForwards
1194+ set *forwards = updatedForwards
1047− yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers}})
1195+ yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers}})
1048 case result.send of
1196 case result.send of
1049 SendDone {delivery, remoteServers, forwardServers} ->
1197 SendDone {delivery, remoteServers, forwardServers} ->
1050 fork (do
1198 fork (do
@@ -1058,8 +1206,8 @@ serve = \req -> case req of
1058 Err _ -> println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)
1206 Err _ -> println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)
1059 Ok _ -> when (any forwardServers (\s -> s == server))
1207 Ok _ -> when (any forwardServers (\s -> s == server))
1060 (removeForward effectiveMsgId server))
1208 (removeForward effectiveMsgId server))
1061− when (isQueued delivery)
1209+ when (count forwardServers < count remoteServers)
1062− (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")))
1210+ (println ("sendMessage: " ++ show (count remoteServers - count forwardServers) ++ " remote server(s) not tracked for " ++ effectiveMsgId ++ " (forwards at capacity " ++ show maxForwardsEntries ++ ")")))
1063 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
1211 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
1064 _ -> forkBroadcastOffline t result.pr
1212 _ -> forkBroadcastOffline t result.pr
1065 yield (case result.send of
1213 yield (case result.send of
@@ -1182,7 +1330,7 @@ backgroundPrune = do
1182 jitter <- randomInt 5000
1330 jitter <- randomInt 5000
1183 sleep (backgroundPruneInterval + jitter)
1331 sleep (backgroundPruneInterval + jitter)
1184 t <- now
1332 t <- now
1185− pruneAndRetry t
1333+ fork (pruneAndRetry t)
1186 backgroundPrune
1334 backgroundPrune
1188pollHeartbeat = do
1336pollHeartbeat = do