@@ -71,8 +71,10 @@ maxMessages = 100000 -- global cap on total stored messages across all recipient
7171clockSkewTolerance : Int<Ms>
7272clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers
7373maxForwardsEntries = 10000
74+maxForwardsPerKey = 50 -- max pending forwards per recipient pubkey (prevents one key from starving others)
7475maxServerNameLen = 128
7576maxMessageIdLen = 128
77+maxCorrelationIdLen = 128
7678maxTokenLen = 256
7779maxChallenges = 10000
7880maxSessions = 10000
@@ -177,7 +179,7 @@ federationLatencyWindowMs : Int<Ms>
177179federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
178180
179181-- Snapshot format version (increment on schema changes to prevent corrupt restores)
180−snapshotVersion = 4 -- bumped for messageTtl field in ForwardEntry
182+snapshotVersion = 5 -- bumped for allowedPeers field in snapshot
181183
182184-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
183185maxGossipEventsPerServer = 500 -- max total gossip events per server per window
@@ -669,6 +671,10 @@ type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: Pubkey
669671-- Last time federation latency metrics were reset (for windowed averages)
670672*federationLatencyResetAt : Timestamp
671673
674+-- Suppresses latency metric recording during startup probe phase (prevents
675+-- unauthenticated probe results from skewing initial /stats readings)
676+*startupProbeActive : Bool
677+
672678-- Total session count across all buckets (avoids O(16) recount on monitoring paths)
673679*totalSessionCount : Int
674680
@@ -903,11 +909,11 @@ route MaintenanceApi where
903909 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
904910 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
905911 GET /maintenance/snapshot headers {authorization: Text}
906− -> {ok: Bool, version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text]}
912+ -> {ok: Bool, version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text]}
907913 = Snapshot
908914 -- Restore: import a previously-exported snapshot. Server must be in drain mode
909915 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
910− POST {version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text]} /maintenance/restore headers {authorization: Text}
916+ POST {version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce], blockedPeers: [Text], allowedPeers: [Text]} /maintenance/restore headers {authorization: Text}
911917 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
912918 = Restore
913919 -- Block a federation peer (rejects all inbound/outbound federation with this server)
@@ -974,7 +980,7 @@ route HealthApi where
974980-- BREAKING CHANGE from flat format — existing scrapers must update field paths.
975981route StatsApi where
976982 GET /stats headers {authorization: Text}
977− -> {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}}
983+ -> {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, senderExpirations: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int}}
978984 = Stats
979985 GET /stats/peers headers {authorization: Text}
980986 -> {peers: [PeerStatus], ok: Bool}
@@ -1061,6 +1067,10 @@ genToken = genTokenN tokenChunks
10611067tooLong : Text -> Int -> Bool
10621068tooLong = \s maxLen -> drop maxLen s != ""
10631069
1070+-- Truncate correlationId to maxCorrelationIdLen to prevent log flooding from malicious peers
1071+sanitizeCorrelationId : Text -> Text
1072+sanitizeCorrelationId = \cid -> take maxCorrelationIdLen cid
1073+
10641074isHexChar : Text -> Bool
10651075isHexChar = \c -> contains c "0123456789abcdef"
10661076
@@ -1453,6 +1463,27 @@ hasSessionForKey = \pubkey b ->
14531463 if any bucket (\s -> s.pubkey == pubkey) then yield (True {})
14541464 else hasSessionForKey pubkey (b + 1)
14551465
1466+-- Check if a pubkey has any non-expired session across all buckets (short-circuits on first match).
1467+hasActiveSessionForKey : PubkeyHex -> Timestamp -> Int -> IO Bool
1468+hasActiveSessionForKey = \pubkey t b ->
1469+ if b > 15 then yield (False {})
1470+ else do
1471+ bucket <- readSessionBucket b
1472+ if any bucket (\s -> s.pubkey == pubkey && s.expiresAt > t) then yield (True {})
1473+ else hasActiveSessionForKey pubkey t (b + 1)
1474+
1475+-- Filter a list of pubkeys to those with active (non-expired) sessions.
1476+-- Checks each key independently (short-circuits per key), avoids materializing all sessions.
1477+filterActiveSessionKeys : [PubkeyHex] -> Timestamp -> IO [PubkeyHex]
1478+filterActiveSessionKeys = \keys t ->
1479+ case headOf keys of
1480+ Nothing {} -> yield []
1481+ Just {value: k} -> do
1482+ let rest = do x <- keys; where x != k; yield x
1483+ hasActive <- hasActiveSessionForKey k t 0
1484+ remaining <- filterActiveSessionKeys rest t
1485+ if hasActive then yield (union [k] remaining) else yield remaining
1486+
14561487-- Count sessions for a specific pubkey across all buckets (avoids materializing full session list).
14571488-- Used in verify fast path to skip readAllSessions.
14581489countSessionsForKey : PubkeyHex -> IO Int
@@ -1467,6 +1498,19 @@ countSessionsInBucket = \pubkey b ->
14671498 rest <- countSessionsInBucket pubkey (b + 1)
14681499 yield (n + rest)
14691500
1501+-- Remove all sessions for a pubkey across all 16 buckets, returning total revoked.
1502+-- Each bucket is filtered in-place (avoids readAllSessions + writeAllSessions repartition).
1503+revokeSessionsForKey : PubkeyHex -> Int -> IO Int
1504+revokeSessionsForKey = \pubkey b ->
1505+ if b > 15 then yield 0
1506+ else do
1507+ bucket <- readSessionBucket b
1508+ let kept = do s <- bucket; where s.pubkey != pubkey; yield s
1509+ let removed = count bucket - count kept
1510+ when (removed > 0) (writeSessionBucket b kept)
1511+ rest <- revokeSessionsForKey pubkey (b + 1)
1512+ yield (removed + rest)
1513+
14701514-- Apply a filter/transform to every session bucket and return the total count.
14711515-- Combines transform + count in one pass, avoiding a redundant 16-bucket traversal.
14721516pruneAndCountAllSessions : ([Session] -> [Session]) -> IO Int
@@ -2453,13 +2497,15 @@ fetchFederation = \server request -> do
24532497 let latency = endTime - startTime
24542498 when (latency > federationFetchTimeoutMs)
24552499 (logWarn ("federation: slow response from " ++ server ++ " (" ++ show latency ++ "ms)"))
2456− atomic do
2500+ -- Skip latency recording during startup probes (unauthenticated, expected to fail)
2501+ probing <- *startupProbeActive
2502+ when (not probing) (atomic do
24572503 total <- *federationLatencyTotal
24582504 cnt <- *federationLatencyCount
24592505 mx <- *federationLatencyMax
24602506 set *federationLatencyTotal = total + latency
24612507 set *federationLatencyCount = cnt + 1
2462− set *federationLatencyMax = if latency > mx then latency else mx
2508+ set *federationLatencyMax = if latency > mx then latency else mx)
24632509 yield result
24642510
24652511clampTtl : Int<S> -> Int<S>
@@ -2485,7 +2531,7 @@ dedupMessageIds = \messageIds ->
24852531 let capped = takeRelation maxMessageIds messageIds in
24862532 fold (\acc mid ->
24872533 if any acc (\a -> a == mid) then acc
2488− else union acc [mid]) [] capped
2534+ else union [mid] acc) [] capped
24892535
24902536
24912537-- Constant-time comparison iteration count. Derived from tokenHexLen (48) + 16
@@ -3387,7 +3433,12 @@ forwardQueuedMessages = \targets t -> do
33873433 relevantMessages <- readSelectedBuckets neededBuckets
33883434 let candidates = forwardCandidates remoteTargets relevantMessages forwards t
33893435 let totalFwdCount <- countAllForwards 0
3390− let capped = takeRelation (maxForwardsEntries - totalFwdCount) candidates
3436+ -- Per-key cap: filter candidates to respect maxForwardsPerKey per recipient
3437+ let perKeyCapped = fold (\acc c ->
3438+ let existingForKey = count (do f <- forwards; where f.toKey == c.toKey; yield f)
3439+ + count (do a <- acc; where a.toKey == c.toKey; yield a)
3440+ in if existingForKey >= maxForwardsPerKey then acc else union acc [c]) [] candidates
3441+ let capped = takeRelation (maxForwardsEntries - totalFwdCount) perKeyCapped
33913442 -- Write back only the affected forward buckets (not all 16)
33923443 let combined = union forwards capped
33933444 writeSelectedFwdBuckets neededFwdBuckets combined
@@ -3632,11 +3683,11 @@ pruneAndRetry = \t -> do
36323683 set *lastPresenceRefresh = t
36333684 yield {extendedActive: [], nearExpiryActive: [], refreshPeers: knownPeers, shouldRefresh: True {}}
36343685 else do
3635− -- Only read session buckets for pubkeys that have near-expiry presence
3686+ -- Check per-key session existence (avoids materializing all 16 session buckets)
36363687 let nearExpiryPubkeys = fold (\acc p -> if any acc (\a -> a == p.pubkey) then acc else union acc [p.pubkey]) [] localNearExpiry
3637− allSessions <- readAllSessions
3638− let toExtend = do p <- localNearExpiry; where any allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
3639− let notExtended = do p <- localNearExpiry; where none allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
3688+ activeKeys <- filterActiveSessionKeys nearExpiryPubkeys t
3689+ let toExtend = do p <- localNearExpiry; where any activeKeys (\k -> k == p.pubkey); yield p
3690+ let notExtended = do p <- localNearExpiry; where none activeKeys (\k -> k == p.pubkey); yield p
36403691 when (count toExtend > 0) (do
36413692 -- Only write to the affected presence buckets (not all 16)
36423693 let affectedBuckets = uniqueBucketsFor toExtend (\e -> e.pubkey)
@@ -4309,7 +4360,11 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTt
43094360 let remainingTtl = if messageTtl > 0 then max 0 (msgExpiresAt - t) else 0
43104361 let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, messageTtl: remainingTtl}
43114362 totalFwdCount <- countAllForwards 0
4312− let cappedFwds = takeRelation (maxForwardsEntries - totalFwdCount) newFwds
4363+ let globalCap = maxForwardsEntries - totalFwdCount
4364+ -- Per-key cap: prevents one key from consuming disproportionate forward slots
4365+ let keyFwdCount = count (do f <- fwdBkt; where f.toKey == to; yield f)
4366+ let keyCap = maxForwardsPerKey - keyFwdCount
4367+ let cappedFwds = takeRelation (min globalCap keyCap) newFwds
43134368 writeFwdBucket fbi (union fwdBkt cappedFwds)
43144369 -- Return whether any forwards were actually enqueued (vs dropped by capacity cap)
43154370 yield (count cappedFwds > 0)
@@ -4649,7 +4704,7 @@ handleVerifyToken = \token scope clientIp respond -> do
46494704handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlationId authorization respond -> do
46504705 t <- now
46514706 let cappedHops = clamp 0 maxGossipHops hopsRemaining
4652− let corrId = if correlationId != "" then correlationId else ""
4707+ let corrId = sanitizeCorrelationId correlationId
46534708 when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " gossip from " ++ fromServer))
46544709 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (cappedHops >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (True {}) (do
46554710 -- Per-peer gossip event volume limiting: counts total events (not just requests) per server.
@@ -4762,7 +4817,8 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
47624817-- twice) and forwardCandidates (won't re-forward to the same server).
47634818handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId messageTtl authorization respond -> do
47644819 t <- now
4765− when (correlationId != "") (logInfo ("[CORR] " ++ correlationId ++ " forward " ++ messageId ++ " from " ++ fromServer))
4820+ let corrId = sanitizeCorrelationId correlationId
4821+ when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " forward " ++ messageId ++ " from " ++ fromServer))
47664822 withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (True {}) (do
47674823 when (fromServer != originServer)
47684824 (logInfo ("forward: relay from " ++ fromServer ++ " (origin: " ++ originServer ++ ") for " ++ toKey))
@@ -4884,7 +4940,8 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
48844940-- recipient has no local presence and no other pending forwards for that message.
48854941handleFederationAck = \messageIds originServer fromServer requestTimestamp correlationId authorization respond -> do
48864942 t <- now
4887− when (correlationId != "") (logInfo ("[CORR] " ++ correlationId ++ " ack from " ++ fromServer))
4943+ let corrId = sanitizeCorrelationId correlationId
4944+ when (corrId != "") (logInfo ("[CORR] " ++ corrId ++ " ack from " ++ fromServer))
48884945 -- Wrap respond to add ackedCount default for withFederationAuth error paths
48894946 let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0}
48904947 -- clearOnAuth=False: defer clearFailure until we confirm the ack matched actual
@@ -4977,7 +5034,8 @@ handleDrain = \phase authorization respond -> do
49775034 ff <- *forwardFailures
49785035 fn <- readAllNonces
49795036 bp <- *blockedPeers
4980− yield {version: snapshotVersion, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp}
5037+ ap <- *allowedPeers
5038+ yield {version: snapshotVersion, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap}
49815039 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
49825040 atomic do set *prunesSinceSnapshot = 0
49835041 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
@@ -5157,7 +5215,7 @@ handleHealth = \authorization respond -> do
51575215 watchdogHealthy: True {},
51585216 draining: result.isDraining})
51595217
5160−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 {}}}
5218+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, senderExpirations: 0}, timing: {pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, drainPhase: 0}}
51615219
51625220handleStats = \authorization respond -> do
51635221 t <- now
@@ -5193,7 +5251,8 @@ handleStats = \authorization respond -> do
51935251 ger <- *gossipEventRates
51945252 vir <- *verifyIpRates
51955253 ukir <- *uniqueKeyIpRates
5196− yield {sc: sc, pc: pc, c: c, aCount: aCount, ao: ao, dCount: dCount, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg, fnCount: fnCount, pac: pac, bp: bp, ger: ger, vir: vir, ukir: ukir}
5254+ se <- *senderExpirations
5255+ yield {sc: sc, pc: pc, c: c, aCount: aCount, ao: ao, dCount: dCount, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg, fnCount: fnCount, pac: pac, bp: bp, ger: ger, vir: vir, ukir: ukir, se: se}
51975256 counters <- atomic do
51985257 sr <- *sendRates
51995258 cr <- *challengeRates
@@ -5237,8 +5296,8 @@ handleStats = \authorization respond -> do
52375296 yield (respond {ok: True {},
52385297 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},
52395298 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},
5240− 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},
5241− timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0}})
5299+ 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},
5300+ 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}})
52425301 else yield (respond zeroStats)
52435302
52445303handleStatsPeers = \authorization respond -> do
@@ -5326,7 +5385,7 @@ handleReady = \respond -> do
53265385 let isReady = lastPrune > 0 && lastPrune >= sa
53275386 yield (respond {ready: isReady})
53285387
5329−emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: []}
5388+emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: [], allowedPeers: []}
53305389
53315390-- GET /maintenance/snapshot: Export federation metadata for persistence.
53325391-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
@@ -5346,26 +5405,27 @@ handleSnapshot = \authorization respond -> do
53465405 f <- readAllForwards
53475406 fn <- readAllNonces
53485407 bp <- *blockedPeers
5349− yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}
5408+ ap <- *allowedPeers
5409+ yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap}
53505410 atomic do
53515411 set *lastSnapshotExportAt = t
53525412 set *prunesSinceSnapshot = 0
5353− 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")
5354− yield (respond {ok: True {}, version: snapshotVersion, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces, blockedPeers: result.blockedPeers})
5413+ 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")
5414+ yield (respond {ok: True {}, version: snapshotVersion, 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})
53555415
53565416-- POST /maintenance/restore: Import a previously-exported snapshot.
53575417-- Requires drain mode to prevent conflicts with live state. Merges imported data
53585418-- with existing state (dedup entries are unioned, presence is merged by freshness).
53595419-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
53605420-- trigger federation forward creation from messages + restored presence.
5361−handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData blockedPeersData authorization respond -> do
5421+handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData blockedPeersData allowedPeersData authorization respond -> do
53625422 let authorized = verifyMaintenanceAuth authorization
53635423 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
53645424 else do
53655425 -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers,
5366− -- version 3 = pre-messageTtl in ForwardEntry (accept all for backwards compatibility;
5367− -- missing fields default to empty list or 0)
5368− if version != 0 && version != 1 && version != 2 && version != 3 && version != snapshotVersion then do
5426+ -- version 3 = pre-messageTtl in ForwardEntry, version 4 = pre-allowedPeers
5427+ -- (accept all for backwards compatibility; missing fields default to empty list or 0)
5428+ if version != 0 && version != 1 && version != 2 && version != 3 && version != 4 && version != snapshotVersion then do
53695429 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
53705430 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
53715431 else do
@@ -5434,6 +5494,11 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
54345494 let mergedBlocked = fold (\acc b -> if any acc (\x -> x == b) then acc else union acc [b]) blocked blockedPeersData
54355495 let cappedBlocked = takeRelation maxBlockedPeers mergedBlocked
54365496 set *blockedPeers = cappedBlocked
5497+ -- Restore allowed peers (union with existing, cap at maxAllowedPeers)
5498+ allowed <- *allowedPeers
5499+ let mergedAllowed = fold (\acc a -> if any acc (\x -> x == a) then acc else union acc [a]) allowed allowedPeersData
5500+ let cappedAllowed = takeRelation maxAllowedPeers mergedAllowed
5501+ set *allowedPeers = cappedAllowed
54375502 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
54385503 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
54395504 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
@@ -5830,11 +5895,10 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do
58305895 else do
58315896 t <- now
58325897 result <- atomic do
5833− allSessions <- readAllSessions
5834− let revokedCount = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
5835− let remaining = do s <- allSessions; where s.pubkey != pubkey; yield s
5836− writeAllSessions remaining
5837− set *totalSessionCount = count remaining
5898+ -- Filter each session bucket in-place (avoids readAllSessions + writeAllSessions repartition)
5899+ revokedCount <- revokeSessionsForKey pubkey 0
5900+ sc <- *totalSessionCount
5901+ set *totalSessionCount = max 0 (sc - revokedCount)
58385902 -- Clear local presence
58395903 let pb = pubkeyBucket pubkey
58405904 presenceBkt <- readPresenceBucket pb
@@ -6065,7 +6129,7 @@ serveInner = \req -> case req of
60656129 Shutdown {authorization, respond} -> handleShutdown authorization respond
60666130 Undrain {authorization, respond} -> handleUndrain authorization respond
60676131 Snapshot {authorization, respond} -> handleSnapshot authorization respond
6068− Restore {version, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, blockedPeers, authorization, respond} -> handleRestore version presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces blockedPeers authorization respond
6132+ Restore {version, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, blockedPeers, allowedPeers, authorization, respond} -> handleRestore version presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces blockedPeers allowedPeers authorization respond
60696133 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
60706134 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
60716135 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
@@ -6264,7 +6328,8 @@ backgroundSnapshot = do
62646328 f <- readAllForwards
62656329 fn <- readAllNonces
62666330 bp <- *blockedPeers
6267− yield {version: snapshotVersion, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}
6331+ ap <- *allowedPeers
6332+ yield {version: snapshotVersion, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp, allowedPeers: ap}
62686333 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
62696334 atomic do
62706335 set *lastAutoSnapshotExportAt = t
@@ -6322,8 +6387,10 @@ backgroundWatchdog = do
63226387-- Surfaces DNS, firewall, and TLS issues before the first real gossip cycle.
63236388-- Probes up to 10 peers sequentially (not forked) to avoid overwhelming the network.
63246389startupProbePeers = do
6390+ atomic do set *startupProbeActive = True {}
63256391 knownPeers <- *knownPeerServers
63266392 startupProbeNext knownPeers 0
6393+ atomic do set *startupProbeActive = False {}
63276394
63286395startupProbeNext = \peers probed ->
63296396 if probed >= 10 then logInfo ("startup: probed " ++ show probed ++ " peers")
@@ -6347,15 +6414,6 @@ startupProbeNext = \peers probed ->
63476414 Ok r -> logInfo ("startup: peer " ++ server ++ " reachable (" ++ show latency ++ "ms)")
63486415 startupProbeNext rest (probed + 1)
63496416
6350−-- Reset federation latency metrics after startup probes so unauthenticated probe
6351−-- results (expected "unauthorized" responses) don't skew initial /stats readings.
6352−resetFederationLatencyMetrics = atomic do
6353− set *federationLatencyTotal = 0
6354− set *federationLatencyCount = 0
6355− set *federationLatencyMax = 0
6356− t <- now
6357− set *federationLatencyResetAt = t
6358−
63596417main = do
63606418 let configOk = listenPort > 0 && maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
63616419 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
@@ -6380,6 +6438,7 @@ main = do
63806438 && ackGracePeriod < unackedMessageMaxAge
63816439 && clockSkewTolerance < challengeExpiry
63826440 && maxMessagesPerRecipient <= maxMessages
6441+ && maxForwardsPerKey <= maxForwardsEntries
63836442 && maxSessionsPerKey <= maxSessions
63846443 && maxChallengesPerKey <= maxChallenges
63856444 && maxPollsPerKey <= maxActivePolls
@@ -6404,6 +6463,8 @@ main = do
64046463 set *serverGeneration = t
64056464 when (maintenanceSecret == "")
64066465 (logWarn "maintenanceSecret is empty — /maintenance/* and /stats endpoints are disabled")
6466+ when (maintenanceSecret != "" && not (tooLong maintenanceSecret 15))
6467+ (logWarn "maintenanceSecret is shorter than 16 characters — use a longer secret for production deployments")
64076468 when (maintenanceSecretPrevious != "")
64086469 (logInfo "maintenanceSecretPrevious is set — both current and previous secrets are accepted (rotation mode)")
64096470 when (matchesHostnameOrSubdomain serverHostN "localhost")
@@ -6420,7 +6481,7 @@ main = do
64206481 logInfo (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")
64216482 logInfo (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")
64226483 logInfo (" max presence: " ++ show maxPresenceEntries)
6423− logInfo (" max forwards: " ++ show maxForwardsEntries)
6484+ logInfo (" max forwards: " ++ show maxForwardsEntries ++ " (per key: " ++ show maxForwardsPerKey ++ ")")
64246485 logInfo (" session expiry: " ++ show sessionExpiry ++ "ms (max lifetime: " ++ show maxSessionLifetime ++ "ms)")
64256486 logInfo (" message max age: " ++ show messageMaxAge ++ "ms")
64266487 logInfo (" global rate limit: " ++ show maxGlobalRequestRate ++ "/" ++ show globalRateWindowMs ++ "ms")
@@ -6447,6 +6508,5 @@ main = do
64476508 when federationEnabled (fork (do
64486509 sleep 5000 -- wait for listener to be ready
64496510 logInfo "startup: probing federation peers..."
6450− startupProbePeers
6451− resetFederationLatencyMetrics))
6511+ startupProbePeers))
64526512 listen listenPort serve