@@ -136,6 +136,9 @@ reconciliationFrequency = 10
136federationLatencyWindowMs : Int<Ms>
136federationLatencyWindowMs : Int<Ms>
137federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
137federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
139+-- Snapshot format version (increment on schema changes to prevent corrupt restores)
139-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
142-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
140maxGossipEventsPerServer = 500 -- max total gossip events per server per window
143maxGossipEventsPerServer = 500 -- max total gossip events per server per window
141gossipEventRateWindowMs : Int<Ms>
144gossipEventRateWindowMs : Int<Ms>
@@ -161,6 +164,10 @@ presenceRefreshInterval = 300000 -- 5 minutes
161presenceRefreshThreshold : Int<Ms>
164presenceRefreshThreshold : Int<Ms>
162presenceRefreshThreshold = presenceRefreshInterval * 2 -- refresh entries expiring within this window
165presenceRefreshThreshold = presenceRefreshInterval * 2 -- refresh entries expiring within this window
163federationEnabled = True {}
166federationEnabled = True {}
167+-- WARNING: DNS rebinding risk — federationUrl resolves DNS at fetch time, not at
168+-- ServerName validation time. A hostname passing isBadServerName could rebind to a
169+-- private IP between validation and fetch. Mitigate at the network layer (e.g.,
170+-- firewall rules blocking outbound connections to RFC1918 ranges from the server process).
164federationUrl = \server -> federationScheme ++ server
171federationUrl = \server -> federationScheme ++ server
165-- Callback verification: outbound federation requests include a random single-use token;
172-- Callback verification: outbound federation requests include a random single-use token;
166-- the receiver calls back to /federation/verify to confirm the sender actually issued it.
173-- the receiver calls back to /federation/verify to confirm the sender actually issued it.
@@ -303,6 +310,10 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
303type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
310type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
304type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
311type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
314+type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId}
315+type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
306-- ============================================
317-- ============================================
307-- Persisted State
318-- Persisted State
308-- ============================================
319-- ============================================
@@ -499,8 +510,8 @@ type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
499-- scope binds the token to a specific request type (gossip/forward/ack) to prevent cross-use
510-- scope binds the token to a specific request type (gossip/forward/ack) to prevent cross-use
500*pendingVerifications : [{token: Text, createdAt: Timestamp, scope: Text}]
511*pendingVerifications : [{token: Text, createdAt: Timestamp, scope: Text}]
502−-- Forward nonce dedup: (messageId, toKey, seenAt) for replay protection beyond message lifetime
513+-- Forward nonce dedup: (messageId, toKey, originServer, seenAt) for replay protection beyond message lifetime
503−*forwardNonces : [{messageId: MessageIdHex, toKey: PubkeyHex, seenAt: Timestamp}]
514+*forwardNonces : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}]
505-- Federation verify per-IP rate limiting (tighter than general IP rate)
516-- Federation verify per-IP rate limiting (tighter than general IP rate)
506*verifyIpRates : [IpRateEntry]
517*verifyIpRates : [IpRateEntry]
@@ -514,6 +525,10 @@ type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
514-- Global federation connection counter (tracks total concurrent outbound connections)
525-- Global federation connection counter (tracks total concurrent outbound connections)
515*federationActiveConnections : Int
526*federationActiveConnections : Int
528+-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
529+-- Polls compare against this to skip expensive presence lookups when nothing changed.
530+*presenceVersion : Timestamp
517-- ============================================
532-- ============================================
519-- ============================================
534-- ============================================
@@ -557,6 +572,10 @@ route ClientApi where
557 POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text, clientIp: Text}
572 POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text, clientIp: Text}
558 -> {ok: Bool, authorized: Bool, ackedCount: Int}
573 -> {ok: Bool, authorized: Bool, ackedCount: Int}
559 = AckMessages
574 = AckMessages
575+ -- Send multiple messages in one request (auth checked once; each message rate-limited independently)
576+ POST {messages: [BatchMessage]} /batch headers {authorization: Text, clientIp: Text}
577+ -> {results: [BatchResult], authorized: Bool}
560 -- Look up which server a pubkey is present on (for federation routing)
579 -- Look up which server a pubkey is present on (for federation routing)
561 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
580 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
562 -> {server: AnyServerName, authorized: Bool}
581 -> {server: AnyServerName, authorized: Bool}
@@ -598,11 +617,11 @@ route MaintenanceApi where
598 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
617 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
599 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
618 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
600 GET /maintenance/snapshot headers {authorization: Text}
619 GET /maintenance/snapshot headers {authorization: Text}
601− -> {ok: Bool, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry]}
620+ -> {ok: Bool, version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry]}
602 = Snapshot
621 = Snapshot
603 -- Restore: import a previously-exported snapshot. Server must be in drain mode
622 -- Restore: import a previously-exported snapshot. Server must be in drain mode
604 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
623 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
605− POST {presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry]} /maintenance/restore headers {authorization: Text}
624+ POST {version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry]} /maintenance/restore headers {authorization: Text}
606 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
625 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
607 = Restore
626 = Restore
@@ -612,7 +631,7 @@ route HealthApi where
612-- Returns counts of in-memory collections (requires maintenance secret)
631-- Returns counts of in-memory collections (requires maintenance secret)
613route StatsApi where
632route StatsApi where
614 GET /stats headers {authorization: Text}
633 GET /stats headers {authorization: Text}
615− -> {sessions: Int, sessionsMax: Int, messages: Int, messagesMax: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pairRates: Int, federationPairRates: Int, pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, federationEnabled: Bool, recentGossip: Int, totalMessageCount: Int, recipientCounts: Int, federationInFlight: Int, messagesPct: Int, presencePct: Int, forwardsPct: Int, sessionsPct: Int, federationAvgLatencyMs: Int, federationMaxLatencyMs: Int, federationCallCount: Int, drainCompletedAt: Timestamp, oldestForwardAge: Int, oldestAckAge: Int, ok: Bool}
634+ -> {sessions: Int, sessionsMax: Int, messages: Int, messagesMax: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pairRates: Int, federationPairRates: Int, pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, federationEnabled: Bool, recentGossip: Int, totalMessageCount: Int, recipientCounts: Int, federationInFlight: Int, hungFederationSlots: Int, messagesPct: Int, presencePct: Int, forwardsPct: Int, sessionsPct: Int, federationAvgLatencyMs: Int, federationMaxLatencyMs: Int, federationCallCount: Int, drainCompletedAt: Timestamp, oldestForwardAge: Int, oldestAckAge: Int, ok: Bool}
617 GET /stats/peers headers {authorization: Text}
636 GET /stats/peers headers {authorization: Text}
618 -> {peers: [PeerStatus], ok: Bool}
637 -> {peers: [PeerStatus], ok: Bool}
@@ -627,6 +646,11 @@ route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | Maintena
627-- Helpers
646-- Helpers
628-- ============================================
647-- ============================================
649+-- Structured log helpers (severity-prefixed for operator filtering)
650+logInfo = \msg -> println ("[INFO] " ++ msg)
651+logWarn = \msg -> println ("[WARN] " ++ msg)
652+logError = \msg -> println ("[ERROR] " ++ msg)
630any : [a] -> (a -> Bool) -> Bool
654any : [a] -> (a -> Bool) -> Bool
631any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
655any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
632none : [a] -> (a -> Bool) -> Bool
656none : [a] -> (a -> Bool) -> Bool
@@ -916,7 +940,7 @@ verifyFederationCallback = \fromServer authorization scope ->
916 result <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
940 result <- fetch (federationUrl fromServer) (VerifyToken {token: token, scope: scope})
917 case result of
941 case result of
918 Err e -> do
942 Err e -> do
919− println ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
943+ logWarn ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
920 yield (False {})
944 yield (False {})
921 Ok r -> yield r.ok
945 Ok r -> yield r.ok
@@ -1020,6 +1044,20 @@ hasSessionForKey = \pubkey b ->
1020 if any bucket (\s -> s.pubkey == pubkey) then yield (True {})
1044 if any bucket (\s -> s.pubkey == pubkey) then yield (True {})
1021 else hasSessionForKey pubkey (b + 1)
1045 else hasSessionForKey pubkey (b + 1)
1047+-- Count sessions for a specific pubkey across all buckets (avoids materializing full session list).
1048+-- Used in verify fast path to skip readAllSessions.
1049+countSessionsForKey : PubkeyHex -> IO Int
1050+countSessionsForKey = \pubkey -> countSessionsInBucket pubkey 0
1052+countSessionsInBucket : PubkeyHex -> Int -> IO Int
1053+countSessionsInBucket = \pubkey b ->
1054+ if b > 15 then yield 0
1056+ bucket <- readSessionBucket b
1057+ let n = count (do s <- bucket; where s.pubkey == pubkey; yield s)
1058+ rest <- countSessionsInBucket pubkey (b + 1)
1023-- Apply a filter/transform to every session bucket
1061-- Apply a filter/transform to every session bucket
1024transformAllSessionBuckets : ([Session] -> [Session]) -> IO ()
1062transformAllSessionBuckets : ([Session] -> [Session]) -> IO ()
1025transformAllSessionBuckets = \f -> do
1063transformAllSessionBuckets = \f -> do
@@ -1124,6 +1162,17 @@ countAllPresence = do
1124 p15 <- *presenceBucket15
1162 p15 <- *presenceBucket15
1125 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)
1163 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)
1165+-- Read only the presence buckets at the given indices (avoids reading all 16)
1166+readSelectedPresenceBuckets : [Int] -> IO [PresenceEntry]
1167+readSelectedPresenceBuckets = \buckets ->
1168+ case headOf buckets of
1169+ Nothing {} -> yield []
1170+ Just {value: b} -> do
1171+ let rest = do x <- buckets; where x != b; yield x
1172+ entries <- readPresenceBucket b
1173+ remaining <- readSelectedPresenceBuckets rest
1174+ yield (union entries remaining)
1127-- Single-pass partitioning: each entry's bucket is computed once (vs 16x in the list-comprehension approach)
1176-- Single-pass partitioning: each entry's bucket is computed once (vs 16x in the list-comprehension approach)
1128writeAllPresence : [PresenceEntry] -> IO ()
1177writeAllPresence : [PresenceEntry] -> IO ()
1129writeAllPresence = \entries -> do
1178writeAllPresence = \entries -> do
@@ -1374,6 +1423,9 @@ hasHungFederationRequest = \server t inFlight ->
1375-- Wrapper around fetch for federation calls: tracks in-flight state, global connection count,
1424-- Wrapper around fetch for federation calls: tracks in-flight state, global connection count,
1376-- logs slow responses, and records latency metrics.
1425-- logs slow responses, and records latency metrics.
1426+-- NOTE: If fetch panics (runtime error), in-flight and connection counters leak until
1427+-- pruneExpired reclaims them (stale threshold: 1.5x federationFetchTimeoutMs). Knot
1428+-- lacks try-finally, so this safety net in pruneExpired is the cleanup mechanism.
1377fetchFederation = \server request -> do
1429fetchFederation = \server request -> do
1378 -- Acquire global federation connection slot
1430 -- Acquire global federation connection slot
1379 atomic do
1431 atomic do
@@ -1390,7 +1442,7 @@ fetchFederation = \server request -> do
1390 endTime <- now
1442 endTime <- now
1391 let latency = endTime - startTime
1443 let latency = endTime - startTime
1392 when (latency > federationFetchTimeoutMs)
1444 when (latency > federationFetchTimeoutMs)
1393− (println ("federation: slow response from " ++ server ++ " (" ++ show latency ++ "ms)"))
1445+ (logWarn ("federation: slow response from " ++ server ++ " (" ++ show latency ++ "ms)"))
1394 atomic do
1446 atomic do
1395 total <- *federationLatencyTotal
1447 total <- *federationLatencyTotal
1396 cnt <- *federationLatencyCount
1448 cnt <- *federationLatencyCount
@@ -1659,6 +1711,7 @@ pruneExpired = \t -> do
1659 tpc <- *totalPresenceCount
1711 tpc <- *totalPresenceCount
1660 set *totalPresenceCount = max 0 (tpc - presenceResults.totalRemoved)
1712 set *totalPresenceCount = max 0 (tpc - presenceResults.totalRemoved)
1661 set *knownPeerServers = presenceResults.keptPeerServers
1713 set *knownPeerServers = presenceResults.keptPeerServers
1714+ when (presenceResults.totalRemoved > 0) (set *presenceVersion = t)
1662 let presenceResult = {expiredLocal: presenceResults.expiredLocal, peerServers: presenceResults.allPeerServers}
1715 let presenceResult = {expiredLocal: presenceResults.expiredLocal, peerServers: presenceResults.allPeerServers}
1663 -- Prune auth state
1716 -- Prune auth state
1664 atomic do
1717 atomic do
@@ -1741,7 +1794,7 @@ pruneExpired = \t -> do
1741 let staleThreshold = federationFetchTimeoutMs * 3 / 2
1794 let staleThreshold = federationFetchTimeoutMs * 3 / 2
1742 let staleInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= staleThreshold; yield r
1795 let staleInFlight = do r <- inFlight; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= staleThreshold; yield r
1743 when (count staleInFlight > 0)
1796 when (count staleInFlight > 0)
1744− (println ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))
1797+ (logWarn ("pruneExpired: cleaning " ++ show (count staleInFlight) ++ " stale in-flight entries (leaked slots)"))
1745 set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < staleThreshold); yield r
1798 set *federationInFlight = do r <- inFlight; where r.count > 0 && (r.firstStartedAt == 0 || t - r.firstStartedAt < staleThreshold); yield r
1746 -- Prune stale federation replay dedup entries
1799 -- Prune stale federation replay dedup entries
1747 recentFedReqs <- *recentFederationRequests
1800 recentFedReqs <- *recentFederationRequests
@@ -1785,7 +1838,7 @@ sendBatchedGossipToServer = \server allEvents hops ->
1785 events: batch, hopsRemaining: hops, fromServer: serverHostN,
1838 events: batch, hopsRemaining: hops, fromServer: serverHostN,
1786 requestTimestamp: t})
1839 requestTimestamp: t})
1787 case result of
1840 case result of
1788− Err e -> println ("federation: gossip to " ++ server ++ " failed: " ++ show e)
1841+ Err e -> logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
1789 Ok _ -> yield {})
1842 Ok _ -> yield {})
1791sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}
1844sendBatchedGossip : [ServerName] -> [GossipEvent] -> Int -> IO {network} {}
@@ -1796,11 +1849,11 @@ broadcastOfflineGossip = \t peerServers expiredLocal ->
1796 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do
1849 when (federationEnabled && count peerServers > 0 && count expiredLocal > 0) (do
1797 failures <- *forwardFailures
1850 failures <- *forwardFailures
1798 if isFederationCircuitOpen failures t
1851 if isFederationCircuitOpen failures t
1799− then println ("federation: circuit breaker open, skipping offline gossip broadcast")
1852+ then logWarn ("federation: circuit breaker open, skipping offline gossip broadcast")
1801 let cappedExpired = takeRelation maxPresenceEntries expiredLocal
1854 let cappedExpired = takeRelation maxPresenceEntries expiredLocal
1802 when (count expiredLocal > maxPresenceEntries)
1855 when (count expiredLocal > maxPresenceEntries)
1803− (println ("gossip: capped offline broadcast to " ++ show maxPresenceEntries ++ " entries (had " ++ show (count expiredLocal) ++ ")"))
1856+ (logWarn ("gossip: capped offline broadcast to " ++ show maxPresenceEntries ++ " entries (had " ++ show (count expiredLocal) ++ ")"))
1804 let gossipEvents = mergeByKey (\a b -> a.pubkey == b.pubkey) (\x -> x.timestamp) [] (offlineGossipEvents cappedExpired)
1857 let gossipEvents = mergeByKey (\a b -> a.pubkey == b.pubkey) (\x -> x.timestamp) [] (offlineGossipEvents cappedExpired)
1805 sendBatchedGossip peerServers gossipEvents gossipOriginHops)
1858 sendBatchedGossip peerServers gossipEvents gossipOriginHops)
@@ -1893,7 +1946,7 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
1893 -- requestTimestamp == 0 means the peer doesn't support this field yet (accept).
1946 -- requestTimestamp == 0 means the peer doesn't support this field yet (accept).
1894 else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance)
1947 else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance)
1896− println ("federation: clock skew " ++ show (t - requestTimestamp) ++ "ms from " ++ fromServer ++ ", rejecting (tolerance: " ++ show clockSkewTolerance ++ "ms)")
1949+ logWarn ("federation: clock skew " ++ show (t - requestTimestamp) ++ "ms from " ++ fromServer ++ ", rejecting (tolerance: " ++ show clockSkewTolerance ++ "ms)")
1897 yield (respond {ok: False {}, error: "clock_skew"})
1950 yield (respond {ok: False {}, error: "clock_skew"})
1898 else if not precondition
1951 else if not precondition
1899 then yield (respond {ok: False {}, error: "invalid_request"})
1952 then yield (respond {ok: False {}, error: "invalid_request"})
@@ -1912,19 +1965,19 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
1912 else yield (False {})
1965 else yield (False {})
1913 if replayDetected
1966 if replayDetected
1915− println ("federation: replay detected from " ++ fromServer ++ " (timestamp: " ++ show requestTimestamp ++ ")")
1968+ logWarn ("federation: replay detected from " ++ fromServer ++ " (timestamp: " ++ show requestTimestamp ++ ")")
1916 yield (respond {ok: False {}, error: "replay_detected"})
1969 yield (respond {ok: False {}, error: "replay_detected"})
1918 globalLimited <- checkGlobalRate t
1971 globalLimited <- checkGlobalRate t
1919 if globalLimited
1972 if globalLimited
1921− println ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)
1974+ logWarn ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)
1922 yield (respond {ok: False {}, error: "rate_limited"})
1975 yield (respond {ok: False {}, error: "rate_limited"})
1924 rateLimited <- checkFederationRate fromServer t
1977 rateLimited <- checkFederationRate fromServer t
1925 if rateLimited
1978 if rateLimited
1927− println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
1980+ logWarn ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
1928 yield (respond {ok: False {}, error: "rate_limited"})
1981 yield (respond {ok: False {}, error: "rate_limited"})
1930 clearFailure fromServer
1983 clearFailure fromServer
@@ -1960,13 +2013,14 @@ ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of
1960 let remainingBuckets = do x <- bucketIds; where x != b; yield x
2013 let remainingBuckets = do x <- bucketIds; where x != b; yield x
1961 let bucketEntries = do e <- entries; where pubkeyBucket e.toKey == b; yield e
2014 let bucketEntries = do e <- entries; where pubkeyBucket e.toKey == b; yield e
1962 atomic do
2015 atomic do
1963− allPresence <- readAllPresence
2016+ -- All entries in this batch share pubkeyBucket == b, so only that presence bucket is needed
2017+ bucketPresence <- readPresenceBucket b
1964 forwards <- *forwards
2018 forwards <- *forwards
1965 msgs <- readMessageBucket b
2019 msgs <- readMessageBucket b
1966 tc <- *totalMessageCount
2020 tc <- *totalMessageCount
1967 rc <- *recipientMsgCounts
2021 rc <- *recipientMsgCounts
1968 let cleanup = fold (\acc rf ->
2022 let cleanup = fold (\acc rf ->
1969− let hasLocal = any allPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in
2023+ let hasLocal = any bucketPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in
1970 let hasOtherFwds = any forwards (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey) in
2024 let hasOtherFwds = any forwards (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey) in
1971 if hasLocal || hasOtherFwds then acc
2025 if hasLocal || hasOtherFwds then acc
1972 else case headOf (do m <- acc.msgs; where m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod; yield m) of
2026 else case headOf (do m <- acc.msgs; where m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod; yield m) of
@@ -2029,7 +2083,7 @@ incrementForwardRetries = \messageId toServer -> atomic do
2029 if shouldDrop
2083 if shouldDrop
2031 set *forwards = do f <- forwards; where not (f.messageId == messageId && f.toServer == toServer); yield f
2085 set *forwards = do f <- forwards; where not (f.messageId == messageId && f.toServer == toServer); yield f
2032− println ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
2086+ logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
2033 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)
2087 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)
2035retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
2089retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
@@ -2049,7 +2103,7 @@ retryForwardsForServer = \entries remaining ->
2049 authorization: auth})
2103 authorization: auth})
2050 case result of
2104 case result of
2051 Err e -> do
2105 Err e -> do
2052− println ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
2106+ logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
2053 recordFailure f.toServer
2107 recordFailure f.toServer
2054 incrementForwardRetries f.messageId f.toServer
2108 incrementForwardRetries f.messageId f.toServer
2055 -- Network error: stop retrying this server
2109 -- Network error: stop retrying this server
@@ -2058,7 +2112,7 @@ retryForwardsForServer = \entries remaining ->
2058 removeForward f.messageId f.toServer
2112 removeForward f.messageId f.toServer
2059 clearFailure f.toServer
2113 clearFailure f.toServer
2061− println ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)
2115+ logWarn ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)
2062 incrementForwardRetries f.messageId f.toServer
2116 incrementForwardRetries f.messageId f.toServer
2063 retryForwardsForServer rest (remaining - 1)
2117 retryForwardsForServer rest (remaining - 1)
@@ -2070,7 +2124,7 @@ retryPendingForwards = \pending -> do
2070 t <- now
2124 t <- now
2071 failures <- *forwardFailures
2125 failures <- *forwardFailures
2072 if isFederationCircuitOpen failures t
2126 if isFederationCircuitOpen failures t
2073− then println ("federation: circuit breaker open (" ++ show (count failures) ++ " peers failing), skipping forward retries")
2127+ then logWarn ("federation: circuit breaker open (" ++ show (count failures) ++ " peers failing), skipping forward retries")
2075 inFlight <- *federationInFlight
2129 inFlight <- *federationInFlight
2076 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2130 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
@@ -2096,7 +2150,7 @@ forwardQueuedMessages = \targets t -> do
2096 let servers = collectPeerServers newForwards (\x -> x.toServer)
2150 let servers = collectPeerServers newForwards (\x -> x.toServer)
2097 failures <- *forwardFailures
2151 failures <- *forwardFailures
2098 if isFederationCircuitOpen failures t
2152 if isFederationCircuitOpen failures t
2099− then println ("federation: circuit breaker open, skipping queued message forwarding")
2153+ then logWarn ("federation: circuit breaker open, skipping queued message forwarding")
2101 inFlight <- *federationInFlight
2155 inFlight <- *federationInFlight
2102 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2156 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
@@ -2118,7 +2172,7 @@ sendBatchedAcksToServer = \serverAcks server ->
2118 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
2172 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
2119 case result of
2173 case result of
2120 Err e -> do
2174 Err e -> do
2121− println ("federation: ack to " ++ server ++ " failed: " ++ show e)
2175+ logWarn ("federation: ack to " ++ server ++ " failed: " ++ show e)
2122 recordFailure server
2176 recordFailure server
2123 -- Increment retry counters; drop entries exceeding maxAckRetries to deliveredIds
2177 -- Increment retry counters; drop entries exceeding maxAckRetries to deliveredIds
2124 dropped <- atomic do
2178 dropped <- atomic do
@@ -2136,7 +2190,7 @@ sendBatchedAcksToServer = \serverAcks server ->
2136 set *deliveredIds = if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined)
2190 set *deliveredIds = if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined)
2137 yield dropCount
2191 yield dropCount
2138 when (dropped > 0)
2192 when (dropped > 0)
2139− (println ("federation: dropped " ++ show dropped ++ " ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
2193+ (logWarn ("federation: dropped " ++ show dropped ++ " ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
2140 Ok _ -> do
2194 Ok _ -> do
2141 clearFailure server
2195 clearFailure server
2142 atomic do
2196 atomic do
@@ -2151,7 +2205,7 @@ sendAcksAndClean = \pendingAcks -> do
2151 t <- now
2205 t <- now
2152 failures <- *forwardFailures
2206 failures <- *forwardFailures
2153 if isFederationCircuitOpen failures t
2207 if isFederationCircuitOpen failures t
2154− then println ("federation: circuit breaker open, skipping ack retries")
2208+ then logWarn ("federation: circuit breaker open, skipping ack retries")
2156 inFlight <- *federationInFlight
2210 inFlight <- *federationInFlight
2157 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2211 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
@@ -2169,7 +2223,7 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
2169 fetchResult <- fetchFederationAck server (do a <- acksForServer evictedToRetry server; yield a.messageId)
2223 fetchResult <- fetchFederationAck server (do a <- acksForServer evictedToRetry server; yield a.messageId)
2170 case fetchResult of
2224 case fetchResult of
2171 Err e -> do
2225 Err e -> do
2172− println ("federation: evicted ack to " ++ server ++ " failed: " ++ show e)
2226+ logWarn ("federation: evicted ack to " ++ server ++ " failed: " ++ show e)
2173 overflow <- atomic do
2227 overflow <- atomic do
2174 ackedIds <- *ackedIds
2228 ackedIds <- *ackedIds
2175 let reinsert = do a <- acksForServer evictedToRetry server; yield {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1}
2229 let reinsert = do a <- acksForServer evictedToRetry server; yield {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1}
@@ -2198,10 +2252,10 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
2198 set *deliveredIds = if deliveredExcess <= 0 then combinedDelivered
2252 set *deliveredIds = if deliveredExcess <= 0 then combinedDelivered
2199 else (evictSmallest combinedDelivered (max 0 deliveredExcess) (\d -> d.deliveredAt)).kept
2253 else (evictSmallest combinedDelivered (max 0 deliveredExcess) (\d -> d.deliveredAt)).kept
2200 when (deliveredExcess > 0)
2254 when (deliveredExcess > 0)
2201− (println ("ackMessages: deliveredIds at capacity (" ++ show maxDeliveredIds ++ "), " ++ show deliveredExcess ++ " oldest dedup entries permanently removed for " ++ server))
2255+ (logWarn ("ackMessages: deliveredIds at capacity (" ++ show maxDeliveredIds ++ "), " ++ show deliveredExcess ++ " oldest dedup entries permanently removed for " ++ server))
2202 yield overflowExcess
2256 yield overflowExcess
2203 when (dropped > 0)
2257 when (dropped > 0)
2204− (println ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries moved to delivered set for " ++ server)))
2258+ (logWarn ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries moved to delivered set for " ++ server)))
2205 Ok _ -> yield {})
2259 Ok _ -> yield {})
2207-- Retry acks sitting in the overflow buffer (last chance before they become deliveredIds)
2261-- Retry acks sitting in the overflow buffer (last chance before they become deliveredIds)
@@ -2210,7 +2264,7 @@ retryOverflowAcks = \overflow -> do
2210 t <- now
2264 t <- now
2211 failures <- *forwardFailures
2265 failures <- *forwardFailures
2212 if isFederationCircuitOpen failures t
2266 if isFederationCircuitOpen failures t
2213− then println ("federation: circuit breaker open, skipping overflow ack retries")
2267+ then logWarn ("federation: circuit breaker open, skipping overflow ack retries")
2215 inFlight <- *federationInFlight
2269 inFlight <- *federationInFlight
2216 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
2270 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight; yield s
@@ -2219,7 +2273,7 @@ retryOverflowAcks = \overflow -> do
2219 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
2273 result <- fetchFederationAck server (do a <- batch; yield a.messageId)
2220 case result of
2274 case result of
2221 Err e -> do
2275 Err e -> do
2222− println ("federation: overflow ack to " ++ server ++ " failed: " ++ show e)
2276+ logWarn ("federation: overflow ack to " ++ server ++ " failed: " ++ show e)
2223 recordFailure server
2277 recordFailure server
2224 -- Increment retry counters in overflow; drop entries exceeding maxAckRetries to deliveredIds
2278 -- Increment retry counters in overflow; drop entries exceeding maxAckRetries to deliveredIds
2225 dropped <- atomic do
2279 dropped <- atomic do
@@ -2237,7 +2291,7 @@ retryOverflowAcks = \overflow -> do
2237 set *deliveredIds = if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined)
2291 set *deliveredIds = if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined)
2238 yield dropCount
2292 yield dropCount
2239 when (dropped > 0)
2293 when (dropped > 0)
2240− (println ("federation: dropped " ++ show dropped ++ " overflow ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
2294+ (logWarn ("federation: dropped " ++ show dropped ++ " overflow ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
2241 Ok _ -> do
2295 Ok _ -> do
2242 clearFailure server
2296 clearFailure server
2243 atomic do
2297 atomic do
@@ -2266,7 +2320,7 @@ waitForInFlightDrain = \deadline -> do
2266 if activeCount <= 0 then yield {}
2320 if activeCount <= 0 then yield {}
2267 else retry
2321 else retry
2268 timedOut <- *drainTimedOut
2322 timedOut <- *drainTimedOut
2269− when timedOut (println "drain: timed out waiting for in-flight federation tasks")
2323+ when timedOut (logWarn "drain: timed out waiting for in-flight federation tasks")
2271-- Full maintenance cycle: prune expired state, retry pending forwards and acks,
2325-- Full maintenance cycle: prune expired state, retry pending forwards and acks,
2272-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
2326-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
@@ -2288,7 +2342,7 @@ pruneAndRetry = \t -> do
2288 set *totalSessionCount = actual
2342 set *totalSessionCount = actual
2289 yield (actual - stored)
2343 yield (actual - stored)
2290 when (presDrift != 0 || sessDrift != 0)
2344 when (presDrift != 0 || sessDrift != 0)
2291− (println ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))
2345+ (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))
2292 -- Message reconciliation is expensive (reads all 16 buckets + rebuilds recipient counts), keep periodic
2346 -- Message reconciliation is expensive (reads all 16 buckets + rebuilds recipient counts), keep periodic
2293 pc <- *pruneCount
2347 pc <- *pruneCount
2294 when (pc > 0 && pc - (pc / reconciliationFrequency) * reconciliationFrequency == 0) (do
2348 when (pc > 0 && pc - (pc / reconciliationFrequency) * reconciliationFrequency == 0) (do
@@ -2300,7 +2354,7 @@ pruneAndRetry = \t -> do
2300 set *recipientMsgCounts = countByRecipient allMsgs
2354 set *recipientMsgCounts = countByRecipient allMsgs
2301 yield (actual - stored)
2355 yield (actual - stored)
2302 when (msgDrift != 0)
2356 when (msgDrift != 0)
2303− (println ("reconciliation: corrected message drift: " ++ show msgDrift)))
2357+ (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift)))
2304 pendingForwards <- atomic do f <- *forwards; yield f
2358 pendingForwards <- atomic do f <- *forwards; yield f
2305 ackState <- atomic do
2359 ackState <- atomic do
2306 ackedIds2 <- *ackedIds
2360 ackedIds2 <- *ackedIds
@@ -2311,19 +2365,22 @@ pruneAndRetry = \t -> do
2311 refreshResult <- atomic do
2365 refreshResult <- atomic do
2312 lastRefresh <- *lastPresenceRefresh
2366 lastRefresh <- *lastPresenceRefresh
2313 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval
2367 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval
2368+ -- Early return: skip expensive readAllPresence + readAllSessions on ~9/10 prune cycles
2369+ if not shouldRefresh then yield {extendedActive: [], nearExpiryActive: [], refreshPeers: [], shouldRefresh: False {}}
2314− allPresence <- readAllPresence
2371+ allPresence <- readAllPresence
2315− allSessions <- readAllSessions
2372+ allSessions <- readAllSessions
2316− -- Extend presence for keys with active sessions nearing expiry (sliding window)
2373+ -- Extend presence for keys with active sessions nearing expiry (sliding window)
2317− let localNearExpiry = if shouldRefresh then (do p <- allPresence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []
2374+ let localNearExpiry = do p <- allPresence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p
2318− let toExtend = do p <- localNearExpiry; where any allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
2375+ let toExtend = do p <- localNearExpiry; where any allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
2319− let notExtended = do p <- localNearExpiry; where none allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
2376+ let notExtended = do p <- localNearExpiry; where none allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
2320− when (count toExtend > 0) (do
2377+ when (count toExtend > 0) (do
2321− 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)
2378+ 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)
2322− writeAllPresence updatedPresence)
2379+ writeAllPresence updatedPresence
2380+ set *presenceVersion = t)
2323− knownPeers <- *knownPeerServers
2381+ knownPeers <- *knownPeerServers
2324− let refreshPeers = if shouldRefresh then knownPeers else []
2325− when shouldRefresh (set *lastPresenceRefresh = t)
2382+ set *lastPresenceRefresh = t
2326− yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
2383+ yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: knownPeers, shouldRefresh: True {}}
2327 -- Reset federation latency metrics periodically for windowed averages (prevents
2384 -- Reset federation latency metrics periodically for windowed averages (prevents
2328 -- lifetime-cumulative average from masking current conditions)
2385 -- lifetime-cumulative average from masking current conditions)
2329 atomic do
2386 atomic do
@@ -2649,7 +2706,7 @@ handleChallenge = \pubkey clientIp respond -> do
2649 preAuth <- checkPreAuth clientIp t
2706 preAuth <- checkPreAuth clientIp t
2650 if preAuth != ""
2707 if preAuth != ""
2652− when (preAuth == "ip_rate_limited") (println ("challenge: IP rate limited for " ++ clientIp))
2709+ when (preAuth == "ip_rate_limited") (logWarn ("challenge: IP rate limited for " ++ clientIp))
2653 yield (respond {challenge: "", ok: False {}, expiresAt: 0})
2710 yield (respond {challenge: "", ok: False {}, expiresAt: 0})
2655 authLimited <- checkAuthGlobalRate t
2712 authLimited <- checkAuthGlobalRate t
@@ -2681,9 +2738,9 @@ handleChallenge = \pubkey clientIp respond -> do
2681 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
2738 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
2682 yield {accepted: True {}, rateLimited: False {}}
2739 yield {accepted: True {}, rateLimited: False {}}
2683 when result.rateLimited
2740 when result.rateLimited
2684− (println ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
2741+ (logWarn ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
2685 when (not result.accepted && not result.rateLimited)
2742 when (not result.accepted && not result.rateLimited)
2686− (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
2743+ (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
2687 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
2744 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
2689-- POST /auth/verify: Verify Ed25519 signature over challenge string.
2746-- POST /auth/verify: Verify Ed25519 signature over challenge string.
@@ -2694,7 +2751,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
2694 preAuth <- checkPreAuth clientIp t
2751 preAuth <- checkPreAuth clientIp t
2695 if preAuth != ""
2752 if preAuth != ""
2697− when (preAuth == "ip_rate_limited") (println ("verify: IP rate limited for " ++ clientIp))
2754+ when (preAuth == "ip_rate_limited") (logWarn ("verify: IP rate limited for " ++ clientIp))
2698 yield (respond {token: "", expiresAt: 0})
2755 yield (respond {token: "", expiresAt: 0})
2700 authRateLimited <- atomic do
2757 authRateLimited <- atomic do
@@ -2706,7 +2763,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
2706 set *challengeRates = rateResult.entries
2763 set *challengeRates = rateResult.entries
2707 yield rateResult.limited
2764 yield rateResult.limited
2708 if authRateLimited then do
2765 if authRateLimited then do
2709− println ("verify: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey)
2766+ logWarn ("verify: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey)
2710 yield (respond {token: "", expiresAt: 0})
2767 yield (respond {token: "", expiresAt: 0})
2712 authGlobalLimited <- checkAuthGlobalRate t
2769 authGlobalLimited <- checkAuthGlobalRate t
@@ -2742,12 +2799,11 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
2742 isDraining <- *draining
2799 isDraining <- *draining
2743 if isDraining then yield (Nothing {})
2800 if isDraining then yield (Nothing {})
2745− -- Read all session buckets for per-key enforcement (revokeOthers, capPerKey)
2746− allSessions <- readAllSessions
2747 tpc <- *totalPresenceCount
2802 tpc <- *totalPresenceCount
2748 knownPeers <- *knownPeerServers
2803 knownPeers <- *knownPeerServers
2804+ -- Count per-key sessions without materializing all sessions (O(sessions) but no union)
2749− let sessionsForKey = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
2805+ sessionsForKey <- countSessionsForKey pubkey
2750− let totalSessions = count allSessions
2806+ tsc <- *totalSessionCount
2751 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp}
2807 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp}
2752 -- Update presence: replace local entry for this pubkey, evict oldest if over capacity
2808 -- Update presence: replace local entry for this pubkey, evict oldest if over capacity
2753 let pb = pubkeyBucket pubkey
2809 let pb = pubkeyBucket pubkey
@@ -2759,18 +2815,19 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
2759 writePresenceBucket pb (union bucketEvicted [newPresenceEntry])
2815 writePresenceBucket pb (union bucketEvicted [newPresenceEntry])
2760 let presenceDelta = count (union bucketEvicted [newPresenceEntry]) - count presenceBkt
2816 let presenceDelta = count (union bucketEvicted [newPresenceEntry]) - count presenceBkt
2761 set *totalPresenceCount = tpc + presenceDelta
2817 set *totalPresenceCount = tpc + presenceDelta
2818+ set *presenceVersion = t
2762 let peerServers = if federationEnabled then knownPeers else []
2819 let peerServers = if federationEnabled then knownPeers else []
2763 -- Fast path: no revocation, under per-key and global limits — only write one bucket
2820 -- Fast path: no revocation, under per-key and global limits — only write one bucket
2764− if not revokeOthers && sessionsForKey < maxSessionsPerKey && totalSessions < maxSessions
2821+ if not revokeOthers && sessionsForKey < maxSessionsPerKey && tsc < maxSessions
2766 let nb = tokenBucket token
2823 let nb = tokenBucket token
2767 currentBucket <- readSessionBucket nb
2824 currentBucket <- readSessionBucket nb
2768 writeSessionBucket nb (union currentBucket [newSession])
2825 writeSessionBucket nb (union currentBucket [newSession])
2769− sc <- *totalSessionCount
2770− set *totalSessionCount = sc + 1
2826+ set *totalSessionCount = tsc + 1
2771 yield (Just {value: {peerServers: peerServers, revokedCount: 0}})
2827 yield (Just {value: {peerServers: peerServers, revokedCount: 0}})
2773− -- Slow path: revocation or capacity pressure — redistribute across all buckets
2829+ -- Slow path: revocation or capacity pressure — need all sessions
2830+ allSessions <- readAllSessions
2774 let revokedCount = if revokeOthers then sessionsForKey else 0
2831 let revokedCount = if revokeOthers then sessionsForKey else 0
2775 let afterRevoke = if revokeOthers then do s <- allSessions; where s.pubkey != pubkey; yield s else allSessions
2832 let afterRevoke = if revokeOthers then do s <- allSessions; where s.pubkey != pubkey; yield s else allSessions
2776 let capped = capPerKeyAndGlobal afterRevoke (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
2833 let capped = capPerKeyAndGlobal afterRevoke (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
@@ -2798,7 +2855,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
2798 Nothing {} -> yield (respond {token: "", expiresAt: 0})
2855 Nothing {} -> yield (respond {token: "", expiresAt: 0})
2799 Just {value: result} -> do
2856 Just {value: result} -> do
2800 when (revokeOthers && result.revokedCount > 0)
2857 when (revokeOthers && result.revokedCount > 0)
2801− (println ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
2858+ (logInfo ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
2802 fork (when federationEnabled
2859 fork (when federationEnabled
2803 (sendBatchedGossip result.peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops))
2860 (sendBatchedGossip result.peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops))
2804 yield (respond {token: token, expiresAt: expiry})
2861 yield (respond {token: token, expiresAt: expiry})
@@ -2825,8 +2882,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
2825 presenceUpdates <- if result.authorized && count cappedWatchKeys > 0 then do
2882 presenceUpdates <- if result.authorized && count cappedWatchKeys > 0 then do
2826 t2 <- now
2883 t2 <- now
2827 atomic do
2884 atomic do
2828− allPresence <- readAllPresence
2885+ -- Read only the presence buckets needed for watched keys (not all 16)
2886+ let neededBuckets = uniqueBucketsFor cappedWatchKeys (\k -> k)
2887+ relevantPresence <- readSelectedPresenceBuckets neededBuckets
2829− yield (do key <- cappedWatchKeys; yield {pubkey: key, online: count (presenceForKey allPresence key t2) > 0})
2888+ yield (do key <- cappedWatchKeys; yield {pubkey: key, online: count (presenceForKey relevantPresence key t2) > 0})
2830 else yield []
2889 else yield []
2831 fork (when federationEnabled (do
2890 fork (when federationEnabled (do
2832 sendAcksAndClean result.pendingAcks
2891 sendAcksAndClean result.pendingAcks
@@ -2847,9 +2906,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2847 then yield (respond {status: MsRejected {}, messageId: ""})
2906 then yield (respond {status: MsRejected {}, messageId: ""})
2849 when (timestamp > t + clockSkewTolerance)
2908 when (timestamp > t + clockSkewTolerance)
2850− (println ("sendMessage: future timestamp detected (+" ++ show (timestamp - t) ++ "ms ahead), clamping to now"))
2909+ (logWarn ("sendMessage: future timestamp detected (+" ++ show (timestamp - t) ++ "ms ahead), clamping to now"))
2851 when (timestamp < t - messageMaxAge / 2)
2910 when (timestamp < t - messageMaxAge / 2)
2852− (println ("sendMessage: stale timestamp (" ++ show ((t - timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
2911+ (logWarn ("sendMessage: stale timestamp (" ++ show ((t - timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
2853 let clampedTs = clampTimestamp timestamp t
2912 let clampedTs = clampTimestamp timestamp t
2854 result <- atomic do
2913 result <- atomic do
2855 draining <- *draining
2914 draining <- *draining
@@ -2935,9 +2994,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2935 set *forwards = union forwards cappedFwds)
2994 set *forwards = union forwards cappedFwds)
2936 yield {send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}, globalCapReached: False {}})
2995 yield {send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}, globalCapReached: False {}})
2937 when result.senderRateLimited
2996 when result.senderRateLimited
2938− (println ("sendMessage: rate limited (" ++ show maxSendRatePerKey ++ "/s per key)"))
2997+ (logWarn ("sendMessage: rate limited (" ++ show maxSendRatePerKey ++ "/s per key)"))
2939 when result.globalCapReached
2998 when result.globalCapReached
2940− (println ("sendMessage: global message cap reached (" ++ show maxMessages ++ "), rejecting"))
2999+ (logWarn ("sendMessage: global message cap reached (" ++ show maxMessages ++ "), rejecting"))
2941 case result.send of
3000 case result.send of
2942 SendDone {delivery, remoteServers} ->
3001 SendDone {delivery, remoteServers} ->
2943 when (count remoteServers > 0 && federationEnabled) (
3002 when (count remoteServers > 0 && federationEnabled) (
@@ -2956,14 +3015,14 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2956 authorization: fwdAuth})
3015 authorization: fwdAuth})
2957 case fwdResult of
3016 case fwdResult of
2958 Err e -> do
3017 Err e -> do
2959− println ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
3018+ logWarn ("federation: send forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
2960 recordFailure server
3019 recordFailure server
2961 Ok response ->
3020 Ok response ->
2962 if response.ok then do
3021 if response.ok then do
2963 removeForward effectiveMsgId server
3022 removeForward effectiveMsgId server
2964 clearFailure server
3023 clearFailure server
2966− println ("federation: send forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))
3025+ logWarn ("federation: send forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))
2967 _ -> yield {}
3026 _ -> yield {}
2968 yield (case result.send of
3027 yield (case result.send of
2969 SendFailed {status} -> respond {status: status, messageId: ""}
3028 SendFailed {status} -> respond {status: status, messageId: ""}
@@ -3077,7 +3136,7 @@ handleRevoke = \targetToken authorization clientIp respond -> do
3077 set *totalSessionCount = max 0 (sc - 1)
3136 set *totalSessionCount = max 0 (sc - 1)
3078 yield (True {})
3137 yield (True {})
3079 else yield (False {})
3138 else yield (False {})
3080− when removed (println ("revoke: session revoked for " ++ callerPubkey))
3139+ when removed (logInfo ("revoke: session revoked for " ++ callerPubkey))
3081 yield (respond {ok: removed, authorized: True {}}))
3140 yield (respond {ok: removed, authorized: True {}}))
3083-- POST /auth/logout: Voluntary disconnect. Removes the caller's session and, if it was
3142-- POST /auth/logout: Voluntary disconnect. Removes the caller's session and, if it was
@@ -3121,6 +3180,7 @@ handleLogout = \authorization clientIp respond -> do
3121 let removedCount = count presenceBkt - count (do p <- presenceBkt; where p.pubkey != pubkey || not (isLocalServer p.server); yield p)
3180 let removedCount = count presenceBkt - count (do p <- presenceBkt; where p.pubkey != pubkey || not (isLocalServer p.server); yield p)
3122 tpc <- *totalPresenceCount
3181 tpc <- *totalPresenceCount
3123 set *totalPresenceCount = max 0 (tpc - removedCount)
3182 set *totalPresenceCount = max 0 (tpc - removedCount)
3183+ set *presenceVersion = t
3124 peerServers <- *knownPeerServers
3184 peerServers <- *knownPeerServers
3125 yield (Just {value: {pubkey: pubkey, gossipOffline: True {}, peerServers: peerServers}})
3185 yield (Just {value: {pubkey: pubkey, gossipOffline: True {}, peerServers: peerServers}})
3126 else yield (Just {value: {pubkey: pubkey, gossipOffline: False {}, peerServers: []}})
3186 else yield (Just {value: {pubkey: pubkey, gossipOffline: False {}, peerServers: []}})
@@ -3128,7 +3188,7 @@ handleLogout = \authorization clientIp respond -> do
3128 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
3188 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
3129 Just {value: r} -> do
3189 Just {value: r} -> do
3130 when r.gossipOffline (do
3190 when r.gossipOffline (do
3131− println ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")
3191+ logInfo ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")
3132 when (federationEnabled && count r.peerServers > 0)
3192 when (federationEnabled && count r.peerServers > 0)
3133 (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
3193 (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
3134 yield (respond {ok: True {}, authorized: True {}})
3194 yield (respond {ok: True {}, authorized: True {}})
@@ -3155,7 +3215,7 @@ handleVerifyToken = \token scope clientIp respond -> do
3155 set *verifyIpRates = result.entries
3215 set *verifyIpRates = result.entries
3156 yield result.limited
3216 yield result.limited
3157 if verifyLimited then do
3217 if verifyLimited then do
3158− println ("federation/verify: rate limited for IP " ++ clientIp)
3218+ logWarn ("federation/verify: rate limited for IP " ++ clientIp)
3159 yield (respond {ok: False {}})
3219 yield (respond {ok: False {}})
3161 found <- atomic do
3221 found <- atomic do
@@ -3164,7 +3224,7 @@ handleVerifyToken = \token scope clientIp respond -> do
3164 let match = any pending (\p -> safeCompare p.token token && p.scope == scope && t - p.createdAt < verificationTokenExpiry)
3224 let match = any pending (\p -> safeCompare p.token token && p.scope == scope && t - p.createdAt < verificationTokenExpiry)
3165 when match (set *pendingVerifications = do p <- pending; where not (safeCompare p.token token); yield p)
3225 when match (set *pendingVerifications = do p <- pending; where not (safeCompare p.token token); yield p)
3166 yield match
3226 yield match
3167− when (not found) (println ("federation/verify: failed attempt from " ++ clientIp ++ " (scope: " ++ scope ++ ")"))
3227+ when (not found) (logWarn ("federation/verify: failed attempt from " ++ clientIp ++ " (scope: " ++ scope ++ ")"))
3168 yield (respond {ok: found})
3228 yield (respond {ok: found})
3170-- POST /federation/gossip: Receive presence gossip from a peer server.
3230-- POST /federation/gossip: Receive presence gossip from a peer server.
@@ -3193,7 +3253,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
3193 set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs; yield r) else newEntries
3253 set *gossipEventRates = if count newEntries > maxGossipEventRateEntries then (do r <- newEntries; where t - r.windowStart < gossipEventRateWindowMs; yield r) else newEntries
3194 yield (False {})
3254 yield (False {})
3195 if eventRateLimited then do
3255 if eventRateLimited then do
3196− println ("gossip: event volume rate limited for " ++ fromServer ++ " (" ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms)")
3256+ logWarn ("gossip: event volume rate limited for " ++ fromServer ++ " (" ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms)")
3197 yield (respond {ok: False {}, error: "event_rate_limited"})
3257 yield (respond {ok: False {}, error: "event_rate_limited"})
3199 let validated = validGossipEvents events t
3259 let validated = validGossipEvents events t
@@ -3202,7 +3262,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
3202 -- Warn if a peer sends both online and offline for the same key in one batch (suspicious)
3262 -- Warn if a peer sends both online and offline for the same key in one batch (suspicious)
3203 let conflictingKeys = do on <- onlineEntries; off <- trustedOffline; where on.pubkey == off.pubkey && on.server == off.server; yield on.pubkey
3263 let conflictingKeys = do on <- onlineEntries; off <- trustedOffline; where on.pubkey == off.pubkey && on.server == off.server; yield on.pubkey
3204 when (count conflictingKeys > 0)
3264 when (count conflictingKeys > 0)
3205− (println ("gossip: warning: " ++ fromServer ++ " sent conflicting online+offline for same key(s)"))
3265+ (logWarn ("gossip: " ++ fromServer ++ " sent conflicting online+offline for same key(s)"))
3206 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
3266 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
3207 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
3267 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
3208 let deduped = mergeByPubkeyServer [] filteredOnline
3268 let deduped = mergeByPubkeyServer [] filteredOnline
@@ -3218,10 +3278,11 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
3218 writeAllPresence gm.finalPresence
3278 writeAllPresence gm.finalPresence
3219 set *totalPresenceCount = count gm.finalPresence
3279 set *totalPresenceCount = count gm.finalPresence
3220 set *knownPeerServers = gm.gossipPeerServers
3280 set *knownPeerServers = gm.gossipPeerServers
3281+ set *presenceVersion = t
3221 yield {gm: gm, preFiltered: count deduped - count capacityDeduped}
3282 yield {gm: gm, preFiltered: count deduped - count capacityDeduped}
3222 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
3283 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
3223 when (not gossipResult.gm.withinCapacity || gossipResult.preFiltered > 0)
3284 when (not gossipResult.gm.withinCapacity || gossipResult.preFiltered > 0)
3224− (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer ++ " (pre-filtered: " ++ show gossipResult.preFiltered ++ ")"))
3285+ (logWarn ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer ++ " (pre-filtered: " ++ show gossipResult.preFiltered ++ ")"))
3225 -- Suppress re-broadcast of events we've already gossiped recently (prevents amplification)
3286 -- Suppress re-broadcast of events we've already gossiped recently (prevents amplification)
3226 freshEvents <- atomic do
3287 freshEvents <- atomic do
3227 recentGossip <- *recentGossip
3288 recentGossip <- *recentGossip
@@ -3258,7 +3319,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3258 t <- now
3319 t <- now
3259 withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
3320 withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
3260 when (fromServer != originServer)
3321 when (fromServer != originServer)
3261− (println ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
3322+ (logInfo ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
3262 -- Relay origin verification: for relay forwards, verify originServer is a known peer
3323 -- Relay origin verification: for relay forwards, verify originServer is a known peer
3263 -- to prevent ack routing (and metadata leakage) to arbitrary servers.
3324 -- to prevent ack routing (and metadata leakage) to arbitrary servers.
3264 originOk <- if fromServer == originServer then yield (True {})
3325 originOk <- if fromServer == originServer then yield (True {})
@@ -3267,7 +3328,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3267 yield (any knownPeers (\p -> p == originServer))
3328 yield (any knownPeers (\p -> p == originServer))
3268 if not originOk
3329 if not originOk
3270− println ("forward: rejected relay with unknown origin " ++ originServer ++ " from " ++ fromServer)
3331+ logWarn ("forward: rejected relay with unknown origin " ++ originServer ++ " from " ++ fromServer)
3271 yield (respond {ok: False {}, error: "unknown_origin"})
3332 yield (respond {ok: False {}, error: "unknown_origin"})
3273 -- Replay protection: reject forwards with timestamps older than messageMaxAge.
3334 -- Replay protection: reject forwards with timestamps older than messageMaxAge.
@@ -3283,7 +3344,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3283 bkt <- readPresenceBucket (pubkeyBucket toKey)
3344 bkt <- readPresenceBucket (pubkeyBucket toKey)
3284 yield (any bkt (\p -> p.pubkey == toKey && p.expiresAt > t))
3345 yield (any bkt (\p -> p.pubkey == toKey && p.expiresAt > t))
3285 if not hasPresence then do
3346 if not hasPresence then do
3286− println ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
3347+ logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
3287 yield (respond {ok: False {}, error: "no_presence"})
3348 yield (respond {ok: False {}, error: "no_presence"})
3289 let clampedTs = clampTimestamp timestamp t
3350 let clampedTs = clampTimestamp timestamp t
@@ -3293,7 +3354,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3293 ackedIds <- *ackedIds
3354 ackedIds <- *ackedIds
3294 deliveredIds <- *deliveredIds
3355 deliveredIds <- *deliveredIds
3295 forwardNonces <- *forwardNonces
3356 forwardNonces <- *forwardNonces
3296− let isDup = checkMessageDup bucket messageId toKey || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds || any forwardNonces (\n -> n.messageId == messageId && n.toKey == toKey)
3357+ let isDup = checkMessageDup bucket messageId toKey || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds || any forwardNonces (\n -> n.messageId == messageId && n.toKey == toKey && n.originServer == originServer)
3297 if isDup
3358 if isDup
3298 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}}
3359 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}}
@@ -3335,16 +3396,16 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3335 signalMessageForKey toKey seqTs
3396 signalMessageForKey toKey seqTs
3336 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3397 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3337 fwdNonces <- *forwardNonces
3398 fwdNonces <- *forwardNonces
3338− let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, seenAt: t}]
3399+ let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
3339 set *forwardNonces = if count updatedNonces > maxForwardNonces then evictOneOldest updatedNonces (\n -> n.seenAt) else updatedNonces
3400 set *forwardNonces = if count updatedNonces > maxForwardNonces then evictOneOldest updatedNonces (\n -> n.seenAt) else updatedNonces
3340 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
3401 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
3341 case result.forward of
3402 case result.forward of
3342 ForwardRejected {} ->
3403 ForwardRejected {} ->
3343 if result.globalCapReached
3404 if result.globalCapReached
3344− then println ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
3405+ then logWarn ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
3345 else if result.recvRateLimited
3406 else if result.recvRateLimited
3346− then println ("forward: recv rate limited (" ++ show maxRecvRatePerKey ++ "/s per key), rejecting for " ++ toKey)
3407+ then logWarn ("forward: recv rate limited (" ++ show maxRecvRatePerKey ++ "/s per key), rejecting for " ++ toKey)
3347− else println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
3408+ else logWarn ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
3348 _ -> yield {}
3409 _ -> yield {}
3349 yield (case result.forward of
3410 yield (case result.forward of
3350 ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}
3411 ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}
@@ -3390,6 +3451,7 @@ handleDrain = \authorization respond -> do
3390 let remainingPresence = do p <- allPresence; where not (isLocalServer p.server) || p.expiresAt <= t; yield p
3451 let remainingPresence = do p <- allPresence; where not (isLocalServer p.server) || p.expiresAt <= t; yield p
3391 writeAllPresence remainingPresence
3452 writeAllPresence remainingPresence
3392 set *totalPresenceCount = count remainingPresence
3453 set *totalPresenceCount = count remainingPresence
3454+ set *presenceVersion = t
3393 set *pollWakeup = t
3455 set *pollWakeup = t
3394 yield {localEntries: localEntries, peerServers: collectPeerServers allPresence (\p -> p.server)}
3456 yield {localEntries: localEntries, peerServers: collectPeerServers allPresence (\p -> p.server)}
3395 let drainedCount = count drainResult.localEntries
3457 let drainedCount = count drainResult.localEntries
@@ -3407,10 +3469,10 @@ handleDrain = \authorization respond -> do
3407 fork (do
3469 fork (do
3408 -- Best-effort flush of pending forwards and acks before stopping
3470 -- Best-effort flush of pending forwards and acks before stopping
3409 when (federationEnabled && pendingForwardCount > 0) (do
3471 when (federationEnabled && pendingForwardCount > 0) (do
3410− println ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards")
3472+ logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards")
3411 retryPendingForwards flushState.forwards)
3473 retryPendingForwards flushState.forwards)
3412 when (federationEnabled && pendingAckCount > 0) (do
3474 when (federationEnabled && pendingAckCount > 0) (do
3413− println ("drain: flushing " ++ show pendingAckCount ++ " pending acks")
3475+ logInfo ("drain: flushing " ++ show pendingAckCount ++ " pending acks")
3414 sendAcksAndClean flushState.pendingAcks)
3476 sendAcksAndClean flushState.pendingAcks)
3415 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
3477 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
3416 let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
3478 let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
@@ -3419,11 +3481,11 @@ handleDrain = \authorization respond -> do
3419 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))
3481 when federationEnabled (waitForInFlightDrain (t + federationFetchTimeoutMs * 2))
3420 t2 <- now
3482 t2 <- now
3421 atomic do set *drainCompletedAt = t2)
3483 atomic do set *drainCompletedAt = t2)
3422− println ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
3484+ logInfo ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
3423 when (pendingForwardCount > 0)
3485 when (pendingForwardCount > 0)
3424− (println ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))
3486+ (logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))
3425 when (pendingAckCount > 0)
3487 when (pendingAckCount > 0)
3426− (println ("drain: flushing " ++ show pendingAckCount ++ " pending acks"))
3488+ (logInfo ("drain: flushing " ++ show pendingAckCount ++ " pending acks"))
3427 yield (respond {ok: True {}, drained: drainedCount})
3489 yield (respond {ok: True {}, drained: drainedCount})
3429-- POST /maintenance/undrain: Resume accepting requests after a drain.
3491-- POST /maintenance/undrain: Resume accepting requests after a drain.
@@ -3437,8 +3499,8 @@ handleUndrain = \authorization respond -> do
3437 set *draining = False {}
3499 set *draining = False {}
3438 set *drainCompletedAt = 0
3500 set *drainCompletedAt = 0
3440− when wasDraining (println "undrain: server resumed, accepting new requests")
3502+ when wasDraining (logInfo "undrain: server resumed, accepting new requests")
3441− when (not wasDraining) (println "undrain: server was not draining")
3503+ when (not wasDraining) (logInfo "undrain: server was not draining")
3442 yield (respond {ok: True {}})
3504 yield (respond {ok: True {}})
3444-- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).
3506-- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).
@@ -3487,7 +3549,7 @@ handleHealth = \respond -> do
3487 capacityHealthy: result.capacityOk,
3549 capacityHealthy: result.capacityOk,
3488 federationHealthy: result.federationOk})
3550 federationHealthy: result.federationOk})
3490−zeroStats = {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pairRates: 0, federationPairRates: 0, pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, drainCompletedAt: 0, oldestForwardAge: 0, oldestAckAge: 0, ok: False {}}
3552+zeroStats = {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pairRates: 0, federationPairRates: 0, pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, hungFederationSlots: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, drainCompletedAt: 0, oldestForwardAge: 0, oldestAckAge: 0, ok: False {}}
3492handleStats = \authorization respond -> do
3554handleStats = \authorization respond -> do
3493 t <- now
3555 t <- now
@@ -3585,7 +3647,7 @@ handleStatsKey = \pubkey authorization respond -> do
3585 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})
3647 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})
3586 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})
3648 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})
3588−emptySnapshot = {ok: False {}, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: []}
3650+emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: []}
3590-- GET /maintenance/snapshot: Export federation metadata for persistence.
3652-- GET /maintenance/snapshot: Export federation metadata for persistence.
3591-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
3653-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
@@ -3605,18 +3667,23 @@ handleSnapshot = \authorization respond -> do
3605 f <- *forwards
3667 f <- *forwards
3606 yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f}
3668 yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f}
3607 atomic do set *lastSnapshotExportAt = t
3669 atomic do set *lastSnapshotExportAt = t
3608− println ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards")
3670+ logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards")
3609− yield (respond {ok: True {}, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards})
3671+ yield (respond {ok: True {}, version: snapshotVersion, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards})
3611-- POST /maintenance/restore: Import a previously-exported snapshot.
3673-- POST /maintenance/restore: Import a previously-exported snapshot.
3612-- Requires drain mode to prevent conflicts with live state. Merges imported data
3674-- Requires drain mode to prevent conflicts with live state. Merges imported data
3613-- with existing state (dedup entries are unioned, presence is merged by freshness).
3675-- with existing state (dedup entries are unioned, presence is merged by freshness).
3614-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
3676-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
3615-- trigger federation forward creation from messages + restored presence.
3677-- trigger federation forward creation from messages + restored presence.
3616−handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData authorization respond -> do
3678+handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData authorization respond -> do
3617 let authorized = verifyMaintenanceAuth authorization
3679 let authorized = verifyMaintenanceAuth authorization
3618 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
3680 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
3682+ -- Version 0 = pre-versioning snapshot (accept for backwards compatibility)
3683+ if version != 0 && version != snapshotVersion then do
3684+ logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
3685+ yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
3620 t <- now
3687 t <- now
3621 isDraining <- *draining
3688 isDraining <- *draining
3622 if not isDraining then yield (respond {ok: False {}, error: "server_must_be_draining", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
3689 if not isDraining then yield (respond {ok: False {}, error: "server_must_be_draining", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
@@ -3629,7 +3696,7 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw
3629 let validAcked = do a <- ackedIdsData; where not (isLocalServer a.originServer); yield a
3696 let validAcked = do a <- ackedIdsData; where not (isLocalServer a.originServer); yield a
3630 let validOverflow = do a <- ackOverflowData; where not (isLocalServer a.originServer); yield a
3697 let validOverflow = do a <- ackOverflowData; where not (isLocalServer a.originServer); yield a
3631 let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow)
3698 let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow)
3632− when (rejectedCount > 0) (println ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))
3699+ when (rejectedCount > 0) (logWarn ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))
3633 result <- atomic do
3700 result <- atomic do
3634 -- Merge presence (existing entries win on conflict via freshness)
3701 -- Merge presence (existing entries win on conflict via freshness)
3635 allPresence <- readAllPresence
3702 allPresence <- readAllPresence
@@ -3638,6 +3705,7 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw
3638 writeAllPresence cappedPresence
3705 writeAllPresence cappedPresence
3639 set *totalPresenceCount = count cappedPresence
3706 set *totalPresenceCount = count cappedPresence
3640 set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)
3707 set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)
3708+ set *presenceVersion = t
3641 -- Union dedup state, respecting capacity limits (using validated data)
3709 -- Union dedup state, respecting capacity limits (using validated data)
3642 ackedIds <- *ackedIds
3710 ackedIds <- *ackedIds
3643 let mergedAcked = union ackedIds validAcked
3711 let mergedAcked = union ackedIds validAcked
@@ -3661,7 +3729,7 @@ handleRestore = \presenceData ackedIdsData ackOverflowData deliveredIdsData forw
3661 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
3729 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
3662 set *forwards = cappedForwards
3730 set *forwards = cappedForwards
3663 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
3731 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
3664− println ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
3732+ logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
3665 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
3733 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
3667-- POST /messages/ack: Explicitly acknowledge received messages by ID.
3735-- POST /messages/ack: Explicitly acknowledge received messages by ID.
@@ -3677,6 +3745,16 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
3677 withSessionAuth authorization t clientIp
3745 withSessionAuth authorization t clientIp
3678 (yield (respond {ok: False {}, authorized: False {}, ackedCount: 0}))
3746 (yield (respond {ok: False {}, authorized: False {}, ackedCount: 0}))
3679 (\pubkey -> do
3747 (\pubkey -> do
3748+ ackRateLimited <- atomic do
3749+ lookupRates <- *lookupRates
3750+ let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == pubkey)
3751+ (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
3752+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3753+ maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
3754+ set *lookupRates = rateResult.entries
3755+ yield rateResult.limited
3756+ if ackRateLimited then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0})
3680 let dedupedIds = dedupMessageIds messageIds
3758 let dedupedIds = dedupMessageIds messageIds
3681 result <- atomic do
3759 result <- atomic do
3682 let b = pubkeyBucket pubkey
3760 let b = pubkeyBucket pubkey
@@ -3705,6 +3783,122 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
3705 retryEvictedAcks result.evictedEntries result.pendingAcks))
3783 retryEvictedAcks result.evictedEntries result.pendingAcks))
3706 yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount}))
3784 yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount}))
3786+-- POST /messages/batch: Send multiple messages in one request.
3787+-- Auth checked once; each message processed independently with its own rate limiting.
3788+-- Partial success is possible — check per-message status in results.
3789+handleSendBatch = \messages authorization clientIp respond -> do
3791+ globalRateLimited <- checkGlobalRate t
3792+ if globalRateLimited then yield (respond {results: [], authorized: False {}})
3794+ withSessionAuth authorization t clientIp
3795+ (yield (respond {results: [], authorized: False {}}))
3797+ let capped = takeRelation maxMessageIds messages
3798+ results <- sendBatchInner senderKey capped t []
3799+ -- Fork federation forwarding for all successful sends
3800+ let forwardTargets = do r <- results; where r.needsForward; yield r
3801+ fork (when (federationEnabled && count forwardTargets > 0)
3802+ (sendToEach (collectPeerServers forwardTargets (\r -> r.forwardServer)) (\server -> do
3803+ let serverMsgs = do r <- forwardTargets; where r.forwardServer == server; yield r
3804+ forEachBatch serverMsgs maxMessageIds (\batch -> do
3806+ forEachBatch batch 1 (\single -> case headOf single of
3807+ Nothing {} -> yield {}
3808+ Just {value: r} -> do
3809+ stillPending <- atomic do
3810+ forwards <- *forwards
3811+ yield (any forwards (\f -> f.messageId == r.messageId && f.toServer == server))
3812+ when stillPending (do
3813+ fwdAuth <- generateFederationAuth "forward"
3814+ fwdResult <- fetchFederation server (ForwardMessage {
3815+ toKey: r.toKey, encryptedBlob: r.blob,
3816+ timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
3817+ fromServer: serverHostN, requestTimestamp: fwdT,
3818+ authorization: fwdAuth})
3821+ logWarn ("federation: batch forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
3822+ recordFailure server
3824+ if response.ok then do
3825+ removeForward r.messageId server
3826+ clearFailure server
3827+ else logWarn ("federation: batch forward to " ++ server ++ " rejected for " ++ r.messageId ++ ": " ++ response.error)))))))
3828+ yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}}))
3830+sendBatchInner = \senderKey messages t acc ->
3831+ let split = splitAtN messages 1 in
3832+ case headOf split.batch of
3833+ Nothing {} -> yield acc
3834+ Just {value: msg} -> do
3835+ effectiveMsgId <- if msg.messageId != "" then yield msg.messageId else genToken
3836+ result <- if t - msg.timestamp > messageMaxAge
3837+ then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0}
3838+ else if senderKey == msg.to
3839+ then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0}
3841+ let clampedTs = clampTimestamp msg.timestamp t
3842+ sendResult <- atomic do
3843+ draining <- *draining
3844+ if draining then yield {status: MsDraining {}, fwdServer: "", needsFwd: False {}}
3846+ let msgBucket = pubkeyBucket msg.to
3847+ bucket <- readMessageBucket msgBucket
3848+ ackedIds <- *ackedIds
3849+ deliveredIds <- *deliveredIds
3850+ let isDup = checkMessageDup bucket effectiveMsgId msg.to || (effectiveMsgId != "" && (hasMessage effectiveMsgId msg.to ackedIds || hasMessage effectiveMsgId msg.to deliveredIds))
3851+ if isDup then yield {status: MsDuplicate {}, fwdServer: "", needsFwd: False {}}
3853+ sendRates <- *sendRates
3854+ let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
3855+ (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})
3856+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3857+ maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
3858+ if rateResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}}
3860+ set *sendRates = rateResult.entries
3861+ recvRates <- *recvRates
3862+ let recvResult = checkRateWindow recvRates (\r -> r.pubkey == msg.to)
3863+ (\ts -> {pubkey: msg.to, windowStart: ts, count: 1})
3864+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3865+ maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
3866+ if recvResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}}
3868+ set *recvRates = recvResult.entries
3869+ totalMsgCount <- *totalMessageCount
3870+ if totalMsgCount >= maxMessages then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}}
3872+ recipientCounts <- *recipientMsgCounts
3873+ let recipientCount = getRecipientCount msg.to recipientCounts
3874+ if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}}
3876+ recipientPresence <- readPresenceBucket (pubkeyBucket msg.to)
3877+ seqCounter <- *seqCounter
3878+ let seqTs = nextSeqTs t seqCounter
3879+ set *seqCounter = seqTs
3880+ writeMessageBucket msgBucket (union bucket
3881+ [{messageId: effectiveMsgId, toKey: msg.to,
3882+ encryptedBlob: msg.encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])
3883+ set *totalMessageCount = totalMsgCount + 1
3884+ set *recipientMsgCounts = adjustRecipientCount recipientCounts msg.to 1
3885+ signalMessageForKey msg.to seqTs
3886+ let toPresence = presenceForKey recipientPresence msg.to t
3887+ let localOnline = hasLocalPresence toPresence
3888+ let remoteServers = collectPeerServers toPresence (\p -> p.server)
3889+ let fwdServer = case headOf remoteServers of Nothing {} -> ""; Just {value: s} -> s
3890+ when (federationEnabled && count remoteServers > 0) (do
3891+ forwards <- *forwards
3892+ let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: msg.to, encryptedBlob: msg.encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}
3893+ let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
3894+ set *forwards = union forwards cappedFwds)
3895+ let delivery = if localOnline then MsDelivered {}
3896+ else if federationEnabled && count remoteServers > 0 then MsFederated {}
3898+ yield {status: delivery, fwdServer: fwdServer, needsFwd: federationEnabled && count remoteServers > 0}
3899+ yield {messageId: effectiveMsgId, status: sendResult.status, needsForward: sendResult.needsFwd, forwardServer: sendResult.fwdServer, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs}
3900+ sendBatchInner senderKey split.rest t (union acc [result])
3708serve = \req -> case req of
3902serve = \req -> case req of
3709 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
3903 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
3710 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
3904 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
@@ -3714,6 +3908,7 @@ serve = \req -> case req of
3714 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond
3908 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond
3715 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
3909 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
3716 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
3910 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
3911+ SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
3717 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
3912 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
3718 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
3913 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
3719 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond
3914 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond
@@ -3723,7 +3918,7 @@ serve = \req -> case req of
3723 Drain {authorization, respond} -> handleDrain authorization respond
3918 Drain {authorization, respond} -> handleDrain authorization respond
3724 Undrain {authorization, respond} -> handleUndrain authorization respond
3919 Undrain {authorization, respond} -> handleUndrain authorization respond
3725 Snapshot {authorization, respond} -> handleSnapshot authorization respond
3920 Snapshot {authorization, respond} -> handleSnapshot authorization respond
3726− Restore {presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, authorization, respond} -> handleRestore presence ackedIds ackOverflow deliveredIds forwardFailures forwards authorization respond
3921+ Restore {version, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, authorization, respond} -> handleRestore version presence ackedIds ackOverflow deliveredIds forwardFailures forwards authorization respond
3727 Health {respond} -> handleHealth respond
3922 Health {respond} -> handleHealth respond
3728 Stats {authorization, respond} -> handleStats authorization respond
3923 Stats {authorization, respond} -> handleStats authorization respond
3729 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
3924 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
@@ -3744,11 +3939,18 @@ backgroundPrune = do
3744 backgroundPrune
3939 backgroundPrune
3746 t <- now
3941 t <- now
3942+ -- Detect backward clock jump: if seqCounter is far ahead of now, reset to prevent
3943+ -- all messages getting unreachable receivedAt values for an extended period
3946+ when (sc > t + clockSkewTolerance) (do
3947+ logWarn ("clock jump detected — seqCounter " ++ show (sc - t) ++ "ms ahead of now, resetting")
3948+ set *seqCounter = t)
3747 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead
3949 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead
3748 sa <- *startedAt
3950 sa <- *startedAt
3749 lastHb <- *lastHeartbeatAt
3951 lastHb <- *lastHeartbeatAt
3750 when (t - sa > healthHeartbeatMaxAge && t - lastHb > healthHeartbeatMaxAge) (do
3952 when (t - sa > healthHeartbeatMaxAge && t - lastHb > healthHeartbeatMaxAge) (do
3751− println "backgroundPrune: heartbeat appears dead, bumping pollWakeup"
3953+ logWarn "backgroundPrune: heartbeat appears dead, bumping pollWakeup"
3752 atomic do
3954 atomic do
3753 set *pollWakeup = t
3955 set *pollWakeup = t
3754 set *lastHeartbeatAt = t)
3956 set *lastHeartbeatAt = t)
@@ -3765,7 +3967,7 @@ backgroundPrune = do
3765 set *pruneCount = pc + 1
3967 set *pruneCount = pc + 1
3766 yield {canRun: True {}, recovered: running > 0, gen: newGen, stuckSince: running}
3968 yield {canRun: True {}, recovered: running > 0, gen: newGen, stuckSince: running}
3767 when status.recovered (do
3969 when status.recovered (do
3768− println ("backgroundPrune: previous cycle stuck for " ++ show (t - status.stuckSince) ++ "ms, forcing recovery")
3970+ logWarn ("backgroundPrune: previous cycle stuck for " ++ show (t - status.stuckSince) ++ "ms, forcing recovery")
3769 -- Reset fork counter: the stuck fork leaked its slot, so reclaim it to prevent permanent lockout
3971 -- Reset fork counter: the stuck fork leaked its slot, so reclaim it to prevent permanent lockout
3770 atomic do set *activePruneForks = 0)
3972 atomic do set *activePruneForks = 0)
3771 if status.canRun
3973 if status.canRun
@@ -3773,7 +3975,7 @@ backgroundPrune = do
3773 -- Guard against unbounded concurrent prune forks from repeated stuck recovery
3975 -- Guard against unbounded concurrent prune forks from repeated stuck recovery
3774 forks <- *activePruneForks
3976 forks <- *activePruneForks
3775 if forks >= 2
3977 if forks >= 2
3776− then println "backgroundPrune: skipping (max concurrent prune forks reached)"
3978+ then logWarn "backgroundPrune: skipping (max concurrent prune forks reached)"
3778 atomic do
3980 atomic do
3779 af <- *activePruneForks
3981 af <- *activePruneForks
@@ -3787,7 +3989,7 @@ backgroundPrune = do
3787 when (currentGen == status.gen) (set *pruneRunning = 0)
3989 when (currentGen == status.gen) (set *pruneRunning = 0)
3788 af <- *activePruneForks
3990 af <- *activePruneForks
3789 set *activePruneForks = max 0 (af - 1))
3991 set *activePruneForks = max 0 (af - 1))
3790− else println "backgroundPrune: skipping (previous cycle still running)"
3992+ else logWarn "backgroundPrune: skipping (previous cycle still running)"
3791 backgroundPrune
3993 backgroundPrune
3793-- Background loop: periodically bumps pollWakeup to unblock long-polls that have been
3995-- Background loop: periodically bumps pollWakeup to unblock long-polls that have been
@@ -3827,14 +4029,16 @@ backgroundSnapshot = do
3827 d <- *deliveredIds
4029 d <- *deliveredIds
3828 ff <- *forwardFailures
4030 ff <- *forwardFailures
3829 yield {messages: tmc, sessions: sc, presence: pc, forwards: count f, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff}
4031 yield {messages: tmc, sessions: sc, presence: pc, forwards: count f, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff}
3830− 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)
4032+ logInfo ("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)
3831 -- Warn if federation state hasn't been exported recently (will be lost on restart)
4033 -- Warn if federation state hasn't been exported recently (will be lost on restart)
3832 t <- now
4034 t <- now
3833 lastExport <- *lastSnapshotExportAt
4035 lastExport <- *lastSnapshotExportAt
3834 when (maintenanceSecret != "" && snapshotStalenessWarning > 0
4036 when (maintenanceSecret != "" && snapshotStalenessWarning > 0
3835 && (lastExport == 0 || t - lastExport > snapshotStalenessWarning)
4037 && (lastExport == 0 || t - lastExport > snapshotStalenessWarning)
3836 && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))
4038 && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))
3837− (println ("WARNING: no snapshot exported in " ++ show (if lastExport == 0 then 0 else (t - lastExport) / 1000) ++ "s — federation state will be lost on restart (presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " deliveredIds=" ++ show result.deliveredIds ++ ")"))
4040+ logError ("no snapshot exported in " ++ show (if lastExport == 0 then 0 else (t - lastExport) / 1000) ++ "s — federation state will be lost on restart (presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " deliveredIds=" ++ show result.deliveredIds ++ ")")
4041+ logError "ACTION REQUIRED: run GET /maintenance/snapshot to persist federation state before restart")
3838 else yield {}
4042 else yield {}
3839 backgroundSnapshot
4043 backgroundSnapshot
@@ -3874,40 +4078,40 @@ main = do
3874 && maxGlobalFederationConcurrency >= maxFederationConcurrency
4078 && maxGlobalFederationConcurrency >= maxFederationConcurrency
3875 && maxGossipEventsPerServer > maxGossipEvents
4079 && maxGossipEventsPerServer > maxGossipEvents
3876 if not configOk
4080 if not configOk
3877− then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
4081+ then logError "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
3878 else if not configRelationsOk
4082 else if not configRelationsOk
3879− then println "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, per-key limits <= global limits), server not started"
4083+ then logError "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, per-key limits <= global limits), server not started"
3881 t <- now
4085 t <- now
3882 atomic do
4086 atomic do
3883 set *startedAt = t
4087 set *startedAt = t
3884 set *serverGeneration = t
4088 set *serverGeneration = t
3885 when (maintenanceSecret == "")
4089 when (maintenanceSecret == "")
3886− (println "WARNING: maintenanceSecret is empty — /maintenance/* and /stats endpoints are disabled")
4090+ (logWarn "maintenanceSecret is empty — /maintenance/* and /stats endpoints are disabled")
3887 when (matchesHostnameOrSubdomain serverHostN "localhost")
4091 when (matchesHostnameOrSubdomain serverHostN "localhost")
3888− (println ("WARNING: serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
4092+ (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
3889− println "NOTE: IP binding is active — sessions created with a ClientIp header will reject requests without one"
4093+ logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
3890− println "Skrepka relay server starting on port 8080..."
4094+ logInfo "Skrepka relay server starting on port 8080..."
3891− println (" federation: " ++ if federationEnabled then "enabled" else "disabled")
4095+ logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")
3892− println (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
4096+ logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
3893− println (" poll timeout: " ++ show pollTimeoutMs ++ "ms")
4097+ logInfo (" poll timeout: " ++ show pollTimeoutMs ++ "ms")
3894− println (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")
4098+ logInfo (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")
3895− println (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")
4099+ logInfo (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")
3896− println (" max presence: " ++ show maxPresenceEntries)
4100+ logInfo (" max presence: " ++ show maxPresenceEntries)
3897− println (" max forwards: " ++ show maxForwardsEntries)
4101+ logInfo (" max forwards: " ++ show maxForwardsEntries)
3898− println (" session expiry: " ++ show sessionExpiry ++ "ms")
4102+ logInfo (" session expiry: " ++ show sessionExpiry ++ "ms")
3899− println (" message max age: " ++ show messageMaxAge ++ "ms")
4103+ logInfo (" message max age: " ++ show messageMaxAge ++ "ms")
3900− println (" global rate limit: " ++ show maxGlobalRequestRate ++ "/" ++ show globalRateWindowMs ++ "ms")
4104+ logInfo (" global rate limit: " ++ show maxGlobalRequestRate ++ "/" ++ show globalRateWindowMs ++ "ms")
3901− println (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
4105+ logInfo (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
3902− println (" pair rate limit: " ++ show maxPairRatePerPair ++ "/" ++ show pairRateWindowMs ++ "ms per sender-recipient")
4106+ logInfo (" pair rate limit: " ++ show maxPairRatePerPair ++ "/" ++ show pairRateWindowMs ++ "ms per sender-recipient")
3903− println (" federation pair rate limit: " ++ show maxFederationPairRatePerPair ++ "/" ++ show federationPairRateWindowMs ++ "ms per server-recipient")
4107+ logInfo (" federation pair rate limit: " ++ show maxFederationPairRatePerPair ++ "/" ++ show federationPairRateWindowMs ++ "ms per server-recipient")
3904− println (" gossip event rate limit: " ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms per server")
4108+ logInfo (" gossip event rate limit: " ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms per server")
3905− println (" global federation concurrency: " ++ show maxGlobalFederationConcurrency)
4109+ logInfo (" global federation concurrency: " ++ show maxGlobalFederationConcurrency)
3906− println (" ack retry limit: " ++ show maxAckRetries ++ " per entry")
4110+ logInfo (" ack retry limit: " ++ show maxAckRetries ++ " per entry")
3907 when (federationCircuitBreakerThreshold > 0)
4111 when (federationCircuitBreakerThreshold > 0)
3908− (println (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
4112+ (logInfo (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
3909 when (autoSnapshotInterval > 0)
4113 when (autoSnapshotInterval > 0)
3910− (println (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
4114+ (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
3911 fork backgroundPrune
4115 fork backgroundPrune
3912 fork pollHeartbeat
4116 fork pollHeartbeat
3913 when (autoSnapshotInterval > 0) (fork backgroundSnapshot)
4117 when (autoSnapshotInterval > 0) (fork backgroundSnapshot)