@@ -158,14 +158,14 @@ forwardNonceRetention : Int<Ms>
158158forwardNonceRetention = messageMaxAge * 2 -- 60 days; extends 30 days beyond message dedup
159159-- Message reconciliation: full scan runs every Nth prune cycle or when cheap drift check detects mismatch.
160160-- 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)
162162
163163-- Federation latency metric window
164164federationLatencyWindowMs : Int<Ms>
165165federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
166166
167167-- 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
169169
170170-- Server build version: included in /health and /stats responses for deployment verification.
171171-- 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
660660*federationLatencyCount : Int
661661*federationLatencyMax : Int<Ms>
662662
663+-- Federation verify callback metrics (exposed in /stats, reset with latency window)
664+*slowVerifyCount : Int
665+*verifyFailCount : Int
666+
663667-- Server instance generation (epoch timestamp, for cursor staleness detection)
664668*serverGeneration : Int
665669
@@ -912,12 +916,12 @@ route MaintenanceApi where
912916 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
913917 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
914918 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}]}
916920 = Snapshot
917921 -- Restore: import a previously-exported snapshot. Server must be in drain mode
918922 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
919923 -- 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}
921925 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
922926 = Restore
923927 -- Block a federation peer (rejects all inbound/outbound federation with this server)
@@ -948,6 +952,10 @@ route MaintenanceApi where
948952 POST {pubkey: PubkeyHex} /maintenance/messages/purge headers {authorization: Text}
949953 -> {ok: Bool, purgedCount: Int}
950954 = 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}
958+ = AdminRevokeIp
951959 -- Read current effective rate limits (compile-time defaults + runtime overrides)
952960 GET /maintenance/rate-limits headers {authorization: Text}
953961 -> {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
989997-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
990998route StatsApi where
991999 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}}
9931001 = Stats
9941002 GET /stats/peers headers {authorization: Text}
9951003 -> {peers: [PeerStatus], ok: Bool}
@@ -1372,11 +1380,17 @@ verifyFederationCallback = \fromServer authorization scope ->
13721380 result <- fetchFederation fromServer (VerifyToken {token: token, scope: scope})
13731381 endTime <- now
13741382 let verifyLatency = endTime - startTime
1375− when (verifyLatency > federationVerifyTimeoutMs)
1376− (logWarn ("federation: slow verify callback to " ++ fromServer ++ " (" ++ show verifyLatency ++ "ms, threshold: " ++ show federationVerifyTimeoutMs ++ "ms)"))
1383+ when (verifyLatency > federationVerifyTimeoutMs) (do
1384+ logWarn ("federation: slow verify callback to " ++ fromServer ++ " (" ++ show verifyLatency ++ "ms, threshold: " ++ show federationVerifyTimeoutMs ++ "ms)")
1385+ atomic do
1386+ svc <- *slowVerifyCount
1387+ set *slowVerifyCount = svc + 1)
13771388 case result of
13781389 Err e -> do
13791390 logWarn ("federation: verify callback to " ++ fromServer ++ " failed: " ++ show e)
1391+ atomic do
1392+ vfc <- *verifyFailCount
1393+ set *verifyFailCount = vfc + 1
13801394 yield (False {})
13811395 Ok r -> yield r.ok
13821396
@@ -2445,6 +2459,17 @@ countAllAcked : Int -> IO Int
24452459countAllAcked = \b -> if b > 15 then yield 0 else do
24462460 bucket <- readAckedBucket b; rest <- countAllAcked (b + 1); yield (count bucket + rest)
24472461
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 {}
2467+ else do
2468+ atomic do
2469+ bucket <- readAckedBucket b
2470+ writeAckedBucket b (do a <- bucket; where a.ackedAt > retainCutoff; yield a)
2471+ pruneAllAckedBuckets (b + 1) retainCutoff
2472+
24482473-- Find the oldest (minimum) ackedAt timestamp across all 16 acked buckets.
24492474-- Returns 0 if no acked entries exist. Avoids materializing all buckets into one list.
24502475computeOldestAckedAt : Int -> IO Timestamp
@@ -2508,6 +2533,16 @@ countAllDelivered : Int -> IO Int
25082533countAllDelivered = \b -> if b > 15 then yield 0 else do
25092534 bucket <- readDelivBucket b; rest <- countAllDelivered (b + 1); yield (count bucket + rest)
25102535
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 {}
2540+ else do
2541+ atomic do
2542+ bucket <- readDelivBucket b
2543+ writeDelivBucket b (do d <- bucket; where d.deliveredAt > retainCutoff; yield d)
2544+ pruneAllDeliveredBuckets (b + 1) retainCutoff
2545+
25112546-- ============================================
25122547-- Federation In-Flight Tracking
25132548-- ============================================
@@ -2952,18 +2987,17 @@ pruneExpired = \t -> do
29522987 -- Clear all forwards when federation is disabled (they can never be retried)
29532988 writeAllForwards (if not federationEnabled then []
29542989 else do f <- forwards; where f.createdAt > retainCutoff; yield f)
2955− -- AckedIds: contends with poll-ack (ackPolledMessages writes here), narrow the lock
2956− atomic do
2957− ackedIds <- readAllAcked
2958− writeAllAcked (do a <- ackedIds; where a.ackedAt > retainCutoff; yield a)
2959− -- DeliveredIds + overflow + failures: smaller collections, grouped together
2990+ -- AckedIds: per-bucket pruning (each bucket in its own atomic block to reduce lock
2991+ -- contention with poll-ack paths that write to ackedIds concurrently).
2992+ pruneAllAckedBuckets 0 retainCutoff
2993+ -- DeliveredIds: per-bucket pruning (same pattern as acked)
2994+ pruneAllDeliveredBuckets 0 retainCutoff
2995+ -- Overflow + failures: smaller collections, grouped together
29602996 atomic do
29612997 forwardFailures <- *forwardFailures
29622998 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f
29632999 ackOverflow <- *ackOverflow
29643000 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)
29673001 -- Prune rate limiters (split into smaller atomic blocks to reduce lock contention)
29683002 atomic do
29693003 sendRates <- *sendRates
@@ -3032,9 +3066,11 @@ pruneExpired = \t -> do
30323066 -- Prune stale deferred ack cleanups (older than messageMaxAge are never useful)
30333067 pendingCleanups <- *pendingAckCleanups
30343068 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.
30363072 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
30383074 -- Reconcile global federation connection counter with actual in-flight state
30393075 -- (self-heals drift from leaked slots where a forked task died without decrementing)
30403076 inFlightRec <- *federationInFlight
@@ -3488,7 +3524,18 @@ retryPendingForwards = \pending -> do
34883524 blocked <- *blockedPeers
34893525 let eligibleServers = do s <- servers; where shouldRetryServerEx s t failures inFlight blocked; yield s
34903526 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
34923539
34933540-- When a remote key comes online, create forward entries for queued messages and send them
34943541forwardQueuedMessages = \targets t -> do
@@ -3785,6 +3832,8 @@ pruneAndRetry = \t -> do
37853832 set *federationLatencyTotal = 0
37863833 set *federationLatencyCount = 0
37873834 set *federationLatencyMax = 0
3835+ set *slowVerifyCount = 0
3836+ set *verifyFailCount = 0
37883837 set *federationLatencyResetAt = t)
37893838 -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
37903839 -- These were queued by handleFederationAck instead of fork+sleep.
@@ -4593,7 +4642,8 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
45934642 (\senderKey -> do
45944643 lookupRateLimited <- atomic do
45954644 lookupRates <- *lookupRates
4596− let rateResult = checkPubkeyRate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
4645+ lkRate <- effectiveLookupRate
4646+ let rateResult = checkPubkeyRate lookupRates senderKey lkRate lookupRateWindowMs maxLookupRateEntries t
45974647 set *lookupRates = rateResult.entries
45984648 yield rateResult.limited
45994649 if lookupRateLimited then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
@@ -5056,6 +5106,13 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
50565106-- Called by the origin server when the sender retracts a message via /messages/cancel.
50575107-- Deletes the message from the recipient's mailbox if it hasn't been polled yet.
50585108-- 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.
50595116handleFederationCancel = \messageId toKey fromServer requestTimestamp correlationId authorization respond -> do
50605117 t <- now
50615118 let corrId = sanitizeCorrelationId correlationId
@@ -5139,8 +5196,9 @@ handleDrain = \phase authorization respond -> do
51395196 fn <- readAllNonces
51405197 bp <- *blockedPeers
51415198 ap <- *allowedPeers
5199+ pac <- *pendingAckCleanups
51425200 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}
51445202 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
51455203 atomic do set *prunesSinceSnapshot = 0
51465204 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
@@ -5266,7 +5324,7 @@ computeAndCacheHealth = \t -> do
52665324 pruneRun <- *pruneRunning
52675325 dp <- *drainPhase
52685326 let warmup = t - sa < healthPruneMaxAge
5269− let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
5327+ let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 5
52705328 let capacityOk = tmc * 10 <= maxMessages * 9
52715329 && presenceCount * 10 <= maxPresenceEntries * 9
52725330 && forwardCount * 10 <= maxForwardsEntries * 9
@@ -5340,7 +5398,7 @@ handleHealth = \authorization respond -> do
53405398 watchdogHealthy: True {},
53415399 draining: result.isDraining})
53425400
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}}
53445402
53455403handleStats = \authorization respond -> do
53465404 t <- now
@@ -5398,8 +5456,10 @@ handleStats = \authorization respond -> do
53985456 flt <- *federationLatencyTotal
53995457 flc <- *federationLatencyCount
54005458 flm <- *federationLatencyMax
5459+ svc <- *slowVerifyCount
5460+ vfc <- *verifyFailCount
54015461 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}
54035463 -- Compute derived metrics
54045464 let sessions = core.sc
54055465 let messages = core.tmc
@@ -5420,7 +5480,7 @@ handleStats = \authorization respond -> do
54205480 let maxFwdDepth = fwdStats.maxDepth
54215481 yield (respond {ok: True {}, version: serverBuildVersion,
54225482 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},
54245484 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},
54255485 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}})
54265486 else yield (respond zeroStats)
@@ -5501,7 +5561,7 @@ handleReady = \respond -> do
55015561 let isReady = lastPrune > 0 && lastPrune >= sa
55025562 yield (respond {ready: isReady})
55035563
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: []}
55055565
55065566-- GET /maintenance/snapshot: Export federation metadata for persistence.
55075567-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
@@ -5522,27 +5582,28 @@ handleSnapshot = \authorization respond -> do
55225582 fn <- readAllNonces
55235583 bp <- *blockedPeers
55245584 ap <- *allowedPeers
5525− yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap}
5585+ pac <- *pendingAckCleanups
5586+ yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap, pendingAckCleanups: pac}
55265587 atomic do
55275588 set *lastSnapshotExportAt = t
55285589 set *prunesSinceSnapshot = 0
55295590 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)
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})
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)
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})
55325593
55335594-- POST /maintenance/restore: Import a previously-exported snapshot.
55345595-- Requires drain mode to prevent conflicts with live state. Merges imported data
55355596-- with existing state (dedup entries are unioned, presence is merged by freshness).
55365597-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
55375598-- 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
55395600 let authorized = verifyMaintenanceAuth authorization
55405601 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
55415602 else do
55425603 -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers,
55435604 -- version 3 = pre-ForwardEntry changes, version 4 = pre-allowedPeers, version 5 = pre-checksum
55445605 -- (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
55465607 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
55475608 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
55485609 else do
@@ -5622,6 +5683,11 @@ handleRestore = \version checksumIn presenceData ackedIdsData ackOverflowData de
56225683 let mergedAllowed = fold (\acc a -> if any acc (\x -> x == a) then acc else union [a] acc) allowed allowedPeersData
56235684 let cappedAllowed = takeRelation maxAllowedPeers mergedAllowed
56245685 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
56255691 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
56265692 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
56275693 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
58125878 set *totalMessageCount = max 0 (tc - 1)
58135879 rc <- readRcBucket b
58145880 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
58165882 fwdBkt <- readFwdBucket b
58175883 let cancelledFwds = do f <- fwdBkt; where f.messageId == messageId && f.toKey == to; yield f
5884+ let forwardTargetServers = collectPeerServers cancelledFwds (\f -> f.toServer)
58185885 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == to); yield f)
58195886 -- 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)
58215888 presenceBkt <- readPresenceBucket b
58225889 let remotePresence = collectPeerServers (presenceForKey presenceBkt to t) (\p -> p.server)
5823− yield {cancelled: True {}, cancelServers: remotePresence}
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
5894+ yield {cancelled: True {}, cancelServers: allCancelServers}
58245895 -- Send federation cancel to remote servers that may have the message
58255896 fork (when (federationEnabled && cancelResult.cancelled && count cancelResult.cancelServers > 0)
58265897 (sendToEach cancelResult.cancelServers (\server -> do
@@ -6102,6 +6173,64 @@ handleAdminPurgeMessages = \pubkey authorization respond -> do
61026173 retryEvictedAcks purgeResult.evictedEntries purgeResult.pendingAcks))
61036174 yield (respond {ok: True {}, purgedCount: purgeResult.purgedCount})
61046175
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})
6185+ else do
6186+ t <- now
6187+ result <- revokeSessionsByIp ip 0 t {revokedCount: 0, affectedKeys: [], cleanups: []}
6188+ -- Gossip offline for pubkeys that lost their last session
6189+ fork (do
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})
6199+
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
6204+ else do
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: []}
6210+ else do
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
6224+
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])
6233+
61056234-- POST /maintenance/nonces/flush: Clear all forward nonces (recovery from corruption).
61066235-- Nonces provide extended replay protection beyond message/ack dedup lifetime.
61076236-- Flushing temporarily reopens a replay window for messages older than 30 days but
@@ -6190,6 +6319,7 @@ requestPath = \req -> case req of
61906319 AdminListSessions {} -> "GET /maintenance/sessions"
61916320 AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"
61926321 AdminPurgeMessages {} -> "POST /maintenance/messages/purge"
6322+ AdminRevokeIp {} -> "POST /maintenance/sessions/revoke-ip"
61936323 GetRateLimits {} -> "GET /maintenance/rate-limits"
61946324 SetRateLimits {} -> "POST /maintenance/rate-limits"
61956325 FlushNonces {} -> "POST /maintenance/nonces/flush"
@@ -6238,6 +6368,7 @@ requestClientIp = \req -> case req of
62386368 AdminListSessions {} -> ""
62396369 AdminRevokeSessions {} -> ""
62406370 AdminPurgeMessages {} -> ""
6371+ AdminRevokeIp {} -> ""
62416372 GetRateLimits {} -> ""
62426373 SetRateLimits {} -> ""
62436374 FlushNonces {} -> ""
@@ -6281,7 +6412,7 @@ serveInner = \req -> case req of
62816412 Shutdown {authorization, respond} -> handleShutdown authorization respond
62826413 Undrain {authorization, respond} -> handleUndrain authorization respond
62836414 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
62856416 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
62866417 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
62876418 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
@@ -6289,6 +6420,7 @@ serveInner = \req -> case req of
62896420 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
62906421 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
62916422 AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond
6423+ AdminRevokeIp {ip, authorization, respond} -> handleAdminRevokeIp ip authorization respond
62926424 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
62936425 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
62946426 FlushNonces {authorization, respond} -> handleFlushNonces authorization respond
@@ -6375,7 +6507,10 @@ backgroundPrune = do
63756507 set *lastHeartbeatAt = t)
63766508 status <- atomic do
63776509 running <- *pruneRunning
6378− if running > 0 && t - running < backgroundPruneInterval * 2
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.
6513+ if running > 0 && t - running < backgroundPruneInterval * 5
63796514 then yield {canRun: False {}, recovered: False {}, gen: 0, stuckSince: 0}
63806515 else do
63816516 set *pruneRunning = t
@@ -6482,8 +6617,9 @@ backgroundSnapshot = do
64826617 fn <- readAllNonces
64836618 bp <- *blockedPeers
64846619 ap <- *allowedPeers
6620+ pac <- *pendingAckCleanups
64856621 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}
64876623 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
64886624 atomic do
64896625 set *lastAutoSnapshotExportAt = t