@@ -139,7 +139,7 @@ federationLatencyWindowMs : Int<Ms>
139federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
139federationLatencyWindowMs = 300000 -- 5 minutes; reset latency metrics after this interval
141-- Snapshot format version (increment on schema changes to prevent corrupt restores)
141-- Snapshot format version (increment on schema changes to prevent corrupt restores)
142−snapshotVersion = 1
142+snapshotVersion = 2
144-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
144-- Per-peer gossip event volume limiting (prevents flooding via large batches of unique pubkeys)
145maxGossipEventsPerServer = 500 -- max total gossip events per server per window
145maxGossipEventsPerServer = 500 -- max total gossip events per server per window
@@ -315,6 +315,9 @@ type FederationInFlightEntry = {server: Text, count: Int, firstStartedAt: Timest
315type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
315type RecentFederationRequest = {server: Text, timestamp: Timestamp, seenAt: Timestamp}
316type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
316type GossipEventRateEntry = {server: Text, windowStart: Timestamp, count: Int}
318+-- Forward nonce for extended replay protection (persisted in snapshots)
319+type ForwardNonce = {messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, seenAt: Timestamp}
318-- Batch send types
321-- Batch send types
319type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId}
322type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId}
320type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
323type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
@@ -516,7 +519,7 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
516*pendingVerifications : [{token: Text, createdAt: Timestamp, scope: Text}]
519*pendingVerifications : [{token: Text, createdAt: Timestamp, scope: Text}]
518-- Forward nonce dedup: (messageId, toKey, originServer, seenAt) for replay protection beyond message lifetime
521-- 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]
521-- Federation verify per-IP rate limiting (tighter than general IP rate)
524-- Federation verify per-IP rate limiting (tighter than general IP rate)
522*verifyIpRates : [IpRateEntry]
525*verifyIpRates : [IpRateEntry]
@@ -549,7 +552,7 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
549route AuthApi where
552route AuthApi where
551 POST {pubkey: PubkeyHex} /challenge headers {clientIp: Text}
554 POST {pubkey: PubkeyHex} /challenge headers {clientIp: Text}
552− -> {challenge: Text, ok: Bool, expiresAt: Timestamp}
555+ -> {challenge: Text, ok: Bool, expiresAt: Timestamp, retryAfterMs: Int}
553 = ReqChallenge
556 = ReqChallenge
554 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /verify headers {clientIp: Text}
557 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /verify headers {clientIp: Text}
555 -> {token: Text, expiresAt: Timestamp}
558 -> {token: Text, expiresAt: Timestamp}
@@ -566,12 +569,12 @@ route AuthApi where
566route ClientApi where
569route ClientApi where
567 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
570 -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
568 POST {cursor: Timestamp, generation: Int, watchKeys: [PubkeyHex]} /poll headers {authorization: Text, clientIp: Text}
571 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}
571 /messages
574 /messages
572 -- Send an E2E encrypted message to a recipient pubkey
575 -- Send an E2E encrypted message to a recipient pubkey
573 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text, clientIp: Text}
576 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}
575 = SendMessage
578 = SendMessage
576 -- Check delivery status of a sent message (caller must know recipient to prove ownership)
579 -- Check delivery status of a sent message (caller must know recipient to prove ownership)
577 POST {to: PubkeyHex, messageId: MessageIdHex} /status headers {authorization: Text, clientIp: Text}
580 POST {to: PubkeyHex, messageId: MessageIdHex} /status headers {authorization: Text, clientIp: Text}
@@ -583,8 +586,12 @@ route ClientApi where
583 = AckMessages
586 = AckMessages
584 -- Send multiple messages in one request (auth checked once; each message rate-limited independently)
587 -- Send multiple messages in one request (auth checked once; each message rate-limited independently)
585 POST {messages: [BatchMessage]} /batch headers {authorization: Text, clientIp: Text}
588 POST {messages: [BatchMessage]} /batch headers {authorization: Text, clientIp: Text}
586− -> {results: [BatchResult], authorized: Bool}
589+ -> {results: [BatchResult], authorized: Bool, retryAfterMs: Int}
587 = SendBatch
590 = 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}
588 -- Look up which server a pubkey is present on (for federation routing)
595 -- Look up which server a pubkey is present on (for federation routing)
589 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
596 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text}
590 -> {server: AnyServerName, authorized: Bool}
597 -> {server: AnyServerName, authorized: Bool}
@@ -603,7 +610,7 @@ route FederationApi where
603 = ForwardMessage
610 = ForwardMessage
604 -- Acknowledge delivery of messages back to the origin server
611 -- Acknowledge delivery of messages back to the origin server
605 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName, requestTimestamp: Timestamp} /ack headers {authorization: Text}
612 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}
607 = FederationAck
614 = FederationAck
609-- Federation callback verification: the receiving server calls back to the sender
616-- Federation callback verification: the receiving server calls back to the sender
@@ -626,16 +633,16 @@ route MaintenanceApi where
626 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
633 -- Messages are excluded (ephemeral by design; clients handle re-delivery).
627 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
634 -- Includes presence, acked/delivered IDs (dedup state), and peer failure records.
628 GET /maintenance/snapshot headers {authorization: Text}
635 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]}
630 = Snapshot
637 = Snapshot
631 -- Restore: import a previously-exported snapshot. Server must be in drain mode
638 -- Restore: import a previously-exported snapshot. Server must be in drain mode
632 -- to prevent conflicts with live state. Run /maintenance/undrain after restore.
639 -- 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}
634 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
641 -> {ok: Bool, error: Text, presenceCount: Int, ackedIdsCount: Int, deliveredIdsCount: Int}
635 = Restore
642 = Restore
637route HealthApi where
644route 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
640-- Returns counts of in-memory collections (requires maintenance secret)
647-- Returns counts of in-memory collections (requires maintenance secret)
641route StatsApi where
648route StatsApi where
@@ -897,6 +904,14 @@ authenticateSession = \sessions token t ->
897 Nothing {} -> Unauthorized {}
904 Nothing {} -> Unauthorized {}
898 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}
905 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}
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)
900-- Extract bearer token from "Authorization: Bearer <token>" header.
915-- Extract bearer token from "Authorization: Bearer <token>" header.
901-- Rejects empty tokens early to avoid unnecessary session bucket scans.
916-- Rejects empty tokens early to avoid unnecessary session bucket scans.
902extractToken = \authHeader ->
917extractToken = \authHeader ->
@@ -1395,11 +1410,18 @@ getRecipientCount : PubkeyHex -> [RecipientCount] -> Int
1395getRecipientCount = \pubkey counts ->
1410getRecipientCount = \pubkey counts ->
1396 fold (\acc c -> if c.pubkey == pubkey then c.count else acc) 0 counts
1411 fold (\acc c -> if c.pubkey == pubkey then c.count else acc) 0 counts
1413+-- Single-pass recipient count adjustment: fold once instead of any + list comprehension (O(n) vs O(2n))
1398adjustRecipientCount : [RecipientCount] -> PubkeyHex -> Int -> [RecipientCount]
1414adjustRecipientCount : [RecipientCount] -> PubkeyHex -> Int -> [RecipientCount]
1399adjustRecipientCount = \counts pubkey delta ->
1415adjustRecipientCount = \counts pubkey delta ->
1400 if delta == 0 then counts
1416 if delta == 0 then counts
1417+ else let result = fold (\acc c ->
1401− else if any counts (\c -> c.pubkey == pubkey)
1418+ if 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)
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
1403 else if delta > 0 then union counts [{pubkey: pubkey, count: delta}]
1425 else if delta > 0 then union counts [{pubkey: pubkey, count: delta}]
1404 else counts
1426 else counts
@@ -1537,19 +1559,21 @@ matchIp = \clientIp sessionIp ->
1537 else if clientIp == "" then False {} -- session has IP but request doesn't — proxy disappeared
1559 else if clientIp == "" then False {} -- session has IP but request doesn't — proxy disappeared
1538 else safeCompareLoop clientIp sessionIp maxServerNameLen 0 0 == 0
1560 else safeCompareLoop clientIp sessionIp maxServerNameLen 0 0 == 0
1540−-- Generic sliding-window rate limiter. Returns whether the request is rate-limited
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).
1541−-- and the updated entries list. Callers provide match/create/update functions for their entry type.
1564+-- Callers provide match/create/update functions for their entry type.
1542checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntries t ->
1565checkRateWindow = \entries matchKey makeNew updateCount maxRate windowMs maxEntries t ->
1543 let entry = headOf (do r <- entries; where matchKey r; yield r) in
1566 let entry = headOf (do r <- entries; where matchKey r; yield r) in
1544 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= windowMs in
1567 let isNewWindow = case entry of Nothing {} -> True {}; Just {value: r} -> t - r.windowStart >= windowMs in
1545 let rateCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count in
1568 let rateCount = case entry of Nothing {} -> 0; Just {value: r} -> if isNewWindow then 0 else r.count in
1546 if rateCount >= maxRate
1569 if rateCount >= maxRate
1570+ then let retryMs = case entry of Nothing {} -> windowMs; Just {value: r} -> max 0 (windowMs - (t - r.windowStart))
1547− then {limited: True {}, entries: entries}
1571+ in {limited: True {}, entries: entries, retryAfterMs: retryMs}
1549 let newEntries = if isNewWindow
1573 let newEntries = if isNewWindow
1550 then union (do r <- entries; where not (matchKey r); yield r) [makeNew t]
1574 then union (do r <- entries; where not (matchKey r); yield r) [makeNew t]
1551 else do r <- entries; yield (if matchKey r then updateCount r else r)
1575 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}
1554-- Extract the item with the smallest key and return both the item and remaining list
1578-- Extract the item with the smallest key and return both the item and remaining list
1555extractOldest = \items getKey ->
1579extractOldest = \items getKey ->
@@ -2595,15 +2619,21 @@ pollOnce = \token cursor clientIp -> do
2595 if wakeup > s.startWakeup || signal > s.startSignal
2619 if wakeup > s.startWakeup || signal > s.startSignal
2596 then yield {}
2620 then yield {}
2597 else retry
2621 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.
2599 t2 <- now
2626 t2 <- now
2600 result <- atomic do
2627 result <- atomic do
2628+ isDraining <- *draining
2629+ if isDraining then yield {authorized: True {}, events: [], cursor: s.clampedCursor}
2601− sessions2 <- readSessionBucket bucket
2631+ sessions2 <- readSessionBucket bucket
2602− case authenticateSession sessions2 token t2 of
2632+ if not (reauthenticateSession sessions2 token s.pubkey t2)
2603− Unauthorized {} -> yield {authorized: False {}, events: [], cursor: s.clampedCursor}
2633+ then yield {authorized: False {}, events: [], cursor: s.clampedCursor}
2604− Authorized {} -> do
2634+ else do
2605− r <- pollNewEvents s.pubkey s.clampedCursor t2
2635+ r <- pollNewEvents s.pubkey s.clampedCursor t2
2606− yield {authorized: True {}, events: r.events, cursor: r.cursor}
2636+ yield {authorized: True {}, events: r.events, cursor: r.cursor}
2607 yield {authorized: result.authorized, events: result.events, cursor: result.cursor}
2637 yield {authorized: result.authorized, events: result.events, cursor: result.cursor}
2608 -- Release poll slot. Safety: the code between acquire and release only contains
2638 -- Release poll slot. Safety: the code between acquire and release only contains
2609 -- atomic reads and STM retry (no IO that can throw), so this path is always reached.
2639 -- atomic reads and STM retry (no IO that can throw), so this path is always reached.
@@ -2704,8 +2734,15 @@ newlyOnlineRemoteKeys = \events appliedKeys ->
2704-- Trust gossip entries if they're first-party (server == fromServer) or updating
2734-- Trust gossip entries if they're first-party (server == fromServer) or updating
2705-- an existing (pubkey, server) presence entry. Third-party gossip cannot create
2735-- an existing (pubkey, server) presence entry. Third-party gossip cannot create
2706-- new entries for keys we haven't seen — only the key's home server can do that.
2736-- 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.
2707trustedGossipEntries = \deduped presence fromServer t ->
2740trustedGossipEntries = \deduped presence fromServer t ->
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
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}
2745+ ; yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp}
2710-- Keys whose presence was actually updated (new or extended expiry)
2747-- Keys whose presence was actually updated (new or extended expiry)
2711appliedOnlineKeys = \entries presence ->
2748appliedOnlineKeys = \entries presence ->
@@ -2741,10 +2778,10 @@ handleChallenge = \pubkey clientIp respond -> do
2741 if preAuth != ""
2778 if preAuth != ""
2743 when (preAuth == "ip_rate_limited") (logWarn ("challenge: IP rate limited for " ++ clientIp))
2780 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})
2746 authLimited <- checkAuthGlobalRate t
2783 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})
2749 challenge <- genToken
2786 challenge <- genToken
2750 result <- atomic do
2787 result <- atomic do
@@ -2757,7 +2794,7 @@ handleChallenge = \pubkey clientIp respond -> do
2757 union others (evictOneOldest forKey (\c -> c.expiresAt))
2794 union others (evictOneOldest forKey (\c -> c.expiresAt))
2758 else active
2795 else active
2759 if count cappedActive >= maxChallenges
2796 if count cappedActive >= maxChallenges
2760− then yield {accepted: False {}, rateLimited: False {}}
2797+ then yield {accepted: False {}, rateLimited: False {}, retryAfterMs: 0}
2762 challengeRates <- *challengeRates
2799 challengeRates <- *challengeRates
2763 let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
2800 let rateResult = checkRateWindow challengeRates (\r -> r.pubkey == pubkey)
@@ -2765,16 +2802,16 @@ handleChallenge = \pubkey clientIp respond -> do
2765 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
2802 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
2766 maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
2803 maxChallengeRatePerKey challengeRateWindowMs maxChallengeRateEntries t
2767 if rateResult.limited
2804 if rateResult.limited
2768− then yield {accepted: False {}, rateLimited: True {}}
2805+ then yield {accepted: False {}, rateLimited: True {}, retryAfterMs: rateResult.retryAfterMs}
2770 set *challengeRates = rateResult.entries
2807 set *challengeRates = rateResult.entries
2771 set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
2808 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}
2773 when result.rateLimited
2810 when result.rateLimited
2774 (logWarn ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
2811 (logWarn ("challenge: rate limited (" ++ show maxChallengeRatePerKey ++ "/s per key), rejecting for " ++ pubkey))
2775 when (not result.accepted && not result.rateLimited)
2812 when (not result.accepted && not result.rateLimited)
2776 (logWarn ("challenge: at capacity (" ++ show maxChallenges ++ "), rejecting for " ++ pubkey))
2813 (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})
2779-- POST /auth/verify: Verify Ed25519 signature over challenge string.
2816-- POST /auth/verify: Verify Ed25519 signature over challenge string.
2780-- On success: creates a session, registers local presence, and gossips online status.
2817-- On success: creates a session, registers local presence, and gossips online status.
@@ -2866,23 +2903,42 @@ verifyAuthInner = \pubkey challenge signature revokeOthers clientIp respond t ->
2866 let capped = capPerKeyAndGlobal afterRevoke (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
2903 let capped = capPerKeyAndGlobal afterRevoke (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
2867 let sessionsWithNew = union capped [newSession]
2904 let sessionsWithNew = union capped [newSession]
2868 set *totalSessionCount = count sessionsWithNew
2905 set *totalSessionCount = count sessionsWithNew
2869− let bFor = \b -> do s <- sessionsWithNew; where tokenBucket s.token == b; yield s
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
2870− writeSessionBucket 0 (bFor 0)
2926+ writeSessionBucket 0 partitioned.b0
2871− writeSessionBucket 1 (bFor 1)
2927+ writeSessionBucket 1 partitioned.b1
2872− writeSessionBucket 2 (bFor 2)
2928+ writeSessionBucket 2 partitioned.b2
2873− writeSessionBucket 3 (bFor 3)
2929+ writeSessionBucket 3 partitioned.b3
2874− writeSessionBucket 4 (bFor 4)
2930+ writeSessionBucket 4 partitioned.b4
2875− writeSessionBucket 5 (bFor 5)
2931+ writeSessionBucket 5 partitioned.b5
2876− writeSessionBucket 6 (bFor 6)
2932+ writeSessionBucket 6 partitioned.b6
2877− writeSessionBucket 7 (bFor 7)
2933+ writeSessionBucket 7 partitioned.b7
2878− writeSessionBucket 8 (bFor 8)
2934+ writeSessionBucket 8 partitioned.b8
2879− writeSessionBucket 9 (bFor 9)
2935+ writeSessionBucket 9 partitioned.b9
2880− writeSessionBucket 10 (bFor 10)
2936+ writeSessionBucket 10 partitioned.b10
2881− writeSessionBucket 11 (bFor 11)
2937+ writeSessionBucket 11 partitioned.b11
2882− writeSessionBucket 12 (bFor 12)
2938+ writeSessionBucket 12 partitioned.b12
2883− writeSessionBucket 13 (bFor 13)
2939+ writeSessionBucket 13 partitioned.b13
2884− writeSessionBucket 14 (bFor 14)
2940+ writeSessionBucket 14 partitioned.b14
2885− writeSessionBucket 15 (bFor 15)
2941+ writeSessionBucket 15 partitioned.b15
2886 yield (Just {value: {peerServers: peerServers, revokedCount: revokedCount}})
2942 yield (Just {value: {peerServers: peerServers, revokedCount: revokedCount}})
2887 case authResult of
2943 case authResult of
2888 Nothing {} -> yield (respond {token: "", expiresAt: 0})
2944 Nothing {} -> yield (respond {token: "", expiresAt: 0})
@@ -2903,10 +2959,10 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
2903 let effectiveCursor = if generation == serverGen then cursor else 0
2959 let effectiveCursor = if generation == serverGen then cursor else 0
2904 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
2960 let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
2905 draining <- *draining
2961 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})
2908 globalRateLimited <- checkGlobalRate t
2964 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})
2911 result <- case extractToken authorization of
2967 result <- case extractToken authorization of
2912 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
2968 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
@@ -2924,7 +2980,7 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
2924 sendAcksAndClean result.pendingAcks
2980 sendAcksAndClean result.pendingAcks
2925 retryEvictedAcks result.evictedEntries result.pendingAcks))
2981 retryEvictedAcks result.evictedEntries result.pendingAcks))
2926 let isHeartbeat = result.authorized && count result.events == 0
2982 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})
2929-- POST /messages: Send an encrypted message to a recipient.
2985-- POST /messages: Send an encrypted message to a recipient.
2930-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
2986-- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
@@ -2932,11 +2988,11 @@ handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
2932handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp respond -> do
2988handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp respond -> do
2933 t <- now
2989 t <- now
2934 globalRateLimited <- checkGlobalRate t
2990 globalRateLimited <- checkGlobalRate t
2935− if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: ""})
2991+ if globalRateLimited then yield (respond {status: MsRateLimited {}, messageId: "", retryAfterMs: 0})
2937 effectiveMsgId <- if messageId != "" then yield messageId else genToken
2993 effectiveMsgId <- if messageId != "" then yield messageId else genToken
2938 if t - timestamp > messageMaxAge
2994 if t - timestamp > messageMaxAge
2939− then yield (respond {status: MsRejected {}, messageId: ""})
2995+ then yield (respond {status: MsRejected {}, messageId: "", retryAfterMs: 0})
2941 when (timestamp > t + clockSkewTolerance)
2997 when (timestamp > t + clockSkewTolerance)
2942 (logWarn ("sendMessage: future timestamp detected (+" ++ show (timestamp - t) ++ "ms ahead), clamping to now"))
2998 (logWarn ("sendMessage: future timestamp detected (+" ++ show (timestamp - t) ++ "ms ahead), clamping to now"))
@@ -2945,12 +3001,12 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2945 let clampedTs = clampTimestamp timestamp t
3001 let clampedTs = clampTimestamp timestamp t
2946 result <- atomic do
3002 result <- atomic do
2947 draining <- *draining
3003 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}
2949 else withSessionAuth authorization t clientIp
3005 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})
2951 (\senderKey -> do
3007 (\senderKey -> do
2952 -- Reject self-sends (sender == recipient wastes a mailbox slot)
3008 -- 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}
2955 let msgBucket = pubkeyBucket to
3011 let msgBucket = pubkeyBucket to
2956 bucket <- readMessageBucket msgBucket
3012 bucket <- readMessageBucket msgBucket
@@ -2959,7 +3015,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2959 -- Dedup check (only when messageId is non-empty; scans only the recipient's bucket)
3015 -- Dedup check (only when messageId is non-empty; scans only the recipient's bucket)
2960 let isDup = checkMessageDup bucket messageId to || (messageId != "" && (hasMessage messageId to ackedIds || hasMessage messageId to deliveredIds))
3016 let isDup = checkMessageDup bucket messageId to || (messageId != "" && (hasMessage messageId to ackedIds || hasMessage messageId to deliveredIds))
2961 if isDup
3017 if isDup
2962− then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}}
3018+ then yield {send: SendDuplicate {}, senderRateLimited: False {}, globalCapReached: False {}, retryAfterMs: 0}
2964 sendRates <- *sendRates
3020 sendRates <- *sendRates
2965 let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
3021 let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
@@ -2967,7 +3023,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2967 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3023 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
2968 maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
3024 maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
2969 if rateResult.limited
3025 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}
2972 set *sendRates = rateResult.entries
3028 set *sendRates = rateResult.entries
2973 pairRates <- *pairRates
3029 pairRates <- *pairRates
@@ -2977,7 +3033,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2977 maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
3033 maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
2978 set *pairRates = pairResult.entries
3034 set *pairRates = pairResult.entries
2979 if pairResult.limited
3035 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}
2982 recvRates <- *recvRates
3038 recvRates <- *recvRates
2983 let recvResult = checkRateWindow recvRates (\r -> r.pubkey == to)
3039 let recvResult = checkRateWindow recvRates (\r -> r.pubkey == to)
@@ -2985,18 +3041,18 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
2985 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3041 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
2986 maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
3042 maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
2987 if recvResult.limited
3043 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}
2990 set *recvRates = recvResult.entries
3046 set *recvRates = recvResult.entries
2991 totalMsgCount <- *totalMessageCount
3047 totalMsgCount <- *totalMessageCount
2992 if totalMsgCount >= maxMessages
3048 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}
2995 -- Recipient count from maintained counts (O(unique_recipients) instead of O(all_messages))
3051 -- Recipient count from maintained counts (O(unique_recipients) instead of O(all_messages))
2996 recipientCounts <- *recipientMsgCounts
3052 recipientCounts <- *recipientMsgCounts
2997 let recipientCount = getRecipientCount to recipientCounts
3053 let recipientCount = getRecipientCount to recipientCounts
2998 if recipientCount >= maxMessagesPerRecipient
3054 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}
3001 -- Hot path: read only recipient's presence bucket
3057 -- Hot path: read only recipient's presence bucket
3002 recipientPresence <- readPresenceBucket (pubkeyBucket to)
3058 recipientPresence <- readPresenceBucket (pubkeyBucket to)
@@ -3025,7 +3081,7 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
3025 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}
3081 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}
3026 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
3082 let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
3027 set *forwards = union forwards cappedFwds)
3083 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})
3029 when result.senderRateLimited
3085 when result.senderRateLimited
3030 (logWarn ("sendMessage: rate limited (" ++ show maxSendRatePerKey ++ "/s per key)"))
3086 (logWarn ("sendMessage: rate limited (" ++ show maxSendRatePerKey ++ "/s per key)"))
3031 when result.globalCapReached
3087 when result.globalCapReached
@@ -3058,9 +3114,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization clientIp
3058 logWarn ("federation: send forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))
3114 logWarn ("federation: send forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))
3059 _ -> yield {}
3115 _ -> yield {}
3060 yield (case result.send of
3116 yield (case result.send of
3061− SendFailed {status} -> respond {status: status, messageId: ""}
3117+ SendFailed {status} -> respond {status: status, messageId: "", retryAfterMs: result.retryAfterMs}
3062− SendDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId}
3118+ SendDuplicate {} -> respond {status: MsDuplicate {}, messageId: effectiveMsgId, retryAfterMs: 0}
3063− SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})
3119+ SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId, retryAfterMs: 0})
3065-- GET /lookup/:pubkey: Look up the best known server for a pubkey.
3121-- GET /lookup/:pubkey: Look up the best known server for a pubkey.
3066-- Returns local server if key is online locally, otherwise the remote with latest expiry.
3122-- Returns local server if key is online locally, otherwise the remote with latest expiry.
@@ -3266,7 +3322,8 @@ handleVerifyToken = \token scope clientIp respond -> do
3266 let match = any pending (\p -> safeCompare p.token token && p.scope == scope && t - p.createdAt < verificationTokenExpiry)
3322 let match = any pending (\p -> safeCompare p.token token && p.scope == scope && t - p.createdAt < verificationTokenExpiry)
3267 when match (set *pendingVerifications = do p <- pending; where not (safeCompare p.token token); yield p)
3323 when match (set *pendingVerifications = do p <- pending; where not (safeCompare p.token token); yield p)
3268 yield match
3324 yield match
3325+ -- Omit scope from log to prevent scope enumeration by attackers probing the endpoint
3269− when (not found) (logWarn ("federation/verify: failed attempt from " ++ clientIp ++ " (scope: " ++ scope ++ ")"))
3326+ when (not found) (logWarn ("federation/verify: failed attempt from " ++ clientIp))
3270 yield (respond {ok: found})
3327 yield (respond {ok: found})
3272-- POST /federation/gossip: Receive presence gossip from a peer server.
3329-- POST /federation/gossip: Receive presence gossip from a peer server.
@@ -3308,6 +3365,14 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
3308 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
3365 let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
3309 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
3366 let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
3310 let deduped = mergeByPubkeyServer [] filteredOnline
3367 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"))
3311 gossipResult <- atomic do
3376 gossipResult <- atomic do
3312 allPresence <- readAllPresence
3377 allPresence <- readAllPresence
3313 tpc <- *totalPresenceCount
3378 tpc <- *totalPresenceCount
@@ -3439,9 +3504,15 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3439 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3504 -- Record forward nonce for replay protection beyond message/ack dedup lifetime
3440 fwdNonces <- *forwardNonces
3505 fwdNonces <- *forwardNonces
3441 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
3506 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
3507+ -- Per-origin eviction: when at capacity, only evict nonces from the
3508+ -- origin server that's causing the overflow. Prevents one peer from
3443− -- for an attacker to force-evict specific nonces via targeted flooding.
3509+ -- force-evicting another peer's nonces via targeted flooding.
3444− set *forwardNonces = if count updatedNonces > maxForwardNonces then (evictSmallest updatedNonces (maxForwardNonces / 10) (\n -> n.seenAt)).kept else updatedNonces
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
3445 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
3516 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}}
3446 case result.forward of
3517 case result.forward of
3447 ForwardRejected {} ->
3518 ForwardRejected {} ->
@@ -3460,7 +3531,9 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3460-- recipient has no local presence and no other pending forwards for that message.
3531-- recipient has no local presence and no other pending forwards for that message.
3461handleFederationAck = \messageIds originServer fromServer requestTimestamp authorization respond -> do
3532handleFederationAck = \messageIds originServer fromServer requestTimestamp authorization respond -> do
3462 t <- now
3533 t <- now
3534+ -- Wrap respond to add ackedCount default for withFederationAuth error paths
3535+ let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0}
3463− withFederationAuth authorization t respond fromServer "ack" requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (do
3536+ withFederationAuth authorization t ackRespond fromServer "ack" requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (do
3464 let dedupedIds = dedupMessageIds messageIds
3537 let dedupedIds = dedupMessageIds messageIds
3465 -- Clear forward entries for the acking server. Extract removed forwards' toKey
3538 -- Clear forward entries for the acking server. Extract removed forwards' toKey
3466 -- so we can reclaim origin messages that are no longer needed.
3539 -- so we can reclaim origin messages that are no longer needed.
@@ -3469,15 +3542,16 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp autho
3469 let removed = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
3542 let removed = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
3470 set *forwards = forwardsExcluding forwards fromServer dedupedIds
3543 set *forwards = forwardsExcluding forwards fromServer dedupedIds
3471 yield removed
3544 yield removed
3545+ let ackedCount = count removedForwards
3472 -- Deferred origin message cleanup: queue for processing in the next prune cycle
3546 -- Deferred origin message cleanup: queue for processing in the next prune cycle
3473 -- after ackGracePeriod elapses. Replaces fork+sleep which accumulated sleeping
3547 -- after ackGracePeriod elapses. Replaces fork+sleep which accumulated sleeping
3474 -- goroutines under high federation ack volume.
3548 -- goroutines under high federation ack volume.
3475− when (count removedForwards > 0) (atomic do
3549+ when (ackedCount > 0) (atomic do
3476 cleanups <- *pendingAckCleanups
3550 cleanups <- *pendingAckCleanups
3477 let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
3551 let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
3478 let combined = union cleanups newCleanups
3552 let combined = union cleanups newCleanups
3479 set *pendingAckCleanups = if count combined > maxPendingAckCleanups then takeRelation maxPendingAckCleanups combined else combined)
3553 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}))
3482-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
3556-- POST /maintenance/drain: Broadcast offline gossip for all local keys (graceful shutdown).
3483-- Call before stopping the server so peers learn keys went offline immediately.
3557-- Call before stopping the server so peers learn keys went offline immediately.
@@ -3555,8 +3629,9 @@ handlePrune = \authorization respond -> do
3555 when authorized (fork (pruneAndRetry t))
3629 when authorized (fork (pruneAndRetry t))
3556 yield (respond {ok: authorized})
3630 yield (respond {ok: authorized})
3558−handleHealth = \respond -> do
3632+handleHealth = \authorization respond -> do
3559 t <- now
3633 t <- now
3634+ let isAuthed = verifyMaintenanceAuth authorization
3560 -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
3635 -- Health checks exempt from global rate limiting (monitoring traffic shouldn't starve real requests)
3561 result <- atomic do
3636 result <- atomic do
3562 lastPrune <- *lastPrunedAt
3637 lastPrune <- *lastPrunedAt
@@ -3586,11 +3661,17 @@ handleHealth = \respond -> do
3586 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
3661 heartbeatOk: warmup || t - lastHb < healthHeartbeatMaxAge,
3587 capacityOk: capacityOk,
3662 capacityOk: capacityOk,
3588 federationOk: fedOk}
3663 federationOk: fedOk}
3589− yield (respond {ok: result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk,
3664+ let allOk = result.pruneOk && result.heartbeatOk && result.capacityOk && result.federationOk
3665+ -- Unauthenticated: return only aggregate ok flag to avoid leaking capacity/federation details
3590− pruneHealthy: result.pruneOk,
3667+ then yield (respond {ok: allOk, pruneHealthy: result.pruneOk,
3591 heartbeatHealthy: result.heartbeatOk,
3668 heartbeatHealthy: result.heartbeatOk,
3592 capacityHealthy: result.capacityOk,
3669 capacityHealthy: result.capacityOk,
3593 federationHealthy: result.federationOk})
3670 federationHealthy: result.federationOk})
3671+ else yield (respond {ok: allOk, pruneHealthy: allOk,
3672+ heartbeatHealthy: allOk,
3673+ capacityHealthy: allOk,
3674+ federationHealthy: allOk})
3595zeroStats = {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 {}}
3676zeroStats = {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 {}}
@@ -3690,7 +3771,7 @@ handleStatsKey = \pubkey authorization respond -> do
3690 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})
3771 yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})
3691 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})
3772 else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})
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: []}
3695-- GET /maintenance/snapshot: Export federation metadata for persistence.
3776-- GET /maintenance/snapshot: Export federation metadata for persistence.
3696-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
3777-- Reads all dedup state (ackedIds, deliveredIds), presence, peer failure records,
@@ -3708,22 +3789,24 @@ handleSnapshot = \authorization respond -> do
3708 d <- *deliveredIds
3789 d <- *deliveredIds
3709 ff <- *forwardFailures
3790 ff <- *forwardFailures
3710 f <- *forwards
3791 f <- *forwards
3792+ fn <- *forwardNonces
3711− yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f}
3793+ yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn}
3712 atomic do set *lastSnapshotExportAt = t
3794 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")
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")
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})
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})
3716-- POST /maintenance/restore: Import a previously-exported snapshot.
3798-- POST /maintenance/restore: Import a previously-exported snapshot.
3717-- Requires drain mode to prevent conflicts with live state. Merges imported data
3799-- Requires drain mode to prevent conflicts with live state. Merges imported data
3718-- with existing state (dedup entries are unioned, presence is merged by freshness).
3800-- with existing state (dedup entries are unioned, presence is merged by freshness).
3719-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
3801-- After restore, call /maintenance/undrain to resume, then /maintenance/prune to
3720-- trigger federation forward creation from messages + restored presence.
3802-- 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
3722 let authorized = verifyMaintenanceAuth authorization
3804 let authorized = verifyMaintenanceAuth authorization
3723 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
3805 if not authorized then yield (respond {ok: False {}, error: "unauthorized", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
3725− -- Version 0 = pre-versioning snapshot (accept for backwards compatibility)
3807+ -- Version 0 = pre-versioning snapshot, version 1 = pre-forwardNonces (accept both for backwards compatibility;
3808+ -- missing forwardNonces field will default to empty list)
3726− if version != 0 && version != snapshotVersion then do
3809+ if version != 0 && version != 1 && version != snapshotVersion then do
3727 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
3810 logError ("restore: incompatible snapshot version " ++ show version ++ " (expected " ++ show snapshotVersion ++ ")")
3728 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
3811 yield (respond {ok: False {}, error: "incompatible_version", presenceCount: 0, ackedIdsCount: 0, deliveredIdsCount: 0})
@@ -3771,6 +3854,12 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
3771 let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) forwards validForwards
3854 let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) forwards validForwards
3772 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
3855 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
3773 set *forwards = cappedForwards
3856 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
3774 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
3863 yield {presenceCount: count cappedPresence, ackedIdsCount: count cappedAcked, deliveredIdsCount: count cappedDelivered}
3775 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
3864 logInfo ("restore: imported " ++ show result.presenceCount ++ " presence, " ++ show result.ackedIdsCount ++ " ackedIds, " ++ show result.deliveredIdsCount ++ " deliveredIds")
3776 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
3865 yield (respond {ok: True {}, error: "", presenceCount: result.presenceCount, ackedIdsCount: result.ackedIdsCount, deliveredIdsCount: result.deliveredIdsCount})
@@ -3826,16 +3915,44 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
3826 retryEvictedAcks result.evictedEntries result.pendingAcks))
3915 retryEvictedAcks result.evictedEntries result.pendingAcks))
3827 yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount}))
3916 yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount}))
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
3923+ globalRateLimited <- checkGlobalRate t
3924+ if globalRateLimited then yield (respond {messages: [], authorized: False {}, count: 0})
3926+ withSessionAuth authorization t clientIp
3927+ (yield (respond {messages: [], authorized: False {}, count: 0}))
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})
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}))
3829-- POST /messages/batch: Send multiple messages in one request.
3946-- POST /messages/batch: Send multiple messages in one request.
3830-- Auth checked once; each message processed independently with its own rate limiting.
3947-- Auth checked once; each message processed independently with its own rate limiting.
3831-- Partial success is possible — check per-message status in results.
3948-- Partial success is possible — check per-message status in results.
3832handleSendBatch = \messages authorization clientIp respond -> do
3949handleSendBatch = \messages authorization clientIp respond -> do
3833 t <- now
3950 t <- now
3834 globalRateLimited <- checkGlobalRate t
3951 globalRateLimited <- checkGlobalRate t
3835− if globalRateLimited then yield (respond {results: [], authorized: False {}})
3952+ if globalRateLimited then yield (respond {results: [], authorized: False {}, retryAfterMs: 0})
3837 withSessionAuth authorization t clientIp
3954 withSessionAuth authorization t clientIp
3838− (yield (respond {results: [], authorized: False {}}))
3955+ (yield (respond {results: [], authorized: False {}, retryAfterMs: 0}))
3839 (\senderKey -> do
3956 (\senderKey -> do
3840 let capped = takeRelation maxMessageIds messages
3957 let capped = takeRelation maxMessageIds messages
3841 results <- sendBatchInner senderKey capped t []
3958 results <- sendBatchInner senderKey capped t []
@@ -3868,7 +3985,9 @@ handleSendBatch = \messages authorization clientIp respond -> do
3868 removeForward r.messageId server
3985 removeForward r.messageId server
3869 clearFailure server
3986 clearFailure server
3870 else logWarn ("federation: batch forward to " ++ server ++ " rejected for " ++ r.messageId ++ ": " ++ response.error)))))))
3987 else logWarn ("federation: batch forward to " ++ server ++ " rejected for " ++ r.messageId ++ ": " ++ response.error)))))))
3988+ -- Aggregate retryAfterMs: use the maximum across all rate-limited messages
3989+ let batchRetryMs = fold (\acc r -> max acc r.retryAfterMs) 0 results
3871− yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}}))
3990+ yield (respond {results: do r <- results; yield {messageId: r.messageId, status: r.status}, authorized: True {}, retryAfterMs: batchRetryMs}))
3873sendBatchInner = \senderKey messages t acc ->
3992sendBatchInner = \senderKey messages t acc ->
3874 let split = splitAtN messages 1 in
3993 let split = splitAtN messages 1 in
@@ -3882,28 +4001,28 @@ sendBatchInner = \senderKey messages t acc ->
3882 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
4001 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
3883 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
4002 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
3884 result <- if t - msg.timestamp > messageMaxAge
4003 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}
3886 else if senderKey == msg.to
4005 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}
3889 let clampedTs = clampTimestamp msg.timestamp t
4008 let clampedTs = clampTimestamp msg.timestamp t
3890 sendResult <- atomic do
4009 sendResult <- atomic do
3891 draining <- *draining
4010 draining <- *draining
3892− if draining then yield {status: MsDraining {}, fwdServer: "", needsFwd: False {}}
4011+ if draining then yield {status: MsDraining {}, fwdServer: "", needsFwd: False {}, retryMs: 0}
3894 let msgBucket = pubkeyBucket msg.to
4013 let msgBucket = pubkeyBucket msg.to
3895 bucket <- readMessageBucket msgBucket
4014 bucket <- readMessageBucket msgBucket
3896 ackedIds <- *ackedIds
4015 ackedIds <- *ackedIds
3897 deliveredIds <- *deliveredIds
4016 deliveredIds <- *deliveredIds
3898 let isDup = checkMessageDup bucket effectiveMsgId msg.to || (effectiveMsgId != "" && (hasMessage effectiveMsgId msg.to ackedIds || hasMessage effectiveMsgId msg.to deliveredIds))
4017 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}
3901 sendRates <- *sendRates
4020 sendRates <- *sendRates
3902 let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
4021 let rateResult = checkRateWindow sendRates (\r -> r.pubkey == senderKey)
3903 (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})
4022 (\ts -> {pubkey: senderKey, windowStart: ts, count: 1})
3904 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
4023 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3905 maxSendRatePerKey sendRateWindowMs maxSendRateEntries t
4024 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}
3908 set *sendRates = rateResult.entries
4027 set *sendRates = rateResult.entries
3909 recvRates <- *recvRates
4028 recvRates <- *recvRates
@@ -3911,7 +4030,7 @@ sendBatchInner = \senderKey messages t acc ->
3911 (\ts -> {pubkey: msg.to, windowStart: ts, count: 1})
4030 (\ts -> {pubkey: msg.to, windowStart: ts, count: 1})
3912 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
4031 (\r -> {pubkey: r.pubkey, windowStart: r.windowStart, count: r.count + 1})
3913 maxRecvRatePerKey recvRateWindowMs maxRecvRateEntries t
4032 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}
3916 set *recvRates = recvResult.entries
4035 set *recvRates = recvResult.entries
3917 -- Per-sender-per-recipient pair rate limiting (prevents one sender filling a recipient's mailbox)
4036 -- Per-sender-per-recipient pair rate limiting (prevents one sender filling a recipient's mailbox)
@@ -3920,15 +4039,15 @@ sendBatchInner = \senderKey messages t acc ->
3920 (\ts -> {sender: senderKey, recipient: msg.to, windowStart: ts, count: 1})
4039 (\ts -> {sender: senderKey, recipient: msg.to, windowStart: ts, count: 1})
3921 (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
4040 (\r -> {sender: r.sender, recipient: r.recipient, windowStart: r.windowStart, count: r.count + 1})
3922 maxPairRatePerPair pairRateWindowMs maxPairRateEntries t
4041 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}
3925 set *pairRates = pairResult.entries
4044 set *pairRates = pairResult.entries
3926 totalMsgCount <- *totalMessageCount
4045 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}
3929 recipientCounts <- *recipientMsgCounts
4048 recipientCounts <- *recipientMsgCounts
3930 let recipientCount = getRecipientCount msg.to recipientCounts
4049 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}
3933 recipientPresence <- readPresenceBucket (pubkeyBucket msg.to)
4052 recipientPresence <- readPresenceBucket (pubkeyBucket msg.to)
3934 seqCounter <- *seqCounter
4053 seqCounter <- *seqCounter
@@ -3952,13 +4071,13 @@ sendBatchInner = \senderKey messages t acc ->
3952 let delivery = if localOnline then MsDelivered {}
4071 let delivery = if localOnline then MsDelivered {}
3953 else if federationEnabled && count remoteServers > 0 then MsFederated {}
4072 else if federationEnabled && count remoteServers > 0 then MsFederated {}
3954 else MsQueued {}
4073 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}
3956 -- Log rate limiting and capacity warnings (matching handleSendMessage)
4075 -- Log rate limiting and capacity warnings (matching handleSendMessage)
3957 case sendResult.status of
4076 case sendResult.status of
3958 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
4077 MsRateLimited {} -> logWarn ("sendBatch: rate limited for " ++ effectiveMsgId)
3959 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")
4078 MsRejected {} -> logWarn ("sendBatch: rejected for " ++ effectiveMsgId ++ " (capacity or mailbox full)")
3960 _ -> yield {}
4079 _ -> 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}
3962 sendBatchInner senderKey split.rest t (union acc [result])
4081 sendBatchInner senderKey split.rest t (union acc [result])
3964serve = \req -> case req of
4083serve = \req -> case req of
@@ -3971,6 +4090,7 @@ serve = \req -> case req of
3971 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
4090 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
3972 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
4091 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
3973 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
4092 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
4093+ ListMessages {authorization, clientIp, respond} -> handleListMessages authorization clientIp respond
3974 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
4094 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
3975 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
4095 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
3976 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp authorization respond
4096 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
3980 Drain {authorization, respond} -> handleDrain authorization respond
4100 Drain {authorization, respond} -> handleDrain authorization respond
3981 Undrain {authorization, respond} -> handleUndrain authorization respond
4101 Undrain {authorization, respond} -> handleUndrain authorization respond
3982 Snapshot {authorization, respond} -> handleSnapshot authorization respond
4102 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
4103+ Restore {version, presence, ackedIds, ackOverflow, deliveredIds, forwardFailures, forwards, forwardNonces, authorization, respond} -> handleRestore version presence ackedIds ackOverflow deliveredIds forwardFailures forwards forwardNonces authorization respond
3984− Health {respond} -> handleHealth respond
4104+ Health {authorization, respond} -> handleHealth authorization respond
3985 Stats {authorization, respond} -> handleStats authorization respond
4105 Stats {authorization, respond} -> handleStats authorization respond
3986 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
4106 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
3987 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
4107 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
@@ -3992,6 +4112,9 @@ serve = \req -> case req of
3993-- Background loop: periodically prunes expired state and retries pending federation ops.
4113-- Background loop: periodically prunes expired state and retries pending federation ops.
3994-- Jitter prevents thundering herd when multiple servers restart simultaneously.
4114-- 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.
3995backgroundPrune = do
4118backgroundPrune = do
3996 jitter <- randomInt 5000
4119 jitter <- randomInt 5000
3997 sleep (backgroundPruneInterval + jitter)
4120 sleep (backgroundPruneInterval + jitter)