@@ -30,6 +30,7 @@ maxPresenceTtl = 3600 -- max presence TTL from gossip
3030maxPubkeyLen = 256
3131maxBlobLen = 65536
3232maxSignatureLen = 256
33+maxIpLen = 45 -- max length of an IP address string (IPv6 = 45 chars); used in matchIp instead of maxServerNameLen
3334maxMessageIds = 100
3435maxBatchSize = 100 -- max messages per batch request (bounds recursion depth in sendBatchInner)
3536maxGossipEvents = 100
@@ -211,6 +212,11 @@ maxLookupRatePerKey = 5 -- max lookups per key per window
211212lookupRateWindowMs : Int<Ms>
212213lookupRateWindowMs = 1000
213214maxLookupRateEntries = 1000
215+-- Client write operation rate limiting (per-key, on ack/delete — separate from read-only lookups)
216+maxClientWriteRatePerKey = 10 -- max ack/delete requests per key per window
217+clientWriteRateWindowMs : Int<Ms>
218+clientWriteRateWindowMs = 1000
219+maxClientWriteRateEntries = 1000
214220-- Federation verify rate limiting (per-IP, tighter than general IP rate)
215221maxVerifyRatePerIp = 5 -- max verify requests per IP per window
216222verifyRateWindowMs : Int<Ms>
@@ -552,7 +558,14 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
552558
553559*lastHeartbeatAt : Timestamp
554560
555−*draining : Bool
561+-- Drain phase: 0=normal (accepting), 1=soft drain (block new auth, serve existing sessions),
562+-- 2=full drain (reject all client requests). Allows staged shutdown:
563+-- phase 1 lets active clients finish while preventing new connections.
564+*drainPhase : Int
565+
566+-- Shutdown requested: when True, background loops exit after drain completes.
567+-- Set by /maintenance/shutdown; cannot be undone (requires process restart).
568+*shutdownRequested : Bool
556569
557570*pruneRunning : Timestamp
558571
@@ -562,6 +575,9 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
562575
563576*lookupRates : [LookupRateEntry]
564577
578+-- Separate rate limiter for client write operations (ack, delete) — prevents ack bursts from starving lookups
579+*clientWriteRates : [LookupRateEntry]
580+
565581*recvRates : [RecvRateEntry]
566582
567583*pairRates : [PairRateEntry]
@@ -687,6 +703,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
687703-- Last time the background watchdog loop ran (for health check liveness detection)
688704*lastWatchdogAt : Timestamp
689705
706+-- Fingerprint of key state counters at last auto-snapshot export.
707+-- backgroundSnapshot compares current counters against this to skip redundant exports.
708+*lastSnapshotFingerprint : Int
709+
690710-- Cached health result: returned when /health is rate-limited instead of false all-healthy.
691711-- Updated on every successful (non-rate-limited) health check.
692712*cachedHealthOk : Bool
@@ -785,12 +805,18 @@ route MaintenanceApi where
785805 POST {} /maintenance/prune headers {authorization: Text}
786806 -> {ok: Bool}
787807 = Prune
788− POST {} /maintenance/drain headers {authorization: Text}
789− -> {ok: Bool, drained: Int}
808+ -- phase: 0 or 2 = full drain (backward compat), 1 = soft drain (auth-only block)
809+ POST {phase: Int} /maintenance/drain headers {authorization: Text}
810+ -> {ok: Bool, drained: Int, phase: Int}
790811 = Drain
791812 POST {} /maintenance/undrain headers {authorization: Text}
792813 -> {ok: Bool}
793814 = Undrain
815+ -- Graceful shutdown: triggers full drain + sets shutdown flag so background loops exit.
816+ -- Call this instead of drain when the process should stop after flush completes.
817+ POST {} /maintenance/shutdown headers {authorization: Text}
818+ -> {ok: Bool, drained: Int}
819+ = Shutdown
794820 -- Snapshot: export federation metadata for persistence across restarts.
795821 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
796822 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
@@ -979,6 +1005,14 @@ isOctalStart = \s -> take 1 s == "0" && isDigitChar (take 1 (drop 1 s)) && isAll
9791005hasOctalOctet : Text -> Bool
9801006hasOctalOctet = \s -> isOctalStart s || hasPatternAfterDot isOctalStart s
9811007
1008+-- Detect percent-encoded characters (e.g., %31%32%37.0.0.1 = 127.0.0.1)
1009+-- Blocks URL-encoded hostnames that could bypass isBadServerName SSRF protection.
1010+containsPercentEncoding : Text -> Bool
1011+containsPercentEncoding = \s ->
1012+ if s == "" then False {}
1013+ else if take 1 s == "%" && isHexChar (take 1 (drop 1 s)) && isHexChar (take 1 (drop 2 s)) then True {}
1014+ else containsPercentEncoding (drop 1 s)
1015+
9821016isHexStart : Text -> Bool
9831017isHexStart = \s -> take 2 s == "0x" || take 2 s == "0X"
9841018hasHexPrefix : Text -> Bool
@@ -1102,6 +1136,7 @@ isBadServerName = \sn ->
11021136 || containsSuffix ".test" sn
11031137 || containsSuffix ".example" sn
11041138 || containsSuffix ".invalid" sn
1139+ || containsPercentEncoding sn
11051140
11061141-- Look up a non-expired session matching the given token (constant-time comparison).
11071142-- Caller passes the appropriate session bucket (sharded by token prefix).
@@ -1298,6 +1333,21 @@ countSessionsInBucket = \pubkey b ->
12981333 rest <- countSessionsInBucket pubkey (b + 1)
12991334 yield (n + rest)
13001335
1336+-- Apply a filter/transform to every session bucket and return the total count.
1337+-- Combines transform + count in one pass, avoiding a redundant 16-bucket traversal.
1338+pruneAndCountAllSessions : ([Session] -> [Session]) -> IO Int
1339+pruneAndCountAllSessions = \f -> pruneAndCountSessionBucket f 0
1340+
1341+pruneAndCountSessionBucket : ([Session] -> [Session]) -> Int -> IO Int
1342+pruneAndCountSessionBucket = \f b ->
1343+ if b > 15 then yield 0
1344+ else do
1345+ bucket <- readSessionBucket b
1346+ let pruned = f bucket
1347+ writeSessionBucket b pruned
1348+ rest <- pruneAndCountSessionBucket f (b + 1)
1349+ yield (count pruned + rest)
1350+
13011351-- Apply a filter/transform to every session bucket
13021352transformAllSessionBuckets : ([Session] -> [Session]) -> IO ()
13031353transformAllSessionBuckets = \f -> do
@@ -1997,6 +2047,48 @@ countAllForwards = \b ->
19972047 rest <- countAllForwards (b + 1)
19982048 yield (count bucket + rest)
19992049
2050+-- Compute forward stats per-bucket without materializing all forwards into one list.
2051+-- Returns total count, oldest age, unique server count, and max per-server queue depth.
2052+computeForwardStats : Int -> Timestamp -> IO {totalCount: Int, oldestAge: Int, serverCount: Int, maxDepth: Int, serverCounts: [{server: Text, count: Int}]}
2053+computeForwardStats = \b t -> computeForwardStatsInner b t {totalCount: 0, oldestAge: 0, serverCount: 0, maxDepth: 0, serverCounts: []}
2054+
2055+computeForwardStatsInner = \b t acc ->
2056+ if b > 15 then yield acc
2057+ else do
2058+ bucket <- readFwdBucket b
2059+ let bucketCount = count bucket
2060+ let bucketOldest = case minBy bucket (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
2061+ -- Aggregate per-server counts: merge this bucket's servers into the running totals
2062+ let updatedServerCounts = fold (\sc f ->
2063+ if any sc (\s -> s.server == f.toServer)
2064+ then do s <- sc; yield (if s.server == f.toServer then {server: s.server, count: s.count + 1} else s)
2065+ else union sc [{server: f.toServer, count: 1}]) acc.serverCounts bucket
2066+ let newMaxDepth = fold (\mx s -> max mx s.count) acc.maxDepth updatedServerCounts
2067+ computeForwardStatsInner (b + 1) t {totalCount: acc.totalCount + bucketCount, oldestAge: max acc.oldestAge bucketOldest, serverCount: count updatedServerCounts, maxDepth: newMaxDepth, serverCounts: updatedServerCounts}
2068+
2069+-- Read only the forward buckets at the given indices (avoids reading all 16)
2070+readSelectedFwdBuckets : [Int] -> IO [ForwardEntry]
2071+readSelectedFwdBuckets = \buckets ->
2072+ case headOf buckets of
2073+ Nothing {} -> yield []
2074+ Just {value: b} -> do
2075+ let rest = do x <- buckets; where x != b; yield x
2076+ entries <- readFwdBucket b
2077+ remaining <- readSelectedFwdBuckets rest
2078+ yield (union entries remaining)
2079+
2080+-- Write forward entries to only the specified bucket indices. Entries are filtered
2081+-- by pubkeyBucket so each bucket gets only its entries. Unspecified buckets are untouched.
2082+writeSelectedFwdBuckets : [Int] -> [ForwardEntry] -> IO ()
2083+writeSelectedFwdBuckets = \buckets entries ->
2084+ case headOf buckets of
2085+ Nothing {} -> yield {}
2086+ Just {value: b} -> do
2087+ let rest = do x <- buckets; where x != b; yield x
2088+ let bucketEntries = do e <- entries; where pubkeyBucket e.toKey == b; yield e
2089+ writeFwdBucket b bucketEntries
2090+ writeSelectedFwdBuckets rest entries
2091+
20002092writeAllForwards : [ForwardEntry] -> IO ()
20012093writeAllForwards = \fwds -> do
20022094 let partitioned = fold (\acc f ->
@@ -2249,11 +2341,12 @@ safeCompare = \a b ->
22492341-- When a session was created with an IP (behind a proxy), requests without an IP header
22502342-- are rejected — prevents bypass if the proxy is removed mid-session.
22512343-- Uses safeCompareLoop to prevent timing side-channels on IP comparison.
2344+-- Uses maxIpLen (45) instead of maxServerNameLen (128) to avoid wasted iterations.
22522345matchIp : Text -> Text -> Bool
22532346matchIp = \clientIp sessionIp ->
22542347 if sessionIp == "" then True {} -- no binding configured for this session
22552348 else if clientIp == "" then False {} -- session has IP but request doesn't — proxy disappeared
2256− else safeCompareLoop clientIp sessionIp maxServerNameLen 0 0 == 0
2349+ else safeCompareLoop clientIp sessionIp maxIpLen 0 0 == 0
22572350
22582351-- Generic sliding-window rate limiter. Returns whether the request is rate-limited,
22592352-- the updated entries list, and retryAfterMs (time until the current window resets, 0 if not limited).
@@ -2542,28 +2635,35 @@ pruneExpired = \t -> do
25422635 set *knownPeerServers = presenceResults.keptPeerServers
25432636 when (presenceResults.totalRemoved > 0) (set *presenceVersion = t)
25442637 let presenceResult = {expiredLocal: presenceResults.expiredLocal, peerServers: presenceResults.allPeerServers}
2545− -- Prune auth state
2638+ -- Prune auth state (challenges in their own block; sessions counted during transform)
25462639 atomic do
25472640 challenges <- *challenges
25482641 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
2549− transformAllSessionBuckets (\bucket -> do x <- bucket; where x.expiresAt > t && t - x.createdAt < maxSessionLifetime; yield x)
2550− sessionCount <- countAllSessions
2551− set *totalSessionCount = sessionCount
2642+ -- Prune sessions and count in a single pass (avoids redundant 16-bucket traversal).
2643+ -- pruneAndCountAllSessions applies the filter to each bucket and sums the survivors.
2644+ sessionCount <- pruneAndCountAllSessions (\bucket -> do x <- bucket; where x.expiresAt > t && t - x.createdAt < maxSessionLifetime; yield x)
2645+ atomic do set *totalSessionCount = sessionCount
25522646 -- Prune messages across all 16 buckets concurrently. Each bucket prune is atomic
25532647 -- and touches only its own message bucket + rcBucket. The only shared write is
25542648 -- *totalMessageCount (a single Int), so STM contention is minimal.
25552649 -- Fork first 15 buckets; run the last one in the foreground to bound concurrency.
25562650 let messageCutoff = t - unackedMessageMaxAge
25572651 pruneMessageBucketsConcurrent 0 messageCutoff
2558− -- Prune federation state
2652+ -- Prune federation state (split into per-collection blocks to reduce lock contention
2653+ -- with poll-ack paths that write to ackedIds/deliveredIds concurrently).
2654+ let retainCutoff = t - messageMaxAge
2655+ -- Forwards: independent from acked/delivered, safe to prune separately
25592656 atomic do
2560− let retainCutoff = t - messageMaxAge
25612657 forwards <- readAllForwards
25622658 -- Clear all forwards when federation is disabled (they can never be retried)
25632659 writeAllForwards (if not federationEnabled then []
25642660 else do f <- forwards; where f.createdAt > retainCutoff; yield f)
2661+ -- AckedIds: contends with poll-ack (ackPolledMessages writes here), narrow the lock
2662+ atomic do
25652663 ackedIds <- readAllAcked
25662664 writeAllAcked (do a <- ackedIds; where a.ackedAt > retainCutoff; yield a)
2665+ -- DeliveredIds + overflow + failures: smaller collections, grouped together
2666+ atomic do
25672667 forwardFailures <- *forwardFailures
25682668 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f
25692669 ackOverflow <- *ackOverflow
@@ -2601,6 +2701,8 @@ pruneExpired = \t -> do
26012701 set *ipRates = do r <- ipRates; where t - r.windowStart < ipRateWindowMs; yield r
26022702 lookupRates <- *lookupRates
26032703 set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs; yield r
2704+ clientWriteRates <- *clientWriteRates
2705+ set *clientWriteRates = do r <- clientWriteRates; where t - r.windowStart < clientWriteRateWindowMs; yield r
26042706 verifyIpRates <- *verifyIpRates
26052707 set *verifyIpRates = do r <- verifyIpRates; where t - r.windowStart < verifyRateWindowMs; yield r
26062708 uniqueKeyIpRates <- *uniqueKeyIpRates
@@ -2794,9 +2896,10 @@ checkClientIpRate = \ip t ->
27942896 yield result.limited
27952897
27962898-- Combined pre-auth gate: drain check, global rate, IP rate. Returns "" if all pass.
2899+-- Blocks at drainPhase >= 1 (soft drain blocks new auth; existing sessions still work).
27972900checkPreAuth = \clientIp t -> do
2798− draining <- *draining
2799− if draining then yield "draining"
2901+ dp <- *drainPhase
2902+ if dp >= 1 then yield "draining"
28002903 else do
28012904 globalLimited <- checkGlobalRate t
28022905 if globalLimited then yield "global_rate_limited"
@@ -2896,8 +2999,8 @@ gossipOfflineIfLastSession : {pubkey: PubkeyHex, gossipOffline: Bool, peerServer
28962999gossipOfflineIfLastSession = \r label -> do
28973000 when r.gossipOffline (do
28983001 t <- now
2899− isDraining <- *draining
2900− if isDraining
3002+ dp <- *drainPhase
3003+ if dp >= 2
29013004 then logInfo (label ++ ": last session for " ++ r.pubkey ++ " (drain active, skipping redundant offline gossip)")
29023005 else do
29033006 logInfo (label ++ ": last session for " ++ r.pubkey ++ ", gossiping offline")
@@ -3055,7 +3158,10 @@ retryPendingForwards = \pending -> do
30553158-- When a remote key comes online, create forward entries for queued messages and send them
30563159forwardQueuedMessages = \targets t -> do
30573160 newForwards <- atomic do
3058− forwards <- readAllForwards
3161+ -- Read only the forward buckets needed for target pubkeys (not all 16).
3162+ -- All entries for a pubkey hash to the same bucket, so dedup is per-bucket complete.
3163+ let neededFwdBuckets = uniqueBucketsFor targets (\target -> target.pubkey)
3164+ forwards <- readSelectedFwdBuckets neededFwdBuckets
30593165 -- Read only the presence buckets needed for target pubkeys (not all 16)
30603166 let targetPresenceBuckets = uniqueBucketsFor targets (\target -> target.pubkey)
30613167 relevantPresence <- readSelectedPresenceBuckets targetPresenceBuckets
@@ -3067,7 +3173,9 @@ forwardQueuedMessages = \targets t -> do
30673173 let candidates = forwardCandidates remoteTargets relevantMessages forwards t
30683174 let totalFwdCount <- countAllForwards 0
30693175 let capped = takeRelation (maxForwardsEntries - totalFwdCount) candidates
3070− writeAllForwards (union forwards capped)
3176+ -- Write back only the affected forward buckets (not all 16)
3177+ let combined = union forwards capped
3178+ writeSelectedFwdBuckets neededFwdBuckets combined
30713179 yield capped
30723180 let servers = collectPeerServers newForwards (\x -> x.toServer)
30733181 failures <- *forwardFailures
@@ -3524,8 +3632,8 @@ pollOnce = \token cursor clientIp -> do
35243632 -- authenticateSession to skip safeCompare on unrelated sessions.
35253633 t2 <- now
35263634 result <- atomic do
3527− isDraining <- *draining
3528− if isDraining then yield {authorized: True {}, events: [], cursor: s.clampedCursor}
3635+ dp <- *drainPhase
3636+ if dp >= 2 then yield {authorized: True {}, events: [], cursor: s.clampedCursor}
35293637 else do
35303638 sessions2 <- readSessionBucket bucket
35313639 if not (reauthenticateSession sessions2 token s.pubkey t2)
@@ -3784,8 +3892,8 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
37843892 token <- genToken
37853893 let newBucket = tokenBucket token
37863894 authResult <- atomic do
3787− isDraining <- *draining
3788− if isDraining then yield (Nothing {})
3895+ dp <- *drainPhase
3896+ if dp >= 1 then yield (Nothing {})
37893897 else do
37903898 tpc <- *totalPresenceCount
37913899 knownPeers <- *knownPeerServers
@@ -3845,8 +3953,8 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
38453953 -- Tells the client their cursor was invalidated so they can handle re-delivery.
38463954 let cursorWasReset = generationMismatch && cursor > 0
38473955 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
3848− draining <- *draining
3849− if draining then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset})
3956+ dp <- *drainPhase
3957+ if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset})
38503958 else do
38513959 globalRateLimited <- checkGlobalRate t
38523960 if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset})
@@ -4030,8 +4138,8 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
40304138 (logWarn ("sendMessage: stale timestamp (" ++ show ((t - timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
40314139 let clampedTs = clampTimestamp timestamp t
40324140 result <- atomic do
4033− draining <- *draining
4034− if draining then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
4141+ dp <- *drainPhase
4142+ if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
40354143 else withSessionAuth authorization t clientIp
40364144 (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
40374145 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId t)
@@ -4055,8 +4163,8 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
40554163 if globalRateLimited then yield (respond {server: "", authorized: False {}})
40564164 else do
40574165 result <- atomic do
4058− draining <- *draining
4059− if draining then yield {server: "", authorized: False {}}
4166+ dp <- *drainPhase
4167+ if dp >= 2 then yield {server: "", authorized: False {}}
40604168 else withSessionAuth authorization t clientIp
40614169 (yield {server: "", authorized: False {}})
40624170 (\senderKey -> do
@@ -4081,8 +4189,8 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
40814189-- "delivered" (acked by recipient), "unknown" (expired or never existed).
40824190handleMessageStatus = \to messageId authorization clientIp respond -> do
40834191 t <- now
4084− draining <- *draining
4085− if draining then yield (respond {status: "", authorized: False {}})
4192+ dp <- *drainPhase
4193+ if dp >= 2 then yield (respond {status: "", authorized: False {}})
40864194 else do
40874195 globalRateLimited <- checkGlobalRate t
40884196 if globalRateLimited then yield (respond {status: "", authorized: False {}})
@@ -4225,8 +4333,13 @@ handleVerifyToken = \token scope clientIp respond -> do
42254333 else do
42264334 found <- atomic do
42274335 pending <- *pendingVerifications
4228− -- Scope is required: callers must specify the operation type (gossip/forward/ack)
4229− let match = any pending (\p -> safeCompare p.token token && p.scope == scope && t - p.createdAt < verificationTokenExpiry)
4336+ -- Pre-filter by scope, expiry, and first hex char before running expensive safeCompare.
4337+ -- Reduces scan from O(n) safeCompare calls to O(n/16) by filtering on token prefix.
4338+ -- The prefix filter is not constant-time, but single-use tokens with 24 bytes of randomness
4339+ -- make first-char leakage irrelevant (16^47 remaining possibilities in 30s window).
4340+ let firstChar = take 1 token
4341+ let candidates = do p <- pending; where p.scope == scope && t - p.createdAt < verificationTokenExpiry && take 1 p.token == firstChar; yield p
4342+ let match = any candidates (\p -> safeCompare p.token token)
42304343 when match (set *pendingVerifications = do p <- pending; where not (safeCompare p.token token); yield p)
42314344 yield match
42324345 -- Omit scope from log to prevent scope enumeration by attackers probing the endpoint
@@ -4480,15 +4593,16 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
44804593 writeAllForwards (forwardsExcluding forwards fromServer dedupedIds)
44814594 yield removed
44824595 let ackedCount = count removedForwards
4483− -- Only clear failure record when actual forwards were matched (proves the peer
4484− -- received and processed messages we sent). Unmatched acks don't prove forward delivery.
4485− when (ackedCount > 0) (clearFailure fromServer)
44864596 -- Guard: warn if peer acks messages we had no forward entries for. This can indicate
44874597 -- a peer acking messages it never received (e.g., the forward response was lost but
44884598 -- the message was never stored), which would cause premature origin message deletion.
44894599 let unmatched = count dedupedIds - ackedCount
44904600 when (unmatched > 0)
44914601 (logWarn ("federation/ack: " ++ fromServer ++ " acked " ++ show unmatched ++ " messageIds with no matching forward entries — skipping cleanup for those"))
4602+ -- Only clear failure record when ALL acked IDs matched forward entries.
4603+ -- Prevents a malicious peer from resetting its backoff by padding ack requests
4604+ -- with known messageIds that no longer have forward entries.
4605+ when (ackedCount > 0 && unmatched == 0) (clearFailure fromServer)
44924606 -- Deferred origin message cleanup: only queue cleanup for messages that actually had
44934607 -- matching forward entries. Messages acked without a forward entry are NOT cleaned up,
44944608 -- preventing premature deletion when a peer acks messages it never actually received.
@@ -4506,20 +4620,28 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
45064620-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
45074621-- Call before stopping the server so peers learn keys went offline immediately.
45084622-- Gossip is sent via fork; allow a few seconds after response before process exit.
4509−handleDrain = \authorization respond -> do
4623+-- phase: 0 or 2 = full drain (backward compat default), 1 = soft drain (block new auth only)
4624+handleDrain = \phase authorization respond -> do
45104625 t <- now
45114626 let authorized = verifyMaintenanceAuth authorization
4512− if not authorized then yield (respond {ok: False {}, drained: 0})
4627+ if not authorized then yield (respond {ok: False {}, drained: 0, phase: 0})
45134628 else do
4514− -- Idempotency: if already draining, return success without re-running drain logic.
4515− -- Prevents duplicate gossip broadcasts, snapshot exports, and flush forks.
4516− alreadyDraining <- *draining
4517− if alreadyDraining then do
4518− logInfo "drain: already draining, skipping"
4519− yield (respond {ok: True {}, drained: 0})
4629+ -- Effective phase: 0 or >=2 means full drain (backward compat for clients sending empty body)
4630+ let effectivePhase = if phase == 1 then 1 else 2
4631+ -- Idempotency: if already at or above requested phase, return success.
4632+ currentPhase <- *drainPhase
4633+ if currentPhase >= effectivePhase then do
4634+ logInfo ("drain: already at phase " ++ show currentPhase ++ ", skipping")
4635+ yield (respond {ok: True {}, drained: 0, phase: currentPhase})
4636+ -- Soft drain (phase 1): block new auth but keep serving existing sessions.
4637+ -- No gossip or flush needed since sessions are still active.
4638+ else if effectivePhase == 1 then do
4639+ atomic do set *drainPhase = 1
4640+ logInfo "drain: soft drain active (new auth blocked, existing sessions served)"
4641+ yield (respond {ok: True {}, drained: 0, phase: 1})
45204642 else do
45214643 drainResult <- atomic do
4522− set *draining = True {}
4644+ set *drainPhase = 2
45234645 allPresence <- readAllPresence
45244646 let localEntries = do p <- allPresence; where isLocalServer p.server && p.expiresAt > t; yield p
45254647 let remainingPresence = do p <- allPresence; where not (isLocalServer p.server) || p.expiresAt <= t; yield p
@@ -4527,7 +4649,10 @@ handleDrain = \authorization respond -> do
45274649 set *totalPresenceCount = count remainingPresence
45284650 set *presenceVersion = t
45294651 set *pollWakeup = t
4530− yield {localEntries: localEntries, peerServers: collectPeerServers allPresence (\p -> p.server)}
4652+ -- Use cached knownPeerServers instead of recomputing from allPresence
4653+ -- (knownPeerServers is updated every prune cycle, equivalent for gossip targets)
4654+ peerServers <- *knownPeerServers
4655+ yield {localEntries: localEntries, peerServers: peerServers}
45314656 let drainedCount = count drainResult.localEntries
45324657 -- Read pending federation state atomically (fix: was read outside atomic)
45334658 flushState <- atomic do
@@ -4585,7 +4710,7 @@ handleDrain = \authorization respond -> do
45854710 (logInfo ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))
45864711 when (pendingAckCount > 0)
45874712 (logInfo ("drain: flushing " ++ show pendingAckCount ++ " pending acks"))
4588− yield (respond {ok: True {}, drained: drainedCount})
4713+ yield (respond {ok: True {}, drained: drainedCount, phase: 2})
45894714
45904715-- POST /maintenance/undrain: Resume accepting requests after a drain.
45914716-- Background loops (prune, heartbeat) automatically resume on next iteration.
@@ -4594,20 +4719,40 @@ handleUndrain = \authorization respond -> do
45944719 if not authorized then yield (respond {ok: False {}})
45954720 else do
45964721 wasDraining <- atomic do
4597− d <- *draining
4598− set *draining = False {}
4722+ d <- *drainPhase
4723+ set *drainPhase = 0
45994724 set *drainCompletedAt = 0
4600− yield d
4725+ set *shutdownRequested = False {}
4726+ yield (d > 0)
46014727 when wasDraining (logInfo "undrain: server resumed, accepting new requests")
46024728 when (not wasDraining) (logInfo "undrain: server was not draining")
46034729 yield (respond {ok: True {}})
46044730
4731+-- POST /maintenance/shutdown: Trigger full drain + set shutdown flag.
4732+-- Background loops will exit after drain flush completes, allowing the process to stop.
4733+-- Unlike drain, shutdown is not undoable (requires process restart).
4734+handleShutdown = \authorization respond -> do
4735+ t <- now
4736+ let authorized = verifyMaintenanceAuth authorization
4737+ if not authorized then yield (respond {ok: False {}, drained: 0})
4738+ else do
4739+ dp <- *drainPhase
4740+ if dp >= 2 then do
4741+ -- Already draining, just set the shutdown flag
4742+ atomic do set *shutdownRequested = True {}
4743+ logInfo "shutdown: flag set (drain already active)"
4744+ yield (respond {ok: True {}, drained: 0})
4745+ else do
4746+ atomic do set *shutdownRequested = True {}
4747+ -- Delegate to handleDrain with phase=2 (full drain)
4748+ handleDrain 2 authorization (\drainResult -> respond {ok: drainResult.ok, drained: drainResult.drained})
4749+
46054750-- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).
46064751-- Accepts either federation or maintenance secret.
46074752handlePrune = \authorization respond -> do
46084753 t <- now
4609− draining <- *draining
4610− let authorized = verifyMaintenanceAuth authorization && not draining
4754+ dp <- *drainPhase
4755+ let authorized = verifyMaintenanceAuth authorization && dp < 2
46114756 when authorized (fork (pruneAndRetry t))
46124757 yield (respond {ok: authorized})
46134758
@@ -4626,8 +4771,8 @@ handleHealth = \authorization respond -> do
46264771 cF <- *cachedHealthFederation
46274772 cS <- *cachedHealthSnapshot
46284773 cW <- *cachedHealthWatchdog
4629− isDraining <- *draining
4630− yield {ok: cOk && not isDraining, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, draining: isDraining}
4774+ dp <- *drainPhase
4775+ yield {ok: cOk && dp == 0, pruneHealthy: cP, heartbeatHealthy: cH, capacityHealthy: cC, federationHealthy: cF, snapshotHealthy: cS, watchdogHealthy: cW, draining: dp > 0}
46314776 yield (respond cached)
46324777 else do
46334778 let isAuthed = verifyMaintenanceAuth authorization
@@ -4645,7 +4790,7 @@ handleHealth = \authorization respond -> do
46454790 forwardFailures <- *forwardFailures
46464791 pruneRun <- *pruneRunning
46474792 lastSnap <- *lastSnapshotExportAt
4648− isDraining <- *draining
4793+ dp <- *drainPhase
46494794 let warmup = t - sa < healthPruneMaxAge
46504795 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
46514796 -- Capacity pressure: flag when any capped collection exceeds 90% full
@@ -4671,7 +4816,7 @@ handleHealth = \authorization respond -> do
46714816 federationOk: fedOk,
46724817 snapshotOk: snapOk,
46734818 watchdogOk: wdOk,
4674− isDraining: isDraining}
4819+ isDraining: dp > 0}
46754820 -- Draining servers report ok=False so load balancers stop routing traffic to them
46764821 let allOk = not result.isDraining && result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk && result.snapshotOk && result.watchdogOk
46774822 -- Cache health result for rate-limited responses
@@ -4714,10 +4859,12 @@ handleStats = \authorization respond -> do
47144859 then do
47154860 -- Split into two atomic reads: core capacity state (consistent) and counters/timing
47164861 -- (slight inconsistency acceptable for monitoring). This narrows the lock window.
4862+ -- Forward stats computed per-bucket to avoid materializing all 16 buckets into one list.
4863+ -- Each bucket is read independently; counts and per-server stats are aggregated.
4864+ fwdStats <- computeForwardStats 0 t
47174865 core <- atomic do
47184866 sc <- *totalSessionCount
47194867 pc <- *totalPresenceCount
4720− f <- readAllForwards
47214868 c <- *challenges
47224869 a <- readAllAcked
47234870 ao <- *ackOverflow
@@ -4734,7 +4881,7 @@ handleStats = \authorization respond -> do
47344881 ger <- *gossipEventRates
47354882 vir <- *verifyIpRates
47364883 ukir <- *uniqueKeyIpRates
4737− yield {sc: sc, pc: pc, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg, fnCount: fnCount, pac: pac, bp: bp, ger: ger, vir: vir, ukir: ukir}
4884+ yield {sc: sc, pc: pc, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg, fnCount: fnCount, pac: pac, bp: bp, ger: ger, vir: vir, ukir: ukir}
47384885 counters <- atomic do
47394886 sr <- *sendRates
47404887 cr <- *challengeRates
@@ -4751,7 +4898,7 @@ handleStats = \authorization respond -> do
47514898 sa <- *startedAt
47524899 lp <- *lastPrunedAt
47534900 lh <- *lastHeartbeatAt
4754− dr <- *draining
4901+ dr <- *drainPhase
47554902 flt <- *federationLatencyTotal
47564903 flc <- *federationLatencyCount
47574904 flm <- *federationLatencyMax
@@ -4761,27 +4908,22 @@ handleStats = \authorization respond -> do
47614908 let sessions = core.sc
47624909 let messages = core.tmc
47634910 let presence = core.pc
4764− let fwds = count core.f
4911+ let fwds = fwdStats.totalCount
47654912 let msgPct = if maxMessages > 0 then messages * 100 / maxMessages else 0
47664913 let presPct = if maxPresenceEntries > 0 then presence * 100 / maxPresenceEntries else 0
47674914 let fwdPct = if maxForwardsEntries > 0 then fwds * 100 / maxForwardsEntries else 0
47684915 let sessPct = if maxSessions > 0 then sessions * 100 / maxSessions else 0
47694916 let fedAvg = if counters.flc > 0 then counters.flt / counters.flc else 0
47704917 let hungFedSlots = count (do r <- core.fi; where r.count > 0 && r.firstStartedAt > 0 && t - r.firstStartedAt >= federationFetchTimeoutMs; yield r)
4771− let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
4918+ let oldestFwdAge = fwdStats.oldestAge
47724919 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt
4773− -- Per-server forward queue summary: count unique servers and max queue depth
4774− -- Two-pass aggregation: collect unique servers, then count per server.
4775− -- Avoids O(n*k) accumulator rebuilding in the fold (no list comprehension per step).
4776− let fwdServers = fold (\acc f -> if any acc (\s -> s == f.toServer) then acc else union acc [f.toServer]) [] core.f
4777− let fwdCounts = do s <- fwdServers; yield {server: s, count: count (do f <- core.f; where f.toServer == s; yield f)}
4778− let fwdServerCount = count fwdCounts
4779− let maxFwdDepth = fold (\acc c -> max acc c.count) 0 fwdCounts
4920+ let fwdServerCount = fwdStats.serverCount
4921+ let maxFwdDepth = fwdStats.maxDepth
47804922 yield (respond {ok: True {},
47814923 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},
47824924 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},
47834925 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},
4784− timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr}})
4926+ timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr > 0}})
47854927 else yield (respond zeroStats)
47864928
47874929handleStatsPeers = \authorization respond -> do
@@ -4872,8 +5014,8 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
48725014 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
48735015 else do
48745016 t <- now
4875− isDraining <- *draining
4876− if not isDraining then yield (respond {ok: False {}, error: "server_must_be_draining", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
5017+ dp <- *drainPhase
5018+ if dp < 2 then yield (respond {ok: False {}, error: "server_must_be_draining", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
48775019 else do
48785020 -- Validate imported data: reject entries with invalid or self-referencing server names.
48795021 -- originServer in ForwardEntry is untyped Text (not ServerName), so not validated by the type system.
@@ -4951,10 +5093,11 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
49515093 withSessionAuth authorization t clientIp
49525094 (yield (respond {ok: False {}, authorized: False {}, ackedCount: 0}))
49535095 (\pubkey -> do
5096+ -- Uses clientWriteRates (separate from lookupRates) so ack bursts don't starve lookups
49545097 ackRateLimited <- atomic do
4955− lookupRates <- *lookupRates
4956− let rateResult = checkPubkeyRate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
4957− set *lookupRates = rateResult.entries
5098+ cwr <- *clientWriteRates
5099+ let rateResult = checkPubkeyRate cwr pubkey maxClientWriteRatePerKey clientWriteRateWindowMs maxClientWriteRateEntries t
5100+ set *clientWriteRates = rateResult.entries
49585101 yield rateResult.limited
49595102 if ackRateLimited then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0})
49605103 else do
@@ -4992,8 +5135,8 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
49925135-- Rate-limited via lookup rate limiter (read-only info lookup).
49935136handleListMessages = \authorization clientIp respond -> do
49945137 t <- now
4995− draining <- *draining
4996− if draining then yield (respond {messages: [], authorized: False {}, count: 0})
5138+ dp <- *drainPhase
5139+ if dp >= 2 then yield (respond {messages: [], authorized: False {}, count: 0})
49975140 else do
49985141 globalRateLimited <- checkGlobalRate t
49995142 if globalRateLimited then yield (respond {messages: [], authorized: False {}, count: 0})
@@ -5022,8 +5165,8 @@ handleListMessages = \authorization clientIp respond -> do
50225165-- their forward entries instead of retrying until prune.
50235166handleDeleteMessages = \messageIds authorization clientIp respond -> do
50245167 t <- now
5025− draining <- *draining
5026− if draining then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0})
5168+ dp <- *drainPhase
5169+ if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0})
50275170 else do
50285171 globalRateLimited <- checkGlobalRate t
50295172 if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0})
@@ -5031,10 +5174,11 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
50315174 withSessionAuth authorization t clientIp
50325175 (yield (respond {ok: False {}, authorized: False {}, deletedCount: 0}))
50335176 (\pubkey -> do
5177+ -- Uses clientWriteRates (separate from lookupRates) so delete bursts don't starve lookups
50345178 deleteRateLimited <- atomic do
5035− lookupRates <- *lookupRates
5036− let rateResult = checkPubkeyRate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
5037− set *lookupRates = rateResult.entries
5179+ cwr <- *clientWriteRates
5180+ let rateResult = checkPubkeyRate cwr pubkey maxClientWriteRatePerKey clientWriteRateWindowMs maxClientWriteRateEntries t
5181+ set *clientWriteRates = rateResult.entries
50385182 yield rateResult.limited
50395183 if deleteRateLimited then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0})
50405184 else do
@@ -5089,7 +5233,10 @@ handleSendBatch = \messages authorization clientIp respond -> do
50895233 -- dedup between messages in the same batch is not caught by STM.
50905234 -- Messages with empty messageId (auto-generated) are always kept.
50915235 let capped = fold (\acc m -> if m.messageId != "" && any acc (\a -> a.messageId == m.messageId) then acc else union acc [m]) [] sized
5092− results <- sendBatchInner senderKey capped t []
5236+ -- Sort by pubkeyBucket(to) so sequential atomic blocks in sendBatchInner benefit
5237+ -- from STM cache locality when multiple messages target the same recipient bucket.
5238+ let sorted = sortBy (\m -> pubkeyBucket m.to) capped
5239+ results <- sendBatchInner senderKey sorted t []
50935240 -- Fork federation forwarding: one forked task per target server (not per message).
50945241 -- Within each task, forwards are sent sequentially to avoid per-message fork overhead.
50955242 let forwardTargets = do r <- results; where count r.remoteServers > 0; yield r
@@ -5125,8 +5272,8 @@ sendBatchInner = \senderKey messages t acc ->
51255272 else do
51265273 let clampedTs = clampTimestamp msg.timestamp t
51275274 sendResult <- atomic do
5128− draining <- *draining
5129− if draining then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
5275+ dp <- *drainPhase
5276+ if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
51305277 else sendMessageCore senderKey msg.to msg.encryptedBlob clampedTs effectiveMsgId t
51315278 case sendResult.status of
51325279 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
@@ -5342,6 +5489,7 @@ requestPath = \req -> case req of
53425489 VerifyToken {} -> "POST /federation/verify"
53435490 Prune {} -> "POST /maintenance/prune"
53445491 Drain {} -> "POST /maintenance/drain"
5492+ Shutdown {} -> "POST /maintenance/shutdown"
53455493 Undrain {} -> "POST /maintenance/undrain"
53465494 Snapshot {} -> "GET /maintenance/snapshot"
53475495 Restore {} -> "POST /maintenance/restore"
@@ -5379,6 +5527,7 @@ requestClientIp = \req -> case req of
53795527 FederationAck {} -> ""
53805528 Prune {} -> ""
53815529 Drain {} -> ""
5530+ Shutdown {} -> ""
53825531 Undrain {} -> ""
53835532 Snapshot {} -> ""
53845533 Restore {} -> ""
@@ -5418,7 +5567,8 @@ serveInner = \req -> case req of
54185567 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
54195568 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
54205569 Prune {authorization, respond} -> handlePrune authorization respond
5421− Drain {authorization, respond} -> handleDrain authorization respond
5570+ Drain {phase, authorization, respond} -> handleDrain phase authorization respond
5571+ Shutdown {authorization, respond} -> handleShutdown authorization respond
54225572 Undrain {authorization, respond} -> handleUndrain authorization respond
54235573 Snapshot {authorization, respond} -> handleSnapshot authorization respond
54245574 Restore {version, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, blockedPeers, authorization, respond} -> handleRestore version presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces blockedPeers authorization respond
@@ -5470,8 +5620,12 @@ serve = \req -> do
54705620backgroundPrune = do
54715621 jitter <- randomInt 5000
54725622 sleep (backgroundPruneInterval + jitter)
5473− draining <- *draining
5474− if draining then do
5623+ dp <- *drainPhase
5624+ -- Check shutdown: if drain completed and shutdown requested, exit the loop
5625+ shutdownReq <- *shutdownRequested
5626+ dca <- *drainCompletedAt
5627+ if shutdownReq && dca > 0 then logInfo "backgroundPrune: shutdown complete, exiting loop"
5628+ else if dp >= 2 then do
54755629 sleep backgroundPruneInterval
54765630 backgroundPrune
54775631 else do
@@ -5545,8 +5699,11 @@ backgroundPrune = do
55455699pollHeartbeat = do
55465700 jitter <- randomInt 2000
55475701 sleep (pollTimeoutMs + jitter)
5548− draining <- *draining
5549− if draining then do
5702+ dp <- *drainPhase
5703+ shutdownReq <- *shutdownRequested
5704+ dca <- *drainCompletedAt
5705+ if shutdownReq && dca > 0 then logInfo "pollHeartbeat: shutdown complete, exiting loop"
5706+ else if dp >= 2 then do
55505707 t <- now
55515708 atomic do set *pollWakeup = t
55525709 -- Don't exit; sleep and retry so undrain can resume the loop
@@ -5569,8 +5726,11 @@ pollHeartbeat = do
55695726-- (e.g., /var/lib/skrepka/snapshot.json) in addition to logging.
55705727backgroundSnapshot = do
55715728 sleep autoSnapshotInterval
5572− draining <- *draining
5573− if not draining then do
5729+ dp <- *drainPhase
5730+ shutdownReq <- *shutdownRequested
5731+ dca <- *drainCompletedAt
5732+ if shutdownReq && dca > 0 then logInfo "backgroundSnapshot: shutdown complete, exiting loop"
5733+ else if dp < 2 then do
55745734 result <- atomic do
55755735 tmc <- *totalMessageCount
55765736 sc <- *totalSessionCount
@@ -5589,7 +5749,12 @@ backgroundSnapshot = do
55895749 when (autoSnapshotExportInterval > 0) (do
55905750 lastAutoExport <- *lastAutoSnapshotExportAt
55915751 let hasState = result.presence > 0 || result.forwards > 0 || result.ackedIds > 0 || result.deliveredIds > 0
5592− when (hasState && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do
5752+ -- Fingerprint: skip export when key counters haven't changed since last export.
5753+ -- Uses multiplicative hash to distinguish {messages:1,sessions:0} from {messages:0,sessions:1}.
5754+ let fingerprint = result.messages + result.sessions * 100003 + result.presence * 100019 + result.forwards * 100043 + result.ackedIds * 100049 + result.deliveredIds * 100057
5755+ lastFp <- *lastSnapshotFingerprint
5756+ let stateChanged = lastFp == 0 || fingerprint != lastFp
5757+ when (hasState && stateChanged && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do
55935758 exportData <- atomic do
55945759 p <- readAllPresence
55955760 a <- readAllAcked
@@ -5601,7 +5766,9 @@ backgroundSnapshot = do
56015766 bp <- *blockedPeers
56025767 yield {version: snapshotVersion, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}
56035768 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
5604− atomic do set *lastAutoSnapshotExportAt = t
5769+ atomic do
5770+ set *lastAutoSnapshotExportAt = t
5771+ set *lastSnapshotFingerprint = fingerprint
56055772 logInfo ("snapshot: auto-exported full state (" ++ show result.presence ++ " presence, " ++ show result.forwards ++ " forwards)")))
56065773 -- Warn if federation state hasn't been exported recently (will be lost on restart)
56075774 lastExport <- *lastSnapshotExportAt
@@ -5679,6 +5846,15 @@ startupProbeNext = \peers probed ->
56795846 Ok r -> logInfo ("startup: peer " ++ server ++ " reachable (" ++ show latency ++ "ms)")
56805847 startupProbeNext rest (probed + 1)
56815848
5849+-- Reset federation latency metrics after startup probes so unauthenticated probe
5850+-- results (expected "unauthorized" responses) don't skew initial /stats readings.
5851+resetFederationLatencyMetrics = atomic do
5852+ set *federationLatencyTotal = 0
5853+ set *federationLatencyCount = 0
5854+ set *federationLatencyMax = 0
5855+ t <- now
5856+ set *federationLatencyResetAt = t
5857+
56825858main = do
56835859 let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
56845860 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
@@ -5767,5 +5943,6 @@ main = do
57675943 when federationEnabled (fork (do
57685944 sleep 5000 -- wait for listener to be ready
57695945 logInfo "startup: probing federation peers..."
5770− startupProbePeers))
5946+ startupProbePeers
5947+ resetFederationLatencyMetrics))
57715948 listen 8080 serve