@@ -139,7 +139,7 @@ federationLatencyWindowMs : Int<Ms>
139139federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
140140
141141-- Snapshot format version (increment on schema changes to prevent corrupt restores)
142−snapshotVersion = 1
142+snapshotVersion = 2
143143
144144-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
145145maxGossipEventsPerServer = 500 -- max total gossip events per server per window
@@ -315,6 +315,9 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
315315type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
316316type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
317317
318+-- Forward nonce for extended replay protection (persisted in snapshots)
319+type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}
320+
318321-- Batch send types
319322type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId}
320323type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
@@ -516,7 +519,7 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
516519*pendingVerifications : [{token: Text, createdAt: Timestamp, scope: Text}]
517520
518521-- Forward nonce dedup: (messageId, toKey, originServer, seenAt) for replay protection beyond message lifetime
519−*forwardNonces : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}]
522+*forwardNonces : [ForwardNonce]
520523
521524-- Federation verify per-IP rate limiting (tighter than general IP rate)
522525*verifyIpRates : [IpRateEntry]
@@ -549,7 +552,7 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
549552route AuthApi where
550553 /auth
551554 POST {pubkey: PubkeyHex} /challenge headers {clientIp: Text}
552− -> {challenge: Text, ok: Bool, expiresAt: Timestamp}
555+ -> {challenge: Text, ok: Bool, expiresAt: Timestamp, retryAfterMs: Int}
553556 = ReqChallenge
554557 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /verify headers {clientIp: Text}
555558 -> {token: Text, expiresAt: Timestamp}
@@ -566,12 +569,12 @@ route AuthApi where
566569route ClientApi where
567570 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
568571 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
569− -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool}
572+ -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, presenceUpdates: [{pubkey: PubkeyHex, online: Bool}], heartbeat: Bool, retryAfterMs: Int}
570573 = Poll
571574 /messages
572575 -- Send an E2E encrypted message to a recipient pubkey
573576 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text, clientIp: Text}
574− -> {status: MessageStatus, messageId: OptionalMessageId}
577+ -> {status: MessageStatus, messageId: OptionalMessageId, retryAfterMs: Int}
575578 = SendMessage
576579 -- Check delivery status of a sent message (caller must know recipient to prove ownership)
577580 POST {to: PubkeyHex, messageId: MessageIdHex} /status headers {authorization: Text, clientIp: Text}
@@ -583,8 +586,12 @@ route ClientApi where
583586 = AckMessages
584587 -- Send multiple messages in one request (auth checked once; each message rate-limited independently)
585588 POST {messages: [BatchMessage]} /batch headers {authorization: Text, clientIp: Text}
586− -> {results: [BatchResult], authorized: Bool}
589+ -> {results: [BatchResult], authorized: Bool, retryAfterMs: Int}
587590 = SendBatch
591+ -- List pending messages without acking (useful after cursor loss or device migration)
592+ GET /messages/list headers {authorization: Text, clientIp: Text}
593+ -> {messages: [{messageId: MessageIdHex, timestamp: Timestamp, receivedAt: Timestamp}], authorized: Bool, count: Int}
594+ = ListMessages
588595 -- Look up which server a pubkey is present on (for federation routing)
589596 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
590597 -> {server: AnyServerName, authorized: Bool}
@@ -603,7 +610,7 @@ route FederationApi where
603610 = ForwardMessage
604611 -- Acknowledge delivery of messages back to the origin server
605612 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp} /ack headers {authorization: Text}
606− -> {ok: Bool, error: Text}
613+ -> {ok: Bool, error: Text, ackedCount: Int}
607614 = FederationAck
608615
609616-- Federation callback verification: the receiving server calls back to the sender
@@ -626,16 +633,16 @@ route MaintenanceApi where
626633 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
627634 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
628635 GET /maintenance/snapshot headers {authorization: Text}
629− -> {ok: Bool, version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry]}
636+ -> {ok: Bool, version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce]}
630637 = Snapshot
631638 -- Restore: import a previously-exported snapshot. Server must be in drain mode
632639 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
633− POST {version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry]} /maintenance/restore headers {authorization: Text}
640+ POST {version: Int, presence: [PresenceEntry], ackedIds: [AckedEntry], ackOverflow: [AckedEntry], deliveredIds: [DeliveredEntry], forwardFailures: [ForwardFailure], forwards: [ForwardEntry], forwardNonces: [ForwardNonce]} /maintenance/restore headers {authorization: Text}
634641 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
635642 = Restore
636643
637644route HealthApi where
638− GET /health -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool} = Health
645+ GET /health headers {authorization: Text} -> {ok: Bool, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, federationHealthy: Bool} = Health
639646
640647-- Returns counts of in-memory collections (requires maintenance secret)
641648route StatsApi where
@@ -897,6 +904,14 @@ authenticateSession = \sessions token t ->
897904 Nothing {} -> Unauthorized {}
898905 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}
899906
907+-- Fast re-authentication for poll Phase 3: pre-filters by known pubkey before
908+-- running safeCompare. Reduces safeCompare calls from O(bucket_size) to
909+-- O(sessions_for_key_in_bucket) (~1-2 vs ~600). Safe in this context because
910+-- the pubkey was already verified in Phase 1.
911+reauthenticateSession : [Session] -> Text -> PubkeyHex -> Timestamp -> Bool
912+reauthenticateSession = \sessions token pubkey t ->
913+ any sessions (\s -> s.pubkey == pubkey && s.expiresAt > t && t - s.createdAt < maxSessionLifetime && safeCompare s.token token)
914+
900915-- Extract bearer token from "Authorization: Bearer <token>" header.
901916-- Rejects empty tokens early to avoid unnecessary session bucket scans.
902917extractToken = \authHeader ->
@@ -1395,11 +1410,18 @@ getRecipientCount : PubkeyHex -> [RecipientCount] -> Int
13951410getRecipientCount = \pubkey counts ->
13961411 fold (\acc c -> if c.pubkey == pubkey then c.count else acc) 0 counts
13971412
1413+-- Single-pass recipient count adjustment: fold once instead of any + list comprehension (O(n) vs O(2n))
13981414adjustRecipientCount : [RecipientCount] -> PubkeyHex -> Int -> [RecipientCount]
13991415adjustRecipientCount = \counts pubkey delta ->
14001416 if delta == 0 then counts
1401− else if any counts (\c -> c.pubkey == pubkey)
1402− then do c <- counts; where not (c.pubkey == pubkey && c.count + delta <= 0); yield (if c.pubkey == pubkey then {pubkey: c.pubkey, count: c.count + delta} else c)
1417+ else let result = fold (\acc c ->
1418+ if c.pubkey == pubkey
1419+ then if c.count + delta > 0
1420+ then {items: union acc.items [{pubkey: c.pubkey, count: c.count + delta}], found: True {}}
1421+ else {items: acc.items, found: True {}}
1422+ else {items: union acc.items [c], found: acc.found})
1423+ {items: [], found: False {}} counts
1424+ in if result.found then result.items
14031425 else if delta > 0 then union counts [{pubkey: pubkey, count: delta}]
14041426 else counts
14051427
@@ -1537,19 +1559,21 @@ matchIp = \clientIp sessionIp ->
15371559 else if clientIp == "" then False {} -- session has IP but request doesn't — proxy disappeared
15381560 else safeCompareLoop clientIp sessionIp maxServerNameLen 0 0 == 0
15391561
1540−-- Generic sliding-window rate limiter. Returns whether the request is rate-limited
1541−-- and the updated entries list. Callers provide match/create/update functions for their entry type.
1562+-- Generic sliding-window rate limiter. Returns whether the request is rate-limited,
1563+-- the updated entries list, and retryAfterMs (time until the current window resets, 0 if not limited).
1564+-- Callers provide match/create/update functions for their entry type.
15421565checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntries t ->
15431566 let entry = headOf (do r <- entries; where matchKey r; yield r) in
15441567 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= windowMs in
15451568 let rateCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count in
15461569 if rateCount >= maxRate
1547− then {limited: True {}, entries: entries}
1570+ then let retryMs = case entry of Nothing {} -> windowMs; Just {value: r} -> max 0 (windowMs - (t - r.windowStart))
1571+ in {limited: True {}, entries: entries, retryAfterMs: retryMs}
15481572 else
15491573 let newEntries = if isNewWindow
15501574 then union (do r <- entries; where not (matchKey r); yield r) [makeNew t]
15511575 else do r <- entries; yield (if matchKey r then updateCount r else r)
1552− in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs; yield r) else newEntries}
1576+ in {limited: False {}, entries: if count newEntries > maxEntries then (do r <- newEntries; where t - r.windowStart < windowMs; yield r) else newEntries, retryAfterMs: 0}
15531577
15541578-- Extract the item with the smallest key and return both the item and remaining list
15551579extractOldest = \items getKey ->
@@ -2595,15 +2619,21 @@ pollOnce = \token cursor clientIp -> do
25952619 if wakeup > s.startWakeup || signal > s.startSignal
25962620 then yield {}
25972621 else retry
2598− -- Phase 3: re-authenticate and read messages after waking
2622+ -- Phase 3: re-authenticate and read messages after waking.
2623+ -- Check drain flag (server may have started draining during our wait).
2624+ -- Uses reauthenticateSession (pre-filters by pubkey) instead of full
2625+ -- authenticateSession to skip safeCompare on unrelated sessions.
25992626 t2 <- now
26002627 result <- atomic do
2601− sessions2 <- readSessionBucket bucket
2602− case authenticateSession sessions2 token t2 of
2603− Unauthorized {} -> yield {authorized: False {}, events: [], cursor: s.clampedCursor}
2604− Authorized {} -> do
2605− r <- pollNewEvents s.pubkey s.clampedCursor t2
2606− yield {authorized: True {}, events: r.events, cursor: r.cursor}
2628+ isDraining <- *draining
2629+ if isDraining then yield {authorized: True {}, events: [], cursor: s.clampedCursor}
2630+ else do
2631+ sessions2 <- readSessionBucket bucket
2632+ if not (reauthenticateSession sessions2 token s.pubkey t2)
2633+ then yield {authorized: False {}, events: [], cursor: s.clampedCursor}
2634+ else do
2635+ r <- pollNewEvents s.pubkey s.clampedCursor t2
2636+ yield {authorized: True {}, events: r.events, cursor: r.cursor}
26072637 yield {authorized: result.authorized, events: result.events, cursor: result.cursor}
26082638 -- Release poll slot. Safety: the code between acquire and release only contains
26092639 -- atomic reads and STM retry (no IO that can throw), so this path is always reached.
@@ -2704,8 +2734,15 @@ newlyOnlineRemoteKeys = \events appliedKeys ->
27042734-- Trust gossip entries if they're first-party (server == fromServer) or updating
27052735-- an existing (pubkey, server) presence entry. Third-party gossip cannot create
27062736-- new entries for keys we haven't seen — only the key's home server can do that.
2737+-- Additional constraint: third-party updates must carry a strictly newer timestamp
2738+-- than the existing entry — prevents stale relays from refreshing presence TTLs
2739+-- beyond what the original server intended.
27072740trustedGossipEntries = \deduped presence fromServer t ->
2708− do e <- deduped; where e.server == fromServer || any presence (\p -> samePubkeyServer p e && p.expiresAt > t); yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp}
2741+ do e <- deduped; where
2742+ if e.server == fromServer then True {} -- first-party: always trusted
2743+ else any presence (\p -> samePubkeyServer p e && p.expiresAt > t && e.timestamp > p.timestamp)
2744+ -- third-party: only update existing entries with strictly newer timestamps
2745+ ; yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp}
27092746
27102747-- Keys whose presence was actually updated (new or extended expiry)
27112748appliedOnlineKeys = \entries presence ->
@@ -2741,10 +2778,10 @@ handleChallenge = \pubkey clientIp respond -> do
27412778 if preAuth != ""
27422779 then do
27432780 when (preAuth == "ip_rate_limited") (logWarn ("challenge: IP rate limited for " ++ clientIp))
2744− yield (respond {challenge: "", ok: False {}, expiresAt: 0})
2781+ yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0})
27452782 else do
27462783 authLimited <- checkAuthGlobalRate t
2747− if authLimited then yield (respond {challenge: "", ok: False {}, expiresAt: 0})
2784+ if authLimited then yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0})
27482785 else do
27492786 challenge <- genToken
27502787 result <- atomic do
@@ -2757,7 +2794,7 @@ handleChallenge = \pubkey clientIp respond -> do
27572794 union others (evictOneOldest forKey (\c -> c.expiresAt))
27582795 else active
27592796 if count cappedActive >= maxChallenges
2760− then yield {accepted: False {}, rateLimited: False {}}
2797+ then yield {accepted: False {}, rateLimited: False {}, retryAfterMs: 0}
27612798 else do
27622799 challengeRates <- *challengeRates
27632800 let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
@@ -2765,16 +2802,16 @@ handleChallenge = \pubkey clientIp respond -> do
27652802 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
27662803 maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
27672804 if rateResult.limited
2768− then yield {accepted: False {}, rateLimited: True {}}
2805+ then yield {accepted: False {}, rateLimited: True {}, retryAfterMs: rateResult.retryAfterMs}
27692806 else do
27702807 set *challengeRates = rateResult.entries
27712808 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
2772− yield {accepted: True {}, rateLimited: False {}}
2809+ yield {accepted: True {}, rateLimited: False {}, retryAfterMs: 0}
27732810 when result.rateLimited
27742811 (logWarn ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
27752812 when (not result.accepted && not result.rateLimited)
27762813 (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
2777− yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
2814+ yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0, retryAfterMs: result.retryAfterMs})
27782815
27792816-- POST /auth/verify: Verify Ed25519 signature over challenge string.
27802817-- On success: creates a session, registers local presence, and gossips online status.
@@ -2866,23 +2903,42 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
28662903 let capped = capPerKeyAndGlobal afterRevoke (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
28672904 let sessionsWithNew = union capped [newSession]
28682905 set *totalSessionCount = count sessionsWithNew
2869− let bFor = \b -> do s <- sessionsWithNew; where tokenBucket s.token == b; yield s
2870− writeSessionBucket 0 (bFor 0)
2871− writeSessionBucket 1 (bFor 1)
2872− writeSessionBucket 2 (bFor 2)
2873− writeSessionBucket 3 (bFor 3)
2874− writeSessionBucket 4 (bFor 4)
2875− writeSessionBucket 5 (bFor 5)
2876− writeSessionBucket 6 (bFor 6)
2877− writeSessionBucket 7 (bFor 7)
2878− writeSessionBucket 8 (bFor 8)
2879− writeSessionBucket 9 (bFor 9)
2880− writeSessionBucket 10 (bFor 10)
2881− writeSessionBucket 11 (bFor 11)
2882− writeSessionBucket 12 (bFor 12)
2883− writeSessionBucket 13 (bFor 13)
2884− writeSessionBucket 14 (bFor 14)
2885− writeSessionBucket 15 (bFor 15)
2906+ -- Single-pass partition by token bucket (O(n) instead of 16x O(n) scan)
2907+ let partitioned = fold (\acc s ->
2908+ let b = tokenBucket s.token in
2909+ {b0: if b == 0 then union [s] acc.b0 else acc.b0,
2910+ b1: if b == 1 then union [s] acc.b1 else acc.b1,
2911+ b2: if b == 2 then union [s] acc.b2 else acc.b2,
2912+ b3: if b == 3 then union [s] acc.b3 else acc.b3,
2913+ b4: if b == 4 then union [s] acc.b4 else acc.b4,
2914+ b5: if b == 5 then union [s] acc.b5 else acc.b5,
2915+ b6: if b == 6 then union [s] acc.b6 else acc.b6,
2916+ b7: if b == 7 then union [s] acc.b7 else acc.b7,
2917+ b8: if b == 8 then union [s] acc.b8 else acc.b8,
2918+ b9: if b == 9 then union [s] acc.b9 else acc.b9,
2919+ b10: if b == 10 then union [s] acc.b10 else acc.b10,
2920+ b11: if b == 11 then union [s] acc.b11 else acc.b11,
2921+ b12: if b == 12 then union [s] acc.b12 else acc.b12,
2922+ b13: if b == 13 then union [s] acc.b13 else acc.b13,
2923+ b14: if b == 14 then union [s] acc.b14 else acc.b14,
2924+ b15: if b == 15 then union [s] acc.b15 else acc.b15})
2925+ {b0: [], b1: [], b2: [], b3: [], b4: [], b5: [], b6: [], b7: [], b8: [], b9: [], b10: [], b11: [], b12: [], b13: [], b14: [], b15: []} sessionsWithNew
2926+ writeSessionBucket 0 partitioned.b0
2927+ writeSessionBucket 1 partitioned.b1
2928+ writeSessionBucket 2 partitioned.b2
2929+ writeSessionBucket 3 partitioned.b3
2930+ writeSessionBucket 4 partitioned.b4
2931+ writeSessionBucket 5 partitioned.b5
2932+ writeSessionBucket 6 partitioned.b6
2933+ writeSessionBucket 7 partitioned.b7
2934+ writeSessionBucket 8 partitioned.b8
2935+ writeSessionBucket 9 partitioned.b9
2936+ writeSessionBucket 10 partitioned.b10
2937+ writeSessionBucket 11 partitioned.b11
2938+ writeSessionBucket 12 partitioned.b12
2939+ writeSessionBucket 13 partitioned.b13
2940+ writeSessionBucket 14 partitioned.b14
2941+ writeSessionBucket 15 partitioned.b15
28862942 yield (Just {value: {peerServers: peerServers, revokedCount: revokedCount}})
28872943 case authResult of
28882944 Nothing {} -> yield (respond {token: "", expiresAt: 0})
@@ -2903,10 +2959,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
29032959 let effectiveCursor = if generation == serverGen then cursor else 0
29042960 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
29052961 draining <- *draining
2906− if draining then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}})
2962+ if draining then yield (respond {events: [], cursor: 0, authorized: True {}, rateLimited: False {}, draining: True {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: 0})
29072963 else do
29082964 globalRateLimited <- checkGlobalRate t
2909− if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}})
2965+ if globalRateLimited then yield (respond {events: [], cursor: 0, authorized: False {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: globalRateWindowMs})
29102966 else do
29112967 result <- case extractToken authorization of
29122968 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
@@ -2924,7 +2980,7 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
29242980 sendAcksAndClean result.pendingAcks
29252981 retryEvictedAcks result.evictedEntries result.pendingAcks))
29262982 let isHeartbeat = result.authorized && count result.events == 0
2927− yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat})
2983+ yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0})
29282984
29292985-- POST /messages: Send an encrypted message to a recipient.
29302986-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
@@ -2932,11 +2988,11 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
29322988handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp respond -> do
29332989 t <- now
29342990 globalRateLimited <- checkGlobalRate t
2935− if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: ""})
2991+ if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0})
29362992 else do
29372993 effectiveMsgId <- if messageId != "" then yield messageId else genToken
29382994 if t - timestamp > messageMaxAge
2939− then yield (respond {status: MsRejected {}, messageId: ""})
2995+ then yield (respond {status: MsRejected {}, messageId: "", retryAfterMs: 0})
29402996 else do
29412997 when (timestamp > t + clockSkewTolerance)
29422998 (logWarn ("sendMessage: future timestamp detected (+" ++ show (timestamp - t) ++ "ms ahead), clamping to now"))
@@ -2945,12 +3001,12 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
29453001 let clampedTs = clampTimestamp timestamp t
29463002 result <- atomic do
29473003 draining <- *draining
2948− if draining then yield {send: SendFailed {status: MsDraining {}}, senderRateLimited: False {}, globalCapReached: False {}}
3004+ if draining then yield {send: SendFailed {status: MsDraining {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}
29493005 else withSessionAuth authorization t clientIp
2950− (yield {send: SendFailed {status: MsUnauthorized {}}, senderRateLimited: False {}, globalCapReached: False {}})
3006+ (yield {send: SendFailed {status: MsUnauthorized {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0})
29513007 (\senderKey -> do
29523008 -- Reject self-sends (sender == recipient wastes a mailbox slot)
2953− if senderKey == to then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}}
3009+ if senderKey == to then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}
29543010 else do
29553011 let msgBucket = pubkeyBucket to
29563012 bucket <- readMessageBucket msgBucket
@@ -2959,7 +3015,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
29593015 -- Dedup check (only when messageId is non-empty; scans only the recipient's bucket)
29603016 let isDup = checkMessageDup bucket messageId to || (messageId != "" && (hasMessage messageId to ackedIds || hasMessage messageId to deliveredIds))
29613017 if isDup
2962− then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}}
3018+ then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}
29633019 else do
29643020 sendRates <- *sendRates
29653021 let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
@@ -2967,7 +3023,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
29673023 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
29683024 maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
29693025 if rateResult.limited
2970− then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}, globalCapReached: False {}}
3026+ then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: True {}, globalCapReached: False {}, retryAfterMs: rateResult.retryAfterMs}
29713027 else do
29723028 set *sendRates = rateResult.entries
29733029 pairRates <- *pairRates
@@ -2977,7 +3033,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
29773033 maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
29783034 set *pairRates = pairResult.entries
29793035 if pairResult.limited
2980− then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}}
3036+ then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: pairResult.retryAfterMs}
29813037 else do
29823038 recvRates <- *recvRates
29833039 let recvResult = checkRateWindow recvRates (\r -> r.pubkey == to)
@@ -2985,18 +3041,18 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
29853041 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
29863042 maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
29873043 if recvResult.limited
2988− then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}}
3044+ then yield {send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: recvResult.retryAfterMs}
29893045 else do
29903046 set *recvRates = recvResult.entries
29913047 totalMsgCount <- *totalMessageCount
29923048 if totalMsgCount >= maxMessages
2993− then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: True {}}
3049+ then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: True {}, retryAfterMs: 0}
29943050 else do
29953051 -- Recipient count from maintained counts (O(unique_recipients) instead of O(all_messages))
29963052 recipientCounts <- *recipientMsgCounts
29973053 let recipientCount = getRecipientCount to recipientCounts
29983054 if recipientCount >= maxMessagesPerRecipient
2999− then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}}
3055+ then yield {send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}
30003056 else do
30013057 -- Hot path: read only recipient's presence bucket
30023058 recipientPresence <- readPresenceBucket (pubkeyBucket to)
@@ -3025,7 +3081,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
30253081 let newFwds = do s <- remoteServers; where none forwards (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}
30263082 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
30273083 set *forwards = union forwards cappedFwds)
3028− yield {send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}, globalCapReached: False {}})
3084+ yield {send: SendDone {delivery: delivery, remoteServers: remoteServers}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0})
30293085 when result.senderRateLimited
30303086 (logWarn ("sendMessage: rate limited (" ++ show maxSendRatePerKey ++ "/s per key)"))
30313087 when result.globalCapReached
@@ -3058,9 +3114,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
30583114 logWarn ("federation: send forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))
30593115 _ -> yield {}
30603116 yield (case result.send of
3061− SendFailed {status} -> respond {status: status, messageId: ""}
3062− SendDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId}
3063− SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})
3117+ SendFailed {status} -> respond {status: status, messageId: "", retryAfterMs: result.retryAfterMs}
3118+ SendDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0}
3119+ SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId, retryAfterMs: 0})
30643120
30653121-- GET /lookup/:pubkey: Look up the best known server for a pubkey.
30663122-- Returns local server if key is online locally, otherwise the remote with latest expiry.
@@ -3266,7 +3322,8 @@ handleVerifyToken = \token scope clientIp respond -> do
32663322 let match = any pending (\p -> safeCompare p.token token && p.scope == scope && t - p.createdAt < verificationTokenExpiry)
32673323 when match (set *pendingVerifications = do p <- pending; where not (safeCompare p.token token); yield p)
32683324 yield match
3269− when (not found) (logWarn ("federation/verify: failed attempt from " ++ clientIp ++ " (scope: " ++ scope ++ ")"))
3325+ -- Omit scope from log to prevent scope enumeration by attackers probing the endpoint
3326+ when (not found) (logWarn ("federation/verify: failed attempt from " ++ clientIp))
32703327 yield (respond {ok: found})
32713328
32723329-- POST /federation/gossip: Receive presence gossip from a peer server.
@@ -3308,6 +3365,14 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
33083365 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
33093366 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
33103367 let deduped = mergeByPubkeyServer [] filteredOnline
3368+ -- Warn when a peer claims keys that are locally authenticated (potential presence hijack)
3369+ localClashes <- atomic do
3370+ let firstParty = do e <- deduped; where e.server == fromServer; yield e
3371+ let targetBuckets = uniqueBucketsFor firstParty (\e -> e.pubkey)
3372+ localPresence <- readSelectedPresenceBuckets targetBuckets
3373+ yield (do e <- firstParty; where any localPresence (\p -> p.pubkey == e.pubkey && isLocalServer p.server && p.expiresAt > t); yield e.pubkey)
3374+ when (count localClashes > 0)
3375+ (logWarn ("gossip: " ++ fromServer ++ " claims " ++ show (count localClashes) ++ " locally-authenticated key(s) — ignoring for local routing"))
33113376 gossipResult <- atomic do
33123377 allPresence <- readAllPresence
33133378 tpc <- *totalPresenceCount
@@ -3439,9 +3504,15 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
34393504 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
34403505 fwdNonces <- *forwardNonces
34413506 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
3442− -- Batch-evict 10% when at capacity to create a buffer; makes it harder
3443− -- for an attacker to force-evict specific nonces via targeted flooding.
3444− set *forwardNonces = if count updatedNonces > maxForwardNonces then (evictSmallest updatedNonces (maxForwardNonces / 10) (\n -> n.seenAt)).kept else updatedNonces
3507+ -- Per-origin eviction: when at capacity, only evict nonces from the
3508+ -- origin server that's causing the overflow. Prevents one peer from
3509+ -- force-evicting another peer's nonces via targeted flooding.
3510+ set *forwardNonces = if count updatedNonces > maxForwardNonces
3511+ then let fromOrigin = do n <- updatedNonces; where n.originServer == originServer; yield n in
3512+ let others = do n <- updatedNonces; where n.originServer != originServer; yield n in
3513+ let evictCount = max 1 (count fromOrigin / 10) in
3514+ union others (evictSmallest fromOrigin evictCount (\n -> n.seenAt)).kept
3515+ else updatedNonces
34453516 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
34463517 case result.forward of
34473518 ForwardRejected {} ->
@@ -3460,7 +3531,9 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
34603531-- recipient has no local presence and no other pending forwards for that message.
34613532handleFederationAck = \messageIds originServer fromServer requestTimestamp authorization respond -> do
34623533 t <- now
3463− withFederationAuth authorization t respond fromServer "ack" requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (do
3534+ -- Wrap respond to add ackedCount default for withFederationAuth error paths
3535+ let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0}
3536+ withFederationAuth authorization t ackRespond fromServer "ack" requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (do
34643537 let dedupedIds = dedupMessageIds messageIds
34653538 -- Clear forward entries for the acking server. Extract removed forwards' toKey
34663539 -- so we can reclaim origin messages that are no longer needed.
@@ -3469,15 +3542,16 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp autho
34693542 let removed = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
34703543 set *forwards = forwardsExcluding forwards fromServer dedupedIds
34713544 yield removed
3545+ let ackedCount = count removedForwards
34723546 -- Deferred origin message cleanup: queue for processing in the next prune cycle
34733547 -- after ackGracePeriod elapses. Replaces fork+sleep which accumulated sleeping
34743548 -- goroutines under high federation ack volume.
3475− when (count removedForwards > 0) (atomic do
3549+ when (ackedCount > 0) (atomic do
34763550 cleanups <- *pendingAckCleanups
34773551 let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
34783552 let combined = union cleanups newCleanups
34793553 set *pendingAckCleanups = if count combined > maxPendingAckCleanups then takeRelation maxPendingAckCleanups combined else combined)
3480− yield (respond {ok: True {}, error: ""}))
3554+ yield (respond {ok: True {}, error: "", ackedCount: ackedCount}))
34813555
34823556-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
34833557-- Call before stopping the server so peers learn keys went offline immediately.
@@ -3555,8 +3629,9 @@ handlePrune = \authorization respond -> do
35553629 when authorized (fork (pruneAndRetry t))
35563630 yield (respond {ok: authorized})
35573631
3558−handleHealth = \respond -> do
3632+handleHealth = \authorization respond -> do
35593633 t <- now
3634+ let isAuthed = verifyMaintenanceAuth authorization
35603635 -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
35613636 result <- atomic do
35623637 lastPrune <- *lastPrunedAt
@@ -3586,11 +3661,17 @@ handleHealth = \respond -> do
35863661 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
35873662 capacityOk: capacityOk,
35883663 federationOk: fedOk}
3589− yield (respond {ok: result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk,
3590− pruneHealthy: result.pruneOk,
3664+ let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk
3665+ -- Unauthenticated: return only aggregate ok flag to avoid leaking capacity/federation details
3666+ if isAuthed
3667+ then yield (respond {ok: allOk, pruneHealthy: result.pruneOk,
35913668 heartbeatHealthy: result.heartbeatOk,
35923669 capacityHealthy: result.capacityOk,
35933670 federationHealthy: result.federationOk})
3671+ else yield (respond {ok: allOk, pruneHealthy: allOk,
3672+ heartbeatHealthy: allOk,
3673+ capacityHealthy: allOk,
3674+ federationHealthy: allOk})
35943675
35953676zeroStats = {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pairRates: 0, federationPairRates: 0, pruneCount: 0, pruneRunning: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, draining: False {}, federationEnabled: False {}, recentGossip: 0, totalMessageCount: 0, recipientCounts: 0, federationInFlight: 0, hungFederationSlots: 0, messagesPct: 0, presencePct: 0, forwardsPct: 0, sessionsPct: 0, federationAvgLatencyMs: 0, federationMaxLatencyMs: 0, federationCallCount: 0, drainCompletedAt: 0, oldestForwardAge: 0, oldestAckAge: 0, ok: False {}}
35963677
@@ -3690,7 +3771,7 @@ handleStatsKey = \pubkey authorization respond -> do
36903771 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})
36913772 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})
36923773
3693−emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: []}
3774+emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: []}
36943775
36953776-- GET /maintenance/snapshot: Export federation metadata for persistence.
36963777-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
@@ -3708,22 +3789,24 @@ handleSnapshot = \authorization respond -> do
37083789 d <- *deliveredIds
37093790 ff <- *forwardFailures
37103791 f <- *forwards
3711− yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f}
3792+ fn <- *forwardNonces
3793+ yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn}
37123794 atomic do set *lastSnapshotExportAt = t
3713− logInfo ("snapshot: exported " ++ show (count result.presence) ++ " presence, " ++ show (count result.ackedIds) ++ " ackedIds, " ++ show (count result.deliveredIds) ++ " deliveredIds, " ++ show (count result.forwards) ++ " forwards")
3714− yield (respond {ok: True {}, version: snapshotVersion, presence: result.presence, ackedIds: result.ackedIds, ackOverflow: result.ackOverflow, deliveredIds: result.deliveredIds, forwardFailures: result.forwardFailures, forwards: result.forwards})
3795+ 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")
3796+ 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})
37153797
37163798-- POST /maintenance/restore: Import a previously-exported snapshot.
37173799-- Requires drain mode to prevent conflicts with live state. Merges imported data
37183800-- with existing state (dedup entries are unioned, presence is merged by freshness).
37193801-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
37203802-- trigger federation forward creation from messages + restored presence.
3721−handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData authorization respond -> do
3803+handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsData forwardFailuresData forwardsData forwardNoncesData authorization respond -> do
37223804 let authorized = verifyMaintenanceAuth authorization
37233805 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
37243806 else do
3725− -- Version 0 = pre-versioning snapshot (accept for backwards compatibility)
3726− if version != 0 && version != snapshotVersion then do
3807+ -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces (accept both for backwards compatibility;
3808+ -- missing forwardNonces field will default to empty list)
3809+ if version != 0 && version != 1 && version != snapshotVersion then do
37273810 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
37283811 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
37293812 else do
@@ -3771,6 +3854,12 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
37713854 let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) forwards validForwards
37723855 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
37733856 set *forwards = cappedForwards
3857+ -- Restore forward nonces (dedup by messageId + toKey + originServer, keep fresher on conflict)
3858+ forwardNonces <- *forwardNonces
3859+ let validNonces = do n <- forwardNoncesData; where t - n.seenAt < forwardNonceRetention; yield n
3860+ let mergedNonces = mergeByKey (\a b -> a.messageId == b.messageId && a.toKey == b.toKey && a.originServer == b.originServer) (\x -> x.seenAt) forwardNonces validNonces
3861+ let cappedNonces = takeRelation maxForwardNonces mergedNonces
3862+ set *forwardNonces = cappedNonces
37743863 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
37753864 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
37763865 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
@@ -3826,16 +3915,44 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
38263915 retryEvictedAcks result.evictedEntries result.pendingAcks))
38273916 yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount}))
38283917
3918+-- GET /messages/list: List pending messages without acking.
3919+-- Useful after cursor loss (app reinstall, new device) to see what's pending.
3920+-- Rate-limited via lookup rate limiter (read-only info lookup).
3921+handleListMessages = \authorization clientIp respond -> do
3922+ t <- now
3923+ globalRateLimited <- checkGlobalRate t
3924+ if globalRateLimited then yield (respond {messages: [], authorized: False {}, count: 0})
3925+ else
3926+ withSessionAuth authorization t clientIp
3927+ (yield (respond {messages: [], authorized: False {}, count: 0}))
3928+ (\pubkey -> do
3929+ listRateLimited <- atomic do
3930+ lookupRates <- *lookupRates
3931+ let rateResult = checkRateWindow lookupRates (\r -> r.pubkey == pubkey)
3932+ (\ts -> {pubkey: pubkey, windowStart: ts, count: 1})
3933+ (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3934+ maxLookupRatePerKey lookupRateWindowMs maxLookupRateEntries t
3935+ set *lookupRates = rateResult.entries
3936+ yield rateResult.limited
3937+ if listRateLimited then yield (respond {messages: [], authorized: True {}, count: 0})
3938+ else do
3939+ result <- atomic do
3940+ bucket <- readMessageBucket (pubkeyBucket pubkey)
3941+ let myMsgs = sortBy (\m -> m.receivedAt) (do m <- bucket; where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge; yield m)
3942+ let capped = takeRelation maxPollEvents myMsgs
3943+ yield {msgs: do m <- capped; yield {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt}, total: count myMsgs}
3944+ yield (respond {messages: result.msgs, authorized: True {}, count: result.total}))
3945+
38293946-- POST /messages/batch: Send multiple messages in one request.
38303947-- Auth checked once; each message processed independently with its own rate limiting.
38313948-- Partial success is possible — check per-message status in results.
38323949handleSendBatch = \messages authorization clientIp respond -> do
38333950 t <- now
38343951 globalRateLimited <- checkGlobalRate t
3835− if globalRateLimited then yield (respond {results: [], authorized: False {}})
3952+ if globalRateLimited then yield (respond {results: [], authorized: False {}, retryAfterMs: 0})
38363953 else
38373954 withSessionAuth authorization t clientIp
3838− (yield (respond {results: [], authorized: False {}}))
3955+ (yield (respond {results: [], authorized: False {}, retryAfterMs: 0}))
38393956 (\senderKey -> do
38403957 let capped = takeRelation maxMessageIds messages
38413958 results <- sendBatchInner senderKey capped t []
@@ -3868,7 +3985,9 @@ handleSendBatch = \messages authorization clientIp respond -> do
38683985 removeForward r.messageId server
38693986 clearFailure server
38703987 else logWarn ("federation: batch forward to " ++ server ++ " rejected for " ++ r.messageId ++ ": " ++ response.error)))))))
3871− yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}}))
3988+ -- Aggregate retryAfterMs: use the maximum across all rate-limited messages
3989+ let batchRetryMs = fold (\acc r -> max acc r.retryAfterMs) 0 results
3990+ yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}, retryAfterMs: batchRetryMs}))
38723991
38733992sendBatchInner = \senderKey messages t acc ->
38743993 let split = splitAtN messages 1 in
@@ -3882,28 +4001,28 @@ sendBatchInner = \senderKey messages t acc ->
38824001 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
38834002 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
38844003 result <- if t - msg.timestamp > messageMaxAge
3885− then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0}
4004+ then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0, retryAfterMs: 0}
38864005 else if senderKey == msg.to
3887− then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0}
4006+ then yield {messageId: "", status: MsRejected {}, needsForward: False {}, forwardServer: "", toKey: "", blob: "", ts: 0, retryAfterMs: 0}
38884007 else do
38894008 let clampedTs = clampTimestamp msg.timestamp t
38904009 sendResult <- atomic do
38914010 draining <- *draining
3892− if draining then yield {status: MsDraining {}, fwdServer: "", needsFwd: False {}}
4011+ if draining then yield {status: MsDraining {}, fwdServer: "", needsFwd: False {}, retryMs: 0}
38934012 else do
38944013 let msgBucket = pubkeyBucket msg.to
38954014 bucket <- readMessageBucket msgBucket
38964015 ackedIds <- *ackedIds
38974016 deliveredIds <- *deliveredIds
38984017 let isDup = checkMessageDup bucket effectiveMsgId msg.to || (effectiveMsgId != "" && (hasMessage effectiveMsgId msg.to ackedIds || hasMessage effectiveMsgId msg.to deliveredIds))
3899− if isDup then yield {status: MsDuplicate {}, fwdServer: "", needsFwd: False {}}
4018+ if isDup then yield {status: MsDuplicate {}, fwdServer: "", needsFwd: False {}, retryMs: 0}
39004019 else do
39014020 sendRates <- *sendRates
39024021 let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
39034022 (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})
39044023 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
39054024 maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
3906− if rateResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}}
4025+ if rateResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: rateResult.retryAfterMs}
39074026 else do
39084027 set *sendRates = rateResult.entries
39094028 recvRates <- *recvRates
@@ -3911,7 +4030,7 @@ sendBatchInner = \senderKey messages t acc ->
39114030 (\ts -> {pubkey: msg.to, windowStart: ts, count: 1})
39124031 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
39134032 maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
3914− if recvResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}}
4033+ if recvResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: recvResult.retryAfterMs}
39154034 else do
39164035 set *recvRates = recvResult.entries
39174036 -- Per-sender-per-recipient pair rate limiting (prevents one sender filling a recipient's mailbox)
@@ -3920,15 +4039,15 @@ sendBatchInner = \senderKey messages t acc ->
39204039 (\ts -> {sender: senderKey, recipient: msg.to, windowStart: ts, count: 1})
39214040 (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
39224041 maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
3923− if pairResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}}
4042+ if pairResult.limited then yield {status: MsRateLimited {}, fwdServer: "", needsFwd: False {}, retryMs: pairResult.retryAfterMs}
39244043 else do
39254044 set *pairRates = pairResult.entries
39264045 totalMsgCount <- *totalMessageCount
3927− if totalMsgCount >= maxMessages then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}}
4046+ if totalMsgCount >= maxMessages then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}, retryMs: 0}
39284047 else do
39294048 recipientCounts <- *recipientMsgCounts
39304049 let recipientCount = getRecipientCount msg.to recipientCounts
3931− if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}}
4050+ if recipientCount >= maxMessagesPerRecipient then yield {status: MsRejected {}, fwdServer: "", needsFwd: False {}, retryMs: 0}
39324051 else do
39334052 recipientPresence <- readPresenceBucket (pubkeyBucket msg.to)
39344053 seqCounter <- *seqCounter
@@ -3952,13 +4071,13 @@ sendBatchInner = \senderKey messages t acc ->
39524071 let delivery = if localOnline then MsDelivered {}
39534072 else if federationEnabled && count remoteServers > 0 then MsFederated {}
39544073 else MsQueued {}
3955− yield {status: delivery, fwdServer: fwdServer, needsFwd: federationEnabled && count remoteServers > 0}
4074+ yield {status: delivery, fwdServer: fwdServer, needsFwd: federationEnabled && count remoteServers > 0, retryMs: 0}
39564075 -- Log rate limiting and capacity warnings (matching handleSendMessage)
39574076 case sendResult.status of
39584077 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
39594078 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")
39604079 _ -> yield {}
3961− yield {messageId: effectiveMsgId, status: sendResult.status, needsForward: sendResult.needsFwd, forwardServer: sendResult.fwdServer, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs}
4080+ yield {messageId: effectiveMsgId, status: sendResult.status, needsForward: sendResult.needsFwd, forwardServer: sendResult.fwdServer, toKey: msg.to, blob: msg.encryptedBlob, ts: clampedTs, retryAfterMs: sendResult.retryMs}
39624081 sendBatchInner senderKey split.rest t (union acc [result])
39634082
39644083serve = \req -> case req of
@@ -3971,6 +4090,7 @@ serve = \req -> case req of
39714090 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
39724091 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
39734092 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
4093+ ListMessages {authorization, clientIp, respond} -> handleListMessages authorization clientIp respond
39744094 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
39754095 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
39764096 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond
@@ -3980,8 +4100,8 @@ serve = \req -> case req of
39804100 Drain {authorization, respond} -> handleDrain authorization respond
39814101 Undrain {authorization, respond} -> handleUndrain authorization respond
39824102 Snapshot {authorization, respond} -> handleSnapshot authorization respond
3983− Restore {version, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, authorization, respond} -> handleRestore version presence ackedIds ackOverflow deliveredIds forwardFailures forwards authorization respond
3984− Health {respond} -> handleHealth respond
4103+ Restore {version, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, authorization, respond} -> handleRestore version presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces authorization respond
4104+ Health {authorization, respond} -> handleHealth authorization respond
39854105 Stats {authorization, respond} -> handleStats authorization respond
39864106 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
39874107 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
@@ -3992,6 +4112,9 @@ serve = \req -> case req of
39924112
39934113-- Background loop: periodically prunes expired state and retries pending federation ops.
39944114-- Jitter prevents thundering herd when multiple servers restart simultaneously.
4115+-- NOTE: This function (and all background loops below) recurse via tail calls.
4116+-- Knot compiles tail calls to loops, so stack growth is bounded. If Knot's tail-call
4117+-- optimization is ever disabled, these loops will overflow the stack over time.
39954118backgroundPrune = do
39964119 jitter <- randomInt 5000
39974120 sleep (backgroundPruneInterval + jitter)