@@ -15,6 +15,7 @@
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-- 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-- transition window, so operators can rotate without restarting. E.g., add a
20-- maintenanceSecretPrevious field that's accepted for a configurable overlap period.
21-- maintenanceSecretPrevious field that's accepted for a configurable overlap period.
@@ -170,6 +171,8 @@ maxGossipEventRateEntries = 100
170maxGlobalFederationConcurrency = 30
171maxGlobalFederationConcurrency = 30
171-- Peer blocklist: max blocked servers tracked at runtime (managed via /maintenance/block, /maintenance/unblock)
172-- Peer blocklist: max blocked servers tracked at runtime (managed via /maintenance/block, /maintenance/unblock)
172maxBlockedPeers = 100
173maxBlockedPeers = 100
174+-- Peer allowlist: when non-empty, only these servers can federate with us (managed via /maintenance/allow, /maintenance/unallow)
175+maxAllowedPeers = 100
174-- Drain timeout: max time for drain flush operations before force-completing
177-- Drain timeout: max time for drain flush operations before force-completing
175drainTimeoutMs : Int<Ms>
178drainTimeoutMs : Int<Ms>
@@ -245,6 +248,12 @@ maxPendingAckCleanups = 10000
245-- Per-key deferred ack cleanup cap (prevents one noisy key from dominating the queue)
248-- Per-key deferred ack cleanup cap (prevents one noisy key from dominating the queue)
246maxPendingAckCleanupsPerKey = 100
249maxPendingAckCleanupsPerKey = 100
251+-- Message expiration notifications: track expired locally-originated messages so senders
252+-- can learn their message was never read (returned in poll response).
253+maxSenderExpirations = 10000
254+maxSenderExpirationsPerKey = 100
255+-- Max pubkeys in a batch lookup request
256+maxBatchLookupKeys = 50
248-- Health endpoint rate limiting (global, since /health has no clientIp)
257-- Health endpoint rate limiting (global, since /health has no clientIp)
249maxHealthRate = 100 -- max health checks per window
258maxHealthRate = 100 -- max health checks per window
250healthRateWindowMs : Int<Ms>
259healthRateWindowMs : Int<Ms>
@@ -328,7 +337,9 @@ type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp, ip: Text,
328-- Stored message; originServer is "" for locally-originated messages, set to sender server for federated forwards.
337-- 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.
338-- expiresAt: computed from sender-specified TTL or default unackedMessageMaxAge.
330-- Messages with expiresAt <= now are pruned and excluded from polls.
339-- Messages with expiresAt <= now are pruned and excluded from polls.
340+-- senderKey: pubkey of the authenticated sender (locally-originated only; "" for federation forwards).
341+-- Used by /messages/cancel to verify the caller is the actual sender.
331−type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp, expiresAt: Timestamp}
342+type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp, expiresAt: Timestamp, senderKey: Text}
332-- Presence: tracks that a pubkey is online at a given server, with expiry
343-- Presence: tracks that a pubkey is online at a given server, with expiry
333type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
344type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
334-- Queued federation forward: message waiting to be delivered to a remote server.
345-- Queued federation forward: message waiting to be delivered to a remote server.
@@ -375,6 +386,10 @@ type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Te
375type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, messageTtl: Int}
386type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId, messageTtl: Int}
376type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
387type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
389+-- Sender expiration notification: tracks when a locally-originated message expired unread.
390+-- Pruned after messageMaxAge. Included in the sender's next poll response.
391+type SenderExpiration = {senderKey: Text, messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}
378-- ============================================
393-- ============================================
379-- Persisted State
394-- Persisted State
380-- ============================================
395-- ============================================
@@ -694,6 +709,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
694-- Blocked federation peers (managed at runtime via /maintenance/block and /maintenance/unblock)
709-- Blocked federation peers (managed at runtime via /maintenance/block and /maintenance/unblock)
695*blockedPeers : [Text]
710*blockedPeers : [Text]
712+-- Allowed federation peers (when non-empty, only these servers can federate with us)
713+-- Managed via /maintenance/allow and /maintenance/unallow. Empty = open federation.
714+*allowedPeers : [Text]
697-- Deferred ack cleanups: origin message reclamation queued by federation ack handlers,
716-- Deferred ack cleanups: origin message reclamation queued by federation ack handlers,
698-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
717-- processed by the prune cycle after ackGracePeriod elapses (replaces fork+sleep pattern).
699*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
718*pendingAckCleanups : [{messageId: MessageIdHex, toKey: PubkeyHex, cleanupAfter: Timestamp}]
@@ -702,6 +721,10 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
702-- (key-rotation abuse bypassing per-key rate limits). Pruned with other rate limiters.
721-- (key-rotation abuse bypassing per-key rate limits). Pruned with other rate limiters.
703*uniqueKeyIpRates : [UniqueKeyIpEntry]
722*uniqueKeyIpRates : [UniqueKeyIpEntry]
724+-- Sender expiration notifications: messages that expired unread, queued for the sender's next poll.
725+-- Populated during prune for locally-originated messages (senderKey != "").
726+*senderExpirations : [SenderExpiration]
705-- Runtime-configurable rate limit overrides (0 = use compile-time default).
728-- Runtime-configurable rate limit overrides (0 = use compile-time default).
706-- Adjusted via /maintenance/rate-limits POST endpoint without recompile+restart.
729-- Adjusted via /maintenance/rate-limits POST endpoint without recompile+restart.
707*rateOverrideGlobalRequest : Int
730*rateOverrideGlobalRequest : Int
@@ -765,7 +788,7 @@ route AuthApi where
765route ClientApi where
788route ClientApi where
766 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
789 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
767 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
790 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
768− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool}
791+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, messageExpirations: [{messageId: MessageIdHex, toKey: PubkeyHex, expiredAt: Timestamp}]}
770 /messages
793 /messages
771 -- Send an E2E encrypted message to a recipient pubkey.
794 -- Send an E2E encrypted message to a recipient pubkey.
@@ -806,6 +829,10 @@ route ClientApi where
806 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
829 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
807 -> {server: AnyServerName, authorized: Bool}
830 -> {server: AnyServerName, authorized: Bool}
808 = LookupPresence
831 = LookupPresence
832+ -- Batch lookup: look up presence for multiple pubkeys in one request
833+ POST {pubkeys: [PubkeyHex]} /lookup/batch headers {authorization: Text, clientIp: Text}
834+ -> {results: [{pubkey: PubkeyHex, server: AnyServerName}], authorized: Bool}
835+ = BatchLookupPresence
810-- Server-to-server federation API (authenticated via domain callback verification)
837-- Server-to-server federation API (authenticated via domain callback verification)
811-- correlationId: optional trace ID for cross-server request tracing (empty = not provided).
838-- correlationId: optional trace ID for cross-server request tracing (empty = not provided).
@@ -896,6 +923,22 @@ route MaintenanceApi where
896 POST {globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int} /maintenance/rate-limits headers {authorization: Text}
923 POST {globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int} /maintenance/rate-limits headers {authorization: Text}
897 -> {ok: Bool}
924 -> {ok: Bool}
898 = SetRateLimits
925 = SetRateLimits
926+ -- Flush forward nonces (recovery from corruption or capacity issues)
927+ POST {} /maintenance/nonces/flush headers {authorization: Text}
928+ -> {ok: Bool, flushedCount: Int}
930+ -- Allow a federation peer (when allowlist is non-empty, only allowed peers can federate)
931+ POST {server: ServerName} /maintenance/allow headers {authorization: Text}
932+ -> {ok: Bool, allowed: Int}
934+ -- Remove a server from the federation allowlist
935+ POST {server: ServerName} /maintenance/unallow headers {authorization: Text}
936+ -> {ok: Bool, removed: Bool}
938+ -- List currently allowed federation peers (empty = open federation)
939+ GET /maintenance/allowed headers {authorization: Text}
940+ -> {peers: [Text], ok: Bool}
900route HealthApi where
943route HealthApi where
901 GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool, draining: Bool} = Health
944 GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool, snapshotHealthy: Bool, watchdogHealthy: Bool, draining: Bool} = Health
@@ -1224,6 +1267,10 @@ verifyMaintenanceAuth = verifyAuthWith (\token -> verifySecret token maintenance
1224isBlockedPeer : Text -> [Text] -> Bool
1267isBlockedPeer : Text -> [Text] -> Bool
1225isBlockedPeer = \server blocked -> any blocked (\b -> b == server)
1268isBlockedPeer = \server blocked -> any blocked (\b -> b == server)
1270+-- Check if a server passes the allowlist (empty allowlist = all allowed)
1271+isAllowedByAllowlist : Text -> [Text] -> Bool
1272+isAllowedByAllowlist = \server allowed -> count allowed == 0 || any allowed (\a -> a == server)
1227-- Generate a single-use token for outbound federation auth.
1274-- Generate a single-use token for outbound federation auth.
1228-- The receiving server calls back to /federation/verify to confirm we issued it.
1275-- The receiving server calls back to /federation/verify to confirm we issued it.
1229-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
1276-- scope binds the token to a request type ("gossip", "forward", "ack") so a token
@@ -1746,6 +1793,8 @@ uniqueBucketsFor = \items getKey ->
1746-- Each bucket prune decrements totalMessageCount and its rcBucket by exactly
1793-- Each bucket prune decrements totalMessageCount and its rcBucket by exactly
1747-- what was removed, preserving concurrent modifications from poll-ack or send.
1794-- what was removed, preserving concurrent modifications from poll-ack or send.
1748-- Uses per-message expiresAt (respects sender-specified TTL).
1795-- Uses per-message expiresAt (respects sender-specified TTL).
1796+-- Also queues expiration notifications for locally-originated messages (senderKey != "")
1797+-- so the sender can learn their message expired unread via the next poll response.
1749pruneMessageBucket = \b t -> atomic do
1798pruneMessageBucket = \b t -> atomic do
1750 bucket <- readMessageBucket b
1799 bucket <- readMessageBucket b
1751 let kept = do m <- bucket; where m.expiresAt > t; yield m
1800 let kept = do m <- bucket; where m.expiresAt > t; yield m
@@ -1757,6 +1806,14 @@ pruneMessageBucket = \b t -> atomic do
1757 rc <- readRcBucket b
1806 rc <- readRcBucket b
1758 let removed = do m <- bucket; where m.expiresAt <= t; yield m
1807 let removed = do m <- bucket; where m.expiresAt <= t; yield m
1759 writeRcBucket b (batchDecrementRecipientCounts rc (countByRecipient removed))
1808 writeRcBucket b (batchDecrementRecipientCounts rc (countByRecipient removed))
1809+ -- Queue expiration notifications for locally-originated messages with known sender
1810+ let localExpired = do m <- removed; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, expiredAt: t}
1811+ when (count localExpired > 0) (do
1812+ se <- *senderExpirations
1813+ let combined = union se localExpired
1814+ set *senderExpirations = if count combined > maxSenderExpirations
1815+ then evictOneOldest combined (\e -> e.expiredAt)
1760 else yield {}
1817 else yield {}
1762-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
1819-- Prune all 16 message buckets with concurrency: fork first 15, run last in foreground.
@@ -2112,6 +2169,20 @@ computeForwardStatsInner = \b t acc ->
2112 let newMaxDepth = fold (\mx s -> max mx s.count) acc.maxDepth updatedServerCounts
2169 let newMaxDepth = fold (\mx s -> max mx s.count) acc.maxDepth updatedServerCounts
2113 computeForwardStatsInner (b + 1) t {totalCount: acc.totalCount + bucketCount, oldestAge: max acc.oldestAge bucketOldest, serverCount: count updatedServerCounts, maxDepth: newMaxDepth, serverCounts: updatedServerCounts}
2170 computeForwardStatsInner (b + 1) t {totalCount: acc.totalCount + bucketCount, oldestAge: max acc.oldestAge bucketOldest, serverCount: count updatedServerCounts, maxDepth: newMaxDepth, serverCounts: updatedServerCounts}
2172+-- Remove forward entries matching a server and message IDs, scanning per-bucket.
2173+-- Only rewrites buckets that actually changed (avoids writeAllForwards repartitioning).
2174+-- Returns all removed forward entries across all buckets.
2175+removeForwardsByServer = \server messageIds b acc ->
2176+ if b > 15 then yield acc
2178+ removed <- atomic do
2179+ bucket <- readFwdBucket b
2180+ let matched = do f <- bucket; where f.toServer == server && any messageIds (\mid -> f.messageId == mid); yield f
2181+ when (count matched > 0)
2182+ (writeFwdBucket b (do f <- bucket; where not (f.toServer == server && any messageIds (\mid -> f.messageId == mid)); yield f))
2184+ removeForwardsByServer server messageIds (b + 1) (if count removed > 0 then union acc removed else acc)
2115-- Read only the forward buckets at the given indices (avoids reading all 16)
2186-- Read only the forward buckets at the given indices (avoids reading all 16)
2116readSelectedFwdBuckets : [Int] -> IO [ForwardEntry]
2187readSelectedFwdBuckets : [Int] -> IO [ForwardEntry]
2117readSelectedFwdBuckets = \buckets ->
2188readSelectedFwdBuckets = \buckets ->
@@ -2669,15 +2740,16 @@ prunePresenceBucketN = \b t -> atomic do
2669 yield {expiredLocal: expiredLocal, keptPeerServers: collectPeerServers kept (\p -> p.server), allPeerServers: collectPeerServers bucket (\p -> p.server), removedCount: count expired}
2740 yield {expiredLocal: expiredLocal, keptPeerServers: collectPeerServers kept (\p -> p.server), allPeerServers: collectPeerServers bucket (\p -> p.server), removedCount: count expired}
2671-- Recursively prune all 16 presence buckets and aggregate results.
2742-- Recursively prune all 16 presence buckets and aggregate results.
2743+-- Collects raw server lists from each bucket; deduplication is done once
2744+-- at the end by the caller (pruneExpired) to avoid O(n*k) per-bucket merges.
2672prunePresenceBuckets = \b t ->
2745prunePresenceBuckets = \b t ->
2673 if b > 15 then yield {expiredLocal: [], keptPeerServers: [], allPeerServers: [], totalRemoved: 0}
2746 if b > 15 then yield {expiredLocal: [], keptPeerServers: [], allPeerServers: [], totalRemoved: 0}
2675 r <- prunePresenceBucketN b t
2748 r <- prunePresenceBucketN b t
2676 rest <- prunePresenceBuckets (b + 1) t
2749 rest <- prunePresenceBuckets (b + 1) t
2677− let mergePeers = \a ps -> fold (\acc s -> if any acc (\x -> x == s) then acc else union acc [s]) a ps
2678 yield {expiredLocal: union r.expiredLocal rest.expiredLocal,
2750 yield {expiredLocal: union r.expiredLocal rest.expiredLocal,
2679− keptPeerServers: mergePeers r.keptPeerServers rest.keptPeerServers,
2751+ keptPeerServers: union r.keptPeerServers rest.keptPeerServers,
2680− allPeerServers: mergePeers r.allPeerServers rest.allPeerServers,
2752+ allPeerServers: union r.allPeerServers rest.allPeerServers,
2681 totalRemoved: r.removedCount + rest.totalRemoved}
2753 totalRemoved: r.removedCount + rest.totalRemoved}
2683-- Remove all expired entries from every collection. Returns expired local presence
2755-- Remove all expired entries from every collection. Returns expired local presence
@@ -2687,12 +2759,17 @@ pruneExpired = \t -> do
2687 -- Each bucket is independent: all entries for a pubkey hash to the same bucket,
2759 -- Each bucket is independent: all entries for a pubkey hash to the same bucket,
2688 -- so per-bucket before/after comparison is equivalent to the global check.
2760 -- so per-bucket before/after comparison is equivalent to the global check.
2689 presenceResults <- prunePresenceBuckets 0 t
2761 presenceResults <- prunePresenceBuckets 0 t
2762+ -- Dedup peer servers once after collecting from all 16 buckets (O(n) total
2763+ -- vs O(n*k) if deduped per-bucket during aggregation).
2764+ let dedupServers = \servers -> fold (\acc s -> if any acc (\x -> x == s) then acc else union acc [s]) [] servers
2765+ let keptPeers = dedupServers presenceResults.keptPeerServers
2766+ let allPeers = dedupServers presenceResults.allPeerServers
2690 atomic do
2767 atomic do
2691 tpc <- *totalPresenceCount
2768 tpc <- *totalPresenceCount
2692 set *totalPresenceCount = max 0 (tpc - presenceResults.totalRemoved)
2769 set *totalPresenceCount = max 0 (tpc - presenceResults.totalRemoved)
2693− set *knownPeerServers = presenceResults.keptPeerServers
2770+ set *knownPeerServers = keptPeers
2694 when (presenceResults.totalRemoved > 0) (set *presenceVersion = t)
2771 when (presenceResults.totalRemoved > 0) (set *presenceVersion = t)
2695− let presenceResult = {expiredLocal: presenceResults.expiredLocal, peerServers: presenceResults.allPeerServers}
2772+ let presenceResult = {expiredLocal: presenceResults.expiredLocal, peerServers: allPeers}
2696 -- Prune auth state (challenges in their own block; sessions counted during transform)
2773 -- Prune auth state (challenges in their own block; sessions counted during transform)
2697 atomic do
2774 atomic do
2698 challenges <- *challenges
2775 challenges <- *challenges
@@ -2797,6 +2874,9 @@ pruneExpired = \t -> do
2797 -- Prune stale deferred ack cleanups (older than messageMaxAge are never useful)
2874 -- Prune stale deferred ack cleanups (older than messageMaxAge are never useful)
2798 pendingCleanups <- *pendingAckCleanups
2875 pendingCleanups <- *pendingAckCleanups
2799 set *pendingAckCleanups = do c <- pendingCleanups; where t - c.cleanupAfter < messageMaxAge; yield c
2876 set *pendingAckCleanups = do c <- pendingCleanups; where t - c.cleanupAfter < messageMaxAge; yield c
2877+ -- Prune old sender expiration notifications (keep for one poll cycle + grace)
2878+ senderExp <- *senderExpirations
2879+ set *senderExpirations = do e <- senderExp; where t - e.expiredAt < sessionExpiry; yield e
2800 -- Reconcile global federation connection counter with actual in-flight state
2880 -- Reconcile global federation connection counter with actual in-flight state
2801 -- (self-heals drift from leaked slots where a forked task died without decrementing)
2881 -- (self-heals drift from leaked slots where a forked task died without decrementing)
2802 inFlightRec <- *federationInFlight
2882 inFlightRec <- *federationInFlight
@@ -2968,13 +3048,19 @@ checkPreAuth = \clientIp t -> do
2968 else yield ""
3048 else yield ""
2970withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition clearOnAuth action -> do
3050withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition clearOnAuth action -> do
2971− -- Check blocklist before spending resources on verification callback
3051+ -- Check blocklist and allowlist before spending resources on verification callback
2972 blocked <- *blockedPeers
3052 blocked <- *blockedPeers
2973 if isBlockedPeer fromServer blocked
3053 if isBlockedPeer fromServer blocked
2975 logWarn ("federation: blocked peer " ++ fromServer ++ " rejected")
3055 logWarn ("federation: blocked peer " ++ fromServer ++ " rejected")
2976 yield (respond {ok: False {}, error: "blocked"})
3056 yield (respond {ok: False {}, error: "blocked"})
3058+ allowed <- *allowedPeers
3059+ if not (isAllowedByAllowlist fromServer allowed)
3061+ logWarn ("federation: peer " ++ fromServer ++ " not in allowlist, rejected")
3062+ yield (respond {ok: False {}, error: "not_allowed"})
2978 verified <- verifyFederationCallback fromServer authorization scope
3064 verified <- verifyFederationCallback fromServer authorization scope
2979 if not verified
3065 if not verified
2980 then yield (respond {ok: False {}, error: "unauthorized"})
3066 then yield (respond {ok: False {}, error: "unauthorized"})
@@ -4017,10 +4103,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
4017 let cursorWasReset = generationMismatch && cursor > 0
4103 let cursorWasReset = generationMismatch && cursor > 0
4018 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
4104 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
4019 dp <- *drainPhase
4105 dp <- *drainPhase
4020− if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset})
4106+ if dp >= 2 then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0, cursorReset: cursorWasReset, messageExpirations: []})
4022 globalRateLimited <- checkGlobalRate t
4108 globalRateLimited <- checkGlobalRate t
4023− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset})
4109+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs, cursorReset: cursorWasReset, messageExpirations: []})
4025 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
4111 -- Snapshot presenceVersion before long-poll wait to skip unchanged lookups
4026 startPresenceVersion <- *presenceVersion
4112 startPresenceVersion <- *presenceVersion
@@ -4043,14 +4129,32 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
4043 sendAcksAndClean result.pendingAcks
4129 sendAcksAndClean result.pendingAcks
4044 retryEvictedAcks result.evictedEntries result.pendingAcks))
4130 retryEvictedAcks result.evictedEntries result.pendingAcks))
4045 let isHeartbeat = result.authorized && count result.events == 0
4131 let isHeartbeat = result.authorized && count result.events == 0
4132+ -- Deliver message expiration notifications: atomically read and drain expirations
4133+ -- for this sender's pubkey so each notification is delivered exactly once.
4134+ expirations <- if not result.authorized then yield []
4135+ else case extractToken authorization of
4136+ Nothing {} -> yield []
4137+ Just {value: token} -> do
4138+ let bucket = tokenBucket token
4139+ sessions <- readSessionBucket bucket
4141+ case authenticateSession sessions token t3 of
4142+ Unauthorized {} -> yield []
4143+ Authorized {pubkey} -> atomic do
4144+ se <- *senderExpirations
4145+ let forKey = do e <- se; where e.senderKey == pubkey; yield {messageId: e.messageId, toKey: e.toKey, expiredAt: e.expiredAt}
4146+ let remaining = do e <- se; where e.senderKey != pubkey; yield e
4147+ set *senderExpirations = remaining
4148+ let capped = takeRelation maxSenderExpirationsPerKey forKey
4046− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0, cursorReset: cursorWasReset})
4150+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0, cursorReset: cursorWasReset, messageExpirations: expirations})
4048-- Core send logic: dedup, rate limiting, store message, create forwards.
4152-- Core send logic: dedup, rate limiting, store message, create forwards.
4049-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
4153-- Called inside an atomic block by both handleSendMessage and sendBatchInner.
4050-- Caller handles auth, drain checks, and messageId generation.
4154-- Caller handles auth, drain checks, and messageId generation.
4051-- messageTtl: sender-specified TTL in ms (0 = default unackedMessageMaxAge).
4155-- messageTtl: sender-specified TTL in ms (0 = default unackedMessageMaxAge).
4052sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTtl t ->
4156sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTtl t ->
4053− if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4157+ if senderKey == to then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
4055 let msgBucket = pubkeyBucket to
4159 let msgBucket = pubkeyBucket to
4056 bucket <- readMessageBucket msgBucket
4160 bucket <- readMessageBucket msgBucket
@@ -4061,31 +4165,31 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTt
4061 -- expire but before the nonce expires.
4165 -- expire but before the nonce expires.
4062 nonceBkt <- readNonceBucket msgBucket
4166 nonceBkt <- readNonceBucket msgBucket
4063 let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt || any nonceBkt (\n -> n.messageId == effectiveMsgId && n.toKey == to)
4167 let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt || any nonceBkt (\n -> n.messageId == effectiveMsgId && n.toKey == to)
4064− if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}
4168+ if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
4066 sendRates <- *sendRates
4170 sendRates <- *sendRates
4067 sendRate <- effectiveSendRate
4171 sendRate <- effectiveSendRate
4068 let rateResult = checkPubkeyRate sendRates senderKey sendRate sendRateWindowMs maxSendRateEntries t
4172 let rateResult = checkPubkeyRate sendRates senderKey sendRate sendRateWindowMs maxSendRateEntries t
4069− if rateResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: rateResult.retryAfterMs}
4173+ if rateResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: rateResult.retryAfterMs, msgExpiresAt: 0}
4071 set *sendRates = rateResult.entries
4175 set *sendRates = rateResult.entries
4072 pairRates <- *pairRates
4176 pairRates <- *pairRates
4073 let pairResult = checkPairRate pairRates senderKey to maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
4177 let pairResult = checkPairRate pairRates senderKey to maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
4074 set *pairRates = pairResult.entries
4178 set *pairRates = pairResult.entries
4075− if pairResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: pairResult.retryAfterMs}
4179+ if pairResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: pairResult.retryAfterMs, msgExpiresAt: 0}
4077 recvRates <- *recvRates
4181 recvRates <- *recvRates
4078 recvRate <- effectiveRecvRate
4182 recvRate <- effectiveRecvRate
4079 let recvResult = checkPubkeyRate recvRates to recvRate recvRateWindowMs maxRecvRateEntries t
4183 let recvResult = checkPubkeyRate recvRates to recvRate recvRateWindowMs maxRecvRateEntries t
4080− if recvResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: recvResult.retryAfterMs}
4184+ if recvResult.limited then yield {status: MsRateLimited {}, remoteServers: [], retryAfterMs: recvResult.retryAfterMs, msgExpiresAt: 0}
4082 set *recvRates = recvResult.entries
4186 set *recvRates = recvResult.entries
4083 totalMsgCount <- *totalMessageCount
4187 totalMsgCount <- *totalMessageCount
4084− if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4188+ if totalMsgCount >= maxMessages then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
4086 recipientCounts <- readRcBucket msgBucket
4190 recipientCounts <- readRcBucket msgBucket
4087 let recipientCount = getRecipientCount to recipientCounts
4191 let recipientCount = getRecipientCount to recipientCounts
4088− if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0}
4192+ if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
4090 recipientPresence <- readPresenceBucket (pubkeyBucket to)
4194 recipientPresence <- readPresenceBucket (pubkeyBucket to)
4091 seqCounter <- *seqCounter
4195 seqCounter <- *seqCounter
@@ -4094,7 +4198,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTt
4094 let msgExpiresAt = computeMessageExpiresAt seqTs messageTtl
4198 let msgExpiresAt = computeMessageExpiresAt seqTs messageTtl
4095 writeMessageBucket msgBucket (union bucket
4199 writeMessageBucket msgBucket (union bucket
4096 [{messageId: effectiveMsgId, toKey: to,
4200 [{messageId: effectiveMsgId, toKey: to,
4097− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt}])
4201+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs, expiresAt: msgExpiresAt, senderKey: senderKey}])
4098 set *totalMessageCount = totalMsgCount + 1
4202 set *totalMessageCount = totalMsgCount + 1
4099 writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)
4203 writeRcBucket msgBucket (adjustRecipientCount recipientCounts to 1)
4100 signalMessageForKey to seqTs
4204 signalMessageForKey to seqTs
@@ -4126,10 +4230,12 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId messageTt
4126 let status = if localOnline then MsDelivered {}
4230 let status = if localOnline then MsDelivered {}
4127 else if federationEnabled && forwardsEnqueued then MsFederated {}
4231 else if federationEnabled && forwardsEnqueued then MsFederated {}
4128 else MsQueued {}
4232 else MsQueued {}
4129− yield {status: status, remoteServers: remoteServers, retryAfterMs: 0}
4233+ yield {status: status, remoteServers: remoteServers, retryAfterMs: 0, msgExpiresAt: msgExpiresAt}
4131−-- Fork federation forwards for messages that need to be delivered to remote servers
4235+-- Fork federation forwards for messages that need to be delivered to remote servers.
4236+-- msgExpiresAt: the stored message's expiration timestamp (used to compute remaining TTL
4237+-- at forward time, ensuring the remote server gets the correct remaining lifetime).
4132−forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs messageTtl remoteServers ->
4238+forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs msgExpiresAt remoteServers ->
4133 when (count remoteServers > 0 && federationEnabled) (
4239 when (count remoteServers > 0 && federationEnabled) (
4134 fork (sendToEach remoteServers (\server -> do
4240 fork (sendToEach remoteServers (\server -> do
4135 stillPending <- atomic do
4241 stillPending <- atomic do
@@ -4139,8 +4245,10 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs messageTtl remoteS
4139 fwdT <- now
4245 fwdT <- now
4140 fwdAuth <- generateFederationAuth "forward"
4246 fwdAuth <- generateFederationAuth "forward"
4141 fwdCorrId <- genTokenN 1
4247 fwdCorrId <- genTokenN 1
4142− -- Compute remaining TTL at forward time (may have decreased since enqueue)
4248+ -- Compute remaining TTL from the stored message's expiration (not the original TTL).
4249+ -- This ensures the remote server gets the correct remaining lifetime, even if
4250+ -- the fork is delayed by network issues or server load.
4143− let fwdTtl = if messageTtl > 0 then max 0 (computeMessageExpiresAt fwdT messageTtl - fwdT) else 0
4251+ let fwdTtl = if msgExpiresAt > 0 then max 0 (msgExpiresAt - fwdT) else 0
4144 fwdResult <- fetchFederation server (ForwardMessage {
4252 fwdResult <- fetchFederation server (ForwardMessage {
4145 toKey: to, encryptedBlob: encryptedBlob,
4253 toKey: to, encryptedBlob: encryptedBlob,
4146 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
4254 timestamp: clampedTs, originServer: serverHostN, messageId: effectiveMsgId,
@@ -4171,11 +4279,13 @@ sendBatchForwardsSequential = \msgs server ->
4171 fwdT <- now
4279 fwdT <- now
4172 fwdAuth <- generateFederationAuth "forward"
4280 fwdAuth <- generateFederationAuth "forward"
4173 fwdCorrId <- genTokenN 1
4281 fwdCorrId <- genTokenN 1
4282+ -- Use stored msgExpiresAt to compute remaining TTL (not original messageTtl)
4283+ let fwdTtl = if r.msgExpiresAt > 0 then max 0 (r.msgExpiresAt - fwdT) else 0
4174 fwdResult <- fetchFederation server (ForwardMessage {
4284 fwdResult <- fetchFederation server (ForwardMessage {
4175 toKey: r.toKey, encryptedBlob: r.blob,
4285 toKey: r.toKey, encryptedBlob: r.blob,
4176 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
4286 timestamp: r.ts, originServer: serverHostN, messageId: r.messageId,
4177 fromServer: serverHostN, requestTimestamp: fwdT,
4287 fromServer: serverHostN, requestTimestamp: fwdT,
4178− authorization: fwdAuth, correlationId: fwdCorrId, messageTtl: r.messageTtl})
4288+ authorization: fwdAuth, correlationId: fwdCorrId, messageTtl: fwdTtl})
4179 case fwdResult of
4289 case fwdResult of
4180 Err e -> do
4290 Err e -> do
4181 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
4291 logWarn ("federation: forward to " ++ server ++ " failed for " ++ r.messageId ++ ": " ++ show e)
@@ -4207,15 +4317,15 @@ handleSendMessage = \to encryptedBlob timestamp messageId messageTtl authorizati
4207 let clampedTs = clampTimestamp timestamp t
4317 let clampedTs = clampTimestamp timestamp t
4208 result <- atomic do
4318 result <- atomic do
4209 dp <- *drainPhase
4319 dp <- *drainPhase
4210− if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
4320+ if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
4211 else withSessionAuth authorization t clientIp
4321 else withSessionAuth authorization t clientIp
4212− (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0})
4322+ (yield {status: MsUnauthorized {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0})
4213 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId messageTtl t)
4323 (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effectiveMsgId messageTtl t)
4214 case result.status of
4324 case result.status of
4215 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
4325 MsRateLimited {} -> logWarn ("sendMessage: rate limited for " ++ effectiveMsgId)
4216 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))
4326 MsRejected {} -> when (result.retryAfterMs == 0) (logWarn ("sendMessage: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)"))
4217 _ -> yield {}
4327 _ -> yield {}
4218− forkSendForwards effectiveMsgId to encryptedBlob clampedTs messageTtl result.remoteServers
4328+ forkSendForwards effectiveMsgId to encryptedBlob clampedTs result.msgExpiresAt result.remoteServers
4219 yield (case result.status of
4329 yield (case result.status of
4220 MsDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0}
4330 MsDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0}
4221 MsDelivered {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId, retryAfterMs: 0}
4331 MsDelivered {} -> respond {status: MsDelivered {}, messageId: effectiveMsgId, retryAfterMs: 0}
@@ -4251,6 +4361,37 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
4251 yield {server: maskedServer, authorized: True {}})
4361 yield {server: maskedServer, authorized: True {}})
4252 yield (respond {server: result.server, authorized: result.authorized})
4362 yield (respond {server: result.server, authorized: result.authorized})
4364+-- POST /lookup/batch: Look up presence for multiple pubkeys in one request.
4365+-- Reads only the needed presence buckets (not all 16), reducing round trips
4366+-- vs sequential single-key lookups. Uses one rate limit check for the whole batch.
4367+handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
4369+ globalRateLimited <- checkGlobalRate t
4370+ if globalRateLimited then yield (respond {results: [], authorized: False {}})
4373+ if dp >= 2 then yield (respond {results: [], authorized: False {}})
4375+ withSessionAuth authorization t clientIp
4376+ (yield (respond {results: [], authorized: False {}}))
4378+ lookupRateLimited <- atomic do
4379+ lookupRates <- *lookupRates
4380+ let rateResult = checkPubkeyRate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
4381+ set *lookupRates = rateResult.entries
4382+ yield rateResult.limited
4383+ if lookupRateLimited then yield (respond {results: [], authorized: True {}})
4385+ let cappedKeys = takeRelation maxBatchLookupKeys pubkeys
4386+ result <- atomic do
4387+ let neededBuckets = uniqueBucketsFor cappedKeys (\k -> k)
4388+ relevantPresence <- readSelectedPresenceBuckets neededBuckets
4389+ yield (do key <- cappedKeys; yield (
4390+ let foundServer = bestServerForKey (presenceForKey relevantPresence key t)
4391+ in let maskedServer = if foundServer == "" then "" else if isLocalServer foundServer then foundServer else "remote"
4392+ in {pubkey: key, server: maskedServer}))
4393+ yield (respond {results: result, authorized: True {}}))
4254-- POST /messages/status: Check delivery status of a previously sent message.
4395-- POST /messages/status: Check delivery status of a previously sent message.
4255-- Caller must provide both messageId and recipient pubkey (proves they know the recipient).
4396-- Caller must provide both messageId and recipient pubkey (proves they know the recipient).
4256-- Returns: "pending" (in mailbox), "forwarding" (federation in progress),
4397-- Returns: "pending" (in mailbox), "forwarding" (federation in progress),
@@ -4614,7 +4755,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
4614 let fwdExpiresAt = computeMessageExpiresAt seqTs messageTtl
4755 let fwdExpiresAt = computeMessageExpiresAt seqTs messageTtl
4615 writeMessageBucket bkt (union bucket
4756 writeMessageBucket bkt (union bucket
4616 [{messageId: messageId, toKey: toKey,
4757 [{messageId: messageId, toKey: toKey,
4617− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt}])
4758+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: ""}])
4618 set *totalMessageCount = totalMsgCount + 1
4759 set *totalMessageCount = totalMsgCount + 1
4619 writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
4760 writeRcBucket bkt (adjustRecipientCount recipientCounts toKey 1)
4620 signalMessageForKey toKey seqTs
4761 signalMessageForKey toKey seqTs
@@ -4657,11 +4798,9 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
4657 let dedupedIds = dedupMessageIds messageIds
4798 let dedupedIds = dedupMessageIds messageIds
4658 -- Clear forward entries for the acking server. Extract removed forwards' toKey
4799 -- Clear forward entries for the acking server. Extract removed forwards' toKey
4659 -- so we can reclaim origin messages that are no longer needed.
4800 -- so we can reclaim origin messages that are no longer needed.
4660− removedForwards <- atomic do
4801+ -- Optimized: scan all 16 forward buckets but only rewrite those that changed.
4661− forwards <- readAllForwards
4802+ -- Avoids the O(n) writeAllForwards repartitioning when only a few buckets are affected.
4662− let removed = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
4803+ removedForwards <- removeForwardsByServer fromServer dedupedIds 0 []
4663− writeAllForwards (forwardsExcluding forwards fromServer dedupedIds)
4665 let ackedCount = count removedForwards
4804 let ackedCount = count removedForwards
4666 -- Guard: warn if peer acks messages we had no forward entries for. This can indicate
4805 -- Guard: warn if peer acks messages we had no forward entries for. This can indicate
4667 -- a peer acking messages it never received (e.g., the forward response was lost but
4806 -- a peer acking messages it never received (e.g., the forward response was lost but
@@ -5010,13 +5149,10 @@ handleStatsPeers = \authorization respond -> do
5010 if authorized
5149 if authorized
5012 failures <- *forwardFailures
5151 failures <- *forwardFailures
5013− forwards <- readAllForwards
5152+ -- Per-bucket aggregation: compute forward counts without materializing all 16 buckets.
5014− -- Single-pass aggregation: accumulate {server, count} pairs in one fold over forwards.
5153+ -- Reuses computeForwardStats which reads one bucket at a time.
5015− -- O(n*k) where k = unique servers (vs previous O(n*k + k*n) two-pass approach).
5154+ fwdStats <- computeForwardStats 0 t
5016− let fwdCounts = fold (\acc f ->
5155+ let fwdCounts = fwdStats.serverCounts
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
5020 let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
5156 let fwdCountFor = \server -> fold (\acc c -> if c.server == server then c.count else acc) 0 fwdCounts
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}
5157 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}
5022 -- Include servers with pending forwards but no failure records
5158 -- Include servers with pending forwards but no failure records
@@ -5337,9 +5473,9 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
5337 yield (respond {ok: True {}, authorized: True {}, deletedCount: deleteResult.deletedCount}))
5473 yield (respond {ok: True {}, authorized: True {}, deletedCount: deleteResult.deletedCount}))
5339-- POST /messages/cancel: Retract a previously sent message before the recipient reads it.
5475-- 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
5476+-- Only works for locally-originated messages where the caller's pubkey matches the
5341−-- the sender (either client-generated or returned only in the send response).
5477+-- stored senderKey. Federation forwards cannot be cancelled via this endpoint.
5342−-- Removes from recipient mailbox, cancels pending forwards, sends federation acks.
5478+-- Removes from recipient mailbox and cancels pending forwards.
5343handleCancelMessage = \messageId to authorization clientIp respond -> do
5479handleCancelMessage = \messageId to authorization clientIp respond -> do
5344 t <- now
5480 t <- now
5345 dp <- *drainPhase
5481 dp <- *drainPhase
@@ -5365,7 +5501,13 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
5365 let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == to; yield m)
5501 let toCancel = headOf (do m <- bucket; where m.messageId == messageId && m.toKey == to; yield m)
5366 case toCancel of
5502 case toCancel of
5367 Nothing {} -> yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5503 Nothing {} -> yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5368− Just {value: msg} -> do
5504+ Just {value: msg} ->
5505+ -- Only allow cancelling locally-originated messages (originServer == "")
5506+ -- where the caller is the actual sender. Federation forwards have no
5507+ -- senderKey and should not be cancellable via this endpoint.
5508+ if msg.originServer != "" then yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5509+ else if msg.senderKey != "" && msg.senderKey != senderKey then yield {cancelled: False {}, pendingAcks: [], evictedEntries: []}
5369 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)
5511 writeMessageBucket b (do m <- bucket; where not (m.messageId == messageId && m.toKey == to); yield m)
5370 tc <- *totalMessageCount
5512 tc <- *totalMessageCount
5371 set *totalMessageCount = max 0 (tc - 1)
5513 set *totalMessageCount = max 0 (tc - 1)
@@ -5374,16 +5516,9 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
5374 -- Cancel pending forwards
5516 -- Cancel pending forwards
5375 fwdBkt <- readFwdBucket b
5517 fwdBkt <- readFwdBucket b
5376 writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toKey == to); yield f)
5518 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
5519+ -- Only locally-originated messages reach this point (originServer == ""),
5378− if msg.originServer != "" then do
5520+ -- so no federation acks are needed.
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: []}
5521+ yield {cancelled: True {}, pendingAcks: [], evictedEntries: []}
5387 fork (when (federationEnabled && count cancelResult.pendingAcks > 0) (do
5522 fork (when (federationEnabled && count cancelResult.pendingAcks > 0) (do
5388 sendAcksAndClean cancelResult.pendingAcks
5523 sendAcksAndClean cancelResult.pendingAcks
5389 retryEvictedAcks cancelResult.evictedEntries cancelResult.pendingAcks))
5524 retryEvictedAcks cancelResult.evictedEntries cancelResult.pendingAcks))
@@ -5441,18 +5576,18 @@ sendBatchInner = \senderKey messages t acc ->
5441 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
5576 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
5442 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
5577 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
5443 result <- if t - msg.timestamp > messageMaxAge
5578 result <- if t - msg.timestamp > messageMaxAge
5444− then yield {messageId: effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, messageTtl: 0, retryAfterMs: 0}
5579+ then yield {messageId: effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, msgExpiresAt: 0, retryAfterMs: 0}
5446 let clampedTs = clampTimestamp msg.timestamp t
5581 let clampedTs = clampTimestamp msg.timestamp t
5447 sendResult <- atomic do
5582 sendResult <- atomic do
5448 dp <- *drainPhase
5583 dp <- *drainPhase
5449− if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0}
5584+ if dp >= 2 then yield {status: MsDraining {}, remoteServers: [], retryAfterMs: 0, msgExpiresAt: 0}
5450 else sendMessageCore senderKey msg.to msg.encryptedBlob clampedTs effectiveMsgId msg.messageTtl t
5585 else sendMessageCore senderKey msg.to msg.encryptedBlob clampedTs effectiveMsgId msg.messageTtl t
5451 case sendResult.status of
5586 case sendResult.status of
5452 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
5587 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
5453 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")
5588 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")
5454 _ -> yield {}
5589 _ -> yield {}
5455− yield {messageId: effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, messageTtl: msg.messageTtl, retryAfterMs: sendResult.retryAfterMs}
5590+ yield {messageId: effectiveMsgId, status: sendResult.status, remoteServers: sendResult.remoteServers, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, msgExpiresAt: sendResult.msgExpiresAt, retryAfterMs: sendResult.retryAfterMs}
5456 sendBatchInner senderKey split.rest t (union acc [result])
5591 sendBatchInner senderKey split.rest t (union acc [result])
5458-- POST /maintenance/block: Add a server to the federation blocklist.
5593-- POST /maintenance/block: Add a server to the federation blocklist.
@@ -5497,10 +5632,55 @@ handleListBlocked = \authorization respond -> do
5497 blocked <- *blockedPeers
5632 blocked <- *blockedPeers
5498 yield (respond {peers: blocked, ok: True {}})
5633 yield (respond {peers: blocked, ok: True {}})
5635+-- POST /maintenance/allow: Add a server to the federation allowlist.
5636+-- When the allowlist is non-empty, only servers on the list can federate with us.
5637+-- Adding the first server transitions from open federation to allowlist-only mode.
5638+handleAllowPeer = \server authorization respond -> do
5639+ let authorized = verifyMaintenanceAuth authorization
5640+ if not authorized then yield (respond {ok: False {}, allowed: 0})
5642+ result <- atomic do
5643+ allowed <- *allowedPeers
5644+ if any allowed (\a -> a == server) then yield {alreadyAllowed: True {}, total: count allowed}
5646+ let updated = if count allowed >= maxAllowedPeers
5647+ then case headOf allowed of
5648+ Nothing {} -> [server]
5649+ Just {value: oldest} -> union (do a <- allowed; where a != oldest; yield a) [server]
5650+ else union allowed [server]
5651+ set *allowedPeers = updated
5652+ yield {alreadyAllowed: False {}, total: count updated}
5653+ when (not result.alreadyAllowed) (logInfo ("allow: added " ++ server ++ " to allowlist (" ++ show result.total ++ " total)"))
5654+ yield (respond {ok: True {}, allowed: result.total})
5656+-- POST /maintenance/unallow: Remove a server from the federation allowlist.
5657+-- Removing the last server transitions back to open federation mode.
5658+handleUnallowPeer = \server authorization respond -> do
5659+ let authorized = verifyMaintenanceAuth authorization
5660+ if not authorized then yield (respond {ok: False {}, removed: False {}})
5662+ removed <- atomic do
5663+ allowed <- *allowedPeers
5664+ let had = any allowed (\a -> a == server)
5665+ set *allowedPeers = do a <- allowed; where a != server; yield a
5667+ when removed (logInfo ("unallow: removed " ++ server ++ " from allowlist"))
5668+ yield (respond {ok: True {}, removed: removed})
5670+-- GET /maintenance/allowed: List currently allowed federation peers.
5671+-- Empty list = open federation (all peers allowed).
5672+handleListAllowed = \authorization respond -> do
5673+ let authorized = verifyMaintenanceAuth authorization
5674+ if not authorized then yield (respond {peers: [], ok: False {}})
5676+ allowed <- *allowedPeers
5677+ yield (respond {peers: allowed, ok: True {}})
5500-- POST /maintenance/ping: Test federation connectivity to a specific peer.
5679-- POST /maintenance/ping: Test federation connectivity to a specific peer.
5501−-- Sends an empty gossip request via fetchFederation, triggering a full verify callback
5680+-- Sends an empty gossip (0 events, hopsRemaining=1) to the peer via the full
5502−-- round-trip (peer calls back to our /federation/verify to authenticate the request).
5681+-- federation auth path. The peer calls back to our /federation/verify to authenticate,
5503−-- This exercises the real federation auth path and tracks in-flight/latency metrics.
5682+-- then processes the empty gossip (no mutations, no re-broadcast since hops=1).
5683+-- This exercises the real bidirectional connectivity and auth flow.
5504handlePingPeer = \server authorization respond -> do
5684handlePingPeer = \server authorization respond -> do
5505 let authorized = verifyMaintenanceAuth authorization
5685 let authorized = verifyMaintenanceAuth authorization
5506 if not authorized then yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"})
5686 if not authorized then yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"})
@@ -5612,6 +5792,21 @@ handleAdminPurgeMessages = \pubkey authorization respond -> do
5612 retryEvictedAcks purgeResult.evictedEntries purgeResult.pendingAcks))
5792 retryEvictedAcks purgeResult.evictedEntries purgeResult.pendingAcks))
5613 yield (respond {ok: True {}, purgedCount: purgeResult.purgedCount})
5793 yield (respond {ok: True {}, purgedCount: purgeResult.purgedCount})
5795+-- POST /maintenance/nonces/flush: Clear all forward nonces (recovery from corruption).
5796+-- Nonces provide extended replay protection beyond message/ack dedup lifetime.
5797+-- Flushing temporarily reopens a replay window for messages older than 30 days but
5798+-- younger than 60 days. Use only for recovery, not routine maintenance.
5799+handleFlushNonces = \authorization respond -> do
5800+ let authorized = verifyMaintenanceAuth authorization
5801+ if not authorized then yield (respond {ok: False {}, flushedCount: 0})
5803+ flushedCount <- atomic do
5804+ total <- countAllNonces 0
5807+ logInfo ("nonces/flush: flushed " ++ show flushedCount ++ " forward nonces")
5808+ yield (respond {ok: True {}, flushedCount: flushedCount})
5615-- GET /maintenance/rate-limits: Return current effective rate limits.
5810-- GET /maintenance/rate-limits: Return current effective rate limits.
5616handleGetRateLimits = \authorization respond -> do
5811handleGetRateLimits = \authorization respond -> do
5617 let authorized = verifyMaintenanceAuth authorization
5812 let authorized = verifyMaintenanceAuth authorization
@@ -5658,6 +5853,7 @@ requestPath = \req -> case req of
5658 DeleteMessages {} -> "POST /messages/delete"
5853 DeleteMessages {} -> "POST /messages/delete"
5659 CancelMessage {} -> "POST /messages/cancel"
5854 CancelMessage {} -> "POST /messages/cancel"
5660 LookupPresence {} -> "GET /lookup"
5855 LookupPresence {} -> "GET /lookup"
5856+ BatchLookupPresence {} -> "POST /lookup/batch"
5661 RecvGossip {} -> "POST /federation/gossip"
5857 RecvGossip {} -> "POST /federation/gossip"
5662 ForwardMessage {} -> "POST /federation/forward"
5858 ForwardMessage {} -> "POST /federation/forward"
5663 FederationAck {} -> "POST /federation/ack"
5859 FederationAck {} -> "POST /federation/ack"
@@ -5677,6 +5873,10 @@ requestPath = \req -> case req of
5677 AdminPurgeMessages {} -> "POST /maintenance/messages/purge"
5873 AdminPurgeMessages {} -> "POST /maintenance/messages/purge"
5678 GetRateLimits {} -> "GET /maintenance/rate-limits"
5874 GetRateLimits {} -> "GET /maintenance/rate-limits"
5679 SetRateLimits {} -> "POST /maintenance/rate-limits"
5875 SetRateLimits {} -> "POST /maintenance/rate-limits"
5876+ FlushNonces {} -> "POST /maintenance/nonces/flush"
5877+ AllowPeer {} -> "POST /maintenance/allow"
5878+ UnallowPeer {} -> "POST /maintenance/unallow"
5879+ ListAllowed {} -> "GET /maintenance/allowed"
5680 Health {} -> "GET /health"
5880 Health {} -> "GET /health"
5681 Stats {} -> "GET /stats"
5881 Stats {} -> "GET /stats"
5682 StatsPeers {} -> "GET /stats/peers"
5882 StatsPeers {} -> "GET /stats/peers"
@@ -5699,6 +5899,7 @@ requestClientIp = \req -> case req of
5699 DeleteMessages {clientIp} -> clientIp
5899 DeleteMessages {clientIp} -> clientIp
5700 CancelMessage {clientIp} -> clientIp
5900 CancelMessage {clientIp} -> clientIp
5701 LookupPresence {clientIp} -> clientIp
5901 LookupPresence {clientIp} -> clientIp
5902+ BatchLookupPresence {clientIp} -> clientIp
5702 VerifyToken {clientIp} -> clientIp
5903 VerifyToken {clientIp} -> clientIp
5703 RecvGossip {} -> ""
5904 RecvGossip {} -> ""
5704 ForwardMessage {} -> ""
5905 ForwardMessage {} -> ""
@@ -5718,6 +5919,10 @@ requestClientIp = \req -> case req of
5718 AdminPurgeMessages {} -> ""
5919 AdminPurgeMessages {} -> ""
5719 GetRateLimits {} -> ""
5920 GetRateLimits {} -> ""
5720 SetRateLimits {} -> ""
5921 SetRateLimits {} -> ""
5922+ FlushNonces {} -> ""
5924+ UnallowPeer {} -> ""
5925+ ListAllowed {} -> ""
5721 Health {} -> ""
5926 Health {} -> ""
5722 Stats {} -> ""
5927 Stats {} -> ""
5723 StatsPeers {} -> ""
5928 StatsPeers {} -> ""
@@ -5743,6 +5948,7 @@ serveInner = \req -> case req of
5743 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
5948 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
5744 CancelMessage {messageId, to, authorization, clientIp, respond} -> handleCancelMessage messageId to authorization clientIp respond
5949 CancelMessage {messageId, to, authorization, clientIp, respond} -> handleCancelMessage messageId to authorization clientIp respond
5745 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
5950 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
5951+ BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
5746 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
5952 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleRecvGossip events hopsRemaining 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
5953 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, messageTtl, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId messageTtl authorization respond
5748 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
5954 FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
@@ -5762,6 +5968,10 @@ serveInner = \req -> case req of
5762 AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond
5968 AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond
5763 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
5969 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
5764 SetRateLimits {globalRequest, ipRate, sendRate, federationRate, authGlobal, recvRate, authorization, respond} -> handleSetRateLimits globalRequest ipRate sendRate federationRate authGlobal recvRate authorization respond
5970 SetRateLimits {globalRequest, ipRate, sendRate, federationRate, authGlobal, recvRate, authorization, respond} -> handleSetRateLimits globalRequest ipRate sendRate federationRate authGlobal recvRate authorization respond
5971+ FlushNonces {authorization, respond} -> handleFlushNonces authorization respond
5972+ AllowPeer {server, authorization, respond} -> handleAllowPeer server authorization respond
5973+ UnallowPeer {server, authorization, respond} -> handleUnallowPeer server authorization respond
5974+ ListAllowed {authorization, respond} -> handleListAllowed authorization respond
5765 Health {authorization, respond} -> handleHealth authorization respond
5975 Health {authorization, respond} -> handleHealth authorization respond
5766 Stats {authorization, respond} -> handleStats authorization respond
5976 Stats {authorization, respond} -> handleStats authorization respond
5767 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
5977 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
@@ -6040,7 +6250,7 @@ resetFederationLatencyMetrics = atomic do
6040 set *federationLatencyResetAt = t
6250 set *federationLatencyResetAt = t
6042main = do
6252main = do
6043− let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
6253+ let configOk = listenPort > 0 && maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
6044 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
6254 && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
6045 && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0
6255 && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0
6046 && maxSessionLifetime > 0
6256 && maxSessionLifetime > 0
@@ -6094,7 +6304,7 @@ main = do
6094 logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
6304 logInfo "IP binding is active — sessions created with a ClientIp header will reject requests without one"
6095 when federationEnabled
6305 when federationEnabled
6096 (logWarn "SECURITY: DNS rebinding risk — federation URLs resolve DNS at fetch time, not at validation time. A peer hostname passing isBadServerName could rebind to a private IP between validation and fetch. Mitigate by blocking outbound connections to RFC1918/link-local/CGNAT ranges at the firewall level.")
6306 (logWarn "SECURITY: DNS rebinding risk — federation URLs resolve DNS at fetch time, not at validation time. A peer hostname passing isBadServerName could rebind to a private IP between validation and fetch. Mitigate by blocking outbound connections to RFC1918/link-local/CGNAT ranges at the firewall level.")
6097− logInfo "Skrepka relay server starting on port 8080..."
6307+ logInfo ("Skrepka relay server starting on port " ++ show listenPort ++ "...")
6098 logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")
6308 logInfo (" federation: " ++ if federationEnabled then "enabled" else "disabled")
6099 logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
6309 logInfo (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
6100 logInfo (" poll timeout: " ++ show pollTimeoutMs ++ "ms")
6310 logInfo (" poll timeout: " ++ show pollTimeoutMs ++ "ms")
@@ -6113,6 +6323,7 @@ main = do
6113 logInfo (" ack retry limit: " ++ show maxAckRetries ++ " per entry")
6323 logInfo (" ack retry limit: " ++ show maxAckRetries ++ " per entry")
6114 when (federationCircuitBreakerThreshold > 0)
6324 when (federationCircuitBreakerThreshold > 0)
6115 (logInfo (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
6325 (logInfo (" circuit breaker: " ++ show federationCircuitBreakerThreshold ++ " failing peers"))
6326+ logInfo (" federation allowlist: " ++ show maxAllowedPeers ++ " max (empty = open federation)")
6116 when (autoSnapshotInterval > 0)
6327 when (autoSnapshotInterval > 0)
6117 (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
6328 (logInfo (" auto-snapshot: " ++ show autoSnapshotInterval ++ "ms"))
6118 when (autoSnapshotExportInterval > 0)
6329 when (autoSnapshotExportInterval > 0)
@@ -6129,4 +6340,4 @@ main = do
6129 logInfo "startup: probing federation peers..."
6340 logInfo "startup: probing federation peers..."
6130 startupProbePeers
6341 startupProbePeers
6131 resetFederationLatencyMetrics))
6342 resetFederationLatencyMetrics))
6132− listen 8080 serve
6343+ listen listenPort serve