@@ -158,14 +158,14 @@ forwardNonceRetention : Int<Ms>
158forwardNonceRetention = messageMaxAge * 2 -- 60 days; extends 30 days beyond message dedup
158forwardNonceRetention = messageMaxAge * 2 -- 60 days; extends 30 days beyond message dedup
159-- Message reconciliation: full scan runs every Nth prune cycle or when cheap drift check detects mismatch.
159-- Message reconciliation: full scan runs every Nth prune cycle or when cheap drift check detects mismatch.
160-- Cheap check (sum rcBucket totals vs totalMessageCount) runs every cycle; full scan only when needed.
160-- Cheap check (sum rcBucket totals vs totalMessageCount) runs every cycle; full scan only when needed.
161−reconciliationFullCycleInterval = 10 -- full reconciliation every 10th prune cycle
161+reconciliationFullCycleInterval = 50 -- full reconciliation every 50th prune cycle (~25 min)
163-- Federation latency metric window
163-- Federation latency metric window
164federationLatencyWindowMs : Int<Ms>
164federationLatencyWindowMs : Int<Ms>
165federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
165federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
167-- Snapshot format version (increment on schema changes to prevent corrupt restores)
167-- Snapshot format version (increment on schema changes to prevent corrupt restores)
168−snapshotVersion = 6 -- bumped for checksum field in snapshot
168+snapshotVersion = 7 -- bumped for pendingAckCleanups field in snapshot
170-- Server build version: included in /health and /stats responses for deployment verification.
170-- Server build version: included in /health and /stats responses for deployment verification.
171-- Update this string when deploying a new build so operators can confirm which version is running.
171-- Update this string when deploying a new build so operators can confirm which version is running.
@@ -660,6 +660,10 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
660*federationLatencyCount : Int
660*federationLatencyCount : Int
661*federationLatencyMax : Int<Ms>
661*federationLatencyMax : Int<Ms>
663+-- Federation verify callback metrics (exposed in /stats, reset with latency window)
664+*slowVerifyCount : Int
665+*verifyFailCount : Int
663-- Server instance generation (epoch timestamp, for cursor staleness detection)
667-- Server instance generation (epoch timestamp, for cursor staleness detection)
664*serverGeneration : Int
668*serverGeneration : Int
@@ -912,12 +916,12 @@ route MaintenanceApi where
912 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
916 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
913 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
917 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
914 GET /maintenance/snapshot headers {authorization: Text}
918 GET /maintenance/snapshot headers {authorization: Text}
915− -> {ok: Bool, version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text]}
919+ -> {ok: Bool, version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]}
916 = Snapshot
920 = Snapshot
917 -- Restore: import a previously-exported snapshot. Server must be in drain mode
921 -- Restore: import a previously-exported snapshot. Server must be in drain mode
918 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
922 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
919 -- checksum: integrity check (0 = skip verification, for pre-v6 snapshots)
923 -- checksum: integrity check (0 = skip verification, for pre-v6 snapshots)
920− POST {version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text]} /maintenance/restore headers {authorization: Text}
924+ POST {version: Int, checksum: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text], pendingAckCleanups: [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]} /maintenance/restore headers {authorization: Text}
921 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
925 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
922 = Restore
926 = Restore
923 -- Block a federation peer (rejects all inbound/outbound federation with this server)
927 -- Block a federation peer (rejects all inbound/outbound federation with this server)
@@ -948,6 +952,10 @@ route MaintenanceApi where
948 POST {pubkey: PubkeyHex} /maintenance/messages/purge headers {authorization: Text}
952 POST {pubkey: PubkeyHex} /maintenance/messages/purge headers {authorization: Text}
949 -> {ok: Bool, purgedCount: Int}
953 -> {ok: Bool, purgedCount: Int}
950 = AdminPurgeMessages
954 = AdminPurgeMessages
955+ -- Revoke all sessions from a specific IP address (key-rotation abuse response)
956+ POST {ip: Text} /maintenance/sessions/revoke-ip headers {authorization: Text}
957+ -> {ok: Bool, revokedCount: Int, affectedKeys: Int}
951 -- Read current effective rate limits (compile-time defaults + runtime overrides)
959 -- Read current effective rate limits (compile-time defaults + runtime overrides)
952 GET /maintenance/rate-limits headers {authorization: Text}
960 GET /maintenance/rate-limits headers {authorization: Text}
953 -> {ok: Bool, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}
961 -> {ok: Bool, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int}
@@ -989,7 +997,7 @@ route HealthApi where
989-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
997-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
990route StatsApi where
998route StatsApi where
991 GET /stats headers {authorization: Text}
999 GET /stats headers {authorization: Text}
992− -> {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, 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}}
1000+ -> {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}}
994 GET /stats/peers headers {authorization: Text}
1002 GET /stats/peers headers {authorization: Text}
995 -> {peers: [PeerStatus], ok: Bool}
1003 -> {peers: [PeerStatus], ok: Bool}
@@ -1372,11 +1380,17 @@ verifyFederationCallback = \fromServer authorization scope ->
1372 result <- fetchFederation fromServer (VerifyToken {token: token, scope: scope})
1380 result <- fetchFederation fromServer (VerifyToken {token: token, scope: scope})
1373 endTime <- now
1381 endTime <- now
1374 let verifyLatency = endTime - startTime
1382 let verifyLatency = endTime - startTime
1375− when (verifyLatency > federationVerifyTimeoutMs)
1383+ when (verifyLatency > federationVerifyTimeoutMs) (do
1376− (logWarn ("federation: slow verify callback to " ++ fromServer ++ " (" ++ show verifyLatency ++ "ms, threshold: " ++ show federationVerifyTimeoutMs ++ "ms)"))
1384+ logWarn ("federation: slow verify callback to " ++ fromServer ++ " (" ++ show verifyLatency ++ "ms, threshold: " ++ show federationVerifyTimeoutMs ++ "ms)")
1386+ svc <- *slowVerifyCount
1387+ set *slowVerifyCount = svc + 1)
1377 case result of
1388 case result of
1378 Err e -> do
1389 Err e -> do
1379 logWarn ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
1390 logWarn ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
1392+ vfc <- *verifyFailCount
1393+ set *verifyFailCount = vfc + 1
1380 yield (False {})
1394 yield (False {})
1381 Ok r -> yield r.ok
1395 Ok r -> yield r.ok
@@ -2445,6 +2459,17 @@ countAllAcked : Int -> IO Int
2445countAllAcked = \b -> if b > 15 then yield 0 else do
2459countAllAcked = \b -> if b > 15 then yield 0 else do
2446 bucket <- readAckedBucket b; rest <- countAllAcked (b + 1); yield (count bucket + rest)
2460 bucket <- readAckedBucket b; rest <- countAllAcked (b + 1); yield (count bucket + rest)
2462+-- Per-bucket pruning for acked entries (avoids readAllAcked/writeAllAcked repartitioning).
2463+-- Each bucket is pruned in its own atomic block to reduce lock contention with poll-ack paths.
2464+pruneAllAckedBuckets : Int -> Timestamp -> IO ()
2465+pruneAllAckedBuckets = \b retainCutoff ->
2466+ if b > 15 then yield {}
2469+ bucket <- readAckedBucket b
2470+ writeAckedBucket b (do a <- bucket; where a.ackedAt > retainCutoff; yield a)
2471+ pruneAllAckedBuckets (b + 1) retainCutoff
2448-- Find the oldest (minimum) ackedAt timestamp across all 16 acked buckets.
2473-- Find the oldest (minimum) ackedAt timestamp across all 16 acked buckets.
2449-- Returns 0 if no acked entries exist. Avoids materializing all buckets into one list.
2474-- Returns 0 if no acked entries exist. Avoids materializing all buckets into one list.
2450computeOldestAckedAt : Int -> IO Timestamp
2475computeOldestAckedAt : Int -> IO Timestamp
@@ -2508,6 +2533,16 @@ countAllDelivered : Int -> IO Int
2508countAllDelivered = \b -> if b > 15 then yield 0 else do
2533countAllDelivered = \b -> if b > 15 then yield 0 else do
2509 bucket <- readDelivBucket b; rest <- countAllDelivered (b + 1); yield (count bucket + rest)
2534 bucket <- readDelivBucket b; rest <- countAllDelivered (b + 1); yield (count bucket + rest)
2536+-- Per-bucket pruning for delivered entries (avoids readAllDelivered/writeAllDelivered repartitioning).
2537+pruneAllDeliveredBuckets : Int -> Timestamp -> IO ()
2538+pruneAllDeliveredBuckets = \b retainCutoff ->
2539+ if b > 15 then yield {}
2542+ bucket <- readDelivBucket b
2543+ writeDelivBucket b (do d <- bucket; where d.deliveredAt > retainCutoff; yield d)
2544+ pruneAllDeliveredBuckets (b + 1) retainCutoff
2511-- ============================================
2546-- ============================================
2512-- Federation In-Flight Tracking
2547-- Federation In-Flight Tracking
2513-- ============================================
2548-- ============================================
@@ -2952,18 +2987,17 @@ pruneExpired = \t -> do
2952 -- Clear all forwards when federation is disabled (they can never be retried)
2987 -- Clear all forwards when federation is disabled (they can never be retried)
2953 writeAllForwards (if not federationEnabled then []
2988 writeAllForwards (if not federationEnabled then []
2954 else do f <- forwards; where f.createdAt > retainCutoff; yield f)
2989 else do f <- forwards; where f.createdAt > retainCutoff; yield f)
2955− -- AckedIds: contends with poll-ack (ackPolledMessages writes here), narrow the lock
2990+ -- AckedIds: per-bucket pruning (each bucket in its own atomic block to reduce lock
2956− atomic do
2991+ -- contention with poll-ack paths that write to ackedIds concurrently).
2957− ackedIds <- readAllAcked
2992+ pruneAllAckedBuckets 0 retainCutoff
2958− writeAllAcked (do a <- ackedIds; where a.ackedAt > retainCutoff; yield a)
2993+ -- DeliveredIds: per-bucket pruning (same pattern as acked)
2994+ pruneAllDeliveredBuckets 0 retainCutoff
2959− -- DeliveredIds + overflow + failures: smaller collections, grouped together
2995+ -- Overflow + failures: smaller collections, grouped together
2960 atomic do
2996 atomic do
2961 forwardFailures <- *forwardFailures
2997 forwardFailures <- *forwardFailures
2962 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f
2998 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f
2963 ackOverflow <- *ackOverflow
2999 ackOverflow <- *ackOverflow
2964 set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a
3000 set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a
2965− deliveredIds <- readAllDelivered
2966− writeAllDelivered (do d <- deliveredIds; where d.deliveredAt > retainCutoff; yield d)
2967 -- Prune rate limiters (split into smaller atomic blocks to reduce lock contention)
3001 -- Prune rate limiters (split into smaller atomic blocks to reduce lock contention)
2968 atomic do
3002 atomic do
2969 sendRates <- *sendRates
3003 sendRates <- *sendRates
@@ -3032,9 +3066,11 @@ pruneExpired = \t -> do
3032 -- Prune stale deferred ack cleanups (older than messageMaxAge are never useful)
3066 -- Prune stale deferred ack cleanups (older than messageMaxAge are never useful)
3033 pendingCleanups <- *pendingAckCleanups
3067 pendingCleanups <- *pendingAckCleanups
3034 set *pendingAckCleanups = do c <- pendingCleanups; where t - c.cleanupAfter < messageMaxAge; yield c
3068 set *pendingAckCleanups = do c <- pendingCleanups; where t - c.cleanupAfter < messageMaxAge; yield c
3035− -- Prune old sender expiration notifications (keep for one poll cycle + grace)
3069+ -- Prune old sender expiration notifications. Uses unackedMessageMaxAge (7 days)
3070+ -- instead of sessionExpiry (1 hour) so mobile clients that don't poll for days
3071+ -- still receive expiration notifications on their next poll.
3036 senderExp <- *senderExpirations
3072 senderExp <- *senderExpirations
3037− set *senderExpirations = do e <- senderExp; where t - e.expiredAt < sessionExpiry; yield e
3073+ set *senderExpirations = do e <- senderExp; where t - e.expiredAt < unackedMessageMaxAge; yield e
3038 -- Reconcile global federation connection counter with actual in-flight state
3074 -- Reconcile global federation connection counter with actual in-flight state
3039 -- (self-heals drift from leaked slots where a forked task died without decrementing)
3075 -- (self-heals drift from leaked slots where a forked task died without decrementing)
3040 inFlightRec <- *federationInFlight
3076 inFlightRec <- *federationInFlight
@@ -3488,7 +3524,18 @@ retryPendingForwards = \pending -> do
3488 blocked <- *blockedPeers
3524 blocked <- *blockedPeers
3489 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
3525 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
3490 let eligibleForwards = do f <- pending; where any eligibleServers (\s -> s == f.toServer); yield f
3526 let eligibleForwards = do f <- pending; where any eligibleServers (\s -> s == f.toServer); yield f
3491− sendForwardsPartitioned eligibleForwards eligibleServers maxForwardRetriesPerServer
3527+ -- Prioritize forwards for recipients with active polls (currently online).
3528+ -- These keys will receive the message sooner, improving delivery latency.
3529+ -- Forwards for offline keys are retried after online-key forwards complete.
3530+ activePolls <- *activePolls
3531+ let onlineForwards = do f <- eligibleForwards; where any activePolls (\p -> p.pubkey == f.toKey); yield f
3532+ let offlineForwards = do f <- eligibleForwards; where none activePolls (\p -> p.pubkey == f.toKey); yield f
3533+ let prioritizedForwards = union onlineForwards offlineForwards
3534+ -- Prioritize servers that have online-key forwards first
3535+ let onlineServers = collectPeerServers onlineForwards (\x -> x.toServer)
3536+ let offlineOnlyServers = do s <- eligibleServers; where none onlineServers (\os -> os == s); yield s
3537+ let prioritizedServers = union onlineServers offlineOnlyServers
3538+ sendForwardsPartitioned prioritizedForwards prioritizedServers maxForwardRetriesPerServer
3493-- When a remote key comes online, create forward entries for queued messages and send them
3540-- When a remote key comes online, create forward entries for queued messages and send them
3494forwardQueuedMessages = \targets t -> do
3541forwardQueuedMessages = \targets t -> do
@@ -3785,6 +3832,8 @@ pruneAndRetry = \t -> do
3785 set *federationLatencyTotal = 0
3832 set *federationLatencyTotal = 0
3786 set *federationLatencyCount = 0
3833 set *federationLatencyCount = 0
3787 set *federationLatencyMax = 0
3834 set *federationLatencyMax = 0
3835+ set *slowVerifyCount = 0
3836+ set *verifyFailCount = 0
3788 set *federationLatencyResetAt = t)
3837 set *federationLatencyResetAt = t)
3789 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
3838 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
3790 -- These were queued by handleFederationAck instead of fork+sleep.
3839 -- These were queued by handleFederationAck instead of fork+sleep.
@@ -4593,7 +4642,8 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
4593 (\senderKey -> do
4642 (\senderKey -> do
4594 lookupRateLimited <- atomic do
4643 lookupRateLimited <- atomic do
4595 lookupRates <- *lookupRates
4644 lookupRates <- *lookupRates
4645+ lkRate <- effectiveLookupRate
4596− let rateResult = checkPubkeyRate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
4646+ let rateResult = checkPubkeyRate lookupRates senderKey lkRate lookupRateWindowMs maxLookupRateEntries t
4597 set *lookupRates = rateResult.entries
4647 set *lookupRates = rateResult.entries
4598 yield rateResult.limited
4648 yield rateResult.limited
4599 if lookupRateLimited then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
4649 if lookupRateLimited then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
@@ -5056,6 +5106,13 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
5056-- Called by the origin server when the sender retracts a message via /messages/cancel.
5106-- Called by the origin server when the sender retracts a message via /messages/cancel.
5057-- Deletes the message from the recipient's mailbox if it hasn't been polled yet.
5107-- Deletes the message from the recipient's mailbox if it hasn't been polled yet.
5058-- If the message was already polled, returns cancelled=False (no-op).
5108-- If the message was already polled, returns cancelled=False (no-op).
5109+-- NOTE: Only the origin server (fromServer == message's originServer) can cancel.
5110+-- Relay cancels (where intermediate server B forwards A's cancel request) are not
5111+-- supported in the current protocol. If server A originates a message, forwards it to B,
5112+-- and B re-forwards to C, then A's /messages/cancel sends federation cancel to B and C
5113+-- directly (via presence/forward targets). B cannot relay the cancel to C on A's behalf
5114+-- because C would reject it (fromServer=B != originServer=A). This means cancel coverage
5115+-- depends on the origin server knowing all forward destinations at cancel time.
5059handleFederationCancel = \messageId toKey fromServer requestTimestamp correlationId authorization respond -> do
5116handleFederationCancel = \messageId toKey fromServer requestTimestamp correlationId authorization respond -> do
5060 t <- now
5117 t <- now
5061 let corrId = sanitizeCorrelationId correlationId
5118 let corrId = sanitizeCorrelationId correlationId
@@ -5139,8 +5196,9 @@ handleDrain = \phase authorization respond -> do
5139 fn <- readAllNonces
5196 fn <- readAllNonces
5140 bp <- *blockedPeers
5197 bp <- *blockedPeers
5141 ap <- *allowedPeers
5198 ap <- *allowedPeers
5199+ pac <- *pendingAckCleanups
5142 let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
5200 let drainCs = snapshotChecksum (count drainResult.localEntries) (count a) (count d) (count flushState.forwards) (count fn)
5143− 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}
5201+ 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}
5144 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
5202 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
5145 atomic do set *prunesSinceSnapshot = 0
5203 atomic do set *prunesSinceSnapshot = 0
5146 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
5204 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
@@ -5266,7 +5324,7 @@ computeAndCacheHealth = \t -> do
5266 pruneRun <- *pruneRunning
5324 pruneRun <- *pruneRunning
5267 dp <- *drainPhase
5325 dp <- *drainPhase
5268 let warmup = t - sa < healthPruneMaxAge
5326 let warmup = t - sa < healthPruneMaxAge
5269− let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
5327+ let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 5
5270 let capacityOk = tmc * 10 <= maxMessages * 9
5328 let capacityOk = tmc * 10 <= maxMessages * 9
5271 && presenceCount * 10 <= maxPresenceEntries * 9
5329 && presenceCount * 10 <= maxPresenceEntries * 9
5272 && forwardCount * 10 <= maxForwardsEntries * 9
5330 && forwardCount * 10 <= maxForwardsEntries * 9
@@ -5340,7 +5398,7 @@ handleHealth = \authorization respond -> do
5340 watchdogHealthy: True {},
5398 watchdogHealthy: True {},
5341 draining: result.isDraining})
5399 draining: result.isDraining})
5343−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, 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}}
5401+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}}
5345handleStats = \authorization respond -> do
5403handleStats = \authorization respond -> do
5346 t <- now
5404 t <- now
@@ -5398,8 +5456,10 @@ handleStats = \authorization respond -> do
5398 flt <- *federationLatencyTotal
5456 flt <- *federationLatencyTotal
5399 flc <- *federationLatencyCount
5457 flc <- *federationLatencyCount
5400 flm <- *federationLatencyMax
5458 flm <- *federationLatencyMax
5459+ svc <- *slowVerifyCount
5460+ vfc <- *verifyFailCount
5401 dca <- *drainCompletedAt
5461 dca <- *drainCompletedAt
5402− 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, dca: dca}
5462+ 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}
5403 -- Compute derived metrics
5463 -- Compute derived metrics
5404 let sessions = core.sc
5464 let sessions = core.sc
5405 let messages = core.tmc
5465 let messages = core.tmc
@@ -5420,7 +5480,7 @@ handleStats = \authorization respond -> do
5420 let maxFwdDepth = fwdStats.maxDepth
5480 let maxFwdDepth = fwdStats.maxDepth
5421 yield (respond {ok: True {}, version: serverBuildVersion,
5481 yield (respond {ok: True {}, version: serverBuildVersion,
5422 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},
5482 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},
5423− federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca, forwardServers: fwdServerCount, maxForwardQueueDepth: maxFwdDepth},
5483+ 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},
5424 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},
5484 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},
5425 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}})
5485 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}})
5426 else yield (respond zeroStats)
5486 else yield (respond zeroStats)
@@ -5501,7 +5561,7 @@ handleReady = \respond -> do
5501 let isReady = lastPrune > 0 && lastPrune >= sa
5561 let isReady = lastPrune > 0 && lastPrune >= sa
5502 yield (respond {ready: isReady})
5562 yield (respond {ready: isReady})
5504−emptySnapshot = {ok: False {}, version: 0, checksum: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: [], allowedPeers: []}
5564+emptySnapshot = {ok: False {}, version: 0, checksum: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: [], allowedPeers: [], pendingAckCleanups: []}
5506-- GET /maintenance/snapshot: Export federation metadata for persistence.
5566-- GET /maintenance/snapshot: Export federation metadata for persistence.
5507-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
5567-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
@@ -5522,27 +5582,28 @@ handleSnapshot = \authorization respond -> do
5522 fn <- readAllNonces
5582 fn <- readAllNonces
5523 bp <- *blockedPeers
5583 bp <- *blockedPeers
5524 ap <- *allowedPeers
5584 ap <- *allowedPeers
5585+ pac <- *pendingAckCleanups
5525− yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap}
5586+ yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac}
5526 atomic do
5587 atomic do
5527 set *lastSnapshotExportAt = t
5588 set *lastSnapshotExportAt = t
5528 set *prunesSinceSnapshot = 0
5589 set *prunesSinceSnapshot = 0
5529 let cs = snapshotChecksum (count result.presence) (count result.ackedIds) (count result.deliveredIds) (count result.forwards) (count result.forwardNonces)
5590 let cs = snapshotChecksum (count result.presence) (count result.ackedIds) (count result.deliveredIds) (count result.forwards) (count result.forwardNonces)
5530− logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards, " ++ show (count result.forwardNonces) ++ " forwardNonces, " ++ show (count result.blockedPeers) ++ " blockedPeers, " ++ show (count result.allowedPeers) ++ " allowedPeers, checksum=" ++ show cs)
5591+ logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards, " ++ show (count result.forwardNonces) ++ " forwardNonces, " ++ show (count result.blockedPeers) ++ " blockedPeers, " ++ show (count result.allowedPeers) ++ " allowedPeers, " ++ show (count result.pendingAckCleanups) ++ " pendingAckCleanups, checksum=" ++ show cs)
5531− yield (respond {ok: True {}, version: snapshotVersion, checksum: cs, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces, blockedPeers: result.blockedPeers, allowedPeers: result.allowedPeers})
5592+ yield (respond {ok: True {}, version: snapshotVersion, checksum: cs, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces, blockedPeers: result.blockedPeers, allowedPeers: result.allowedPeers, pendingAckCleanups: result.pendingAckCleanups})
5533-- POST /maintenance/restore: Import a previously-exported snapshot.
5594-- POST /maintenance/restore: Import a previously-exported snapshot.
5534-- Requires drain mode to prevent conflicts with live state. Merges imported data
5595-- Requires drain mode to prevent conflicts with live state. Merges imported data
5535-- with existing state (dedup entries are unioned, presence is merged by freshness).
5596-- with existing state (dedup entries are unioned, presence is merged by freshness).
5536-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
5597-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
5537-- trigger federation forward creation from messages + restored presence.
5598-- trigger federation forward creation from messages + restored presence.
5538−handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData blockedPeersData allowedPeersData authorization respond -> do
5599+handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData blockedPeersData allowedPeersData pendingAckCleanupsData authorization respond -> do
5539 let authorized = verifyMaintenanceAuth authorization
5600 let authorized = verifyMaintenanceAuth authorization
5540 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
5601 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
5542 -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers,
5603 -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers,
5543 -- version 3 = pre-ForwardEntry changes, version 4 = pre-allowedPeers, version 5 = pre-checksum
5604 -- version 3 = pre-ForwardEntry changes, version 4 = pre-allowedPeers, version 5 = pre-checksum
5544 -- (accept all for backwards compatibility; missing fields default to empty list or 0)
5605 -- (accept all for backwards compatibility; missing fields default to empty list or 0)
5545− if version != 0 && version != 1 && version != 2 && version != 3 && version != 4 && version != 5 && version != snapshotVersion then do
5606+ if version != 0 && version != 1 && version != 2 && version != 3 && version != 4 && version != 5 && version != 6 && version != snapshotVersion then do
5546 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
5607 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
5547 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
5608 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
@@ -5622,6 +5683,11 @@ handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData de
5622 let mergedAllowed = fold (\acc a -> if any acc (\x -> x == a) then acc else union [a] acc) allowed allowedPeersData
5683 let mergedAllowed = fold (\acc a -> if any acc (\x -> x == a) then acc else union [a] acc) allowed allowedPeersData
5623 let cappedAllowed = takeRelation maxAllowedPeers mergedAllowed
5684 let cappedAllowed = takeRelation maxAllowedPeers mergedAllowed
5624 set *allowedPeers = cappedAllowed
5685 set *allowedPeers = cappedAllowed
5686+ -- Restore pending ack cleanups (union with existing, cap at maxPendingAckCleanups)
5687+ -- pre-v7 snapshots have empty pendingAckCleanups (field defaults to [])
5688+ existingCleanups <- *pendingAckCleanups
5689+ let mergedCleanups = union existingCleanups pendingAckCleanupsData
5690+ set *pendingAckCleanups = takeRelation maxPendingAckCleanups mergedCleanups
5625 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
5691 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
5626 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
5692 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
5627 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
5693 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
@@ -5812,15 +5878,20 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
5812 set *totalMessageCount = max 0 (tc - 1)
5878 set *totalMessageCount = max 0 (tc - 1)
5813 rc <- readRcBucket b
5879 rc <- readRcBucket b
5814 writeRcBucket b (adjustRecipientCount rc to (0 - 1))
5880 writeRcBucket b (adjustRecipientCount rc to (0 - 1))
5815− -- Cancel pending forwards and collect servers that already received the message
5881+ -- Cancel pending forwards and collect their target servers before deleting
5816 fwdBkt <- readFwdBucket b
5882 fwdBkt <- readFwdBucket b
5817 let cancelledFwds = do f <- fwdBkt; where f.messageId == messageId && f.toKey == to; yield f
5883 let cancelledFwds = do f <- fwdBkt; where f.messageId == messageId && f.toKey == to; yield f
5884+ let forwardTargetServers = collectPeerServers cancelledFwds (\f -> f.toServer)
5818 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == to); yield f)
5885 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == to); yield f)
5819 -- Collect servers from recipient's presence that may have received the forward
5886 -- Collect servers from recipient's presence that may have received the forward
5820− -- (forwards that completed are no longer in fwdBucket, so check presence)
5887+ -- (completed forwards are no longer in fwdBucket, so check presence too)
5821 presenceBkt <- readPresenceBucket b
5888 presenceBkt <- readPresenceBucket b
5822 let remotePresence = collectPeerServers (presenceForKey presenceBkt to t) (\p -> p.server)
5889 let remotePresence = collectPeerServers (presenceForKey presenceBkt to t) (\p -> p.server)
5890+ -- Union forward targets (servers that haven't received it yet or already have it)
5891+ -- with presence servers (servers that already received and stored the message).
5892+ -- This ensures cancel reaches both pending and completed forward destinations.
5893+ let allCancelServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union [s] acc) remotePresence forwardTargetServers
5823− yield {cancelled: True {}, cancelServers: remotePresence}
5894+ yield {cancelled: True {}, cancelServers: allCancelServers}
5824 -- Send federation cancel to remote servers that may have the message
5895 -- Send federation cancel to remote servers that may have the message
5825 fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)
5896 fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)
5826 (sendToEach cancelResult.cancelServers (\server -> do
5897 (sendToEach cancelResult.cancelServers (\server -> do
@@ -6102,6 +6173,64 @@ handleAdminPurgeMessages = \pubkey authorization respond -> do
6102 retryEvictedAcks purgeResult.evictedEntries purgeResult.pendingAcks))
6173 retryEvictedAcks purgeResult.evictedEntries purgeResult.pendingAcks))
6103 yield (respond {ok: True {}, purgedCount: purgeResult.purgedCount})
6174 yield (respond {ok: True {}, purgedCount: purgeResult.purgedCount})
6176+-- POST /maintenance/sessions/revoke-ip: Revoke all sessions from a specific IP address.
6177+-- Useful for abuse response when key-rotation abuse is detected from a single IP
6178+-- (one IP authenticating many distinct pubkeys to bypass per-key rate limits).
6179+-- Scans all 16 session buckets, removes matching sessions, and gossips offline
6180+-- for any pubkey that lost its last session.
6181+handleAdminRevokeIp = \ip authorization respond -> do
6182+ let authorized = verifyMaintenanceAuth authorization
6183+ if not authorized then yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0})
6184+ else if ip == "" then yield (respond {ok: False {}, revokedCount: 0, affectedKeys: 0})
6187+ result <- revokeSessionsByIp ip 0 t {revokedCount: 0, affectedKeys: [], cleanups: []}
6188+ -- Gossip offline for pubkeys that lost their last session
6190+ offlines <- do c <- result.cleanups; where c.gossipOffline; yield c
6191+ when (count offlines > 0 && federationEnabled) (do
6192+ peerServers <- *knownPeerServers
6193+ when (count peerServers > 0) (do
6194+ let gossipEvents = do c <- offlines; yield {eventType: GossipOffline {}, pubkey: c.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
6195+ sendBatchedGossip peerServers gossipEvents gossipOriginHops)))
6196+ when (result.revokedCount > 0)
6197+ (logInfo ("admin: revoked " ++ show result.revokedCount ++ " sessions from IP " ++ ip ++ " (" ++ show (count result.affectedKeys) ++ " keys affected)"))
6198+ yield (respond {ok: True {}, revokedCount: result.revokedCount, affectedKeys: count result.affectedKeys})
6200+-- Per-bucket IP session revocation: scans one session bucket at a time, collecting
6201+-- affected pubkeys and checking for last-session cleanup needs.
6202+revokeSessionsByIp = \ip b t acc ->
6203+ if b > 15 then yield acc
6205+ bucketResult <- atomic do
6206+ bucket <- readSessionBucket b
6207+ let matching = do s <- bucket; where s.ip == ip; yield s
6208+ let matchCount = count matching
6209+ if matchCount == 0 then yield {removed: 0, pubkeys: []}
6211+ writeSessionBucket b (do s <- bucket; where s.ip != ip; yield s)
6212+ sc <- *totalSessionCount
6213+ set *totalSessionCount = max 0 (sc - matchCount)
6214+ -- Collect unique affected pubkeys from this bucket
6215+ let pks = fold (\a s -> if any a (\x -> x == s.pubkey) then a else union [s.pubkey] a) [] matching
6216+ yield {removed: matchCount, pubkeys: pks}
6217+ -- For each newly seen pubkey, check if it was their last session
6218+ let newKeys = do k <- bucketResult.pubkeys; where none acc.affectedKeys (\a -> a == k); yield k
6219+ cleanups <- checkCleanupForKeys newKeys t []
6220+ let newAcc = {revokedCount: acc.revokedCount + bucketResult.removed,
6221+ affectedKeys: fold (\a k -> if any a (\x -> x == k) then a else union [k] a) acc.affectedKeys bucketResult.pubkeys,
6222+ cleanups: union acc.cleanups cleanups}
6223+ revokeSessionsByIp ip (b + 1) t newAcc
6225+-- Check last-session cleanup for a list of pubkeys (used by revokeSessionsByIp)
6226+checkCleanupForKeys = \keys t acc ->
6227+ case headOf keys of
6228+ Nothing {} -> yield acc
6229+ Just {value: k} -> do
6230+ let rest = do x <- keys; where x != k; yield x
6231+ cleanup <- checkLastSessionCleanup k t
6232+ checkCleanupForKeys rest t (union acc [cleanup])
6105-- POST /maintenance/nonces/flush: Clear all forward nonces (recovery from corruption).
6234-- POST /maintenance/nonces/flush: Clear all forward nonces (recovery from corruption).
6106-- Nonces provide extended replay protection beyond message/ack dedup lifetime.
6235-- Nonces provide extended replay protection beyond message/ack dedup lifetime.
6107-- Flushing temporarily reopens a replay window for messages older than 30 days but
6236-- Flushing temporarily reopens a replay window for messages older than 30 days but
@@ -6190,6 +6319,7 @@ requestPath = \req -> case req of
6190 AdminListSessions {} -> "GET /maintenance/sessions"
6319 AdminListSessions {} -> "GET /maintenance/sessions"
6191 AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"
6320 AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"
6192 AdminPurgeMessages {} -> "POST /maintenance/messages/purge"
6321 AdminPurgeMessages {} -> "POST /maintenance/messages/purge"
6322+ AdminRevokeIp {} -> "POST /maintenance/sessions/revoke-ip"
6193 GetRateLimits {} -> "GET /maintenance/rate-limits"
6323 GetRateLimits {} -> "GET /maintenance/rate-limits"
6194 SetRateLimits {} -> "POST /maintenance/rate-limits"
6324 SetRateLimits {} -> "POST /maintenance/rate-limits"
6195 FlushNonces {} -> "POST /maintenance/nonces/flush"
6325 FlushNonces {} -> "POST /maintenance/nonces/flush"
@@ -6238,6 +6368,7 @@ requestClientIp = \req -> case req of
6238 AdminListSessions {} -> ""
6368 AdminListSessions {} -> ""
6239 AdminRevokeSessions {} -> ""
6369 AdminRevokeSessions {} -> ""
6240 AdminPurgeMessages {} -> ""
6370 AdminPurgeMessages {} -> ""
6371+ AdminRevokeIp {} -> ""
6241 GetRateLimits {} -> ""
6372 GetRateLimits {} -> ""
6242 SetRateLimits {} -> ""
6373 SetRateLimits {} -> ""
6243 FlushNonces {} -> ""
6374 FlushNonces {} -> ""
@@ -6281,7 +6412,7 @@ serveInner = \req -> case req of
6281 Shutdown {authorization, respond} -> handleShutdown authorization respond
6412 Shutdown {authorization, respond} -> handleShutdown authorization respond
6282 Undrain {authorization, respond} -> handleUndrain authorization respond
6413 Undrain {authorization, respond} -> handleUndrain authorization respond
6283 Snapshot {authorization, respond} -> handleSnapshot authorization respond
6414 Snapshot {authorization, respond} -> handleSnapshot authorization respond
6284− Restore {version, checksum, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, blockedPeers, allowedPeers, authorization, respond} -> handleRestore version checksum presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces blockedPeers allowedPeers authorization respond
6415+ Restore {version, checksum, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, blockedPeers, allowedPeers, pendingAckCleanups, authorization, respond} -> handleRestore version checksum presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces blockedPeers allowedPeers pendingAckCleanups authorization respond
6285 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
6416 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
6286 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
6417 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
6287 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
6418 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
@@ -6289,6 +6420,7 @@ serveInner = \req -> case req of
6289 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
6420 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
6290 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
6421 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
6291 AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond
6422 AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond
6423+ AdminRevokeIp {ip, authorization, respond} -> handleAdminRevokeIp ip authorization respond
6292 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
6424 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
6293 SetRateLimits {globalRequest, ipRate, sendRate, federationRate, authGlobal, recvRate, challengeRate, pairRate, lookupRate, clientWriteRate, authorization, respond} -> handleSetRateLimits globalRequest ipRate sendRate federationRate authGlobal recvRate challengeRate pairRate lookupRate clientWriteRate authorization respond
6425 SetRateLimits {globalRequest, ipRate, sendRate, federationRate, authGlobal, recvRate, challengeRate, pairRate, lookupRate, clientWriteRate, authorization, respond} -> handleSetRateLimits globalRequest ipRate sendRate federationRate authGlobal recvRate challengeRate pairRate lookupRate clientWriteRate authorization respond
6294 FlushNonces {authorization, respond} -> handleFlushNonces authorization respond
6426 FlushNonces {authorization, respond} -> handleFlushNonces authorization respond
@@ -6375,7 +6507,10 @@ backgroundPrune = do
6375 set *lastHeartbeatAt = t)
6507 set *lastHeartbeatAt = t)
6376 status <- atomic do
6508 status <- atomic do
6377 running <- *pruneRunning
6509 running <- *pruneRunning
6510+ -- 5x threshold: allows heavy prune cycles (reconciliation after restore, large nonce
6511+ -- pruning) to complete without triggering false stuck-recovery. Original 2x was too
6512+ -- tight and could cause spurious fork resets during legitimate long-running prune cycles.
6378− if running > 0 && t - running < backgroundPruneInterval * 2
6513+ if running > 0 && t - running < backgroundPruneInterval * 5
6379 then yield {canRun: False {}, recovered: False {}, gen: 0, stuckSince: 0}
6514 then yield {canRun: False {}, recovered: False {}, gen: 0, stuckSince: 0}
6381 set *pruneRunning = t
6516 set *pruneRunning = t
@@ -6482,8 +6617,9 @@ backgroundSnapshot = do
6482 fn <- readAllNonces
6617 fn <- readAllNonces
6483 bp <- *blockedPeers
6618 bp <- *blockedPeers
6484 ap <- *allowedPeers
6619 ap <- *allowedPeers
6620+ pac <- *pendingAckCleanups
6485 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
6621 let exportCs = snapshotChecksum (count p) (count a) (count d) (count f) (count fn)
6486− yield {version: snapshotVersion, checksum: exportCs, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap}
6622+ 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}
6487 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
6623 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
6488 atomic do
6624 atomic do
6489 set *lastAutoSnapshotExportAt = t
6625 set *lastAutoSnapshotExportAt = t