@@ -18,6 +18,10 @@ serverHost = "localhost:8080"
1818federationSecret = "" -- must be set to enable federation
1919maintenanceSecret = "" -- set to enable /maintenance/prune independently of federation
2020federationScheme = "https://" -- scheme for federation HTTP calls
21+-- NOTE: Outbound federation HTTP calls have no timeout. A slow/unresponsive peer can hold
22+-- a concurrency slot (maxFederationConcurrency) indefinitely. Add a fetch timeout if the
23+-- Knot runtime supports it; until then, exponential backoff on failed fetches mitigates
24+-- but does not fully prevent resource exhaustion from hung connections.
2125-- Per-peer federation secrets (overrides global federationSecret for specific peers)
2226-- Format: [{server: "peer.example.com", secret: "peer-specific-secret"}]
2327federationPeerSecrets : [{server: Text, secret: Text}]
@@ -125,6 +129,9 @@ maxLookupRateEntries = 1000
125129-- Unacked message aging (messages pruned after this regardless of ack status)
126130unackedMessageMaxAge : Int<Ms>
127131unackedMessageMaxAge = 604800000 -- 7 days
132+-- Grace period before federation ack can delete the original message (allows local client to poll)
133+ackGracePeriod : Int<Ms>
134+ackGracePeriod = 30000 -- 30 seconds
128135-- Health check thresholds (background task freshness)
129136healthPruneMaxAge : Int<Ms>
130137healthPruneMaxAge = backgroundPruneInterval * 3
@@ -163,7 +170,7 @@ type ValidGossipEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Times
163170-- Status returned to the sender after SendMessage
164171-- Delivered: recipient is online locally; Federated: forwarded to remote server;
165172-- Queued: stored but recipient location unknown; Unauthorized/Rejected: not accepted
166−data MessageStatus = MsDelivered {} | MsFederated {} | MsQueued {} | MsUnauthorized {} | MsRejected {} | MsRateLimited {} | MsDuplicate {}
173+data MessageStatus = MsDelivered {} | MsFederated {} | MsQueued {} | MsUnauthorized {} | MsRejected {} | MsRateLimited {} | MsDuplicate {} | MsDraining {}
167174
168175impl ToJSON MessageStatus where
169176 toJson ms = toJson (case ms of
@@ -173,7 +180,8 @@ impl ToJSON MessageStatus where
173180 MsUnauthorized {} -> "unauthorized"
174181 MsRejected {} -> "rejected"
175182 MsRateLimited {} -> "rate_limited"
176− MsDuplicate {} -> "duplicate")
183+ MsDuplicate {} -> "duplicate"
184+ MsDraining {} -> "draining")
177185
178186-- Internal delivery status (same as MessageStatus but without error variants)
179187data DeliveryStatus = DsDelivered {} | DsFederated {} | DsQueued {}
@@ -244,6 +252,8 @@ type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
244252
245253*messages : [Message]
246254
255+*totalMessageCount : Int
256+
247257*presence : [PresenceEntry]
248258
249259-- Monotonic counter ensuring unique receivedAt ordering even within the same millisecond
@@ -314,6 +324,8 @@ type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
314324
315325*pruneRunning : Timestamp
316326
327+*pruneGeneration : Int
328+
317329*ipRates : [IpRateEntry]
318330
319331*lookupRates : [LookupRateEntry]
@@ -345,7 +357,7 @@ route AuthApi where
345357route ClientApi where
346358 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
347359 POST {cursor: Timestamp} /poll headers {authorization: Text}
348− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool}
360+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool}
349361 = Poll
350362 /messages
351363 -- Send an E2E encrypted message to a recipient pubkey
@@ -894,12 +906,16 @@ pruneExpired = \t -> do
894906 atomic do
895907 let messageCutoff = t - unackedMessageMaxAge
896908 messages <- *messages
897− set *messages = do m <- messages; where m.receivedAt > messageCutoff; yield m
909+ let prunedMessages = do m <- messages; where m.receivedAt > messageCutoff; yield m
910+ set *messages = prunedMessages
911+ set *totalMessageCount = count prunedMessages
898912 -- Prune federation state
899913 atomic do
900914 let retainCutoff = t - messageMaxAge
901915 forwards <- *forwards
902− set *forwards = do f <- forwards; where f.createdAt > retainCutoff; yield f
916+ -- Clear all forwards when federation is disabled (they can never be retried)
917+ set *forwards = if not federationEnabled then []
918+ else do f <- forwards; where f.createdAt > retainCutoff; yield f
903919 ackedIds <- *ackedIds
904920 set *ackedIds = do a <- ackedIds; where a.ackedAt > retainCutoff; yield a
905921 forwardFailures <- *forwardFailures
@@ -1269,10 +1285,10 @@ pruneAndRetry = \t -> do
12691285 retryPendingForwards result.pendingForwards
12701286 sendAcksAndClean result.pendingAcks
12711287 retryOverflowAcks result.pendingOverflow))
1288+ -- Only gossip extended sessions; keys nearing expiry without active sessions will naturally
1289+ -- trigger offline gossip when they expire (avoids wasted federation traffic)
12721290 when result.shouldRefresh (do
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
1291+ let refreshEvents = do p <- result.extendedActive; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000)}
12761292 when (count result.refreshPeers > 0 && count refreshEvents > 0)
12771293 (fork (sendBatchedGossip result.refreshPeers refreshEvents gossipOriginHops)))
12781294
@@ -1338,6 +1354,8 @@ ackPolledMessages = \pubkey cursor t -> do
13381354 if count toAck > 0
13391355 then do
13401356 set *messages = diff messages toAck
1357+ totalCount <- *totalMessageCount
1358+ set *totalMessageCount = totalCount - count toAck
13411359 let ackedMsgIds = do m <- toAck; yield m.messageId
13421360 forwards <- *forwards
13431361 set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
@@ -1635,6 +1653,10 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
16351653 else yield {valid: True {}}
16361654 if validation.valid
16371655 then do
1656+ -- Reject if server started draining after challenge was issued
1657+ isDraining <- *draining
1658+ if isDraining then yield (respond {token: "", expiresAt: 0})
1659+ else do
16381660 token <- genToken
16391661 peerServers <- atomic do
16401662 sessions <- *sessions
@@ -1660,10 +1682,10 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
16601682handlePoll = \cursor authorization respond -> do
16611683 t <- now
16621684 draining <- *draining
1663− if draining then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: False {}})
1685+ if draining then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}})
16641686 else do
16651687 globalRateLimited <- checkGlobalRate t
1666− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}})
1688+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}})
16671689 else do
16681690 result <- case extractToken authorization of
16691691 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
@@ -1671,7 +1693,7 @@ handlePoll = \cursor authorization respond -> do
16711693 fork (when federationEnabled (do
16721694 sendAcksAndClean result.pendingAcks
16731695 retryEvictedAcks result.evictedEntries result.pendingAcks))
1674− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}})
1696+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}})
16751697
16761698-- POST /messages: Send an encrypted message to a recipient.
16771699-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
@@ -1679,7 +1701,7 @@ handlePoll = \cursor authorization respond -> do
16791701handleSendMessage = \to encryptedBlob timestamp messageId authorization respond -> do
16801702 t <- now
16811703 draining <- *draining
1682− if draining then yield (respond {status: MsRejected {}, messageId: ""})
1704+ if draining then yield (respond {status: MsDraining {}, messageId: ""})
16831705 else do
16841706 globalRateLimited <- checkGlobalRate t
16851707 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: ""})
@@ -1690,6 +1712,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
16901712 withSessionAuth authorization t
16911713 (yield {send: SendFailed {status: MsUnauthorized {}}, senderRateLimited: False {}, globalCapReached: False {}})
16921714 (\senderKey -> do
1715+ -- Reject self-sends (sender == recipient wastes a mailbox slot)
1716+ if senderKey == to then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}}
1717+ else do
16931718 messages <- *messages
16941719 ackedIds <- *ackedIds
16951720 deliveredIds <- *deliveredIds
@@ -1714,10 +1739,10 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
17141739 then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}}
17151740 else do
17161741 set *recvRates = recvResult.entries
1717− let msgStats = messageCounts messages to t
1718− if msgStats.total >= maxMessages
1742+ totalMsgCount <- *totalMessageCount
1743+ if totalMsgCount >= maxMessages
17191744 then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: True {}}
1720− else if msgStats.forKey >= maxMessagesPerRecipient
1745+ else if countMessagesFor messages to t >= maxMessagesPerRecipient
17211746 then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}}
17221747 else do
17231748 presence <- *presence
@@ -1727,6 +1752,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
17271752 set *messages = union messages
17281753 [{messageId: effectiveMsgId, toKey: to,
17291754 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
1755+ set *totalMessageCount = totalMsgCount + 1
17301756 signalMessageForKey to seqTs
17311757 let toPresence = presenceForKey presence to t
17321758 let localOnline = hasLocalPresence toPresence
@@ -1855,10 +1881,10 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
18551881 if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds
18561882 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}}
18571883 else do
1858− let msgStats = messageCounts messages toKey t
1859− if msgStats.total >= maxMessages
1884+ totalMsgCount <- *totalMessageCount
1885+ if totalMsgCount >= maxMessages
18601886 then yield {forward: ForwardRejected {}, globalCapReached: True {}}
1861− else if msgStats.forKey >= maxMessagesPerRecipient
1887+ else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
18621888 then yield {forward: ForwardRejected {}, globalCapReached: False {}}
18631889 else do
18641890 recvRates <- *recvRates
@@ -1877,6 +1903,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
18771903 [{messageId: messageId, toKey: toKey,
18781904 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
18791905 set *messages = updatedMessages
1906+ set *totalMessageCount = totalMsgCount + 1
18801907 signalMessageForKey toKey seqTs
18811908 yield {forward: ForwardStored {}, globalCapReached: False {}}
18821909 case result.forward of
@@ -1908,7 +1935,11 @@ handleFederationAck = \messageIds originServer fromServer authorization respond
19081935 where none updatedForwards (\rf -> rf.messageId == f.messageId && rf.toKey == f.toKey)
19091936 where none presence (\p -> p.pubkey == f.toKey && isLocalServer p.server && p.expiresAt > t)
19101937 yield f
1911− set *messages = do m <- messages; where none removable (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m
1938+ -- Only remove messages stored long enough for local clients to have polled them (grace period)
1939+ let pastGrace = do f <- removable; where any messages (\m -> m.messageId == f.messageId && m.toKey == f.toKey && t - m.receivedAt >= ackGracePeriod); yield f
1940+ let updatedMessages = do m <- messages; where none pastGrace (\f -> m.messageId == f.messageId && m.toKey == f.toKey); yield m
1941+ set *messages = updatedMessages
1942+ set *totalMessageCount = count updatedMessages
19121943 yield (respond {ok: True {}, error: ""}))
19131944
19141945-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
@@ -1953,40 +1984,38 @@ handlePrune = \authorization respond -> do
19531984
19541985handleHealth = \respond -> do
19551986 t <- now
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})
1987+ -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
1988+ result <- atomic do
1989+ lastPrune <- *lastPrunedAt
1990+ lastHb <- *lastHeartbeatAt
1991+ sa <- *startedAt
1992+ presence <- *presence
1993+ forwards <- *forwards
1994+ sessions <- *sessions
1995+ activePolls <- *activePolls
1996+ messages <- *messages
1997+ lastFedSuccess <- *lastFederationSuccessAt
1998+ forwardFailures <- *forwardFailures
1999+ let warmup = t - sa < healthPruneMaxAge
2000+ -- Capacity pressure: flag when any capped collection exceeds 90% full
2001+ let capacityOk = count messages * 10 <= maxMessages * 9
2002+ && count presence * 10 <= maxPresenceEntries * 9
2003+ && count forwards * 10 <= maxForwardsEntries * 9
2004+ && count sessions * 10 <= maxSessions * 9
2005+ && count activePolls * 10 <= maxActivePolls * 9
2006+ -- Federation: healthy if disabled, in warmup, no peer issues, or recent success
2007+ let fedOk = not federationEnabled || warmup
2008+ || count forwardFailures == 0
2009+ || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
2010+ yield {pruneOk: warmup || t - lastPrune < healthPruneMaxAge,
2011+ heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
2012+ capacityOk: capacityOk,
2013+ federationOk: fedOk}
2014+ yield (respond {ok: result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk,
2015+ pruneHealthy: result.pruneOk,
2016+ heartbeatHealthy: result.heartbeatOk,
2017+ capacityHealthy: result.capacityOk,
2018+ federationHealthy: result.federationOk})
19902019
19912020handleStats = \authorization respond -> do
19922021 let authorized = verifyMaintenanceAuth authorization
@@ -2092,18 +2121,25 @@ backgroundPrune = do
20922121 status <- atomic do
20932122 running <- *pruneRunning
20942123 if running > 0 && t - running < backgroundPruneInterval * 3
2095− then yield {canRun: False {}, recovered: False {}}
2124+ then yield {canRun: False {}, recovered: False {}, gen: 0}
20962125 else do
20972126 set *pruneRunning = t
2127+ gen <- *pruneGeneration
2128+ let newGen = gen + 1
2129+ set *pruneGeneration = newGen
20982130 pc <- *pruneCount
20992131 set *pruneCount = pc + 1
2100− yield {canRun: True {}, recovered: running > 0}
2132+ yield {canRun: True {}, recovered: running > 0, gen: newGen}
21012133 when status.recovered
21022134 (println "backgroundPrune: previous cycle appears stuck, forcing new cycle")
21032135 if status.canRun
21042136 then fork (do
21052137 pruneAndRetry t
2106− atomic do set *pruneRunning = 0)
2138+ -- Only clear running flag if this is still the latest generation
2139+ -- (prevents a stuck old fork from clearing the flag after a recovery cycle started)
2140+ atomic do
2141+ currentGen <- *pruneGeneration
2142+ when (currentGen == status.gen) (set *pruneRunning = 0))
21072143 else println "backgroundPrune: skipping (previous cycle still running)"
21082144 backgroundPrune
21092145