@@ -274,6 +274,12 @@ maxSenderExpirations = 10000
274274maxSenderExpirationsPerKey = 100
275275-- Max pubkeys in a batch lookup request
276276maxBatchLookupKeys = 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
277283-- Health endpoint rate limiting (global, since /health has no clientIp)
278284maxHealthRate = 100 -- max health checks per window
279285healthRateWindowMs : Int<Ms>
@@ -534,6 +540,11 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
534540
535541*lastPresenceRefresh : Timestamp
536542
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
547+
537548*challengeRates : [ChallengeRateEntry]
538549
539550-- Separate revoke rate limiter (prevents challenge floods from exhausting revoke budget)
@@ -744,6 +755,9 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
744755-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
745756*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
746757
758+-- Cached successful verify callbacks per-server (short-lived, reduces outbound HTTP calls)
759+*verificationCache : [{server: Text, scope: Text, verifiedAt: Timestamp}]
760+
747761-- Per-IP unique key tracking: detects when a single IP authenticates many distinct pubkeys
748762-- (key-rotation abuse bypassing per-key rate limits). Pruned with other rate limiters.
749763*uniqueKeyIpRates : [UniqueKeyIpEntry]
@@ -765,6 +779,15 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
765779*rateOverrideLookupRate : Int
766780*rateOverrideClientWriteRate : Int
767781
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
786+*rateLimitedIp : Int
787+*rateLimitedAuth : Int
788+*rateLimitedPerKey : Int
789+*rateLimitedFederation : Int
790+
768791-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
769792-- Polls compare against this to skip expensive presence lookups when nothing changed.
770793*presenceVersion : Timestamp
@@ -958,6 +981,12 @@ route MaintenanceApi where
958981 POST {ip: Text} /maintenance/sessions/revoke-ip headers {authorization: Text}
959982 -> {ok: Bool, revokedCount: Int, affectedKeys: Int}
960983 = 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
961990 -- Read current effective rate limits (compile-time defaults + runtime overrides)
962991 GET /maintenance/rate-limits headers {authorization: Text}
963992 -> {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
9991028-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
10001029route StatsApi where
10011030 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}}
10031032 = Stats
10041033 GET /stats/peers headers {authorization: Text}
10051034 -> {peers: [PeerStatus], ok: Bool}
@@ -1086,9 +1115,19 @@ genToken = genTokenN tokenChunks
10861115tooLong : Text -> Int -> Bool
10871116tooLong = \s maxLen -> drop maxLen s != ""
10881117
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.
10901122sanitizeCorrelationId : Text -> Text
1091−sanitizeCorrelationId = \cid -> take maxCorrelationIdLen cid
1123+sanitizeCorrelationId = \cid -> stripControlChars (take maxCorrelationIdLen cid)
1124+
1125+stripControlChars : Text -> Text
1126+stripControlChars = \s ->
1127+ if s == "" then ""
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)
10921131
10931132isHexChar : Text -> Bool
10941133isHexChar = \c -> contains c "0123456789abcdef"
@@ -1264,7 +1303,8 @@ isBadServerName = \sn ->
12641303 || matchesHostnameOrSubdomain sn "internal"
12651304 || matchesHostnameOrSubdomain sn "metadata"
12661305 || 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
12681308 || take 2 sn == "0." || fc == "["
12691309 || is172Private sn
12701310 || take 7 sn == "198.18." || take 7 sn == "198.19."
@@ -1367,9 +1407,11 @@ generateFederationAuth = \scope server -> do
13671407 else updated
13681408 yield ("Bearer " ++ token)
13691409
1370−-- Verify an inbound federation token. First checks if a pre-shared key is configured
1371−-- for the sender — if so, verifies via constant-time comparison (no network round-trip).
1372−-- Otherwise falls back to callback verification via the sender's /federation/verify endpoint.
1410+-- Verify an inbound federation token. Checks a short-lived per-server cache first
1411+-- to reduce outbound HTTP calls (a peer sending N requests/second would otherwise
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.
13731415-- No retry: a failed verify rejects the request. The peer's federation rate limit + circuit breaker
13741416-- handle unreliable peers; retrying here doubles verify traffic and blocks with a 1s sleep per op,
13751417-- which a malicious peer can exploit for amplification.
@@ -1378,6 +1420,13 @@ verifyFederationCallback = \fromServer authorization scope ->
13781420 case extractToken authorization of
13791421 Nothing {} -> yield (False {})
13801422 Just {value: token} -> do
1423+ -- Check verify cache: skip the outbound HTTP call if we recently verified this server+scope
1424+ t <- now
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 {})
1429+ else do
13811430 startTime <- now
13821431 result <- fetchFederation fromServer (VerifyToken {token: token, scope: scope})
13831432 endTime <- now
@@ -1394,7 +1443,15 @@ verifyFederationCallback = \fromServer authorization scope ->
13941443 vfc <- *verifyFailCount
13951444 set *verifyFailCount = vfc + 1
13961445 yield (False {})
1397− Ok r -> yield r.ok
1446+ Ok r -> do
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)
1453+ else updated)
1454+ yield r.ok
13981455
13991456-- ============================================
14001457-- Session Bucket Helpers
@@ -1799,15 +1856,17 @@ transformAllPresenceBuckets = \f -> do
17991856 b14 <- *presenceBucket14; set *presenceBucket14 = f b14
18001857 b15 <- *presenceBucket15; set *presenceBucket15 = f b15
18011858
1802−-- Scan presence buckets one at a time for local entries nearing expiry.
1803−-- Avoids materializing all 16 buckets when few keys are near expiry.
1804−findNearExpiryPresence : Int -> Timestamp -> IO [PresenceEntry]
1805−findNearExpiryPresence = \b t ->
1806− if b > 15 then yield []
1859+-- Scan presence buckets for local entries nearing 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.
1863+findNearExpiryPresence : Int -> Int -> Timestamp -> IO [PresenceEntry]
1864+findNearExpiryPresence = \b endBucket t ->
1865+ if b > endBucket then yield []
18071866 else do
18081867 bucket <- readPresenceBucket b
18091868 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
18111870 yield (union nearExpiry rest)
18121871
18131872-- ============================================
@@ -3201,6 +3260,9 @@ checkFederationRate = \server t -> atomic do
32013260 fedRate <- effectiveFederationRate
32023261 let result = checkServerRate federationRates server fedRate federationRateWindowMs maxFederationRateEntries t
32033262 set *federationRates = result.entries
3263+ when result.limited (do
3264+ rlf <- *rateLimitedFederation
3265+ set *rateLimitedFederation = rlf + 1)
32043266 yield result.limited
32053267
32063268checkGlobalRate = \t -> atomic do
@@ -3213,7 +3275,10 @@ checkGlobalRate = \t -> atomic do
32133275 set *globalRateCount = 1
32143276 yield (False {})
32153277 else if rateCount >= effectiveRate
3216− then yield (True {})
3278+ then do
3279+ rlg <- *rateLimitedGlobal
3280+ set *rateLimitedGlobal = rlg + 1
3281+ yield (True {})
32173282 else do
32183283 set *globalRateCount = rateCount + 1
32193284 yield (False {})
@@ -3229,7 +3294,10 @@ checkAuthGlobalRate = \t -> atomic do
32293294 set *authGlobalRateCount = 1
32303295 yield (False {})
32313296 else if rateCount >= authRate
3232− then yield (True {})
3297+ then do
3298+ rla <- *rateLimitedAuth
3299+ set *rateLimitedAuth = rla + 1
3300+ yield (True {})
32333301 else do
32343302 set *authGlobalRateCount = rateCount + 1
32353303 yield (False {})
@@ -3257,6 +3325,9 @@ checkClientIpRate = \ip t ->
32573325 ipRate <- effectiveIpRate
32583326 let result = checkIpRate ipRates ip ipRate ipRateWindowMs maxIpRateEntries t
32593327 set *ipRates = result.entries
3328+ when result.limited (do
3329+ rli <- *rateLimitedIp
3330+ set *rateLimitedIp = rli + 1)
32603331 yield result.limited
32613332
32623333-- 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
33263397 if globalLimited
33273398 then do
33283399 logWarn ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)
3329− yield (respond {ok: False {}, error: "rate_limited"})
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:...".
3402+ yield (respond {ok: False {}, error: "rate_limited:" ++ show globalRateWindowMs ++ "ms"})
33303403 else do
33313404 rateLimited <- checkFederationRate fromServer t
33323405 if rateLimited
33333406 then do
33343407 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"})
33363409 else do
33373410 when clearOnAuth (clearFailure fromServer)
33383411 action
@@ -3663,7 +3736,9 @@ sendBatchedAcksToServer = \serverAcks server ->
36633736
36643737-- Send delivery acknowledgments back to origin servers and remove acked entries on success
36653738sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
3666−sendAcksAndClean = \pendingAcks -> do
3739+sendAcksAndClean = \pendingAcks ->
3740+ if count pendingAcks == 0 then yield {}
3741+ else do
36673742 let servers = collectPeerServers pendingAcks (\a -> a.originServer)
36683743 t <- now
36693744 failures <- *forwardFailures
@@ -3849,10 +3924,14 @@ pruneAndRetry = \t -> do
38493924 -- Early return: skip expensive reads on ~9/10 prune cycles
38503925 if not shouldRefresh then yield {extendedActive: [], nearExpiryActive: [], refreshPeers: [], shouldRefresh: False {}}
38513926 else do
3852− -- Optimized: scan presence buckets one at a time to find near-expiry local entries,
3853− -- then only read session buckets for the pubkeys found. Avoids materializing all 16
3854− -- presence + 16 session buckets when few keys are near expiry (common case).
3855− localNearExpiry <- findNearExpiryPresence 0 t
3927+ -- Round-robin: scan 4 of 16 presence buckets per refresh cycle.
3928+ -- Spreads the gossip traffic across 4 refresh intervals instead of
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
3934+ localNearExpiry <- findNearExpiryPresence startBucket endBucket t
38563935 if count localNearExpiry == 0 then do
38573936 knownPeers <- *knownPeerServers
38583937 set *lastPresenceRefresh = t
@@ -3883,6 +3962,15 @@ pruneAndRetry = \t -> do
38833962 set *federationLatencyMax = 0
38843963 set *slowVerifyCount = 0
38853964 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
38863974 set *federationLatencyResetAt = t)
38873975 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
38883976 -- These were queued by handleFederationAck instead of fork+sleep.
@@ -4446,9 +4534,12 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
44464534 relevantPresence <- readSelectedPresenceBuckets neededBuckets
44474535 yield (do key <- cappedWatchKeys; yield {pubkey: key, online: count (presenceForKey relevantPresence key t2) > 0})
44484536 else yield []
4449− fork (when federationEnabled (do
4450− sendAcksAndClean result.pendingAcks
4451− retryEvictedAcks result.evictedEntries result.pendingAcks))
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.
4539+ when (federationEnabled && (count result.pendingAcks > 0 || count result.evictedEntries > 0))
4540+ (fork (do
4541+ sendAcksAndClean result.pendingAcks
4542+ retryEvictedAcks result.evictedEntries result.pendingAcks))
44524543 let isHeartbeat = result.authorized && count result.events == 0
44534544 -- Deliver message expiration notifications: atomically read and drain expirations
44544545 -- for this sender's pubkey so each notification is delivered exactly once.
@@ -4698,7 +4789,10 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
46984789 yield rateResult.limited
46994790 if lookupRateLimited then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
47004791 else do
4701− let cappedKeys = takeRelation maxBatchLookupKeys pubkeys
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
4795+ let cappedKeys = takeRelation maxBatchLookupKeys dedupedKeys
47024796 result <- atomic do
47034797 let neededBuckets = uniqueBucketsFor cappedKeys (\k -> k)
47044798 relevantPresence <- readSelectedPresenceBuckets neededBuckets
@@ -4904,7 +4998,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
49044998 yield (False {})
49054999 if eventRateLimited then do
49065000 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"})
49085002 else do
49095003 let validated = validGossipEvents events t
49105004 let onlineEntries = validated.online
@@ -5141,15 +5235,23 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
51415235 -- Deferred origin message cleanup: only queue cleanup for messages that actually had
51425236 -- matching forward entries. Messages acked without a forward entry are NOT cleaned up,
51435237 -- preventing premature deletion when a peer acks messages it never actually received.
5144− when (ackedCount > 0) (atomic do
5145− cleanups <- *pendingAckCleanups
5146− 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
5148− let perKeyCapped = fold (\acc c ->
5149− let existingForKey = count (do x <- acc; where x.toKey == c.toKey; yield x)
5150− in if existingForKey >= maxPendingAckCleanupsPerKey then acc else union acc [c])
5151− cleanups newCleanups
5152− set *pendingAckCleanups = if count perKeyCapped > maxPendingAckCleanups then takeRelation maxPendingAckCleanups perKeyCapped else perKeyCapped)
5238+ when (ackedCount > 0) (do
5239+ capWarning <- atomic do
5240+ cleanups <- *pendingAckCleanups
5241+ let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
5242+ -- Per-key cap: limit cleanups per recipient to prevent one noisy key from dominating the queue
5243+ let perKeyCapped = fold (\acc c ->
5244+ let existingForKey = count (do x <- acc; where x.toKey == c.toKey; yield x)
5245+ in if existingForKey >= maxPendingAckCleanupsPerKey then acc else union acc [c])
5246+ cleanups newCleanups
5247+ let perKeyDropped = count newCleanups - (count perKeyCapped - count cleanups)
5248+ let globalCapped = count perKeyCapped > maxPendingAckCleanups
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")))
51535255 yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
51545256
51555257-- POST /federation/cancel: Cancel a previously forwarded message.
@@ -5249,7 +5351,9 @@ handleDrain = \phase authorization respond -> do
52495351 pac <- *pendingAckCleanups
52505352 let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
52515353 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
52525355 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
5356+ println "[SNAPSHOT_END]"
52535357 atomic do set *prunesSinceSnapshot = 0
52545358 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
52555359 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
@@ -5453,7 +5557,7 @@ handleHealth = \authorization respond -> do
54535557 watchdogHealthy: True {},
54545558 draining: result.isDraining})
54555559
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}}
54575561
54585562handleStats = \authorization respond -> do
54595563 t <- now
@@ -5514,7 +5618,12 @@ handleStats = \authorization respond -> do
55145618 svc <- *slowVerifyCount
55155619 vfc <- *verifyFailCount
55165620 dca <- *drainCompletedAt
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}
5621+ rlg <- *rateLimitedGlobal
5622+ rli <- *rateLimitedIp
5623+ rla <- *rateLimitedAuth
5624+ rlk <- *rateLimitedPerKey
5625+ rlf <- *rateLimitedFederation
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}
55185627 -- Compute derived metrics
55195628 let sessions = core.sc
55205629 let messages = core.tmc
@@ -5536,7 +5645,7 @@ handleStats = \authorization respond -> do
55365645 yield (respond {ok: True {}, version: serverBuildVersion,
55375646 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},
55385647 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},
55405649 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}})
55415650 else yield (respond zeroStats)
55425651
@@ -5896,6 +6005,14 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
58966005-- Only works for locally-originated messages where the caller's pubkey matches the
58976006-- stored senderKey. Federation forwards cannot be cancelled via this endpoint.
58986007-- 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).
58996016handleCancelMessage = \messageId to authorization clientIp respond -> do
59006017 t <- now
59016018 dp <- *drainPhase
@@ -6287,6 +6404,29 @@ checkCleanupForKeys = \keys t acc ->
62876404 cleanup <- checkLastSessionCleanup k t
62886405 checkCleanupForKeys rest t (union acc [cleanup])
62896406
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})
6413+ else do
6414+ let effectiveLimit = if limit <= 0 then 100 else clamp 1 500 limit
6415+ allSessions <- readAllSessions
6416+ t <- now
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}])
6424+ [] active
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})
6429+
62906430-- POST /maintenance/nonces/flush: Clear all forward nonces (recovery from corruption).
62916431-- Nonces provide extended replay protection beyond message/ack dedup lifetime.
62926432-- Flushing temporarily reopens a replay window for messages older than 30 days but
@@ -6374,6 +6514,7 @@ requestPath = \req -> case req of
63746514 PingPeer {} -> "POST /maintenance/ping"
63756515 AdminListSessions {} -> "GET /maintenance/sessions"
63766516 AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"
6517+ AdminListAllSessions {} -> "POST /maintenance/sessions/list"
63776518 AdminPurgeMessages {} -> "POST /maintenance/messages/purge"
63786519 AdminRevokeIp {} -> "POST /maintenance/sessions/revoke-ip"
63796520 GetRateLimits {} -> "GET /maintenance/rate-limits"
@@ -6423,6 +6564,7 @@ requestClientIp = \req -> case req of
64236564 PingPeer {} -> ""
64246565 AdminListSessions {} -> ""
64256566 AdminRevokeSessions {} -> ""
6567+ AdminListAllSessions {} -> ""
64266568 AdminPurgeMessages {} -> ""
64276569 AdminRevokeIp {} -> ""
64286570 GetRateLimits {} -> ""
@@ -6475,6 +6617,7 @@ serveInner = \req -> case req of
64756617 PingPeer {server, authorization, respond} -> handlePingPeer server authorization respond
64766618 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
64776619 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
6620+ AdminListAllSessions {limit, authorization, respond} -> handleAdminListAllSessions limit authorization respond
64786621 AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond
64796622 AdminRevokeIp {ip, authorization, respond} -> handleAdminRevokeIp ip authorization respond
64806623 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
@@ -6550,9 +6693,14 @@ backgroundPrune = do
65506693 -- Bump server generation so clients reset their cursors (avoids stuck polls
65516694 -- from messages with inflated receivedAt timestamps during the forward jump)
65526695 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.
65546701 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)
65566704 -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead
65576705 sa <- *startedAt
65586706 lastHb <- *lastHeartbeatAt
@@ -6677,6 +6825,7 @@ backgroundSnapshot = do
66776825 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
66786826 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}
66796827 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
6828+ println "[SNAPSHOT_END]"
66806829 atomic do
66816830 set *lastAutoSnapshotExportAt = t
66826831 set *lastSnapshotFingerprint = fingerprint
@@ -6690,9 +6839,7 @@ backgroundSnapshot = do
66906839 when (maintenanceSecret != "" && snapshotStalenessWarning > 0
66916840 && staleCycles > 0 && pss > staleCycles
66926841 && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))
6693− (do
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 ++ ")")
6695− logError "ACTION REQUIRED: run GET /maintenance/snapshot to persist federation state before restart")
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"))
66966843 else yield {}
66976844 backgroundSnapshot
66986845