@@ -15,6 +15,9 @@
1515
1616-- Server identity and federation credentials
1717serverHost = "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.
1821maintenanceSecret = "" -- set to enable /maintenance/* and /stats endpoints
1922federationScheme = "https://" -- scheme for federation HTTP calls
2023-- NOTE: Outbound federation HTTP calls have no runtime-level timeout. The fetchFederation
@@ -23,12 +26,20 @@ federationScheme = "https://" -- scheme for federation HTTP calls
2326-- slow peer from consuming all concurrency slots, though truly hung connections still hold
2427-- one slot until the forked task terminates.
2528-- 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.
2633messageMaxAge : Int<Ms>
2734messageMaxAge = 2592000000 -- 30 days
2835maxPresenceTtl : Int<S>
2936maxPresenceTtl = 3600 -- max presence TTL from gossip
3037maxPubkeyLen = 256
3138maxBlobLen = 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
3243maxSignatureLen = 256
3344maxIpLen = 45 -- max length of an IP address string (IPv6 = 45 chars); used in matchIp instead of maxServerNameLen
3445maxMessageIds = 100
@@ -148,7 +159,7 @@ federationLatencyWindowMs : Int<Ms>
148159federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
149160
150161-- Snapshot format version (increment on schema changes to prevent corrupt restores)
151−snapshotVersion = 3
162+snapshotVersion = 4 -- bumped for messageTtl field in ForwardEntry
152163
153164-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
154165maxGossipEventsPerServer = 500 -- max total gossip events per server per window
@@ -314,12 +325,16 @@ type Timestamp = Int<Ms>
314325type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp, ip: Text}
315326-- Authenticated session; token is a random hex string
316327type 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
318−type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp}
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.
331+type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp, expiresAt: Timestamp}
319332-- Presence: tracks that a pubkey is online at a given server, with expiry
320333type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
321−-- Queued federation forward: message waiting to be delivered to a remote server
322−type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int}
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.
337+type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, retries: Int, messageTtl: Int}
323338-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
324339type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName, retries: Int}
325340-- 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}
357372type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}
358373
359374-- 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}
361376type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
362377
363378-- ============================================
@@ -707,6 +722,11 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
707722-- backgroundSnapshot compares current counters against this to skip redundant exports.
708723*lastSnapshotFingerprint : Int
709724
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
729+
710730-- Cached health result: returned when /health is rate-limited instead of false all-healthy.
711731-- Updated on every successful (non-rate-limited) health check.
712732*cachedHealthOk : Bool
@@ -728,10 +748,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
728748route AuthApi where
729749 /auth
730750 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}
732752 = ReqChallenge
733753 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}
735755 = VerifyAuth
736756 POST {targetToken: Text} /revoke headers {authorization: Text, clientIp: Text}
737757 -> {ok: Bool, authorized: Bool}
@@ -748,8 +768,11 @@ route ClientApi where
748768 -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool}
749769 = Poll
750770 /messages
751− -- Send an E2E encrypted message to a recipient pubkey
752− POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text, clientIp: Text}
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.
775+ POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, messageTtl: Int} / headers {authorization: Text, clientIp: Text}
753776 -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int}
754777 = SendMessage
755778 -- Check delivery status of a sent message (caller must know recipient to prove ownership)
@@ -772,6 +795,13 @@ route ClientApi where
772795 POST {messageIds: [MessageIdHex]} /delete headers {authorization: Text, clientIp: Text}
773796 -> {ok: Bool, authorized: Bool, deletedCount: Int}
774797 = 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}
804+ = CancelMessage
775805 -- Look up which server a pubkey is present on (for federation routing)
776806 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
777807 -> {server: AnyServerName, authorized: Bool}
@@ -786,8 +816,10 @@ route FederationApi where
786816 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /gossip headers {authorization: Text}
787817 -> {ok: Bool, error: Text}
788818 = RecvGossip
789− -- Receive a forwarded message from a peer server on behalf of a sender
790− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /forward headers {authorization: Text}
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.
822+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text, messageTtl: Int} /forward headers {authorization: Text}
791823 -> {ok: Bool, error: Text}
792824 = ForwardMessage
793825 -- Acknowledge delivery of messages back to the origin server
@@ -884,8 +916,19 @@ route StatsApi where
884916 GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text}
885917 -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, forwardDetails: [{server: Text, retries: Int, createdAt: Timestamp}], activePolls: Int, ok: Bool}
886918 = 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}
922+ = TopRecipients
923+
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.
928+route ReadyApi where
929+ GET /ready -> {ready: Bool} = Ready
887930
888−route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | MaintenanceApi | HealthApi | StatsApi
931+route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | MaintenanceApi | HealthApi | StatsApi | ReadyApi
889932
890933-- ============================================
891934-- Helpers
@@ -1107,6 +1150,7 @@ isBadServerName = \sn ->
11071150 || matchesHostnameOrSubdomain sn "localhost"
11081151 || matchesHostname sn "localhost.localdomain"
11091152 || matchesHostnameOrSubdomain sn "local"
1153+ || containsSuffix ".localhost" sn -- RFC 6761: foo.localhost resolves to loopback
11101154 || matchesHostnameOrSubdomain sn "internal"
11111155 || matchesHostnameOrSubdomain sn "metadata"
11121156 || take 4 sn == "127." || take 3 sn == "10."
@@ -1692,24 +1736,26 @@ transformSelectedBuckets = \buckets f ->
16921736 transformSelectedBuckets rest f
16931737
16941738-- Collect unique bucket indices for a set of pubkeys
1739+-- O(1) per step with union [b] acc (set, order irrelevant, max 16 entries)
16951740uniqueBucketsFor = \items getKey ->
16961741 fold (\acc x ->
16971742 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
16991744
17001745-- Prune a single message bucket and atomically adjust shared counters.
17011746-- Each bucket prune decrements totalMessageCount and its rcBucket by exactly
17021747-- what was removed, preserving concurrent modifications from poll-ack or send.
1703−pruneMessageBucket = \b messageCutoff -> atomic do
1748+-- Uses per-message expiresAt (respects sender-specified TTL).
1749+pruneMessageBucket = \b t -> atomic do
17041750 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
17061752 let removedCount = count bucket - count kept
17071753 writeMessageBucket b kept
17081754 if removedCount > 0 then do
17091755 tc <- *totalMessageCount
17101756 set *totalMessageCount = max 0 (tc - removedCount)
17111757 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
17131759 writeRcBucket b (batchDecrementRecipientCounts rc (countByRecipient removed))
17141760 else yield {}
17151761
@@ -1720,15 +1766,15 @@ pruneMessageBucket = \b messageCutoff -> atomic do
17201766-- is fast (list filter + counter decrement), (2) prune runs every ~30s, not on the
17211767-- hot path, (3) Knot's fork is fire-and-forget so we can't collect per-bucket
17221768-- deltas and apply them in a single final atomic block.
1723−pruneMessageBucketsConcurrent = \b messageCutoff ->
1769+pruneMessageBucketsConcurrent = \b t ->
17241770 if b > 15 then yield {}
17251771 else do
17261772 bucket <- readMessageBucket b
1727− if count bucket == 0 then pruneMessageBucketsConcurrent (b + 1) messageCutoff
1773+ if count bucket == 0 then pruneMessageBucketsConcurrent (b + 1) t
17281774 else if b < 15 then do
1729− fork (pruneMessageBucket b messageCutoff)
1730− pruneMessageBucketsConcurrent (b + 1) messageCutoff
1731− else pruneMessageBucket b messageCutoff
1775+ fork (pruneMessageBucket b t)
1776+ pruneMessageBucketsConcurrent (b + 1) t
1777+ else pruneMessageBucket b t
17321778
17331779-- Per-bucket message reconciliation: reads one bucket per atomic block instead of all 16
17341780-- at once, reducing lock contention during reconciliation. Each pubkey maps to exactly one
@@ -2294,6 +2340,13 @@ fetchFederation = \server request -> do
22942340clampTtl : Int<S> -> Int<S>
22952341clampTtl = clamp 1 maxPresenceTtl
22962342
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
2349+
22972350clampTimestamp : Timestamp -> Timestamp -> Timestamp
22982351clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
22992352
@@ -2502,13 +2555,18 @@ removeOneForKey = \items pubkey ->
25022555
25032556-- Extract unique remote (non-local) server names from a list of items.
25042557-- 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.
25052563collectPeerServers : [a] -> (a -> Text) -> [Text]
25062564collectPeerServers = \items getServer ->
25072565 fold (\acc item ->
25082566 let s = getServer item in
25092567 if isLocalServer s then acc
25102568 else if any acc (\a -> a == s) then acc
2511− else union acc [s]) [] items
2569+ else union [s] acc) [] items
25122570
25132571samePubkeyServer = \a b -> a.pubkey == b.pubkey && a.server == b.server
25142572pubkeyServer = \e -> {pubkey: e.pubkey, server: e.server}
@@ -2647,8 +2705,9 @@ pruneExpired = \t -> do
26472705 -- and touches only its own message bucket + rcBucket. The only shared write is
26482706 -- *totalMessageCount (a single Int), so STM contention is minimal.
26492707 -- Fork first 15 buckets; run the last one in the foreground to bound concurrency.
2650− let messageCutoff = t - unackedMessageMaxAge
2651− pruneMessageBucketsConcurrent 0 messageCutoff
2708+ -- Uses per-message expiresAt instead of a global cutoff, so sender-specified TTLs
2709+ -- are respected during pruning.
2710+ pruneMessageBucketsConcurrent 0 t
26522711 -- Prune federation state (split into per-collection blocks to reduce lock contention
26532712 -- with poll-ack paths that write to ackedIds/deliveredIds concurrently).
26542713 let retainCutoff = t - messageMaxAge
@@ -3123,7 +3182,7 @@ retryForwardsForServer = \entries remaining ->
31233182 toKey: f.toKey, encryptedBlob: f.encryptedBlob,
31243183 timestamp: f.timestamp, originServer: f.originServer, messageId: f.messageId,
31253184 fromServer: serverHostN, requestTimestamp: t,
3126− authorization: auth, correlationId: corrId})
3185+ authorization: auth, correlationId: corrId, messageTtl: f.messageTtl})
31273186 case result of
31283187 Err e -> do
31293188 logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
@@ -3360,7 +3419,10 @@ waitForInFlightDrain = \deadline -> do
33603419-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
33613420-- nearing expiry. Called by both the background task and the /maintenance/prune endpoint.
33623421pruneAndRetry = \t -> do
3363− atomic do set *lastPrunedAt = t
3422+ atomic do
3423+ set *lastPrunedAt = t
3424+ pss <- *prunesSinceSnapshot
3425+ set *prunesSinceSnapshot = pss + 1
33643426 pr <- pruneExpired t
33653427 -- Counter reconciliation: reconcile all counters every prune cycle.
33663428 -- Presence and session counters are cheap (just counting list lengths across buckets).
@@ -3537,16 +3599,14 @@ ackPolledMessages = \pubkey cursor t -> do
35373599 recipientCounts <- readRcBucket b
35383600 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
35393601 let ackedMsgIds = do m <- toAck; yield m.messageId
3540− let fb = pubkeyBucket pubkey
3541− fwdBkt <- readFwdBucket fb
3542− writeFwdBucket fb (do f <- fwdBkt; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
3602+ fwdBkt <- readFwdBucket b
3603+ writeFwdBucket b (do f <- fwdBkt; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
35433604 let ackedFederated = federatedAckEntries toAck t
3544− let ab = pubkeyBucket pubkey
3545− ackedBkt <- readAckedBucket ab
3605+ ackedBkt <- readAckedBucket b
35463606 let combinedAckedIds = union ackedBkt ackedFederated
35473607 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
35483608 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
3549− writeAckedBucket ab eviction.kept
3609+ writeAckedBucket b eviction.kept
35503610 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
35513611 else yield {pendingAcks: [], evictedEntries: []}
35523612
@@ -3679,9 +3739,9 @@ bestServerForKey = \entries ->
36793739 Just {value: p} -> p.server
36803740 Nothing {} -> ""
36813741
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)
36833743pollableMessages = \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
36853745
36863746-- Build ack entries for messages that came from a remote origin server (need federation ack)
36873747federatedAckEntries = \ackedMessages t ->
@@ -3703,9 +3763,10 @@ forwardCandidates = \targets messages forwards t ->
37033763 where m.toKey == target.pubkey
37043764 where none serverFwdIds (\fid -> fid == m.messageId)
37053765 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
37063767 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
37073768 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp,
3708− originServer: origin, retries: 0})) [] targets
3769+ originServer: origin, retries: 0, messageTtl: remainingTtl})) [] targets
37093770
37103771
37113772challengeMatch = \challenges pubkey challenge ip t ->
@@ -3798,10 +3859,10 @@ handleChallenge = \pubkey clientIp respond -> do
37983859 if preAuth != ""
37993860 then do
38003861 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})
38023863 else do
38033864 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"})
38053866 else do
38063867 challenge <- genToken
38073868 result <- atomic do
@@ -3814,21 +3875,21 @@ handleChallenge = \pubkey clientIp respond -> do
38143875 union others (evictOneOldest forKey (\c -> c.expiresAt))
38153876 else active
38163877 if count cappedActive >= maxChallenges
3817− then yield {accepted: False {}, rateLimited: False {}, retryAfterMs: 0}
3878+ then yield {accepted: False {}, rateLimited: False {}, retryAfterMs: 0, error: "capacity"}
38183879 else do
38193880 challengeRates <- *challengeRates
38203881 let rateResult = checkPubkeyRate challengeRates pubkey maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
38213882 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"}
38233884 else do
38243885 set *challengeRates = rateResult.entries
38253886 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: ""}
38273888 when result.rateLimited
38283889 (logWarn ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
38293890 when (not result.accepted && not result.rateLimited)
38303891 (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})
38323893
38333894-- POST /auth/verify: Verify Ed25519 signature over challenge string.
38343895-- On success: creates a session, registers local presence, and gossips online status.
@@ -3839,7 +3900,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
38393900 if preAuth != ""
38403901 then do
38413902 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})
38433904 else do
38443905 authRateLimited <- atomic do
38453906 challengeRates <- *challengeRates
@@ -3848,10 +3909,10 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
38483909 yield rateResult.limited
38493910 if authRateLimited then do
38503911 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"})
38523913 else do
38533914 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"})
38553916 else do
38563917 -- Per-IP unique key abuse detection: reject IPs authenticating too many distinct pubkeys
38573918 uniqueKeyLimited <- atomic do
@@ -3863,7 +3924,7 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
38633924 yield result.limited
38643925 if uniqueKeyLimited then do
38653926 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"})
38673928 else verifyAuthInner pubkey challenge signature revokeOthers clientIp respond t
38683929
38693930verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t -> do
@@ -3923,6 +3984,8 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
39233984 yield (Just {value: {peerServers: peerServers, revokedCount: 0}})
39243985 else do
39253986 -- 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.
39263989 allSessions <- readAllSessions
39273990 let revokedCount = if revokeOthers then sessionsForKey else 0
39283991 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 ->
39323995 writeAllSessions sessionsWithNew
39333996 yield (Just {value: {peerServers: peerServers, revokedCount: revokedCount}})
39343997 case authResult of
3935− Nothing {} -> yield (respond {token: "", expiresAt: 0})
3998+ Nothing {} -> yield (respond {token: "", expiresAt: 0, error: "draining"})
39363999 Just {value: result} -> do
39374000 when (revokeOthers && result.revokedCount > 0)
39384001 (logInfo ("verify: revokeOthers revoked " ++ show result.revokedCount ++ " sessions for " ++ pubkey))
39394002 fork (when federationEnabled
39404003 (sendBatchedGossip result.peerServers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}] gossipOriginHops))
3941− yield (respond {token: token, expiresAt: expiry})
3942− else yield (respond {token: "", expiresAt: 0})
4004+ yield (respond {token: token, expiresAt: expiry, error: ""})
4005+ else yield (respond {token: "", expiresAt: 0, error: "invalid"})
39434006
39444007-- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor
39454008-- 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
39854048-- Core send logic: dedup, rate limiting, store message, create forwards.
39864049-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
39874050-- Caller handles auth, drain checks, and messageId generation.
3988−sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
4051+-- messageTtl: sender-specified TTL in ms (0 = default unackedMessageMaxAge).
4052+sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTtl t ->
39894053 if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
39904054 else do
39914055 let msgBucket = pubkeyBucket to
@@ -4027,9 +4091,10 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
40274091 seqCounter <- *seqCounter
40284092 let seqTs = nextSeqTs t seqCounter
40294093 set *seqCounter = seqTs
4094+ let msgExpiresAt = computeMessageExpiresAt seqTs messageTtl
40304095 writeMessageBucket msgBucket (union bucket
40314096 [{messageId: effectiveMsgId, toKey: to,
4032− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}])
4097+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt}])
40334098 set *totalMessageCount = totalMsgCount + 1
40344099 writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)
40354100 signalMessageForKey to seqTs
@@ -4050,7 +4115,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
40504115 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
40514116 let fbi = pubkeyBucket to
40524117 fwdBkt <- readFwdBucket fbi
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}
4118+ let remainingTtl = if messageTtl > 0 then max 0 (msgExpiresAt - t) else 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}
40544120 totalFwdCount <- countAllForwards 0
40554121 let cappedFwds = takeRelation (maxForwardsEntries - totalFwdCount) newFwds
40564122 writeFwdBucket fbi (union fwdBkt cappedFwds)
@@ -4063,7 +4129,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
40634129 yield {status: status, remoteServers: remoteServers, retryAfterMs: 0}
40644130
40654131-- 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 ->
40674133 when (count remoteServers > 0 && federationEnabled) (
40684134 fork (sendToEach remoteServers (\server -> do
40694135 stillPending <- atomic do
@@ -4073,11 +4139,13 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
40734139 fwdT <- now
40744140 fwdAuth <- generateFederationAuth "forward"
40754141 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
40764144 fwdResult <- fetchFederation server (ForwardMessage {
40774145 toKey: to, encryptedBlob: encryptedBlob,
40784146 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
40794147 fromServer: serverHostN, requestTimestamp: fwdT,
4080− authorization: fwdAuth, correlationId: fwdCorrId})
4148+ authorization: fwdAuth, correlationId: fwdCorrId, messageTtl: fwdTtl})
40814149 case fwdResult of
40824150 Err e -> do
40834151 logWarn ("federation: forward to " ++ server ++ " failed for " ++ effectiveMsgId ++ ": " ++ show e)
@@ -4107,7 +4175,7 @@ sendBatchForwardsSequential = \msgs server ->
41074175 toKey: r.toKey, encryptedBlob: r.blob,
41084176 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
41094177 fromServer: serverHostN, requestTimestamp: fwdT,
4110− authorization: fwdAuth, correlationId: fwdCorrId})
4178+ authorization: fwdAuth, correlationId: fwdCorrId, messageTtl: r.messageTtl})
41114179 case fwdResult of
41124180 Err e -> do
41134181 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
@@ -4123,7 +4191,7 @@ sendBatchForwardsSequential = \msgs server ->
41234191-- POST /messages: Send an encrypted message to a recipient.
41244192-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
41254193-- 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
41274195 t <- now
41284196 globalRateLimited <- checkGlobalRate t
41294197 if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0})
@@ -4142,12 +4210,12 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
41424210 if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
41434211 else withSessionAuth authorization t clientIp
41444212 (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)
41464214 case result.status of
41474215 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
41484216 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))
41494217 _ -> yield {}
4150− forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.remoteServers
4218+ forkSendForwards effectiveMsgId to encryptedBlob clampedTs messageTtl result.remoteServers
41514219 yield (case result.status of
41524220 MsDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0}
41534221 MsDelivered {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId, retryAfterMs: 0}
@@ -4469,7 +4537,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
44694537-- (fromServer != originServer) where an intermediate server re-forwards because
44704538-- the recipient moved. Cascade is bounded by dedup (same messageId won't be stored
44714539-- 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
44734541 t <- now
44744542 when (correlationId != "") (logInfo ("[CORR] " ++ correlationId ++ " forward " ++ messageId ++ " from " ++ fromServer))
44754543 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
44994567 -- Verify recipient has active presence on this server. Peers should only forward
45004568 -- to us if gossip indicated the key is here. Accepting messages for completely
45014569 -- unknown keys wastes mailbox slots that will never be drained by polling.
4502− let msgBucket = pubkeyBucket toKey
4503− bkt <- readPresenceBucket msgBucket
4504− let hasPresence = any bkt (\p -> p.pubkey == toKey && p.expiresAt > t)
4570+ -- All sharded collections (presence, messages, acked, delivered, nonces) use
4571+ -- pubkeyBucket(toKey), so a single index serves all reads in this block.
4572+ let bkt = pubkeyBucket toKey
4573+ presenceBkt <- readPresenceBucket bkt
4574+ let hasPresence = any presenceBkt (\p -> p.pubkey == toKey && p.expiresAt > t)
45054575 if not hasPresence
45064576 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: True {}}
45074577 else do
4508− bucket <- readMessageBucket msgBucket
4509− ackedBkt <- readAckedBucket msgBucket
4510− delivBkt <- readDelivBucket msgBucket
4578+ bucket <- readMessageBucket bkt
4579+ ackedBkt <- readAckedBucket bkt
4580+ delivBkt <- readDelivBucket bkt
45114581 -- 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
45134583 -- Nonce check ignores originServer: a (messageId, toKey) pair is globally unique,
45144584 -- so replays from any origin should be rejected. This also catches locally-originated
45154585 -- messages (originServer="") whose nonces were recorded by sendMessageCore.
@@ -4521,7 +4591,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
45214591 if totalMsgCount >= maxMessages
45224592 then yield {forward: ForwardRejected {}, globalCapReached: True {}, recvRateLimited: False {}, noPresence: False {}}
45234593 else do
4524− recipientCounts <- readRcBucket msgBucket
4594+ recipientCounts <- readRcBucket bkt
45254595 let recipientCount = getRecipientCount toKey recipientCounts
45264596 if recipientCount >= maxMessagesPerRecipient
45274597 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
@@ -4541,18 +4611,18 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
45414611 seqCounter <- *seqCounter
45424612 let seqTs = nextSeqTs t seqCounter
45434613 set *seqCounter = seqTs
4544− writeMessageBucket msgBucket (union bucket
4614+ let fwdExpiresAt = computeMessageExpiresAt seqTs messageTtl
4615+ writeMessageBucket bkt (union bucket
45454616 [{messageId: messageId, toKey: toKey,
4546− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}])
4617+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt}])
45474618 set *totalMessageCount = totalMsgCount + 1
4548− writeRcBucket msgBucket (adjustRecipientCount recipientCounts toKey 1)
4619+ writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
45494620 signalMessageForKey toKey seqTs
45504621 -- Record forward nonce for replay protection beyond message/ack dedup lifetime.
45514622 -- Sharded: writes only to the bucket for this recipient's pubkey.
4552− let fnb = pubkeyBucket toKey
4553− fwdNonces <- readNonceBucket fnb
4623+ fwdNonces <- readNonceBucket bkt
45544624 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
45564626 then evictNonces updatedNonces maxNoncesPerBucket forwardNonceRetention t
45574627 else updatedNonces)
45584628 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
@@ -4674,6 +4744,7 @@ handleDrain = \phase authorization respond -> do
46744744 bp <- *blockedPeers
46754745 yield {version: snapshotVersion, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp}
46764746 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
4747+ atomic do set *prunesSinceSnapshot = 0
46774748 logInfo ("drain: auto-snapshot exported (" ++ show (count drainResult.localEntries) ++ " presence, " ++ show (count flushState.forwards) ++ " forwards)")
46784749 -- Fork best-effort flush of pending federation ops, then gossip, then wait for in-flight.
46794750 -- Operator polls /stats for drainCompletedAt > 0 to know when safe to stop.
@@ -4789,7 +4860,7 @@ handleHealth = \authorization respond -> do
47894860 lastFedSuccess <- *lastFederationSuccessAt
47904861 forwardFailures <- *forwardFailures
47914862 pruneRun <- *pruneRunning
4792− lastSnap <- *lastSnapshotExportAt
4863+ -- lastSnapshotExportAt no longer used in health check (replaced by prunesSinceSnapshot)
47934864 dp <- *drainPhase
47944865 let warmup = t - sa < healthPruneMaxAge
47954866 let pruneStuck = pruneRun > 0 && t - pruneRun > backgroundPruneInterval * 2
@@ -4805,8 +4876,12 @@ handleHealth = \authorization respond -> do
48054876 || (lastFedSuccess > 0 && t - lastFedSuccess < backgroundPruneInterval * 5)
48064877 -- Snapshot staleness: unhealthy if maintenance secret is configured but no snapshot
48074878 -- 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
48084883 let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup
4809− || (lastSnap > 0 && t - lastSnap <= snapshotStalenessWarning)
4884+ || staleThresholdCycles == 0 || pss <= staleThresholdCycles
48104885 || (tmc == 0 && presenceCount == 0 && forwardCount == 0)
48114886 -- Watchdog: healthy if recently bumped or in warmup (watchdog starts after 10x prune interval)
48124887 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
@@ -4936,10 +5011,12 @@ handleStatsPeers = \authorization respond -> do
49365011 then do
49375012 failures <- *forwardFailures
49385013 forwards <- readAllForwards
4939− -- Count pending forwards per server in a single pass
4940− -- Two-pass aggregation: collect unique servers, then count per server.
4941− let fwdServers = fold (\acc f -> if any acc (\s -> s == f.toServer) then acc else union acc [f.toServer]) [] forwards
4942− let fwdCounts = do s <- fwdServers; yield {server: s, count: count (do f <- forwards; where f.toServer == s; yield f)}
5014+ -- Single-pass aggregation: accumulate {server, count} pairs in one fold over forwards.
5015+ -- O(n*k) where k = unique servers (vs previous O(n*k + k*n) two-pass approach).
5016+ let fwdCounts = fold (\acc 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
49435020 let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
49445021 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}
49455022 -- Include servers with pending forwards but no failure records
@@ -4973,6 +5050,44 @@ handleStatsKey = \pubkey authorization respond -> do
49735050 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, forwardDetails: result.forwardDetails, activePolls: result.activePolls, ok: True {}})
49745051 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, forwardDetails: [], activePolls: 0, ok: False {}})
49755052
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
5057+ t <- now
5058+ globalRateLimited <- checkGlobalRate t
5059+ if globalRateLimited then yield (respond {recipients: [], ok: False {}})
5060+ else do
5061+ let authorized = verifyMaintenanceAuth authorization
5062+ if authorized
5063+ then do
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 {}})
5070+
5071+-- Collect RecipientCount entries from all 16 rc buckets
5072+collectAllRcEntries : Int -> IO [RecipientCount]
5073+collectAllRcEntries = \b ->
5074+ if b > 15 then yield []
5075+ else do
5076+ bucket <- readRcBucket b
5077+ rest <- collectAllRcEntries (b + 1)
5078+ yield (union bucket rest)
5079+
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
5084+ t <- now
5085+ sa <- *startedAt
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})
5090+
49765091emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: []}
49775092
49785093-- GET /maintenance/snapshot: Export federation metadata for persistence.
@@ -4994,7 +5109,9 @@ handleSnapshot = \authorization respond -> do
49945109 fn <- readAllNonces
49955110 bp <- *blockedPeers
49965111 yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}
4997− atomic do set *lastSnapshotExportAt = t
5112+ atomic do
5113+ set *lastSnapshotExportAt = t
5114+ set *prunesSinceSnapshot = 0
49985115 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")
49995116 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})
50005117
@@ -5007,9 +5124,10 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
50075124 let authorized = verifyMaintenanceAuth authorization
50085125 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
50095126 else do
5010− -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces, version 2 = pre-blockedPeers
5011− -- (accept all for backwards compatibility; missing fields default to empty list)
5012− if version != 0 && version != 1 && version != 2 && version != snapshotVersion then do
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;
5129+ -- missing fields default to empty list or 0)
5130+ if version != 0 && version != 1 && version != 2 && version != 3 && version != snapshotVersion then do
50135131 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
50145132 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
50155133 else do
@@ -5030,7 +5148,9 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
50305148 -- snapshot could inject presence pointing to private IPs, causing forward attempts
50315149 -- to those addresses.
50325150 let validPresence = do p <- presenceData; where not (isLocalServer p.server) && not (isBadServerName p.server); yield p
5033− let validDeliveredIds = do d <- deliveredIdsData; where not (isLocalServer d.originServer); yield d
5151+ -- DeliveredEntry has no originServer field (only messageId, toKey, deliveredAt),
5152+ -- so no server-name validation is needed here (unlike AckedEntry/ForwardEntry).
5153+ let validDeliveredIds = deliveredIdsData
50345154 let validFailures = do f <- forwardFailuresData; where not (isBadServerName f.server) && not (isLocalServer f.server); yield f
50355155 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)
50365156 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
51535273 else do
51545274 result <- atomic do
51555275 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)
51575277 let capped = takeRelation maxPollEvents myMsgs
51585278 yield {msgs: do m <- capped; yield {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt}, total: count myMsgs}
51595279 yield (respond {messages: result.msgs, authorized: True {}, count: result.total}))
@@ -5216,6 +5336,59 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
52165336 retryEvictedAcks deleteResult.evictedEntries deleteResult.pendingAcks))
52175337 yield (respond {ok: True {}, authorized: True {}, deletedCount: deleteResult.deletedCount}))
52185338
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
5344+ t <- now
5345+ dp <- *drainPhase
5346+ if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}})
5347+ else do
5348+ globalRateLimited <- checkGlobalRate t
5349+ if globalRateLimited then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}})
5350+ else
5351+ withSessionAuth authorization t clientIp
5352+ (yield (respond {ok: False {}, authorized: False {}, cancelled: False {}}))
5353+ (\senderKey -> do
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 {}})
5361+ else do
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)
5366+ case toCancel of
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}))
5391+
52195392-- POST /messages/batch: Send multiple messages in one request.
52205393-- Auth checked once; each message processed independently with its own rate limiting.
52215394-- Partial success is possible — check per-message status in results.
@@ -5268,18 +5441,18 @@ sendBatchInner = \senderKey messages t acc ->
52685441 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
52695442 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
52705443 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}
52725445 else do
52735446 let clampedTs = clampTimestamp msg.timestamp t
52745447 sendResult <- atomic do
52755448 dp <- *drainPhase
52765449 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
52785451 case sendResult.status of
52795452 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
52805453 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")
52815454 _ -> 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}
52835456 sendBatchInner senderKey split.rest t (union acc [result])
52845457
52855458-- POST /maintenance/block: Add a server to the federation blocklist.
@@ -5355,9 +5528,10 @@ handleAdminListSessions = \pubkey authorization respond -> do
53555528 else do
53565529 result <- atomic do
53575530 allSessions <- readAllSessions
5358− -- Show only shard prefix (4 hex chars = 16 bits) for identification without aiding brute-force.
5359− -- Previous 8-char prefix exposed 32 bits, narrowing brute-force from 192 to 160 bits.
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}
5531+ -- Return only metadata (IP, timestamps) — no token fragments. Even a 4-char prefix
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.
5534+ let keySessions = do s <- allSessions; where s.pubkey == pubkey; yield {token: "***", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt}
53615535 yield {sessions: keySessions, total: count keySessions}
53625536 yield (respond {sessions: result.sessions, count: result.total, ok: True {}})
53635537
@@ -5482,6 +5656,7 @@ requestPath = \req -> case req of
54825656 SendBatch {} -> "POST /messages/batch"
54835657 ListMessages {} -> "GET /messages/list"
54845658 DeleteMessages {} -> "POST /messages/delete"
5659+ CancelMessage {} -> "POST /messages/cancel"
54855660 LookupPresence {} -> "GET /lookup"
54865661 RecvGossip {} -> "POST /federation/gossip"
54875662 ForwardMessage {} -> "POST /federation/forward"
@@ -5506,6 +5681,8 @@ requestPath = \req -> case req of
55065681 Stats {} -> "GET /stats"
55075682 StatsPeers {} -> "GET /stats/peers"
55085683 StatsKey {} -> "GET /stats/key"
5684+ TopRecipients {} -> "GET /stats/top-recipients"
5685+ Ready {} -> "GET /ready"
55095686
55105687-- Extract client IP for access logging (empty for federation/maintenance routes)
55115688requestClientIp = \req -> case req of
@@ -5520,6 +5697,7 @@ requestClientIp = \req -> case req of
55205697 SendBatch {clientIp} -> clientIp
55215698 ListMessages {clientIp} -> clientIp
55225699 DeleteMessages {clientIp} -> clientIp
5700+ CancelMessage {clientIp} -> clientIp
55235701 LookupPresence {clientIp} -> clientIp
55245702 VerifyToken {clientIp} -> clientIp
55255703 RecvGossip {} -> ""
@@ -5544,6 +5722,8 @@ requestClientIp = \req -> case req of
55445722 Stats {} -> ""
55455723 StatsPeers {} -> ""
55465724 StatsKey {} -> ""
5725+ TopRecipients {} -> ""
5726+ Ready {} -> ""
55475727
55485728-- Slow request threshold for access logging (ms)
55495729slowRequestThreshold : Int<Ms>
@@ -5555,15 +5735,16 @@ serveInner = \req -> case req of
55555735 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
55565736 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
55575737 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
55595739 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
55605740 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
55615741 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
55625742 ListMessages {authorization, clientIp, respond} -> handleListMessages authorization clientIp respond
55635743 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
5744+ CancelMessage {messageId, to, authorization, clientIp, respond} -> handleCancelMessage messageId to authorization clientIp respond
55645745 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
55655746 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
55675748 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
55685749 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
55695750 Prune {authorization, respond} -> handlePrune authorization respond
@@ -5585,6 +5766,8 @@ serveInner = \req -> case req of
55855766 Stats {authorization, respond} -> handleStats authorization respond
55865767 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
55875768 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
5769+ TopRecipients {authorization, respond} -> handleTopRecipients authorization respond
5770+ Ready {respond} -> handleReady respond
55885771
55895772-- Access-logged request handler: wraps serveInner with timing and correlation ID.
55905773-- Logs all non-poll requests with a 6-hex-char request ID for log correlation.
@@ -5769,17 +5952,18 @@ backgroundSnapshot = do
57695952 atomic do
57705953 set *lastAutoSnapshotExportAt = t
57715954 set *lastSnapshotFingerprint = fingerprint
5955+ set *prunesSinceSnapshot = 0
57725956 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)
5774− lastExport <- *lastSnapshotExportAt
5775− lastAutoExport <- *lastAutoSnapshotExportAt
5776− -- Consider auto-exports as valid exports for staleness check
5777− let effectiveLastExport = max lastExport lastAutoExport
5957+ -- Warn if federation state hasn't been exported recently (will be lost on restart).
5958+ -- Uses prune-cycle counter instead of wall-clock comparison to survive clock jumps.
5959+ -- Threshold: snapshotStalenessWarning / backgroundPruneInterval prune cycles.
5960+ pss <- *prunesSinceSnapshot
5961+ let staleCycles = if backgroundPruneInterval > 0 then snapshotStalenessWarning / backgroundPruneInterval else 0
57785962 when (maintenanceSecret != "" && snapshotStalenessWarning > 0
5779− && (effectiveLastExport == 0 || t - effectiveLastExport > snapshotStalenessWarning)
5963+ && staleCycles > 0 && pss > staleCycles
57805964 && (result.ackedIds > 0 || result.deliveredIds > 0 || result.forwards > 0 || result.presence > 0))
57815965 (do
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 ++ ")")
57835967 logError "ACTION REQUIRED: run GET /maintenance/snapshot to persist federation state before restart")
57845968 else yield {}
57855969 backgroundSnapshot