@@ -92,6 +92,8 @@ authGlobalRateWindowMs = 1000
92-- Session and gossip timing
92-- Session and gossip timing
93sessionExpiry : Int<Ms>
93sessionExpiry : Int<Ms>
94sessionExpiry = 3600000 -- 1 hour
94sessionExpiry = 3600000 -- 1 hour
95+maxSessionLifetime : Int<Ms>
96+maxSessionLifetime = 86400000 -- 24 hours; absolute session lifetime (forces re-auth even if sliding window keeps refreshing)
95onlineGossipTtl : Int<S>
97onlineGossipTtl : Int<S>
96onlineGossipTtl = 3600 -- 1 hour; TTL announced when a key authenticates
98onlineGossipTtl = 3600 -- 1 hour; TTL announced when a key authenticates
97offlineGossipMinTtl : Int<S>
99offlineGossipMinTtl : Int<S>
@@ -195,6 +197,9 @@ unackedMessageMaxAge = 604800000 -- 7 days
195-- Grace period before federation ack can delete the original message (allows local client to poll)
197-- Grace period before federation ack can delete the original message (allows local client to poll)
196ackGracePeriod : Int<Ms>
198ackGracePeriod : Int<Ms>
197ackGracePeriod = 30000 -- 30 seconds
199ackGracePeriod = 30000 -- 30 seconds
200+-- Deferred ack cleanup capacity
201+maxPendingAckCleanups = 10000
198-- Health check thresholds (background task freshness)
203-- Health check thresholds (background task freshness)
199healthPruneMaxAge : Int<Ms>
204healthPruneMaxAge : Int<Ms>
200healthPruneMaxAge = backgroundPruneInterval * 3
205healthPruneMaxAge = backgroundPruneInterval * 3
@@ -271,7 +276,7 @@ type Timestamp = Int<Ms>
271-- Pending challenge for Ed25519 challenge-response auth
276-- Pending challenge for Ed25519 challenge-response auth
272type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp, ip: Text}
277type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp, ip: Text}
273-- Authenticated session; token is a random hex string
278-- Authenticated session; token is a random hex string
274−type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp, ip: Text}
279+type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp, ip: Text, createdAt: Timestamp}
275-- Stored message; originServer is "" for locally-originated messages, set to sender server for federated forwards
280-- Stored message; originServer is "" for locally-originated messages, set to sender server for federated forwards
276type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp}
281type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp}
277-- Presence: tracks that a pubkey is online at a given server, with expiry
282-- Presence: tracks that a pubkey is online at a given server, with expiry
@@ -525,6 +530,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
525-- Global federation connection counter (tracks total concurrent outbound connections)
530-- Global federation connection counter (tracks total concurrent outbound connections)
526*federationActiveConnections : Int
531*federationActiveConnections : Int
533+-- Deferred ack cleanups: origin message reclamation queued by federation ack handlers,
534+-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
535+*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
528-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
537-- Monotonic counter bumped on any presence change (gossip, auth, logout, prune, restore, drain)
529-- Polls compare against this to skip expensive presence lookups when nothing changed.
538-- Polls compare against this to skip expensive presence lookups when nothing changed.
530*presenceVersion : Timestamp
539*presenceVersion : Timestamp
@@ -884,7 +893,7 @@ isBadServerName = \sn ->
884-- Caller passes the appropriate session bucket (sharded by token prefix).
893-- Caller passes the appropriate session bucket (sharded by token prefix).
885authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
894authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
886authenticateSession = \sessions token t ->
895authenticateSession = \sessions token t ->
887− case headOf (do s <- sessions; where s.expiresAt > t && safeCompare s.token token; yield s) of
896+ case headOf (do s <- sessions; where s.expiresAt > t && t - s.createdAt < maxSessionLifetime && safeCompare s.token token; yield s) of
888 Nothing {} -> Unauthorized {}
897 Nothing {} -> Unauthorized {}
889 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}
898 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}
@@ -1360,6 +1369,23 @@ pruneMessageBucket = \b messageCutoff -> atomic do
1360 set *recipientMsgCounts = batchDecrementRecipientCounts rc (countByRecipient removed)
1369 set *recipientMsgCounts = batchDecrementRecipientCounts rc (countByRecipient removed)
1361 else yield {}
1370 else yield {}
1372+-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16
1373+-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one
1374+-- bucket, so per-bucket recipient counts are disjoint and can be merged without overlap.
1375+reconcileMessageCounts = \b acc ->
1378+ let total = fold (\a c -> a + c.count) 0 acc
1379+ stored <- *totalMessageCount
1380+ set *totalMessageCount = total
1381+ set *recipientMsgCounts = acc
1382+ yield (total - stored)
1384+ bucketCounts <- atomic do
1385+ bucket <- readMessageBucket b
1386+ yield (countByRecipient bucket)
1387+ reconcileMessageCounts (b + 1) (union acc bucketCounts)
1363-- ============================================
1389-- ============================================
1364-- Recipient Count Helpers
1390-- Recipient Count Helpers
1365-- ============================================
1391-- ============================================
@@ -1717,7 +1743,7 @@ pruneExpired = \t -> do
1717 atomic do
1743 atomic do
1718 challenges <- *challenges
1744 challenges <- *challenges
1719 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
1745 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
1720− transformAllSessionBuckets (\bucket -> do x <- bucket; where x.expiresAt > t; yield x)
1746+ transformAllSessionBuckets (\bucket -> do x <- bucket; where x.expiresAt > t && t - x.createdAt < maxSessionLifetime; yield x)
1721 sessionCount <- countAllSessions
1747 sessionCount <- countAllSessions
1722 set *totalSessionCount = sessionCount
1748 set *totalSessionCount = sessionCount
1723 -- Prune messages across all 16 buckets. Each bucket prune is atomic and adjusts
1749 -- Prune messages across all 16 buckets. Each bucket prune is atomic and adjusts
@@ -1808,6 +1834,9 @@ pruneExpired = \t -> do
1808 -- Prune gossip event rate entries
1834 -- Prune gossip event rate entries
1809 gossipEventRates <- *gossipEventRates
1835 gossipEventRates <- *gossipEventRates
1810 set *gossipEventRates = do r <- gossipEventRates; where t - r.windowStart < gossipEventRateWindowMs; yield r
1836 set *gossipEventRates = do r <- gossipEventRates; where t - r.windowStart < gossipEventRateWindowMs; yield r
1837+ -- Prune stale deferred ack cleanups (older than messageMaxAge are never useful)
1838+ pendingCleanups <- *pendingAckCleanups
1839+ set *pendingAckCleanups = do c <- pendingCleanups; where t - c.cleanupAfter < messageMaxAge; yield c
1811 -- Reconcile global federation connection counter with actual in-flight state
1840 -- Reconcile global federation connection counter with actual in-flight state
1812 -- (self-heals drift from leaked slots where a forked task died without decrementing)
1841 -- (self-heals drift from leaked slots where a forked task died without decrementing)
1813 inFlightRec <- *federationInFlight
1842 inFlightRec <- *federationInFlight
@@ -2136,8 +2165,10 @@ retryPendingForwards = \pending -> do
2136forwardQueuedMessages = \targets t -> do
2165forwardQueuedMessages = \targets t -> do
2137 newForwards <- atomic do
2166 newForwards <- atomic do
2138 forwards <- *forwards
2167 forwards <- *forwards
2139− allPresence <- readAllPresence
2168+ -- Read only the presence buckets needed for target pubkeys (not all 16)
2169+ let targetPresenceBuckets = uniqueBucketsFor targets (\target -> target.pubkey)
2170+ relevantPresence <- readSelectedPresenceBuckets targetPresenceBuckets
2140− let localOnline = localOnlineKeys allPresence t
2171+ let localOnline = localOnlineKeys relevantPresence t
2141 let remoteTargets = remoteOnlyTargets targets localOnline
2172 let remoteTargets = remoteOnlyTargets targets localOnline
2142 -- Read only the message buckets containing target pubkeys (not all 16)
2173 -- Read only the message buckets containing target pubkeys (not all 16)
2143 let neededBuckets = uniqueBucketsFor remoteTargets (\target -> target.pubkey)
2174 let neededBuckets = uniqueBucketsFor remoteTargets (\target -> target.pubkey)
@@ -2343,16 +2374,10 @@ pruneAndRetry = \t -> do
2343 yield (actual - stored)
2374 yield (actual - stored)
2344 when (presDrift != 0 || sessDrift != 0)
2375 when (presDrift != 0 || sessDrift != 0)
2345 (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))
2376 (logInfo ("reconciliation: corrected drift — presence: " ++ show presDrift ++ ", sessions: " ++ show sessDrift))
2346− -- Message reconciliation is expensive (reads all 16 buckets + rebuilds recipient counts), keep periodic
2377+ -- Message reconciliation: per-bucket reads reduce lock contention vs reading all 16 at once
2347 pc <- *pruneCount
2378 pc <- *pruneCount
2348 when (pc > 0 && pc - (pc / reconciliationFrequency) * reconciliationFrequency == 0) (do
2379 when (pc > 0 && pc - (pc / reconciliationFrequency) * reconciliationFrequency == 0) (do
2349− msgDrift <- atomic do
2380+ msgDrift <- reconcileMessageCounts 0 []
2350− allMsgs <- readAllMessages
2351− let actual = count allMsgs
2352− stored <- *totalMessageCount
2353− set *totalMessageCount = actual
2354− set *recipientMsgCounts = countByRecipient allMsgs
2355− yield (actual - stored)
2356 when (msgDrift != 0)
2381 when (msgDrift != 0)
2357 (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift)))
2382 (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift)))
2358 pendingForwards <- atomic do f <- *forwards; yield f
2383 pendingForwards <- atomic do f <- *forwards; yield f
@@ -2390,6 +2415,15 @@ pruneAndRetry = \t -> do
2390 set *federationLatencyCount = 0
2415 set *federationLatencyCount = 0
2391 set *federationLatencyMax = 0
2416 set *federationLatencyMax = 0
2392 set *federationLatencyResetAt = t)
2417 set *federationLatencyResetAt = t)
2418+ -- Process deferred ack cleanups: reclaim origin messages whose grace period has elapsed.
2419+ -- These were queued by handleFederationAck instead of fork+sleep.
2420+ readyCleanups <- atomic do
2421+ cleanups <- *pendingAckCleanups
2422+ let ready = do c <- cleanups; where t >= c.cleanupAfter; yield c
2423+ let remaining = do c <- cleanups; where t < c.cleanupAfter; yield c
2424+ set *pendingAckCleanups = remaining
2426+ when (count readyCleanups > 0) (ackCleanupForwards readyCleanups t)
2393 let result = {pendingForwards: pendingForwards, pendingAcks: ackState.pendingAcks, pendingOverflow: ackState.pendingOverflow, extendedActive: refreshResult.extendedActive, nearExpiryActive: refreshResult.nearExpiryActive, refreshPeers: refreshResult.refreshPeers, shouldRefresh: refreshResult.shouldRefresh}
2427 let result = {pendingForwards: pendingForwards, pendingAcks: ackState.pendingAcks, pendingOverflow: ackState.pendingOverflow, extendedActive: refreshResult.extendedActive, nearExpiryActive: refreshResult.nearExpiryActive, refreshPeers: refreshResult.refreshPeers, shouldRefresh: refreshResult.shouldRefresh}
2394 -- Fork federation I/O so hung peers don't block next prune cycle
2428 -- Fork federation I/O so hung peers don't block next prune cycle
2395 fork (do
2429 fork (do
@@ -2515,7 +2549,7 @@ pollOnce = \token cursor clientIp -> do
2515 then yield (Nothing {})
2549 then yield (Nothing {})
2517 -- Refresh session expiry (sliding window keeps active sessions alive)
2551 -- Refresh session expiry (sliding window keeps active sessions alive)
2518− writeSessionBucket bucket (do s <- sessions; yield (if safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip} else s))
2552+ writeSessionBucket bucket (do s <- sessions; yield (if safeCompare s.token token then {token: s.token, pubkey: s.pubkey, expiresAt: t + sessionExpiry, ip: s.ip, createdAt: s.createdAt} else s))
2519 seqCounter <- *seqCounter
2553 seqCounter <- *seqCounter
2520 let clampedCursor = clamp 0 (max seqCounter t) cursor
2554 let clampedCursor = clamp 0 (max seqCounter t) cursor
2521 ackResult <- if clampedCursor > 0
2555 ackResult <- if clampedCursor > 0
@@ -2712,8 +2746,7 @@ handleChallenge = \pubkey clientIp respond -> do
2712 authLimited <- checkAuthGlobalRate t
2746 authLimited <- checkAuthGlobalRate t
2713 if authLimited then yield (respond {challenge: "", ok: False {}, expiresAt: 0})
2747 if authLimited then yield (respond {challenge: "", ok: False {}, expiresAt: 0})
2715− rawChallenge <- genToken
2749+ challenge <- genToken
2716− let challenge = serverHostN ++ ":" ++ rawChallenge
2717 result <- atomic do
2750 result <- atomic do
2718 challenges <- *challenges
2751 challenges <- *challenges
2719 let active = do c <- challenges; where c.expiresAt > t; yield c
2752 let active = do c <- challenges; where c.expiresAt > t; yield c
@@ -2804,7 +2837,7 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
2804 -- Count per-key sessions without materializing all sessions (O(sessions) but no union)
2837 -- Count per-key sessions without materializing all sessions (O(sessions) but no union)
2805 sessionsForKey <- countSessionsForKey pubkey
2838 sessionsForKey <- countSessionsForKey pubkey
2806 tsc <- *totalSessionCount
2839 tsc <- *totalSessionCount
2807− let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp}
2840+ let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
2808 -- Update presence: replace local entry for this pubkey, evict oldest if over capacity
2841 -- Update presence: replace local entry for this pubkey, evict oldest if over capacity
2809 let pb = pubkeyBucket pubkey
2842 let pb = pubkeyBucket pubkey
2810 presenceBkt <- readPresenceBucket pb
2843 presenceBkt <- readPresenceBucket pb
@@ -3101,7 +3134,13 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
3101 let inAcked = hasMessage messageId to ackedIds
3134 let inAcked = hasMessage messageId to ackedIds
3102 let inDelivered = hasMessage messageId to deliveredIds
3135 let inDelivered = hasMessage messageId to deliveredIds
3103 if inAcked || inDelivered then yield "delivered"
3136 if inAcked || inDelivered then yield "delivered"
3138+ -- Extended check: forward nonces have 60-day retention (2x message lifetime)
3139+ -- and 100k capacity, providing delivery evidence after ackedIds/deliveredIds are pruned
3140+ forwardNonces <- *forwardNonces
3141+ let inNonces = any forwardNonces (\n -> n.messageId == messageId && n.toKey == to)
3142+ if inNonces then yield "delivered"
3104− else yield "unknown"
3143+ else yield "unknown"
3105 yield (respond {status: result, authorized: True {}}))
3144 yield (respond {status: result, authorized: True {}}))
3107-- POST /auth/revoke: Revoke a specific session token.
3146-- POST /auth/revoke: Revoke a specific session token.
@@ -3115,6 +3154,9 @@ handleRevoke = \targetToken authorization clientIp respond -> do
3115 withSessionAuth authorization t clientIp
3154 withSessionAuth authorization t clientIp
3116 (yield (respond {ok: False {}, authorized: False {}}))
3155 (yield (respond {ok: False {}, authorized: False {}}))
3117 (\callerPubkey -> do
3156 (\callerPubkey -> do
3157+ authLimited <- checkAuthGlobalRate t
3158+ if authLimited then yield (respond {ok: False {}, authorized: True {}})
3118 revokeLimited <- atomic do
3160 revokeLimited <- atomic do
3119 challengeRates <- *challengeRates
3161 challengeRates <- *challengeRates
3120 let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == callerPubkey)
3162 let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == callerPubkey)
@@ -3397,7 +3439,9 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3397 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3439 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3398 fwdNonces <- *forwardNonces
3440 fwdNonces <- *forwardNonces
3399 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
3441 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
3442+ -- Batch-evict 10% when at capacity to create a buffer; makes it harder
3443+ -- for an attacker to force-evict specific nonces via targeted flooding.
3400− set *forwardNonces = if count updatedNonces > maxForwardNonces then evictOneOldest updatedNonces (\n -> n.seenAt) else updatedNonces
3444+ set *forwardNonces = if count updatedNonces > maxForwardNonces then (evictSmallest updatedNonces (maxForwardNonces / 10) (\n -> n.seenAt)).kept else updatedNonces
3401 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
3445 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
3402 case result.forward of
3446 case result.forward of
3403 ForwardRejected {} ->
3447 ForwardRejected {} ->
@@ -3425,15 +3469,14 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp autho
3425 let removed = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
3469 let removed = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
3426 set *forwards = forwardsExcluding forwards fromServer dedupedIds
3470 set *forwards = forwardsExcluding forwards fromServer dedupedIds
3427 yield removed
3471 yield removed
3428− -- Best-effort origin message cleanup (forked to avoid slowing the ack response).
3472+ -- Deferred origin message cleanup: queue for processing in the next prune cycle
3429− -- Reclaims messages past the ack grace period when the recipient has no local
3473+ -- after ackGracePeriod elapses. Replaces fork+sleep which accumulated sleeping
3430− -- presence and no other pending forwards remain. Without this, confirmed-delivered
3474+ -- goroutines under high federation ack volume.
3431− -- messages sit for up to unackedMessageMaxAge (7 days).
3475+ when (count removedForwards > 0) (atomic do
3432− when (count removedForwards > 0) (fork (do
3476+ cleanups <- *pendingAckCleanups
3433− sleep ackGracePeriod
3477+ let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
3434− t2 <- now
3478+ let combined = union cleanups newCleanups
3435− -- Batched by message bucket: entries for the same bucket share one atomic block.
3479+ set *pendingAckCleanups = if count combined > maxPendingAckCleanups then takeRelation maxPendingAckCleanups combined else combined)
3436− ackCleanupForwards removedForwards t2))
3437 yield (respond {ok: True {}, error: ""}))
3480 yield (respond {ok: True {}, error: ""}))
3439-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
3482-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
@@ -3833,6 +3876,11 @@ sendBatchInner = \senderKey messages t acc ->
3833 Nothing {} -> yield acc
3876 Nothing {} -> yield acc
3834 Just {value: msg} -> do
3877 Just {value: msg} -> do
3835 effectiveMsgId <- if msg.messageId != "" then yield msg.messageId else genToken
3878 effectiveMsgId <- if msg.messageId != "" then yield msg.messageId else genToken
3879+ -- Log timestamp anomalies (matching handleSendMessage warnings)
3880+ when (msg.timestamp > t + clockSkewTolerance)
3881+ (logWarn ("sendBatch: future timestamp detected (+" ++ show (msg.timestamp - t) ++ "ms ahead) for " ++ effectiveMsgId))
3882+ when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
3883+ (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
3836 result <- if t - msg.timestamp > messageMaxAge
3884 result <- if t - msg.timestamp > messageMaxAge
3837 then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0}
3885 then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0}
3838 else if senderKey == msg.to
3886 else if senderKey == msg.to
@@ -3866,36 +3914,50 @@ sendBatchInner = \senderKey messages t acc ->
3866 if recvResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}}
3914 if recvResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}}
3868 set *recvRates = recvResult.entries
3916 set *recvRates = recvResult.entries
3869− totalMsgCount <- *totalMessageCount
3917+ -- Per-sender-per-recipient pair rate limiting (prevents one sender filling a recipient's mailbox)
3918+ pairRates <- *pairRates
3919+ let pairResult = checkRateWindow pairRates (\r -> r.sender == senderKey && r.recipient == msg.to)
3920+ (\ts -> {sender: senderKey, recipient: msg.to, windowStart: ts, count: 1})
3921+ (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
3922+ maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
3870− if totalMsgCount >= maxMessages then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}}
3923+ if pairResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}}
3872− recipientCounts <- *recipientMsgCounts
3925+ set *pairRates = pairResult.entries
3873− let recipientCount = getRecipientCount msg.to recipientCounts
3926+ totalMsgCount <- *totalMessageCount
3874− if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}}
3927+ if totalMsgCount >= maxMessages then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}}
3929+ recipientCounts <- *recipientMsgCounts
3930+ let recipientCount = getRecipientCount msg.to recipientCounts
3931+ if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}}
3876− recipientPresence <- readPresenceBucket (pubkeyBucket msg.to)
3933+ recipientPresence <- readPresenceBucket (pubkeyBucket msg.to)
3877− seqCounter <- *seqCounter
3934+ seqCounter <- *seqCounter
3878− let seqTs = nextSeqTs t seqCounter
3935+ let seqTs = nextSeqTs t seqCounter
3879− set *seqCounter = seqTs
3936+ set *seqCounter = seqTs
3880− writeMessageBucket msgBucket (union bucket
3937+ writeMessageBucket msgBucket (union bucket
3881− [{messageId: effectiveMsgId, toKey: msg.to,
3938+ [{messageId: effectiveMsgId, toKey: msg.to,
3882− encryptedBlob: msg.encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])
3939+ encryptedBlob: msg.encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])
3883− set *totalMessageCount = totalMsgCount + 1
3940+ set *totalMessageCount = totalMsgCount + 1
3884− set *recipientMsgCounts = adjustRecipientCount recipientCounts msg.to 1
3941+ set *recipientMsgCounts = adjustRecipientCount recipientCounts msg.to 1
3885− signalMessageForKey msg.to seqTs
3942+ signalMessageForKey msg.to seqTs
3886− let toPresence = presenceForKey recipientPresence msg.to t
3943+ let toPresence = presenceForKey recipientPresence msg.to t
3887− let localOnline = hasLocalPresence toPresence
3944+ let localOnline = hasLocalPresence toPresence
3888− let remoteServers = collectPeerServers toPresence (\p -> p.server)
3945+ let remoteServers = collectPeerServers toPresence (\p -> p.server)
3889− let fwdServer = case headOf remoteServers of Nothing {} -> ""; Just {value: s} -> s
3946+ let fwdServer = case headOf remoteServers of Nothing {} -> ""; Just {value: s} -> s
3890− when (federationEnabled && count remoteServers > 0) (do
3947+ when (federationEnabled && count remoteServers > 0) (do
3891− forwards <- *forwards
3948+ forwards <- *forwards
3892− let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: msg.to, encryptedBlob: msg.encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}
3949+ let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: msg.to, encryptedBlob: msg.encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}
3893− let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
3950+ let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
3894− set *forwards = union forwards cappedFwds)
3951+ set *forwards = union forwards cappedFwds)
3895− let delivery = if localOnline then MsDelivered {}
3952+ let delivery = if localOnline then MsDelivered {}
3896− else if federationEnabled && count remoteServers > 0 then MsFederated {}
3953+ else if federationEnabled && count remoteServers > 0 then MsFederated {}
3897− else MsQueued {}
3954+ else MsQueued {}
3898− yield {status: delivery, fwdServer: fwdServer, needsFwd: federationEnabled && count remoteServers > 0}
3955+ yield {status: delivery, fwdServer: fwdServer, needsFwd: federationEnabled && count remoteServers > 0}
3956+ -- Log rate limiting and capacity warnings (matching handleSendMessage)
3957+ case sendResult.status of
3958+ MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
3959+ MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")
3899 yield {messageId: effectiveMsgId, status: sendResult.status, needsForward: sendResult.needsFwd, forwardServer: sendResult.fwdServer, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs}
3961 yield {messageId: effectiveMsgId, status: sendResult.status, needsForward: sendResult.needsFwd, forwardServer: sendResult.fwdServer, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs}
3900 sendBatchInner senderKey split.rest t (union acc [result])
3962 sendBatchInner senderKey split.rest t (union acc [result])
@@ -4028,8 +4090,9 @@ backgroundSnapshot = do
4028 ao <- *ackOverflow
4090 ao <- *ackOverflow
4029 d <- *deliveredIds
4091 d <- *deliveredIds
4030 ff <- *forwardFailures
4092 ff <- *forwardFailures
4093+ pac <- *pendingAckCleanups
4031− yield {messages: tmc, sessions: sc, presence: pc, forwards: count f, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff}
4094+ yield {messages: tmc, sessions: sc, presence: pc, forwards: count f, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff, pendingAckCleanups: count pac}
4032− logInfo ("snapshot: messages=" ++ show result.messages ++ " sessions=" ++ show result.sessions ++ " presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " ackOverflow=" ++ show result.ackOverflow ++ " deliveredIds=" ++ show result.deliveredIds ++ " forwardFailures=" ++ show result.forwardFailures)
4095+ logInfo ("snapshot: messages=" ++ show result.messages ++ " sessions=" ++ show result.sessions ++ " presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " ackOverflow=" ++ show result.ackOverflow ++ " deliveredIds=" ++ show result.deliveredIds ++ " forwardFailures=" ++ show result.forwardFailures ++ " pendingAckCleanups=" ++ show result.pendingAckCleanups)
4033 -- Warn if federation state hasn't been exported recently (will be lost on restart)
4096 -- Warn if federation state hasn't been exported recently (will be lost on restart)
4034 t <- now
4097 t <- now
4035 lastExport <- *lastSnapshotExportAt
4098 lastExport <- *lastSnapshotExportAt
@@ -4042,10 +4105,34 @@ backgroundSnapshot = do
4042 else yield {}
4105 else yield {}
4043 backgroundSnapshot
4106 backgroundSnapshot
4108+-- Background watchdog: monitors prune and heartbeat loops, restarts if they appear dead.
4109+-- Knot lacks try/catch, so panicked loops can't self-recover. This watchdog detects
4110+-- liveness by checking counter freshness and forks a new loop instance as needed.
4111+backgroundWatchdog = do
4112+ sleep (backgroundPruneInterval * 5)
4115+ let warmup = t - sa < backgroundPruneInterval * 10
4116+ if warmup then backgroundWatchdog
4118+ -- Check prune loop health
4119+ lastPrune <- *lastPrunedAt
4120+ when (t - lastPrune > healthPruneMaxAge * 3) (do
4121+ logError "watchdog: prune loop appears dead, restarting"
4122+ atomic do set *pruneRunning = 0
4123+ fork backgroundPrune)
4124+ -- Check heartbeat loop health
4125+ lastHb <- *lastHeartbeatAt
4126+ when (t - lastHb > healthHeartbeatMaxAge * 3) (do
4127+ logError "watchdog: heartbeat loop appears dead, restarting"
4128+ fork pollHeartbeat)
4045main = do
4131main = do
4046 let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
4132 let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
4047 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
4133 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
4048 && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0
4134 && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0
4135+ && maxSessionLifetime > 0
4049 && maxPresenceTtl > 0 && maxForwardsEntries > 0 && maxAckedIds > 0
4136 && maxPresenceTtl > 0 && maxForwardsEntries > 0 && maxAckedIds > 0
4050 && maxForwardRetries > 0 && maxGossipHops > 0 && maxPollsPerKey > 0
4137 && maxForwardRetries > 0 && maxGossipHops > 0 && maxPollsPerKey > 0
4051 && gossipOriginHops > 0 && maxThirdPartyGossipHops > 0
4138 && gossipOriginHops > 0 && maxThirdPartyGossipHops > 0
@@ -4059,6 +4146,7 @@ main = do
4059 -- (e.g., messages pruned before their stated max age, backoff that never increases)
4146 -- (e.g., messages pruned before their stated max age, backoff that never increases)
4060 let configRelationsOk = unackedMessageMaxAge <= messageMaxAge
4147 let configRelationsOk = unackedMessageMaxAge <= messageMaxAge
4061 && challengeExpiry < sessionExpiry
4148 && challengeExpiry < sessionExpiry
4149+ && sessionExpiry <= maxSessionLifetime
4062 && retryBackoffBase <= retryBackoffMax
4150 && retryBackoffBase <= retryBackoffMax
4063 && backgroundPruneInterval < healthPruneMaxAge
4151 && backgroundPruneInterval < healthPruneMaxAge
4064 && ackGracePeriod < unackedMessageMaxAge
4152 && ackGracePeriod < unackedMessageMaxAge
@@ -4080,7 +4168,7 @@ main = do
4080 if not configOk
4168 if not configOk
4081 then logError "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
4169 then logError "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
4082 else if not configRelationsOk
4170 else if not configRelationsOk
4083− then logError "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, per-key limits <= global limits), server not started"
4171+ then logError "FATAL: invalid configuration — relationship invariants violated (check: unackedMessageMaxAge <= messageMaxAge, challengeExpiry < sessionExpiry, sessionExpiry <= maxSessionLifetime, retryBackoffBase <= retryBackoffMax, backgroundPruneInterval < healthPruneMaxAge, pollTimeoutMs < sessionExpiry, gossipDedupWindow < presenceRefreshInterval, federationFetchTimeoutMs <= backgroundPruneInterval, maxAckOverflow <= maxAckedIds, gossipOriginHops <= maxGossipHops, maxThirdPartyGossipHops <= maxGossipHops, verificationTokenExpiry <= federationFetchTimeoutMs, forwardNonceRetention >= messageMaxAge, maxGlobalFederationConcurrency >= maxFederationConcurrency, maxGossipEventsPerServer > maxGossipEvents, per-key limits <= global limits), server not started"
4085 t <- now
4173 t <- now
4086 atomic do
4174 atomic do
@@ -4099,7 +4187,7 @@ main = do
4099 logInfo (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")
4187 logInfo (" max sessions: " ++ show maxSessions ++ " (per key: " ++ show maxSessionsPerKey ++ ")")
4100 logInfo (" max presence: " ++ show maxPresenceEntries)
4188 logInfo (" max presence: " ++ show maxPresenceEntries)
4101 logInfo (" max forwards: " ++ show maxForwardsEntries)
4189 logInfo (" max forwards: " ++ show maxForwardsEntries)
4102− logInfo (" session expiry: " ++ show sessionExpiry ++ "ms")
4190+ logInfo (" session expiry: " ++ show sessionExpiry ++ "ms (max lifetime: " ++ show maxSessionLifetime ++ "ms)")
4103 logInfo (" message max age: " ++ show messageMaxAge ++ "ms")
4191 logInfo (" message max age: " ++ show messageMaxAge ++ "ms")
4104 logInfo (" global rate limit: " ++ show maxGlobalRequestRate ++ "/" ++ show globalRateWindowMs ++ "ms")
4192 logInfo (" global rate limit: " ++ show maxGlobalRequestRate ++ "/" ++ show globalRateWindowMs ++ "ms")
4105 logInfo (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
4193 logInfo (" auth rate limit: " ++ show maxAuthGlobalRate ++ "/" ++ show authGlobalRateWindowMs ++ "ms")
@@ -4114,5 +4202,6 @@ main = do
4114 (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
4202 (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
4115 fork backgroundPrune
4203 fork backgroundPrune
4116 fork pollHeartbeat
4204 fork pollHeartbeat
4205+ fork backgroundWatchdog
4117 when (autoSnapshotInterval > 0) (fork backgroundSnapshot)
4206 when (autoSnapshotInterval > 0) (fork backgroundSnapshot)
4118 listen 8080 serve
4207 listen 8080 serve