@@ -136,7 +136,9 @@ maxRecentFederationRequests = 1000
136maxForwardNonces = 100000
136maxForwardNonces = 100000
137forwardNonceRetention : Int<Ms>
137forwardNonceRetention : Int<Ms>
138forwardNonceRetention = messageMaxAge * 2 -- 60 days; extends 30 days beyond message dedup
138forwardNonceRetention = messageMaxAge * 2 -- 60 days; extends 30 days beyond message dedup
139−-- (reconciliationFrequency removed: message counters now reconcile every prune cycle)
139+-- Message reconciliation: full scan runs every Nth prune cycle or when cheap drift check detects mismatch.
140+-- Cheap check (sum rcBucket totals vs totalMessageCount) runs every cycle; full scan only when needed.
141+reconciliationFullCycleInterval = 10 -- full reconciliation every 10th prune cycle
141-- Federation latency metric window
143-- Federation latency metric window
142federationLatencyWindowMs : Int<Ms>
144federationLatencyWindowMs : Int<Ms>
@@ -163,6 +165,11 @@ backgroundPruneInterval : Int<Ms>
163backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
165backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
164autoSnapshotInterval : Int<Ms>
166autoSnapshotInterval : Int<Ms>
165autoSnapshotInterval = 300000 -- 5 minutes; periodic state summary to stdout (0 to disable)
167autoSnapshotInterval = 300000 -- 5 minutes; periodic state summary to stdout (0 to disable)
168+-- Full snapshot auto-export interval: emit [SNAPSHOT_DATA] JSON to stdout for sidecar capture.
169+-- 0 to disable. When enabled, a crash loses at most this interval of federation state.
170+-- Only emits when there's meaningful state to persist (presence/forwards/ackedIds > 0).
171+autoSnapshotExportInterval : Int<Ms>
172+autoSnapshotExportInterval = 3600000 -- 1 hour; full state export to stdout (0 to disable)
166snapshotStalenessWarning : Int<Ms>
173snapshotStalenessWarning : Int<Ms>
167snapshotStalenessWarning = 3600000 -- 1 hour; warn if no snapshot exported via /maintenance/snapshot
174snapshotStalenessWarning = 3600000 -- 1 hour; warn if no snapshot exported via /maintenance/snapshot
168pollTimeoutMs : Int<Ms>
175pollTimeoutMs : Int<Ms>
@@ -577,6 +584,9 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
577-- Tracks when /maintenance/snapshot was last exported (for staleness warnings)
584-- Tracks when /maintenance/snapshot was last exported (for staleness warnings)
578*lastSnapshotExportAt : Timestamp
585*lastSnapshotExportAt : Timestamp
587+-- Tracks when the last auto-export [SNAPSHOT_DATA] was emitted to stdout
588+*lastAutoSnapshotExportAt : Timestamp
580-- Per-peer gossip event volume rate limiting
590-- Per-peer gossip event volume rate limiting
581*gossipEventRates : [GossipEventRateEntry]
591*gossipEventRates : [GossipEventRateEntry]
@@ -594,6 +604,15 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
594-- (key-rotation abuse bypassing per-key rate limits). Pruned with other rate limiters.
604-- (key-rotation abuse bypassing per-key rate limits). Pruned with other rate limiters.
595*uniqueKeyIpRates : [UniqueKeyIpEntry]
605*uniqueKeyIpRates : [UniqueKeyIpEntry]
607+-- Runtime-configurable rate limit overrides (0 = use compile-time default).
608+-- Adjusted via /maintenance/rate-limits POST endpoint without recompile+restart.
609+*rateOverrideGlobalRequest : Int
610+*rateOverrideIpRate : Int
611+*rateOverrideSendRate : Int
612+*rateOverrideFederationRate : Int
613+*rateOverrideAuthGlobal : Int
614+*rateOverrideRecvRate : Int
597-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
616-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
598-- Polls compare against this to skip expensive presence lookups when nothing changed.
617-- Polls compare against this to skip expensive presence lookups when nothing changed.
599*presenceVersion : Timestamp
618*presenceVersion : Timestamp
@@ -601,6 +620,16 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
601-- Last time the background watchdog loop ran (for health check liveness detection)
620-- Last time the background watchdog loop ran (for health check liveness detection)
602*lastWatchdogAt : Timestamp
621*lastWatchdogAt : Timestamp
623+-- Cached health result: returned when /health is rate-limited instead of false all-healthy.
624+-- Updated on every successful (non-rate-limited) health check.
625+*cachedHealthOk : Bool
626+*cachedHealthPrune : Bool
627+*cachedHealthHeartbeat : Bool
628+*cachedHealthCapacity : Bool
629+*cachedHealthFederation : Bool
630+*cachedHealthSnapshot : Bool
631+*cachedHealthWatchdog : Bool
604-- ============================================
633-- ============================================
606-- ============================================
635-- ============================================
@@ -662,18 +691,20 @@ route ClientApi where
662 = LookupPresence
691 = LookupPresence
664-- Server-to-server federation API (authenticated via domain callback verification)
693-- Server-to-server federation API (authenticated via domain callback verification)
694+-- correlationId: optional trace ID for cross-server request tracing (empty = not provided).
695+-- Peers that don't support it send "" — logged when present but not enforced.
665route FederationApi where
696route FederationApi where
666 /federation
697 /federation
667 -- Receive presence gossip events from a peer server
698 -- Receive presence gossip events from a peer server
668− POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp} /gossip headers {authorization: Text}
699+ POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /gossip headers {authorization: Text}
669 -> {ok: Bool, error: Text}
700 -> {ok: Bool, error: Text}
670 = RecvGossip
701 = RecvGossip
671 -- Receive a forwarded message from a peer server on behalf of a sender
702 -- Receive a forwarded message from a peer server on behalf of a sender
672− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp} /forward headers {authorization: Text}
703+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /forward headers {authorization: Text}
673 -> {ok: Bool, error: Text}
704 -> {ok: Bool, error: Text}
674 = ForwardMessage
705 = ForwardMessage
675 -- Acknowledge delivery of messages back to the origin server
706 -- Acknowledge delivery of messages back to the origin server
676− POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp} /ack headers {authorization: Text}
707+ POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /ack headers {authorization: Text}
677 -> {ok: Bool, error: Text, ackedCount: Int}
708 -> {ok: Bool, error: Text, ackedCount: Int}
678 = FederationAck
709 = FederationAck
@@ -728,6 +759,14 @@ route MaintenanceApi where
728 POST {pubkey: PubkeyHex} /maintenance/sessions/revoke-all headers {authorization: Text}
759 POST {pubkey: PubkeyHex} /maintenance/sessions/revoke-all headers {authorization: Text}
729 -> {ok: Bool, revokedCount: Int}
760 -> {ok: Bool, revokedCount: Int}
730 = AdminRevokeSessions
761 = AdminRevokeSessions
762+ -- Read current effective rate limits (compile-time defaults + runtime overrides)
763+ GET /maintenance/rate-limits headers {authorization: Text}
764+ -> {ok: Bool, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int}
766+ -- Update rate limits at runtime (0 = reset to compile-time default)
767+ POST {globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int} /maintenance/rate-limits headers {authorization: Text}
732route HealthApi where
771route HealthApi where
733 GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool} = Health
772 GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool} = Health
@@ -1343,6 +1382,18 @@ readSelectedPresenceBuckets = \buckets ->
1343 remaining <- readSelectedPresenceBuckets rest
1382 remaining <- readSelectedPresenceBuckets rest
1344 yield (union entries remaining)
1383 yield (union entries remaining)
1385+-- Write presence entries to only the specified bucket indices. Entries are filtered
1386+-- by pubkeyBucket so each bucket gets only its entries. Unspecified buckets are untouched.
1387+writeSelectedPresenceBuckets : [Int] -> [PresenceEntry] -> IO ()
1388+writeSelectedPresenceBuckets = \buckets entries ->
1389+ case headOf buckets of
1390+ Nothing {} -> yield {}
1391+ Just {value: b} -> do
1392+ let rest = do x <- buckets; where x != b; yield x
1393+ let bucketEntries = do e <- entries; where pubkeyBucket e.pubkey == b; yield e
1394+ writePresenceBucket b bucketEntries
1395+ writeSelectedPresenceBuckets rest entries
1346-- Single-pass partitioning: each entry's bucket is computed once (vs 16x in the list-comprehension approach)
1397-- Single-pass partitioning: each entry's bucket is computed once (vs 16x in the list-comprehension approach)
1347writeAllPresence : [PresenceEntry] -> IO ()
1398writeAllPresence : [PresenceEntry] -> IO ()
1348writeAllPresence = \entries -> do
1399writeAllPresence = \entries -> do
@@ -1655,6 +1706,16 @@ countAllRcEntries = \b ->
1655 rest <- countAllRcEntries (b + 1)
1706 rest <- countAllRcEntries (b + 1)
1656 yield (count bucket + rest)
1707 yield (count bucket + rest)
1709+-- Sum total message count from rcBuckets (cheap: sums .count fields, doesn't scan messages)
1710+sumRcBucketTotals : Int -> IO Int
1711+sumRcBucketTotals = \b ->
1712+ if b > 15 then yield 0
1714+ bucket <- readRcBucket b
1715+ let bucketTotal = fold (\acc c -> acc + c.count) 0 bucket
1716+ rest <- sumRcBucketTotals (b + 1)
1717+ yield (bucketTotal + rest)
1658-- Check message dedup only (without counting — count comes from rcBuckets)
1719-- Check message dedup only (without counting — count comes from rcBuckets)
1659checkMessageDup : [Message] -> Text -> PubkeyHex -> Bool
1720checkMessageDup : [Message] -> Text -> PubkeyHex -> Bool
1660checkMessageDup = \messages messageId toKey ->
1721checkMessageDup = \messages messageId toKey ->
@@ -1867,6 +1928,21 @@ evictSmallest = \items n getKey ->
1867 let rest = evictSmallest r.items (n - 1) getKey in
1928 let rest = evictSmallest r.items (n - 1) getKey in
1868 {kept: rest.kept, evicted: union [c] rest.evicted}
1929 {kept: rest.kept, evicted: union [c] rest.evicted}
1931+-- Single-pass eviction by timestamp cutoff: keeps items where getKey >= cutoff.
1932+-- O(n) vs O(n*k) for evictSmallest. Used for forward nonce eviction on the hot path.
1933+evictByCutoff = \items cutoff getKey ->
1934+ do x <- items; where getKey x >= cutoff; yield x
1936+-- Evict nonces with fallback: first try expired-nonce eviction (O(n)), then if still
1937+-- over capacity, evict oldest 1% via evictSmallest (small k keeps O(n*k) bounded).
1938+-- Minimizes replay window: expired nonces are removed first (no window), and the
1939+-- 1% fallback creates a window of only ~1000 entries vs ~10000 with the old 10%.
1940+evictNonces = \items maxCount retention t ->
1941+ let afterExpired = evictByCutoff items (t - retention) (\n -> n.seenAt) in
1942+ if count afterExpired <= maxCount then afterExpired
1943+ else let evictCount = max 1 (count afterExpired / 100) in
1944+ (evictSmallest afterExpired evictCount (\n -> n.seenAt)).kept
1870-- Enforce per-key capacity: if a key has >= maxPerKey items, evict the oldest for that key.
1946-- Enforce per-key capacity: if a key has >= maxPerKey items, evict the oldest for that key.
1871-- Also prunes expired items.
1947-- Also prunes expired items.
1872capPerKey = \items matchKey maxPerKey getExpiry t ->
1948capPerKey = \items matchKey maxPerKey getExpiry t ->
@@ -2147,10 +2223,11 @@ sendBatchedGossipToServer = \server allEvents hops ->
2147 forEachBatch allEvents maxGossipEvents (\batch -> do
2223 forEachBatch allEvents maxGossipEvents (\batch -> do
2148 t <- now
2224 t <- now
2149 auth <- generateFederationAuth "gossip"
2225 auth <- generateFederationAuth "gossip"
2226+ corrId <- genTokenN 1
2150 result <- fetchFederation server (RecvGossip {
2227 result <- fetchFederation server (RecvGossip {
2151 authorization: auth,
2228 authorization: auth,
2152 events: batch, hopsRemaining: hops, fromServer: serverHostN,
2229 events: batch, hopsRemaining: hops, fromServer: serverHostN,
2153− requestTimestamp: t})
2230+ requestTimestamp: t, correlationId: corrId})
2154 case result of
2231 case result of
2155 Err e -> logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
2232 Err e -> logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
2156 Ok _ -> yield {})
2233 Ok _ -> yield {})
@@ -2190,21 +2267,48 @@ mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
2190 withinCapacity: capResult.withinCapacity}
2267 withinCapacity: capResult.withinCapacity}
2270+-- Effective rate limit readers: return runtime override if set (> 0), else compile-time default
2271+effectiveGlobalRequestRate = do
2272+ o <- *rateOverrideGlobalRequest
2273+ yield (if o > 0 then o else maxGlobalRequestRate)
2275+effectiveIpRate = do
2276+ o <- *rateOverrideIpRate
2277+ yield (if o > 0 then o else maxIpRatePerIp)
2279+effectiveSendRate = do
2280+ o <- *rateOverrideSendRate
2281+ yield (if o > 0 then o else maxSendRatePerKey)
2283+effectiveFederationRate = do
2284+ o <- *rateOverrideFederationRate
2285+ yield (if o > 0 then o else maxFederationRatePerServer)
2287+effectiveAuthGlobalRate = do
2288+ o <- *rateOverrideAuthGlobal
2289+ yield (if o > 0 then o else maxAuthGlobalRate)
2291+effectiveRecvRate = do
2292+ o <- *rateOverrideRecvRate
2293+ yield (if o > 0 then o else maxRecvRatePerKey)
2193checkFederationRate = \server t -> atomic do
2295checkFederationRate = \server t -> atomic do
2194 federationRates <- *federationRates
2296 federationRates <- *federationRates
2297+ fedRate <- effectiveFederationRate
2195− let result = checkServerRate federationRates server maxFederationRatePerServer federationRateWindowMs maxFederationRateEntries t
2298+ let result = checkServerRate federationRates server fedRate federationRateWindowMs maxFederationRateEntries t
2196 set *federationRates = result.entries
2299 set *federationRates = result.entries
2197 yield result.limited
2300 yield result.limited
2199checkGlobalRate = \t -> atomic do
2302checkGlobalRate = \t -> atomic do
2200 windowStart <- *globalRateWindowStart
2303 windowStart <- *globalRateWindowStart
2201 rateCount <- *globalRateCount
2304 rateCount <- *globalRateCount
2305+ effectiveRate <- effectiveGlobalRequestRate
2202 if t - windowStart >= globalRateWindowMs
2306 if t - windowStart >= globalRateWindowMs
2204 set *globalRateWindowStart = t
2308 set *globalRateWindowStart = t
2205 set *globalRateCount = 1
2309 set *globalRateCount = 1
2206 yield (False {})
2310 yield (False {})
2207− else if rateCount >= maxGlobalRequestRate
2311+ else if rateCount >= effectiveRate
2208 then yield (True {})
2312 then yield (True {})
2210 set *globalRateCount = rateCount + 1
2314 set *globalRateCount = rateCount + 1
@@ -2214,12 +2318,13 @@ checkGlobalRate = \t -> atomic do
2214checkAuthGlobalRate = \t -> atomic do
2318checkAuthGlobalRate = \t -> atomic do
2215 windowStart <- *authGlobalRateWindowStart
2319 windowStart <- *authGlobalRateWindowStart
2216 rateCount <- *authGlobalRateCount
2320 rateCount <- *authGlobalRateCount
2321+ authRate <- effectiveAuthGlobalRate
2217 if t - windowStart >= authGlobalRateWindowMs
2322 if t - windowStart >= authGlobalRateWindowMs
2219 set *authGlobalRateWindowStart = t
2324 set *authGlobalRateWindowStart = t
2220 set *authGlobalRateCount = 1
2325 set *authGlobalRateCount = 1
2221 yield (False {})
2326 yield (False {})
2222− else if rateCount >= maxAuthGlobalRate
2327+ else if rateCount >= authRate
2223 then yield (True {})
2328 then yield (True {})
2225 set *authGlobalRateCount = rateCount + 1
2330 set *authGlobalRateCount = rateCount + 1
@@ -2245,7 +2350,8 @@ checkClientIpRate = \ip t ->
2245 if ip == "" then yield (False {})
2350 if ip == "" then yield (False {})
2246 else atomic do
2351 else atomic do
2247 ipRates <- *ipRates
2352 ipRates <- *ipRates
2353+ ipRate <- effectiveIpRate
2248− let result = checkIpRate ipRates ip maxIpRatePerIp ipRateWindowMs maxIpRateEntries t
2354+ let result = checkIpRate ipRates ip ipRate ipRateWindowMs maxIpRateEntries t
2249 set *ipRates = result.entries
2355 set *ipRates = result.entries
2250 yield result.limited
2356 yield result.limited
@@ -2426,11 +2532,12 @@ retryForwardsForServer = \entries remaining ->
2426 let rest = do x <- entries; where x.messageId != f.messageId; yield x
2532 let rest = do x <- entries; where x.messageId != f.messageId; yield x
2427 t <- now
2533 t <- now
2428 auth <- generateFederationAuth "forward"
2534 auth <- generateFederationAuth "forward"
2535+ corrId <- genTokenN 1
2429 result <- fetchFederation f.toServer (ForwardMessage {
2536 result <- fetchFederation f.toServer (ForwardMessage {
2430 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
2537 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
2431 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
2538 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
2432 fromServer: serverHostN, requestTimestamp: t,
2539 fromServer: serverHostN, requestTimestamp: t,
2433− authorization: auth})
2540+ authorization: auth, correlationId: corrId})
2434 case result of
2541 case result of
2435 Err e -> do
2542 Err e -> do
2436 logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
2543 logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
@@ -2493,11 +2600,12 @@ forwardQueuedMessages = \targets t -> do
2493fetchFederationAck = \origin messageIds -> do
2600fetchFederationAck = \origin messageIds -> do
2494 t <- now
2601 t <- now
2495 auth <- generateFederationAuth "ack"
2602 auth <- generateFederationAuth "ack"
2603+ corrId <- genTokenN 1
2496 fetchFederation origin (FederationAck {
2604 fetchFederation origin (FederationAck {
2497 authorization: auth,
2605 authorization: auth,
2498 messageIds: messageIds,
2606 messageIds: messageIds,
2499 originServer: origin, fromServer: serverHostN,
2607 originServer: origin, fromServer: serverHostN,
2500− requestTimestamp: t})
2608+ requestTimestamp: t, correlationId: corrId})
2502sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
2610sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
2503sendBatchedAcksToServer = \serverAcks server ->
2611sendBatchedAcksToServer = \serverAcks server ->
@@ -2679,10 +2787,20 @@ pruneAndRetry = \t -> do
2679 yield (actual - stored)
2787 yield (actual - stored)
2680 when (presDrift != 0 || sessDrift != 0)
2788 when (presDrift != 0 || sessDrift != 0)
2681 (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))
2789 (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))
2682− -- Message reconciliation: per-bucket reads reduce lock contention vs reading all 16 at once
2790+ -- Message reconciliation: cheap check first (sum rcBucket .count fields vs totalMessageCount).
2791+ -- Full scan only when drift detected or every Nth cycle (reconciliationFullCycleInterval).
2793+ let isFullCycle = pc > 0 && (pc - (pc / reconciliationFullCycleInterval) * reconciliationFullCycleInterval) == 0
2794+ cheapMsgDrift <- atomic do
2795+ rcTotal <- sumRcBucketTotals 0
2796+ stored <- *totalMessageCount
2797+ yield (rcTotal - stored)
2798+ if cheapMsgDrift != 0 || isFullCycle
2683− msgDrift <- reconcileMessageCounts 0 0
2800+ msgDrift <- reconcileMessageCounts 0 0
2684− when (msgDrift != 0)
2801+ when (msgDrift != 0)
2685− (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))
2802+ (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))
2686 pendingForwards <- atomic do f <- *forwards; yield f
2804 pendingForwards <- atomic do f <- *forwards; yield f
2687 ackState <- atomic do
2805 ackState <- atomic do
2688 ackedIds2 <- *ackedIds
2806 ackedIds2 <- *ackedIds
@@ -3270,7 +3388,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
3270 if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}
3388 if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}
3272 sendRates <- *sendRates
3390 sendRates <- *sendRates
3391+ sendRate <- effectiveSendRate
3273− let rateResult = checkPubkeyRate sendRates senderKey maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
3392+ let rateResult = checkPubkeyRate sendRates senderKey sendRate sendRateWindowMs maxSendRateEntries t
3274 if rateResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: rateResult.retryAfterMs}
3393 if rateResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: rateResult.retryAfterMs}
3276 set *sendRates = rateResult.entries
3395 set *sendRates = rateResult.entries
@@ -3280,7 +3399,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
3280 if pairResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: pairResult.retryAfterMs}
3399 if pairResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: pairResult.retryAfterMs}
3282 recvRates <- *recvRates
3401 recvRates <- *recvRates
3402+ recvRate <- effectiveRecvRate
3283− let recvResult = checkPubkeyRate recvRates to maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
3403+ let recvResult = checkPubkeyRate recvRates to recvRate recvRateWindowMs maxRecvRateEntries t
3284 if recvResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: recvResult.retryAfterMs}
3404 if recvResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: recvResult.retryAfterMs}
3286 set *recvRates = recvResult.entries
3406 set *recvRates = recvResult.entries
@@ -3307,20 +3427,23 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
3307 -- would miss locally-originated messages entirely.
3427 -- would miss locally-originated messages entirely.
3308 fwdNonces <- *forwardNonces
3428 fwdNonces <- *forwardNonces
3309 let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, originServer: "", seenAt: t}]
3429 let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, originServer: "", seenAt: t}]
3430+ -- O(n) cutoff eviction: remove expired nonces first, avoiding O(n*k) evictSmallest on hot path
3310 set *forwardNonces = if count updatedNonces > maxForwardNonces
3431 set *forwardNonces = if count updatedNonces > maxForwardNonces
3311− then let evictCount = max 1 (count updatedNonces / 10) in
3432+ then evictNonces updatedNonces maxForwardNonces forwardNonceRetention t
3312− (evictSmallest updatedNonces evictCount (\n -> n.seenAt)).kept
3313 else updatedNonces
3433 else updatedNonces
3314 let toPresence = presenceForKey recipientPresence to t
3434 let toPresence = presenceForKey recipientPresence to t
3315 let localOnline = hasLocalPresence toPresence
3435 let localOnline = hasLocalPresence toPresence
3316 let remoteServers = collectPeerServers toPresence (\p -> p.server)
3436 let remoteServers = collectPeerServers toPresence (\p -> p.server)
3317− when (federationEnabled && count remoteServers > 0) (do
3437+ let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
3318 forwards <- *forwards
3438 forwards <- *forwards
3319 let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}
3439 let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}
3320 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
3440 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
3321− set *forwards = union forwards cappedFwds)
3441+ set *forwards = union forwards cappedFwds
3442+ -- Return whether any forwards were actually enqueued (vs dropped by capacity cap)
3443+ yield (count cappedFwds > 0)
3444+ else yield (False {})
3322 let status = if localOnline then MsDelivered {}
3445 let status = if localOnline then MsDelivered {}
3323− else if federationEnabled && count remoteServers > 0 then MsFederated {}
3446+ else if federationEnabled && forwardsEnqueued then MsFederated {}
3324 else MsQueued {}
3447 else MsQueued {}
3325 yield {status: status, remoteServers: remoteServers, retryAfterMs: 0}
3448 yield {status: status, remoteServers: remoteServers, retryAfterMs: 0}
@@ -3334,11 +3457,12 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
3334 when stillPending (do
3457 when stillPending (do
3335 fwdT <- now
3458 fwdT <- now
3336 fwdAuth <- generateFederationAuth "forward"
3459 fwdAuth <- generateFederationAuth "forward"
3460+ fwdCorrId <- genTokenN 1
3337 fwdResult <- fetchFederation server (ForwardMessage {
3461 fwdResult <- fetchFederation server (ForwardMessage {
3338 toKey: to, encryptedBlob: encryptedBlob,
3462 toKey: to, encryptedBlob: encryptedBlob,
3339 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
3463 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
3340 fromServer: serverHostN, requestTimestamp: fwdT,
3464 fromServer: serverHostN, requestTimestamp: fwdT,
3341− authorization: fwdAuth})
3465+ authorization: fwdAuth, correlationId: fwdCorrId})
3342 case fwdResult of
3466 case fwdResult of
3343 Err e -> do
3467 Err e -> do
3344 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
3468 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
@@ -3363,11 +3487,12 @@ sendBatchForwardsSequential = \msgs server ->
3363 when stillPending (do
3487 when stillPending (do
3364 fwdT <- now
3488 fwdT <- now
3365 fwdAuth <- generateFederationAuth "forward"
3489 fwdAuth <- generateFederationAuth "forward"
3490+ fwdCorrId <- genTokenN 1
3366 fwdResult <- fetchFederation server (ForwardMessage {
3491 fwdResult <- fetchFederation server (ForwardMessage {
3367 toKey: r.toKey, encryptedBlob: r.blob,
3492 toKey: r.toKey, encryptedBlob: r.blob,
3368 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
3493 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
3369 fromServer: serverHostN, requestTimestamp: fwdT,
3494 fromServer: serverHostN, requestTimestamp: fwdT,
3370− authorization: fwdAuth})
3495+ authorization: fwdAuth, correlationId: fwdCorrId})
3371 case fwdResult of
3496 case fwdResult of
3372 Err e -> do
3497 Err e -> do
3373 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
3498 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
@@ -3635,9 +3760,11 @@ handleVerifyToken = \token scope clientIp respond -> do
3635-- Validates events, merges into presence table (last-write-wins), and re-gossips
3760-- Validates events, merges into presence table (last-write-wins), and re-gossips
3636-- to other known peers with decremented hop count. Triggers message forwarding
3761-- to other known peers with decremented hop count. Triggers message forwarding
3637-- when a remote key comes online.
3762-- when a remote key comes online.
3638−handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorization respond -> do
3763+handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlationId authorization respond -> do
3639 t <- now
3764 t <- now
3640 let cappedHops = clamp 0 maxGossipHops hopsRemaining
3765 let cappedHops = clamp 0 maxGossipHops hopsRemaining
3766+ let corrId = if correlationId != "" then correlationId else ""
3767+ when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " gossip from " ++ fromServer))
3641 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
3768 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
3642 -- Per-peer gossip event volume limiting: counts total events (not just requests) per server.
3769 -- Per-peer gossip event volume limiting: counts total events (not just requests) per server.
3643 -- A single gossip request carries up to maxGossipEvents entries; without this, a peer
3770 -- A single gossip request carries up to maxGossipEvents entries; without this, a peer
@@ -3682,18 +3809,28 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
3682 yield (do e <- firstParty; where any localPresence (\p -> p.pubkey == e.pubkey && isLocalServer p.server && p.expiresAt > t); yield e.pubkey)
3809 yield (do e <- firstParty; where any localPresence (\p -> p.pubkey == e.pubkey && isLocalServer p.server && p.expiresAt > t); yield e.pubkey)
3683 when (count localClashes > 0)
3810 when (count localClashes > 0)
3684 (logWarn ("gossip: " ++ fromServer ++ " claims " ++ show (count localClashes) ++ " locally-authenticated key(s) — ignoring for local routing"))
3811 (logWarn ("gossip: " ++ fromServer ++ " claims " ++ show (count localClashes) ++ " locally-authenticated key(s) — ignoring for local routing"))
3812+ -- Per-bucket gossip merge: read only presence buckets affected by this gossip batch
3813+ -- instead of all 16. All entries for a pubkey hash to the same bucket, so per-bucket
3814+ -- merge is equivalent to global merge. Reduces lock scope for small gossip batches.
3815+ let allGossipEntries = union deduped trustedOffline
3816+ let affectedBuckets = uniqueBucketsFor allGossipEntries (\e -> e.pubkey)
3685 gossipResult <- atomic do
3817 gossipResult <- atomic do
3686− allPresence <- readAllPresence
3818+ selectedPresence <- readSelectedPresenceBuckets affectedBuckets
3687 tpc <- *totalPresenceCount
3819 tpc <- *totalPresenceCount
3688 -- Pre-filter: when near capacity, reject gossip entries for pubkeys not already in presence.
3820 -- Pre-filter: when near capacity, reject gossip entries for pubkeys not already in presence.
3689 -- This prevents a malicious peer from spiking presence above maxPresenceEntries via large batches.
3821 -- This prevents a malicious peer from spiking presence above maxPresenceEntries via large batches.
3690 let capacityDeduped = if tpc >= maxPresenceEntries
3822 let capacityDeduped = if tpc >= maxPresenceEntries
3691− then do d <- deduped; where any allPresence (\p -> p.pubkey == d.pubkey); yield d
3823+ then do d <- deduped; where any selectedPresence (\p -> p.pubkey == d.pubkey); yield d
3692 else deduped
3824 else deduped
3693− let gm = mergeGossipPresence allPresence capacityDeduped trustedOffline fromServer t
3825+ let gm = mergeGossipPresence selectedPresence capacityDeduped trustedOffline fromServer t
3694− writeAllPresence gm.finalPresence
3826+ -- Write back only affected buckets (not writeAllPresence, which would zero unaffected buckets)
3827+ writeSelectedPresenceBuckets affectedBuckets gm.finalPresence
3828+ let delta = count gm.finalPresence - count selectedPresence
3695− set *totalPresenceCount = count gm.finalPresence
3829+ set *totalPresenceCount = max 0 (tpc + delta)
3830+ -- Merge new peer servers with existing known peers (selected buckets may not contain all peers)
3831+ knownPeers <- *knownPeerServers
3696− set *knownPeerServers = gm.gossipPeerServers
3832+ let newPeers = gm.gossipPeerServers
3833+ set *knownPeerServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union acc [s]) knownPeers newPeers
3697 set *presenceVersion = t
3834 set *presenceVersion = t
3698 yield {gm: gm, preFiltered: count deduped - count capacityDeduped}
3835 yield {gm: gm, preFiltered: count deduped - count capacityDeduped}
3699 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
3836 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
@@ -3732,8 +3869,9 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
3732-- (fromServer != originServer) where an intermediate server re-forwards because
3869-- (fromServer != originServer) where an intermediate server re-forwards because
3733-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored
3870-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored
3734-- twice) and forwardCandidates (won't re-forward to the same server).
3871-- twice) and forwardCandidates (won't re-forward to the same server).
3735−handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond -> do
3872+handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId authorization respond -> do
3736 t <- now
3873 t <- now
3874+ when (correlationId != "") (logInfo ("[CORR] " ++ correlationId ++ " forward " ++ messageId ++ " from " ++ fromServer))
3737 withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
3875 withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
3738 when (fromServer != originServer)
3876 when (fromServer != originServer)
3739 (logInfo ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
3877 (logInfo ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
@@ -3811,13 +3949,9 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3811 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3949 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3812 fwdNonces <- *forwardNonces
3950 fwdNonces <- *forwardNonces
3813 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
3951 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
3814− -- Global age-based eviction: evict oldest nonces regardless of origin.
3952+ -- Evict expired nonces first (O(n)), then 1% LRU fallback to minimize replay window
3815− -- Previous per-originServer partitioning could be gamed by a relay peer
3816− -- forging varied originServer values to spread nonces across partitions
3817− -- and crowd out legitimate entries.
3818 set *forwardNonces = if count updatedNonces > maxForwardNonces
3953 set *forwardNonces = if count updatedNonces > maxForwardNonces
3819− then let evictCount = max 1 (count updatedNonces / 10) in
3954+ then evictNonces updatedNonces maxForwardNonces forwardNonceRetention t
3820− (evictSmallest updatedNonces evictCount (\n -> n.seenAt)).kept
3821 else updatedNonces
3955 else updatedNonces
3822 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
3956 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
3823 if result.noPresence then do
3957 if result.noPresence then do
@@ -3839,8 +3973,9 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3839-- POST /federation/ack: Peer server confirms delivery of messages we originated.
3973-- POST /federation/ack: Peer server confirms delivery of messages we originated.
3840-- Removes corresponding forward entries. Also cleans up the original message if the
3974-- Removes corresponding forward entries. Also cleans up the original message if the
3841-- recipient has no local presence and no other pending forwards for that message.
3975-- recipient has no local presence and no other pending forwards for that message.
3842−handleFederationAck = \messageIds originServer fromServer requestTimestamp authorization respond -> do
3976+handleFederationAck = \messageIds originServer fromServer requestTimestamp correlationId authorization respond -> do
3843 t <- now
3977 t <- now
3978+ when (correlationId != "") (logInfo ("[CORR] " ++ correlationId ++ " ack from " ++ fromServer))
3844 -- Wrap respond to add ackedCount default for withFederationAuth error paths
3979 -- Wrap respond to add ackedCount default for withFederationAuth error paths
3845 let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0}
3980 let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0}
3846 withFederationAuth authorization t ackRespond fromServer "ack" requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (do
3981 withFederationAuth authorization t ackRespond fromServer "ack" requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (do
@@ -3973,7 +4108,18 @@ handleHealth = \authorization respond -> do
3973 -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
4108 -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
3974 -- but have their own dedicated rate limiter to prevent DDoS on /health from saturating reads.
4109 -- but have their own dedicated rate limiter to prevent DDoS on /health from saturating reads.
3975 healthLimited <- checkHealthRate t
4110 healthLimited <- checkHealthRate t
4111+ if healthLimited then do
4112+ -- Return cached last-known health (prevents DDoS on /health from masking real failures)
4113+ cached <- atomic do
4114+ cOk <- *cachedHealthOk
4115+ cP <- *cachedHealthPrune
4116+ cH <- *cachedHealthHeartbeat
4117+ cC <- *cachedHealthCapacity
4118+ cF <- *cachedHealthFederation
4119+ cS <- *cachedHealthSnapshot
4120+ cW <- *cachedHealthWatchdog
3976− if healthLimited then yield (respond {ok: True {}, pruneHealthy: True {}, heartbeatHealthy: True {}, capacityHealthy: True {}, federationHealthy: True {}, snapshotHealthy: True {}, watchdogHealthy: True {}})
4121+ yield {ok: cOk, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW}
4122+ yield (respond cached)
3978 let isAuthed = verifyMaintenanceAuth authorization
4124 let isAuthed = verifyMaintenanceAuth authorization
3979 result <- atomic do
4125 result <- atomic do
@@ -4016,6 +4162,15 @@ handleHealth = \authorization respond -> do
4016 snapshotOk: snapOk,
4162 snapshotOk: snapOk,
4017 watchdogOk: wdOk}
4163 watchdogOk: wdOk}
4018 let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
4164 let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
4165+ -- Cache health result for rate-limited responses
4167+ set *cachedHealthOk = allOk
4168+ set *cachedHealthPrune = result.pruneOk
4169+ set *cachedHealthHeartbeat = result.heartbeatOk
4170+ set *cachedHealthCapacity = result.capacityOk
4171+ set *cachedHealthFederation = result.federationOk
4172+ set *cachedHealthSnapshot = result.snapshotOk
4173+ set *cachedHealthWatchdog = result.watchdogOk
4019 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
4174 -- Unauthenticated: return real ok boolean so load balancers can route around unhealthy nodes,
4020 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
4175 -- but hide subsystem breakdown to prevent targeted load-correlation attacks.
4021 if isAuthed
4176 if isAuthed
@@ -4372,7 +4527,12 @@ handleSendBatch = \messages authorization clientIp respond -> do
4372 -- Fork federation forwarding: one forked task per target server (not per message).
4527 -- Fork federation forwarding: one forked task per target server (not per message).
4373 -- Within each task, forwards are sent sequentially to avoid per-message fork overhead.
4528 -- Within each task, forwards are sent sequentially to avoid per-message fork overhead.
4374 let forwardTargets = do r <- results; where count r.remoteServers > 0; yield r
4529 let forwardTargets = do r <- results; where count r.remoteServers > 0; yield r
4375− let allServers = collectPeerServers forwardTargets (\r -> case headOf r.remoteServers of Nothing {} -> ""; Just {value: s} -> s)
4530+ -- Flatten all remote servers from all results (previous code used headOf, missing 2nd+ servers per message)
4531+ let allServers = fold (\acc r ->
4532+ fold (\innerAcc s ->
4533+ if isLocalServer s then innerAcc
4534+ else if any innerAcc (\a -> a == s) then innerAcc
4535+ else union innerAcc [s]) acc r.remoteServers) [] forwardTargets
4376 fork (when (federationEnabled && count forwardTargets > 0)
4536 fork (when (federationEnabled && count forwardTargets > 0)
4377 (sendToEach allServers (\server -> do
4537 (sendToEach allServers (\server -> do
4378 let serverMsgs = do r <- forwardTargets; where any r.remoteServers (\s -> s == server); yield r
4538 let serverMsgs = do r <- forwardTargets; where any r.remoteServers (\s -> s == server); yield r
@@ -4461,10 +4621,11 @@ handlePingPeer = \server authorization respond -> do
4462 startTime <- now
4622 startTime <- now
4463 auth <- generateFederationAuth "gossip"
4623 auth <- generateFederationAuth "gossip"
4624+ pingCorrId <- genTokenN 1
4464 result <- fetchFederation server (RecvGossip {
4625 result <- fetchFederation server (RecvGossip {
4465 authorization: auth,
4626 authorization: auth,
4466 events: [], hopsRemaining: 1, fromServer: serverHostN,
4627 events: [], hopsRemaining: 1, fromServer: serverHostN,
4467− requestTimestamp: startTime})
4628+ requestTimestamp: startTime, correlationId: pingCorrId})
4468 endTime <- now
4629 endTime <- now
4469 let latency = endTime - startTime
4630 let latency = endTime - startTime
4470 case result of
4631 case result of
@@ -4516,6 +4677,37 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do
4516 (fork (sendBatchedGossip result.peerServers [{eventType: GossipOffline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))
4677 (fork (sendBatchedGossip result.peerServers [{eventType: GossipOffline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))
4517 yield (respond {ok: True {}, revokedCount: result.revokedCount})
4678 yield (respond {ok: True {}, revokedCount: result.revokedCount})
4680+-- GET /maintenance/rate-limits: Return current effective rate limits.
4681+handleGetRateLimits = \authorization respond -> do
4682+ let authorized = verifyMaintenanceAuth authorization
4683+ if not authorized then yield (respond {ok: False {}, globalRequest: 0, ipRate: 0, sendRate: 0, federationRate: 0, authGlobal: 0, recvRate: 0})
4685+ result <- atomic do
4686+ gr <- effectiveGlobalRequestRate
4687+ ir <- effectiveIpRate
4688+ sr <- effectiveSendRate
4689+ fr <- effectiveFederationRate
4690+ ar <- effectiveAuthGlobalRate
4691+ rr <- effectiveRecvRate
4692+ yield {globalRequest: gr, ipRate: ir, sendRate: sr, federationRate: fr, authGlobal: ar, recvRate: rr}
4693+ yield (respond {ok: True {}, globalRequest: result.globalRequest, ipRate: result.ipRate, sendRate: result.sendRate, federationRate: result.federationRate, authGlobal: result.authGlobal, recvRate: result.recvRate})
4695+-- POST /maintenance/rate-limits: Update rate limits at runtime. 0 = reset to compile-time default.
4696+-- Negative values are clamped to 0. Changes take effect on the next rate check cycle.
4697+handleSetRateLimits = \globalRequest ipRate sendRate federationRate authGlobal recvRate authorization respond -> do
4698+ let authorized = verifyMaintenanceAuth authorization
4699+ if not authorized then yield (respond {ok: False {}})
4702+ set *rateOverrideGlobalRequest = max 0 globalRequest
4703+ set *rateOverrideIpRate = max 0 ipRate
4704+ set *rateOverrideSendRate = max 0 sendRate
4705+ set *rateOverrideFederationRate = max 0 federationRate
4706+ set *rateOverrideAuthGlobal = max 0 authGlobal
4707+ set *rateOverrideRecvRate = max 0 recvRate
4708+ logInfo ("rate-limits: updated — global=" ++ show globalRequest ++ " ip=" ++ show ipRate ++ " send=" ++ show sendRate ++ " federation=" ++ show federationRate ++ " auth=" ++ show authGlobal ++ " recv=" ++ show recvRate ++ " (0 = default)")
4709+ yield (respond {ok: True {}})
4519-- Extract request path for access logging (excludes poll to reduce log volume)
4711-- Extract request path for access logging (excludes poll to reduce log volume)
4520requestPath = \req -> case req of
4712requestPath = \req -> case req of
4521 ReqChallenge {} -> "POST /auth/challenge"
4713 ReqChallenge {} -> "POST /auth/challenge"
@@ -4545,6 +4737,8 @@ requestPath = \req -> case req of
4545 PingPeer {} -> "POST /maintenance/ping"
4737 PingPeer {} -> "POST /maintenance/ping"
4546 AdminListSessions {} -> "GET /maintenance/sessions"
4738 AdminListSessions {} -> "GET /maintenance/sessions"
4547 AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"
4739 AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"
4740+ GetRateLimits {} -> "GET /maintenance/rate-limits"
4741+ SetRateLimits {} -> "POST /maintenance/rate-limits"
4548 Health {} -> "GET /health"
4742 Health {} -> "GET /health"
4549 Stats {} -> "GET /stats"
4743 Stats {} -> "GET /stats"
4550 StatsPeers {} -> "GET /stats/peers"
4744 StatsPeers {} -> "GET /stats/peers"
@@ -4579,6 +4773,8 @@ requestClientIp = \req -> case req of
4579 PingPeer {} -> ""
4773 PingPeer {} -> ""
4580 AdminListSessions {} -> ""
4774 AdminListSessions {} -> ""
4581 AdminRevokeSessions {} -> ""
4775 AdminRevokeSessions {} -> ""
4776+ GetRateLimits {} -> ""
4777+ SetRateLimits {} -> ""
4582 Health {} -> ""
4778 Health {} -> ""
4583 Stats {} -> ""
4779 Stats {} -> ""
4584 StatsPeers {} -> ""
4780 StatsPeers {} -> ""
@@ -4601,9 +4797,9 @@ serveInner = \req -> case req of
4601 ListMessages {authorization, clientIp, respond} -> handleListMessages authorization clientIp respond
4797 ListMessages {authorization, clientIp, respond} -> handleListMessages authorization clientIp respond
4602 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
4798 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
4603 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
4799 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
4604− RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
4800+ RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
4605− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond
4801+ ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId authorization respond
4606− FederationAck {messageIds, originServer, fromServer, requestTimestamp, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp authorization respond
4802+ FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
4607 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
4803 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
4608 Prune {authorization, respond} -> handlePrune authorization respond
4804 Prune {authorization, respond} -> handlePrune authorization respond
4609 Drain {authorization, respond} -> handleDrain authorization respond
4805 Drain {authorization, respond} -> handleDrain authorization respond
@@ -4616,6 +4812,8 @@ serveInner = \req -> case req of
4616 PingPeer {server, authorization, respond} -> handlePingPeer server authorization respond
4812 PingPeer {server, authorization, respond} -> handlePingPeer server authorization respond
4617 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
4813 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
4618 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
4814 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
4815+ GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
4816+ SetRateLimits {globalRequest, ipRate, sendRate, federationRate, authGlobal, recvRate, authorization, respond} -> handleSetRateLimits globalRequest ipRate sendRate federationRate authGlobal recvRate authorization respond
4619 Health {authorization, respond} -> handleHealth authorization respond
4817 Health {authorization, respond} -> handleHealth authorization respond
4620 Stats {authorization, respond} -> handleStats authorization respond
4818 Stats {authorization, respond} -> handleStats authorization respond
4621 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
4819 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
@@ -4661,13 +4859,24 @@ backgroundPrune = do
4661 backgroundPrune
4859 backgroundPrune
4663 t <- now
4861 t <- now
4664− -- Detect backward clock jump: if seqCounter is far ahead of now, reset to prevent
4862+ -- Detect clock jumps in both directions:
4665− -- all messages getting unreachable receivedAt values for an extended period
4863+ -- Backward: seqCounter ahead of now (clock corrected back after forward jump)
4864+ -- Forward: now jumped far ahead of last prune (VM suspend, NTP step)
4865+ -- Forward jumps cause inflated receivedAt timestamps; clients with those cursors
4866+ -- miss new messages until real time catches up. Log a warning so operators can
4867+ -- advise affected clients to reset their cursor (generation change handles this).
4666 atomic do
4868 atomic do
4667 sc <- *seqCounter
4869 sc <- *seqCounter
4870+ lp <- *lastPrunedAt
4668 when (sc > t + clockSkewTolerance) (do
4871 when (sc > t + clockSkewTolerance) (do
4669 logWarn ("clock jump detected — seqCounter " ++ show (sc - t) ++ "ms ahead of now, resetting")
4872 logWarn ("clock jump detected — seqCounter " ++ show (sc - t) ++ "ms ahead of now, resetting")
4670− set *seqCounter = t)
4873+ set *seqCounter = t
4874+ -- Bump server generation so clients reset their cursors (avoids stuck polls
4875+ -- from messages with inflated receivedAt timestamps during the forward jump)
4876+ set *serverGeneration = t)
4877+ -- Forward jump detection: now is far ahead of last prune (more than 10x interval)
4878+ when (lp > 0 && t - lp > backgroundPruneInterval * 10) (do
4879+ logWarn ("clock forward jump detected — " ++ show (t - lp) ++ "ms since last prune (expected ~" ++ show backgroundPruneInterval ++ "ms), messages during this period may have inflated timestamps"))
4671 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead
4880 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead
4672 sa <- *startedAt
4881 sa <- *startedAt
4673 lastHb <- *lastHeartbeatAt
4882 lastHb <- *lastHeartbeatAt
@@ -4737,9 +4946,8 @@ pollHeartbeat = do
4738-- Background loop: periodically logs a state summary to stdout.
4947-- Background loop: periodically logs a state summary to stdout.
4739-- Operators can capture stdout to detect state drift; use /maintenance/snapshot for full export.
4948-- Operators can capture stdout to detect state drift; use /maintenance/snapshot for full export.
4740−-- The drain handler now auto-exports a [SNAPSHOT_DATA] JSON line for log-shipping pipelines,
4949+-- When autoSnapshotExportInterval > 0, periodically emits [SNAPSHOT_DATA] JSON to stdout
4741−-- but a crash outside drain still loses state. Mitigate with frequent /maintenance/snapshot calls
4950+-- for sidecar capture, so a crash outside drain loses at most one export interval of state.
4742−-- or a sidecar that watches for [SNAPSHOT_DATA] lines and writes them to disk.
4743-- TODO: If Knot adds file I/O primitives, auto-export snapshots to a file
4951-- TODO: If Knot adds file I/O primitives, auto-export snapshots to a file
4744-- (e.g., /var/lib/skrepka/snapshot.json) in addition to logging.
4952-- (e.g., /var/lib/skrepka/snapshot.json) in addition to logging.
4745backgroundSnapshot = do
4953backgroundSnapshot = do
@@ -4758,14 +4966,36 @@ backgroundSnapshot = do
4758 pac <- *pendingAckCleanups
4966 pac <- *pendingAckCleanups
4759 yield {messages: tmc, sessions: sc, presence: pc, forwards: count f, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff, pendingAckCleanups: count pac}
4967 yield {messages: tmc, sessions: sc, presence: pc, forwards: count f, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff, pendingAckCleanups: count pac}
4760 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 ++ " pendingAckCleanups=" ++ show result.pendingAckCleanups)
4968 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 ++ " pendingAckCleanups=" ++ show result.pendingAckCleanups)
4761− -- Warn if federation state hasn't been exported recently (will be lost on restart)
4762 t <- now
4969 t <- now
4970+ -- Periodic full snapshot auto-export: emit [SNAPSHOT_DATA] JSON to stdout so a
4971+ -- sidecar process can persist it. Only when there's meaningful state to save.
4972+ when (autoSnapshotExportInterval > 0) (do
4973+ lastAutoExport <- *lastAutoSnapshotExportAt
4974+ let hasState = result.presence > 0 || result.forwards > 0 || result.ackedIds > 0 || result.deliveredIds > 0
4975+ when (hasState && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do
4976+ exportData <- atomic do
4977+ p <- readAllPresence
4981+ ff <- *forwardFailures
4983+ fn <- *forwardNonces
4984+ bp <- *blockedPeers
4985+ yield {version: snapshotVersion, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}
4986+ println ("[SNAPSHOT_DATA] " ++ toJson exportData)
4987+ atomic do set *lastAutoSnapshotExportAt = t
4988+ logInfo ("snapshot: auto-exported full state (" ++ show result.presence ++ " presence, " ++ show result.forwards ++ " forwards)")))
4989+ -- Warn if federation state hasn't been exported recently (will be lost on restart)
4763 lastExport <- *lastSnapshotExportAt
4990 lastExport <- *lastSnapshotExportAt
4991+ lastAutoExport <- *lastAutoSnapshotExportAt
4992+ -- Consider auto-exports as valid exports for staleness check
4993+ let effectiveLastExport = max lastExport lastAutoExport
4764 when (maintenanceSecret != "" && snapshotStalenessWarning > 0
4994 when (maintenanceSecret != "" && snapshotStalenessWarning > 0
4765− && (lastExport == 0 || t - lastExport > snapshotStalenessWarning)
4995+ && (effectiveLastExport == 0 || t - effectiveLastExport > snapshotStalenessWarning)
4766 && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))
4996 && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))
4768− 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 ++ ")")
4998+ logError ("no snapshot exported in " ++ show (if effectiveLastExport == 0 then 0 else (t - effectiveLastExport) / 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 ++ ")")
4769 logError "ACTION REQUIRED: run GET /maintenance/snapshot to persist federation state before restart")
4999 logError "ACTION REQUIRED: run GET /maintenance/snapshot to persist federation state before restart")
4770 else yield {}
5000 else yield {}
4771 backgroundSnapshot
5001 backgroundSnapshot
@@ -4821,7 +5051,7 @@ startupProbeNext = \peers probed ->
4821 result <- fetchFederation server (RecvGossip {
5051 result <- fetchFederation server (RecvGossip {
4822 authorization: auth,
5052 authorization: auth,
4823 events: [], hopsRemaining: 1, fromServer: serverHostN,
5053 events: [], hopsRemaining: 1, fromServer: serverHostN,
4824− requestTimestamp: startTime})
5054+ requestTimestamp: startTime, correlationId: ""})
4825 endTime <- now
5055 endTime <- now
4826 let latency = endTime - startTime
5056 let latency = endTime - startTime
4827 case result of
5057 case result of
@@ -4905,6 +5135,8 @@ main = do
4905 (logInfo (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
5135 (logInfo (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
4906 when (autoSnapshotInterval > 0)
5136 when (autoSnapshotInterval > 0)
4907 (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
5137 (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
5138+ when (autoSnapshotExportInterval > 0)
5139+ (logInfo (" auto-snapshot export: " ++ show autoSnapshotExportInterval ++ "ms (full [SNAPSHOT_DATA] to stdout)"))
4908 fork backgroundPrune
5140 fork backgroundPrune
4909 fork pollHeartbeat
5141 fork pollHeartbeat
4910 fork backgroundWatchdog
5142 fork backgroundWatchdog