@@ -136,6 +136,14 @@ reconciliationFrequency = 10
136136federationLatencyWindowMs : Int<Ms>
137137federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
138138
139+-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
140+maxGossipEventsPerServer = 500 -- max total gossip events per server per window
141+gossipEventRateWindowMs : Int<Ms>
142+gossipEventRateWindowMs = 60000 -- 1 minute
143+maxGossipEventRateEntries = 100
144+-- Global federation connection cap (bounds total concurrent outbound connections across all paths)
145+maxGlobalFederationConcurrency = 30
146+
139147-- Background task intervals
140148backgroundPruneInterval : Int<Ms>
141149backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
@@ -293,6 +301,7 @@ type RecipientCount = {pubkey: PubkeyHex, count: Int}
293301type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
294302-- Tracks recent federation request (server, timestamp) pairs for replay detection
295303type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
304+type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
296305
297306-- ============================================
298307-- Persisted State
@@ -499,6 +508,12 @@ type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Time
499508-- Tracks when /maintenance/snapshot was last exported (for staleness warnings)
500509*lastSnapshotExportAt : Timestamp
501510
511+-- Per-peer gossip event volume rate limiting
512+*gossipEventRates : [GossipEventRateEntry]
513+
514+-- Global federation connection counter (tracks total concurrent outbound connections)
515+*federationActiveConnections : Int
516+
502517-- ============================================
503518-- Routes
504519-- ============================================
@@ -527,7 +542,7 @@ route AuthApi where
527542route ClientApi where
528543 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
529544 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
530− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}]}
545+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool}
531546 = Poll
532547 /messages
533548 -- Send an E2E encrypted message to a recipient pubkey
@@ -538,6 +553,10 @@ route ClientApi where
538553 POST {to: PubkeyHex, messageId: MessageIdHex} /status headers {authorization: Text, clientIp: Text}
539554 -> {status: Text, authorized: Bool}
540555 = MessageStatus
556+ -- Explicitly acknowledge received messages by ID (alternative to cursor-based ack in poll)
557+ POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text, clientIp: Text}
558+ -> {ok: Bool, authorized: Bool, ackedCount: Int}
559+ = AckMessages
541560 -- Look up which server a pubkey is present on (for federation routing)
542561 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
543562 -> {server: AnyServerName, authorized: Bool}
@@ -886,7 +905,9 @@ generateFederationAuth = \scope -> do
886905-- Verify an inbound federation token by callback to the sender's /federation/verify endpoint.
887906-- Single-use: the sender consumes the token on verification.
888907-- scope must match the scope the token was generated with; empty scope accepts any (backward compat).
889−-- Retries once on network error to handle transient failures (e.g., peer restarting).
908+-- No retry: a failed verify rejects the request. The peer's federation rate limit + circuit breaker
909+-- handle unreliable peers; retrying here doubles verify traffic and blocks with a 1s sleep per op,
910+-- which a malicious peer can exploit for amplification.
890911verifyFederationCallback : Text -> Text -> Text -> IO {network} Bool
891912verifyFederationCallback = \fromServer authorization scope ->
892913 case extractToken authorization of
@@ -895,14 +916,8 @@ verifyFederationCallback = \fromServer authorization scope ->
895916 result <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
896917 case result of
897918 Err e -> do
898− println ("federation: verify callback to " ++ fromServer ++ " failed (attempt 1): " ++ show e)
899− sleep 1000
900− result2 <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
901− case result2 of
902− Err e2 -> do
903− println ("federation: verify callback to " ++ fromServer ++ " failed (attempt 2): " ++ show e2)
904− yield (False {})
905− Ok r -> yield r.ok
919+ println ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
920+ yield (False {})
906921 Ok r -> yield r.ok
907922
908923-- ============================================
@@ -1357,12 +1372,21 @@ hasHungFederationRequest : Text -> Timestamp -> [FederationInFlightEntry] -> Boo
13571372hasHungFederationRequest = \server t inFlight ->
13581373 any inFlight (\r -> r.server == server && r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs)
13591374
1360−-- Wrapper around fetch for federation calls: tracks in-flight state, logs slow responses, records failures
1375+-- Wrapper around fetch for federation calls: tracks in-flight state, global connection count,
1376+-- logs slow responses, and records latency metrics.
13611377fetchFederation = \server request -> do
1378+ -- Acquire global federation connection slot
1379+ atomic do
1380+ ac <- *federationActiveConnections
1381+ set *federationActiveConnections = ac + 1
13621382 markFederationInFlight server
13631383 startTime <- now
13641384 result <- fetch (federationUrl server) request
13651385 clearFederationInFlight server
1386+ -- Release global federation connection slot
1387+ atomic do
1388+ ac <- *federationActiveConnections
1389+ set *federationActiveConnections = max 0 (ac - 1)
13661390 endTime <- now
13671391 let latency = endTime - startTime
13681392 when (latency > federationFetchTimeoutMs)
@@ -1588,26 +1612,54 @@ sendToEachN = \servers action remaining ->
15881612 Nothing {} -> yield {}
15891613 Just {value: server} -> do
15901614 let rest = do s <- servers; where s != server; yield s
1591− if remaining > 0
1615+ -- Check global federation connection cap: fall back to sequential when at limit
1616+ globalActive <- *federationActiveConnections
1617+ let effectiveRemaining = if globalActive >= maxGlobalFederationConcurrency then 0 else remaining
1618+ if effectiveRemaining > 0
15921619 then do
15931620 fork (action server)
1594− sendToEachN rest action (remaining - 1)
1621+ sendToEachN rest action (effectiveRemaining - 1)
15951622 else do
15961623 action server
15971624 sendToEachN rest action 0
15981625
1626+-- Per-bucket presence pruning: prune a single presence bucket atomically.
1627+-- Returns expired local entries (for offline gossip), peer servers in kept entries,
1628+-- all peer servers (before pruning), and count of removed entries.
1629+prunePresenceBucketN = \b t -> atomic do
1630+ bucket <- readPresenceBucket b
1631+ let kept = do x <- bucket; where x.expiresAt > t; yield x
1632+ let expired = do x <- bucket; where x.expiresAt <= t; yield x
1633+ -- Expired local entries: only those without a surviving local presence entry for the same key
1634+ -- (all entries for a pubkey live in the same bucket, so this per-bucket check is complete)
1635+ let expiredLocal = do x <- expired; where isLocalServer x.server && none kept (\k -> k.pubkey == x.pubkey && isLocalServer k.server); yield x
1636+ writePresenceBucket b kept
1637+ yield {expiredLocal: expiredLocal, keptPeerServers: collectPeerServers kept (\p -> p.server), allPeerServers: collectPeerServers bucket (\p -> p.server), removedCount: count expired}
1638+
1639+-- Recursively prune all 16 presence buckets and aggregate results.
1640+prunePresenceBuckets = \b t ->
1641+ if b > 15 then yield {expiredLocal: [], keptPeerServers: [], allPeerServers: [], totalRemoved: 0}
1642+ else do
1643+ r <- prunePresenceBucketN b t
1644+ rest <- prunePresenceBuckets (b + 1) t
1645+ let mergePeers = \a ps -> fold (\acc s -> if any acc (\x -> x == s) then acc else union acc [s]) a ps
1646+ yield {expiredLocal: union r.expiredLocal rest.expiredLocal,
1647+ keptPeerServers: mergePeers r.keptPeerServers rest.keptPeerServers,
1648+ allPeerServers: mergePeers r.allPeerServers rest.allPeerServers,
1649+ totalRemoved: r.removedCount + rest.totalRemoved}
1650+
15991651-- Remove all expired entries from every collection. Returns expired local presence
16001652-- entries (for broadcasting offline gossip) and the set of peer servers known before pruning.
16011653pruneExpired = \t -> do
1602− -- Prune presence (must be atomic for before/after comparison)
1603− presenceResult <- atomic do
1604− presenceBefore <- readAllPresence
1605− let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)
1606− let presenceAfter = do x <- presenceBefore; where x.expiresAt > t; yield x
1607− writeAllPresence presenceAfter
1608− set *totalPresenceCount = count presenceAfter
1609− set *knownPeerServers = collectPeerServers presenceAfter (\p -> p.server)
1610− yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
1654+ -- Prune presence per-bucket (16 smaller atomic blocks instead of one large one).
1655+ -- Each bucket is independent: all entries for a pubkey hash to the same bucket,
1656+ -- so per-bucket before/after comparison is equivalent to the global check.
1657+ presenceResults <- prunePresenceBuckets 0 t
1658+ atomic do
1659+ tpc <- *totalPresenceCount
1660+ set *totalPresenceCount = max 0 (tpc - presenceResults.totalRemoved)
1661+ set *knownPeerServers = presenceResults.keptPeerServers
1662+ let presenceResult = {expiredLocal: presenceResults.expiredLocal, peerServers: presenceResults.allPeerServers}
16111663 -- Prune auth state
16121664 atomic do
16131665 challenges <- *challenges
@@ -1700,6 +1752,13 @@ pruneExpired = \t -> do
17001752 -- Prune expired forward nonces (extended replay protection)
17011753 forwardNonces <- *forwardNonces
17021754 set *forwardNonces = do n <- forwardNonces; where t - n.seenAt < forwardNonceRetention; yield n
1755+ -- Prune gossip event rate entries
1756+ gossipEventRates <- *gossipEventRates
1757+ set *gossipEventRates = do r <- gossipEventRates; where t - r.windowStart < gossipEventRateWindowMs; yield r
1758+ -- Reconcile global federation connection counter with actual in-flight state
1759+ -- (self-heals drift from leaked slots where a forked task died without decrementing)
1760+ inFlightRec <- *federationInFlight
1761+ set *federationActiveConnections = fold (\acc r -> acc + r.count) 0 inFlightRec
17031762 yield presenceResult
17041763
17051764
@@ -2215,7 +2274,22 @@ waitForInFlightDrain = \deadline -> do
22152274pruneAndRetry = \t -> do
22162275 atomic do set *lastPrunedAt = t
22172276 pr <- pruneExpired t
2218− -- Periodic counter reconciliation: recount from actual data every N prune cycles to self-heal drift
2277+ -- Counter reconciliation: always reconcile presence and session counters (cheap: just
2278+ -- counting list lengths across buckets). Message reconciliation stays periodic because
2279+ -- it requires reading all messages and rebuilding per-recipient counts.
2280+ presDrift <- atomic do
2281+ actual <- countAllPresence
2282+ stored <- *totalPresenceCount
2283+ set *totalPresenceCount = actual
2284+ yield (actual - stored)
2285+ sessDrift <- atomic do
2286+ actual <- countAllSessions
2287+ stored <- *totalSessionCount
2288+ set *totalSessionCount = actual
2289+ yield (actual - stored)
2290+ when (presDrift != 0 || sessDrift != 0)
2291+ (println ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))
2292+ -- Message reconciliation is expensive (reads all 16 buckets + rebuilds recipient counts), keep periodic
22192293 pc <- *pruneCount
22202294 when (pc > 0 && pc - (pc / reconciliationFrequency) * reconciliationFrequency == 0) (do
22212295 msgDrift <- atomic do
@@ -2225,18 +2299,8 @@ pruneAndRetry = \t -> do
22252299 set *totalMessageCount = actual
22262300 set *recipientMsgCounts = countByRecipient allMsgs
22272301 yield (actual - stored)
2228− presDrift <- atomic do
2229− actual <- countAllPresence
2230− stored <- *totalPresenceCount
2231− set *totalPresenceCount = actual
2232− yield (actual - stored)
2233− sessDrift <- atomic do
2234− actual <- countAllSessions
2235− stored <- *totalSessionCount
2236− set *totalSessionCount = actual
2237− yield (actual - stored)
2238− when (msgDrift != 0 || presDrift != 0 || sessDrift != 0)
2239− (println ("reconciliation: corrected drift — messages: " ++ show msgDrift ++ ", presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift)))
2302+ when (msgDrift != 0)
2303+ (println ("reconciliation: corrected message drift: " ++ show msgDrift)))
22402304 pendingForwards <- atomic do f <- *forwards; yield f
22412305 ackState <- atomic do
22422306 ackedIds2 <- *ackedIds
@@ -3111,6 +3175,27 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
31113175 t <- now
31123176 let cappedHops = clamp 0 maxGossipHops hopsRemaining
31133177 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
3178+ -- Per-peer gossip event volume limiting: counts total events (not just requests) per server.
3179+ -- A single gossip request carries up to maxGossipEvents entries; without this, a peer
3180+ -- sending 50 req/s * 100 events = 5000 presence mutations/second.
3181+ eventRateLimited <- atomic do
3182+ ger <- *gossipEventRates
3183+ let eventCount = count events
3184+ let entry = headOf (do r <- ger; where r.server == fromServer; yield r)
3185+ let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= gossipEventRateWindowMs
3186+ let currentCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count
3187+ if currentCount + eventCount > maxGossipEventsPerServer
3188+ then yield (True {})
3189+ else do
3190+ let newEntries = if isNewWindow
3191+ then union (do r <- ger; where r.server != fromServer; yield r) [{server: fromServer, windowStart: t, count: eventCount}]
3192+ else do r <- ger; yield (if r.server == fromServer then {server: r.server, windowStart: r.windowStart, count: r.count + eventCount} else r)
3193+ set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs; yield r) else newEntries
3194+ yield (False {})
3195+ if eventRateLimited then do
3196+ println ("gossip: event volume rate limited for " ++ fromServer ++ " (" ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms)")
3197+ yield (respond {ok: False {}, error: "event_rate_limited"})
3198+ else do
31143199 let validated = validGossipEvents events t
31153200 let onlineEntries = validated.online
31163201 let trustedOffline = trustedOfflineEvents validated.offline fromServer
@@ -3191,6 +3276,16 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
31913276 if t - timestamp > messageMaxAge
31923277 then yield (respond {ok: False {}, error: "stale_timestamp"})
31933278 else do
3279+ -- Verify recipient has active presence on this server. Peers should only forward
3280+ -- to us if gossip indicated the key is here. Accepting messages for completely
3281+ -- unknown keys wastes mailbox slots that will never be drained by polling.
3282+ hasPresence <- atomic do
3283+ bkt <- readPresenceBucket (pubkeyBucket toKey)
3284+ yield (any bkt (\p -> p.pubkey == toKey && p.expiresAt > t))
3285+ if not hasPresence then do
3286+ println ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
3287+ yield (respond {ok: False {}, error: "no_presence"})
3288+ else do
31943289 let clampedTs = clampTimestamp timestamp t
31953290 result <- atomic do
31963291 let msgBucket = pubkeyBucket toKey
@@ -3526,6 +3621,15 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw
35263621 isDraining <- *draining
35273622 if not isDraining then yield (respond {ok: False {}, error: "server_must_be_draining", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
35283623 else do
3624+ -- Validate imported data: reject entries with invalid or self-referencing server names.
3625+ -- originServer in ForwardEntry is untyped Text (not ServerName), so not validated by the type system.
3626+ -- Also reject entries pointing to the local server (nonsensical: we don't forward to ourselves
3627+ -- or ack to ourselves).
3628+ let validForwards = do f <- forwardsData; where not (isLocalServer f.toServer) && (f.originServer == "" || not (isBadServerName f.originServer)) && not (isLocalServer f.originServer); yield f
3629+ let validAcked = do a <- ackedIdsData; where not (isLocalServer a.originServer); yield a
3630+ let validOverflow = do a <- ackOverflowData; where not (isLocalServer a.originServer); yield a
3631+ let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow)
3632+ when (rejectedCount > 0) (println ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))
35293633 result <- atomic do
35303634 -- Merge presence (existing entries win on conflict via freshness)
35313635 allPresence <- readAllPresence
@@ -3534,13 +3638,13 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw
35343638 writeAllPresence cappedPresence
35353639 set *totalPresenceCount = count cappedPresence
35363640 set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)
3537− -- Union dedup state, respecting capacity limits
3641+ -- Union dedup state, respecting capacity limits (using validated data)
35383642 ackedIds <- *ackedIds
3539− let mergedAcked = union ackedIds ackedIdsData
3643+ let mergedAcked = union ackedIds validAcked
35403644 let cappedAcked = takeRelation maxAckedIds mergedAcked
35413645 set *ackedIds = cappedAcked
35423646 ackOverflow <- *ackOverflow
3543− let mergedOverflow = union ackOverflow ackOverflowData
3647+ let mergedOverflow = union ackOverflow validOverflow
35443648 set *ackOverflow = takeRelation maxAckOverflow mergedOverflow
35453649 deliveredIds <- *deliveredIds
35463650 let mergedDelivered = union deliveredIds deliveredIdsData
@@ -3551,15 +3655,56 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw
35513655 let mergedFailures = mergeByKey (\a b -> a.server == b.server) (\x -> x.failedAt) forwardFailures forwardFailuresData
35523656 let cappedFailures = takeRelation maxForwardFailures mergedFailures
35533657 set *forwardFailures = cappedFailures
3554− -- Restore forward entries (dedup by messageId + toServer, keep fresher on conflict)
3658+ -- Restore forward entries (dedup by messageId + toServer, keep fresher on conflict; using validated data)
35553659 forwards <- *forwards
3556− let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) forwards forwardsData
3660+ let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) forwards validForwards
35573661 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
35583662 set *forwards = cappedForwards
35593663 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
35603664 println ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
35613665 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
35623666
3667+-- POST /messages/ack: Explicitly acknowledge received messages by ID.
3668+-- Alternative to the implicit cursor-based ack in poll. Benefits:
3669+-- 1. Reduces the window for duplicate delivery (ack immediately without waiting for next poll)
3670+-- 2. Selective ack: acknowledge specific messages without acking all messages up to a cursor
3671+-- Rate-limited via global rate limiter; requires session auth.
3672+handleAckMessages = \messageIds authorization clientIp respond -> do
3673+ t <- now
3674+ globalRateLimited <- checkGlobalRate t
3675+ if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, ackedCount: 0})
3676+ else
3677+ withSessionAuth authorization t clientIp
3678+ (yield (respond {ok: False {}, authorized: False {}, ackedCount: 0}))
3679+ (\pubkey -> do
3680+ let dedupedIds = dedupMessageIds messageIds
3681+ result <- atomic do
3682+ let b = pubkeyBucket pubkey
3683+ bucket <- readMessageBucket b
3684+ let toAck = do m <- bucket; where m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid); yield m
3685+ let ackCount = count toAck
3686+ if ackCount > 0 then do
3687+ writeMessageBucket b (diff bucket toAck)
3688+ totalCount <- *totalMessageCount
3689+ set *totalMessageCount = max 0 (totalCount - ackCount)
3690+ recipientCounts <- *recipientMsgCounts
3691+ set *recipientMsgCounts = adjustRecipientCount recipientCounts pubkey (0 - ackCount)
3692+ let ackedMsgIds = do m <- toAck; yield m.messageId
3693+ forwards <- *forwards
3694+ set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
3695+ let ackedFederated = federatedAckEntries toAck t
3696+ ackedIds <- *ackedIds
3697+ let combinedAckedIds = union ackedIds ackedFederated
3698+ let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
3699+ let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
3700+ set *ackedIds = eviction.kept
3701+ yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
3702+ else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}
3703+ fork (when federationEnabled (do
3704+ sendAcksAndClean result.pendingAcks
3705+ retryEvictedAcks result.evictedEntries result.pendingAcks))
3706+ yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount}))
3707+
35633708serve = \req -> case req of
35643709 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
35653710 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
@@ -3568,6 +3713,7 @@ serve = \req -> case req of
35683713 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond
35693714 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond
35703715 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
3716+ AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
35713717 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
35723718 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
35733719 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond
@@ -3703,6 +3849,8 @@ main = do
37033849 && maxAckRetries > 0 && maxPairRatePerPair > 0 && pairRateWindowMs > 0
37043850 && maxFederationPairRatePerPair > 0 && federationPairRateWindowMs > 0
37053851 && reconciliationFrequency > 0 && maxForwardNonces > 0
3852+ && maxGossipEventsPerServer > 0 && gossipEventRateWindowMs > 0
3853+ && maxGlobalFederationConcurrency > 0
37063854 -- Relationship invariants: these misconfigurations cause subtle bugs
37073855 -- (e.g., messages pruned before their stated max age, backoff that never increases)
37083856 let configRelationsOk = unackedMessageMaxAge <= messageMaxAge
@@ -3723,10 +3871,12 @@ main = do
37233871 && maxThirdPartyGossipHops <= maxGossipHops
37243872 && verificationTokenExpiry <= federationFetchTimeoutMs
37253873 && forwardNonceRetention >= messageMaxAge
3874+ && maxGlobalFederationConcurrency >= maxFederationConcurrency
3875+ && maxGossipEventsPerServer > maxGossipEvents
37263876 if not configOk
37273877 then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
37283878 else if not configRelationsOk
3729− then println "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, per-key limits <= global limits), server not started"
3879+ then println "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, per-key limits <= global limits), server not started"
37303880 else do
37313881 t <- now
37323882 atomic do
@@ -3751,6 +3901,8 @@ main = do
37513901 println (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
37523902 println (" pair rate limit: " ++ show maxPairRatePerPair ++ "/" ++ show pairRateWindowMs ++ "ms per sender-recipient")
37533903 println (" federation pair rate limit: " ++ show maxFederationPairRatePerPair ++ "/" ++ show federationPairRateWindowMs ++ "ms per server-recipient")
3904+ println (" gossip event rate limit: " ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms per server")
3905+ println (" global federation concurrency: " ++ show maxGlobalFederationConcurrency)
37543906 println (" ack retry limit: " ++ show maxAckRetries ++ " per entry")
37553907 when (federationCircuitBreakerThreshold > 0)
37563908 (println (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))