@@ -31,11 +31,13 @@ maxTokenLen = 256
3131maxChallenges = 10000
3232maxSessions = 10000
3333maxAckedIds = 10000
34+maxAckOverflow = 200
3435maxChallengesPerKey = 5
3536maxSessionsPerKey = 10
3637challengeExpiry : Int<Ms>
3738challengeExpiry = 60000 -- 1 minute
38−challengeRateLimit = 50
39+maxChallengeRatePerKey = 5
40+maxChallengeRateEntries = 1000
3941challengeRateWindowMs : Int<Ms>
4042challengeRateWindowMs = 1000
4143maxSendRatePerKey = 20
@@ -61,6 +63,8 @@ backgroundPruneInterval : Int<Ms>
6163backgroundPruneInterval = 30000 -- 30 seconds
6264pollTimeoutMs : Int<Ms>
6365pollTimeoutMs = 25000 -- long-poll heartbeat interval
66+maxPollsPerKey = 3
67+maxActivePolls = 5000
6468presenceRefreshInterval : Int<Ms>
6569presenceRefreshInterval = 300000 -- 5 minutes
6670presenceRefreshThreshold : Int<Ms>
@@ -130,6 +134,8 @@ type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: T
130134type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
131135type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}
132136type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
137+type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
138+type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
133139data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
134140data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName]} | SendDuplicate {}
135141data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
@@ -160,8 +166,7 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
160166
161167*lastPresenceRefresh : Timestamp
162168
163−*challengeRateTs : Timestamp
164−*challengeRateCount : Int
169+*challengeRates : [ChallengeRateEntry]
165170
166171*pollWakeup : Timestamp
167172
@@ -182,6 +187,10 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
182187*messageSignal14 : Timestamp
183188*messageSignal15 : Timestamp
184189
190+*activePolls : [ActivePoll]
191+
192+*ackOverflow : [AckedEntry]
193+
185194-- ============================================
186195-- Routes
187196-- ============================================
@@ -212,7 +221,7 @@ route FederationApi where
212221 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName} /gossip headers {authorization: Text}
213222 -> {ok: Bool}
214223 = RecvGossip
215− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex} /forward headers {authorization: Text}
224+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName} /forward headers {authorization: Text}
216225 -> {ok: Bool}
217226 = ForwardMessage
218227 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName} /ack headers {authorization: Text}
@@ -536,6 +545,14 @@ capGlobal = \items maxCount getKey ->
536545capPerKeyAndGlobal = \items matchKey maxPerKey getExpiry t maxGlobal ->
537546 capGlobal (capPerKey items matchKey maxPerKey getExpiry t) maxGlobal getExpiry
538547
548+removeOneForKey = \items pubkey ->
549+ let result = fold (\acc x ->
550+ if not acc.removed && x.pubkey == pubkey
551+ then {removed: True {}, items: acc.items}
552+ else {removed: acc.removed, items: union acc.items [x]}
553+ ) {removed: False {}, items: []} items
554+ in result.items
555+
539556collectPeerServers : [a] -> (a -> Text) -> [Text]
540557collectPeerServers = \items getServer ->
541558 fold (\acc item ->
@@ -635,6 +652,12 @@ pruneExpired = \t -> do
635652 set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r
636653 forwardFailures <- *forwardFailures
637654 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < retryBackoffMax * 2; yield f
655+ challengeRates <- *challengeRates
656+ set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r
657+ activePolls <- *activePolls
658+ set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p
659+ ackOverflow <- *ackOverflow
660+ set *ackOverflow = do a <- ackOverflow; where a.ackedAt > messageCutoff; yield a
638661 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
639662
640663maybePrune = \t -> do
@@ -760,7 +783,7 @@ retryForwardsForServer = \entries remaining ->
760783 result <- fetch (federationUrl f.toServer) (ForwardMessage {
761784 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
762785 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
763− authorization: federationAuth})
786+ fromServer: serverHostN, authorization: federationAuth})
764787 case result of
765788 Err _ -> do
766789 println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId)
@@ -796,11 +819,11 @@ forwardQueuedMessages = \targets t -> do
796819 sendToEach servers (\server ->
797820 retryForwardsForServer (forwardsForServer newForwards server) maxForwardRetriesPerServer)
798821
799−fetchFederationAck = \server messageIds ->
800− fetch (federationUrl server) (FederationAck {
822+fetchFederationAck = \origin messageIds ->
823+ fetch (federationUrl origin) (FederationAck {
801824 authorization: federationAuth,
802825 messageIds: messageIds,
803− originServer: server, fromServer: serverHostN})
826+ originServer: origin, fromServer: serverHostN})
804827
805828sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
806829sendBatchedAcksToServer = \serverAcks server ->
@@ -827,19 +850,40 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
827850 fetchResult <- fetchFederationAck server (do a <- acksForServer evictedToRetry server; yield a.messageId)
828851 case fetchResult of
829852 Err _ -> do
830− dropped <- atomic do
853+ overflow <- atomic do
831854 ackedIds <- *ackedIds
832855 let reinsert = acksForServer evictedToRetry server
833856 let combined = union ackedIds reinsert
834857 let excess = count combined - maxAckedIds
835− let afterEviction = if excess <= 0 then combined
836− else evictOldestAckedIds combined excess
837− set *ackedIds = afterEviction
838− yield excess
839− when (dropped > 0)
840− (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
858+ if excess <= 0
859+ then do set *ackedIds = combined; yield []
860+ else do
861+ let evicted = takeRelation excess (sortBy (\a -> a.ackedAt) combined)
862+ set *ackedIds = diff combined evicted
863+ yield evicted
864+ when (count overflow > 0) (do
865+ dropped <- atomic do
866+ ackOverflow <- *ackOverflow
867+ let updated = union ackOverflow overflow
868+ let overflowExcess = count updated - maxAckOverflow
869+ if overflowExcess <= 0
870+ then do set *ackOverflow = updated; yield 0
871+ else do set *ackOverflow = evictOldestAckedIds updated overflowExcess; yield overflowExcess
872+ when (dropped > 0)
873+ (println ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server)))
841874 Ok _ -> yield {})
842875
876+retryOverflowAcks = \overflow ->
877+ let servers = collectPeerServers overflow (\a -> a.originServer) in
878+ sendToEach servers (\server ->
879+ forEachBatch (acksForServer overflow server) maxMessageIds (\batch -> do
880+ result <- fetchFederationAck server (do a <- batch; yield a.messageId)
881+ case result of
882+ Err _ -> println ("federation: overflow ack to " ++ server ++ " failed")
883+ Ok _ -> atomic do
884+ ackOverflow <- *ackOverflow
885+ set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))
886+
843887pruneAndRetry = \t -> do
844888 result <- atomic do
845889 set *lastPrunedAt = t
@@ -847,17 +891,20 @@ pruneAndRetry = \t -> do
847891 forwards <- *forwards
848892 ackedIds2 <- *ackedIds
849893 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a
894+ ackOverflow <- *ackOverflow
895+ let pendingOverflow = do a <- ackOverflow; where not (isLocalServer a.originServer); yield a
850896 lastRefresh <- *lastPresenceRefresh
851897 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval
852898 presence <- *presence
853899 let localActive = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []
854900 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []
855901 when shouldRefresh (set *lastPresenceRefresh = t)
856− yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks, localActive: localActive, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
902+ yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks, pendingOverflow: pendingOverflow, localActive: localActive, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
857903 broadcastOfflineGossip t result.peerServers result.expiredLocal
858904 when federationEnabled (do
859905 retryPendingForwards result.pendingForwards
860− sendAcksAndClean result.pendingAcks)
906+ sendAcksAndClean result.pendingAcks
907+ retryOverflowAcks result.pendingOverflow)
861908 when result.shouldRefresh (do
862909 let refreshEvents = do p <- result.localActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.timestamp, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000)}
863910 when (count result.refreshPeers > 0 && count refreshEvents > 0)
@@ -959,31 +1006,51 @@ pollOnce = \token cursor -> do
9591006 case setup of
9601007 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
9611008 Just {value: s} -> do
962− -- Phase 1: check for existing messages (no blocking)
963− initialMessages <- *messages
964− let initial = pollableMessages initialMessages s.pubkey s.clampedCursor t
965− if count initial > 0
966− then do
967− result <- atomic do pollNewEvents s.pubkey s.clampedCursor t
968− yield {authorized: True {}, events: result.events, cursor: result.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
1009+ -- Acquire poll slot (limit concurrent polls per key)
1010+ acquired <- atomic do
1011+ activePolls <- *activePolls
1012+ let pollsForKey = count (do p <- activePolls; where p.pubkey == s.pubkey; yield p)
1013+ if pollsForKey >= maxPollsPerKey
1014+ then yield (False {})
1015+ else do
1016+ set *activePolls = if count activePolls >= maxActivePolls
1017+ then union (evictOneOldest activePolls (\p -> p.startedAt)) [{pubkey: s.pubkey, startedAt: t}]
1018+ else union activePolls [{pubkey: s.pubkey, startedAt: t}]
1019+ yield (True {})
1020+ if not acquired
1021+ then yield {authorized: True {}, events: [], cursor: cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
9691022 else do
970− -- Phase 2: wait for signal or heartbeat (no *messages dependency)
1023+ pollResult <- do
1024+ -- Phase 1: check for existing messages (no blocking)
1025+ initialMessages <- *messages
1026+ let initial = pollableMessages initialMessages s.pubkey s.clampedCursor t
1027+ if count initial > 0
1028+ then do
1029+ result <- atomic do pollNewEvents s.pubkey s.clampedCursor t
1030+ yield {authorized: True {}, events: result.events, cursor: result.cursor}
1031+ else do
1032+ -- Phase 2: wait for signal or heartbeat (no *messages dependency)
1033+ atomic do
1034+ wakeup <- *pollWakeup
1035+ signal <- readMessageSignal s.pubkey
1036+ if wakeup > s.startWakeup || signal > s.startSignal
1037+ then yield {}
1038+ else retry
1039+ -- Phase 3: re-authenticate and read messages after waking
1040+ t2 <- now
1041+ result <- atomic do
1042+ sessions2 <- *sessions
1043+ case authenticateSession sessions2 token t2 of
1044+ Unauthorized {} -> yield {authorized: False {}, events: [], cursor: s.clampedCursor}
1045+ Authorized {} -> do
1046+ r <- pollNewEvents s.pubkey s.clampedCursor t2
1047+ yield {authorized: True {}, events: r.events, cursor: r.cursor}
1048+ yield {authorized: result.authorized, events: result.events, cursor: result.cursor}
1049+ -- Release poll slot
9711050 atomic do
972− wakeup <- *pollWakeup
973− signal <- readMessageSignal s.pubkey
974− if wakeup > s.startWakeup || signal > s.startSignal
975− then yield {}
976− else retry
977− -- Phase 3: re-authenticate and read messages after waking
978− t2 <- now
979− result <- atomic do
980− sessions2 <- *sessions
981− case authenticateSession sessions2 token t2 of
982− Unauthorized {} -> yield {authorized: False {}, events: [], cursor: s.clampedCursor}
983− Authorized {} -> do
984− r <- pollNewEvents s.pubkey s.clampedCursor t2
985− yield {authorized: True {}, events: r.events, cursor: r.cursor}
986− yield {authorized: result.authorized, events: result.events, cursor: result.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
1051+ activePolls <- *activePolls
1052+ set *activePolls = removeOneForKey activePolls s.pubkey
1053+ yield {authorized: pollResult.authorized, events: pollResult.events, cursor: pollResult.cursor, pendingAcks: s.pendingAcks, evictedEntries: s.evictedEntries}
9871054
9881055-- ============================================
9891056-- Derived Relations
@@ -1093,19 +1160,21 @@ serve = \req -> case req of
10931160 else if count active >= maxChallenges
10941161 then yield {pr: pr, accepted: False {}, rateLimited: False {}}
10951162 else do
1096− rateTs <- *challengeRateTs
1097− rateCount <- *challengeRateCount
1098− let newWindow = t - rateTs >= challengeRateWindowMs
1099− let currentCount = if newWindow then 0 else rateCount
1100− if currentCount >= challengeRateLimit
1163+ challengeRates <- *challengeRates
1164+ let senderEntry = headOf (do r <- challengeRates; where r.pubkey == pubkey; yield r)
1165+ let isNewWindow = case senderEntry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= challengeRateWindowMs
1166+ let rateCount = case senderEntry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count
1167+ if rateCount >= maxChallengeRatePerKey
11011168 then yield {pr: pr, accepted: False {}, rateLimited: True {}}
11021169 else do
1103− set *challengeRateTs = if newWindow then t else rateTs
1104− set *challengeRateCount = currentCount + 1
1170+ let newRates = if isNewWindow
1171+ then union (do r <- challengeRates; where r.pubkey != pubkey; yield r) [{pubkey: pubkey, windowStart: t, count: 1}]
1172+ else do r <- challengeRates; yield (if r.pubkey == pubkey then {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1} else r)
1173+ set *challengeRates = if count newRates > maxChallengeRateEntries then (do r <- newRates; where t - r.windowStart < challengeRateWindowMs; yield r) else newRates
11051174 set *challenges = union active [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
11061175 yield {pr: pr, accepted: True {}, rateLimited: False {}}
11071176 when result.rateLimited
1108− (println ("challenge: rate limited (" ++ show challengeRateLimit ++ "/s), rejecting for " ++ pubkey))
1177+ (println ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
11091178 when (not result.accepted && not result.rateLimited)
11101179 (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
11111180 forkBroadcastOffline t result.pr
@@ -1211,7 +1280,7 @@ serve = \req -> case req of
12111280 fwdResult <- fetch (federationUrl server) (ForwardMessage {
12121281 toKey: to, encryptedBlob: encryptedBlob,
12131282 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
1214− authorization: federationAuth})
1283+ fromServer: serverHostN, authorization: federationAuth})
12151284 case fwdResult of
12161285 Err _ -> do
12171286 println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId)
@@ -1269,9 +1338,9 @@ serve = \req -> case req of
12691338 (forwardQueuedMessages newlyOnlineRemote t))
12701339 yield (respond {ok: True {}}))
12711340
1272− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
1341+ ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, authorization, respond} -> do
12731342 t <- now
1274− withFederationAuth authorization t respond (not (isLocalServer originServer)) (do
1343+ withFederationAuth authorization t respond (not (isLocalServer originServer) && fromServer == originServer) (do
12751344 let clampedTs = clampTimestamp timestamp t
12761345 result <- atomic do
12771346 pr <- maybePrune t