@@ -136,6 +136,14 @@ reconciliationFrequency = 10
136federationLatencyWindowMs : Int<Ms>
136federationLatencyWindowMs : Int<Ms>
137federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
137federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
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
139-- Background task intervals
147-- Background task intervals
140backgroundPruneInterval : Int<Ms>
148backgroundPruneInterval : Int<Ms>
141backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
149backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
@@ -293,6 +301,7 @@ type RecipientCount = {pubkey: PubkeyHex, count: Int}
293type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
301type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
294-- Tracks recent federation request (server, timestamp) pairs for replay detection
302-- Tracks recent federation request (server, timestamp) pairs for replay detection
295type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
303type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
304+type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
297-- ============================================
306-- ============================================
298-- Persisted State
307-- Persisted State
@@ -499,6 +508,12 @@ type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Time
499-- Tracks when /maintenance/snapshot was last exported (for staleness warnings)
508-- Tracks when /maintenance/snapshot was last exported (for staleness warnings)
500*lastSnapshotExportAt : Timestamp
509*lastSnapshotExportAt : Timestamp
511+-- Per-peer gossip event volume rate limiting
512+*gossipEventRates : [GossipEventRateEntry]
514+-- Global federation connection counter (tracks total concurrent outbound connections)
515+*federationActiveConnections : Int
502-- ============================================
517-- ============================================
504-- ============================================
519-- ============================================
@@ -527,7 +542,7 @@ route AuthApi where
527route ClientApi where
542route ClientApi where
528 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
543 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
529 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
544 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}
532 /messages
547 /messages
533 -- Send an E2E encrypted message to a recipient pubkey
548 -- Send an E2E encrypted message to a recipient pubkey
@@ -538,6 +553,10 @@ route ClientApi where
538 POST {to: PubkeyHex, messageId: MessageIdHex} /status headers {authorization: Text, clientIp: Text}
553 POST {to: PubkeyHex, messageId: MessageIdHex} /status headers {authorization: Text, clientIp: Text}
539 -> {status: Text, authorized: Bool}
554 -> {status: Text, authorized: Bool}
540 = MessageStatus
555 = 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}
541 -- Look up which server a pubkey is present on (for federation routing)
560 -- Look up which server a pubkey is present on (for federation routing)
542 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
561 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
543 -> {server: AnyServerName, authorized: Bool}
562 -> {server: AnyServerName, authorized: Bool}
@@ -886,7 +905,9 @@ generateFederationAuth = \scope -> do
886-- Verify an inbound federation token by callback to the sender's /federation/verify endpoint.
905-- Verify an inbound federation token by callback to the sender's /federation/verify endpoint.
887-- Single-use: the sender consumes the token on verification.
906-- Single-use: the sender consumes the token on verification.
888-- scope must match the scope the token was generated with; empty scope accepts any (backward compat).
907-- 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.
890verifyFederationCallback : Text -> Text -> Text -> IO {network} Bool
911verifyFederationCallback : Text -> Text -> Text -> IO {network} Bool
891verifyFederationCallback = \fromServer authorization scope ->
912verifyFederationCallback = \fromServer authorization scope ->
892 case extractToken authorization of
913 case extractToken authorization of
@@ -895,14 +916,8 @@ verifyFederationCallback = \fromServer authorization scope ->
895 result <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
916 result <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
896 case result of
917 case result of
897 Err e -> do
918 Err e -> do
898− println ("federation: verify callback to " ++ fromServer ++ " failed (attempt 1): " ++ show e)
919+ println ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
900− result2 <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
903− println ("federation: verify callback to " ++ fromServer ++ " failed (attempt 2): " ++ show e2)
904− yield (False {})
920+ yield (False {})
906 Ok r -> yield r.ok
921 Ok r -> yield r.ok
908-- ============================================
923-- ============================================
@@ -1357,12 +1372,21 @@ hasHungFederationRequest : Text -> Timestamp -> [FederationInFlightEntry] -> Boo
1357hasHungFederationRequest = \server t inFlight ->
1372hasHungFederationRequest = \server t inFlight ->
1358 any inFlight (\r -> r.server == server && r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs)
1373 any inFlight (\r -> r.server == server && r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs)
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.
1361fetchFederation = \server request -> do
1377fetchFederation = \server request -> do
1378+ -- Acquire global federation connection slot
1380+ ac <- *federationActiveConnections
1381+ set *federationActiveConnections = ac + 1
1362 markFederationInFlight server
1382 markFederationInFlight server
1363 startTime <- now
1383 startTime <- now
1364 result <- fetch (federationUrl server) request
1384 result <- fetch (federationUrl server) request
1365 clearFederationInFlight server
1385 clearFederationInFlight server
1386+ -- Release global federation connection slot
1388+ ac <- *federationActiveConnections
1389+ set *federationActiveConnections = max 0 (ac - 1)
1366 endTime <- now
1390 endTime <- now
1367 let latency = endTime - startTime
1391 let latency = endTime - startTime
1368 when (latency > federationFetchTimeoutMs)
1392 when (latency > federationFetchTimeoutMs)
@@ -1588,26 +1612,54 @@ sendToEachN = \servers action remaining ->
1588 Nothing {} -> yield {}
1612 Nothing {} -> yield {}
1589 Just {value: server} -> do
1613 Just {value: server} -> do
1590 let rest = do s <- servers; where s != server; yield s
1614 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
1593 fork (action server)
1620 fork (action server)
1594− sendToEachN rest action (remaining - 1)
1621+ sendToEachN rest action (effectiveRemaining - 1)
1596 action server
1623 action server
1597 sendToEachN rest action 0
1624 sendToEachN rest action 0
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}
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}
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}
1599-- Remove all expired entries from every collection. Returns expired local presence
1651-- Remove all expired entries from every collection. Returns expired local presence
1600-- entries (for broadcasting offline gossip) and the set of peer servers known before pruning.
1652-- entries (for broadcasting offline gossip) and the set of peer servers known before pruning.
1601pruneExpired = \t -> do
1653pruneExpired = \t -> do
1602− -- Prune presence (must be atomic for before/after comparison)
1654+ -- Prune presence per-bucket (16 smaller atomic blocks instead of one large one).
1603− presenceResult <- atomic do
1655+ -- Each bucket is independent: all entries for a pubkey hash to the same bucket,
1604− presenceBefore <- readAllPresence
1656+ -- so per-bucket before/after comparison is equivalent to the global check.
1605− let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)
1657+ presenceResults <- prunePresenceBuckets 0 t
1606− let presenceAfter = do x <- presenceBefore; where x.expiresAt > t; yield x
1658+ atomic do
1607− writeAllPresence presenceAfter
1659+ tpc <- *totalPresenceCount
1608− set *totalPresenceCount = count presenceAfter
1660+ set *totalPresenceCount = max 0 (tpc - presenceResults.totalRemoved)
1609− set *knownPeerServers = collectPeerServers presenceAfter (\p -> p.server)
1661+ set *knownPeerServers = presenceResults.keptPeerServers
1610− yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
1662+ let presenceResult = {expiredLocal: presenceResults.expiredLocal, peerServers: presenceResults.allPeerServers}
1611 -- Prune auth state
1663 -- Prune auth state
1612 atomic do
1664 atomic do
1613 challenges <- *challenges
1665 challenges <- *challenges
@@ -1700,6 +1752,13 @@ pruneExpired = \t -> do
1700 -- Prune expired forward nonces (extended replay protection)
1752 -- Prune expired forward nonces (extended replay protection)
1701 forwardNonces <- *forwardNonces
1753 forwardNonces <- *forwardNonces
1702 set *forwardNonces = do n <- forwardNonces; where t - n.seenAt < forwardNonceRetention; yield n
1754 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
1703 yield presenceResult
1762 yield presenceResult
@@ -2215,7 +2274,22 @@ waitForInFlightDrain = \deadline -> do
2215pruneAndRetry = \t -> do
2274pruneAndRetry = \t -> do
2216 atomic do set *lastPrunedAt = t
2275 atomic do set *lastPrunedAt = t
2217 pr <- pruneExpired t
2276 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
2219 pc <- *pruneCount
2293 pc <- *pruneCount
2220 when (pc > 0 && pc - (pc / reconciliationFrequency) * reconciliationFrequency == 0) (do
2294 when (pc > 0 && pc - (pc / reconciliationFrequency) * reconciliationFrequency == 0) (do
2221 msgDrift <- atomic do
2295 msgDrift <- atomic do
@@ -2225,18 +2299,8 @@ pruneAndRetry = \t -> do
2225 set *totalMessageCount = actual
2299 set *totalMessageCount = actual
2226 set *recipientMsgCounts = countByRecipient allMsgs
2300 set *recipientMsgCounts = countByRecipient allMsgs
2227 yield (actual - stored)
2301 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)
2302+ when (msgDrift != 0)
2239− (println ("reconciliation: corrected drift — messages: " ++ show msgDrift ++ ", presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift)))
2303+ (println ("reconciliation: corrected message drift: " ++ show msgDrift)))
2240 pendingForwards <- atomic do f <- *forwards; yield f
2304 pendingForwards <- atomic do f <- *forwards; yield f
2241 ackState <- atomic do
2305 ackState <- atomic do
2242 ackedIds2 <- *ackedIds
2306 ackedIds2 <- *ackedIds
@@ -3111,6 +3175,27 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
3111 t <- now
3175 t <- now
3112 let cappedHops = clamp 0 maxGossipHops hopsRemaining
3176 let cappedHops = clamp 0 maxGossipHops hopsRemaining
3113 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
3177 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 {})
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
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"})
3114 let validated = validGossipEvents events t
3199 let validated = validGossipEvents events t
3115 let onlineEntries = validated.online
3200 let onlineEntries = validated.online
3116 let trustedOffline = trustedOfflineEvents validated.offline fromServer
3201 let trustedOffline = trustedOfflineEvents validated.offline fromServer
@@ -3191,6 +3276,16 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3191 if t - timestamp > messageMaxAge
3276 if t - timestamp > messageMaxAge
3192 then yield (respond {ok: False {}, error: "stale_timestamp"})
3277 then yield (respond {ok: False {}, error: "stale_timestamp"})
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"})
3194 let clampedTs = clampTimestamp timestamp t
3289 let clampedTs = clampTimestamp timestamp t
3195 result <- atomic do
3290 result <- atomic do
3196 let msgBucket = pubkeyBucket toKey
3291 let msgBucket = pubkeyBucket toKey
@@ -3526,6 +3621,15 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw
3526 isDraining <- *draining
3621 isDraining <- *draining
3527 if not isDraining then yield (respond {ok: False {}, error: "server_must_be_draining", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
3622 if not isDraining then yield (respond {ok: False {}, error: "server_must_be_draining", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
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"))
3529 result <- atomic do
3633 result <- atomic do
3530 -- Merge presence (existing entries win on conflict via freshness)
3634 -- Merge presence (existing entries win on conflict via freshness)
3531 allPresence <- readAllPresence
3635 allPresence <- readAllPresence
@@ -3534,13 +3638,13 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw
3534 writeAllPresence cappedPresence
3638 writeAllPresence cappedPresence
3535 set *totalPresenceCount = count cappedPresence
3639 set *totalPresenceCount = count cappedPresence
3536 set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)
3640 set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)
3537− -- Union dedup state, respecting capacity limits
3641+ -- Union dedup state, respecting capacity limits (using validated data)
3538 ackedIds <- *ackedIds
3642 ackedIds <- *ackedIds
3539− let mergedAcked = union ackedIds ackedIdsData
3643+ let mergedAcked = union ackedIds validAcked
3540 let cappedAcked = takeRelation maxAckedIds mergedAcked
3644 let cappedAcked = takeRelation maxAckedIds mergedAcked
3541 set *ackedIds = cappedAcked
3645 set *ackedIds = cappedAcked
3542 ackOverflow <- *ackOverflow
3646 ackOverflow <- *ackOverflow
3543− let mergedOverflow = union ackOverflow ackOverflowData
3647+ let mergedOverflow = union ackOverflow validOverflow
3544 set *ackOverflow = takeRelation maxAckOverflow mergedOverflow
3648 set *ackOverflow = takeRelation maxAckOverflow mergedOverflow
3545 deliveredIds <- *deliveredIds
3649 deliveredIds <- *deliveredIds
3546 let mergedDelivered = union deliveredIds deliveredIdsData
3650 let mergedDelivered = union deliveredIds deliveredIdsData
@@ -3551,15 +3655,56 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw
3551 let mergedFailures = mergeByKey (\a b -> a.server == b.server) (\x -> x.failedAt) forwardFailures forwardFailuresData
3655 let mergedFailures = mergeByKey (\a b -> a.server == b.server) (\x -> x.failedAt) forwardFailures forwardFailuresData
3552 let cappedFailures = takeRelation maxForwardFailures mergedFailures
3656 let cappedFailures = takeRelation maxForwardFailures mergedFailures
3553 set *forwardFailures = cappedFailures
3657 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)
3555 forwards <- *forwards
3659 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
3557 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
3661 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
3558 set *forwards = cappedForwards
3662 set *forwards = cappedForwards
3559 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
3663 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
3560 println ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
3664 println ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
3561 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
3665 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
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
3674+ globalRateLimited <- checkGlobalRate t
3675+ if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, ackedCount: 0})
3677+ withSessionAuth authorization t clientIp
3678+ (yield (respond {ok: False {}, authorized: False {}, ackedCount: 0}))
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}))
3563serve = \req -> case req of
3708serve = \req -> case req of
3564 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
3709 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
3565 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
3710 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
@@ -3568,6 +3713,7 @@ serve = \req -> case req of
3568 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond
3713 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond
3569 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond
3714 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond
3570 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
3715 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
3716+ AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
3571 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
3717 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
3572 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
3718 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
3573 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond
3719 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
3703 && maxAckRetries > 0 && maxPairRatePerPair > 0 && pairRateWindowMs > 0
3849 && maxAckRetries > 0 && maxPairRatePerPair > 0 && pairRateWindowMs > 0
3704 && maxFederationPairRatePerPair > 0 && federationPairRateWindowMs > 0
3850 && maxFederationPairRatePerPair > 0 && federationPairRateWindowMs > 0
3705 && reconciliationFrequency > 0 && maxForwardNonces > 0
3851 && reconciliationFrequency > 0 && maxForwardNonces > 0
3852+ && maxGossipEventsPerServer > 0 && gossipEventRateWindowMs > 0
3853+ && maxGlobalFederationConcurrency > 0
3706 -- Relationship invariants: these misconfigurations cause subtle bugs
3854 -- Relationship invariants: these misconfigurations cause subtle bugs
3707 -- (e.g., messages pruned before their stated max age, backoff that never increases)
3855 -- (e.g., messages pruned before their stated max age, backoff that never increases)
3708 let configRelationsOk = unackedMessageMaxAge <= messageMaxAge
3856 let configRelationsOk = unackedMessageMaxAge <= messageMaxAge
@@ -3723,10 +3871,12 @@ main = do
3723 && maxThirdPartyGossipHops <= maxGossipHops
3871 && maxThirdPartyGossipHops <= maxGossipHops
3724 && verificationTokenExpiry <= federationFetchTimeoutMs
3872 && verificationTokenExpiry <= federationFetchTimeoutMs
3725 && forwardNonceRetention >= messageMaxAge
3873 && forwardNonceRetention >= messageMaxAge
3874+ && maxGlobalFederationConcurrency >= maxFederationConcurrency
3875+ && maxGossipEventsPerServer > maxGossipEvents
3726 if not configOk
3876 if not configOk
3727 then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
3877 then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
3728 else if not configRelationsOk
3878 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"
3731 t <- now
3881 t <- now
3732 atomic do
3882 atomic do
@@ -3751,6 +3901,8 @@ main = do
3751 println (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
3901 println (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
3752 println (" pair rate limit: " ++ show maxPairRatePerPair ++ "/" ++ show pairRateWindowMs ++ "ms per sender-recipient")
3902 println (" pair rate limit: " ++ show maxPairRatePerPair ++ "/" ++ show pairRateWindowMs ++ "ms per sender-recipient")
3753 println (" federation pair rate limit: " ++ show maxFederationPairRatePerPair ++ "/" ++ show federationPairRateWindowMs ++ "ms per server-recipient")
3903 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)
3754 println (" ack retry limit: " ++ show maxAckRetries ++ " per entry")
3906 println (" ack retry limit: " ++ show maxAckRetries ++ " per entry")
3755 when (federationCircuitBreakerThreshold > 0)
3907 when (federationCircuitBreakerThreshold > 0)
3756 (println (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
3908 (println (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))