@@ -185,6 +185,10 @@ reconciliationFullCycleInterval = 50 -- full reconciliation every 50th prune cyc
185185-- (sum of rcBucket totals vs totalMessageCount) exceeds this threshold. Small drifts from
186186-- concurrent bucket pruning self-correct on the next cycle. 0 = always reconcile on drift.
187187reconciliationDriftThreshold = 10
188+-- Lightweight reconciliation interval: presence/session/forward counter reconciliation
189+-- runs every Nth prune cycle. Between cycles, cached counters are trusted. Message
190+-- reconciliation has its own interval (reconciliationFullCycleInterval).
191+lightReconciliationInterval = 5
188192
189193-- Federation latency metric window
190194federationLatencyWindowMs : Int<Ms>
@@ -937,6 +941,14 @@ maxPeerLatencyEntries = 100
937941-- Used by health checks to avoid scanning all 16 forward buckets on every /health request.
938942*cachedForwardStaleCount : Int
939943
944+-- Cached adaptive rate scale factor (0-100%). Updated every prune cycle.
945+-- Avoids 3 STM reads per rate check on the hot path. 100 = no scaling (default).
946+*cachedAdaptiveScale : Int
947+
948+-- Gossip send metrics (windowed, reset with federation latency metrics)
949+*gossipSendSuccess : Int
950+*gossipSendFail : Int
951+
940952-- ============================================
941953-- Routes
942954-- ============================================
@@ -1183,7 +1195,7 @@ route HealthApi where
11831195-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
11841196route StatsApi where
11851197 GET /stats headers {authorization: Text}
1186− -> {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, topFailingPeers: [{server: Text, failures: Int}], pendingCancels: 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, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int}}
1198+ -> {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, messageBucketMax: Int, messageBucketMin: 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, topFailingPeers: [{server: Text, failures: Int}], pendingCancels: Int, gossipSuccess: Int, gossipFail: 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, senderDeliveries: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int}}
11871199 = Stats
11881200 GET /stats/peers headers {authorization: Text}
11891201 -> {peers: [PeerStatus], ok: Bool}
@@ -1671,16 +1683,31 @@ isBlockedPeer = \server blocked -> any blocked (\b -> b == server)
16711683isAllowedByAllowlist : Text -> [Text] -> Bool
16721684isAllowedByAllowlist = \server allowed -> count allowed == 0 || any allowed (\a -> a == server)
16731685
1686+-- Split text at the first colon. Returns {before, after}. Used to parse "Signed <sig>:<nonce>".
1687+-- Safe: signature hex and nonce are both [0-9a-f]+ (no colons), so exactly one colon separates them.
1688+splitOnColon : Text -> {before: Text, after: Text}
1689+splitOnColon = \s -> splitOnColonInner s 0
1690+
1691+splitOnColonInner : Text -> Int -> {before: Text, after: Text}
1692+splitOnColonInner = \s pos ->
1693+ if drop pos s == "" then {before: s, after: ""}
1694+ else if take 1 (drop pos s) == ":" then {before: take pos s, after: drop (pos + 1) s}
1695+ else splitOnColonInner s (pos + 1)
1696+
16741697-- Generate a single-use token for outbound federation auth.
16751698-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
16761699-- issued for one operation cannot authorize a different one.
1700+-- When signing is enabled, includes a random nonce in the signed message to prevent
1701+-- replay attacks (two requests with the same timestamp+scope produce different signatures).
1702+-- Format: "Signed <sig>:<nonce>" — backward compatible (old format has no colon).
16771703generateFederationAuth : Text -> Timestamp -> IO {network} Text
16781704generateFederationAuth = \scope requestTimestamp ->
16791705 if serverFederationPrivateKeyHex != ""
16801706 then do
1681− let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN)
1707+ nonce <- genTokenN 2
1708+ let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN ++ "|" ++ nonce)
16821709 let sig = sign (bytesFromHex serverFederationPrivateKeyHex) message
1683− yield ("Signed " ++ bytesToHex sig)
1710+ yield ("Signed " ++ bytesToHex sig ++ ":" ++ nonce)
16841711 else do
16851712 token <- genToken
16861713 t <- now
@@ -2701,6 +2728,37 @@ pruneNonceBucketRange = \b endB t ->
27012728 when (count pruned < count bucket) (writeNonceBucket b pruned)
27022729 pruneNonceBucketRange (b + 1) endB t
27032730
2731+-- Record a forward nonce and handle capacity eviction with LRU logging.
2732+-- Usable inside or outside atomic blocks (reads/writes nonce bucket refs).
2733+-- Returns eviction info for deferred logging when called inside atomic.
2734+updateForwardNonce : Int -> MessageIdHex -> PubkeyHex -> Timestamp -> IO {lruEvicted: Int, oldestRetainedAge: Int}
2735+updateForwardNonce = \nb messageId toKey t -> do
2736+ fwdNonces <- readNonceBucket nb
2737+ let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, seenAt: t}]
2738+ let evictedNonces = if count updatedNonces > maxNoncesPerBucket
2739+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
2740+ else updatedNonces
2741+ let lruEvicted = count updatedNonces - count evictedNonces
2742+ let oldestRetainedAge = if lruEvicted > 0
2743+ then case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> t - n.seenAt
2744+ else 0
2745+ writeNonceBucket nb evictedNonces
2746+ yield {lruEvicted: lruEvicted, oldestRetainedAge: oldestRetainedAge}
2747+
2748+-- Log nonce LRU eviction if any occurred. Call outside atomic blocks.
2749+logNonceEviction : Int -> Int -> Int -> IO ()
2750+logNonceEviction = \nb lruEvicted oldestRetainedAge ->
2751+ when (lruEvicted > 0) (do
2752+ let level = if oldestRetainedAge < messageMaxAge then "error" else "warn"
2753+ logKV level "nonce_lru_eviction" ["bucket=" ++ show nb, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket, "oldestRetainedMs=" ++ show oldestRetainedAge])
2754+
2755+-- Full record-and-log helper for use outside atomic blocks.
2756+recordForwardNonce : MessageIdHex -> PubkeyHex -> Timestamp -> IO ()
2757+recordForwardNonce = \messageId toKey t -> do
2758+ let nb = pubkeyBucket toKey
2759+ result <- updateForwardNonce nb messageId toKey t
2760+ logNonceEviction nb result.lruEvicted result.oldestRetainedAge
2761+
27042762-- ============================================
27052763-- Forward Bucket Helpers
27062764-- ============================================
@@ -2772,6 +2830,19 @@ countAllForwards = \b ->
27722830 rest <- countAllForwards (b + 1)
27732831 yield (count bucket + rest)
27742832
2833+-- Compute min and max message counts across all 16 buckets (for stats bucket skew detection).
2834+-- Returns {bucketMin, bucketMax}. O(16) bucket reads.
2835+messageBucketMinMax : IO {bucketMin: Int, bucketMax: Int}
2836+messageBucketMinMax = messageBucketMinMaxInner 0 999999999 0
2837+
2838+messageBucketMinMaxInner : Int -> Int -> Int -> IO {bucketMin: Int, bucketMax: Int}
2839+messageBucketMinMaxInner = \b currentMin currentMax ->
2840+ if b > 15 then yield {bucketMin: if currentMin == 999999999 then 0 else currentMin, bucketMax: currentMax}
2841+ else do
2842+ bucket <- readMessageBucket b
2843+ let c = count bucket
2844+ messageBucketMinMaxInner (b + 1) (min currentMin c) (max currentMax c)
2845+
27752846-- Compute forward stats per-bucket without materializing all forwards into one list.
27762847-- Returns total count, oldest age, unique server count, and max per-server queue depth.
27772848computeForwardStats : Int -> Timestamp -> IO {totalCount: Int, oldestAge: Int, serverCount: Int, maxDepth: Int, serverCounts: [{server: Text, count: Int}]}
@@ -3429,6 +3500,13 @@ hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey
34293500-- inflate effective presence lifetime by up to clockSkewTolerance ms. Clamping to now removes
34303501-- this inflation: the peer's declared TTL is honored, but the expiry clock starts from now,
34313502-- not from a future timestamp.
3503+-- Filter seenBy entries to only known peers + ourselves. Unknown entries are dropped
3504+-- to prevent gossip suppression attacks (a malicious peer injecting arbitrary server
3505+-- names into seenBy to block propagation to those servers).
3506+filterSeenBy : [Text] -> [Text] -> [Text]
3507+filterSeenBy = \seenBy knownPeers ->
3508+ do s <- seenBy; where s == serverHostN || any knownPeers (\p -> p == s); yield s
3509+
34323510validGossipEvents : [GossipEvent] -> Timestamp -> {online: [ValidGossipEntry], offline: [ValidGossipEntry]}
34333511validGossipEvents = \events t ->
34343512 let validate = \e ->
@@ -3517,15 +3595,9 @@ pruneExpired = \t -> do
35173595 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
35183596 -- Prune sessions and count in a single pass (avoids redundant 16-bucket traversal).
35193597 -- pruneAndCountAllSessions applies the filter to each bucket and sums the survivors.
3520− oldSessionCount <- *totalSessionCount
35213598 sessionCount <- pruneAndCountAllSessions (\bucket -> do x <- bucket; where x.expiresAt > t && t - x.createdAt < maxSessionLifetime; yield x)
35223599 atomic do
35233600 set *totalSessionCount = sessionCount
3524− -- Bump presenceVersion when sessions are pruned: if a pubkey's last session expired,
3525− -- its presence will be pruned in a subsequent cycle. Bumping here ensures polls
3526− -- detect the transition sooner (within one prune interval) rather than waiting for
3527− -- the presence entry's separate TTL to expire.
3528− when (sessionCount < oldSessionCount) (set *presenceVersion = t)
35293601 -- Prune messages across all 16 buckets concurrently. Each bucket prune is atomic
35303602 -- and touches only its own message bucket + rcBucket. The only shared write is
35313603 -- *totalMessageCount (a single Int), so STM contention is minimal.
@@ -3681,12 +3753,21 @@ forEachBatch = \items batchSize process ->
36813753 forEachBatch split.rest batchSize process
36823754
36833755sendBatchedGossipToServer : ServerName -> [GossipEvent] -> Int -> IO {network} {}
3684−sendBatchedGossipToServer = \server allEvents hops ->
3756+sendBatchedGossipToServer = \server allEvents hops -> do
3757+ -- Check backoff before sending: skip servers that recently failed (same backoff as forwards).
3758+ -- Without this, gossip to a dead peer generates unbounded log spam every refresh cycle.
3759+ t0 <- now
3760+ failures <- *forwardFailures
3761+ inFlight <- *federationInFlight
3762+ blocked <- *blockedPeers
3763+ if not (shouldRetryServerEx server t0 failures inFlight blocked) then yield {}
3764+ else do
36853765 -- Filter out events this server has already seen (via seenBy tracking).
36863766 -- Add ourselves and the target server to seenBy so downstream relays skip both.
3687− -- Cap seenBy at maxSeenByLen to bound per-event memory and bandwidth.
3688− let unseen = do e <- allEvents; where none e.seenBy (\s -> s == server); yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: takeRelation maxSeenByLen (union e.seenBy [server])}
3689− in if count unseen == 0 then yield {}
3767+ -- Cap seenBy at maxSeenByLen and filter to known peers (defense-in-depth against spoofed seenBy).
3768+ knownPeersForSeen <- *knownPeerServers
3769+ let unseen = do e <- allEvents; where none e.seenBy (\s -> s == server); yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: filterSeenBy (takeRelation maxSeenByLen (union e.seenBy [server])) knownPeersForSeen}
3770+ if count unseen == 0 then yield {}
36903771 else forEachBatch unseen maxGossipEvents (\batch -> do
36913772 t <- now
36923773 auth <- generateFederationAuth "gossip" t
@@ -3696,8 +3777,13 @@ sendBatchedGossipToServer = \server allEvents hops ->
36963777 events: batch, hopsRemaining: hops, fromServer: serverHostN,
36973778 requestTimestamp: t, correlationId: corrId, protocolVersion: federationProtocolVersion})
36983779 case result of
3699− Err e -> logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
3700− Ok _ -> yield {})
3780+ Err e -> do
3781+ logWarn ("federation: gossip to " ++ server ++ " failed: " ++ show e)
3782+ recordFailure server
3783+ atomic do gs <- *gossipSendFail; set *gossipSendFail = gs + 1
3784+ Ok _ -> do
3785+ clearFailure server
3786+ atomic do gs <- *gossipSendSuccess; set *gossipSendSuccess = gs + 1)
37013787
37023788-- Sends gossip events to each peer server. Per-server seenBy filtering happens inside
37033789-- sendBatchedGossipToServer: if all events for a server are already in its seenBy list,
@@ -3764,14 +3850,14 @@ adaptiveRateScale = \msgCount sessCount presCount ->
37643850
37653851-- Apply adaptive scaling to a base rate limit. Reads cached counters (O(1), no bucket scans)
37663852-- and scales the rate proportionally. Returns at least 1 to prevent division-by-zero in callers.
3853+-- Apply adaptive scaling to a base rate limit. Reads cached scale factor (O(1), updated
3854+-- every prune cycle) instead of 3 counter reads per call. Returns at least 1 to prevent
3855+-- division-by-zero in callers.
37673856adaptiveRate : Int -> IO Int
37683857adaptiveRate = \baseRate -> do
3769− scale <- atomic do
3770− tmc <- *totalMessageCount
3771− sc <- *totalSessionCount
3772− pc <- *totalPresenceCount
3773− yield (adaptiveRateScale tmc sc pc)
3774− yield (max 1 (baseRate * scale / 100))
3858+ scale <- *cachedAdaptiveScale
3859+ let effectiveScale = if scale <= 0 then 100 else scale
3860+ yield (max 1 (baseRate * effectiveScale / 100))
37753861
37763862-- Effective rate limit readers: return runtime override if set (> 0), else compile-time default.
37773863-- All per-key and global rates are scaled by adaptiveRate under high utilization.
@@ -3985,6 +4071,7 @@ fetchAndCachePeerKey = \server t -> do
39854071-- Verify an inbound Ed25519 signed federation request.
39864072-- Reconstructs the signed message from (scope, requestTimestamp, fromServer),
39874073-- fetches the peer's public key (cached), and verifies the signature.
4074+-- Verify signature without nonce (backward compatible with old peers).
39884075verifyFederationSignature : Text -> Text -> Text -> Timestamp -> IO {network} Bool
39894076verifyFederationSignature = \fromServer signatureHex scope requestTimestamp -> do
39904077 t <- now
@@ -3994,6 +4081,16 @@ verifyFederationSignature = \fromServer signatureHex scope requestTimestamp -> d
39944081 let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer)
39954082 yield (verify (bytesFromHex pubkeyHex) message (bytesFromHex signatureHex))
39964083
4084+-- Verify signature with nonce (new format: prevents replay of same-millisecond requests).
4085+verifyFederationSignatureWithNonce : Text -> Text -> Text -> Text -> Timestamp -> IO {network} Bool
4086+verifyFederationSignatureWithNonce = \fromServer signatureHex nonce scope requestTimestamp -> do
4087+ t <- now
4088+ pubkeyHex <- fetchPeerPublicKey fromServer t
4089+ if pubkeyHex == "" then yield (False {})
4090+ else do
4091+ let message = textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer ++ "|" ++ nonce)
4092+ yield (verify (bytesFromHex pubkeyHex) message (bytesFromHex signatureHex))
4093+
39974094-- Federation auth wrapper. Intentionally does NOT call checkPreAuth because:
39984095-- (a) Soft drain (phase 1) blocks new client auth but must not block federation —
39994096-- peers need to deliver messages and acks to existing recipients during drain.
@@ -4016,12 +4113,18 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
40164113 yield (respond {ok: False {}, error: "not_allowed"})
40174114 else do
40184115 -- Dispatch on auth scheme: "Signed " = Ed25519 signature, "Bearer " = verify callback
4116+ -- New format "Signed <sig>:<nonce>" includes a random nonce to prevent replay of
4117+ -- same-millisecond requests. Old format "Signed <sig>" (no colon) is backward compatible.
40194118 verified <- if take 7 authorization == "Signed "
40204119 then do
4021− let signatureHex = drop 7 authorization
4022− if signatureHex == "" || tooLong signatureHex maxSignatureLen
4120+ let payload = drop 7 authorization
4121+ if payload == "" || tooLong payload (maxSignatureLen + 13)
40234122 then yield (False {})
4024− else verifyFederationSignature fromServer signatureHex scope requestTimestamp
4123+ else
4124+ let parts = splitOnColon payload in
4125+ if parts.after != ""
4126+ then verifyFederationSignatureWithNonce fromServer parts.before parts.after scope requestTimestamp
4127+ else verifyFederationSignature fromServer payload scope requestTimestamp
40254128 else verifyFederationCallback fromServer authorization scope
40264129 if not verified
40274130 then yield (respond {ok: False {}, error: "unauthorized"})
@@ -4752,7 +4855,12 @@ waitForInFlightDrain = \deadline -> do
47524855 if activeCount <= 0 then yield {}
47534856 else retry
47544857 timedOut <- *drainTimedOut
4755− when timedOut (logWarn "drain: timed out waiting for in-flight federation tasks")
4858+ when timedOut (do
4859+ logWarn "drain: timed out waiting for in-flight federation tasks"
4860+ t2 <- now
4861+ inFlight <- *federationInFlight
4862+ let hungSlots = do r <- inFlight; where r.count > 0; yield r
4863+ when (count hungSlots > 0) (logStaleInFlight hungSlots t2))
47564864
47574865-- Full maintenance cycle: prune expired state, retry pending forwards and acks,
47584866-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
@@ -4773,6 +4881,8 @@ pruneAndRetry = \t -> do
47734881 sc <- *totalSessionCount
47744882 pc <- *totalPresenceCount
47754883 fc <- *totalForwardCount
4884+ -- Update cached adaptive rate scale (avoids 3 reads per rate check on hot path)
4885+ set *cachedAdaptiveScale = adaptiveRateScale tmc sc pc
47764886 yield {messages: tmc, sessions: sc, presence: pc, forwards: fc}
47774887 when (capacityCounts.messages * 4 >= maxMessages * 3)
47784888 (logEvent "warn" "capacity_warning" ("messages at " ++ show (capacityCounts.messages * 100 / maxMessages) ++ "% (" ++ show capacityCounts.messages ++ "/" ++ show maxMessages ++ ")"))
@@ -4782,32 +4892,33 @@ pruneAndRetry = \t -> do
47824892 (logEvent "warn" "capacity_warning" ("presence at " ++ show (capacityCounts.presence * 100 / maxPresenceEntries) ++ "% (" ++ show capacityCounts.presence ++ "/" ++ show maxPresenceEntries ++ ")"))
47834893 when (capacityCounts.forwards * 4 >= maxForwardsEntries * 3)
47844894 (logEvent "warn" "capacity_warning" ("forwards at " ++ show (capacityCounts.forwards * 100 / maxForwardsEntries) ++ "% (" ++ show capacityCounts.forwards ++ "/" ++ show maxForwardsEntries ++ ")"))
4785− -- Counter reconciliation: reconcile all counters every prune cycle.
4786− -- Presence and session counters are cheap (just counting list lengths across buckets).
4787− -- Message reconciliation is heavier (reads all messages, rebuilds per-recipient counts)
4788− -- but runs every cycle to minimize the drift window from concurrent bucket pruning.
4789− presDrift <- atomic do
4790− actual <- countAllPresence
4791− stored <- *totalPresenceCount
4792− set *totalPresenceCount = actual
4793− yield (actual - stored)
4794− sessDrift <- atomic do
4795− actual <- countAllSessions
4796− stored <- *totalSessionCount
4797− set *totalSessionCount = actual
4798− yield (actual - stored)
4799− fwdDrift <- atomic do
4800− actual <- countAllForwards 0
4801− stored <- *totalForwardCount
4802− set *totalForwardCount = actual
4803− yield (actual - stored)
4804− when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0)
4805− (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift))
4895+ -- Counter reconciliation: presence/session/forward counters reconciled every Nth cycle
4896+ -- (lightReconciliationInterval). Between cycles, cached counters are trusted — drift
4897+ -- from concurrent operations is typically small and self-corrects.
4898+ pc <- *pruneCount
4899+ let isLightReconcCycle = pc > 0 && (pc - (pc / lightReconciliationInterval) * lightReconciliationInterval) == 0
4900+ when isLightReconcCycle (do
4901+ presDrift <- atomic do
4902+ actual <- countAllPresence
4903+ stored <- *totalPresenceCount
4904+ set *totalPresenceCount = actual
4905+ yield (actual - stored)
4906+ sessDrift <- atomic do
4907+ actual <- countAllSessions
4908+ stored <- *totalSessionCount
4909+ set *totalSessionCount = actual
4910+ yield (actual - stored)
4911+ fwdDrift <- atomic do
4912+ actual <- countAllForwards 0
4913+ stored <- *totalForwardCount
4914+ set *totalForwardCount = actual
4915+ yield (actual - stored)
4916+ when (presDrift != 0 || sessDrift != 0 || fwdDrift != 0)
4917+ (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift ++ ", forwards: " ++ show fwdDrift)))
48064918 -- Message reconciliation: cheap check first (sum rcBucket .count fields vs totalMessageCount).
48074919 -- Full scan only when drift exceeds threshold or every Nth cycle.
48084920 -- Small drifts from concurrent bucket pruning self-correct; only trigger the expensive
48094921 -- full scan when drift is significant (> reconciliationDriftThreshold) or periodic.
4810− pc <- *pruneCount
48114922 let isFullCycle = pc > 0 && (pc - (pc / reconciliationFullCycleInterval) * reconciliationFullCycleInterval) == 0
48124923 cheapMsgDrift <- atomic do
48134924 rcTotal <- sumRcBucketTotals 0
@@ -4885,6 +4996,9 @@ pruneAndRetry = \t -> do
48854996 set *verificationCache = do c <- cache; where t - c.verifiedAt < verificationCacheMaxMs; yield c
48864997 -- Reset per-peer latency tracking with global window
48874998 set *peerLatencies = []
4999+ -- Reset gossip send metrics (windowed with federation metrics)
5000+ set *gossipSendSuccess = 0
5001+ set *gossipSendFail = 0
48885002 -- Reset client operation latency metrics (windowed with federation metrics)
48895003 set *clientPollLatencyTotal = 0
48905004 set *clientPollLatencyCount = 0
@@ -5050,7 +5164,6 @@ pollNewEvents = \pubkey clampedCursor t -> do
50505164 let selectedLow = takeRelation reservedSlots sortedLow
50515165 -- Merge and re-sort to maintain correct cursor advancement (lastReceivedAt must
50525166 -- reflect the actual latest receivedAt across both pools for cursor consistency).
5053− let allPollable = sortBy (\m -> (0 - m.priority) * 10000000000000 + m.receivedAt) raw
50545167 let limitedMsgs = sortBy (\m -> (0 - m.priority) * 10000000000000 + m.receivedAt) (union selectedHigh selectedLow)
50555168 let evts = do
50565169 m <- limitedMsgs
@@ -5627,25 +5740,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId expiresIn
56275740 -- if the message expires from messages/ackedIds/deliveredIds and a federation
56285741 -- peer forwards it back. Without this, the nonce check in handleForwardMessage
56295742 -- would miss locally-originated messages entirely.
5630− -- Sharded: writes only to the bucket for this recipient's pubkey.
5631− let nb = pubkeyBucket to
5632− fwdNonces <- readNonceBucket nb
5633− let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, seenAt: t}]
5634− let evictedNonces = if count updatedNonces > maxNoncesPerBucket
5635− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
5636− else updatedNonces
5637− -- Log LRU nonce eviction: evicted nonces create a bounded replay window
5638− -- where messages with those nonces could be re-accepted until natural expiry.
5639− -- oldestRetainedAge: age of the oldest surviving nonce (new dedup coverage floor).
5640− -- If > messageMaxAge, evicted nonces were in the extended-protection zone only
5641− -- (message/acked/delivered dedup still covers the gap). If <= messageMaxAge,
5642− -- some evicted nonces were still within primary dedup coverage (lower risk).
5643− let lruEvicted = count updatedNonces - count evictedNonces
5644− when (lruEvicted > 0) (do
5645− let oldestRetainedAge = case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> t - n.seenAt
5646− let level = if oldestRetainedAge < messageMaxAge then "error" else "warn"
5647− logKV level "nonce_lru_eviction" ["bucket=" ++ show nb, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket, "oldestRetainedMs=" ++ show oldestRetainedAge])
5648− writeNonceBucket nb evictedNonces
5743+ recordForwardNonce effectiveMsgId to t
56495744 let toPresence = presenceForKey recipientPresence to t
56505745 let localOnline = hasLocalPresence toPresence
56515746 let remoteServers = collectPeerServers toPresence (\p -> p.server)
@@ -6044,16 +6139,13 @@ handleVerifyToken = \token scope clientIp respond -> do
60446139 else do
60456140 found <- atomic do
60466141 pending <- *pendingVerifications
6047− -- Pre-filter by scope, expiry, and first hex char before running expensive safeCompare.
6048− -- Reduces scan from O(n) safeCompare calls to O(n/16) by filtering on token prefix.
6049− -- The prefix filter is not constant-time, but single-use tokens with 24 bytes of randomness
6050− -- make first-char leakage irrelevant (16^47 remaining possibilities in 30s window).
6051− let firstChar = take 1 token
6052− let isCandidate = \p -> p.scope == scope && t - p.createdAt < verificationTokenExpiry && take 1 p.token == firstChar
6142+ -- Filter by scope and expiry only — no token prefix optimization.
6143+ -- The previous first-char pre-filter leaked one hex digit via timing side-channel.
6144+ -- safeCompare must run on all candidates to prevent information leakage.
6145+ -- Bounded by maxPendingVerifications and further filtered by scope+expiry.
6146+ let isCandidate = \p -> p.scope == scope && t - p.createdAt < verificationTokenExpiry
60536147 let candidates = do p <- pending; where isCandidate p; yield p
60546148 let match = any candidates (\p -> safeCompare p.token token)
6055− -- Optimized removal: run safeCompare only on candidates (O(n/16)) instead of
6056− -- all pending (O(n)). Non-candidates are kept without constant-time comparison.
60576149 when match (set *pendingVerifications = union
60586150 (do p <- pending; where not (isCandidate p); yield p)
60596151 (do p <- candidates; where not (safeCompare p.token token); yield p))
@@ -6073,8 +6165,9 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
60736165 let cappedHops = clamp 0 maxGossipHops hopsRemaining
60746166 let corrId = sanitizeCorrelationId correlationId
60756167 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " gossip from " ++ fromServer))
6076− -- Cap inbound seenBy lists to prevent memory waste from malicious peers
6077− let cappedEvents = do e <- events; yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: takeRelation maxSeenByLen e.seenBy}
6168+ -- Cap inbound seenBy lists and filter to known peers to prevent gossip suppression attacks
6169+ knownPeersForSeenBy <- *knownPeerServers
6170+ let cappedEvents = do e <- events; yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl, seenBy: filterSeenBy (takeRelation maxSeenByLen e.seenBy) knownPeersForSeenBy}
60786171 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count cappedEvents <= maxGossipEvents) (True {}) (do
60796172 -- Per-peer gossip event volume limiting: counts total events (not just requests) per server.
60806173 -- A single gossip request carries up to maxGossipEvents entries; without this, a peer
@@ -6287,18 +6380,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
62876380 writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
62886381 signalMessageForKey toKey seqTs
62896382 -- Record forward nonce for replay protection beyond message/ack dedup lifetime.
6290− -- Sharded: writes only to the bucket for this recipient's pubkey.
6291− fwdNonces <- readNonceBucket bkt
6292− let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, seenAt: t}]
6293− let evictedNonces = if count updatedNonces > maxNoncesPerBucket
6294− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
6295− else updatedNonces
6296− let lruEvicted = count updatedNonces - count evictedNonces
6297− when (lruEvicted > 0) (do
6298− let oldestRetainedAge = case minBy evictedNonces (\n -> n.seenAt) of Nothing {} -> 0; Just {value: n} -> t - n.seenAt
6299− let level = if oldestRetainedAge < messageMaxAge then "error" else "warn"
6300− logKV level "nonce_lru_eviction" ["bucket=" ++ show bkt, "evicted=" ++ show lruEvicted, "capacity=" ++ show maxNoncesPerBucket, "oldestRetainedMs=" ++ show oldestRetainedAge])
6301− writeNonceBucket bkt evictedNonces
6383+ recordForwardNonce messageId toKey t
63026384 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
63036385 if result.noPresence then do
63046386 logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
@@ -7067,7 +7149,7 @@ handleHealth = \authorization respond -> do
70677149 draining: result.isDraining,
70687150 degradedPeers: 0})
70697151
7070−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, topFailingPeers: [], pendingCancels: 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, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0}}
7152+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, messageBucketMax: 0, messageBucketMin: 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, topFailingPeers: [], pendingCancels: 0, gossipSuccess: 0, gossipFail: 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, senderDeliveries: 0, rejectedGlobal: 0, rejectedIp: 0, rejectedAuth: 0, rejectedPerKey: 0, rejectedFederation: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0, totalEntries: 0, activeRequests: 0, peakActiveRequests: 0, pollAvgMs: 0, pollMaxMs: 0, pollCount: 0, sendAvgMs: 0, sendMaxMs: 0, sendCount: 0}}
70717153
70727154handleStats = \authorization respond -> do
70737155 t <- now
@@ -7082,6 +7164,7 @@ handleStats = \authorization respond -> do
70827164 -- Forward stats computed per-bucket to avoid materializing all 16 buckets into one list.
70837165 -- Each bucket is read independently; counts and per-server stats are aggregated.
70847166 fwdStats <- computeForwardStats 0 t
7167+ bucketStats <- messageBucketMinMax
70857168 core <- atomic do
70867169 sc <- *totalSessionCount
70877170 pc <- *totalPresenceCount
@@ -7143,7 +7226,9 @@ handleStats = \authorization respond -> do
71437226 cslt <- *clientSendLatencyTotal
71447227 cslc <- *clientSendLatencyCount
71457228 cslm <- *clientSendLatencyMax
7146− yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, svc: svc, vfc: vfc, dca: dca, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf, arq: arq, parq: parq, cplt: cplt, cplc: cplc, cplm: cplm, cslt: cslt, cslc: cslc, cslm: cslm}
7229+ gss <- *gossipSendSuccess
7230+ gsf <- *gossipSendFail
7231+ yield {sr: sr, cr: cr, fr: fr, fpr: fpr, grc: grc, ir: ir, lr: lr, rr: rr, prr: prr, pc: pc, pr: pr, hc: hc, sa: sa, lp: lp, lh: lh, dr: dr, flt: flt, flc: flc, flm: flm, svc: svc, vfc: vfc, dca: dca, rlg: rlg, rli: rli, rla: rla, rlk: rlk, rlf: rlf, arq: arq, parq: parq, cplt: cplt, cplc: cplc, cplm: cplm, cslt: cslt, cslc: cslc, cslm: cslm, gss: gss, gsf: gsf}
71477232 -- Compute derived metrics
71487233 let sessions = core.sc
71497234 let messages = core.tmc
@@ -7168,8 +7253,8 @@ handleStats = \authorization respond -> do
71687253 let pollAvg = if counters.cplc > 0 then counters.cplt / counters.cplc else 0
71697254 let sendAvg = if counters.cslc > 0 then counters.cslt / counters.cslc else 0
71707255 yield (respond {ok: True {}, version: serverBuildVersion,
7171− 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},
7172− 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, topFailingPeers: topFailingPeers, pendingCancels: count core.pfc},
7256+ 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, messageBucketMax: bucketStats.bucketMax, messageBucketMin: bucketStats.bucketMin},
7257+ 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, topFailingPeers: topFailingPeers, pendingCancels: count core.pfc, gossipSuccess: counters.gss, gossipFail: counters.gsf},
71737258 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, senderDeliveries: count core.sdl, rejectedGlobal: counters.rlg, rejectedIp: counters.rli, rejectedAuth: counters.rla, rejectedPerKey: counters.rlk, rejectedFederation: counters.rlf},
71747259 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, totalEntries: sessions + messages + presence + fwds + count core.c + core.aCount + count core.ao + core.dCount + core.fnCount + count core.ap + count core.fi, activeRequests: counters.arq, peakActiveRequests: counters.parq, pollAvgMs: pollAvg, pollMaxMs: counters.cplm, pollCount: counters.cplc, sendAvgMs: sendAvg, sendMaxMs: counters.cslm, sendCount: counters.cslc}})
71757260 else yield (respond zeroStats)
@@ -7539,13 +7624,13 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
75397624 bucket <- readMessageBucket b
75407625 let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == to; yield m)
75417626 case toCancel of
7542− Nothing {} -> yield {cancelled: False {}, cancelServers: []}
7627+ Nothing {} -> yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}
75437628 Just {value: msg} ->
75447629 -- Only allow cancelling locally-originated messages (originServer == "")
75457630 -- where the caller is the actual sender. Federation forwards have no
75467631 -- senderKey and should not be cancellable via this endpoint.
7547− if msg.originServer != "" then yield {cancelled: False {}, cancelServers: []}
7548− else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, cancelServers: []}
7632+ if msg.originServer != "" then yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}
7633+ else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, cancelServers: [], nonceLruEvicted: 0, nonceOldestAge: 0}
75497634 else do
75507635 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)
75517636 tc <- *totalMessageCount
@@ -7568,12 +7653,10 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
75687653 -- Record cancel nonce: prevents replayed federation forwards of the
75697654 -- cancelled message from being accepted after the message is removed
75707655 -- from all dedup sets (messages, ackedIds, deliveredIds).
7571− nonceBkt <- readNonceBucket b
7572− let updatedNonces = union nonceBkt [{messageId: messageId, toKey: to, seenAt: t}]
7573− writeNonceBucket b (if count updatedNonces > maxNoncesPerBucket
7574− then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
7575− else updatedNonces)
7576− yield {cancelled: True {}, cancelServers: allCancelServers}
7656+ nonceResult <- updateForwardNonce b messageId to t
7657+ yield {cancelled: True {}, cancelServers: allCancelServers, nonceLruEvicted: nonceResult.lruEvicted, nonceOldestAge: nonceResult.oldestRetainedAge}
7658+ -- Log nonce eviction outside atomic block (println can't run inside STM)
7659+ logNonceEviction (pubkeyBucket to) cancelResult.nonceLruEvicted cancelResult.nonceOldestAge
75777660 -- Send federation batch cancel to remote servers that may have the message.
75787661 -- Uses FederationBatchCancel (single-entry batch) instead of FederationCancel
75797662 -- for code path consistency with handleCancelBatch.