@@ -80,6 +80,10 @@ maxFederationRateEntries = 100
8080maxGlobalRequestRate = 1000 -- max total requests per window
8181globalRateWindowMs : Int<Ms>
8282globalRateWindowMs = 1000
83+-- Auth-tier global rate limiting (prevents auth flooding from starving client operations)
84+maxAuthGlobalRate = 200 -- max auth requests (challenge + verify + revoke) per window
85+authGlobalRateWindowMs : Int<Ms>
86+authGlobalRateWindowMs = 1000
8387-- Session and gossip timing
8488sessionExpiry : Int<Ms>
8589sessionExpiry = 3600000 -- 1 hour
@@ -402,6 +406,16 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
402406-- Last time federation latency metrics were reset (for windowed averages)
403407*federationLatencyResetAt : Timestamp
404408
409+-- Total session count across all buckets (avoids O(16) recount on monitoring paths)
410+*totalSessionCount : Int
411+
412+-- Auth-tier global rate limiting state
413+*authGlobalRateCount : Int
414+*authGlobalRateWindowStart : Timestamp
415+
416+-- Max concurrent background prune forks (prevents resource exhaustion from stuck recovery)
417+*activePruneForks : Int
418+
405419-- ============================================
406420-- Routes
407421-- ============================================
@@ -421,6 +435,10 @@ route AuthApi where
421435 POST {targetToken: Text} /revoke headers {authorization: Text, clientIp: Text}
422436 -> {ok: Bool, authorized: Bool}
423437 = RevokeSession
438+ -- Voluntary disconnect: removes session and gossips offline if last session for this key
439+ POST {} /logout headers {authorization: Text, clientIp: Text}
440+ -> {ok: Bool, authorized: Bool}
441+ = Logout
424442
425443-- Client-facing API (requires session token in Authorization header)
426444route ClientApi where
@@ -433,6 +451,10 @@ route ClientApi where
433451 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text, clientIp: Text}
434452 -> {status: MessageStatus, messageId: OptionalMessageId}
435453 = SendMessage
454+ -- Check delivery status of a sent message (caller must know recipient to prove ownership)
455+ POST {to: PubkeyHex, messageId: MessageIdHex} /status headers {authorization: Text, clientIp: Text}
456+ -> {status: Text, authorized: Bool}
457+ = MessageStatus
436458 -- Look up which server a pubkey is present on (for federation routing)
437459 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
438460 -> {server: AnyServerName, authorized: Bool}
@@ -677,10 +699,12 @@ hasDotBeforePort = \s ->
677699-- IPv6 shorthand, octal/hex IP tricks, bare IPs without dots, and .arpa domains.
678700isBadServerName : Text -> Bool
679701isBadServerName = \sn ->
680− sn == ""
681− || take 1 sn == "."
682− || take 1 sn == ":"
702+ let fc = take 1 sn in
703+ fc == ""
704+ || fc == "."
705+ || fc == ":"
683706 || containsBadUrlChar sn
707+ || not (hasDotBeforePort sn)
684708 || matchesHostnameOrSubdomain sn "localhost"
685709 || matchesHostname sn "localhost.localdomain"
686710 || matchesHostnameOrSubdomain sn "local"
@@ -688,7 +712,7 @@ isBadServerName = \sn ->
688712 || matchesHostnameOrSubdomain sn "metadata"
689713 || take 4 sn == "127." || take 3 sn == "10."
690714 || take 8 sn == "192.168." || take 8 sn == "169.254."
691− || take 2 sn == "0." || take 1 sn == "["
715+ || take 2 sn == "0." || fc == "["
692716 || is172Private sn
693717 || take 7 sn == "198.18." || take 7 sn == "198.19."
694718 || take 8 sn == "192.0.0."
@@ -701,9 +725,8 @@ isBadServerName = \sn ->
701725 || containsMultipleColons sn
702726 || hasOctalOctet sn
703727 || hasHexPrefix sn
704− || (isDigitChar (take 1 sn) && isAllDigitsOrDotsUntilColonOrEnd sn)
728+ || (isDigitChar fc && isAllDigitsOrDotsUntilColonOrEnd sn)
705729 || isHighOctetPrefix sn
706− || not (hasDotBeforePort sn)
707730 || containsSuffix ".arpa" sn
708731 || containsSuffix ".onion" sn
709732 || containsSuffix ".i2p" sn
@@ -1302,6 +1325,8 @@ pruneExpired = \t -> do
13021325 challenges <- *challenges
13031326 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
13041327 transformAllSessionBuckets (\bucket -> do x <- bucket; where x.expiresAt > t; yield x)
1328+ sessionCount <- countAllSessions
1329+ set *totalSessionCount = sessionCount
13051330 -- Prune messages across all 16 buckets. Each bucket prune is atomic (short lock).
13061331 -- Recipient counts are rebuilt incrementally per-bucket — since messages are sharded
13071332 -- by recipient pubkey, each bucket's counts are disjoint, so simple union merges them.
@@ -1342,22 +1367,29 @@ pruneExpired = \t -> do
13421367 set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a
13431368 deliveredIds <- *deliveredIds
13441369 set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > retainCutoff; yield d
1345− -- Prune rate limiters and ephemeral state
1370+ -- Prune rate limiters (split into smaller atomic blocks to reduce lock contention)
13461371 atomic do
13471372 sendRates <- *sendRates
13481373 set *sendRates = do r <- sendRates; where t - r.windowStart < sendRateWindowMs; yield r
13491374 challengeRates <- *challengeRates
13501375 set *challengeRates = do r <- challengeRates; where t - r.windowStart < challengeRateWindowMs; yield r
1376+ recvRates <- *recvRates
1377+ set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r
1378+ -- Reset auth-tier global rate window
1379+ authWS <- *authGlobalRateWindowStart
1380+ when (t - authWS >= authGlobalRateWindowMs) (do
1381+ set *authGlobalRateWindowStart = t
1382+ set *authGlobalRateCount = 0)
1383+ atomic do
13511384 federationRates <- *federationRates
13521385 set *federationRates = do r <- federationRates; where t - r.windowStart < federationRateWindowMs; yield r
1353− activePolls <- *activePolls
1354− set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p
13551386 ipRates <- *ipRates
13561387 set *ipRates = do r <- ipRates; where t - r.windowStart < ipRateWindowMs; yield r
13571388 lookupRates <- *lookupRates
13581389 set *lookupRates = do r <- lookupRates; where t - r.windowStart < lookupRateWindowMs; yield r
1359− recvRates <- *recvRates
1360− set *recvRates = do r <- recvRates; where t - r.windowStart < recvRateWindowMs; yield r
1390+ atomic do
1391+ activePolls <- *activePolls
1392+ set *activePolls = do p <- activePolls; where t - p.startedAt < pollTimeoutMs * 3; yield p
13611393 recentGossip <- *recentGossip
13621394 set *recentGossip = do r <- recentGossip; where t - r.seenAt < gossipDedupWindow; yield r
13631395 -- Clean up stale in-flight entries: remove zeroed entries and reset leaked slots
@@ -1445,6 +1477,21 @@ checkGlobalRate = \t -> atomic do
14451477 set *globalRateCount = rateCount + 1
14461478 yield (False {})
14471479
1480+-- Auth-tier global rate limiting (shared budget for challenge + verify + revoke)
1481+checkAuthGlobalRate = \t -> atomic do
1482+ windowStart <- *authGlobalRateWindowStart
1483+ rateCount <- *authGlobalRateCount
1484+ if t - windowStart >= authGlobalRateWindowMs
1485+ then do
1486+ set *authGlobalRateWindowStart = t
1487+ set *authGlobalRateCount = 1
1488+ yield (False {})
1489+ else if rateCount >= maxAuthGlobalRate
1490+ then yield (True {})
1491+ else do
1492+ set *authGlobalRateCount = rateCount + 1
1493+ yield (False {})
1494+
14481495-- IP-based rate limiting; skips if clientIp header is empty (no reverse proxy)
14491496checkIpRate = \ip t ->
14501497 if ip == "" then yield (False {})
@@ -1587,12 +1634,14 @@ retryPendingForwards = \pending -> do
15871634-- When a remote key comes online, create forward entries for queued messages and send them
15881635forwardQueuedMessages = \targets t -> do
15891636 newForwards <- atomic do
1590− allMessages <- readAllMessages
15911637 forwards <- *forwards
15921638 presence <- *presence
15931639 let localOnline = localOnlineKeys presence t
15941640 let remoteTargets = remoteOnlyTargets targets localOnline
1595− let candidates = forwardCandidates remoteTargets allMessages forwards t
1641+ -- Read only the message buckets containing target pubkeys (not all 16)
1642+ let neededBuckets = uniqueBucketsFor remoteTargets (\target -> target.pubkey)
1643+ relevantMessages <- readSelectedBuckets neededBuckets
1644+ let candidates = forwardCandidates remoteTargets relevantMessages forwards t
15961645 let capped = takeRelation (maxForwardsEntries - count forwards) candidates
15971646 let updatedForwards = union forwards capped
15981647 set *forwards = updatedForwards
@@ -2068,6 +2117,9 @@ handleChallenge = \pubkey clientIp respond -> do
20682117 when (preAuth == "ip_rate_limited") (println ("challenge: IP rate limited for " ++ clientIp))
20692118 yield (respond {challenge: "", ok: False {}, expiresAt: 0})
20702119 else do
2120+ authLimited <- checkAuthGlobalRate t
2121+ if authLimited then yield (respond {challenge: "", ok: False {}, expiresAt: 0})
2122+ else do
20712123 rawChallenge <- genToken
20722124 let challenge = serverHostN ++ ":" ++ rawChallenge
20732125 result <- atomic do
@@ -2121,11 +2173,13 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
21212173 if authRateLimited then do
21222174 println ("verify: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey)
21232175 yield (respond {token: "", expiresAt: 0})
2124− else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
2176+ else do
2177+ authGlobalLimited <- checkAuthGlobalRate t
2178+ if authGlobalLimited then yield (respond {token: "", expiresAt: 0})
2179+ else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
21252180
21262181verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do
21272182 let expiry = t + sessionExpiry
2128− let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
21292183 validation <- atomic do
21302184 challenges <- *challenges
21312185 let hasMatch = challengeMatch challenges pubkey challenge clientIp t
@@ -2134,11 +2188,17 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
21342188 -- by submitting a bad signature from the same IP. Rate limiter still constrains
21352189 -- brute-force attempts; not consuming on IP mismatch prevents remote attackers
21362190 -- from burning challenges.
2137− if not hasMatch || not sigValid
2191+ -- verify() is inside the atomic block to prevent a TOCTOU gap between challenge
2192+ -- validation and consumption. It's pure (no IO) so safe in STM.
2193+ if not hasMatch
21382194 then yield {valid: False {}}
21392195 else do
2140− set *challenges = challengesExcluding challenges pubkey challenge
2141− yield {valid: True {}}
2196+ let sigValid = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
2197+ if not sigValid
2198+ then yield {valid: False {}}
2199+ else do
2200+ set *challenges = challengesExcluding challenges pubkey challenge
2201+ yield {valid: True {}}
21422202 if validation.valid
21432203 then do
21442204 token <- genToken
@@ -2159,6 +2219,8 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
21592219 let nb = tokenBucket token
21602220 currentBucket <- readSessionBucket nb
21612221 writeSessionBucket nb (union currentBucket [newSession])
2222+ sc <- *totalSessionCount
2223+ set *totalSessionCount = sc + 1
21622224 let presenceWithout = presenceWithoutLocalKey presence pubkey
21632225 let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
21642226 let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
@@ -2172,6 +2234,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
21722234 let afterRevoke = if revokeOthers then do s <- allSessions; where s.pubkey != pubkey; yield s else allSessions
21732235 let capped = capPerKeyAndGlobal afterRevoke (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
21742236 let sessionsWithNew = union capped [newSession]
2237+ set *totalSessionCount = count sessionsWithNew
21752238 let bFor = \b -> do s <- sessionsWithNew; where tokenBucket s.token == b; yield s
21762239 writeSessionBucket 0 (bFor 0)
21772240 writeSessionBucket 1 (bFor 1)
@@ -2380,6 +2443,39 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
23802443 yield {server: maskedServer, authorized: True {}})
23812444 yield (respond {server: result.server, authorized: result.authorized})
23822445
2446+-- POST /messages/status: Check delivery status of a previously sent message.
2447+-- Caller must provide both messageId and recipient pubkey (proves they know the recipient).
2448+-- Returns: "pending" (in mailbox), "forwarding" (federation in progress),
2449+-- "delivered" (acked by recipient), "unknown" (expired or never existed).
2450+handleMessageStatus = \to messageId authorization clientIp respond -> do
2451+ t <- now
2452+ globalRateLimited <- checkGlobalRate t
2453+ if globalRateLimited then yield (respond {status: "", authorized: False {}})
2454+ else
2455+ withSessionAuth authorization t clientIp
2456+ (yield (respond {status: "", authorized: False {}}))
2457+ (\senderKey -> do
2458+ result <- atomic do
2459+ let msgBucket = pubkeyBucket to
2460+ bucket <- readMessageBucket msgBucket
2461+ -- Check if message is still in mailbox (pending delivery)
2462+ let inMailbox = any bucket (\m -> m.messageId == messageId && m.toKey == to)
2463+ if inMailbox then yield "pending"
2464+ else do
2465+ -- Check if being forwarded to a remote server
2466+ forwards <- *forwards
2467+ let inForwards = any forwards (\f -> f.messageId == messageId && f.toKey == to)
2468+ if inForwards then yield "forwarding"
2469+ else do
2470+ -- Check if already delivered (acked or in terminal delivered set)
2471+ ackedIds <- *ackedIds
2472+ deliveredIds <- *deliveredIds
2473+ let inAcked = hasMessage messageId to ackedIds
2474+ let inDelivered = hasMessage messageId to deliveredIds
2475+ if inAcked || inDelivered then yield "delivered"
2476+ else yield "unknown"
2477+ yield (respond {status: result, authorized: True {}}))
2478+
23832479-- POST /auth/revoke: Revoke a specific session token.
23842480-- Caller authenticates via Authorization header, then specifies the target token to remove.
23852481-- Only removes the target if it belongs to the same pubkey as the caller (prevents cross-user revocation).
@@ -2408,11 +2504,59 @@ handleRevoke = \targetToken authorization clientIp respond -> do
24082504 if hasTarget
24092505 then do
24102506 writeSessionBucket targetBucket (do s <- sessions; where not (safeCompare s.token targetToken && s.pubkey == callerPubkey); yield s)
2507+ sc <- *totalSessionCount
2508+ set *totalSessionCount = max 0 (sc - 1)
24112509 yield (True {})
24122510 else yield (False {})
24132511 when removed (println ("revoke: session revoked for " ++ callerPubkey))
24142512 yield (respond {ok: removed, authorized: True {}}))
24152513
2514+-- POST /auth/logout: Voluntary disconnect. Removes the caller's session and, if it was
2515+-- the last active session for this pubkey, clears local presence and gossips offline.
2516+-- This gives faster offline detection than waiting for session expiry (1 hour).
2517+handleLogout = \authorization clientIp respond -> do
2518+ t <- now
2519+ globalRateLimited <- checkGlobalRate t
2520+ if globalRateLimited then yield (respond {ok: False {}, authorized: False {}})
2521+ else
2522+ case extractToken authorization of
2523+ Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
2524+ Just {value: token} -> do
2525+ let bucket = tokenBucket token
2526+ result <- atomic do
2527+ sessions <- readSessionBucket bucket
2528+ case authenticateSession sessions token t of
2529+ Unauthorized {} -> yield (Nothing {})
2530+ Authorized {pubkey, ip} ->
2531+ if clientIp != "" && ip != "" && clientIp != ip
2532+ then yield (Nothing {})
2533+ else do
2534+ -- Remove the caller's session
2535+ writeSessionBucket bucket (do s <- sessions; where not (safeCompare s.token token); yield s)
2536+ sc <- *totalSessionCount
2537+ set *totalSessionCount = max 0 (sc - 1)
2538+ -- Check if this was the last session for this key
2539+ allSessions <- readAllSessions
2540+ let remaining = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
2541+ if remaining == 0
2542+ then do
2543+ -- Remove local presence immediately
2544+ presence <- *presence
2545+ set *presence = presenceWithoutLocalKey presence pubkey
2546+ yield (Just {value: {pubkey: pubkey, gossipOffline: True {}}})
2547+ else yield (Just {value: {pubkey: pubkey, gossipOffline: False {}}})
2548+ case result of
2549+ Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
2550+ Just {value: r} -> do
2551+ when r.gossipOffline (do
2552+ println ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")
2553+ when federationEnabled (do
2554+ presence <- *presence
2555+ let peerServers = collectPeerServers presence (\p -> p.server)
2556+ when (count peerServers > 0)
2557+ (fork (sendBatchedGossip peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))))
2558+ yield (respond {ok: True {}, authorized: True {}})
2559+
24162560-- POST /federation/gossip: Receive presence gossip from a peer server.
24172561-- Validates events, merges into presence table (last-write-wins), and re-gossips
24182562-- to other known peers with decremented hop count. Triggers message forwarding
@@ -2532,33 +2676,12 @@ handleFederationAck = \messageIds originServer fromServer authorization respond
25322676 t <- now
25332677 withFederationAuth authorization t respond fromServer (isLocalServer originServer && not (isLocalServer fromServer)) (do
25342678 let dedupedIds = dedupMessageIds messageIds
2679+ -- Only clear forward entries; origin messages are retained until the recipient
2680+ -- polls from this server or they age out via unackedMessageMaxAge. This prevents
2681+ -- a compromised peer from deleting messages by sending unverified acks.
25352682 atomic do
25362683 forwards <- *forwards
2537− let ackedForwards = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
2538− let updatedForwards = forwardsExcluding forwards fromServer dedupedIds
2539− set *forwards = updatedForwards
2540− -- Read only message buckets containing acked recipients (not all 16)
2541− let neededBuckets = uniqueBucketsFor ackedForwards (\f -> f.toKey)
2542− relevantMessages <- readSelectedBuckets neededBuckets
2543− presence <- *presence
2544− let removable = do
2545− f <- ackedForwards
2546− where none updatedForwards (\rf -> rf.messageId == f.messageId && rf.toKey == f.toKey)
2547− where none presence (\p -> p.pubkey == f.toKey && isLocalServer p.server && p.expiresAt > t)
2548− yield f
2549− -- Only remove messages stored long enough for local clients to have polled them (grace period)
2550− let pastGrace = do f <- removable; where any relevantMessages (\m -> m.messageId == f.messageId && m.toKey == f.toKey && t - m.receivedAt >= ackGracePeriod); yield f
2551− let removed = count pastGrace
2552− -- Remove matched messages from only the affected buckets
2553− when (removed > 0) (do
2554− transformSelectedBuckets neededBuckets (\bucket ->
2555− do m <- bucket; where none pastGrace (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m)
2556− totalMsgCount <- *totalMessageCount
2557− set *totalMessageCount = totalMsgCount - removed
2558− -- Batch-adjust per-recipient counts (single pass instead of per-entry fold)
2559− recipientCounts <- *recipientMsgCounts
2560− let removedCounts = countByRecipient pastGrace
2561− set *recipientMsgCounts = batchDecrementRecipientCounts recipientCounts removedCounts)
2684+ set *forwards = forwardsExcluding forwards fromServer dedupedIds
25622685 yield (respond {ok: True {}, error: ""}))
25632686
25642687-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
@@ -2579,14 +2702,20 @@ handleDrain = \authorization respond -> do
25792702 let drainedCount = count drainResult.localEntries
25802703 forwards <- *forwards
25812704 ackedIds <- *ackedIds
2705+ let pendingAcks = do a <- ackedIds; where not (isLocalServer a.originServer); yield a
25822706 let pendingForwardCount = count forwards
2583− let pendingAckCount = count ackedIds
2584− -- Fork gossip so drain responds immediately (hung peers can't block the response).
2585− -- After gossip, wait for any in-flight federation requests (from prior forks) to
2586− -- complete before signaling drainCompletedAt. Operator polls /stats for drainCompletedAt > 0
2587− -- to know when it's truly safe to stop the process.
2707+ let pendingAckCount = count pendingAcks
2708+ -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
2709+ -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
25882710 atomic do set *drainCompletedAt = 0
25892711 fork (do
2712+ -- Best-effort flush of pending forwards and acks before stopping
2713+ when (federationEnabled && pendingForwardCount > 0) (do
2714+ println ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards")
2715+ retryPendingForwards forwards)
2716+ when (federationEnabled && pendingAckCount > 0) (do
2717+ println ("drain: flushing " ++ show pendingAckCount ++ " pending acks")
2718+ sendAcksAndClean pendingAcks)
25902719 when (federationEnabled && drainedCount > 0 && count drainResult.peerServers > 0) (do
25912720 let gossipEvents = do p <- drainResult.localEntries; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}
25922721 sendBatchedGossip drainResult.peerServers gossipEvents gossipOriginHops)
@@ -2596,9 +2725,9 @@ handleDrain = \authorization respond -> do
25962725 atomic do set *drainCompletedAt = t2)
25972726 println ("drain: broadcast offline for " ++ show drainedCount ++ " local keys")
25982727 when (pendingForwardCount > 0)
2599− (println ("drain: warning: " ++ show pendingForwardCount ++ " pending forwards will not be retried"))
2728+ (println ("drain: flushing " ++ show pendingForwardCount ++ " pending forwards"))
26002729 when (pendingAckCount > 0)
2601− (println ("drain: warning: " ++ show pendingAckCount ++ " pending acks will not be sent"))
2730+ (println ("drain: flushing " ++ show pendingAckCount ++ " pending acks"))
26022731 yield (respond {ok: True {}, drained: drainedCount})
26032732
26042733-- POST /maintenance/undrain: Resume accepting requests after a drain.
@@ -2634,7 +2763,7 @@ handleHealth = \respond -> do
26342763 sa <- *startedAt
26352764 presence <- *presence
26362765 forwards <- *forwards
2637− sessionCount <- countAllSessions
2766+ sessionCount <- *totalSessionCount
26382767 activePolls <- *activePolls
26392768 tmc <- *totalMessageCount
26402769 lastFedSuccess <- *lastFederationSuccessAt
@@ -2665,14 +2794,17 @@ handleHealth = \respond -> do
26652794zeroStats = {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, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, drainCompletedAt: 0, ok: False {}}
26662795
26672796handleStats = \authorization respond -> do
2797+ t <- now
2798+ globalRateLimited <- checkGlobalRate t
2799+ if globalRateLimited then yield (respond zeroStats)
2800+ else do
26682801 let authorized = verifyMaintenanceAuth authorization
26692802 if authorized
26702803 then do
2671− t <- now
26722804 -- Split into two atomic reads: core capacity state (consistent) and counters/timing
26732805 -- (slight inconsistency acceptable for monitoring). This narrows the lock window.
26742806 core <- atomic do
2675− sc <- countAllSessions
2807+ sc <- *totalSessionCount
26762808 p <- *presence
26772809 f <- *forwards
26782810 c <- *challenges
@@ -2716,20 +2848,26 @@ handleStats = \authorization respond -> do
27162848 else yield (respond zeroStats)
27172849
27182850handleStatsPeers = \authorization respond -> do
2851+ t <- now
2852+ globalRateLimited <- checkGlobalRate t
2853+ if globalRateLimited then yield (respond {peers: [], ok: False {}})
2854+ else do
27192855 let authorized = verifyMaintenanceAuth authorization
27202856 if authorized
27212857 then do
2722− t <- now
27232858 failures <- *forwardFailures
27242859 let peers = do f <- failures; yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoff f.failures, retriable: shouldRetryServer f.server t failures}
27252860 yield (respond {peers: peers, ok: True {}})
27262861 else yield (respond {peers: [], ok: False {}})
27272862
27282863handleStatsKey = \pubkey authorization respond -> do
2864+ t <- now
2865+ globalRateLimited <- checkGlobalRate t
2866+ if globalRateLimited then yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})
2867+ else do
27292868 let authorized = verifyMaintenanceAuth authorization
27302869 if authorized
27312870 then do
2732− t <- now
27332871 result <- atomic do
27342872 keyBucket <- readMessageBucket (pubkeyBucket pubkey)
27352873 presence <- *presence
@@ -2749,8 +2887,10 @@ serve = \req -> case req of
27492887 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
27502888 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
27512889 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
2890+ Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
27522891 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond
27532892 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond
2893+ MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
27542894 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
27552895 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer authorization respond
27562896 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer authorization respond
@@ -2801,13 +2941,24 @@ backgroundPrune = do
28012941 when status.recovered
28022942 (println ("backgroundPrune: previous cycle stuck for " ++ show (t - status.stuckSince) ++ "ms, forcing recovery"))
28032943 if status.canRun
2804− then fork (do
2805− pruneAndRetry t
2806− -- Only clear running flag if this is still the latest generation
2807− -- (prevents a stuck old fork from clearing the flag after a recovery cycle started)
2808− atomic do
2809− currentGen <- *pruneGeneration
2810− when (currentGen == status.gen) (set *pruneRunning = 0))
2944+ then do
2945+ -- Guard against unbounded concurrent prune forks from repeated stuck recovery
2946+ forks <- *activePruneForks
2947+ if forks >= 2
2948+ then println "backgroundPrune: skipping (max concurrent prune forks reached)"
2949+ else do
2950+ atomic do
2951+ af <- *activePruneForks
2952+ set *activePruneForks = af + 1
2953+ fork (do
2954+ pruneAndRetry t
2955+ -- Only clear running flag if this is still the latest generation
2956+ -- (prevents a stuck old fork from clearing the flag after a recovery cycle started)
2957+ atomic do
2958+ currentGen <- *pruneGeneration
2959+ when (currentGen == status.gen) (set *pruneRunning = 0)
2960+ af <- *activePruneForks
2961+ set *activePruneForks = max 0 (af - 1))
28112962 else println "backgroundPrune: skipping (previous cycle still running)"
28122963 backgroundPrune
28132964
@@ -2851,6 +3002,16 @@ main = do
28513002 println (" federation: " ++ if federationEnabled then "enabled" else "disabled")
28523003 println (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
28533004 println (" poll timeout: " ++ show pollTimeoutMs ++ "ms")
3005+ println (" max messages: " ++ show maxMessages ++ " (per recipient: " ++ show maxMessagesPerRecipient ++ ")")
3006+ println (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")
3007+ println (" max presence: " ++ show maxPresenceEntries)
3008+ println (" max forwards: " ++ show maxForwardsEntries)
3009+ println (" session expiry: " ++ show sessionExpiry ++ "ms")
3010+ println (" message max age: " ++ show messageMaxAge ++ "ms")
3011+ println (" global rate limit: " ++ show maxGlobalRequestRate ++ "/" ++ show globalRateWindowMs ++ "ms")
3012+ println (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
3013+ when (count federationPeerSecrets > 0)
3014+ (println (" per-peer secrets: " ++ show (count federationPeerSecrets) ++ " configured"))
28543015 fork backgroundPrune
28553016 fork pollHeartbeat
28563017 listen 8080 serve