@@ -322,6 +322,8 @@ type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
322322
323323*recentGossip : [RecentGossipEntry]
324324
325+*lastFederationSuccessAt : Timestamp
326+
325327-- ============================================
326328-- Routes
327329-- ============================================
@@ -381,7 +383,7 @@ route MaintenanceApi where
381383 = Drain
382384
383385route HealthApi where
384− GET /health -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool} = Health
386+ GET /health -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool} = Health
385387
386388-- Returns counts of in-memory collections (requires federation or maintenance secret)
387389route StatsApi where
@@ -708,7 +710,11 @@ safeCompareLoop = \s1 s2 n result ->
708710safeCompare : Text -> Text -> Bool
709711safeCompare = \a b ->
710712 if a == "" || b == "" then False {}
711− else safeCompareLoop a b maxTokenLen 0 == 0
713+ else
714+ let lenA = count (chars a) in
715+ let lenB = count (chars b) in
716+ let lenDiff = if lenA == lenB then 0 else 1 in
717+ safeCompareLoop a b maxTokenLen 0 + lenDiff == 0
712718
713719-- Generic sliding-window rate limiter. Returns whether the request is rate-limited
714720-- and the updated entries list. Callers provide match/create/update functions for their entry type.
@@ -922,18 +928,6 @@ pruneExpired = \t -> do
922928 set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
923929 yield presenceResult
924930
925−-- Opportunistic pruning: runs at most once per second, triggered by request handlers
926−maybePrune = \t -> do
927− shouldRun <- atomic do
928− lastPruned <- *lastPrunedAt
929− if t - lastPruned < 1000
930− then yield (False {})
931− else do
932− set *lastPrunedAt = t
933− yield (True {})
934− if not shouldRun
935− then yield {expiredLocal: [], peerServers: []}
936− else pruneExpired t
937931
938932splitAtN = \items n ->
939933 fold (\acc e ->
@@ -967,8 +961,6 @@ broadcastOfflineGossip = \t peerServers expiredLocal ->
967961 let gossipEvents = mergeByKey (\a b -> a.pubkey == b.pubkey) (\x -> x.timestamp) [] (offlineGossipEvents expiredLocal)
968962 sendBatchedGossip peerServers gossipEvents gossipOriginHops)
969963
970−forkBroadcastOffline = \t result ->
971− fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
972964
973965effectiveGossipEvents = \entries appliedKeys eventType -> do
974966 e <- entries
@@ -987,9 +979,6 @@ mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
987979 gossipPeerServers: collectPeerServers capResult.finalPresence (\p -> p.server),
988980 withinCapacity: capResult.withinCapacity}
989981
990−maybePruneAndBroadcast = \t -> do
991− pruneResult <- maybePrune t
992− forkBroadcastOffline t pruneResult
993982
994983checkFederationRate = \server t -> atomic do
995984 federationRates <- *federationRates
@@ -1040,26 +1029,20 @@ checkPreAuth = \clientIp t -> do
10401029
10411030withFederationAuth = \authorization t respond fromServer precondition action ->
10421031 if not (verifyFederationAuthForServer fromServer authorization)
1043− then do
1044− maybePruneAndBroadcast t
1045− yield (respond {ok: False {}, error: "unauthorized"})
1032+ then yield (respond {ok: False {}, error: "unauthorized"})
10461033 else if not precondition
1047− then do
1048− maybePruneAndBroadcast t
1049− yield (respond {ok: False {}, error: "invalid_request"})
1034+ then yield (respond {ok: False {}, error: "invalid_request"})
10501035 else do
10511036 globalLimited <- checkGlobalRate t
10521037 if globalLimited
10531038 then do
10541039 println ("federation: global rate limited (" ++ show maxGlobalRequestRate ++ "/s), rejecting for " ++ fromServer)
1055− maybePruneAndBroadcast t
10561040 yield (respond {ok: False {}, error: "rate_limited"})
10571041 else do
10581042 rateLimited <- checkFederationRate fromServer t
10591043 if rateLimited
10601044 then do
10611045 println ("federation: rate limited (" ++ show maxFederationRatePerServer ++ "/s per server), rejecting for " ++ fromServer)
1062− maybePruneAndBroadcast t
10631046 yield (respond {ok: False {}, error: "rate_limited"})
10641047 else do
10651048 clearFailure fromServer
@@ -1102,9 +1085,12 @@ recordFailure = \server -> do
11021085 Just {value: f} -> union (do x <- failures; where x.server != server; yield x) [{server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}]
11031086 set *forwardFailures = if count updated > maxForwardFailures then evictOneOldest updated (\x -> x.failedAt) else updated
11041087
1105−clearFailure = \server -> atomic do
1106− failures <- *forwardFailures
1107− set *forwardFailures = do f <- failures; where f.server != server; yield f
1088+clearFailure = \server -> do
1089+ t <- now
1090+ atomic do
1091+ failures <- *forwardFailures
1092+ set *forwardFailures = do f <- failures; where f.server != server; yield f
1093+ set *lastFederationSuccessAt = t
11081094
11091095retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
11101096retryForwardsForServer = \entries remaining ->
@@ -1266,19 +1252,29 @@ pruneAndRetry = \t -> do
12661252 lastRefresh <- *lastPresenceRefresh
12671253 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval
12681254 presence <- *presence
1269− let localActive = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []
1255+ sessions <- *sessions
1256+ -- Extend presence for keys with active sessions nearing expiry (sliding window)
1257+ let localNearExpiry = if shouldRefresh then (do p <- presence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p) else []
1258+ let toExtend = do p <- localNearExpiry; where any sessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
1259+ let notExtended = do p <- localNearExpiry; where none sessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
1260+ when (count toExtend > 0) (
1261+ set *presence = do pp <- presence; yield (if any toExtend (\e -> e.pubkey == pp.pubkey && isLocalServer pp.server) then {pubkey: pp.pubkey, server: pp.server, expiresAt: t + sessionExpiry, timestamp: t} else pp))
12701262 let refreshPeers = if shouldRefresh then collectPeerServers presence (\p -> p.server) else []
12711263 when shouldRefresh (set *lastPresenceRefresh = t)
1272− yield {pendingForwards: forwards, pendingAcks: pendingAcks, pendingOverflow: pendingOverflow, localActive: localActive, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
1273− broadcastOfflineGossip t pr.peerServers pr.expiredLocal
1274− when federationEnabled (do
1275− retryPendingForwards result.pendingForwards
1276− sendAcksAndClean result.pendingAcks
1277− retryOverflowAcks result.pendingOverflow)
1264+ yield {pendingForwards: forwards, pendingAcks: pendingAcks, pendingOverflow: pendingOverflow, extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: refreshPeers, shouldRefresh: shouldRefresh}
1265+ -- Fork federation I/O so hung peers don't block next prune cycle
1266+ fork (do
1267+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal
1268+ when federationEnabled (do
1269+ retryPendingForwards result.pendingForwards
1270+ sendAcksAndClean result.pendingAcks
1271+ retryOverflowAcks result.pendingOverflow))
12781272 when result.shouldRefresh (do
1279− let refreshEvents = do p <- result.localActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.timestamp, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000)}
1273+ let extendedEvents = do p <- result.extendedActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000)}
1274+ let nearExpiryEvents = do p <- result.nearExpiryActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.timestamp, ttl: clampTtl ((p.expiresAt - p.timestamp) / 1000)}
1275+ let refreshEvents = union extendedEvents nearExpiryEvents
12801276 when (count result.refreshPeers > 0 && count refreshEvents > 0)
1281− (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops))
1277+ (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
12821278
12831279hexCharIndex : Text -> Int
12841280hexCharIndex = \c ->
@@ -1379,6 +1375,8 @@ pollOnce = \token cursor -> do
13791375 case authenticateSession sessions token t of
13801376 Unauthorized {} -> yield (Nothing {})
13811377 Authorized {pubkey} -> do
1378+ -- Refresh session expiry (sliding window keeps active sessions alive)
1379+ set *sessions = do s <- sessions; yield (if s.token == token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry} else s)
13821380 seqCounter <- *seqCounter
13831381 let clampedCursor = clamp 0 (max seqCounter t) cursor
13841382 ackResult <- if clampedCursor > 0
@@ -1570,7 +1568,6 @@ handleChallenge = \pubkey clientIp respond -> do
15701568 yield (respond {challenge: "", ok: False {}, expiresAt: 0})
15711569 else do
15721570 challenge <- genToken
1573− pr <- maybePrune t
15741571 result <- atomic do
15751572 challenges <- *challenges
15761573 let active = do c <- challenges; where c.expiresAt > t; yield c
@@ -1598,7 +1595,6 @@ handleChallenge = \pubkey clientIp respond -> do
15981595 (println ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
15991596 when (not result.accepted && not result.rateLimited)
16001597 (println ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
1601− forkBroadcastOffline t pr
16021598 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
16031599
16041600-- POST /auth/verify: Verify Ed25519 signature over challenge string.
@@ -1628,7 +1624,6 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
16281624verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do
16291625 let expiry = t + sessionExpiry
16301626 let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
1631− pr <- maybePrune t
16321627 validation <- atomic do
16331628 challenges <- *challenges
16341629 let hasMatch = challengeMatch challenges pubkey challenge clientIp t
@@ -1654,15 +1649,10 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
16541649 yield (if federationEnabled
16551650 then collectPeerServers updatedPresence (\p -> p.server)
16561651 else [])
1657− let expiredLocal = do p <- pr.expiredLocal; where p.pubkey != pubkey; yield p
1658− fork (do
1659− when federationEnabled
1660− (sendBatchedGossip peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops)
1661− broadcastOfflineGossip t pr.peerServers expiredLocal)
1652+ fork (when federationEnabled
1653+ (sendBatchedGossip peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops))
16621654 yield (respond {token: token, expiresAt: expiry})
1663− else do
1664− forkBroadcastOffline t pr
1665− yield (respond {token: "", expiresAt: 0})
1655+ else yield (respond {token: "", expiresAt: 0})
16661656
16671657-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor
16681658-- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.
@@ -1678,8 +1668,6 @@ handlePoll = \cursor authorization respond -> do
16781668 result <- case extractToken authorization of
16791669 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
16801670 Just {value: token} -> pollOnce token cursor
1681− tAfterPoll <- now
1682− maybePruneAndBroadcast tAfterPoll
16831671 fork (when federationEnabled (do
16841672 sendAcksAndClean result.pendingAcks
16851673 retryEvictedAcks result.evictedEntries result.pendingAcks))
@@ -1698,7 +1686,6 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
16981686 else do
16991687 effectiveMsgId <- if messageId != "" then yield messageId else genToken
17001688 let clampedTs = clampTimestamp timestamp t
1701− pr <- maybePrune t
17021689 result <- atomic do
17031690 withSessionAuth authorization t
17041691 (yield {send: SendFailed {status: MsUnauthorized {}}, senderRateLimited: False {}, globalCapReached: False {}})
@@ -1760,9 +1747,13 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
17601747 (println ("sendMessage: global message cap reached (" ++ show maxMessages ++ "), rejecting"))
17611748 case result.send of
17621749 SendDone {delivery, remoteServers} ->
1763− fork (do
1764− when (count remoteServers > 0 && federationEnabled) (
1765− sendToEach remoteServers (\server -> do
1750+ when (count remoteServers > 0 && federationEnabled) (
1751+ fork (sendToEach remoteServers (\server -> do
1752+ -- Check if forward still exists (recipient may have acked during fork delay)
1753+ stillPending <- atomic do
1754+ forwards <- *forwards
1755+ yield (any forwards (\f -> f.messageId == effectiveMsgId && f.toServer == server))
1756+ when stillPending (do
17661757 fwdResult <- fetch (federationUrl server) (ForwardMessage {
17671758 toKey: to, encryptedBlob: encryptedBlob,
17681759 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
@@ -1773,9 +1764,8 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
17731764 recordFailure server
17741765 Ok _ -> do
17751766 removeForward effectiveMsgId server
1776− clearFailure server))
1777− broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
1778− _ -> forkBroadcastOffline t pr
1767+ clearFailure server))))
1768+ _ -> yield {}
17791769 yield (case result.send of
17801770 SendFailed {status} -> respond {status: status, messageId: ""}
17811771 SendDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId}
@@ -1791,7 +1781,6 @@ handleLookupPresence = \pubkey authorization respond -> do
17911781 globalRateLimited <- checkGlobalRate t
17921782 if globalRateLimited then yield (respond {server: "", authorized: False {}})
17931783 else do
1794− pr <- maybePrune t
17951784 result <- atomic do
17961785 withSessionAuth authorization t
17971786 (yield {server: "", authorized: False {}})
@@ -1811,7 +1800,6 @@ handleLookupPresence = \pubkey authorization respond -> do
18111800 -- not which specific peer server a key is homed on (prevents metadata leakage)
18121801 let maskedServer = if foundServer == "" then "" else if isLocalServer foundServer then foundServer else "remote"
18131802 yield {server: maskedServer, authorized: True {}})
1814− forkBroadcastOffline t pr
18151803 yield (respond {server: result.server, authorized: result.authorized})
18161804
18171805-- POST /federation/gossip: Receive presence gossip from a peer server.
@@ -1828,7 +1816,6 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do
18281816 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
18291817 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
18301818 let deduped = mergeByPubkeyServer [] filteredOnline
1831− pr <- maybePrune t
18321819 gossipResult <- atomic do
18331820 presence <- *presence
18341821 let gm = mergeGossipPresence presence deduped trustedOffline fromServer t
@@ -1850,7 +1837,6 @@ handleRecvGossip = \events hopsRemaining fromServer authorization respond -> do
18501837 fork (do
18511838 when (cappedHops > 1 && federationEnabled && count freshEvents > 0) (
18521839 sendBatchedGossip (gossipTargets gossipResult.gm.gossipPeerServers fromServer) freshEvents (cappedHops - 1))
1853− broadcastOfflineGossip t pr.peerServers pr.expiredLocal
18541840 when (federationEnabled && count newlyOnlineRemote > 0)
18551841 (forwardQueuedMessages newlyOnlineRemote t))
18561842 yield (respond {ok: True {}, error: ""}))
@@ -1862,7 +1848,6 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
18621848 t <- now
18631849 withFederationAuth authorization t respond fromServer (not (isLocalServer originServer) && fromServer == originServer) (do
18641850 let clampedTs = clampTimestamp timestamp t
1865− pr <- maybePrune t
18661851 result <- atomic do
18671852 messages <- *messages
18681853 ackedIds <- *ackedIds
@@ -1900,7 +1885,6 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
19001885 then println ("forward: global message cap reached (" ++ show maxMessages ++ "), rejecting for " ++ toKey)
19011886 else println ("forward: rejected for " ++ toKey ++ " (mailbox full, max " ++ show maxMessagesPerRecipient ++ ")")
19021887 _ -> yield {}
1903− forkBroadcastOffline t pr
19041888 yield (case result.forward of
19051889 ForwardRejected {} -> respond {ok: False {}, error: "mailbox_full"}
19061890 _ -> respond {ok: True {}, error: ""}))
@@ -1912,7 +1896,6 @@ handleFederationAck = \messageIds originServer fromServer authorization respond
19121896 t <- now
19131897 withFederationAuth authorization t respond fromServer (isLocalServer originServer && not (isLocalServer fromServer)) (do
19141898 let dedupedIds = dedupMessageIds messageIds
1915− pr <- maybePrune t
19161899 atomic do
19171900 forwards <- *forwards
19181901 let ackedForwards = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
@@ -1926,7 +1909,6 @@ handleFederationAck = \messageIds originServer fromServer authorization respond
19261909 where none presence (\p -> p.pubkey == f.toKey && isLocalServer p.server && p.expiresAt > t)
19271910 yield f
19281911 set *messages = do m <- messages; where none removable (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m
1929− forkBroadcastOffline t pr
19301912 yield (respond {ok: True {}, error: ""}))
19311913
19321914-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
@@ -1949,9 +1931,10 @@ handleDrain = \authorization respond -> do
19491931 ackedIds <- *ackedIds
19501932 let pendingForwardCount = count forwards
19511933 let pendingAckCount = count ackedIds
1952− fork (when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
1934+ -- Block on gossip (don't fork) so drain completes before process exit
1935+ when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
19531936 let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
1954− sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops))
1937+ sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
19551938 println ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
19561939 when (pendingForwardCount > 0)
19571940 (println ("drain: warning: " ++ show pendingForwardCount ++ " pending forwards will not be retried"))
@@ -1963,42 +1946,55 @@ handleDrain = \authorization respond -> do
19631946-- Accepts either federation or maintenance secret.
19641947handlePrune = \authorization respond -> do
19651948 t <- now
1966− let authorized = verifyMaintenanceAuth authorization
1949+ draining <- *draining
1950+ let authorized = verifyMaintenanceAuth authorization && not draining
19671951 when authorized (fork (pruneAndRetry t))
19681952 yield (respond {ok: authorized})
19691953
19701954handleHealth = \respond -> do
19711955 t <- now
1972− result <- atomic do
1973− lastPrune <- *lastPrunedAt
1974− lastHb <- *lastHeartbeatAt
1975− sa <- *startedAt
1976− presence <- *presence
1977− forwards <- *forwards
1978− sessions <- *sessions
1979− activePolls <- *activePolls
1980− messages <- *messages
1981− let warmup = t - sa < healthPruneMaxAge
1982− -- Capacity pressure: flag when any capped collection exceeds 90% full
1983− let capacityOk = count messages * 10 <= maxMessages * 9
1984− && count presence * 10 <= maxPresenceEntries * 9
1985− && count forwards * 10 <= maxForwardsEntries * 9
1986− && count sessions * 10 <= maxSessions * 9
1987− && count activePolls * 10 <= maxActivePolls * 9
1988− yield {pruneOk: warmup || t - lastPrune < healthPruneMaxAge,
1989− heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
1990− capacityOk: capacityOk}
1991− yield (respond {ok: result.pruneOk && result.heartbeatOk && result.capacityOk,
1992− pruneHealthy: result.pruneOk,
1993− heartbeatHealthy: result.heartbeatOk,
1994− capacityHealthy: result.capacityOk})
1956+ globalLimited <- checkGlobalRate t
1957+ if globalLimited then yield (respond {ok: False {}, pruneHealthy: False {}, heartbeatHealthy: False {}, capacityHealthy: False {}, federationHealthy: False {}})
1958+ else do
1959+ result <- atomic do
1960+ lastPrune <- *lastPrunedAt
1961+ lastHb <- *lastHeartbeatAt
1962+ sa <- *startedAt
1963+ presence <- *presence
1964+ forwards <- *forwards
1965+ sessions <- *sessions
1966+ activePolls <- *activePolls
1967+ messages <- *messages
1968+ lastFedSuccess <- *lastFederationSuccessAt
1969+ forwardFailures <- *forwardFailures
1970+ let warmup = t - sa < healthPruneMaxAge
1971+ -- Capacity pressure: flag when any capped collection exceeds 90% full
1972+ let capacityOk = count messages * 10 <= maxMessages * 9
1973+ && count presence * 10 <= maxPresenceEntries * 9
1974+ && count forwards * 10 <= maxForwardsEntries * 9
1975+ && count sessions * 10 <= maxSessions * 9
1976+ && count activePolls * 10 <= maxActivePolls * 9
1977+ -- Federation: healthy if disabled, in warmup, no peer issues, or recent success
1978+ let fedOk = not federationEnabled || warmup
1979+ || count forwardFailures == 0
1980+ || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
1981+ yield {pruneOk: warmup || t - lastPrune < healthPruneMaxAge,
1982+ heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
1983+ capacityOk: capacityOk,
1984+ federationOk: fedOk}
1985+ yield (respond {ok: result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk,
1986+ pruneHealthy: result.pruneOk,
1987+ heartbeatHealthy: result.heartbeatOk,
1988+ capacityHealthy: result.capacityOk,
1989+ federationHealthy: result.federationOk})
19951990
19961991handleStats = \authorization respond -> do
19971992 let authorized = verifyMaintenanceAuth authorization
19981993 if authorized
19991994 then do
20001995 t <- now
2001− result <- atomic do
1996+ -- Read all refs atomically for a consistent snapshot, count outside to minimize lock time
1997+ snapshot <- atomic do
20021998 s <- *sessions
20031999 m <- *messages
20042000 p <- *presence
@@ -2021,7 +2017,8 @@ handleStats = \authorization respond -> do
20212017 sa <- *startedAt
20222018 lp <- *lastPrunedAt
20232019 lh <- *lastHeartbeatAt
2024− yield {sessions: count s, messages: count m, presence: count p, forwards: count f, challenges: count c, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff, activePolls: count ap, sendRates: count sr, challengeRates: count cr, federationRates: count fr, globalRateCount: grc, ipRates: count ir, lookupRates: count lr, recvRates: count rr, pruneCount: pc, heartbeatCount: hc, uptime: t - sa, lastPruneAt: lp, lastHeartbeatAt: lh}
2020+ yield {s: s, m: m, p: p, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, sr: sr, cr: cr, fr: fr, grc: grc, ir: ir, lr: lr, rr: rr, pc: pc, hc: hc, sa: sa, lp: lp, lh: lh}
2021+ let result = {sessions: count snapshot.s, messages: count snapshot.m, presence: count snapshot.p, forwards: count snapshot.f, challenges: count snapshot.c, ackedIds: count snapshot.a, ackOverflow: count snapshot.ao, deliveredIds: count snapshot.d, forwardFailures: count snapshot.ff, activePolls: count snapshot.ap, sendRates: count snapshot.sr, challengeRates: count snapshot.cr, federationRates: count snapshot.fr, globalRateCount: snapshot.grc, ipRates: count snapshot.ir, lookupRates: count snapshot.lr, recvRates: count snapshot.rr, pruneCount: snapshot.pc, heartbeatCount: snapshot.hc, uptime: t - snapshot.sa, lastPruneAt: snapshot.lp, lastHeartbeatAt: snapshot.lh}
20252022 yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, ok: True {}})
20262023 else yield (respond {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pruneCount: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, ok: False {}})
20272024
@@ -2084,6 +2081,14 @@ backgroundPrune = do
20842081 if draining then println "backgroundPrune: stopping (server draining)"
20852082 else do
20862083 t <- now
2084+ -- Heartbeat watchdog: bump pollWakeup if heartbeat loop appears dead
2085+ sa <- *startedAt
2086+ lastHb <- *lastHeartbeatAt
2087+ when (t - sa > healthHeartbeatMaxAge && t - lastHb > healthHeartbeatMaxAge) (do
2088+ println "backgroundPrune: heartbeat appears dead, bumping pollWakeup"
2089+ atomic do
2090+ set *pollWakeup = t
2091+ set *lastHeartbeatAt = t)
20872092 status <- atomic do
20882093 running <- *pruneRunning
20892094 if running > 0 && t - running < backgroundPruneInterval * 3