@@ -93,6 +93,7 @@ offlineGossipMinTtl : Int<S>
93offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events
93offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events
94maxFederationConcurrency = 10 -- max concurrent outbound federation requests per cycle
94maxFederationConcurrency = 10 -- max concurrent outbound federation requests per cycle
95maxForwardRetriesPerServer = 100
95maxForwardRetriesPerServer = 100
96+maxForwardRetries = 10 -- max retries per individual forward entry before dropping
96maxGossipHops = 5 -- max hops a gossip event can propagate across servers
97maxGossipHops = 5 -- max hops a gossip event can propagate across servers
97gossipOriginHops = 2 -- hops set when this server originates a gossip event
98gossipOriginHops = 2 -- hops set when this server originates a gossip event
98gossipDedupWindow : Int<Ms>
99gossipDedupWindow : Int<Ms>
@@ -114,6 +115,8 @@ maxFederationInFlight = 100
114-- servers have active (non-expired) failure records. Prevents wasted retries and log spam
115-- servers have active (non-expired) failure records. Prevents wasted retries and log spam
115-- when the local network is down. Set to 0 to disable.
116-- when the local network is down. Set to 0 to disable.
116federationCircuitBreakerThreshold = 3
117federationCircuitBreakerThreshold = 3
118+-- Federation replay dedup: track recent (server, timestamp) pairs to reject exact replays
119+maxRecentFederationRequests = 1000
118-- Federation latency metric window
121-- Federation latency metric window
119federationLatencyWindowMs : Int<Ms>
122federationLatencyWindowMs : Int<Ms>
@@ -122,6 +125,8 @@ federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after thi
122-- Background task intervals
125-- Background task intervals
123backgroundPruneInterval : Int<Ms>
126backgroundPruneInterval : Int<Ms>
124backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
127backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
128+autoSnapshotInterval : Int<Ms>
129+autoSnapshotInterval = 300000 -- 5 minutes; periodic state summary to stdout (0 to disable)
125pollTimeoutMs : Int<Ms>
130pollTimeoutMs : Int<Ms>
126pollTimeoutMs = 25000 -- long-poll heartbeat interval (how long a poll blocks before returning empty)
131pollTimeoutMs = 25000 -- long-poll heartbeat interval (how long a poll blocks before returning empty)
127maxPollsPerKey = 3 -- max concurrent long-poll connections per pubkey
132maxPollsPerKey = 3 -- max concurrent long-poll connections per pubkey
@@ -232,7 +237,7 @@ type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHe
232-- Presence: tracks that a pubkey is online at a given server, with expiry
237-- Presence: tracks that a pubkey is online at a given server, with expiry
233type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
238type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
234-- Queued federation forward: message waiting to be delivered to a remote server
239-- Queued federation forward: message waiting to be delivered to a remote server
235−type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text}
240+type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int}
236-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
241-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
237type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
242type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
238-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
243-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
@@ -259,6 +264,8 @@ type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFa
259type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
264type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
260type RecipientCount = {pubkey: PubkeyHex, count: Int}
265type RecipientCount = {pubkey: PubkeyHex, count: Int}
261type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
266type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timestamp}
267+-- Tracks recent federation request (server, timestamp) pairs for replay detection
268+type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
263-- ============================================
270-- ============================================
264-- Persisted State
271-- Persisted State
@@ -308,7 +315,26 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
309*totalMessageCount : Int
316*totalMessageCount : Int
318+-- Presence sharded into 16 buckets by first hex char of pubkey.
319+-- Hot-path operations (send, lookup, poll) scan only one bucket (~1/16 of entries).
320+*presenceBucket0 : [PresenceEntry]
321+*presenceBucket1 : [PresenceEntry]
322+*presenceBucket2 : [PresenceEntry]
323+*presenceBucket3 : [PresenceEntry]
324+*presenceBucket4 : [PresenceEntry]
325+*presenceBucket5 : [PresenceEntry]
326+*presenceBucket6 : [PresenceEntry]
327+*presenceBucket7 : [PresenceEntry]
328+*presenceBucket8 : [PresenceEntry]
329+*presenceBucket9 : [PresenceEntry]
330+*presenceBucket10 : [PresenceEntry]
331+*presenceBucket11 : [PresenceEntry]
332+*presenceBucket12 : [PresenceEntry]
333+*presenceBucket13 : [PresenceEntry]
334+*presenceBucket14 : [PresenceEntry]
311−*presence : [PresenceEntry]
335+*presenceBucket15 : [PresenceEntry]
337+*totalPresenceCount : Int
313-- Monotonic counter ensuring unique receivedAt ordering even within the same millisecond
339-- Monotonic counter ensuring unique receivedAt ordering even within the same millisecond
314*seqCounter : Timestamp
340*seqCounter : Timestamp
@@ -423,6 +449,9 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
423-- Signal for drain timeout (set by forked timer, watched by STM retry in waitForInFlightDrain)
449-- Signal for drain timeout (set by forked timer, watched by STM retry in waitForInFlightDrain)
424*drainTimedOut : Bool
450*drainTimedOut : Bool
452+-- Recent federation requests for replay detection (dedup by server+timestamp)
453+*recentFederationRequests : [RecentFederationRequest]
426-- ============================================
455-- ============================================
428-- ============================================
457-- ============================================
@@ -924,6 +953,129 @@ transformAllSessionBuckets = \f -> do
924 b14 <- *sessionBucket14; set *sessionBucket14 = f b14
953 b14 <- *sessionBucket14; set *sessionBucket14 = f b14
925 b15 <- *sessionBucket15; set *sessionBucket15 = f b15
954 b15 <- *sessionBucket15; set *sessionBucket15 = f b15
956+-- ============================================
957+-- Presence Bucket Helpers
958+-- ============================================
959+-- Presence is sharded into 16 buckets by the first hex char of the pubkey.
960+-- Hot-path operations (send, lookup, poll) scan only one bucket (~1/16 of entries).
962+readPresenceBucket : Int -> IO [PresenceEntry]
963+readPresenceBucket = \b ->
964+ if b == 0 then do s <- *presenceBucket0; yield s
965+ else if b == 1 then do s <- *presenceBucket1; yield s
966+ else if b == 2 then do s <- *presenceBucket2; yield s
967+ else if b == 3 then do s <- *presenceBucket3; yield s
968+ else if b == 4 then do s <- *presenceBucket4; yield s
969+ else if b == 5 then do s <- *presenceBucket5; yield s
970+ else if b == 6 then do s <- *presenceBucket6; yield s
971+ else if b == 7 then do s <- *presenceBucket7; yield s
972+ else if b == 8 then do s <- *presenceBucket8; yield s
973+ else if b == 9 then do s <- *presenceBucket9; yield s
974+ else if b == 10 then do s <- *presenceBucket10; yield s
975+ else if b == 11 then do s <- *presenceBucket11; yield s
976+ else if b == 12 then do s <- *presenceBucket12; yield s
977+ else if b == 13 then do s <- *presenceBucket13; yield s
978+ else if b == 14 then do s <- *presenceBucket14; yield s
979+ else do s <- *presenceBucket15; yield s
981+writePresenceBucket : Int -> [PresenceEntry] -> IO ()
982+writePresenceBucket = \b entries ->
983+ if b == 0 then set *presenceBucket0 = entries
984+ else if b == 1 then set *presenceBucket1 = entries
985+ else if b == 2 then set *presenceBucket2 = entries
986+ else if b == 3 then set *presenceBucket3 = entries
987+ else if b == 4 then set *presenceBucket4 = entries
988+ else if b == 5 then set *presenceBucket5 = entries
989+ else if b == 6 then set *presenceBucket6 = entries
990+ else if b == 7 then set *presenceBucket7 = entries
991+ else if b == 8 then set *presenceBucket8 = entries
992+ else if b == 9 then set *presenceBucket9 = entries
993+ else if b == 10 then set *presenceBucket10 = entries
994+ else if b == 11 then set *presenceBucket11 = entries
995+ else if b == 12 then set *presenceBucket12 = entries
996+ else if b == 13 then set *presenceBucket13 = entries
997+ else if b == 14 then set *presenceBucket14 = entries
998+ else set *presenceBucket15 = entries
1000+readAllPresence : IO [PresenceEntry]
1001+readAllPresence = do
1002+ p0 <- *presenceBucket0
1003+ p1 <- *presenceBucket1
1004+ p2 <- *presenceBucket2
1005+ p3 <- *presenceBucket3
1006+ p4 <- *presenceBucket4
1007+ p5 <- *presenceBucket5
1008+ p6 <- *presenceBucket6
1009+ p7 <- *presenceBucket7
1010+ p8 <- *presenceBucket8
1011+ p9 <- *presenceBucket9
1012+ p10 <- *presenceBucket10
1013+ p11 <- *presenceBucket11
1014+ p12 <- *presenceBucket12
1015+ p13 <- *presenceBucket13
1016+ p14 <- *presenceBucket14
1017+ p15 <- *presenceBucket15
1018+ yield (union p0 (union p1 (union p2 (union p3 (union p4 (union p5 (union p6 (union p7 (union p8 (union p9 (union p10 (union p11 (union p12 (union p13 (union p14 p15)))))))))))))))
1020+countAllPresence : IO Int
1021+countAllPresence = do
1022+ p0 <- *presenceBucket0
1023+ p1 <- *presenceBucket1
1024+ p2 <- *presenceBucket2
1025+ p3 <- *presenceBucket3
1026+ p4 <- *presenceBucket4
1027+ p5 <- *presenceBucket5
1028+ p6 <- *presenceBucket6
1029+ p7 <- *presenceBucket7
1030+ p8 <- *presenceBucket8
1031+ p9 <- *presenceBucket9
1032+ p10 <- *presenceBucket10
1033+ p11 <- *presenceBucket11
1034+ p12 <- *presenceBucket12
1035+ p13 <- *presenceBucket13
1036+ p14 <- *presenceBucket14
1037+ p15 <- *presenceBucket15
1038+ yield (count p0 + count p1 + count p2 + count p3 + count p4 + count p5 + count p6 + count p7 + count p8 + count p9 + count p10 + count p11 + count p12 + count p13 + count p14 + count p15)
1040+writeAllPresence : [PresenceEntry] -> IO ()
1041+writeAllPresence = \entries -> do
1042+ let bFor = \b -> do p <- entries; where pubkeyBucket p.pubkey == b; yield p
1043+ writePresenceBucket 0 (bFor 0)
1044+ writePresenceBucket 1 (bFor 1)
1045+ writePresenceBucket 2 (bFor 2)
1046+ writePresenceBucket 3 (bFor 3)
1047+ writePresenceBucket 4 (bFor 4)
1048+ writePresenceBucket 5 (bFor 5)
1049+ writePresenceBucket 6 (bFor 6)
1050+ writePresenceBucket 7 (bFor 7)
1051+ writePresenceBucket 8 (bFor 8)
1052+ writePresenceBucket 9 (bFor 9)
1053+ writePresenceBucket 10 (bFor 10)
1054+ writePresenceBucket 11 (bFor 11)
1055+ writePresenceBucket 12 (bFor 12)
1056+ writePresenceBucket 13 (bFor 13)
1057+ writePresenceBucket 14 (bFor 14)
1058+ writePresenceBucket 15 (bFor 15)
1060+transformAllPresenceBuckets : ([PresenceEntry] -> [PresenceEntry]) -> IO ()
1061+transformAllPresenceBuckets = \f -> do
1062+ b0 <- *presenceBucket0; set *presenceBucket0 = f b0
1063+ b1 <- *presenceBucket1; set *presenceBucket1 = f b1
1064+ b2 <- *presenceBucket2; set *presenceBucket2 = f b2
1065+ b3 <- *presenceBucket3; set *presenceBucket3 = f b3
1066+ b4 <- *presenceBucket4; set *presenceBucket4 = f b4
1067+ b5 <- *presenceBucket5; set *presenceBucket5 = f b5
1068+ b6 <- *presenceBucket6; set *presenceBucket6 = f b6
1069+ b7 <- *presenceBucket7; set *presenceBucket7 = f b7
1070+ b8 <- *presenceBucket8; set *presenceBucket8 = f b8
1071+ b9 <- *presenceBucket9; set *presenceBucket9 = f b9
1072+ b10 <- *presenceBucket10; set *presenceBucket10 = f b10
1073+ b11 <- *presenceBucket11; set *presenceBucket11 = f b11
1074+ b12 <- *presenceBucket12; set *presenceBucket12 = f b12
1075+ b13 <- *presenceBucket13; set *presenceBucket13 = f b13
1076+ b14 <- *presenceBucket14; set *presenceBucket14 = f b14
1077+ b15 <- *presenceBucket15; set *presenceBucket15 = f b15
927-- ============================================
1079-- ============================================
928-- Message Bucket Helpers
1080-- Message Bucket Helpers
929-- ============================================
1081-- ============================================
@@ -1059,7 +1211,7 @@ adjustRecipientCount : [RecipientCount] -> PubkeyHex -> Int -> [RecipientCount]
1059adjustRecipientCount = \counts pubkey delta ->
1211adjustRecipientCount = \counts pubkey delta ->
1060 if delta == 0 then counts
1212 if delta == 0 then counts
1061 else if any counts (\c -> c.pubkey == pubkey)
1213 else if any counts (\c -> c.pubkey == pubkey)
1062− then do c <- counts; yield (if c.pubkey == pubkey then {pubkey: c.pubkey, count: max 0 (c.count + delta)} else c)
1214+ then do c <- counts; where not (c.pubkey == pubkey && c.count + delta <= 0); yield (if c.pubkey == pubkey then {pubkey: c.pubkey, count: c.count + delta} else c)
1063 else if delta > 0 then union counts [{pubkey: pubkey, count: delta}]
1215 else if delta > 0 then union counts [{pubkey: pubkey, count: delta}]
1064 else counts
1216 else counts
@@ -1078,7 +1230,7 @@ recomputeRecipientCounts = countByRecipient
1078-- instead of calling adjustRecipientCount in a loop.
1230-- instead of calling adjustRecipientCount in a loop.
1079batchDecrementRecipientCounts : [RecipientCount] -> [RecipientCount] -> [RecipientCount]
1231batchDecrementRecipientCounts : [RecipientCount] -> [RecipientCount] -> [RecipientCount]
1080batchDecrementRecipientCounts = \counts removedCounts ->
1232batchDecrementRecipientCounts = \counts removedCounts ->
1081− do c <- counts; yield {pubkey: c.pubkey, count: max 0 (c.count - getRecipientCount c.pubkey removedCounts)}
1233+ do c <- counts; let adj = c.count - getRecipientCount c.pubkey removedCounts; where adj > 0; yield {pubkey: c.pubkey, count: adj}
1083-- Check message dedup only (without counting — count comes from recipientMsgCounts)
1235-- Check message dedup only (without counting — count comes from recipientMsgCounts)
1084checkMessageDup : [Message] -> Text -> PubkeyHex -> Bool
1236checkMessageDup : [Message] -> Text -> PubkeyHex -> Bool
@@ -1344,10 +1496,11 @@ sendToEachN = \servers action remaining ->
1344pruneExpired = \t -> do
1496pruneExpired = \t -> do
1345 -- Prune presence (must be atomic for before/after comparison)
1497 -- Prune presence (must be atomic for before/after comparison)
1346 presenceResult <- atomic do
1498 presenceResult <- atomic do
1347− presenceBefore <- *presence
1499+ presenceBefore <- readAllPresence
1348 let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)
1500 let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)
1349 let presenceAfter = do x <- presenceBefore; where x.expiresAt > t; yield x
1501 let presenceAfter = do x <- presenceBefore; where x.expiresAt > t; yield x
1502+ writeAllPresence presenceAfter
1350− set *presence = presenceAfter
1503+ set *totalPresenceCount = count presenceAfter
1351 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
1504 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
1352 -- Prune auth state
1505 -- Prune auth state
1353 atomic do
1506 atomic do
@@ -1430,6 +1583,9 @@ pruneExpired = \t -> do
1430 when (count staleInFlight > 0)
1583 when (count staleInFlight > 0)
1431 (println ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))
1584 (println ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))
1432 set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < federationFetchTimeoutMs * 2); yield r
1585 set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < federationFetchTimeoutMs * 2); yield r
1586+ -- Prune stale federation replay dedup entries
1587+ recentFedReqs <- *recentFederationRequests
1588+ set *recentFederationRequests = do r <- recentFedReqs; where t - r.seenAt < clockSkewTolerance * 2; yield r
1433 yield presenceResult
1589 yield presenceResult
@@ -1567,6 +1723,23 @@ withFederationAuth = \authorization t respond fromServer requestTimestamp precon
1567 else if not precondition
1723 else if not precondition
1568 then yield (respond {ok: False {}, error: "invalid_request"})
1724 then yield (respond {ok: False {}, error: "invalid_request"})
1726+ -- Replay detection: reject duplicate (server, timestamp) pairs within the skew window
1727+ replayDetected <- if requestTimestamp > 0
1729+ recent <- *recentFederationRequests
1730+ let isDup = any recent (\r -> r.server == fromServer && r.timestamp == requestTimestamp)
1731+ if isDup then yield (True {})
1733+ let cleaned = do r <- recent; where t - r.seenAt < clockSkewTolerance * 2; yield r
1734+ let updated = union cleaned [{server: fromServer, timestamp: requestTimestamp, seenAt: t}]
1735+ set *recentFederationRequests = if count updated > maxRecentFederationRequests then evictOneOldest updated (\r -> r.seenAt) else updated
1737+ else yield (False {})
1740+ println ("federation: replay detected from " ++ fromServer ++ " (timestamp: " ++ show requestTimestamp ++ ")")
1741+ yield (respond {ok: False {}, error: "replay_detected"})
1570 globalLimited <- checkGlobalRate t
1743 globalLimited <- checkGlobalRate t
1571 if globalLimited
1744 if globalLimited
@@ -1604,29 +1777,35 @@ removeForward = \mid srv -> atomic do
1604 set *forwards = forwardsExcluding forwards srv [mid]
1777 set *forwards = forwardsExcluding forwards srv [mid]
1606-- Reclaim origin messages after federation ack confirms delivery.
1779-- Reclaim origin messages after federation ack confirms delivery.
1607−-- Iterates removedForwards with one atomic block per message (short locks).
1780+-- Batched by message bucket: all entries for the same bucket are processed in one
1781+-- atomic block (reduces transaction overhead compared to one-per-entry).
1608−ackCleanupForwards = \entries t -> case headOf entries of
1782+ackCleanupForwards = \entries t ->
1783+ let bucketIds = uniqueBucketsFor entries (\e -> e.toKey) in
1784+ ackCleanupByBucket bucketIds entries t
1786+ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of
1609 Nothing {} -> yield {}
1787 Nothing {} -> yield {}
1610− Just {value: rf} -> do
1788+ Just {value: b} -> do
1611− let rest = do x <- entries; where x.messageId != rf.messageId || x.toKey != rf.toKey; yield x
1789+ let remainingBuckets = do x <- bucketIds; where x != b; yield x
1790+ let bucketEntries = do e <- entries; where pubkeyBucket e.toKey == b; yield e
1612 atomic do
1791 atomic do
1613− presence <- *presence
1792+ allPresence <- readAllPresence
1614 forwards <- *forwards
1793 forwards <- *forwards
1794+ msgs <- readMessageBucket b
1795+ tc <- *totalMessageCount
1796+ rc <- *recipientMsgCounts
1797+ let cleanup = fold (\acc rf ->
1615− let hasLocal = any presence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t)
1798+ let hasLocal = any allPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in
1616− let hasOtherForwards = any forwards (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey)
1799+ let hasOtherFwds = any forwards (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey) in
1617− when (not hasLocal && not hasOtherForwards) (do
1800+ if hasLocal || hasOtherFwds then acc
1618− let b = pubkeyBucket rf.toKey
1619− bucket <- readMessageBucket b
1620− let msg = headOf (do m <- bucket; where m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod; yield m)
1801+ else case headOf (do m <- acc.msgs; where m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod; yield m) of
1622− Nothing {} -> yield {}
1802+ Nothing {} -> acc
1623− Just {} -> do
1803+ Just {} -> {msgs: (do m <- acc.msgs; where not (m.messageId == rf.messageId && m.toKey == rf.toKey); yield m), removed: acc.removed + 1, rc: adjustRecipientCount acc.rc rf.toKey (0 - 1)}
1624− writeMessageBucket b (do m <- bucket; where not (m.messageId == rf.messageId && m.toKey == rf.toKey); yield m)
1804+ ) {msgs: msgs, removed: 0, rc: rc} bucketEntries
1625− tc <- *totalMessageCount
1805+ writeMessageBucket b cleanup.msgs
1626− set *totalMessageCount = max 0 (tc - 1)
1806+ set *totalMessageCount = max 0 (tc - cleanup.removed)
1627− rc <- *recipientMsgCounts
1628− set *recipientMsgCounts = adjustRecipientCount rc rf.toKey (0 - 1))
1807+ set *recipientMsgCounts = cleanup.rc
1629− ackCleanupForwards rest t
1808+ ackCleanupByBucket remainingBuckets entries t
1631pow2 : Int -> Int
1810pow2 : Int -> Int
1632pow2 = \n -> if n <= 0 then 1 else 2 * pow2 (n - 1)
1811pow2 = \n -> if n <= 0 then 1 else 2 * pow2 (n - 1)
@@ -1672,6 +1851,16 @@ clearFailure = \server -> do
1672 set *forwardFailures = do f <- failures; where f.server != server; yield f
1851 set *forwardFailures = do f <- failures; where f.server != server; yield f
1673 set *lastFederationSuccessAt = t
1852 set *lastFederationSuccessAt = t
1854+-- Atomically increment a forward entry's retry count; drop if maxForwardRetries exceeded
1855+incrementForwardRetries = \messageId toServer -> atomic do
1856+ forwards <- *forwards
1857+ let shouldDrop = any forwards (\f -> f.messageId == messageId && f.toServer == toServer && f.retries + 1 >= maxForwardRetries)
1860+ set *forwards = do f <- forwards; where not (f.messageId == messageId && f.toServer == toServer); yield f
1861+ println ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
1862+ else set *forwards = do f <- forwards; yield (if f.messageId == messageId && f.toServer == toServer then {messageId: f.messageId, toServer: f.toServer, createdAt: f.createdAt, toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, retries: f.retries + 1} else f)
1675retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
1864retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
1676retryForwardsForServer = \entries remaining ->
1865retryForwardsForServer = \entries remaining ->
1677 if remaining <= 0
1866 if remaining <= 0
@@ -1690,13 +1879,15 @@ retryForwardsForServer = \entries remaining ->
1690 Err e -> do
1879 Err e -> do
1691 println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
1880 println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
1692 recordFailure f.toServer
1881 recordFailure f.toServer
1882+ incrementForwardRetries f.messageId f.toServer
1693− -- Network error: stop retrying, server is unreachable
1883+ -- Network error: stop retrying this server
1694 Ok response -> do
1884 Ok response -> do
1695 if response.ok then do
1885 if response.ok then do
1696 removeForward f.messageId f.toServer
1886 removeForward f.messageId f.toServer
1697 clearFailure f.toServer
1887 clearFailure f.toServer
1699 println ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)
1889 println ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)
1890+ incrementForwardRetries f.messageId f.toServer
1700 retryForwardsForServer rest (remaining - 1)
1891 retryForwardsForServer rest (remaining - 1)
1702-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
1893-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
@@ -1719,8 +1910,8 @@ retryPendingForwards = \pending -> do
1719forwardQueuedMessages = \targets t -> do
1910forwardQueuedMessages = \targets t -> do
1720 newForwards <- atomic do
1911 newForwards <- atomic do
1721 forwards <- *forwards
1912 forwards <- *forwards
1722− presence <- *presence
1913+ allPresence <- readAllPresence
1723− let localOnline = localOnlineKeys presence t
1914+ let localOnline = localOnlineKeys allPresence t
1724 let remoteTargets = remoteOnlyTargets targets localOnline
1915 let remoteTargets = remoteOnlyTargets targets localOnline
1725 -- Read only the message buckets containing target pubkeys (not all 16)
1916 -- Read only the message buckets containing target pubkeys (not all 16)
1726 let neededBuckets = uniqueBucketsFor remoteTargets (\target -> target.pubkey)
1917 let neededBuckets = uniqueBucketsFor remoteTargets (\target -> target.pubkey)
@@ -1886,15 +2077,16 @@ pruneAndRetry = \t -> do
1886 refreshResult <- atomic do
2077 refreshResult <- atomic do
1887 lastRefresh <- *lastPresenceRefresh
2078 lastRefresh <- *lastPresenceRefresh
1888 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval
2079 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval
1889− presence <- *presence
2080+ allPresence <- readAllPresence
1890 allSessions <- readAllSessions
2081 allSessions <- readAllSessions
1891 -- Extend presence for keys with active sessions nearing expiry (sliding window)
2082 -- Extend presence for keys with active sessions nearing expiry (sliding window)
1892− let localNearExpiry = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []
2083+ let localNearExpiry = if shouldRefresh then (do p <- allPresence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []
1893 let toExtend = do p <- localNearExpiry; where any allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
2084 let toExtend = do p <- localNearExpiry; where any allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
1894 let notExtended = do p <- localNearExpiry; where none allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
2085 let notExtended = do p <- localNearExpiry; where none allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
1895− when (count toExtend > 0) (
2086+ when (count toExtend > 0) (do
1896− set *presence = do pp <- presence; yield (if any toExtend (\e -> e.pubkey == pp.pubkey && isLocalServer pp.server) then {pubkey: pp.pubkey, server: pp.server, expiresAt: t + sessionExpiry, timestamp: t} else pp))
2087+ let updatedPresence = do pp <- allPresence; yield (if any toExtend (\e -> e.pubkey == pp.pubkey && isLocalServer pp.server) then {pubkey: pp.pubkey, server: pp.server, expiresAt: t + sessionExpiry, timestamp: t} else pp)
2088+ writeAllPresence updatedPresence)
1897− let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []
2089+ let refreshPeers = if shouldRefresh then collectPeerServers allPresence (\p -> p.server) else []
1898 when shouldRefresh (set *lastPresenceRefresh = t)
2090 when shouldRefresh (set *lastPresenceRefresh = t)
1899 yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
2091 yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
1900 -- Reset federation latency metrics periodically for windowed averages (prevents
2092 -- Reset federation latency metrics periodically for windowed averages (prevents
@@ -2148,7 +2340,7 @@ forwardCandidates = \targets messages forwards t ->
2148 let origin = if m.originServer == "" then serverHostN else m.originServer
2340 let origin = if m.originServer == "" then serverHostN else m.originServer
2149 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
2341 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
2150 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,
2342 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,
2151− originServer: origin})) [] targets
2343+ originServer: origin, retries: 0})) [] targets
2154challengeMatch = \challenges pubkey challenge ip t ->
2346challengeMatch = \challenges pubkey challenge ip t ->
@@ -2315,10 +2507,22 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
2316 -- Read all session buckets for per-key enforcement (revokeOthers, capPerKey)
2508 -- Read all session buckets for per-key enforcement (revokeOthers, capPerKey)
2317 allSessions <- readAllSessions
2509 allSessions <- readAllSessions
2318− presence <- *presence
2510+ allPresence <- readAllPresence
2511+ tpc <- *totalPresenceCount
2319 let sessionsForKey = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
2512 let sessionsForKey = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
2320 let totalSessions = count allSessions
2513 let totalSessions = count allSessions
2321 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp}
2514 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp}
2515+ -- Update presence: replace local entry for this pubkey, evict oldest if over capacity
2516+ let pb = pubkeyBucket pubkey
2517+ presenceBkt <- readPresenceBucket pb
2518+ let bucketWithout = do p <- presenceBkt; where p.pubkey != pubkey || not (isLocalServer p.server); yield p
2519+ let hadLocalEntry = count presenceBkt > count bucketWithout
2520+ let bucketEvicted = if not hadLocalEntry && tpc >= maxPresenceEntries then evictOneOldest bucketWithout (\p -> p.expiresAt) else bucketWithout
2521+ let newPresenceEntry = {pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}
2522+ writePresenceBucket pb (union bucketEvicted [newPresenceEntry])
2523+ let presenceDelta = count (union bucketEvicted [newPresenceEntry]) - count presenceBkt
2524+ set *totalPresenceCount = tpc + presenceDelta
2525+ let peerServers = if federationEnabled then collectPeerServers allPresence (\p -> p.server) else []
2322 -- Fast path: no revocation, under per-key and global limits — only write one bucket
2526 -- Fast path: no revocation, under per-key and global limits — only write one bucket
2323 if not revokeOthers && sessionsForKey < maxSessionsPerKey && totalSessions < maxSessions
2527 if not revokeOthers && sessionsForKey < maxSessionsPerKey && totalSessions < maxSessions
@@ -2327,13 +2531,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
2327 writeSessionBucket nb (union currentBucket [newSession])
2531 writeSessionBucket nb (union currentBucket [newSession])
2328 sc <- *totalSessionCount
2532 sc <- *totalSessionCount
2329 set *totalSessionCount = sc + 1
2533 set *totalSessionCount = sc + 1
2330− let presenceWithout = presenceWithoutLocalKey presence pubkey
2331− let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
2332− let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
2333− set *presence = updatedPresence
2334− yield (Just {value: {peerServers: if federationEnabled
2534+ yield (Just {value: {peerServers: peerServers, revokedCount: 0}})
2335− then collectPeerServers updatedPresence (\p -> p.server)
2336− else [], revokedCount: 0}})
2338 -- Slow path: revocation or capacity pressure — redistribute across all buckets
2536 -- Slow path: revocation or capacity pressure — redistribute across all buckets
2339 let revokedCount = if revokeOthers then sessionsForKey else 0
2537 let revokedCount = if revokeOthers then sessionsForKey else 0
@@ -2358,13 +2556,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
2358 writeSessionBucket 13 (bFor 13)
2556 writeSessionBucket 13 (bFor 13)
2359 writeSessionBucket 14 (bFor 14)
2557 writeSessionBucket 14 (bFor 14)
2360 writeSessionBucket 15 (bFor 15)
2558 writeSessionBucket 15 (bFor 15)
2361− let presenceWithout = presenceWithoutLocalKey presence pubkey
2362− let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
2363− let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
2364− set *presence = updatedPresence
2365− yield (Just {value: {peerServers: if federationEnabled
2366− then collectPeerServers updatedPresence (\p -> p.server)
2367− else [], revokedCount: revokedCount}})
2559+ yield (Just {value: {peerServers: peerServers, revokedCount: revokedCount}})
2368 case authResult of
2560 case authResult of
2369 Nothing {} -> yield (respond {token: "", expiresAt: 0})
2561 Nothing {} -> yield (respond {token: "", expiresAt: 0})
2370 Just {value: result} -> do
2562 Just {value: result} -> do
@@ -2396,8 +2588,8 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
2396 presenceUpdates <- if result.authorized && count cappedWatchKeys > 0 then do
2588 presenceUpdates <- if result.authorized && count cappedWatchKeys > 0 then do
2397 t2 <- now
2589 t2 <- now
2398 atomic do
2590 atomic do
2399− presence <- *presence
2591+ allPresence <- readAllPresence
2400− yield (do key <- cappedWatchKeys; yield {pubkey: key, online: count (presenceForKey presence key t2) > 0})
2592+ yield (do key <- cappedWatchKeys; yield {pubkey: key, online: count (presenceForKey allPresence key t2) > 0})
2401 else yield []
2593 else yield []
2402 fork (when federationEnabled (do
2594 fork (when federationEnabled (do
2403 sendAcksAndClean result.pendingAcks
2595 sendAcksAndClean result.pendingAcks
@@ -2463,7 +2655,8 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2463 if recipientCount >= maxMessagesPerRecipient
2655 if recipientCount >= maxMessagesPerRecipient
2464 then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}}
2656 then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}}
2466− presence <- *presence
2658+ -- Hot path: read only recipient's presence bucket
2659+ recipientPresence <- readPresenceBucket (pubkeyBucket to)
2467 seqCounter <- *seqCounter
2660 seqCounter <- *seqCounter
2468 let seqTs = nextSeqTs t seqCounter
2661 let seqTs = nextSeqTs t seqCounter
2469 set *seqCounter = seqTs
2662 set *seqCounter = seqTs
@@ -2473,7 +2666,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2473 set *totalMessageCount = totalMsgCount + 1
2666 set *totalMessageCount = totalMsgCount + 1
2474 set *recipientMsgCounts = adjustRecipientCount recipientCounts to 1
2667 set *recipientMsgCounts = adjustRecipientCount recipientCounts to 1
2475 signalMessageForKey to seqTs
2668 signalMessageForKey to seqTs
2476− let toPresence = presenceForKey presence to t
2669+ let toPresence = presenceForKey recipientPresence to t
2477 let localOnline = hasLocalPresence toPresence
2670 let localOnline = hasLocalPresence toPresence
2478 let remoteServers = collectPeerServers toPresence (\p -> p.server)
2671 let remoteServers = collectPeerServers toPresence (\p -> p.server)
2479 let delivery = if localOnline then DsDelivered {}
2672 let delivery = if localOnline then DsDelivered {}
@@ -2482,7 +2675,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2482 -- Create forward entries atomically with message storage (prevents race with ack)
2675 -- Create forward entries atomically with message storage (prevents race with ack)
2483 when (federationEnabled && count remoteServers > 0) (do
2676 when (federationEnabled && count remoteServers > 0) (do
2484 forwards <- *forwards
2677 forwards <- *forwards
2485− let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN}
2678+ let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}
2486 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
2679 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
2487 set *forwards = union forwards cappedFwds)
2680 set *forwards = union forwards cappedFwds)
2488 yield {send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}, globalCapReached: False {}})
2681 yield {send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}, globalCapReached: False {}})
@@ -2543,8 +2736,9 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
2543 if rateResult.limited
2736 if rateResult.limited
2544 then yield {server: "", authorized: True {}}
2737 then yield {server: "", authorized: True {}}
2546− presence <- *presence
2739+ -- Hot path: read only the target key's presence bucket
2740+ lookupBkt <- readPresenceBucket (pubkeyBucket pubkey)
2547− let foundServer = bestServerForKey (presenceForKey presence pubkey t)
2741+ let foundServer = bestServerForKey (presenceForKey lookupBkt pubkey t)
2548 -- Mask remote server names: clients only need to know local/remote/unknown,
2742 -- Mask remote server names: clients only need to know local/remote/unknown,
2549 -- not which specific peer server a key is homed on (prevents metadata leakage)
2743 -- not which specific peer server a key is homed on (prevents metadata leakage)
2550 let maskedServer = if foundServer == "" then "" else if isLocalServer foundServer then foundServer else "remote"
2744 let maskedServer = if foundServer == "" then "" else if isLocalServer foundServer then foundServer else "remote"
@@ -2563,26 +2757,37 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
2563 withSessionAuth authorization t clientIp
2757 withSessionAuth authorization t clientIp
2564 (yield (respond {status: "", authorized: False {}}))
2758 (yield (respond {status: "", authorized: False {}}))
2565 (\senderKey -> do
2759 (\senderKey -> do
2760+ -- Per-key rate limiting (reuses lookup rate limiter — both are info lookups)
2761+ statusRateLimited <- atomic do
2762+ lookupRates <- *lookupRates
2763+ let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == senderKey)
2764+ (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})
2765+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
2766+ maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
2767+ set *lookupRates = rateResult.entries
2768+ yield rateResult.limited
2769+ if statusRateLimited then yield (respond {status: "", authorized: True {}})
2566− result <- atomic do
2771+ result <- atomic do
2567− let msgBucket = pubkeyBucket to
2772+ let msgBucket = pubkeyBucket to
2568− bucket <- readMessageBucket msgBucket
2773+ bucket <- readMessageBucket msgBucket
2569− -- Check if message is still in mailbox (pending delivery)
2774+ -- Check if message is still in mailbox (pending delivery)
2570− let inMailbox = any bucket (\m -> m.messageId == messageId && m.toKey == to)
2775+ let inMailbox = any bucket (\m -> m.messageId == messageId && m.toKey == to)
2571− if inMailbox then yield "pending"
2776+ if inMailbox then yield "pending"
2573− -- Check if being forwarded to a remote server
2574− forwards <- *forwards
2575− let inForwards = any forwards (\f -> f.messageId == messageId && f.toKey == to)
2576− if inForwards then yield "forwarding"
2778+ -- Check if being forwarded to a remote server
2779+ forwards <- *forwards
2780+ let inForwards = any forwards (\f -> f.messageId == messageId && f.toKey == to)
2781+ if inForwards then yield "forwarding"
2578− -- Check if already delivered (acked or in terminal delivered set)
2783+ -- Check if already delivered (acked or in terminal delivered set)
2579− ackedIds <- *ackedIds
2784+ ackedIds <- *ackedIds
2580− deliveredIds <- *deliveredIds
2785+ deliveredIds <- *deliveredIds
2581− let inAcked = hasMessage messageId to ackedIds
2786+ let inAcked = hasMessage messageId to ackedIds
2582− let inDelivered = hasMessage messageId to deliveredIds
2787+ let inDelivered = hasMessage messageId to deliveredIds
2583− if inAcked || inDelivered then yield "delivered"
2788+ if inAcked || inDelivered then yield "delivered"
2584− else yield "unknown"
2789+ else yield "unknown"
2585− yield (respond {status: result, authorized: True {}}))
2790+ yield (respond {status: result, authorized: True {}}))
2587-- POST /auth/revoke: Revoke a specific session token.
2792-- POST /auth/revoke: Revoke a specific session token.
2588-- Caller authenticates via Authorization header, then specifies the target token to remove.
2793-- Caller authenticates via Authorization header, then specifies the target token to remove.
@@ -2650,21 +2855,25 @@ handleLogout = \authorization clientIp respond -> do
2650 hasRemaining <- hasSessionForKey pubkey 0
2855 hasRemaining <- hasSessionForKey pubkey 0
2651 if not hasRemaining
2856 if not hasRemaining
2653− -- Remove local presence immediately
2858+ -- Remove local presence immediately (only this key's bucket)
2654− presence <- *presence
2859+ let pb = pubkeyBucket pubkey
2655− set *presence = presenceWithoutLocalKey presence pubkey
2860+ presenceBkt <- readPresenceBucket pb
2861+ writePresenceBucket pb (do p <- presenceBkt; where p.pubkey != pubkey || not (isLocalServer p.server); yield p)
2862+ let removedCount = count presenceBkt - count (do p <- presenceBkt; where p.pubkey != pubkey || not (isLocalServer p.server); yield p)
2863+ tpc <- *totalPresenceCount
2864+ set *totalPresenceCount = max 0 (tpc - removedCount)
2865+ -- Capture peer servers inside atomic block (fix: was read outside)
2866+ allPresence <- readAllPresence
2867+ let peerServers = collectPeerServers allPresence (\p -> p.server)
2656− yield (Just {value: {pubkey: pubkey, gossipOffline: True {}}})
2868+ yield (Just {value: {pubkey: pubkey, gossipOffline: True {}, peerServers: peerServers}})
2657− else yield (Just {value: {pubkey: pubkey, gossipOffline: False {}}})
2869+ else yield (Just {value: {pubkey: pubkey, gossipOffline: False {}, peerServers: []}})
2658 case result of
2870 case result of
2659 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
2871 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
2660 Just {value: r} -> do
2872 Just {value: r} -> do
2661 when r.gossipOffline (do
2873 when r.gossipOffline (do
2662 println ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")
2874 println ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")
2663− when federationEnabled (do
2875+ when (federationEnabled && count r.peerServers > 0)
2664− presence <- *presence
2665− let peerServers = collectPeerServers presence (\p -> p.server)
2666− when (count peerServers > 0)
2667− (fork (sendBatchedGossip peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))))
2876+ (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
2668 yield (respond {ok: True {}, authorized: True {}})
2877 yield (respond {ok: True {}, authorized: True {}})
2670-- POST /federation/gossip: Receive presence gossip from a peer server.
2879-- POST /federation/gossip: Receive presence gossip from a peer server.
@@ -2686,9 +2895,10 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
2686 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
2895 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
2687 let deduped = mergeByPubkeyServer [] filteredOnline
2896 let deduped = mergeByPubkeyServer [] filteredOnline
2688 gossipResult <- atomic do
2897 gossipResult <- atomic do
2689− presence <- *presence
2898+ allPresence <- readAllPresence
2690− let gm = mergeGossipPresence presence deduped trustedOffline fromServer t
2899+ let gm = mergeGossipPresence allPresence deduped trustedOffline fromServer t
2900+ writeAllPresence gm.finalPresence
2691− set *presence = gm.finalPresence
2901+ set *totalPresenceCount = count gm.finalPresence
2692 yield {gm: gm}
2902 yield {gm: gm}
2693 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
2903 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
2694 when (not gossipResult.gm.withinCapacity)
2904 when (not gossipResult.gm.withinCapacity)
@@ -2800,8 +3010,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp autho
2800 when (count removedForwards > 0) (fork (do
3010 when (count removedForwards > 0) (fork (do
2801 sleep ackGracePeriod
3011 sleep ackGracePeriod
2802 t2 <- now
3012 t2 <- now
2803− -- Each forward is cleaned up in its own atomic block (short locks, one per message).
3013+ -- Batched by message bucket: entries for the same bucket share one atomic block.
2804− -- Uses explicit recursion since fold is pure and can't sequence atomic IO.
2805 ackCleanupForwards removedForwards t2))
3014 ackCleanupForwards removedForwards t2))
2806 yield (respond {ok: True {}, error: ""}))
3015 yield (respond {ok: True {}, error: ""}))
@@ -2815,17 +3024,22 @@ handleDrain = \authorization respond -> do
2816 drainResult <- atomic do
3025 drainResult <- atomic do
2817 set *draining = True {}
3026 set *draining = True {}
2818− presence <- *presence
3027+ allPresence <- readAllPresence
2819− let localEntries = do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p
3028+ let localEntries = do p <- allPresence; where isLocalServer p.server && p.expiresAt > t; yield p
2820− set *presence = do p <- presence; where not (isLocalServer p.server) || p.expiresAt <= t; yield p
3029+ let remainingPresence = do p <- allPresence; where not (isLocalServer p.server) || p.expiresAt <= t; yield p
3030+ writeAllPresence remainingPresence
3031+ set *totalPresenceCount = count remainingPresence
2821 set *pollWakeup = t
3032 set *pollWakeup = t
2822− yield {localEntries: localEntries, peerServers: collectPeerServers presence (\p -> p.server)}
3033+ yield {localEntries: localEntries, peerServers: collectPeerServers allPresence (\p -> p.server)}
2823 let drainedCount = count drainResult.localEntries
3034 let drainedCount = count drainResult.localEntries
3035+ -- Read pending federation state atomically (fix: was read outside atomic)
3036+ flushState <- atomic do
2824− forwards <- *forwards
3037+ f <- *forwards
2825− ackedIds <- *ackedIds
3038+ a <- *ackedIds
2826− let pendingAcks = do a <- ackedIds; where not (isLocalServer a.originServer); yield a
3039+ let pa = do x <- a; where not (isLocalServer x.originServer); yield x
3040+ yield {forwards: f, pendingAcks: pa}
2827− let pendingForwardCount = count forwards
3041+ let pendingForwardCount = count flushState.forwards
2828− let pendingAckCount = count pendingAcks
3042+ let pendingAckCount = count flushState.pendingAcks
2829 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
3043 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
2830 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
3044 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
2831 atomic do set *drainCompletedAt = 0
3045 atomic do set *drainCompletedAt = 0
@@ -2833,10 +3047,10 @@ handleDrain = \authorization respond -> do
2833 -- Best-effort flush of pending forwards and acks before stopping
3047 -- Best-effort flush of pending forwards and acks before stopping
2834 when (federationEnabled && pendingForwardCount > 0) (do
3048 when (federationEnabled && pendingForwardCount > 0) (do
2835 println ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards")
3049 println ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards")
2836− retryPendingForwards forwards)
3050+ retryPendingForwards flushState.forwards)
2837 when (federationEnabled && pendingAckCount > 0) (do
3051 when (federationEnabled && pendingAckCount > 0) (do
2838 println ("drain: flushing " ++ show pendingAckCount ++ " pending acks")
3052 println ("drain: flushing " ++ show pendingAckCount ++ " pending acks")
2839− sendAcksAndClean pendingAcks)
3053+ sendAcksAndClean flushState.pendingAcks)
2840 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
3054 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
2841 let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
3055 let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
2842 sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
3056 sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
@@ -2882,7 +3096,7 @@ handleHealth = \respond -> do
2882 lastPrune <- *lastPrunedAt
3096 lastPrune <- *lastPrunedAt
2883 lastHb <- *lastHeartbeatAt
3097 lastHb <- *lastHeartbeatAt
2884 sa <- *startedAt
3098 sa <- *startedAt
2885− presence <- *presence
3099+ presenceCount <- *totalPresenceCount
2886 forwards <- *forwards
3100 forwards <- *forwards
2887 sessionCount <- *totalSessionCount
3101 sessionCount <- *totalSessionCount
2888 activePolls <- *activePolls
3102 activePolls <- *activePolls
@@ -2894,7 +3108,7 @@ handleHealth = \respond -> do
2894 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
3108 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
2895 -- Capacity pressure: flag when any capped collection exceeds 90% full
3109 -- Capacity pressure: flag when any capped collection exceeds 90% full
2896 let capacityOk = tmc * 10 <= maxMessages * 9
3110 let capacityOk = tmc * 10 <= maxMessages * 9
2897− && count presence * 10 <= maxPresenceEntries * 9
3111+ && presenceCount * 10 <= maxPresenceEntries * 9
2898 && count forwards * 10 <= maxForwardsEntries * 9
3112 && count forwards * 10 <= maxForwardsEntries * 9
2899 && sessionCount * 10 <= maxSessions * 9
3113 && sessionCount * 10 <= maxSessions * 9
2900 && count activePolls * 10 <= maxActivePolls * 9
3114 && count activePolls * 10 <= maxActivePolls * 9
@@ -2926,7 +3140,7 @@ handleStats = \authorization respond -> do
2926 -- (slight inconsistency acceptable for monitoring). This narrows the lock window.
3140 -- (slight inconsistency acceptable for monitoring). This narrows the lock window.
2927 core <- atomic do
3141 core <- atomic do
2928 sc <- *totalSessionCount
3142 sc <- *totalSessionCount
2929− p <- *presence
3143+ pc <- *totalPresenceCount
2930 f <- *forwards
3144 f <- *forwards
2931 c <- *challenges
3145 c <- *challenges
2932 a <- *ackedIds
3146 a <- *ackedIds
@@ -2938,7 +3152,7 @@ handleStats = \authorization respond -> do
2938 rc <- *recipientMsgCounts
3152 rc <- *recipientMsgCounts
2939 fi <- *federationInFlight
3153 fi <- *federationInFlight
2940 rg <- *recentGossip
3154 rg <- *recentGossip
2941− yield {sc: sc, p: p, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg}
3155+ yield {sc: sc, pc: pc, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg}
2942 counters <- atomic do
3156 counters <- atomic do
2943 sr <- *sendRates
3157 sr <- *sendRates
2944 cr <- *challengeRates
3158 cr <- *challengeRates
@@ -2959,7 +3173,7 @@ handleStats = \authorization respond -> do
2959 flm <- *federationLatencyMax
3173 flm <- *federationLatencyMax
2960 dca <- *drainCompletedAt
3174 dca <- *drainCompletedAt
2961 yield {sr: sr, cr: cr, fr: fr, grc: grc, ir: ir, lr: lr, rr: rr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, dca: dca}
3175 yield {sr: sr, cr: cr, fr: fr, grc: grc, ir: ir, lr: lr, rr: rr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, dca: dca}
2962− let result = {sessions: core.sc, messages: core.tmc, presence: count core.p, forwards: count core.f, challenges: count core.c, ackedIds: count core.a, ackOverflow: count core.ao, deliveredIds: count core.d, forwardFailures: count core.ff, activePolls: count core.ap, sendRates: count counters.sr, challengeRates: count counters.cr, federationRates: count counters.fr, globalRateCount: counters.grc, ipRates: count counters.ir, lookupRates: count counters.lr, recvRates: count counters.rr, pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr, recentGossip: count core.rg, totalMessageCount: core.tmc, recipientCounts: count core.rc, federationInFlight: count core.fi, federationLatencyTotal: counters.flt, federationLatencyCount: counters.flc, federationLatencyMax: counters.flm, drainCompletedAt: counters.dca}
3176+ let result = {sessions: core.sc, messages: core.tmc, presence: core.pc, forwards: count core.f, challenges: count core.c, ackedIds: count core.a, ackOverflow: count core.ao, deliveredIds: count core.d, forwardFailures: count core.ff, activePolls: count core.ap, sendRates: count counters.sr, challengeRates: count counters.cr, federationRates: count counters.fr, globalRateCount: counters.grc, ipRates: count counters.ir, lookupRates: count counters.lr, recvRates: count counters.rr, pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr, recentGossip: count core.rg, totalMessageCount: core.tmc, recipientCounts: count core.rc, federationInFlight: count core.fi, federationLatencyTotal: counters.flt, federationLatencyCount: counters.flc, federationLatencyMax: counters.flm, drainCompletedAt: counters.dca}
2963 let msgPct = if maxMessages > 0 then result.messages * 100 / maxMessages else 0
3177 let msgPct = if maxMessages > 0 then result.messages * 100 / maxMessages else 0
2964 let presPct = if maxPresenceEntries > 0 then result.presence * 100 / maxPresenceEntries else 0
3178 let presPct = if maxPresenceEntries > 0 then result.presence * 100 / maxPresenceEntries else 0
2965 let fwdPct = if maxForwardsEntries > 0 then result.forwards * 100 / maxForwardsEntries else 0
3179 let fwdPct = if maxForwardsEntries > 0 then result.forwards * 100 / maxForwardsEntries else 0
@@ -2991,12 +3205,13 @@ handleStatsKey = \pubkey authorization respond -> do
2992 result <- atomic do
3206 result <- atomic do
2993 keyBucket <- readMessageBucket (pubkeyBucket pubkey)
3207 keyBucket <- readMessageBucket (pubkeyBucket pubkey)
2994− presence <- *presence
3208+ -- Single bucket read: all entries for this pubkey are in the same presence bucket
3209+ keyPresenceBkt <- readPresenceBucket (pubkeyBucket pubkey)
2995 allSessions <- readAllSessions
3210 allSessions <- readAllSessions
2996 forwards <- *forwards
3211 forwards <- *forwards
2997 activePolls <- *activePolls
3212 activePolls <- *activePolls
2998 let keyMessages = count (do m <- keyBucket; where m.toKey == pubkey; yield m)
3213 let keyMessages = count (do m <- keyBucket; where m.toKey == pubkey; yield m)
2999− let keyPresence = do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield {server: p.server, expiresAt: p.expiresAt}
3214+ let keyPresence = do p <- keyPresenceBkt; where p.pubkey == pubkey && p.expiresAt > t; yield {server: p.server, expiresAt: p.expiresAt}
3000 let keySessions = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
3215 let keySessions = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
3001 let keyForwards = count (do f <- forwards; where f.toKey == pubkey; yield f)
3216 let keyForwards = count (do f <- forwards; where f.toKey == pubkey; yield f)
3002 let keyPolls = count (do p <- activePolls; where p.pubkey == pubkey; yield p)
3217 let keyPolls = count (do p <- activePolls; where p.pubkey == pubkey; yield p)
@@ -3016,7 +3231,7 @@ handleSnapshot = \authorization respond -> do
3016 if not authorized then yield (respond emptySnapshot)
3231 if not authorized then yield (respond emptySnapshot)
3018 result <- atomic do
3233 result <- atomic do
3019− p <- *presence
3234+ p <- readAllPresence
3020 a <- *ackedIds
3235 a <- *ackedIds
3021 ao <- *ackOverflow
3236 ao <- *ackOverflow
3022 d <- *deliveredIds
3237 d <- *deliveredIds
@@ -3040,10 +3255,11 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw
3041 result <- atomic do
3256 result <- atomic do
3042 -- Merge presence (existing entries win on conflict via freshness)
3257 -- Merge presence (existing entries win on conflict via freshness)
3043− presence <- *presence
3258+ allPresence <- readAllPresence
3044− let mergedPresence = mergeByPubkeyServer presence (do p <- presenceData; where p.expiresAt > t; yield p)
3259+ let mergedPresence = mergeByPubkeyServer allPresence (do p <- presenceData; where p.expiresAt > t; yield p)
3045 let cappedPresence = takeRelation maxPresenceEntries mergedPresence
3260 let cappedPresence = takeRelation maxPresenceEntries mergedPresence
3261+ writeAllPresence cappedPresence
3046− set *presence = cappedPresence
3262+ set *totalPresenceCount = count cappedPresence
3047 -- Union dedup state, respecting capacity limits
3263 -- Union dedup state, respecting capacity limits
3048 ackedIds <- *ackedIds
3264 ackedIds <- *ackedIds
3049 let mergedAcked = union ackedIds ackedIdsData
3265 let mergedAcked = union ackedIds ackedIdsData
@@ -3167,6 +3383,26 @@ pollHeartbeat = do
3167 set *heartbeatCount = hc + 1
3383 set *heartbeatCount = hc + 1
3168 pollHeartbeat
3384 pollHeartbeat
3386+-- Background loop: periodically logs a state summary to stdout.
3387+-- Operators can capture stdout to detect state drift; use /maintenance/snapshot for full export.
3388+backgroundSnapshot = do
3389+ sleep autoSnapshotInterval
3390+ draining <- *draining
3391+ if not draining then do
3392+ result <- atomic do
3393+ tmc <- *totalMessageCount
3394+ sc <- *totalSessionCount
3395+ pc <- *totalPresenceCount
3400+ ff <- *forwardFailures
3401+ yield {messages: tmc, sessions: sc, presence: pc, forwards: count f, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff}
3402+ println ("snapshot: messages=" ++ show result.messages ++ " sessions=" ++ show result.sessions ++ " presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " ackOverflow=" ++ show result.ackOverflow ++ " deliveredIds=" ++ show result.deliveredIds ++ " forwardFailures=" ++ show result.forwardFailures)
3170main = do
3406main = do
3171 let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
3407 let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
3172 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
3408 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
@@ -3184,11 +3420,15 @@ main = do
3184 && maxSessionsPerKey <= maxSessions
3420 && maxSessionsPerKey <= maxSessions
3185 && maxChallengesPerKey <= maxChallenges
3421 && maxChallengesPerKey <= maxChallenges
3186 && maxPollsPerKey <= maxActivePolls
3422 && maxPollsPerKey <= maxActivePolls
3423+ && pollTimeoutMs < sessionExpiry
3424+ && gossipDedupWindow < presenceRefreshInterval
3425+ && federationFetchTimeoutMs <= backgroundPruneInterval
3426+ && maxAckOverflow <= maxAckedIds
3187 let peerSecretsOk = none federationPeerSecrets (\p -> isBadServerName (normalizeServerName p.server) || p.secret == "")
3427 let peerSecretsOk = none federationPeerSecrets (\p -> isBadServerName (normalizeServerName p.server) || p.secret == "")
3188 if not configOk
3428 if not configOk
3189 then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
3429 then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
3190 else if not configRelationsOk
3430 else if not configRelationsOk
3191− then println "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, per-key limits <= global limits), server not started"
3431+ 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, per-key limits <= global limits), server not started"
3192 else if not peerSecretsOk
3432 else if not peerSecretsOk
3193 then println "FATAL: invalid federationPeerSecrets — server names must pass validation and secrets must be non-empty"
3433 then println "FATAL: invalid federationPeerSecrets — server names must pass validation and secrets must be non-empty"
@@ -3212,6 +3452,9 @@ main = do
3212 (println (" per-peer secrets: " ++ show (count federationPeerSecrets) ++ " configured"))
3452 (println (" per-peer secrets: " ++ show (count federationPeerSecrets) ++ " configured"))
3213 when (federationCircuitBreakerThreshold > 0)
3453 when (federationCircuitBreakerThreshold > 0)
3214 (println (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
3454 (println (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
3455+ when (autoSnapshotInterval > 0)
3456+ (println (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
3215 fork backgroundPrune
3457 fork backgroundPrune
3216 fork pollHeartbeat
3458 fork pollHeartbeat
3459+ when (autoSnapshotInterval > 0) (fork backgroundSnapshot)
3217 listen 8080 serve
3460 listen 8080 serve