@@ -15,6 +15,9 @@
16-- Server identity and federation credentials
16-- Server identity and federation credentials
17serverHost = "localhost:8080"
17serverHost = "localhost:8080"
18+-- TODO: Support runtime secret rotation: accept both old and new secrets during a
19+-- transition window, so operators can rotate without restarting. E.g., add a
20+-- maintenanceSecretPrevious field that's accepted for a configurable overlap period.
18maintenanceSecret = "" -- set to enable /maintenance/* and /stats endpoints
21maintenanceSecret = "" -- set to enable /maintenance/* and /stats endpoints
19federationScheme = "https://" -- scheme for federation HTTP calls
22federationScheme = "https://" -- scheme for federation HTTP calls
20-- NOTE: Outbound federation HTTP calls have no runtime-level timeout. The fetchFederation
23-- NOTE: Outbound federation HTTP calls have no runtime-level timeout. The fetchFederation
@@ -23,12 +26,20 @@ federationScheme = "https://" -- scheme for federation HTTP calls
23-- slow peer from consuming all concurrency slots, though truly hung connections still hold
26-- slow peer from consuming all concurrency slots, though truly hung connections still hold
24-- one slot until the forked task terminates.
27-- one slot until the forked task terminates.
25-- Message and data size limits
28-- Message and data size limits
29+-- TODO: If Knot adds HTTP-layer configuration, enforce a request body size limit
30+-- (e.g., 256KB) to prevent large-payload DoS. Currently field-level length checks
31+-- (maxBlobLen, maxPubkeyLen, etc.) validate after parsing, so a multi-GB JSON body
32+-- would be fully parsed before rejection. A framework-level limit would reject early.
26messageMaxAge : Int<Ms>
33messageMaxAge : Int<Ms>
27messageMaxAge = 2592000000 -- 30 days
34messageMaxAge = 2592000000 -- 30 days
28maxPresenceTtl : Int<S>
35maxPresenceTtl : Int<S>
29maxPresenceTtl = 3600 -- max presence TTL from gossip
36maxPresenceTtl = 3600 -- max presence TTL from gossip
30maxPubkeyLen = 256
37maxPubkeyLen = 256
31maxBlobLen = 65536
38maxBlobLen = 65536
39+-- Minimum message TTL (sender-specified). Prevents very short TTLs that would expire
40+-- before the recipient polls, wasting mailbox slots and federation bandwidth.
41+minMessageTtl : Int<Ms>
42+minMessageTtl = 60000 -- 1 minute
32maxSignatureLen = 256
43maxSignatureLen = 256
33maxIpLen = 45 -- max length of an IP address string (IPv6 = 45 chars); used in matchIp instead of maxServerNameLen
44maxIpLen = 45 -- max length of an IP address string (IPv6 = 45 chars); used in matchIp instead of maxServerNameLen
34maxMessageIds = 100
45maxMessageIds = 100
@@ -148,7 +159,7 @@ federationLatencyWindowMs : Int<Ms>
148federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
159federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
150-- Snapshot format version (increment on schema changes to prevent corrupt restores)
161-- Snapshot format version (increment on schema changes to prevent corrupt restores)
151−snapshotVersion = 3
162+snapshotVersion = 4 -- bumped for messageTtl field in ForwardEntry
153-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
164-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
154maxGossipEventsPerServer = 500 -- max total gossip events per server per window
165maxGossipEventsPerServer = 500 -- max total gossip events per server per window
@@ -314,12 +325,16 @@ type Timestamp = Int<Ms>
314type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp, ip: Text}
325type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp, ip: Text}
315-- Authenticated session; token is a random hex string
326-- Authenticated session; token is a random hex string
316type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp, ip: Text, createdAt: Timestamp}
327type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp, ip: Text, createdAt: Timestamp}
317−-- Stored message; originServer is "" for locally-originated messages, set to sender server for federated forwards
328+-- Stored message; originServer is "" for locally-originated messages, set to sender server for federated forwards.
329+-- expiresAt: computed from sender-specified TTL or default unackedMessageMaxAge.
330+-- Messages with expiresAt <= now are pruned and excluded from polls.
318−type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp}
331+type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp, expiresAt: Timestamp}
319-- Presence: tracks that a pubkey is online at a given server, with expiry
332-- Presence: tracks that a pubkey is online at a given server, with expiry
320type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
333type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
321−-- Queued federation forward: message waiting to be delivered to a remote server
334+-- Queued federation forward: message waiting to be delivered to a remote server.
335+-- messageTtl: remaining TTL in ms at time of enqueue (0 = default). Passed to the
336+-- receiving server so it can compute its own expiresAt.
322−type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int}
337+type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int, messageTtl: Int}
323-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
338-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
324type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}
339type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}
325-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
340-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
@@ -357,7 +372,7 @@ type UniqueKeyIpEntry = {ip: Text, keys: [PubkeyHex], windowStart: Timestamp}
357type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}
372type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}
359-- Batch send types
374-- Batch send types
360−type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId}
375+type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, messageTtl: Int}
361type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
376type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
363-- ============================================
378-- ============================================
@@ -707,6 +722,11 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
707-- backgroundSnapshot compares current counters against this to skip redundant exports.
722-- backgroundSnapshot compares current counters against this to skip redundant exports.
708*lastSnapshotFingerprint : Int
723*lastSnapshotFingerprint : Int
725+-- Prune cycles since last snapshot export (wall-clock agnostic staleness tracking).
726+-- Incremented each prune cycle; reset to 0 on snapshot export (manual, auto, or drain).
727+-- More robust than wall-clock comparison: survives clock jumps without false alarms.
728+*prunesSinceSnapshot : Int
710-- Cached health result: returned when /health is rate-limited instead of false all-healthy.
730-- Cached health result: returned when /health is rate-limited instead of false all-healthy.
711-- Updated on every successful (non-rate-limited) health check.
731-- Updated on every successful (non-rate-limited) health check.
712*cachedHealthOk : Bool
732*cachedHealthOk : Bool
@@ -728,10 +748,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
728route AuthApi where
748route AuthApi where
730 POST {pubkey: PubkeyHex} /challenge headers {clientIp: Text}
750 POST {pubkey: PubkeyHex} /challenge headers {clientIp: Text}
731− -> {challenge: Text, ok: Bool, expiresAt: Timestamp, retryAfterMs: Int}
751+ -> {challenge: Text, ok: Bool, expiresAt: Timestamp, retryAfterMs: Int, error: Text}
732 = ReqChallenge
752 = ReqChallenge
733 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /verify headers {clientIp: Text}
753 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /verify headers {clientIp: Text}
734− -> {token: Text, expiresAt: Timestamp}
754+ -> {token: Text, expiresAt: Timestamp, error: Text}
735 = VerifyAuth
755 = VerifyAuth
736 POST {targetToken: Text} /revoke headers {authorization: Text, clientIp: Text}
756 POST {targetToken: Text} /revoke headers {authorization: Text, clientIp: Text}
737 -> {ok: Bool, authorized: Bool}
757 -> {ok: Bool, authorized: Bool}
@@ -748,8 +768,11 @@ route ClientApi where
748 -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool}
768 -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool}
750 /messages
770 /messages
751− -- Send an E2E encrypted message to a recipient pubkey
771+ -- Send an E2E encrypted message to a recipient pubkey.
772+ -- messageTtl: optional TTL in milliseconds (0 = default unackedMessageMaxAge).
773+ -- Clamped to [minMessageTtl, unackedMessageMaxAge]. Short TTLs for ephemeral data
774+ -- (typing indicators, read receipts) free mailbox slots faster than the default 7 days.
752− POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text, clientIp: Text}
775+ POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, messageTtl: Int} / headers {authorization: Text, clientIp: Text}
753 -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int}
776 -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int}
754 = SendMessage
777 = SendMessage
755 -- Check delivery status of a sent message (caller must know recipient to prove ownership)
778 -- Check delivery status of a sent message (caller must know recipient to prove ownership)
@@ -772,6 +795,13 @@ route ClientApi where
772 POST {messageIds: [MessageIdHex]} /delete headers {authorization: Text, clientIp: Text}
795 POST {messageIds: [MessageIdHex]} /delete headers {authorization: Text, clientIp: Text}
773 -> {ok: Bool, authorized: Bool, deletedCount: Int}
796 -> {ok: Bool, authorized: Bool, deletedCount: Int}
774 = DeleteMessages
797 = DeleteMessages
798+ -- Cancel (retract) a sent message before the recipient reads it.
799+ -- Caller provides messageId and recipient pubkey — knowing the messageId proves
800+ -- sender identity (either client-generated or returned only to the sender).
801+ -- Removes from recipient mailbox, cancels pending forwards, sends federation acks.
802+ POST {messageId: MessageIdHex, to: PubkeyHex} /cancel headers {authorization: Text, clientIp: Text}
803+ -> {ok: Bool, authorized: Bool, cancelled: Bool}
775 -- Look up which server a pubkey is present on (for federation routing)
805 -- Look up which server a pubkey is present on (for federation routing)
776 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
806 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
777 -> {server: AnyServerName, authorized: Bool}
807 -> {server: AnyServerName, authorized: Bool}
@@ -786,8 +816,10 @@ route FederationApi where
786 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /gossip headers {authorization: Text}
816 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /gossip headers {authorization: Text}
787 -> {ok: Bool, error: Text}
817 -> {ok: Bool, error: Text}
788 = RecvGossip
818 = RecvGossip
789− -- Receive a forwarded message from a peer server on behalf of a sender
819+ -- Receive a forwarded message from a peer server on behalf of a sender.
820+ -- messageTtl: remaining TTL in ms (0 = default). Allows short-lived messages to
821+ -- expire on the destination server without waiting for the full unackedMessageMaxAge.
790− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /forward headers {authorization: Text}
822+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, messageTtl: Int} /forward headers {authorization: Text}
791 -> {ok: Bool, error: Text}
823 -> {ok: Bool, error: Text}
792 = ForwardMessage
824 = ForwardMessage
793 -- Acknowledge delivery of messages back to the origin server
825 -- Acknowledge delivery of messages back to the origin server
@@ -884,8 +916,19 @@ route StatsApi where
884 GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text}
916 GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text}
885 -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, forwardDetails: [{server: Text, retries: Int, createdAt: Timestamp}], activePolls: Int, ok: Bool}
917 -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, forwardDetails: [{server: Text, retries: Int, createdAt: Timestamp}], activePolls: Int, ok: Bool}
886 = StatsKey
918 = StatsKey
919+ -- Top recipients by pending message count (for abuse investigation — find inbox-bombing targets)
920+ GET /stats/top-recipients headers {authorization: Text}
921+ -> {recipients: [RecipientCount], ok: Bool}
924+-- Readiness probe: returns false during startup warmup (before first prune completes),
925+-- true once the server is fully operational. Unlike /health, this is unauthenticated
926+-- and doesn't check subsystem liveness — it only indicates whether the server has
927+-- completed initialization. Use for Kubernetes readinessProbe or load balancer warmup.
929+ GET /ready -> {ready: Bool} = Ready
888−route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | MaintenanceApi | HealthApi | StatsApi
931+route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | MaintenanceApi | HealthApi | StatsApi | ReadyApi
890-- ============================================
933-- ============================================
891-- Helpers
934-- Helpers
@@ -1107,6 +1150,7 @@ isBadServerName = \sn ->
1107 || matchesHostnameOrSubdomain sn "localhost"
1150 || matchesHostnameOrSubdomain sn "localhost"
1108 || matchesHostname sn "localhost.localdomain"
1151 || matchesHostname sn "localhost.localdomain"
1109 || matchesHostnameOrSubdomain sn "local"
1152 || matchesHostnameOrSubdomain sn "local"
1153+ || containsSuffix ".localhost" sn -- RFC 6761: foo.localhost resolves to loopback
1110 || matchesHostnameOrSubdomain sn "internal"
1154 || matchesHostnameOrSubdomain sn "internal"
1111 || matchesHostnameOrSubdomain sn "metadata"
1155 || matchesHostnameOrSubdomain sn "metadata"
1112 || take 4 sn == "127." || take 3 sn == "10."
1156 || take 4 sn == "127." || take 3 sn == "10."
@@ -1692,24 +1736,26 @@ transformSelectedBuckets = \buckets f ->
1692 transformSelectedBuckets rest f
1736 transformSelectedBuckets rest f
1694-- Collect unique bucket indices for a set of pubkeys
1738-- Collect unique bucket indices for a set of pubkeys
1739+-- O(1) per step with union [b] acc (set, order irrelevant, max 16 entries)
1695uniqueBucketsFor = \items getKey ->
1740uniqueBucketsFor = \items getKey ->
1696 fold (\acc x ->
1741 fold (\acc x ->
1697 let b = pubkeyBucket (getKey x) in
1742 let b = pubkeyBucket (getKey x) in
1698− if any acc (\i -> i == b) then acc else union acc [b]) [] items
1743+ if any acc (\i -> i == b) then acc else union [b] acc) [] items
1700-- Prune a single message bucket and atomically adjust shared counters.
1745-- Prune a single message bucket and atomically adjust shared counters.
1701-- Each bucket prune decrements totalMessageCount and its rcBucket by exactly
1746-- Each bucket prune decrements totalMessageCount and its rcBucket by exactly
1702-- what was removed, preserving concurrent modifications from poll-ack or send.
1747-- what was removed, preserving concurrent modifications from poll-ack or send.
1748+-- Uses per-message expiresAt (respects sender-specified TTL).
1703−pruneMessageBucket = \b messageCutoff -> atomic do
1749+pruneMessageBucket = \b t -> atomic do
1704 bucket <- readMessageBucket b
1750 bucket <- readMessageBucket b
1705− let kept = do m <- bucket; where m.receivedAt > messageCutoff; yield m
1751+ let kept = do m <- bucket; where m.expiresAt > t; yield m
1706 let removedCount = count bucket - count kept
1752 let removedCount = count bucket - count kept
1707 writeMessageBucket b kept
1753 writeMessageBucket b kept
1708 if removedCount > 0 then do
1754 if removedCount > 0 then do
1709 tc <- *totalMessageCount
1755 tc <- *totalMessageCount
1710 set *totalMessageCount = max 0 (tc - removedCount)
1756 set *totalMessageCount = max 0 (tc - removedCount)
1711 rc <- readRcBucket b
1757 rc <- readRcBucket b
1712− let removed = do m <- bucket; where m.receivedAt <= messageCutoff; yield m
1758+ let removed = do m <- bucket; where m.expiresAt <= t; yield m
1713 writeRcBucket b (batchDecrementRecipientCounts rc (countByRecipient removed))
1759 writeRcBucket b (batchDecrementRecipientCounts rc (countByRecipient removed))
1714 else yield {}
1760 else yield {}
@@ -1720,15 +1766,15 @@ pruneMessageBucket = \b messageCutoff -> atomic do
1720-- is fast (list filter + counter decrement), (2) prune runs every ~30s, not on the
1766-- is fast (list filter + counter decrement), (2) prune runs every ~30s, not on the
1721-- hot path, (3) Knot's fork is fire-and-forget so we can't collect per-bucket
1767-- hot path, (3) Knot's fork is fire-and-forget so we can't collect per-bucket
1722-- deltas and apply them in a single final atomic block.
1768-- deltas and apply them in a single final atomic block.
1723−pruneMessageBucketsConcurrent = \b messageCutoff ->
1769+pruneMessageBucketsConcurrent = \b t ->
1724 if b > 15 then yield {}
1770 if b > 15 then yield {}
1726 bucket <- readMessageBucket b
1772 bucket <- readMessageBucket b
1727− if count bucket == 0 then pruneMessageBucketsConcurrent (b + 1) messageCutoff
1773+ if count bucket == 0 then pruneMessageBucketsConcurrent (b + 1) t
1728 else if b < 15 then do
1774 else if b < 15 then do
1729− fork (pruneMessageBucket b messageCutoff)
1775+ fork (pruneMessageBucket b t)
1730− pruneMessageBucketsConcurrent (b + 1) messageCutoff
1776+ pruneMessageBucketsConcurrent (b + 1) t
1731− else pruneMessageBucket b messageCutoff
1777+ else pruneMessageBucket b t
1733-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16
1779-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16
1734-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one
1780-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one
@@ -2294,6 +2340,13 @@ fetchFederation = \server request -> do
2294clampTtl : Int<S> -> Int<S>
2340clampTtl : Int<S> -> Int<S>
2295clampTtl = clamp 1 maxPresenceTtl
2341clampTtl = clamp 1 maxPresenceTtl
2343+-- Compute message expiresAt from a sender-specified TTL (in ms).
2344+-- 0 = default (unackedMessageMaxAge). Non-zero values are clamped to [minMessageTtl, unackedMessageMaxAge].
2345+computeMessageExpiresAt : Timestamp -> Int -> Timestamp
2346+computeMessageExpiresAt = \receivedAt messageTtl ->
2347+ if messageTtl <= 0 then receivedAt + unackedMessageMaxAge
2348+ else receivedAt + clamp minMessageTtl unackedMessageMaxAge messageTtl
2297clampTimestamp : Timestamp -> Timestamp -> Timestamp
2350clampTimestamp : Timestamp -> Timestamp -> Timestamp
2298clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
2351clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
@@ -2502,13 +2555,18 @@ removeOneForKey = \items pubkey ->
2503-- Extract unique remote (non-local) server names from a list of items.
2556-- Extract unique remote (non-local) server names from a list of items.
2504-- O(n*k) where k = unique servers; acceptable since k is typically small (<20).
2557-- O(n*k) where k = unique servers; acceptable since k is typically small (<20).
2558+-- NOTE: `union acc [s]` copies acc (O(|acc|) per step, O(n*k) total). Using
2559+-- `union [s] acc` would be O(1) per step but reverses insertion order. Since this
2560+-- is a set (order doesn't matter), prepend is preferred for large collections.
2561+-- TODO: Switch to `union [s] acc` across set-accumulating folds (collectPeerServers,
2562+-- uniqueBucketsFor, dedupMessageIds) if Knot confirms union copies left operand.
2505collectPeerServers : [a] -> (a -> Text) -> [Text]
2563collectPeerServers : [a] -> (a -> Text) -> [Text]
2506collectPeerServers = \items getServer ->
2564collectPeerServers = \items getServer ->
2507 fold (\acc item ->
2565 fold (\acc item ->
2508 let s = getServer item in
2566 let s = getServer item in
2509 if isLocalServer s then acc
2567 if isLocalServer s then acc
2510 else if any acc (\a -> a == s) then acc
2568 else if any acc (\a -> a == s) then acc
2511− else union acc [s]) [] items
2569+ else union [s] acc) [] items
2513samePubkeyServer = \a b -> a.pubkey == b.pubkey && a.server == b.server
2571samePubkeyServer = \a b -> a.pubkey == b.pubkey && a.server == b.server
2514pubkeyServer = \e -> {pubkey: e.pubkey, server: e.server}
2572pubkeyServer = \e -> {pubkey: e.pubkey, server: e.server}
@@ -2647,8 +2705,9 @@ pruneExpired = \t -> do
2647 -- and touches only its own message bucket + rcBucket. The only shared write is
2705 -- and touches only its own message bucket + rcBucket. The only shared write is
2648 -- *totalMessageCount (a single Int), so STM contention is minimal.
2706 -- *totalMessageCount (a single Int), so STM contention is minimal.
2649 -- Fork first 15 buckets; run the last one in the foreground to bound concurrency.
2707 -- Fork first 15 buckets; run the last one in the foreground to bound concurrency.
2650− let messageCutoff = t - unackedMessageMaxAge
2708+ -- Uses per-message expiresAt instead of a global cutoff, so sender-specified TTLs
2709+ -- are respected during pruning.
2651− pruneMessageBucketsConcurrent 0 messageCutoff
2710+ pruneMessageBucketsConcurrent 0 t
2652 -- Prune federation state (split into per-collection blocks to reduce lock contention
2711 -- Prune federation state (split into per-collection blocks to reduce lock contention
2653 -- with poll-ack paths that write to ackedIds/deliveredIds concurrently).
2712 -- with poll-ack paths that write to ackedIds/deliveredIds concurrently).
2654 let retainCutoff = t - messageMaxAge
2713 let retainCutoff = t - messageMaxAge
@@ -3123,7 +3182,7 @@ retryForwardsForServer = \entries remaining ->
3123 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
3182 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
3124 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
3183 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
3125 fromServer: serverHostN, requestTimestamp: t,
3184 fromServer: serverHostN, requestTimestamp: t,
3126− authorization: auth, correlationId: corrId})
3185+ authorization: auth, correlationId: corrId, messageTtl: f.messageTtl})
3127 case result of
3186 case result of
3128 Err e -> do
3187 Err e -> do
3129 logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
3188 logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
@@ -3360,7 +3419,10 @@ waitForInFlightDrain = \deadline -> do
3360-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
3419-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
3361-- nearing expiry. Called by both the background task and the /maintenance/prune endpoint.
3420-- nearing expiry. Called by both the background task and the /maintenance/prune endpoint.
3362pruneAndRetry = \t -> do
3421pruneAndRetry = \t -> do
3363− atomic do set *lastPrunedAt = t
3423+ set *lastPrunedAt = t
3424+ pss <- *prunesSinceSnapshot
3425+ set *prunesSinceSnapshot = pss + 1
3364 pr <- pruneExpired t
3426 pr <- pruneExpired t
3365 -- Counter reconciliation: reconcile all counters every prune cycle.
3427 -- Counter reconciliation: reconcile all counters every prune cycle.
3366 -- Presence and session counters are cheap (just counting list lengths across buckets).
3428 -- Presence and session counters are cheap (just counting list lengths across buckets).
@@ -3537,16 +3599,14 @@ ackPolledMessages = \pubkey cursor t -> do
3537 recipientCounts <- readRcBucket b
3599 recipientCounts <- readRcBucket b
3538 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
3600 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
3539 let ackedMsgIds = do m <- toAck; yield m.messageId
3601 let ackedMsgIds = do m <- toAck; yield m.messageId
3540− let fb = pubkeyBucket pubkey
3541− fwdBkt <- readFwdBucket fb
3602+ fwdBkt <- readFwdBucket b
3542− writeFwdBucket fb (do f <- fwdBkt; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
3603+ writeFwdBucket b (do f <- fwdBkt; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
3543 let ackedFederated = federatedAckEntries toAck t
3604 let ackedFederated = federatedAckEntries toAck t
3544− let ab = pubkeyBucket pubkey
3545− ackedBkt <- readAckedBucket ab
3605+ ackedBkt <- readAckedBucket b
3546 let combinedAckedIds = union ackedBkt ackedFederated
3606 let combinedAckedIds = union ackedBkt ackedFederated
3547 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
3607 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
3548 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
3608 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
3549− writeAckedBucket ab eviction.kept
3609+ writeAckedBucket b eviction.kept
3550 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
3610 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
3551 else yield {pendingAcks: [], evictedEntries: []}
3611 else yield {pendingAcks: [], evictedEntries: []}
@@ -3679,9 +3739,9 @@ bestServerForKey = \entries ->
3679 Just {value: p} -> p.server
3739 Just {value: p} -> p.server
3680 Nothing {} -> ""
3740 Nothing {} -> ""
3682−-- Messages for a pubkey newer than cursor and within messageMaxAge
3742+-- Messages for a pubkey newer than cursor and not expired (per-message expiresAt)
3683pollableMessages = \messages pubkey cursor t ->
3743pollableMessages = \messages pubkey cursor t ->
3684− do m <- messages; where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > cursor; yield m
3744+ do m <- messages; where m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor; yield m
3686-- Build ack entries for messages that came from a remote origin server (need federation ack)
3746-- Build ack entries for messages that came from a remote origin server (need federation ack)
3687federatedAckEntries = \ackedMessages t ->
3747federatedAckEntries = \ackedMessages t ->
@@ -3703,9 +3763,10 @@ forwardCandidates = \targets messages forwards t ->
3703 where m.toKey == target.pubkey
3763 where m.toKey == target.pubkey
3704 where none serverFwdIds (\fid -> fid == m.messageId)
3764 where none serverFwdIds (\fid -> fid == m.messageId)
3705 let origin = if m.originServer == "" then serverHostN else m.originServer
3765 let origin = if m.originServer == "" then serverHostN else m.originServer
3766+ let remainingTtl = if m.expiresAt > 0 && m.expiresAt > t then max 0 (m.expiresAt - t) else 0
3706 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
3767 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
3707 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,
3768 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,
3708− originServer: origin, retries: 0})) [] targets
3769+ originServer: origin, retries: 0, messageTtl: remainingTtl})) [] targets
3711challengeMatch = \challenges pubkey challenge ip t ->
3772challengeMatch = \challenges pubkey challenge ip t ->
@@ -3798,10 +3859,10 @@ handleChallenge = \pubkey clientIp respond -> do
3798 if preAuth != ""
3859 if preAuth != ""
3800 when (preAuth == "ip_rate_limited") (logWarn ("challenge: IP rate limited for " ++ clientIp))
3861 when (preAuth == "ip_rate_limited") (logWarn ("challenge: IP rate limited for " ++ clientIp))
3801− yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0})
3862+ yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: preAuth})
3803 authLimited <- checkAuthGlobalRate t
3864 authLimited <- checkAuthGlobalRate t
3804− if authLimited then yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0})
3865+ if authLimited then yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: "auth_rate_limited"})
3806 challenge <- genToken
3867 challenge <- genToken
3807 result <- atomic do
3868 result <- atomic do
@@ -3814,21 +3875,21 @@ handleChallenge = \pubkey clientIp respond -> do
3814 union others (evictOneOldest forKey (\c -> c.expiresAt))
3875 union others (evictOneOldest forKey (\c -> c.expiresAt))
3815 else active
3876 else active
3816 if count cappedActive >= maxChallenges
3877 if count cappedActive >= maxChallenges
3817− then yield {accepted: False {}, rateLimited: False {}, retryAfterMs: 0}
3878+ then yield {accepted: False {}, rateLimited: False {}, retryAfterMs: 0, error: "capacity"}
3819 challengeRates <- *challengeRates
3880 challengeRates <- *challengeRates
3820 let rateResult = checkPubkeyRate challengeRates pubkey maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
3881 let rateResult = checkPubkeyRate challengeRates pubkey maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
3821 if rateResult.limited
3882 if rateResult.limited
3822− then yield {accepted: False {}, rateLimited: True {}, retryAfterMs: rateResult.retryAfterMs}
3883+ then yield {accepted: False {}, rateLimited: True {}, retryAfterMs: rateResult.retryAfterMs, error: "rate_limited"}
3824 set *challengeRates = rateResult.entries
3885 set *challengeRates = rateResult.entries
3825 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
3886 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
3826− yield {accepted: True {}, rateLimited: False {}, retryAfterMs: 0}
3887+ yield {accepted: True {}, rateLimited: False {}, retryAfterMs: 0, error: ""}
3827 when result.rateLimited
3888 when result.rateLimited
3828 (logWarn ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
3889 (logWarn ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
3829 when (not result.accepted && not result.rateLimited)
3890 when (not result.accepted && not result.rateLimited)
3830 (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
3891 (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
3831− yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0, retryAfterMs: result.retryAfterMs})
3892+ yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0, retryAfterMs: result.retryAfterMs, error: result.error})
3833-- POST /auth/verify: Verify Ed25519 signature over challenge string.
3894-- POST /auth/verify: Verify Ed25519 signature over challenge string.
3834-- On success: creates a session, registers local presence, and gossips online status.
3895-- On success: creates a session, registers local presence, and gossips online status.
@@ -3839,7 +3900,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
3839 if preAuth != ""
3900 if preAuth != ""
3841 when (preAuth == "ip_rate_limited") (logWarn ("verify: IP rate limited for " ++ clientIp))
3902 when (preAuth == "ip_rate_limited") (logWarn ("verify: IP rate limited for " ++ clientIp))
3842− yield (respond {token: "", expiresAt: 0})
3903+ yield (respond {token: "", expiresAt: 0, error: preAuth})
3844 authRateLimited <- atomic do
3905 authRateLimited <- atomic do
3845 challengeRates <- *challengeRates
3906 challengeRates <- *challengeRates
@@ -3848,10 +3909,10 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
3848 yield rateResult.limited
3909 yield rateResult.limited
3849 if authRateLimited then do
3910 if authRateLimited then do
3850 logWarn ("verify: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey)
3911 logWarn ("verify: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey)
3851− yield (respond {token: "", expiresAt: 0})
3912+ yield (respond {token: "", expiresAt: 0, error: "rate_limited"})
3853 authGlobalLimited <- checkAuthGlobalRate t
3914 authGlobalLimited <- checkAuthGlobalRate t
3854− if authGlobalLimited then yield (respond {token: "", expiresAt: 0})
3915+ if authGlobalLimited then yield (respond {token: "", expiresAt: 0, error: "auth_rate_limited"})
3856 -- Per-IP unique key abuse detection: reject IPs authenticating too many distinct pubkeys
3917 -- Per-IP unique key abuse detection: reject IPs authenticating too many distinct pubkeys
3857 uniqueKeyLimited <- atomic do
3918 uniqueKeyLimited <- atomic do
@@ -3863,7 +3924,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
3863 yield result.limited
3924 yield result.limited
3864 if uniqueKeyLimited then do
3925 if uniqueKeyLimited then do
3865 logWarn ("verify: unique key per IP rate limited (" ++ show maxUniqueKeysPerIp ++ " keys/" ++ show uniqueKeyIpWindowMs ++ "ms) for IP " ++ clientIp)
3926 logWarn ("verify: unique key per IP rate limited (" ++ show maxUniqueKeysPerIp ++ " keys/" ++ show uniqueKeyIpWindowMs ++ "ms) for IP " ++ clientIp)
3866− yield (respond {token: "", expiresAt: 0})
3927+ yield (respond {token: "", expiresAt: 0, error: "unique_key_limited"})
3867 else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
3928 else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
3869verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do
3930verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do
@@ -3923,6 +3984,8 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
3923 yield (Just {value: {peerServers: peerServers, revokedCount: 0}})
3984 yield (Just {value: {peerServers: peerServers, revokedCount: 0}})
3925 -- Slow path: revocation or capacity pressure — need all sessions
3986 -- Slow path: revocation or capacity pressure — need all sessions
3987+ -- TODO: Iterate bucket-by-bucket to avoid materializing all 16 buckets.
3988+ -- Currently O(total_sessions); could be O(sessions_in_bucket) for revoke.
3926 allSessions <- readAllSessions
3989 allSessions <- readAllSessions
3927 let revokedCount = if revokeOthers then sessionsForKey else 0
3990 let revokedCount = if revokeOthers then sessionsForKey else 0
3928 let afterRevoke = if revokeOthers then do s <- allSessions; where s.pubkey != pubkey; yield s else allSessions
3991 let afterRevoke = if revokeOthers then do s <- allSessions; where s.pubkey != pubkey; yield s else allSessions
@@ -3932,14 +3995,14 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
3932 writeAllSessions sessionsWithNew
3995 writeAllSessions sessionsWithNew
3933 yield (Just {value: {peerServers: peerServers, revokedCount: revokedCount}})
3996 yield (Just {value: {peerServers: peerServers, revokedCount: revokedCount}})
3934 case authResult of
3997 case authResult of
3935− Nothing {} -> yield (respond {token: "", expiresAt: 0})
3998+ Nothing {} -> yield (respond {token: "", expiresAt: 0, error: "draining"})
3936 Just {value: result} -> do
3999 Just {value: result} -> do
3937 when (revokeOthers && result.revokedCount > 0)
4000 when (revokeOthers && result.revokedCount > 0)
3938 (logInfo ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
4001 (logInfo ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
3939 fork (when federationEnabled
4002 fork (when federationEnabled
3940 (sendBatchedGossip result.peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops))
4003 (sendBatchedGossip result.peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops))
3941− yield (respond {token: token, expiresAt: expiry})
4004+ yield (respond {token: token, expiresAt: expiry, error: ""})
3942− else yield (respond {token: "", expiresAt: 0})
4005+ else yield (respond {token: "", expiresAt: 0, error: "invalid"})
3944-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor
4007-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor
3945-- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.
4008-- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.
@@ -3985,7 +4048,8 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
3985-- Core send logic: dedup, rate limiting, store message, create forwards.
4048-- Core send logic: dedup, rate limiting, store message, create forwards.
3986-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
4049-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
3987-- Caller handles auth, drain checks, and messageId generation.
4050-- Caller handles auth, drain checks, and messageId generation.
4051+-- messageTtl: sender-specified TTL in ms (0 = default unackedMessageMaxAge).
3988−sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
4052+sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTtl t ->
3989 if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4053 if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
3991 let msgBucket = pubkeyBucket to
4055 let msgBucket = pubkeyBucket to
@@ -4027,9 +4091,10 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
4027 seqCounter <- *seqCounter
4091 seqCounter <- *seqCounter
4028 let seqTs = nextSeqTs t seqCounter
4092 let seqTs = nextSeqTs t seqCounter
4029 set *seqCounter = seqTs
4093 set *seqCounter = seqTs
4094+ let msgExpiresAt = computeMessageExpiresAt seqTs messageTtl
4030 writeMessageBucket msgBucket (union bucket
4095 writeMessageBucket msgBucket (union bucket
4031 [{messageId: effectiveMsgId, toKey: to,
4096 [{messageId: effectiveMsgId, toKey: to,
4032− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])
4097+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt}])
4033 set *totalMessageCount = totalMsgCount + 1
4098 set *totalMessageCount = totalMsgCount + 1
4034 writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)
4099 writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)
4035 signalMessageForKey to seqTs
4100 signalMessageForKey to seqTs
@@ -4050,7 +4115,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
4050 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
4115 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
4051 let fbi = pubkeyBucket to
4116 let fbi = pubkeyBucket to
4052 fwdBkt <- readFwdBucket fbi
4117 fwdBkt <- readFwdBucket fbi
4118+ let remainingTtl = if messageTtl > 0 then max 0 (msgExpiresAt - t) else 0
4053− let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}
4119+ let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, messageTtl: remainingTtl}
4054 totalFwdCount <- countAllForwards 0
4120 totalFwdCount <- countAllForwards 0
4055 let cappedFwds = takeRelation (maxForwardsEntries - totalFwdCount) newFwds
4121 let cappedFwds = takeRelation (maxForwardsEntries - totalFwdCount) newFwds
4056 writeFwdBucket fbi (union fwdBkt cappedFwds)
4122 writeFwdBucket fbi (union fwdBkt cappedFwds)
@@ -4063,7 +4129,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
4063 yield {status: status, remoteServers: remoteServers, retryAfterMs: 0}
4129 yield {status: status, remoteServers: remoteServers, retryAfterMs: 0}
4065-- Fork federation forwards for messages that need to be delivered to remote servers
4131-- Fork federation forwards for messages that need to be delivered to remote servers
4066−forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
4132+forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs messageTtl remoteServers ->
4067 when (count remoteServers > 0 && federationEnabled) (
4133 when (count remoteServers > 0 && federationEnabled) (
4068 fork (sendToEach remoteServers (\server -> do
4134 fork (sendToEach remoteServers (\server -> do
4069 stillPending <- atomic do
4135 stillPending <- atomic do
@@ -4073,11 +4139,13 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
4073 fwdT <- now
4139 fwdT <- now
4074 fwdAuth <- generateFederationAuth "forward"
4140 fwdAuth <- generateFederationAuth "forward"
4075 fwdCorrId <- genTokenN 1
4141 fwdCorrId <- genTokenN 1
4142+ -- Compute remaining TTL at forward time (may have decreased since enqueue)
4143+ let fwdTtl = if messageTtl > 0 then max 0 (computeMessageExpiresAt fwdT messageTtl - fwdT) else 0
4076 fwdResult <- fetchFederation server (ForwardMessage {
4144 fwdResult <- fetchFederation server (ForwardMessage {
4077 toKey: to, encryptedBlob: encryptedBlob,
4145 toKey: to, encryptedBlob: encryptedBlob,
4078 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
4146 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
4079 fromServer: serverHostN, requestTimestamp: fwdT,
4147 fromServer: serverHostN, requestTimestamp: fwdT,
4080− authorization: fwdAuth, correlationId: fwdCorrId})
4148+ authorization: fwdAuth, correlationId: fwdCorrId, messageTtl: fwdTtl})
4081 case fwdResult of
4149 case fwdResult of
4082 Err e -> do
4150 Err e -> do
4083 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
4151 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
@@ -4107,7 +4175,7 @@ sendBatchForwardsSequential = \msgs server ->
4107 toKey: r.toKey, encryptedBlob: r.blob,
4175 toKey: r.toKey, encryptedBlob: r.blob,
4108 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
4176 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
4109 fromServer: serverHostN, requestTimestamp: fwdT,
4177 fromServer: serverHostN, requestTimestamp: fwdT,
4110− authorization: fwdAuth, correlationId: fwdCorrId})
4178+ authorization: fwdAuth, correlationId: fwdCorrId, messageTtl: r.messageTtl})
4111 case fwdResult of
4179 case fwdResult of
4112 Err e -> do
4180 Err e -> do
4113 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
4181 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
@@ -4123,7 +4191,7 @@ sendBatchForwardsSequential = \msgs server ->
4123-- POST /messages: Send an encrypted message to a recipient.
4191-- POST /messages: Send an encrypted message to a recipient.
4124-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
4192-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
4125-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
4193-- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
4126−handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp respond -> do
4194+handleSendMessage = \to encryptedBlob timestamp messageId messageTtl authorization clientIp respond -> do
4127 t <- now
4195 t <- now
4128 globalRateLimited <- checkGlobalRate t
4196 globalRateLimited <- checkGlobalRate t
4129 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0})
4197 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0})
@@ -4142,12 +4210,12 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
4142 if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
4210 if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
4143 else withSessionAuth authorization t clientIp
4211 else withSessionAuth authorization t clientIp
4144 (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
4212 (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
4145− (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId t)
4213+ (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId messageTtl t)
4146 case result.status of
4214 case result.status of
4147 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
4215 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
4148 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))
4216 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))
4149 _ -> yield {}
4217 _ -> yield {}
4150− forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.remoteServers
4218+ forkSendForwards effectiveMsgId to encryptedBlob clampedTs messageTtl result.remoteServers
4151 yield (case result.status of
4219 yield (case result.status of
4152 MsDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0}
4220 MsDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0}
4153 MsDelivered {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId, retryAfterMs: 0}
4221 MsDelivered {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId, retryAfterMs: 0}
@@ -4469,7 +4537,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
4469-- (fromServer != originServer) where an intermediate server re-forwards because
4537-- (fromServer != originServer) where an intermediate server re-forwards because
4470-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored
4538-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored
4471-- twice) and forwardCandidates (won't re-forward to the same server).
4539-- twice) and forwardCandidates (won't re-forward to the same server).
4472−handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId authorization respond -> do
4540+handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId messageTtl authorization respond -> do
4473 t <- now
4541 t <- now
4474 when (correlationId != "") (logInfo ("[CORR] " ++ correlationId ++ " forward " ++ messageId ++ " from " ++ fromServer))
4542 when (correlationId != "") (logInfo ("[CORR] " ++ correlationId ++ " forward " ++ messageId ++ " from " ++ fromServer))
4475 withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (True {}) (do
4543 withFederationAuth authorization t respond fromServer "forward" requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (True {}) (do
@@ -4499,17 +4567,19 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
4499 -- Verify recipient has active presence on this server. Peers should only forward
4567 -- Verify recipient has active presence on this server. Peers should only forward
4500 -- to us if gossip indicated the key is here. Accepting messages for completely
4568 -- to us if gossip indicated the key is here. Accepting messages for completely
4501 -- unknown keys wastes mailbox slots that will never be drained by polling.
4569 -- unknown keys wastes mailbox slots that will never be drained by polling.
4570+ -- All sharded collections (presence, messages, acked, delivered, nonces) use
4571+ -- pubkeyBucket(toKey), so a single index serves all reads in this block.
4502− let msgBucket = pubkeyBucket toKey
4572+ let bkt = pubkeyBucket toKey
4503− bkt <- readPresenceBucket msgBucket
4573+ presenceBkt <- readPresenceBucket bkt
4504− let hasPresence = any bkt (\p -> p.pubkey == toKey && p.expiresAt > t)
4574+ let hasPresence = any presenceBkt (\p -> p.pubkey == toKey && p.expiresAt > t)
4505 if not hasPresence
4575 if not hasPresence
4506 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: True {}}
4576 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: True {}}
4508− bucket <- readMessageBucket msgBucket
4578+ bucket <- readMessageBucket bkt
4509− ackedBkt <- readAckedBucket msgBucket
4579+ ackedBkt <- readAckedBucket bkt
4510− delivBkt <- readDelivBucket msgBucket
4580+ delivBkt <- readDelivBucket bkt
4511 -- Sharded nonce check: reads only the bucket for this recipient's pubkey (O(n/16) vs O(n)).
4581 -- Sharded nonce check: reads only the bucket for this recipient's pubkey (O(n/16) vs O(n)).
4512− nonceBkt <- readNonceBucket (pubkeyBucket toKey)
4582+ nonceBkt <- readNonceBucket bkt
4513 -- Nonce check ignores originServer: a (messageId, toKey) pair is globally unique,
4583 -- Nonce check ignores originServer: a (messageId, toKey) pair is globally unique,
4514 -- so replays from any origin should be rejected. This also catches locally-originated
4584 -- so replays from any origin should be rejected. This also catches locally-originated
4515 -- messages (originServer="") whose nonces were recorded by sendMessageCore.
4585 -- messages (originServer="") whose nonces were recorded by sendMessageCore.
@@ -4521,7 +4591,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
4521 if totalMsgCount >= maxMessages
4591 if totalMsgCount >= maxMessages
4522 then yield {forward: ForwardRejected {}, globalCapReached: True {}, recvRateLimited: False {}, noPresence: False {}}
4592 then yield {forward: ForwardRejected {}, globalCapReached: True {}, recvRateLimited: False {}, noPresence: False {}}
4524− recipientCounts <- readRcBucket msgBucket
4594+ recipientCounts <- readRcBucket bkt
4525 let recipientCount = getRecipientCount toKey recipientCounts
4595 let recipientCount = getRecipientCount toKey recipientCounts
4526 if recipientCount >= maxMessagesPerRecipient
4596 if recipientCount >= maxMessagesPerRecipient
4527 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
4597 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
@@ -4541,18 +4611,18 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
4541 seqCounter <- *seqCounter
4611 seqCounter <- *seqCounter
4542 let seqTs = nextSeqTs t seqCounter
4612 let seqTs = nextSeqTs t seqCounter
4543 set *seqCounter = seqTs
4613 set *seqCounter = seqTs
4614+ let fwdExpiresAt = computeMessageExpiresAt seqTs messageTtl
4544− writeMessageBucket msgBucket (union bucket
4615+ writeMessageBucket bkt (union bucket
4545 [{messageId: messageId, toKey: toKey,
4616 [{messageId: messageId, toKey: toKey,
4546− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}])
4617+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt}])
4547 set *totalMessageCount = totalMsgCount + 1
4618 set *totalMessageCount = totalMsgCount + 1
4548− writeRcBucket msgBucket (adjustRecipientCount recipientCounts toKey 1)
4619+ writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
4549 signalMessageForKey toKey seqTs
4620 signalMessageForKey toKey seqTs
4550 -- Record forward nonce for replay protection beyond message/ack dedup lifetime.
4621 -- Record forward nonce for replay protection beyond message/ack dedup lifetime.
4551 -- Sharded: writes only to the bucket for this recipient's pubkey.
4622 -- Sharded: writes only to the bucket for this recipient's pubkey.
4552− let fnb = pubkeyBucket toKey
4553− fwdNonces <- readNonceBucket fnb
4623+ fwdNonces <- readNonceBucket bkt
4554 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
4624 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
4555− writeNonceBucket fnb (if count updatedNonces > maxNoncesPerBucket
4625+ writeNonceBucket bkt (if count updatedNonces > maxNoncesPerBucket
4556 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
4626 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
4557 else updatedNonces)
4627 else updatedNonces)
4558 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
4628 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
@@ -4674,6 +4744,7 @@ handleDrain = \phase authorization respond -> do
4674 bp <- *blockedPeers
4744 bp <- *blockedPeers
4675 yield {version: snapshotVersion, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp}
4745 yield {version: snapshotVersion, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp}
4676 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
4746 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
4747+ atomic do set *prunesSinceSnapshot = 0
4677 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
4748 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
4678 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
4749 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
4679 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
4750 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
@@ -4789,7 +4860,7 @@ handleHealth = \authorization respond -> do
4789 lastFedSuccess <- *lastFederationSuccessAt
4860 lastFedSuccess <- *lastFederationSuccessAt
4790 forwardFailures <- *forwardFailures
4861 forwardFailures <- *forwardFailures
4791 pruneRun <- *pruneRunning
4862 pruneRun <- *pruneRunning
4792− lastSnap <- *lastSnapshotExportAt
4863+ -- lastSnapshotExportAt no longer used in health check (replaced by prunesSinceSnapshot)
4793 dp <- *drainPhase
4864 dp <- *drainPhase
4794 let warmup = t - sa < healthPruneMaxAge
4865 let warmup = t - sa < healthPruneMaxAge
4795 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
4866 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
@@ -4805,8 +4876,12 @@ handleHealth = \authorization respond -> do
4805 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
4876 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
4806 -- Snapshot staleness: unhealthy if maintenance secret is configured but no snapshot
4877 -- Snapshot staleness: unhealthy if maintenance secret is configured but no snapshot
4807 -- has been exported within the staleness window (federation state will be lost on restart)
4878 -- has been exported within the staleness window (federation state will be lost on restart)
4879+ -- Snapshot staleness: uses prune-cycle counter (clock-jump resilient).
4880+ -- Stale when prunesSinceSnapshot exceeds threshold computed from staleness/prune interval.
4881+ pss <- *prunesSinceSnapshot
4882+ let staleThresholdCycles = if backgroundPruneInterval > 0 then snapshotStalenessWarning / backgroundPruneInterval else 0
4808 let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup
4883 let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup
4809− || (lastSnap > 0 && t - lastSnap <= snapshotStalenessWarning)
4884+ || staleThresholdCycles == 0 || pss <= staleThresholdCycles
4810 || (tmc == 0 && presenceCount == 0 && forwardCount == 0)
4885 || (tmc == 0 && presenceCount == 0 && forwardCount == 0)
4811 -- Watchdog: healthy if recently bumped or in warmup (watchdog starts after 10x prune interval)
4886 -- Watchdog: healthy if recently bumped or in warmup (watchdog starts after 10x prune interval)
4812 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
4887 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
@@ -4936,10 +5011,12 @@ handleStatsPeers = \authorization respond -> do
4937 failures <- *forwardFailures
5012 failures <- *forwardFailures
4938 forwards <- readAllForwards
5013 forwards <- readAllForwards
4939− -- Count pending forwards per server in a single pass
5014+ -- Single-pass aggregation: accumulate {server, count} pairs in one fold over forwards.
4940− -- Two-pass aggregation: collect unique servers, then count per server.
5015+ -- O(n*k) where k = unique servers (vs previous O(n*k + k*n) two-pass approach).
4941− let fwdServers = fold (\acc f -> if any acc (\s -> s == f.toServer) then acc else union acc [f.toServer]) [] forwards
5016+ let fwdCounts = fold (\acc f ->
4942− let fwdCounts = do s <- fwdServers; yield {server: s, count: count (do f <- forwards; where f.toServer == s; yield f)}
5017+ if any acc (\c -> c.server == f.toServer)
5018+ then do c <- acc; yield (if c.server == f.toServer then {server: c.server, count: c.count + 1} else c)
5019+ else union acc [{server: f.toServer, count: 1}]) [] forwards
4943 let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
5020 let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
4944 let peers = do f <- failures; yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoffFor f.failures f.server, retriable: shouldRetryServer f.server t failures, pendingForwards: fwdCountFor f.server}
5021 let peers = do f <- failures; yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoffFor f.failures f.server, retriable: shouldRetryServer f.server t failures, pendingForwards: fwdCountFor f.server}
4945 -- Include servers with pending forwards but no failure records
5022 -- Include servers with pending forwards but no failure records
@@ -4973,6 +5050,44 @@ handleStatsKey = \pubkey authorization respond -> do
4973 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, forwardDetails: result.forwardDetails, activePolls: result.activePolls, ok: True {}})
5050 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, forwardDetails: result.forwardDetails, activePolls: result.activePolls, ok: True {}})
4974 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, forwardDetails: [], activePolls: 0, ok: False {}})
5051 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, forwardDetails: [], activePolls: 0, ok: False {}})
5053+-- GET /stats/top-recipients: Return recipients sorted by pending message count (descending).
5054+-- Uses rcBucket data (per-recipient counts) — O(total_unique_recipients) with no message scan.
5055+-- Helps operators identify inbox-bombing targets without knowing the recipient key in advance.
5056+handleTopRecipients = \authorization respond -> do
5058+ globalRateLimited <- checkGlobalRate t
5059+ if globalRateLimited then yield (respond {recipients: [], ok: False {}})
5061+ let authorized = verifyMaintenanceAuth authorization
5064+ -- Collect all recipient counts across 16 buckets and sort by count descending
5065+ allRc <- collectAllRcEntries 0
5066+ let sorted = sortBy (\c -> 0 - c.count) allRc
5067+ let top = takeRelation 20 sorted
5068+ yield (respond {recipients: top, ok: True {}})
5069+ else yield (respond {recipients: [], ok: False {}})
5071+-- Collect RecipientCount entries from all 16 rc buckets
5072+collectAllRcEntries : Int -> IO [RecipientCount]
5073+collectAllRcEntries = \b ->
5074+ if b > 15 then yield []
5076+ bucket <- readRcBucket b
5077+ rest <- collectAllRcEntries (b + 1)
5078+ yield (union bucket rest)
5080+-- GET /ready: Readiness probe — returns true once the server has completed startup.
5081+-- Unlike /health, this is unauthenticated and doesn't check subsystem liveness.
5082+-- Returns false until the first prune cycle completes (server fully initialized).
5083+handleReady = \respond -> do
5086+ lastPrune <- *lastPrunedAt
5087+ -- Ready once the first prune cycle has run (indicates all background loops are alive)
5088+ let isReady = lastPrune > 0 && lastPrune >= sa
5089+ yield (respond {ready: isReady})
4976emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: []}
5091emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: []}
4978-- GET /maintenance/snapshot: Export federation metadata for persistence.
5093-- GET /maintenance/snapshot: Export federation metadata for persistence.
@@ -4994,7 +5109,9 @@ handleSnapshot = \authorization respond -> do
4994 fn <- readAllNonces
5109 fn <- readAllNonces
4995 bp <- *blockedPeers
5110 bp <- *blockedPeers
4996 yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}
5111 yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}
4997− atomic do set *lastSnapshotExportAt = t
5113+ set *lastSnapshotExportAt = t
5114+ set *prunesSinceSnapshot = 0
4998 logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards, " ++ show (count result.forwardNonces) ++ " forwardNonces, " ++ show (count result.blockedPeers) ++ " blockedPeers")
5115 logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards, " ++ show (count result.forwardNonces) ++ " forwardNonces, " ++ show (count result.blockedPeers) ++ " blockedPeers")
4999 yield (respond {ok: True {}, version: snapshotVersion, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces, blockedPeers: result.blockedPeers})
5116 yield (respond {ok: True {}, version: snapshotVersion, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards, forwardNonces: result.forwardNonces, blockedPeers: result.blockedPeers})
@@ -5007,9 +5124,10 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
5007 let authorized = verifyMaintenanceAuth authorization
5124 let authorized = verifyMaintenanceAuth authorization
5008 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
5125 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
5010− -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers
5127+ -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers,
5128+ -- version 3 = pre-messageTtl in ForwardEntry (accept all for backwards compatibility;
5011− -- (accept all for backwards compatibility; missing fields default to empty list)
5129+ -- missing fields default to empty list or 0)
5012− if version != 0 && version != 1 && version != 2 && version != snapshotVersion then do
5130+ if version != 0 && version != 1 && version != 2 && version != 3 && version != snapshotVersion then do
5013 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
5131 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
5014 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
5132 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
@@ -5030,7 +5148,9 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
5030 -- snapshot could inject presence pointing to private IPs, causing forward attempts
5148 -- snapshot could inject presence pointing to private IPs, causing forward attempts
5031 -- to those addresses.
5149 -- to those addresses.
5032 let validPresence = do p <- presenceData; where not (isLocalServer p.server) && not (isBadServerName p.server); yield p
5150 let validPresence = do p <- presenceData; where not (isLocalServer p.server) && not (isBadServerName p.server); yield p
5151+ -- DeliveredEntry has no originServer field (only messageId, toKey, deliveredAt),
5152+ -- so no server-name validation is needed here (unlike AckedEntry/ForwardEntry).
5033− let validDeliveredIds = do d <- deliveredIdsData; where not (isLocalServer d.originServer); yield d
5153+ let validDeliveredIds = deliveredIdsData
5034 let validFailures = do f <- forwardFailuresData; where not (isBadServerName f.server) && not (isLocalServer f.server); yield f
5154 let validFailures = do f <- forwardFailuresData; where not (isBadServerName f.server) && not (isLocalServer f.server); yield f
5035 let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow) + (count presenceData - count validPresence) + (count deliveredIdsData - count validDeliveredIds) + (count forwardFailuresData - count validFailures)
5155 let rejectedCount = (count forwardsData - count validForwards) + (count ackedIdsData - count validAcked) + (count ackOverflowData - count validOverflow) + (count presenceData - count validPresence) + (count deliveredIdsData - count validDeliveredIds) + (count forwardFailuresData - count validFailures)
5036 when (rejectedCount > 0) (logWarn ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))
5156 when (rejectedCount > 0) (logWarn ("restore: rejected " ++ show rejectedCount ++ " imported entries with invalid or local server references"))
@@ -5153,7 +5273,7 @@ handleListMessages = \authorization clientIp respond -> do
5154 result <- atomic do
5274 result <- atomic do
5155 bucket <- readMessageBucket (pubkeyBucket pubkey)
5275 bucket <- readMessageBucket (pubkeyBucket pubkey)
5156− let myMsgs = sortBy (\m -> m.receivedAt) (do m <- bucket; where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge; yield m)
5276+ let myMsgs = sortBy (\m -> m.receivedAt) (do m <- bucket; where m.toKey == pubkey && m.expiresAt > t; yield m)
5157 let capped = takeRelation maxPollEvents myMsgs
5277 let capped = takeRelation maxPollEvents myMsgs
5158 yield {msgs: do m <- capped; yield {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt}, total: count myMsgs}
5278 yield {msgs: do m <- capped; yield {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt}, total: count myMsgs}
5159 yield (respond {messages: result.msgs, authorized: True {}, count: result.total}))
5279 yield (respond {messages: result.msgs, authorized: True {}, count: result.total}))
@@ -5216,6 +5336,59 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
5216 retryEvictedAcks deleteResult.evictedEntries deleteResult.pendingAcks))
5336 retryEvictedAcks deleteResult.evictedEntries deleteResult.pendingAcks))
5217 yield (respond {ok: True {}, authorized: True {}, deletedCount: deleteResult.deletedCount}))
5337 yield (respond {ok: True {}, authorized: True {}, deletedCount: deleteResult.deletedCount}))
5339+-- POST /messages/cancel: Retract a previously sent message before the recipient reads it.
5340+-- Caller provides messageId and recipient pubkey. Knowing the messageId is proof of being
5341+-- the sender (either client-generated or returned only in the send response).
5342+-- Removes from recipient mailbox, cancels pending forwards, sends federation acks.
5343+handleCancelMessage = \messageId to authorization clientIp respond -> do
5346+ if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}})
5348+ globalRateLimited <- checkGlobalRate t
5349+ if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}})
5351+ withSessionAuth authorization t clientIp
5352+ (yield (respond {ok: False {}, authorized: False {}, cancelled: False {}}))
5354+ -- Uses clientWriteRates (like ack/delete) so cancel bursts don't starve lookups
5355+ cancelRateLimited <- atomic do
5356+ cwr <- *clientWriteRates
5357+ let rateResult = checkPubkeyRate cwr senderKey maxClientWriteRatePerKey clientWriteRateWindowMs maxClientWriteRateEntries t
5358+ set *clientWriteRates = rateResult.entries
5359+ yield rateResult.limited
5360+ if cancelRateLimited then yield (respond {ok: False {}, authorized: True {}, cancelled: False {}})
5362+ cancelResult <- atomic do
5363+ let b = pubkeyBucket to
5364+ bucket <- readMessageBucket b
5365+ let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == to; yield m)
5367+ Nothing {} -> yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5368+ Just {value: msg} -> do
5369+ writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)
5370+ tc <- *totalMessageCount
5371+ set *totalMessageCount = max 0 (tc - 1)
5372+ rc <- readRcBucket b
5373+ writeRcBucket b (adjustRecipientCount rc to (0 - 1))
5374+ -- Cancel pending forwards
5375+ fwdBkt <- readFwdBucket b
5376+ writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == to); yield f)
5377+ -- Create ack entries for federated messages so origin server learns about cancel
5378+ if msg.originServer != "" then do
5379+ let ackedFederated = [{messageId: msg.messageId, toKey: msg.toKey, ackedAt: t, originServer: msg.originServer, retries: 0}]
5380+ ackedBkt <- readAckedBucket b
5381+ let combinedAcked = union ackedBkt ackedFederated
5382+ let excessCount = max 0 (count combinedAcked - maxAckedIds)
5383+ let eviction = evictSmallest combinedAcked excessCount (\a -> a.ackedAt)
5384+ writeAckedBucket b eviction.kept
5385+ yield {cancelled: True {}, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
5386+ else yield {cancelled: True {}, pendingAcks: [], evictedEntries: []}
5387+ fork (when (federationEnabled && count cancelResult.pendingAcks > 0) (do
5388+ sendAcksAndClean cancelResult.pendingAcks
5389+ retryEvictedAcks cancelResult.evictedEntries cancelResult.pendingAcks))
5390+ yield (respond {ok: True {}, authorized: True {}, cancelled: cancelResult.cancelled}))
5219-- POST /messages/batch: Send multiple messages in one request.
5392-- POST /messages/batch: Send multiple messages in one request.
5220-- Auth checked once; each message processed independently with its own rate limiting.
5393-- Auth checked once; each message processed independently with its own rate limiting.
5221-- Partial success is possible — check per-message status in results.
5394-- Partial success is possible — check per-message status in results.
@@ -5268,18 +5441,18 @@ sendBatchInner = \senderKey messages t acc ->
5268 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
5441 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
5269 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
5442 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
5270 result <- if t - msg.timestamp > messageMaxAge
5443 result <- if t - msg.timestamp > messageMaxAge
5271− then yield {messageId: effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, retryAfterMs: 0}
5444+ then yield {messageId: effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, messageTtl: 0, retryAfterMs: 0}
5273 let clampedTs = clampTimestamp msg.timestamp t
5446 let clampedTs = clampTimestamp msg.timestamp t
5274 sendResult <- atomic do
5447 sendResult <- atomic do
5275 dp <- *drainPhase
5448 dp <- *drainPhase
5276 if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
5449 if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
5277− else sendMessageCore senderKey msg.to msg.encryptedBlob clampedTs effectiveMsgId t
5450+ else sendMessageCore senderKey msg.to msg.encryptedBlob clampedTs effectiveMsgId msg.messageTtl t
5278 case sendResult.status of
5451 case sendResult.status of
5279 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
5452 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
5280 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")
5453 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")
5281 _ -> yield {}
5454 _ -> yield {}
5282− yield {messageId: effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryAfterMs}
5455+ yield {messageId: effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, messageTtl: msg.messageTtl, retryAfterMs: sendResult.retryAfterMs}
5283 sendBatchInner senderKey split.rest t (union acc [result])
5456 sendBatchInner senderKey split.rest t (union acc [result])
5285-- POST /maintenance/block: Add a server to the federation blocklist.
5458-- POST /maintenance/block: Add a server to the federation blocklist.
@@ -5355,9 +5528,10 @@ handleAdminListSessions = \pubkey authorization respond -> do
5356 result <- atomic do
5529 result <- atomic do
5357 allSessions <- readAllSessions
5530 allSessions <- readAllSessions
5358− -- Show only shard prefix (4 hex chars = 16 bits) for identification without aiding brute-force.
5531+ -- Return only metadata (IP, timestamps) — no token fragments. Even a 4-char prefix
5359− -- Previous 8-char prefix exposed 32 bits, narrowing brute-force from 192 to 160 bits.
5532+ -- (16 bits) narrows brute-force search space and can be correlated with token bucket
5533+ -- indices. Sessions are identifiable by (ip, createdAt) pair without token exposure.
5360− let keySessions = do s <- allSessions; where s.pubkey == pubkey; yield {token: take 4 s.token ++ "...", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt}
5534+ let keySessions = do s <- allSessions; where s.pubkey == pubkey; yield {token: "***", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt}
5361 yield {sessions: keySessions, total: count keySessions}
5535 yield {sessions: keySessions, total: count keySessions}
5362 yield (respond {sessions: result.sessions, count: result.total, ok: True {}})
5536 yield (respond {sessions: result.sessions, count: result.total, ok: True {}})
@@ -5482,6 +5656,7 @@ requestPath = \req -> case req of
5482 SendBatch {} -> "POST /messages/batch"
5656 SendBatch {} -> "POST /messages/batch"
5483 ListMessages {} -> "GET /messages/list"
5657 ListMessages {} -> "GET /messages/list"
5484 DeleteMessages {} -> "POST /messages/delete"
5658 DeleteMessages {} -> "POST /messages/delete"
5659+ CancelMessage {} -> "POST /messages/cancel"
5485 LookupPresence {} -> "GET /lookup"
5660 LookupPresence {} -> "GET /lookup"
5486 RecvGossip {} -> "POST /federation/gossip"
5661 RecvGossip {} -> "POST /federation/gossip"
5487 ForwardMessage {} -> "POST /federation/forward"
5662 ForwardMessage {} -> "POST /federation/forward"
@@ -5506,6 +5681,8 @@ requestPath = \req -> case req of
5506 Stats {} -> "GET /stats"
5681 Stats {} -> "GET /stats"
5507 StatsPeers {} -> "GET /stats/peers"
5682 StatsPeers {} -> "GET /stats/peers"
5508 StatsKey {} -> "GET /stats/key"
5683 StatsKey {} -> "GET /stats/key"
5684+ TopRecipients {} -> "GET /stats/top-recipients"
5685+ Ready {} -> "GET /ready"
5510-- Extract client IP for access logging (empty for federation/maintenance routes)
5687-- Extract client IP for access logging (empty for federation/maintenance routes)
5511requestClientIp = \req -> case req of
5688requestClientIp = \req -> case req of
@@ -5520,6 +5697,7 @@ requestClientIp = \req -> case req of
5520 SendBatch {clientIp} -> clientIp
5697 SendBatch {clientIp} -> clientIp
5521 ListMessages {clientIp} -> clientIp
5698 ListMessages {clientIp} -> clientIp
5522 DeleteMessages {clientIp} -> clientIp
5699 DeleteMessages {clientIp} -> clientIp
5700+ CancelMessage {clientIp} -> clientIp
5523 LookupPresence {clientIp} -> clientIp
5701 LookupPresence {clientIp} -> clientIp
5524 VerifyToken {clientIp} -> clientIp
5702 VerifyToken {clientIp} -> clientIp
5525 RecvGossip {} -> ""
5703 RecvGossip {} -> ""
@@ -5544,6 +5722,8 @@ requestClientIp = \req -> case req of
5544 Stats {} -> ""
5722 Stats {} -> ""
5545 StatsPeers {} -> ""
5723 StatsPeers {} -> ""
5546 StatsKey {} -> ""
5724 StatsKey {} -> ""
5725+ TopRecipients {} -> ""
5548-- Slow request threshold for access logging (ms)
5728-- Slow request threshold for access logging (ms)
5549slowRequestThreshold : Int<Ms>
5729slowRequestThreshold : Int<Ms>
@@ -5555,15 +5735,16 @@ serveInner = \req -> case req of
5555 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
5735 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
5556 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
5736 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
5557 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond
5737 Poll {cursor, generation, watchKeys, authorization, clientIp, respond} -> handlePoll cursor generation watchKeys authorization clientIp respond
5558− SendMessage {to, encryptedBlob, timestamp, messageId, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId authorization clientIp respond
5738+ SendMessage {to, encryptedBlob, timestamp, messageId, messageTtl, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp messageId messageTtl authorization clientIp respond
5559 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
5739 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
5560 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
5740 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
5561 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
5741 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
5562 ListMessages {authorization, clientIp, respond} -> handleListMessages authorization clientIp respond
5742 ListMessages {authorization, clientIp, respond} -> handleListMessages authorization clientIp respond
5563 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
5743 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
5744+ CancelMessage {messageId, to, authorization, clientIp, respond} -> handleCancelMessage messageId to authorization clientIp respond
5564 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
5745 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
5565 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
5746 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
5566− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId authorization respond
5747+ ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, messageTtl, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId messageTtl authorization respond
5567 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
5748 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
5568 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
5749 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
5569 Prune {authorization, respond} -> handlePrune authorization respond
5750 Prune {authorization, respond} -> handlePrune authorization respond
@@ -5585,6 +5766,8 @@ serveInner = \req -> case req of
5585 Stats {authorization, respond} -> handleStats authorization respond
5766 Stats {authorization, respond} -> handleStats authorization respond
5586 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
5767 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
5587 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
5768 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
5769+ TopRecipients {authorization, respond} -> handleTopRecipients authorization respond
5770+ Ready {respond} -> handleReady respond
5589-- Access-logged request handler: wraps serveInner with timing and correlation ID.
5772-- Access-logged request handler: wraps serveInner with timing and correlation ID.
5590-- Logs all non-poll requests with a 6-hex-char request ID for log correlation.
5773-- Logs all non-poll requests with a 6-hex-char request ID for log correlation.
@@ -5769,17 +5952,18 @@ backgroundSnapshot = do
5769 atomic do
5952 atomic do
5770 set *lastAutoSnapshotExportAt = t
5953 set *lastAutoSnapshotExportAt = t
5771 set *lastSnapshotFingerprint = fingerprint
5954 set *lastSnapshotFingerprint = fingerprint
5955+ set *prunesSinceSnapshot = 0
5772 logInfo ("snapshot: auto-exported full state (" ++ show result.presence ++ " presence, " ++ show result.forwards ++ " forwards)")))
5956 logInfo ("snapshot: auto-exported full state (" ++ show result.presence ++ " presence, " ++ show result.forwards ++ " forwards)")))
5773− -- Warn if federation state hasn't been exported recently (will be lost on restart)
5957+ -- Warn if federation state hasn't been exported recently (will be lost on restart).
5774− lastExport <- *lastSnapshotExportAt
5958+ -- Uses prune-cycle counter instead of wall-clock comparison to survive clock jumps.
5775− lastAutoExport <- *lastAutoSnapshotExportAt
5959+ -- Threshold: snapshotStalenessWarning / backgroundPruneInterval prune cycles.
5776− -- Consider auto-exports as valid exports for staleness check
5960+ pss <- *prunesSinceSnapshot
5777− let effectiveLastExport = max lastExport lastAutoExport
5961+ let staleCycles = if backgroundPruneInterval > 0 then snapshotStalenessWarning / backgroundPruneInterval else 0
5778 when (maintenanceSecret != "" && snapshotStalenessWarning > 0
5962 when (maintenanceSecret != "" && snapshotStalenessWarning > 0
5779− && (effectiveLastExport == 0 || t - effectiveLastExport > snapshotStalenessWarning)
5963+ && staleCycles > 0 && pss > staleCycles
5780 && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))
5964 && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))
5782− logError ("no snapshot exported in " ++ show (if effectiveLastExport == 0 then 0 else (t - effectiveLastExport) / 1000) ++ "s — federation state will be lost on restart (presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " deliveredIds=" ++ show result.deliveredIds ++ ")")
5966+ logError ("no snapshot exported in " ++ show pss ++ " prune cycles (~" ++ show (pss * backgroundPruneInterval / 1000) ++ "s) — federation state will be lost on restart (presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " deliveredIds=" ++ show result.deliveredIds ++ ")")
5783 logError "ACTION REQUIRED: run GET /maintenance/snapshot to persist federation state before restart")
5967 logError "ACTION REQUIRED: run GET /maintenance/snapshot to persist federation state before restart")
5784 else yield {}
5968 else yield {}
5785 backgroundSnapshot
5969 backgroundSnapshot