@@ -274,6 +274,12 @@ maxSenderExpirations = 10000
274maxSenderExpirationsPerKey = 100
274maxSenderExpirationsPerKey = 100
275-- Max pubkeys in a batch lookup request
275-- Max pubkeys in a batch lookup request
276maxBatchLookupKeys = 50
276maxBatchLookupKeys = 50
277+-- Verify callback caching: cache successful verifications per-server to reduce
278+-- outbound HTTP calls. A peer sending N requests/second forces N verify callbacks;
279+-- caching the success for a short window amortizes the cost.
280+verificationCacheMs : Int<Ms>
281+verificationCacheMs = 2000 -- 2 seconds; short enough to limit replay window
282+maxVerificationCache = 100
277-- Health endpoint rate limiting (global, since /health has no clientIp)
283-- Health endpoint rate limiting (global, since /health has no clientIp)
278maxHealthRate = 100 -- max health checks per window
284maxHealthRate = 100 -- max health checks per window
279healthRateWindowMs : Int<Ms>
285healthRateWindowMs : Int<Ms>
@@ -534,6 +540,11 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
535*lastPresenceRefresh : Timestamp
541*lastPresenceRefresh : Timestamp
543+-- Round-robin presence refresh: tracks which bucket group (0-3) to scan next.
544+-- Each refresh cycle scans 4 buckets (0-3, 4-7, 8-11, 12-15), cycling through
545+-- all 16 over 4 refresh intervals. Spreads gossip traffic across prune cycles.
546+*presenceRefreshGroup : Int
537*challengeRates : [ChallengeRateEntry]
548*challengeRates : [ChallengeRateEntry]
539-- Separate revoke rate limiter (prevents challenge floods from exhausting revoke budget)
550-- Separate revoke rate limiter (prevents challenge floods from exhausting revoke budget)
@@ -744,6 +755,9 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
744-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
755-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
745*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
756*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
758+-- Cached successful verify callbacks per-server (short-lived, reduces outbound HTTP calls)
759+*verificationCache : [{server: Text, scope: Text, verifiedAt: Timestamp}]
747-- Per-IP unique key tracking: detects when a single IP authenticates many distinct pubkeys
761-- Per-IP unique key tracking: detects when a single IP authenticates many distinct pubkeys
748-- (key-rotation abuse bypassing per-key rate limits). Pruned with other rate limiters.
762-- (key-rotation abuse bypassing per-key rate limits). Pruned with other rate limiters.
749*uniqueKeyIpRates : [UniqueKeyIpEntry]
763*uniqueKeyIpRates : [UniqueKeyIpEntry]
@@ -765,6 +779,15 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
765*rateOverrideLookupRate : Int
779*rateOverrideLookupRate : Int
766*rateOverrideClientWriteRate : Int
780*rateOverrideClientWriteRate : Int
782+-- Rate-limit rejection counters: track how many requests were rejected by each rate limiter.
783+-- Exposed via /stats for operators to tune rate limits without parsing logs.
784+-- Reset on each federation latency window reset (same cadence as other metrics).
785+*rateLimitedGlobal : Int
787+*rateLimitedAuth : Int
788+*rateLimitedPerKey : Int
789+*rateLimitedFederation : Int
768-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
791-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
769-- Polls compare against this to skip expensive presence lookups when nothing changed.
792-- Polls compare against this to skip expensive presence lookups when nothing changed.
770*presenceVersion : Timestamp
793*presenceVersion : Timestamp
@@ -958,6 +981,12 @@ route MaintenanceApi where
958 POST {ip: Text} /maintenance/sessions/revoke-ip headers {authorization: Text}
981 POST {ip: Text} /maintenance/sessions/revoke-ip headers {authorization: Text}
959 -> {ok: Bool, revokedCount: Int, affectedKeys: Int}
982 -> {ok: Bool, revokedCount: Int, affectedKeys: Int}
960 = AdminRevokeIp
983 = AdminRevokeIp
984+ -- List all active session pubkeys grouped by IP (admin debugging, abuse investigation).
985+ -- Helps detect suspicious multi-key activity without knowing the pubkey in advance.
986+ -- limit: max entries to return (default 100, capped at 500). 0 = default.
987+ POST {limit: Int} /maintenance/sessions/list headers {authorization: Text}
988+ -> {sessions: [{pubkey: PubkeyHex, count: Int, ips: [Text], latestExpiry: Timestamp}], ok: Bool, total: Int}
989+ = AdminListAllSessions
961 -- Read current effective rate limits (compile-time defaults + runtime overrides)
990 -- Read current effective rate limits (compile-time defaults + runtime overrides)
962 GET /maintenance/rate-limits headers {authorization: Text}
991 GET /maintenance/rate-limits headers {authorization: Text}
963 -> {ok: Bool, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}
992 -> {ok: Bool, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}
@@ -999,7 +1028,7 @@ route HealthApi where
999-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1028-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
1000route StatsApi where
1029route StatsApi where
1001 GET /stats headers {authorization: Text}
1030 GET /stats headers {authorization: Text}
1002− -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, activePolls: Int, activePollsMax: Int, recipientCounts: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, senderExpirations: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int}}
1031+ -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, messagesMax: Int, messagesPct: Int, totalMessageCount: Int, presence: Int, presenceMax: Int, presencePct: Int, forwards: Int, forwardsMax: Int, forwardsPct: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, activePolls: Int, activePollsMax: Int, recipientCounts: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, senderExpirations: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int}}
1004 GET /stats/peers headers {authorization: Text}
1033 GET /stats/peers headers {authorization: Text}
1005 -> {peers: [PeerStatus], ok: Bool}
1034 -> {peers: [PeerStatus], ok: Bool}
@@ -1086,9 +1115,19 @@ genToken = genTokenN tokenChunks
1086tooLong : Text -> Int -> Bool
1115tooLong : Text -> Int -> Bool
1087tooLong = \s maxLen -> drop maxLen s != ""
1116tooLong = \s maxLen -> drop maxLen s != ""
1089−-- Truncate correlationId to maxCorrelationIdLen to prevent log flooding from malicious peers
1118+-- Sanitize correlationId: truncate to maxCorrelationIdLen and strip control characters
1119+-- (newlines, carriage returns, tabs) to prevent log injection attacks.
1120+-- A malicious peer could embed "\n[SNAPSHOT_DATA] ..." to spoof snapshot lines
1121+-- parsed by sidecar processes, or inject fake [ERROR] entries.
1090sanitizeCorrelationId : Text -> Text
1122sanitizeCorrelationId : Text -> Text
1091−sanitizeCorrelationId = \cid -> take maxCorrelationIdLen cid
1123+sanitizeCorrelationId = \cid -> stripControlChars (take maxCorrelationIdLen cid)
1125+stripControlChars : Text -> Text
1126+stripControlChars = \s ->
1128+ else let c = take 1 s in
1129+ if c == "\n" || c == "\r" || c == "\t" then stripControlChars (drop 1 s)
1130+ else c ++ stripControlChars (drop 1 s)
1093isHexChar : Text -> Bool
1132isHexChar : Text -> Bool
1094isHexChar = \c -> contains c "0123456789abcdef"
1133isHexChar = \c -> contains c "0123456789abcdef"
@@ -1264,7 +1303,8 @@ isBadServerName = \sn ->
1264 || matchesHostnameOrSubdomain sn "internal"
1303 || matchesHostnameOrSubdomain sn "internal"
1265 || matchesHostnameOrSubdomain sn "metadata"
1304 || matchesHostnameOrSubdomain sn "metadata"
1266 || take 4 sn == "127." || take 3 sn == "10."
1305 || take 4 sn == "127." || take 3 sn == "10."
1267− || take 8 sn == "192.168." || take 8 sn == "169.254."
1306+ || take 8 sn == "192.168."
1307+ || take 8 sn == "169.254." -- link-local; also covers AWS EC2 metadata endpoint 169.254.169.254
1268 || take 2 sn == "0." || fc == "["
1308 || take 2 sn == "0." || fc == "["
1269 || is172Private sn
1309 || is172Private sn
1270 || take 7 sn == "198.18." || take 7 sn == "198.19."
1310 || take 7 sn == "198.18." || take 7 sn == "198.19."
@@ -1367,9 +1407,11 @@ generateFederationAuth = \scope server -> do
1367 else updated
1407 else updated
1368 yield ("Bearer " ++ token)
1408 yield ("Bearer " ++ token)
1370−-- Verify an inbound federation token. First checks if a pre-shared key is configured
1410+-- Verify an inbound federation token. Checks a short-lived per-server cache first
1371−-- for the sender — if so, verifies via constant-time comparison (no network round-trip).
1411+-- to reduce outbound HTTP calls (a peer sending N requests/second would otherwise
1372−-- Otherwise falls back to callback verification via the sender's /federation/verify endpoint.
1412+-- force N verify callbacks). Cache hit means the server was verified within the last
1413+-- verificationCacheMs — the token itself is still single-use (consumed on first verify),
1414+-- but subsequent requests from the same server in the same window are trusted.
1373-- No retry: a failed verify rejects the request. The peer's federation rate limit + circuit breaker
1415-- No retry: a failed verify rejects the request. The peer's federation rate limit + circuit breaker
1374-- handle unreliable peers; retrying here doubles verify traffic and blocks with a 1s sleep per op,
1416-- handle unreliable peers; retrying here doubles verify traffic and blocks with a 1s sleep per op,
1375-- which a malicious peer can exploit for amplification.
1417-- which a malicious peer can exploit for amplification.
@@ -1378,6 +1420,13 @@ verifyFederationCallback = \fromServer authorization scope ->
1378 case extractToken authorization of
1420 case extractToken authorization of
1379 Nothing {} -> yield (False {})
1421 Nothing {} -> yield (False {})
1380 Just {value: token} -> do
1422 Just {value: token} -> do
1423+ -- Check verify cache: skip the outbound HTTP call if we recently verified this server+scope
1425+ cached <- atomic do
1426+ cache <- *verificationCache
1427+ yield (any cache (\c -> c.server == fromServer && c.scope == scope && t - c.verifiedAt < verificationCacheMs))
1428+ if cached then yield (True {})
1381 startTime <- now
1430 startTime <- now
1382 result <- fetchFederation fromServer (VerifyToken {token: token, scope: scope})
1431 result <- fetchFederation fromServer (VerifyToken {token: token, scope: scope})
1383 endTime <- now
1432 endTime <- now
@@ -1394,7 +1443,15 @@ verifyFederationCallback = \fromServer authorization scope ->
1394 vfc <- *verifyFailCount
1443 vfc <- *verifyFailCount
1395 set *verifyFailCount = vfc + 1
1444 set *verifyFailCount = vfc + 1
1396 yield (False {})
1445 yield (False {})
1447+ when r.ok (atomic do
1448+ cache <- *verificationCache
1449+ let cleaned = do c <- cache; where t - c.verifiedAt < verificationCacheMs; yield c
1450+ let updated = union cleaned [{server: fromServer, scope: scope, verifiedAt: t}]
1451+ set *verificationCache = if count updated > maxVerificationCache
1452+ then evictOneOldest updated (\c -> c.verifiedAt)
1397− Ok r -> yield r.ok
1454+ yield r.ok
1399-- ============================================
1456-- ============================================
1400-- Session Bucket Helpers
1457-- Session Bucket Helpers
@@ -1799,15 +1856,17 @@ transformAllPresenceBuckets = \f -> do
1799 b14 <- *presenceBucket14; set *presenceBucket14 = f b14
1856 b14 <- *presenceBucket14; set *presenceBucket14 = f b14
1800 b15 <- *presenceBucket15; set *presenceBucket15 = f b15
1857 b15 <- *presenceBucket15; set *presenceBucket15 = f b15
1802−-- Scan presence buckets one at a time for local entries nearing expiry.
1859+-- Scan presence buckets for local entries nearing expiry.
1803−-- Avoids materializing all 16 buckets when few keys are near expiry.
1860+-- Supports partial scans (startBucket to endBucket) for round-robin refresh:
1861+-- instead of scanning all 16 buckets every refresh cycle, the caller can
1862+-- scan 4 buckets per cycle, cycling through all 16 over 4 refresh intervals.
1804−findNearExpiryPresence : Int -> Timestamp -> IO [PresenceEntry]
1863+findNearExpiryPresence : Int -> Int -> Timestamp -> IO [PresenceEntry]
1805−findNearExpiryPresence = \b t ->
1864+findNearExpiryPresence = \b endBucket t ->
1806− if b > 15 then yield []
1865+ if b > endBucket then yield []
1808 bucket <- readPresenceBucket b
1867 bucket <- readPresenceBucket b
1809 let nearExpiry = do p <- bucket; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p
1868 let nearExpiry = do p <- bucket; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p
1810− rest <- findNearExpiryPresence (b + 1) t
1869+ rest <- findNearExpiryPresence (b + 1) endBucket t
1811 yield (union nearExpiry rest)
1870 yield (union nearExpiry rest)
1813-- ============================================
1872-- ============================================
@@ -3201,6 +3260,9 @@ checkFederationRate = \server t -> atomic do
3201 fedRate <- effectiveFederationRate
3260 fedRate <- effectiveFederationRate
3202 let result = checkServerRate federationRates server fedRate federationRateWindowMs maxFederationRateEntries t
3261 let result = checkServerRate federationRates server fedRate federationRateWindowMs maxFederationRateEntries t
3203 set *federationRates = result.entries
3262 set *federationRates = result.entries
3263+ when result.limited (do
3264+ rlf <- *rateLimitedFederation
3265+ set *rateLimitedFederation = rlf + 1)
3204 yield result.limited
3266 yield result.limited
3206checkGlobalRate = \t -> atomic do
3268checkGlobalRate = \t -> atomic do
@@ -3213,7 +3275,10 @@ checkGlobalRate = \t -> atomic do
3213 set *globalRateCount = 1
3275 set *globalRateCount = 1
3214 yield (False {})
3276 yield (False {})
3215 else if rateCount >= effectiveRate
3277 else if rateCount >= effectiveRate
3279+ rlg <- *rateLimitedGlobal
3280+ set *rateLimitedGlobal = rlg + 1
3216− then yield (True {})
3281+ yield (True {})
3218 set *globalRateCount = rateCount + 1
3283 set *globalRateCount = rateCount + 1
3219 yield (False {})
3284 yield (False {})
@@ -3229,7 +3294,10 @@ checkAuthGlobalRate = \t -> atomic do
3229 set *authGlobalRateCount = 1
3294 set *authGlobalRateCount = 1
3230 yield (False {})
3295 yield (False {})
3231 else if rateCount >= authRate
3296 else if rateCount >= authRate
3298+ rla <- *rateLimitedAuth
3299+ set *rateLimitedAuth = rla + 1
3232− then yield (True {})
3300+ yield (True {})
3234 set *authGlobalRateCount = rateCount + 1
3302 set *authGlobalRateCount = rateCount + 1
3235 yield (False {})
3303 yield (False {})
@@ -3257,6 +3325,9 @@ checkClientIpRate = \ip t ->
3257 ipRate <- effectiveIpRate
3325 ipRate <- effectiveIpRate
3258 let result = checkIpRate ipRates ip ipRate ipRateWindowMs maxIpRateEntries t
3326 let result = checkIpRate ipRates ip ipRate ipRateWindowMs maxIpRateEntries t
3259 set *ipRates = result.entries
3327 set *ipRates = result.entries
3328+ when result.limited (do
3329+ rli <- *rateLimitedIp
3330+ set *rateLimitedIp = rli + 1)
3260 yield result.limited
3331 yield result.limited
3262-- Combined pre-auth gate: drain check, global rate, IP rate. Returns "" if all pass.
3333-- Combined pre-auth gate: drain check, global rate, IP rate. Returns "" if all pass.
@@ -3326,13 +3397,15 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
3326 if globalLimited
3397 if globalLimited
3328 logWarn ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)
3399 logWarn ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)
3400+ -- Encode retryAfterMs in error string so peers can make smarter retry decisions.
3401+ -- Format: "rate_limited:NNNms" — peers that don't parse it still see "rate_limited:...".
3329− yield (respond {ok: False {}, error: "rate_limited"})
3402+ yield (respond {ok: False {}, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
3331 rateLimited <- checkFederationRate fromServer t
3404 rateLimited <- checkFederationRate fromServer t
3332 if rateLimited
3405 if rateLimited
3334 logWarn ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
3407 logWarn ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
3335− yield (respond {ok: False {}, error: "rate_limited"})
3408+ yield (respond {ok: False {}, error: "rate_limited:" ++ show federationRateWindowMs ++ "ms"})
3337 when clearOnAuth (clearFailure fromServer)
3410 when clearOnAuth (clearFailure fromServer)
@@ -3663,7 +3736,9 @@ sendBatchedAcksToServer = \serverAcks server ->
3664-- Send delivery acknowledgments back to origin servers and remove acked entries on success
3737-- Send delivery acknowledgments back to origin servers and remove acked entries on success
3665sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
3738sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
3666−sendAcksAndClean = \pendingAcks -> do
3739+sendAcksAndClean = \pendingAcks ->
3740+ if count pendingAcks == 0 then yield {}
3667 let servers = collectPeerServers pendingAcks (\a -> a.originServer)
3742 let servers = collectPeerServers pendingAcks (\a -> a.originServer)
3668 t <- now
3743 t <- now
3669 failures <- *forwardFailures
3744 failures <- *forwardFailures
@@ -3849,10 +3924,14 @@ pruneAndRetry = \t -> do
3849 -- Early return: skip expensive reads on ~9/10 prune cycles
3924 -- Early return: skip expensive reads on ~9/10 prune cycles
3850 if not shouldRefresh then yield {extendedActive: [], nearExpiryActive: [], refreshPeers: [], shouldRefresh: False {}}
3925 if not shouldRefresh then yield {extendedActive: [], nearExpiryActive: [], refreshPeers: [], shouldRefresh: False {}}
3852− -- Optimized: scan presence buckets one at a time to find near-expiry local entries,
3927+ -- Round-robin: scan 4 of 16 presence buckets per refresh cycle.
3853− -- then only read session buckets for the pubkeys found. Avoids materializing all 16
3928+ -- Spreads the gossip traffic across 4 refresh intervals instead of
3854− -- presence + 16 session buckets when few keys are near expiry (common case).
3929+ -- scanning all 16 buckets every cycle.
3930+ group <- *presenceRefreshGroup
3931+ let startBucket = (group * 4) - ((group * 4) / 16) * 16
3932+ let endBucket = min 15 (startBucket + 3)
3933+ set *presenceRefreshGroup = if group >= 3 then 0 else group + 1
3855− localNearExpiry <- findNearExpiryPresence 0 t
3934+ localNearExpiry <- findNearExpiryPresence startBucket endBucket t
3856 if count localNearExpiry == 0 then do
3935 if count localNearExpiry == 0 then do
3857 knownPeers <- *knownPeerServers
3936 knownPeers <- *knownPeerServers
3858 set *lastPresenceRefresh = t
3937 set *lastPresenceRefresh = t
@@ -3883,6 +3962,15 @@ pruneAndRetry = \t -> do
3883 set *federationLatencyMax = 0
3962 set *federationLatencyMax = 0
3884 set *slowVerifyCount = 0
3963 set *slowVerifyCount = 0
3885 set *verifyFailCount = 0
3964 set *verifyFailCount = 0
3965+ -- Reset rate-limit rejection counters (windowed with federation latency metrics)
3966+ set *rateLimitedGlobal = 0
3967+ set *rateLimitedIp = 0
3968+ set *rateLimitedAuth = 0
3969+ set *rateLimitedPerKey = 0
3970+ set *rateLimitedFederation = 0
3971+ -- Prune expired verify cache entries
3972+ cache <- *verificationCache
3973+ set *verificationCache = do c <- cache; where t - c.verifiedAt < verificationCacheMs; yield c
3886 set *federationLatencyResetAt = t)
3974 set *federationLatencyResetAt = t)
3887 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
3975 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
3888 -- These were queued by handleFederationAck instead of fork+sleep.
3976 -- These were queued by handleFederationAck instead of fork+sleep.
@@ -4446,9 +4534,12 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
4446 relevantPresence <- readSelectedPresenceBuckets neededBuckets
4534 relevantPresence <- readSelectedPresenceBuckets neededBuckets
4447 yield (do key <- cappedWatchKeys; yield {pubkey: key, online: count (presenceForKey relevantPresence key t2) > 0})
4535 yield (do key <- cappedWatchKeys; yield {pubkey: key, online: count (presenceForKey relevantPresence key t2) > 0})
4448 else yield []
4536 else yield []
4537+ -- Short-circuit: skip fork overhead when there are no pending acks or evicted entries.
4538+ -- Every poll with cursor > 0 reaches this path, but most polls have zero acks.
4449− fork (when federationEnabled (do
4539+ when (federationEnabled && (count result.pendingAcks > 0 || count result.evictedEntries > 0))
4450− sendAcksAndClean result.pendingAcks
4541+ sendAcksAndClean result.pendingAcks
4451− retryEvictedAcks result.evictedEntries result.pendingAcks))
4542+ retryEvictedAcks result.evictedEntries result.pendingAcks))
4452 let isHeartbeat = result.authorized && count result.events == 0
4543 let isHeartbeat = result.authorized && count result.events == 0
4453 -- Deliver message expiration notifications: atomically read and drain expirations
4544 -- Deliver message expiration notifications: atomically read and drain expirations
4454 -- for this sender's pubkey so each notification is delivered exactly once.
4545 -- for this sender's pubkey so each notification is delivered exactly once.
@@ -4698,7 +4789,10 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
4698 yield rateResult.limited
4789 yield rateResult.limited
4699 if lookupRateLimited then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
4790 if lookupRateLimited then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
4792+ -- Deduplicate input pubkeys to avoid redundant presence bucket reads
4793+ -- (a client could send 50 copies of the same pubkey, each triggering a scan)
4794+ let dedupedKeys = fold (\acc k -> if any acc (\a -> a == k) then acc else union acc [k]) [] pubkeys
4701− let cappedKeys = takeRelation maxBatchLookupKeys pubkeys
4795+ let cappedKeys = takeRelation maxBatchLookupKeys dedupedKeys
4702 result <- atomic do
4796 result <- atomic do
4703 let neededBuckets = uniqueBucketsFor cappedKeys (\k -> k)
4797 let neededBuckets = uniqueBucketsFor cappedKeys (\k -> k)
4704 relevantPresence <- readSelectedPresenceBuckets neededBuckets
4798 relevantPresence <- readSelectedPresenceBuckets neededBuckets
@@ -4904,7 +4998,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
4904 yield (False {})
4998 yield (False {})
4905 if eventRateLimited then do
4999 if eventRateLimited then do
4906 logWarn ("gossip: event volume rate limited for " ++ fromServer ++ " (" ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms)")
5000 logWarn ("gossip: event volume rate limited for " ++ fromServer ++ " (" ++ show maxGossipEventsPerServer ++ " events/" ++ show gossipEventRateWindowMs ++ "ms)")
4907− yield (respond {ok: False {}, error: "event_rate_limited"})
5001+ yield (respond {ok: False {}, error: "event_rate_limited:" ++ show gossipEventRateWindowMs ++ "ms"})
4909 let validated = validGossipEvents events t
5003 let validated = validGossipEvents events t
4910 let onlineEntries = validated.online
5004 let onlineEntries = validated.online
@@ -5141,15 +5235,23 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
5141 -- Deferred origin message cleanup: only queue cleanup for messages that actually had
5235 -- Deferred origin message cleanup: only queue cleanup for messages that actually had
5142 -- matching forward entries. Messages acked without a forward entry are NOT cleaned up,
5236 -- matching forward entries. Messages acked without a forward entry are NOT cleaned up,
5143 -- preventing premature deletion when a peer acks messages it never actually received.
5237 -- preventing premature deletion when a peer acks messages it never actually received.
5144− when (ackedCount > 0) (atomic do
5238+ when (ackedCount > 0) (do
5239+ capWarning <- atomic do
5145− cleanups <- *pendingAckCleanups
5240+ cleanups <- *pendingAckCleanups
5146− let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
5241+ let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
5147− -- Per-key cap: limit cleanups per recipient to prevent one noisy key from dominating the queue
5242+ -- Per-key cap: limit cleanups per recipient to prevent one noisy key from dominating the queue
5148− let perKeyCapped = fold (\acc c ->
5243+ let perKeyCapped = fold (\acc c ->
5149− let existingForKey = count (do x <- acc; where x.toKey == c.toKey; yield x)
5244+ let existingForKey = count (do x <- acc; where x.toKey == c.toKey; yield x)
5150− in if existingForKey >= maxPendingAckCleanupsPerKey then acc else union acc [c])
5245+ in if existingForKey >= maxPendingAckCleanupsPerKey then acc else union acc [c])
5151− cleanups newCleanups
5246+ cleanups newCleanups
5247+ let perKeyDropped = count newCleanups - (count perKeyCapped - count cleanups)
5248+ let globalCapped = count perKeyCapped > maxPendingAckCleanups
5152− set *pendingAckCleanups = if count perKeyCapped > maxPendingAckCleanups then takeRelation maxPendingAckCleanups perKeyCapped else perKeyCapped)
5249+ set *pendingAckCleanups = if globalCapped then takeRelation maxPendingAckCleanups perKeyCapped else perKeyCapped
5250+ yield {perKeyDropped: perKeyDropped, globalCapped: globalCapped}
5251+ when (capWarning.perKeyDropped > 0)
5252+ (logWarn ("federation/ack: dropped " ++ show capWarning.perKeyDropped ++ " deferred cleanups (per-key cap " ++ show maxPendingAckCleanupsPerKey ++ " reached) — origin messages may not be reclaimed"))
5253+ when capWarning.globalCapped
5254+ (logWarn ("federation/ack: pendingAckCleanups at global capacity (" ++ show maxPendingAckCleanups ++ ") — oldest entries truncated, origin messages may not be reclaimed")))
5153 yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
5255 yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
5155-- POST /federation/cancel: Cancel a previously forwarded message.
5257-- POST /federation/cancel: Cancel a previously forwarded message.
@@ -5249,7 +5351,9 @@ handleDrain = \phase authorization respond -> do
5249 pac <- *pendingAckCleanups
5351 pac <- *pendingAckCleanups
5250 let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
5352 let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
5251 yield {version: snapshotVersion, checksum: drainCs, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac}
5353 yield {version: snapshotVersion, checksum: drainCs, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac}
5354+ -- Sentinel-wrapped: sidecar processes should verify [SNAPSHOT_END] follows to detect truncation
5252 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
5355 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
5356+ println "[SNAPSHOT_END]"
5253 atomic do set *prunesSinceSnapshot = 0
5357 atomic do set *prunesSinceSnapshot = 0
5254 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
5358 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
5255 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
5359 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
@@ -5453,7 +5557,7 @@ handleHealth = \authorization respond -> do
5453 watchdogHealthy: True {},
5557 watchdogHealthy: True {},
5454 draining: result.isDraining})
5558 draining: result.isDraining})
5456−zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0}}
5560+zeroStats = {ok: False {}, version: serverBuildVersion, capacity: {sessions: 0, sessionsMax: 0, sessionsPct: 0, messages: 0, messagesMax: 0, messagesPct: 0, totalMessageCount: 0, presence: 0, presenceMax: 0, presencePct: 0, forwards: 0, forwardsMax: 0, forwardsPct: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardNonces: 0, forwardNoncesMax: 0, activePolls: 0, activePollsMax: 0, recipientCounts: 0}, federation: {enabled: False {}, inFlight: 0, hungSlots: 0, avgLatencyMs: 0, maxLatencyMs: 0, callCount: 0, slowVerifies: 0, verifyFailures: 0, forwardFailures: 0, oldestForwardAge: 0, oldestAckAge: 0, recentGossip: 0, blockedPeers: 0, drainCompletedAt: 0, forwardServers: 0, maxForwardQueueDepth: 0}, rates: {send: 0, challenge: 0, federation: 0, federationPair: 0, global: 0, ip: 0, lookup: 0, recv: 0, pair: 0, gossipEvent: 0, verifyIp: 0, uniqueKeyIp: 0, pendingAckCleanups: 0, senderExpirations: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0}}
5458handleStats = \authorization respond -> do
5562handleStats = \authorization respond -> do
5459 t <- now
5563 t <- now
@@ -5514,7 +5618,12 @@ handleStats = \authorization respond -> do
5514 svc <- *slowVerifyCount
5618 svc <- *slowVerifyCount
5515 vfc <- *verifyFailCount
5619 vfc <- *verifyFailCount
5516 dca <- *drainCompletedAt
5620 dca <- *drainCompletedAt
5621+ rlg <- *rateLimitedGlobal
5622+ rli <- *rateLimitedIp
5623+ rla <- *rateLimitedAuth
5624+ rlk <- *rateLimitedPerKey
5625+ rlf <- *rateLimitedFederation
5517− yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, svc: svc, vfc: vfc, dca: dca}
5626+ yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, svc: svc, vfc: vfc, dca: dca, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf}
5518 -- Compute derived metrics
5627 -- Compute derived metrics
5519 let sessions = core.sc
5628 let sessions = core.sc
5520 let messages = core.tmc
5629 let messages = core.tmc
@@ -5536,7 +5645,7 @@ handleStats = \authorization respond -> do
5536 yield (respond {ok: True {}, version: serverBuildVersion,
5645 yield (respond {ok: True {}, version: serverBuildVersion,
5537 capacity: {sessions: sessions, sessionsMax: maxSessions, sessionsPct: sessPct, messages: messages, messagesMax: maxMessages, messagesPct: msgPct, totalMessageCount: messages, presence: presence, presenceMax: maxPresenceEntries, presencePct: presPct, forwards: fwds, forwardsMax: maxForwardsEntries, forwardsPct: fwdPct, challenges: count core.c, challengesMax: maxChallenges, ackedIds: core.aCount, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: core.dCount, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
5646 capacity: {sessions: sessions, sessionsMax: maxSessions, sessionsPct: sessPct, messages: messages, messagesMax: maxMessages, messagesPct: msgPct, totalMessageCount: messages, presence: presence, presenceMax: maxPresenceEntries, presencePct: presPct, forwards: fwds, forwardsMax: maxForwardsEntries, forwardsPct: fwdPct, challenges: count core.c, challengesMax: maxChallenges, ackedIds: core.aCount, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: core.dCount, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
5538 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, slowVerifies: counters.svc, verifyFailures: counters.vfc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca, forwardServers: fwdServerCount, maxForwardQueueDepth: maxFwdDepth},
5647 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, slowVerifies: counters.svc, verifyFailures: counters.vfc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca, forwardServers: fwdServerCount, maxForwardQueueDepth: maxFwdDepth},
5539− rates: {send: count counters.sr, challenge: count counters.cr, federation: count counters.fr, federationPair: count counters.fpr, global: counters.grc, ip: count counters.ir, lookup: count counters.lr, recv: count counters.rr, pair: count counters.prr, gossipEvent: count core.ger, verifyIp: count core.vir, uniqueKeyIp: count core.ukir, pendingAckCleanups: count core.pac, senderExpirations: count core.se},
5648+ rates: {send: count counters.sr, challenge: count counters.cr, federation: count counters.fr, federationPair: count counters.fpr, global: counters.grc, ip: count counters.ir, lookup: count counters.lr, recv: count counters.rr, pair: count counters.prr, gossipEvent: count core.ger, verifyIp: count core.vir, uniqueKeyIp: count core.ukir, pendingAckCleanups: count core.pac, senderExpirations: count core.se, rejectedGlobal: counters.rlg, rejectedIp: counters.rli, rejectedAuth: counters.rla, rejectedPerKey: counters.rlk, rejectedFederation: counters.rlf},
5540 timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0, drainPhase: counters.dr}})
5649 timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0, drainPhase: counters.dr}})
5541 else yield (respond zeroStats)
5650 else yield (respond zeroStats)
@@ -5896,6 +6005,14 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
5896-- Only works for locally-originated messages where the caller's pubkey matches the
6005-- Only works for locally-originated messages where the caller's pubkey matches the
5897-- stored senderKey. Federation forwards cannot be cancelled via this endpoint.
6006-- stored senderKey. Federation forwards cannot be cancelled via this endpoint.
5898-- Removes from recipient mailbox and cancels pending forwards.
6007-- Removes from recipient mailbox and cancels pending forwards.
6008+-- NOTE: Race condition between cancel and re-send: after this endpoint deletes the
6009+-- local message and sends federation cancel, there's a window where:
6010+-- 1. The federation cancel hasn't arrived at the remote server yet
6011+-- 2. The sender immediately re-sends (with a new messageId)
6012+-- 3. The recipient receives the new message before the cancel is processed
6013+-- This can cause a UX inconsistency where the recipient sees the "retracted" message
6014+-- briefly before the cancel arrives. Not fixable at the relay layer — requires
6015+-- higher-level protocol support (e.g., cancel tokens in the encrypted payload).
5899handleCancelMessage = \messageId to authorization clientIp respond -> do
6016handleCancelMessage = \messageId to authorization clientIp respond -> do
5900 t <- now
6017 t <- now
5901 dp <- *drainPhase
6018 dp <- *drainPhase
@@ -6287,6 +6404,29 @@ checkCleanupForKeys = \keys t acc ->
6287 cleanup <- checkLastSessionCleanup k t
6404 cleanup <- checkLastSessionCleanup k t
6288 checkCleanupForKeys rest t (union acc [cleanup])
6405 checkCleanupForKeys rest t (union acc [cleanup])
6407+-- POST /maintenance/sessions/list: List all active pubkeys with session counts and IPs.
6408+-- Useful for detecting suspicious multi-key activity (e.g., one IP authenticating many keys)
6409+-- without needing to know the pubkey in advance. Groups sessions by pubkey.
6410+handleAdminListAllSessions = \limit authorization respond -> do
6411+ let authorized = verifyMaintenanceAuth authorization
6412+ if not authorized then yield (respond {sessions: [], ok: False {}, total: 0})
6414+ let effectiveLimit = if limit <= 0 then 100 else clamp 1 500 limit
6415+ allSessions <- readAllSessions
6417+ let active = do s <- allSessions; where s.expiresAt > t && t - s.createdAt < maxSessionLifetime; yield s
6418+ -- Group by pubkey: accumulate count, unique IPs, and latest expiry per pubkey.
6419+ -- O(n*k) where k = unique pubkeys; acceptable since readAllSessions already materialized all.
6420+ let grouped = fold (\acc s ->
6421+ if any acc (\a -> a.pubkey == s.pubkey)
6422+ then do a <- acc; yield (if a.pubkey == s.pubkey then {pubkey: a.pubkey, count: a.count + 1, ips: if any a.ips (\i -> i == s.ip) || s.ip == "" then a.ips else union a.ips [s.ip], latestExpiry: max a.latestExpiry s.expiresAt} else a)
6423+ else union acc [{pubkey: s.pubkey, count: 1, ips: if s.ip == "" then [] else [s.ip], latestExpiry: s.expiresAt}])
6425+ -- Sort by count descending (most sessions first) for abuse investigation
6426+ let sorted = sortBy (\g -> 0 - g.count) grouped
6427+ let capped = takeRelation effectiveLimit sorted
6428+ yield (respond {sessions: capped, ok: True {}, total: count grouped})
6290-- POST /maintenance/nonces/flush: Clear all forward nonces (recovery from corruption).
6430-- POST /maintenance/nonces/flush: Clear all forward nonces (recovery from corruption).
6291-- Nonces provide extended replay protection beyond message/ack dedup lifetime.
6431-- Nonces provide extended replay protection beyond message/ack dedup lifetime.
6292-- Flushing temporarily reopens a replay window for messages older than 30 days but
6432-- Flushing temporarily reopens a replay window for messages older than 30 days but
@@ -6374,6 +6514,7 @@ requestPath = \req -> case req of
6374 PingPeer {} -> "POST /maintenance/ping"
6514 PingPeer {} -> "POST /maintenance/ping"
6375 AdminListSessions {} -> "GET /maintenance/sessions"
6515 AdminListSessions {} -> "GET /maintenance/sessions"
6376 AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"
6516 AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"
6517+ AdminListAllSessions {} -> "POST /maintenance/sessions/list"
6377 AdminPurgeMessages {} -> "POST /maintenance/messages/purge"
6518 AdminPurgeMessages {} -> "POST /maintenance/messages/purge"
6378 AdminRevokeIp {} -> "POST /maintenance/sessions/revoke-ip"
6519 AdminRevokeIp {} -> "POST /maintenance/sessions/revoke-ip"
6379 GetRateLimits {} -> "GET /maintenance/rate-limits"
6520 GetRateLimits {} -> "GET /maintenance/rate-limits"
@@ -6423,6 +6564,7 @@ requestClientIp = \req -> case req of
6423 PingPeer {} -> ""
6564 PingPeer {} -> ""
6424 AdminListSessions {} -> ""
6565 AdminListSessions {} -> ""
6425 AdminRevokeSessions {} -> ""
6566 AdminRevokeSessions {} -> ""
6567+ AdminListAllSessions {} -> ""
6426 AdminPurgeMessages {} -> ""
6568 AdminPurgeMessages {} -> ""
6427 AdminRevokeIp {} -> ""
6569 AdminRevokeIp {} -> ""
6428 GetRateLimits {} -> ""
6570 GetRateLimits {} -> ""
@@ -6475,6 +6617,7 @@ serveInner = \req -> case req of
6475 PingPeer {server, authorization, respond} -> handlePingPeer server authorization respond
6617 PingPeer {server, authorization, respond} -> handlePingPeer server authorization respond
6476 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
6618 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
6477 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
6619 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
6620+ AdminListAllSessions {limit, authorization, respond} -> handleAdminListAllSessions limit authorization respond
6478 AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond
6621 AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond
6479 AdminRevokeIp {ip, authorization, respond} -> handleAdminRevokeIp ip authorization respond
6622 AdminRevokeIp {ip, authorization, respond} -> handleAdminRevokeIp ip authorization respond
6480 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
6623 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
@@ -6550,9 +6693,14 @@ backgroundPrune = do
6550 -- Bump server generation so clients reset their cursors (avoids stuck polls
6693 -- Bump server generation so clients reset their cursors (avoids stuck polls
6551 -- from messages with inflated receivedAt timestamps during the forward jump)
6694 -- from messages with inflated receivedAt timestamps during the forward jump)
6552 set *serverGeneration = t)
6695 set *serverGeneration = t)
6553− -- Forward jump detection: now is far ahead of last prune (more than 10x interval)
6696+ -- Forward jump detection: now is far ahead of last prune (more than 10x interval).
6697+ -- Bump server generation so clients reset their cursors: messages stored during the
6698+ -- jump window have inflated receivedAt timestamps that would appear as "already seen"
6699+ -- to clients with pre-jump cursors. Without a generation bump, these messages are
6700+ -- silently skipped until real time catches up to the inflated timestamps.
6554 when (lp > 0 && t - lp > backgroundPruneInterval * 10) (do
6701 when (lp > 0 && t - lp > backgroundPruneInterval * 10) (do
6555− logWarn ("clock forward jump detected — " ++ show (t - lp) ++ "ms since last prune (expected ~" ++ show backgroundPruneInterval ++ "ms), messages during this period may have inflated timestamps"))
6702+ logWarn ("clock forward jump detected — " ++ show (t - lp) ++ "ms since last prune (expected ~" ++ show backgroundPruneInterval ++ "ms), bumping server generation to reset client cursors")
6703+ set *serverGeneration = t)
6556 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead
6704 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead
6557 sa <- *startedAt
6705 sa <- *startedAt
6558 lastHb <- *lastHeartbeatAt
6706 lastHb <- *lastHeartbeatAt
@@ -6677,6 +6825,7 @@ backgroundSnapshot = do
6677 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
6825 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
6678 yield {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac}
6826 yield {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac}
6679 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
6827 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
6828+ println "[SNAPSHOT_END]"
6680 atomic do
6829 atomic do
6681 set *lastAutoSnapshotExportAt = t
6830 set *lastAutoSnapshotExportAt = t
6682 set *lastSnapshotFingerprint = fingerprint
6831 set *lastSnapshotFingerprint = fingerprint
@@ -6690,9 +6839,7 @@ backgroundSnapshot = do
6690 when (maintenanceSecret != "" && snapshotStalenessWarning > 0
6839 when (maintenanceSecret != "" && snapshotStalenessWarning > 0
6691 && staleCycles > 0 && pss > staleCycles
6840 && staleCycles > 0 && pss > staleCycles
6692 && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))
6841 && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))
6694− logError ("no snapshot exported in " ++ show pss ++ " prune cycles (~" ++ show (pss * backgroundPruneInterval / 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 ++ ")")
6842+ (logError ("no snapshot exported in " ++ show pss ++ " prune cycles (~" ++ show (pss * backgroundPruneInterval / 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 ++ "). ACTION REQUIRED: run GET /maintenance/snapshot to persist state before restart"))
6695− logError "ACTION REQUIRED: run GET /maintenance/snapshot to persist federation state before restart")
6696 else yield {}
6843 else yield {}
6697 backgroundSnapshot
6844 backgroundSnapshot