@@ -134,6 +134,7 @@ maxRecentFederationRequests = 1000
134134-- Forward nonce dedup: tracks (messageId, toKey) pairs to catch replays
135135-- beyond the lifetime of messages/ackedIds/deliveredIds
136136maxForwardNonces = 100000
137+maxNoncesPerBucket = maxForwardNonces / 16 + 1 -- per-bucket cap for 16-way sharded nonces
137138forwardNonceRetention : Int<Ms>
138139forwardNonceRetention = messageMaxAge * 2 -- 60 days; extends 30 days beyond message dedup
139140-- Message reconciliation: full scan runs every Nth prune cycle or when cheap drift check detects mismatch.
@@ -848,7 +849,7 @@ route HealthApi where
848849-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
849850route StatsApi where
850851 GET /stats headers {authorization: Text}
851− -> {ok: Bool, 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}, 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}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool}}
852+ -> {ok: Bool, 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}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool}}
852853 = Stats
853854 GET /stats/peers headers {authorization: Text}
854855 -> {peers: [PeerStatus], ok: Bool}
@@ -2786,7 +2787,7 @@ checkPreAuth = \clientIp t -> do
27862787 if ipLimited then yield "ip_rate_limited"
27872788 else yield ""
27882789
2789−withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do
2790+withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition clearOnAuth action -> do
27902791 -- Check blocklist before spending resources on verification callback
27912792 blocked <- *blockedPeers
27922793 if isBlockedPeer fromServer blocked
@@ -2835,7 +2836,7 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
28352836 logWarn ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
28362837 yield (respond {ok: False {}, error: "rate_limited"})
28372838 else do
2838− clearFailure fromServer
2839+ when clearOnAuth (clearFailure fromServer)
28392840 action
28402841
28412842withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
@@ -2851,6 +2852,40 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
28512852 then onUnauth
28522853 else onAuth pubkey
28532854
2855+-- Check if a pubkey has no remaining sessions and, if so, clean up local presence.
2856+-- Must be called inside an atomic block after the session has already been removed.
2857+-- Returns {pubkey, gossipOffline, peerServers} indicating whether offline gossip is needed.
2858+checkLastSessionCleanup : PubkeyHex -> Timestamp -> IO {pubkey: PubkeyHex, gossipOffline: Bool, peerServers: [Text]}
2859+checkLastSessionCleanup = \pubkey t -> do
2860+ hasRemaining <- hasSessionForKey pubkey 0
2861+ if not hasRemaining
2862+ then do
2863+ let pb = pubkeyBucket pubkey
2864+ presenceBkt <- readPresenceBucket pb
2865+ let kept = do p <- presenceBkt; where p.pubkey != pubkey || not (isLocalServer p.server); yield p
2866+ writePresenceBucket pb kept
2867+ let removedCount = count presenceBkt - count kept
2868+ tpc <- *totalPresenceCount
2869+ set *totalPresenceCount = max 0 (tpc - removedCount)
2870+ set *presenceVersion = t
2871+ peerServers <- *knownPeerServers
2872+ yield {pubkey: pubkey, gossipOffline: True {}, peerServers: peerServers}
2873+ else yield {pubkey: pubkey, gossipOffline: False {}, peerServers: []}
2874+
2875+-- If the last session was removed, gossip offline to federation peers.
2876+-- Skips gossip during drain (handleDrain already broadcast offline for all local keys).
2877+gossipOfflineIfLastSession : {pubkey: PubkeyHex, gossipOffline: Bool, peerServers: [Text]} -> Text -> IO {network} ()
2878+gossipOfflineIfLastSession = \r label -> do
2879+ when r.gossipOffline (do
2880+ t <- now
2881+ isDraining <- *draining
2882+ if isDraining
2883+ then logInfo (label ++ ": last session for " ++ r.pubkey ++ " (drain active, skipping redundant offline gossip)")
2884+ else do
2885+ logInfo (label ++ ": last session for " ++ r.pubkey ++ ", gossiping offline")
2886+ when (federationEnabled && count r.peerServers > 0)
2887+ (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
2888+
28542889removeForward = \mid srv toKey -> atomic do
28552890 let b = pubkeyBucket toKey
28562891 fwdBkt <- readFwdBucket b
@@ -3876,9 +3911,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
38763911 let nb = pubkeyBucket to
38773912 fwdNonces <- readNonceBucket nb
38783913 let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, originServer: "", seenAt: t}]
3879− let maxPerBucket = maxForwardNonces / 16 + 1
3880− writeNonceBucket nb (if count updatedNonces > maxPerBucket
3881− then evictNonces updatedNonces maxPerBucket forwardNonceRetention t
3914+ writeNonceBucket nb (if count updatedNonces > maxNoncesPerBucket
3915+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
38823916 else updatedNonces)
38833917 let toPresence = presenceForKey recipientPresence to t
38843918 let localOnline = hasLocalPresence toPresence
@@ -4025,6 +4059,9 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
40254059-- "delivered" (acked by recipient), "unknown" (expired or never existed).
40264060handleMessageStatus = \to messageId authorization clientIp respond -> do
40274061 t <- now
4062+ draining <- *draining
4063+ if draining then yield (respond {status: "", authorized: False {}})
4064+ else do
40284065 globalRateLimited <- checkGlobalRate t
40294066 if globalRateLimited then yield (respond {status: "", authorized: False {}})
40304067 else
@@ -4097,32 +4134,13 @@ handleRevoke = \targetToken authorization clientIp respond -> do
40974134 writeSessionBucket targetBucket (do s <- sessions; where not (safeCompare s.token targetToken && s.pubkey == callerPubkey); yield s)
40984135 sc <- *totalSessionCount
40994136 set *totalSessionCount = max 0 (sc - 1)
4100− -- Check if this was the last session for this key (same logic as handleLogout)
4101− hasRemaining <- hasSessionForKey callerPubkey 0
4102− if not hasRemaining
4103− then do
4104− let pb = pubkeyBucket callerPubkey
4105− presenceBkt <- readPresenceBucket pb
4106− writePresenceBucket pb (do p <- presenceBkt; where p.pubkey != callerPubkey || not (isLocalServer p.server); yield p)
4107− let removedCount = count presenceBkt - count (do p <- presenceBkt; where p.pubkey != callerPubkey || not (isLocalServer p.server); yield p)
4108− tpc <- *totalPresenceCount
4109− set *totalPresenceCount = max 0 (tpc - removedCount)
4110− set *presenceVersion = t
4111− peerServers <- *knownPeerServers
4112− yield (Just {value: {pubkey: callerPubkey, gossipOffline: True {}, peerServers: peerServers}})
4113− else yield (Just {value: {pubkey: callerPubkey, gossipOffline: False {}, peerServers: []}})
4137+ cleanup <- checkLastSessionCleanup callerPubkey t
4138+ yield (Just {value: cleanup})
41144139 else yield (Nothing {})
41154140 case result of
41164141 Nothing {} -> yield (respond {ok: False {}, authorized: True {}})
41174142 Just {value: r} -> do
4118− when r.gossipOffline (do
4119− isDraining <- *draining
4120− if isDraining
4121− then logInfo ("revoke: last session for " ++ r.pubkey ++ " (drain active, skipping redundant offline gossip)")
4122− else do
4123− logInfo ("revoke: last session for " ++ r.pubkey ++ ", gossiping offline")
4124− when (federationEnabled && count r.peerServers > 0)
4125− (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
4143+ gossipOfflineIfLastSession r "revoke"
41264144 when (not r.gossipOffline) (logInfo ("revoke: session revoked for " ++ r.pubkey))
41274145 yield (respond {ok: True {}, authorized: True {}}))
41284146
@@ -4153,37 +4171,12 @@ handleLogout = \authorization clientIp respond -> do
41534171 writeSessionBucket bucket (do s <- sessions; where not (safeCompare s.token token); yield s)
41544172 sc <- *totalSessionCount
41554173 set *totalSessionCount = max 0 (sc - 1)
4156− -- Check if this was the last session for this key.
4157− -- Short-circuit scan: reads buckets one at a time, stopping as soon as
4158− -- a matching session is found. Avoids reading all 16 buckets when the
4159− -- user has sessions in other buckets (common case: multi-device).
4160− hasRemaining <- hasSessionForKey pubkey 0
4161− if not hasRemaining
4162− then do
4163− -- Remove local presence immediately (only this key's bucket)
4164− let pb = pubkeyBucket pubkey
4165− presenceBkt <- readPresenceBucket pb
4166− writePresenceBucket pb (do p <- presenceBkt; where p.pubkey != pubkey || not (isLocalServer p.server); yield p)
4167− let removedCount = count presenceBkt - count (do p <- presenceBkt; where p.pubkey != pubkey || not (isLocalServer p.server); yield p)
4168− tpc <- *totalPresenceCount
4169− set *totalPresenceCount = max 0 (tpc - removedCount)
4170− set *presenceVersion = t
4171− peerServers <- *knownPeerServers
4172− yield (Just {value: {pubkey: pubkey, gossipOffline: True {}, peerServers: peerServers}})
4173− else yield (Just {value: {pubkey: pubkey, gossipOffline: False {}, peerServers: []}})
4174+ cleanup <- checkLastSessionCleanup pubkey t
4175+ yield (Just {value: cleanup})
41744176 case result of
41754177 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
41764178 Just {value: r} -> do
4177− when r.gossipOffline (do
4178− -- Skip offline gossip during drain: handleDrain already broadcast offline for all local keys,
4179− -- so per-key gossip here would be a redundant federation round-trip.
4180− isDraining <- *draining
4181− if isDraining
4182− then logInfo ("logout: last session for " ++ r.pubkey ++ " (drain active, skipping redundant offline gossip)")
4183− else do
4184− logInfo ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")
4185− when (federationEnabled && count r.peerServers > 0)
4186− (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
4179+ gossipOfflineIfLastSession r "logout"
41874180 yield (respond {ok: True {}, authorized: True {}})
41884181
41894182-- POST /federation/verify: Confirm a token was issued by this server.
@@ -4227,7 +4220,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
42274220 let cappedHops = clamp 0 maxGossipHops hopsRemaining
42284221 let corrId = if correlationId != "" then correlationId else ""
42294222 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " gossip from " ++ fromServer))
4230− withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
4223+ withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (True {}) (do
42314224 -- Per-peer gossip event volume limiting: counts total events (not just requests) per server.
42324225 -- A single gossip request carries up to maxGossipEvents entries; without this, a peer
42334226 -- sending 50 req/s * 100 events = 5000 presence mutations/second.
@@ -4263,18 +4256,26 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
42634256 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
42644257 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
42654258 let deduped = mergeByPubkeyServer [] filteredOnline
4266− -- Warn when a peer claims keys that are locally authenticated (potential presence hijack)
4259+ -- Reject first-party gossip entries that claim keys locally authenticated on this server.
4260+ -- A peer announcing presence for a key we have a live session for could be a presence
4261+ -- hijack attempt: the attacker's server would receive forwarded message copies.
4262+ -- Filter these out before the merge to prevent duplicate routing.
42674263 localClashes <- atomic do
42684264 let firstParty = do e <- deduped; where e.server == fromServer; yield e
42694265 let targetBuckets = uniqueBucketsFor firstParty (\e -> e.pubkey)
42704266 localPresence <- readSelectedPresenceBuckets targetBuckets
42714267 yield (do e <- firstParty; where any localPresence (\p -> p.pubkey == e.pubkey && isLocalServer p.server && p.expiresAt > t); yield e.pubkey)
42724268 when (count localClashes > 0)
4273− (logWarn ("gossip: " ++ fromServer ++ " claims " ++ show (count localClashes) ++ " locally-authenticated key(s) — ignoring for local routing"))
4269+ (logWarn ("gossip: " ++ fromServer ++ " claims " ++ show (count localClashes) ++ " locally-authenticated key(s) — filtering from merge"))
4270+ -- Remove clashing entries from deduped before merge (only first-party claims for
4271+ -- locally-authenticated keys; third-party relays are handled by trustedGossipEntries)
4272+ let safeDeduped = if count localClashes > 0
4273+ then do d <- deduped; where not (d.server == fromServer && any localClashes (\pk -> pk == d.pubkey)); yield d
4274+ else deduped
42744275 -- Per-bucket gossip merge: read only presence buckets affected by this gossip batch
42754276 -- instead of all 16. All entries for a pubkey hash to the same bucket, so per-bucket
42764277 -- merge is equivalent to global merge. Reduces lock scope for small gossip batches.
4277− let allGossipEntries = union deduped trustedOffline
4278+ let allGossipEntries = union safeDeduped trustedOffline
42784279 let affectedBuckets = uniqueBucketsFor allGossipEntries (\e -> e.pubkey)
42794280 gossipResult <- atomic do
42804281 selectedPresence <- readSelectedPresenceBuckets affectedBuckets
@@ -4282,8 +4283,8 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
42824283 -- Pre-filter: when near capacity, reject gossip entries for pubkeys not already in presence.
42834284 -- This prevents a malicious peer from spiking presence above maxPresenceEntries via large batches.
42844285 let capacityDeduped = if tpc >= maxPresenceEntries
4285− then do d <- deduped; where any selectedPresence (\p -> p.pubkey == d.pubkey); yield d
4286− else deduped
4286+ then do d <- safeDeduped; where any selectedPresence (\p -> p.pubkey == d.pubkey); yield d
4287+ else safeDeduped
42874288 let gm = mergeGossipPresence selectedPresence capacityDeduped trustedOffline fromServer t
42884289 -- Write back only affected buckets (not writeAllPresence, which would zero unaffected buckets)
42894290 writeSelectedPresenceBuckets affectedBuckets gm.finalPresence
@@ -4294,7 +4295,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
42944295 let newPeers = gm.gossipPeerServers
42954296 set *knownPeerServers = fold (\acc s -> if any acc (\a -> a == s) then acc else union acc [s]) knownPeers newPeers
42964297 set *presenceVersion = t
4297− yield {gm: gm, preFiltered: count deduped - count capacityDeduped}
4298+ yield {gm: gm, preFiltered: count safeDeduped - count capacityDeduped}
42984299 let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
42994300 when (not gossipResult.gm.withinCapacity || gossipResult.preFiltered > 0)
43004301 (logWarn ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer ++ " (pre-filtered: " ++ show gossipResult.preFiltered ++ ")"))
@@ -4334,7 +4335,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
43344335handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId authorization respond -> do
43354336 t <- now
43364337 when (correlationId != "") (logInfo ("[CORR] " ++ correlationId ++ " forward " ++ messageId ++ " from " ++ fromServer))
4337− withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
4338+ withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (True {}) (do
43384339 when (fromServer != originServer)
43394340 (logInfo ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
43404341 -- Relay origin verification: for relay forwards, verify originServer is a known peer
@@ -4414,9 +4415,8 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
44144415 let fnb = pubkeyBucket toKey
44154416 fwdNonces <- readNonceBucket fnb
44164417 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
4417− let maxPerBucket = maxForwardNonces / 16 + 1
4418− writeNonceBucket fnb (if count updatedNonces > maxPerBucket
4419− then evictNonces updatedNonces maxPerBucket forwardNonceRetention t
4418+ writeNonceBucket fnb (if count updatedNonces > maxNoncesPerBucket
4419+ then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
44204420 else updatedNonces)
44214421 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
44224422 if result.noPresence then do
@@ -4443,7 +4443,10 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
44434443 when (correlationId != "") (logInfo ("[CORR] " ++ correlationId ++ " ack from " ++ fromServer))
44444444 -- Wrap respond to add ackedCount default for withFederationAuth error paths
44454445 let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0}
4446− withFederationAuth authorization t ackRespond fromServer "ack" requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (do
4446+ -- clearOnAuth=False: defer clearFailure until we confirm the ack matched actual
4447+ -- forward entries. Prevents a peer from resetting its failure/backoff counters by
4448+ -- spamming acks for messageIds we never forwarded to it.
4449+ withFederationAuth authorization t ackRespond fromServer "ack" requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (False {}) (do
44474450 let dedupedIds = dedupMessageIds messageIds
44484451 -- Clear forward entries for the acking server. Extract removed forwards' toKey
44494452 -- so we can reclaim origin messages that are no longer needed.
@@ -4453,6 +4456,9 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
44534456 writeAllForwards (forwardsExcluding forwards fromServer dedupedIds)
44544457 yield removed
44554458 let ackedCount = count removedForwards
4459+ -- Only clear failure record when actual forwards were matched (proves the peer
4460+ -- received and processed messages we sent). Unmatched acks don't prove forward delivery.
4461+ when (ackedCount > 0) (clearFailure fromServer)
44564462 -- Guard: warn if peer acks messages we had no forward entries for. This can indicate
44574463 -- a peer acking messages it never received (e.g., the forward response was lost but
44584464 -- the message was never stored), which would cause premature origin message deletion.
@@ -4672,7 +4678,7 @@ handleHealth = \authorization respond -> do
46724678 watchdogHealthy: True {},
46734679 draining: result.isDraining})
46744680
4675−zeroStats = {ok: False {}, 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}, 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}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}}}
4681+zeroStats = {ok: False {}, 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}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}}}
46764682
46774683handleStats = \authorization respond -> do
46784684 t <- now
@@ -4740,9 +4746,15 @@ handleStats = \authorization respond -> do
47404746 let hungFedSlots = count (do r <- core.fi; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs; yield r)
47414747 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
47424748 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt
4749+ -- Per-server forward queue summary: count unique servers and max queue depth
4750+ let fwdCounts = fold (\acc f ->
4751+ if any acc (\a -> a.server == f.toServer) then do a <- acc; yield (if a.server == f.toServer then {server: a.server, count: a.count + 1} else a)
4752+ else union acc [{server: f.toServer, count: 1}]) [] core.f
4753+ let fwdServerCount = count fwdCounts
4754+ let maxFwdDepth = fold (\acc c -> max acc c.count) 0 fwdCounts
47434755 yield (respond {ok: True {},
47444756 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: count core.a, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: count core.d, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
4745− 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},
4757+ 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},
47464758 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},
47474759 timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr}})
47484760 else yield (respond zeroStats)
@@ -4780,12 +4792,12 @@ handleStatsKey = \pubkey authorization respond -> do
47804792 keyBucket <- readMessageBucket (pubkeyBucket pubkey)
47814793 -- Single bucket read: all entries for this pubkey are in the same presence bucket
47824794 keyPresenceBkt <- readPresenceBucket (pubkeyBucket pubkey)
4783− allSessions <- readAllSessions
4795+ -- Count sessions without materializing all 16 buckets (O(n/16) best case)
4796+ keySessions <- countSessionsForKey pubkey
47844797 forwards <- readFwdBucket (pubkeyBucket pubkey)
47854798 activePolls <- *activePolls
47864799 let keyMessages = count (do m <- keyBucket; where m.toKey == pubkey; yield m)
47874800 let keyPresence = do p <- keyPresenceBkt; where p.pubkey == pubkey && p.expiresAt > t; yield {server: p.server, expiresAt: p.expiresAt}
4788− let keySessions = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
47894801 let keyForwardEntries = do f <- forwards; where f.toKey == pubkey; yield f
47904802 let keyForwards = count keyForwardEntries
47914803 let keyForwardDetails = do f <- keyForwardEntries; yield {server: f.toServer, retries: f.retries, createdAt: f.createdAt}
@@ -4845,12 +4857,18 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
48454857 let validForwards = do f <- forwardsData; where not (isLocalServer f.toServer) && (f.originServer == "" || not (isBadServerName f.originServer)) && not (isLocalServer f.originServer); yield f
48464858 let validAcked = do a <- ackedIdsData; where not (isLocalServer a.originServer); yield a
48474859 let validOverflow = do a <- ackOverflowData; where not (isLocalServer a.originServer); yield a
4848− let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow)
4860+ -- Validate presence entries: reject entries with bad server names (SSRF risk) or
4861+ -- self-referencing local server (nonsensical in a snapshot import). The server field
4862+ -- in PresenceEntry uses AnyServerName (no isBadServerName check), so a corrupted
4863+ -- snapshot could inject presence pointing to private IPs, causing forward attempts
4864+ -- to those addresses.
4865+ let validPresence = do p <- presenceData; where not (isLocalServer p.server) && not (isBadServerName p.server); yield p
4866+ let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow) + (count presenceData - count validPresence)
48494867 when (rejectedCount > 0) (logWarn ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))
48504868 result <- atomic do
48514869 -- Merge presence (existing entries win on conflict via freshness)
48524870 allPresence <- readAllPresence
4853− let mergedPresence = mergeByPubkeyServer allPresence (do p <- presenceData; where p.expiresAt > t; yield p)
4871+ let mergedPresence = mergeByPubkeyServer allPresence (do p <- validPresence; where p.expiresAt > t; yield p)
48544872 let cappedPresence = takeRelation maxPresenceEntries mergedPresence
48554873 writeAllPresence cappedPresence
48564874 set *totalPresenceCount = count cappedPresence
@@ -4947,6 +4965,9 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
49474965-- Rate-limited via lookup rate limiter (read-only info lookup).
49484966handleListMessages = \authorization clientIp respond -> do
49494967 t <- now
4968+ draining <- *draining
4969+ if draining then yield (respond {messages: [], authorized: False {}, count: 0})
4970+ else do
49504971 globalRateLimited <- checkGlobalRate t
49514972 if globalRateLimited then yield (respond {messages: [], authorized: False {}, count: 0})
49524973 else
@@ -4969,11 +4990,14 @@ handleListMessages = \authorization clientIp respond -> do
49694990
49704991-- POST /messages/delete: Delete pending messages by ID (recipient-only).
49714992-- The recipient knows their pubkey (toKey) and can be authenticated via session.
4972−-- Deletes matching messages, adjusts counters, and cancels pending forwards.
4973−-- NOTE: Deleted messages won't trigger federation acks to origin servers,
4974−-- potentially leaving stale forward entries on origin servers until prune.
4993+-- Deletes matching messages, adjusts counters, cancels pending forwards, and
4994+-- sends federation acks for federated messages so origin servers can reclaim
4995+-- their forward entries instead of retrying until prune.
49754996handleDeleteMessages = \messageIds authorization clientIp respond -> do
49764997 t <- now
4998+ draining <- *draining
4999+ if draining then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0})
5000+ else do
49775001 globalRateLimited <- checkGlobalRate t
49785002 if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0})
49795003 else
@@ -4988,7 +5012,7 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
49885012 if deleteRateLimited then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0})
49895013 else do
49905014 let dedupedIds = dedupMessageIds messageIds
4991− deletedCount <- atomic do
5015+ deleteResult <- atomic do
49925016 let b = pubkeyBucket pubkey
49935017 bucket <- readMessageBucket b
49945018 let toDelete = do m <- bucket; where m.toKey == pubkey && any dedupedIds (\mid -> mid == m.messageId); yield m
@@ -5004,9 +5028,22 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
50045028 let dfb = pubkeyBucket pubkey
50055029 fwdBkt <- readFwdBucket dfb
50065030 writeFwdBucket dfb (do f <- fwdBkt; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
5007− yield removed
5008− else yield 0
5009− yield (respond {ok: True {}, authorized: True {}, deletedCount: deletedCount}))
5031+ -- Create ack entries for federated messages so origin servers learn about the delete
5032+ let federatedDeleted = do m <- toDelete; where m.originServer != ""; yield m
5033+ let ackedFederated = federatedAckEntries federatedDeleted t
5034+ if count ackedFederated > 0 then do
5035+ ackedBkt <- readAckedBucket b
5036+ let combinedAcked = union ackedBkt ackedFederated
5037+ let excessCount = max 0 (count combinedAcked - maxAckedIds)
5038+ let eviction = evictSmallest combinedAcked excessCount (\a -> a.ackedAt)
5039+ writeAckedBucket b eviction.kept
5040+ yield {deletedCount: removed, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
5041+ else yield {deletedCount: removed, pendingAcks: [], evictedEntries: []}
5042+ else yield {deletedCount: 0, pendingAcks: [], evictedEntries: []}
5043+ fork (when federationEnabled (do
5044+ sendAcksAndClean deleteResult.pendingAcks
5045+ retryEvictedAcks deleteResult.evictedEntries deleteResult.pendingAcks))
5046+ yield (respond {ok: True {}, authorized: True {}, deletedCount: deleteResult.deletedCount}))
50105047
50115048-- POST /messages/batch: Send multiple messages in one request.
50125049-- Auth checked once; each message processed independently with its own rate limiting.
@@ -5660,6 +5697,8 @@ main = do
56605697 (logWarn "maintenanceSecret is empty — /maintenance/* and /stats endpoints are disabled")
56615698 when (matchesHostnameOrSubdomain serverHostN "localhost")
56625699 (logWarn ("serverHost is " ++ serverHost ++ " — peers cannot reach this server for federation"))
5700+ when (federationEnabled && isBadServerName serverHostN)
5701+ (logError ("serverHost '" ++ serverHost ++ "' would be rejected by isBadServerName — federation peers cannot forward messages here. Set serverHost to a public DNS name."))
56635702 logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
56645703 when federationEnabled
56655704 (logWarn "SECURITY: DNS rebinding risk — federation URLs resolve DNS at fetch time, not at validation time. A peer hostname passing isBadServerName could rebind to a private IP between validation and fetch. Mitigate by blocking outbound connections to RFC1918/link-local/CGNAT ranges at the firewall level.")