@@ -93,6 +93,7 @@ offlineGossipMinTtl : Int<S>
9393offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events
9494maxFederationConcurrency = 10 -- max concurrent outbound federation requests per cycle
9595maxForwardRetriesPerServer = 100
96+maxForwardRetries = 10 -- max retries per individual forward entry before dropping
9697maxGossipHops = 5 -- max hops a gossip event can propagate across servers
9798gossipOriginHops = 2 -- hops set when this server originates a gossip event
9899gossipDedupWindow : Int<Ms>
@@ -114,6 +115,8 @@ maxFederationInFlight = 100
114115-- servers have active (non-expired) failure records. Prevents wasted retries and log spam
115116-- when the local network is down. Set to 0 to disable.
116117federationCircuitBreakerThreshold = 3
118+-- Federation replay dedup: track recent (server, timestamp) pairs to reject exact replays
119+maxRecentFederationRequests = 1000
117120
118121-- Federation latency metric window
119122federationLatencyWindowMs : Int<Ms>
@@ -122,6 +125,8 @@ federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after thi
122125-- Background task intervals
123126backgroundPruneInterval : Int<Ms>
124127backgroundPruneInterval = 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)
125130pollTimeoutMs : Int<Ms>
126131pollTimeoutMs = 25000 -- long-poll heartbeat interval (how long a poll blocks before returning empty)
127132maxPollsPerKey = 3 -- max concurrent long-poll connections per pubkey
@@ -232,7 +237,7 @@ type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHe
232237-- Presence: tracks that a pubkey is online at a given server, with expiry
233238type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
234239-- 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}
236241-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
237242type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
238243-- 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
259264type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
260265type RecipientCount = {pubkey: PubkeyHex, count: Int}
261266type 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}
262269
263270-- ============================================
264271-- Persisted State
@@ -308,7 +315,26 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
308315
309316*totalMessageCount : Int
310317
311−*presence : [PresenceEntry]
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]
335+*presenceBucket15 : [PresenceEntry]
336+
337+*totalPresenceCount : Int
312338
313339-- Monotonic counter ensuring unique receivedAt ordering even within the same millisecond
314340*seqCounter : Timestamp
@@ -423,6 +449,9 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
423449-- Signal for drain timeout (set by forked timer, watched by STM retry in waitForInFlightDrain)
424450*drainTimedOut : Bool
425451
452+-- Recent federation requests for replay detection (dedup by server+timestamp)
453+*recentFederationRequests : [RecentFederationRequest]
454+
426455-- ============================================
427456-- Routes
428457-- ============================================
@@ -924,6 +953,129 @@ transformAllSessionBuckets = \f -> do
924953 b14 <- *sessionBucket14; set *sessionBucket14 = f b14
925954 b15 <- *sessionBucket15; set *sessionBucket15 = f b15
926955
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).
961+
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
980+
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
999+
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)))))))))))))))
1019+
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)
1039+
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)
1059+
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
1078+
9271079-- ============================================
9281080-- Message Bucket Helpers
9291081-- ============================================
@@ -1059,7 +1211,7 @@ adjustRecipientCount : [RecipientCount] -> PubkeyHex -> Int -> [RecipientCount]
10591211adjustRecipientCount = \counts pubkey delta ->
10601212 if delta == 0 then counts
10611213 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)
10631215 else if delta > 0 then union counts [{pubkey: pubkey, count: delta}]
10641216 else counts
10651217
@@ -1078,7 +1230,7 @@ recomputeRecipientCounts = countByRecipient
10781230-- instead of calling adjustRecipientCount in a loop.
10791231batchDecrementRecipientCounts : [RecipientCount] -> [RecipientCount] -> [RecipientCount]
10801232batchDecrementRecipientCounts = \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}
10821234
10831235-- Check message dedup only (without counting — count comes from recipientMsgCounts)
10841236checkMessageDup : [Message] -> Text -> PubkeyHex -> Bool
@@ -1344,10 +1496,11 @@ sendToEachN = \servers action remaining ->
13441496pruneExpired = \t -> do
13451497 -- Prune presence (must be atomic for before/after comparison)
13461498 presenceResult <- atomic do
1347− presenceBefore <- *presence
1499+ presenceBefore <- readAllPresence
13481500 let peerServersBefore = collectPeerServers presenceBefore (\p -> p.server)
13491501 let presenceAfter = do x <- presenceBefore; where x.expiresAt > t; yield x
1350− set *presence = presenceAfter
1502+ writeAllPresence presenceAfter
1503+ set *totalPresenceCount = count presenceAfter
13511504 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
13521505 -- Prune auth state
13531506 atomic do
@@ -1430,6 +1583,9 @@ pruneExpired = \t -> do
14301583 when (count staleInFlight > 0)
14311584 (println ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))
14321585 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
14331589 yield presenceResult
14341590
14351591
@@ -1567,6 +1723,23 @@ withFederationAuth = \authorization t respond fromServer requestTimestamp precon
15671723 else if not precondition
15681724 then yield (respond {ok: False {}, error: "invalid_request"})
15691725 else do
1726+ -- Replay detection: reject duplicate (server, timestamp) pairs within the skew window
1727+ replayDetected <- if requestTimestamp > 0
1728+ then atomic do
1729+ recent <- *recentFederationRequests
1730+ let isDup = any recent (\r -> r.server == fromServer && r.timestamp == requestTimestamp)
1731+ if isDup then yield (True {})
1732+ else do
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
1736+ yield (False {})
1737+ else yield (False {})
1738+ if replayDetected
1739+ then do
1740+ println ("federation: replay detected from " ++ fromServer ++ " (timestamp: " ++ show requestTimestamp ++ ")")
1741+ yield (respond {ok: False {}, error: "replay_detected"})
1742+ else do
15701743 globalLimited <- checkGlobalRate t
15711744 if globalLimited
15721745 then do
@@ -1604,29 +1777,35 @@ removeForward = \mid srv -> atomic do
16041777 set *forwards = forwardsExcluding forwards srv [mid]
16051778
16061779-- Reclaim origin messages after federation ack confirms delivery.
1607−-- Iterates removedForwards with one atomic block per message (short locks).
1608−ackCleanupForwards = \entries t -> case headOf entries of
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).
1782+ackCleanupForwards = \entries t ->
1783+ let bucketIds = uniqueBucketsFor entries (\e -> e.toKey) in
1784+ ackCleanupByBucket bucketIds entries t
1785+
1786+ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of
16091787 Nothing {} -> yield {}
1610− Just {value: rf} -> do
1611− let rest = do x <- entries; where x.messageId != rf.messageId || x.toKey != rf.toKey; yield x
1788+ Just {value: b} -> do
1789+ let remainingBuckets = do x <- bucketIds; where x != b; yield x
1790+ let bucketEntries = do e <- entries; where pubkeyBucket e.toKey == b; yield e
16121791 atomic do
1613− presence <- *presence
1792+ allPresence <- readAllPresence
16141793 forwards <- *forwards
1615− let hasLocal = any presence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t)
1616− let hasOtherForwards = any forwards (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey)
1617− when (not hasLocal && not hasOtherForwards) (do
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)
1621− case msg of
1622− Nothing {} -> yield {}
1623− Just {} -> do
1624− writeMessageBucket b (do m <- bucket; where not (m.messageId == rf.messageId && m.toKey == rf.toKey); yield m)
1625− tc <- *totalMessageCount
1626− set *totalMessageCount = max 0 (tc - 1)
1627− rc <- *recipientMsgCounts
1628− set *recipientMsgCounts = adjustRecipientCount rc rf.toKey (0 - 1))
1629− ackCleanupForwards rest t
1794+ msgs <- readMessageBucket b
1795+ tc <- *totalMessageCount
1796+ rc <- *recipientMsgCounts
1797+ let cleanup = fold (\acc rf ->
1798+ let hasLocal = any allPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in
1799+ let hasOtherFwds = any forwards (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey) in
1800+ if hasLocal || hasOtherFwds then acc
1801+ else case headOf (do m <- acc.msgs; where m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod; yield m) of
1802+ Nothing {} -> acc
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)}
1804+ ) {msgs: msgs, removed: 0, rc: rc} bucketEntries
1805+ writeMessageBucket b cleanup.msgs
1806+ set *totalMessageCount = max 0 (tc - cleanup.removed)
1807+ set *recipientMsgCounts = cleanup.rc
1808+ ackCleanupByBucket remainingBuckets entries t
16301809
16311810pow2 : Int -> Int
16321811pow2 = \n -> if n <= 0 then 1 else 2 * pow2 (n - 1)
@@ -1672,6 +1851,16 @@ clearFailure = \server -> do
16721851 set *forwardFailures = do f <- failures; where f.server != server; yield f
16731852 set *lastFederationSuccessAt = t
16741853
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)
1858+ if shouldDrop
1859+ then do
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)
1863+
16751864retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
16761865retryForwardsForServer = \entries remaining ->
16771866 if remaining <= 0
@@ -1690,13 +1879,15 @@ retryForwardsForServer = \entries remaining ->
16901879 Err e -> do
16911880 println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
16921881 recordFailure f.toServer
1693− -- Network error: stop retrying, server is unreachable
1882+ incrementForwardRetries f.messageId f.toServer
1883+ -- Network error: stop retrying this server
16941884 Ok response -> do
16951885 if response.ok then do
16961886 removeForward f.messageId f.toServer
16971887 clearFailure f.toServer
1698− else
1888+ else do
16991889 println ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)
1890+ incrementForwardRetries f.messageId f.toServer
17001891 retryForwardsForServer rest (remaining - 1)
17011892
17021893-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
@@ -1719,8 +1910,8 @@ retryPendingForwards = \pending -> do
17191910forwardQueuedMessages = \targets t -> do
17201911 newForwards <- atomic do
17211912 forwards <- *forwards
1722− presence <- *presence
1723− let localOnline = localOnlineKeys presence t
1913+ allPresence <- readAllPresence
1914+ let localOnline = localOnlineKeys allPresence t
17241915 let remoteTargets = remoteOnlyTargets targets localOnline
17251916 -- Read only the message buckets containing target pubkeys (not all 16)
17261917 let neededBuckets = uniqueBucketsFor remoteTargets (\target -> target.pubkey)
@@ -1886,15 +2077,16 @@ pruneAndRetry = \t -> do
18862077 refreshResult <- atomic do
18872078 lastRefresh <- *lastPresenceRefresh
18882079 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval
1889− presence <- *presence
2080+ allPresence <- readAllPresence
18902081 allSessions <- readAllSessions
18912082 -- 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 []
18932084 let toExtend = do p <- localNearExpiry; where any allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
18942085 let notExtended = do p <- localNearExpiry; where none allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
1895− when (count toExtend > 0) (
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))
1897− let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []
2086+ when (count toExtend > 0) (do
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)
2089+ let refreshPeers = if shouldRefresh then collectPeerServers allPresence (\p -> p.server) else []
18982090 when shouldRefresh (set *lastPresenceRefresh = t)
18992091 yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
19002092 -- Reset federation latency metrics periodically for windowed averages (prevents
@@ -2148,7 +2340,7 @@ forwardCandidates = \targets messages forwards t ->
21482340 let origin = if m.originServer == "" then serverHostN else m.originServer
21492341 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
21502342 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,
2151− originServer: origin})) [] targets
2343+ originServer: origin, retries: 0})) [] targets
21522344
21532345
21542346challengeMatch = \challenges pubkey challenge ip t ->
@@ -2315,10 +2507,22 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
23152507 else do
23162508 -- Read all session buckets for per-key enforcement (revokeOthers, capPerKey)
23172509 allSessions <- readAllSessions
2318− presence <- *presence
2510+ allPresence <- readAllPresence
2511+ tpc <- *totalPresenceCount
23192512 let sessionsForKey = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
23202513 let totalSessions = count allSessions
23212514 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 []
23222526 -- Fast path: no revocation, under per-key and global limits — only write one bucket
23232527 if not revokeOthers && sessionsForKey < maxSessionsPerKey && totalSessions < maxSessions
23242528 then do
@@ -2327,13 +2531,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
23272531 writeSessionBucket nb (union currentBucket [newSession])
23282532 sc <- *totalSessionCount
23292533 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
2335− then collectPeerServers updatedPresence (\p -> p.server)
2336− else [], revokedCount: 0}})
2534+ yield (Just {value: {peerServers: peerServers, revokedCount: 0}})
23372535 else do
23382536 -- Slow path: revocation or capacity pressure — redistribute across all buckets
23392537 let revokedCount = if revokeOthers then sessionsForKey else 0
@@ -2358,13 +2556,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
23582556 writeSessionBucket 13 (bFor 13)
23592557 writeSessionBucket 14 (bFor 14)
23602558 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}})
23682560 case authResult of
23692561 Nothing {} -> yield (respond {token: "", expiresAt: 0})
23702562 Just {value: result} -> do
@@ -2396,8 +2588,8 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
23962588 presenceUpdates <- if result.authorized && count cappedWatchKeys > 0 then do
23972589 t2 <- now
23982590 atomic do
2399− presence <- *presence
2400− yield (do key <- cappedWatchKeys; yield {pubkey: key, online: count (presenceForKey presence key t2) > 0})
2591+ allPresence <- readAllPresence
2592+ yield (do key <- cappedWatchKeys; yield {pubkey: key, online: count (presenceForKey allPresence key t2) > 0})
24012593 else yield []
24022594 fork (when federationEnabled (do
24032595 sendAcksAndClean result.pendingAcks
@@ -2463,7 +2655,8 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
24632655 if recipientCount >= maxMessagesPerRecipient
24642656 then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}}
24652657 else do
2466− presence <- *presence
2658+ -- Hot path: read only recipient's presence bucket
2659+ recipientPresence <- readPresenceBucket (pubkeyBucket to)
24672660 seqCounter <- *seqCounter
24682661 let seqTs = nextSeqTs t seqCounter
24692662 set *seqCounter = seqTs
@@ -2473,7 +2666,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
24732666 set *totalMessageCount = totalMsgCount + 1
24742667 set *recipientMsgCounts = adjustRecipientCount recipientCounts to 1
24752668 signalMessageForKey to seqTs
2476− let toPresence = presenceForKey presence to t
2669+ let toPresence = presenceForKey recipientPresence to t
24772670 let localOnline = hasLocalPresence toPresence
24782671 let remoteServers = collectPeerServers toPresence (\p -> p.server)
24792672 let delivery = if localOnline then DsDelivered {}
@@ -2482,7 +2675,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
24822675 -- Create forward entries atomically with message storage (prevents race with ack)
24832676 when (federationEnabled && count remoteServers > 0) (do
24842677 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}
24862679 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
24872680 set *forwards = union forwards cappedFwds)
24882681 yield {send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}, globalCapReached: False {}})
@@ -2543,8 +2736,9 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
25432736 if rateResult.limited
25442737 then yield {server: "", authorized: True {}}
25452738 else do
2546− presence <- *presence
2547− let foundServer = bestServerForKey (presenceForKey presence pubkey t)
2739+ -- Hot path: read only the target key's presence bucket
2740+ lookupBkt <- readPresenceBucket (pubkeyBucket pubkey)
2741+ let foundServer = bestServerForKey (presenceForKey lookupBkt pubkey t)
25482742 -- Mask remote server names: clients only need to know local/remote/unknown,
25492743 -- not which specific peer server a key is homed on (prevents metadata leakage)
25502744 let maskedServer = if foundServer == "" then "" else if isLocalServer foundServer then foundServer else "remote"
@@ -2563,26 +2757,37 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
25632757 withSessionAuth authorization t clientIp
25642758 (yield (respond {status: "", authorized: False {}}))
25652759 (\senderKey -> do
2566− result <- atomic do
2567− let msgBucket = pubkeyBucket to
2568− bucket <- readMessageBucket msgBucket
2569− -- Check if message is still in mailbox (pending delivery)
2570− let inMailbox = any bucket (\m -> m.messageId == messageId && m.toKey == to)
2571− if inMailbox then yield "pending"
2572− else do
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"
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 {}})
2770+ else do
2771+ result <- atomic do
2772+ let msgBucket = pubkeyBucket to
2773+ bucket <- readMessageBucket msgBucket
2774+ -- Check if message is still in mailbox (pending delivery)
2775+ let inMailbox = any bucket (\m -> m.messageId == messageId && m.toKey == to)
2776+ if inMailbox then yield "pending"
25772777 else do
2578− -- Check if already delivered (acked or in terminal delivered set)
2579− ackedIds <- *ackedIds
2580− deliveredIds <- *deliveredIds
2581− let inAcked = hasMessage messageId to ackedIds
2582− let inDelivered = hasMessage messageId to deliveredIds
2583− if inAcked || inDelivered then yield "delivered"
2584− else yield "unknown"
2585− yield (respond {status: result, authorized: True {}}))
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"
2782+ else do
2783+ -- Check if already delivered (acked or in terminal delivered set)
2784+ ackedIds <- *ackedIds
2785+ deliveredIds <- *deliveredIds
2786+ let inAcked = hasMessage messageId to ackedIds
2787+ let inDelivered = hasMessage messageId to deliveredIds
2788+ if inAcked || inDelivered then yield "delivered"
2789+ else yield "unknown"
2790+ yield (respond {status: result, authorized: True {}}))
25862791
25872792-- POST /auth/revoke: Revoke a specific session token.
25882793-- Caller authenticates via Authorization header, then specifies the target token to remove.
@@ -2650,21 +2855,25 @@ handleLogout = \authorization clientIp respond -> do
26502855 hasRemaining <- hasSessionForKey pubkey 0
26512856 if not hasRemaining
26522857 then do
2653− -- Remove local presence immediately
2654− presence <- *presence
2655− set *presence = presenceWithoutLocalKey presence pubkey
2656− yield (Just {value: {pubkey: pubkey, gossipOffline: True {}}})
2657− else yield (Just {value: {pubkey: pubkey, gossipOffline: False {}}})
2858+ -- Remove local presence immediately (only this key's bucket)
2859+ let pb = pubkeyBucket 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)
2868+ yield (Just {value: {pubkey: pubkey, gossipOffline: True {}, peerServers: peerServers}})
2869+ else yield (Just {value: {pubkey: pubkey, gossipOffline: False {}, peerServers: []}})
26582870 case result of
26592871 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
26602872 Just {value: r} -> do
26612873 when r.gossipOffline (do
26622874 println ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")
2663− when federationEnabled (do
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))))
2875+ when (federationEnabled && count r.peerServers > 0)
2876+ (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
26682877 yield (respond {ok: True {}, authorized: True {}})
26692878
26702879-- POST /federation/gossip: Receive presence gossip from a peer server.
@@ -2686,9 +2895,10 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
26862895 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
26872896 let deduped = mergeByPubkeyServer [] filteredOnline
26882897 gossipResult <- atomic do
2689− presence <- *presence
2690− let gm = mergeGossipPresence presence deduped trustedOffline fromServer t
2691− set *presence = gm.finalPresence
2898+ allPresence <- readAllPresence
2899+ let gm = mergeGossipPresence allPresence deduped trustedOffline fromServer t
2900+ writeAllPresence gm.finalPresence
2901+ set *totalPresenceCount = count gm.finalPresence
26922902 yield {gm: gm}
26932903 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
26942904 when (not gossipResult.gm.withinCapacity)
@@ -2800,8 +3010,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp autho
28003010 when (count removedForwards > 0) (fork (do
28013011 sleep ackGracePeriod
28023012 t2 <- now
2803− -- Each forward is cleaned up in its own atomic block (short locks, one per message).
2804− -- Uses explicit recursion since fold is pure and can't sequence atomic IO.
3013+ -- Batched by message bucket: entries for the same bucket share one atomic block.
28053014 ackCleanupForwards removedForwards t2))
28063015 yield (respond {ok: True {}, error: ""}))
28073016
@@ -2815,17 +3024,22 @@ handleDrain = \authorization respond -> do
28153024 else do
28163025 drainResult <- atomic do
28173026 set *draining = True {}
2818− presence <- *presence
2819− let localEntries = do p <- presence; 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
3027+ allPresence <- readAllPresence
3028+ let localEntries = do p <- allPresence; where 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
28213032 set *pollWakeup = t
2822− yield {localEntries: localEntries, peerServers: collectPeerServers presence (\p -> p.server)}
3033+ yield {localEntries: localEntries, peerServers: collectPeerServers allPresence (\p -> p.server)}
28233034 let drainedCount = count drainResult.localEntries
2824− forwards <- *forwards
2825− ackedIds <- *ackedIds
2826− let pendingAcks = do a <- ackedIds; where not (isLocalServer a.originServer); yield a
2827− let pendingForwardCount = count forwards
2828− let pendingAckCount = count pendingAcks
3035+ -- Read pending federation state atomically (fix: was read outside atomic)
3036+ flushState <- atomic do
3037+ f <- *forwards
3038+ a <- *ackedIds
3039+ let pa = do x <- a; where not (isLocalServer x.originServer); yield x
3040+ yield {forwards: f, pendingAcks: pa}
3041+ let pendingForwardCount = count flushState.forwards
3042+ let pendingAckCount = count flushState.pendingAcks
28293043 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
28303044 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
28313045 atomic do set *drainCompletedAt = 0
@@ -2833,10 +3047,10 @@ handleDrain = \authorization respond -> do
28333047 -- Best-effort flush of pending forwards and acks before stopping
28343048 when (federationEnabled && pendingForwardCount > 0) (do
28353049 println ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards")
2836− retryPendingForwards forwards)
3050+ retryPendingForwards flushState.forwards)
28373051 when (federationEnabled && pendingAckCount > 0) (do
28383052 println ("drain: flushing " ++ show pendingAckCount ++ " pending acks")
2839− sendAcksAndClean pendingAcks)
3053+ sendAcksAndClean flushState.pendingAcks)
28403054 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
28413055 let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
28423056 sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
@@ -2882,7 +3096,7 @@ handleHealth = \respond -> do
28823096 lastPrune <- *lastPrunedAt
28833097 lastHb <- *lastHeartbeatAt
28843098 sa <- *startedAt
2885− presence <- *presence
3099+ presenceCount <- *totalPresenceCount
28863100 forwards <- *forwards
28873101 sessionCount <- *totalSessionCount
28883102 activePolls <- *activePolls
@@ -2894,7 +3108,7 @@ handleHealth = \respond -> do
28943108 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
28953109 -- Capacity pressure: flag when any capped collection exceeds 90% full
28963110 let capacityOk = tmc * 10 <= maxMessages * 9
2897− && count presence * 10 <= maxPresenceEntries * 9
3111+ && presenceCount * 10 <= maxPresenceEntries * 9
28983112 && count forwards * 10 <= maxForwardsEntries * 9
28993113 && sessionCount * 10 <= maxSessions * 9
29003114 && count activePolls * 10 <= maxActivePolls * 9
@@ -2926,7 +3140,7 @@ handleStats = \authorization respond -> do
29263140 -- (slight inconsistency acceptable for monitoring). This narrows the lock window.
29273141 core <- atomic do
29283142 sc <- *totalSessionCount
2929− p <- *presence
3143+ pc <- *totalPresenceCount
29303144 f <- *forwards
29313145 c <- *challenges
29323146 a <- *ackedIds
@@ -2938,7 +3152,7 @@ handleStats = \authorization respond -> do
29383152 rc <- *recipientMsgCounts
29393153 fi <- *federationInFlight
29403154 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}
29423156 counters <- atomic do
29433157 sr <- *sendRates
29443158 cr <- *challengeRates
@@ -2959,7 +3173,7 @@ handleStats = \authorization respond -> do
29593173 flm <- *federationLatencyMax
29603174 dca <- *drainCompletedAt
29613175 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}
29633177 let msgPct = if maxMessages > 0 then result.messages * 100 / maxMessages else 0
29643178 let presPct = if maxPresenceEntries > 0 then result.presence * 100 / maxPresenceEntries else 0
29653179 let fwdPct = if maxForwardsEntries > 0 then result.forwards * 100 / maxForwardsEntries else 0
@@ -2991,12 +3205,13 @@ handleStatsKey = \pubkey authorization respond -> do
29913205 then do
29923206 result <- atomic do
29933207 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)
29953210 allSessions <- readAllSessions
29963211 forwards <- *forwards
29973212 activePolls <- *activePolls
29983213 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}
30003215 let keySessions = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
30013216 let keyForwards = count (do f <- forwards; where f.toKey == pubkey; yield f)
30023217 let keyPolls = count (do p <- activePolls; where p.pubkey == pubkey; yield p)
@@ -3016,7 +3231,7 @@ handleSnapshot = \authorization respond -> do
30163231 if not authorized then yield (respond emptySnapshot)
30173232 else do
30183233 result <- atomic do
3019− p <- *presence
3234+ p <- readAllPresence
30203235 a <- *ackedIds
30213236 ao <- *ackOverflow
30223237 d <- *deliveredIds
@@ -3040,10 +3255,11 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw
30403255 else do
30413256 result <- atomic do
30423257 -- Merge presence (existing entries win on conflict via freshness)
3043− presence <- *presence
3044− let mergedPresence = mergeByPubkeyServer presence (do p <- presenceData; where p.expiresAt > t; yield p)
3258+ allPresence <- readAllPresence
3259+ let mergedPresence = mergeByPubkeyServer allPresence (do p <- presenceData; where p.expiresAt > t; yield p)
30453260 let cappedPresence = takeRelation maxPresenceEntries mergedPresence
3046− set *presence = cappedPresence
3261+ writeAllPresence cappedPresence
3262+ set *totalPresenceCount = count cappedPresence
30473263 -- Union dedup state, respecting capacity limits
30483264 ackedIds <- *ackedIds
30493265 let mergedAcked = union ackedIds ackedIdsData
@@ -3167,6 +3383,26 @@ pollHeartbeat = do
31673383 set *heartbeatCount = hc + 1
31683384 pollHeartbeat
31693385
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
3396+ f <- *forwards
3397+ a <- *ackedIds
3398+ ao <- *ackOverflow
3399+ d <- *deliveredIds
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)
3403+ else yield {}
3404+ backgroundSnapshot
3405+
31703406main = do
31713407 let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
31723408 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
@@ -3184,11 +3420,15 @@ main = do
31843420 && maxSessionsPerKey <= maxSessions
31853421 && maxChallengesPerKey <= maxChallenges
31863422 && maxPollsPerKey <= maxActivePolls
3423+ && pollTimeoutMs < sessionExpiry
3424+ && gossipDedupWindow < presenceRefreshInterval
3425+ && federationFetchTimeoutMs <= backgroundPruneInterval
3426+ && maxAckOverflow <= maxAckedIds
31873427 let peerSecretsOk = none federationPeerSecrets (\p -> isBadServerName (normalizeServerName p.server) || p.secret == "")
31883428 if not configOk
31893429 then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
31903430 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"
31923432 else if not peerSecretsOk
31933433 then println "FATAL: invalid federationPeerSecrets — server names must pass validation and secrets must be non-empty"
31943434 else do
@@ -3212,6 +3452,9 @@ main = do
32123452 (println (" per-peer secrets: " ++ show (count federationPeerSecrets) ++ " configured"))
32133453 when (federationCircuitBreakerThreshold > 0)
32143454 (println (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
3455+ when (autoSnapshotInterval > 0)
3456+ (println (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
32153457 fork backgroundPrune
32163458 fork pollHeartbeat
3459+ when (autoSnapshotInterval > 0) (fork backgroundSnapshot)
32173460 listen 8080 serve