@@ -431,10 +431,43 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
431-- Monotonic counter ensuring unique receivedAt ordering even within the same millisecond
431-- Monotonic counter ensuring unique receivedAt ordering even within the same millisecond
432*seqCounter : Timestamp
432*seqCounter : Timestamp
434+-- Forwards sharded into 16 buckets by pubkeyBucket(toKey).
435+-- Hot-path operations (sendMessageCore dedup, ackPolledMessages) scan one bucket.
436+*fwdBucket0 : [ForwardEntry]
437+*fwdBucket1 : [ForwardEntry]
438+*fwdBucket2 : [ForwardEntry]
439+*fwdBucket3 : [ForwardEntry]
440+*fwdBucket4 : [ForwardEntry]
441+*fwdBucket5 : [ForwardEntry]
442+*fwdBucket6 : [ForwardEntry]
443+*fwdBucket7 : [ForwardEntry]
444+*fwdBucket8 : [ForwardEntry]
445+*fwdBucket9 : [ForwardEntry]
446+*fwdBucket10 : [ForwardEntry]
447+*fwdBucket11 : [ForwardEntry]
448+*fwdBucket12 : [ForwardEntry]
449+*fwdBucket13 : [ForwardEntry]
450+*fwdBucket14 : [ForwardEntry]
434−*forwards : [ForwardEntry]
451+*fwdBucket15 : [ForwardEntry]
436−-- Messages acked by local recipients, pending federation ack to origin server
453+-- Messages acked by local recipients, pending federation ack to origin server.
454+-- Sharded into 16 buckets by pubkeyBucket(toKey) for O(n/16) dedup in sendMessageCore.
455+*ackedBucket0 : [AckedEntry]
456+*ackedBucket1 : [AckedEntry]
457+*ackedBucket2 : [AckedEntry]
458+*ackedBucket3 : [AckedEntry]
459+*ackedBucket4 : [AckedEntry]
460+*ackedBucket5 : [AckedEntry]
461+*ackedBucket6 : [AckedEntry]
462+*ackedBucket7 : [AckedEntry]
463+*ackedBucket8 : [AckedEntry]
464+*ackedBucket9 : [AckedEntry]
465+*ackedBucket10 : [AckedEntry]
466+*ackedBucket11 : [AckedEntry]
467+*ackedBucket12 : [AckedEntry]
468+*ackedBucket13 : [AckedEntry]
469+*ackedBucket14 : [AckedEntry]
437−*ackedIds : [AckedEntry]
470+*ackedBucket15 : [AckedEntry]
439*lastPrunedAt : Timestamp
472*lastPrunedAt : Timestamp
@@ -477,8 +510,24 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
477-- Overflow buffer: acks evicted from ackedIds before they could be sent to origin
510-- Overflow buffer: acks evicted from ackedIds before they could be sent to origin
478*ackOverflow : [AckedEntry]
511*ackOverflow : [AckedEntry]
480−-- Final dedup set: messages confirmed delivered (ack sent or overflow drained)
513+-- Final dedup set: messages confirmed delivered (ack sent or overflow drained).
514+-- Sharded into 16 buckets by pubkeyBucket(toKey) for O(n/16) dedup in sendMessageCore.
515+*delivBucket0 : [DeliveredEntry]
516+*delivBucket1 : [DeliveredEntry]
517+*delivBucket2 : [DeliveredEntry]
518+*delivBucket3 : [DeliveredEntry]
519+*delivBucket4 : [DeliveredEntry]
520+*delivBucket5 : [DeliveredEntry]
521+*delivBucket6 : [DeliveredEntry]
522+*delivBucket7 : [DeliveredEntry]
523+*delivBucket8 : [DeliveredEntry]
524+*delivBucket9 : [DeliveredEntry]
525+*delivBucket10 : [DeliveredEntry]
526+*delivBucket11 : [DeliveredEntry]
527+*delivBucket12 : [DeliveredEntry]
528+*delivBucket13 : [DeliveredEntry]
529+*delivBucket14 : [DeliveredEntry]
481−*deliveredIds : [DeliveredEntry]
530+*delivBucket15 : [DeliveredEntry]
483*federationRates : [FederationRateEntry]
532*federationRates : [FederationRateEntry]
@@ -575,8 +624,24 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}
575-- scope binds the token to a specific request type (gossip/forward/ack) to prevent cross-use
624-- scope binds the token to a specific request type (gossip/forward/ack) to prevent cross-use
576*pendingVerifications : [{token: Text, createdAt: Timestamp, scope: Text}]
625*pendingVerifications : [{token: Text, createdAt: Timestamp, scope: Text}]
578−-- Forward nonce dedup: (messageId, toKey, originServer, seenAt) for replay protection beyond message lifetime
627+-- Forward nonce dedup sharded into 16 buckets by pubkeyBucket(toKey).
628+-- Reduces hot-path scan in handleForwardMessage from O(100k) to O(6k).
629+*nonceBucket0 : [ForwardNonce]
630+*nonceBucket1 : [ForwardNonce]
631+*nonceBucket2 : [ForwardNonce]
632+*nonceBucket3 : [ForwardNonce]
633+*nonceBucket4 : [ForwardNonce]
634+*nonceBucket5 : [ForwardNonce]
635+*nonceBucket6 : [ForwardNonce]
636+*nonceBucket7 : [ForwardNonce]
637+*nonceBucket8 : [ForwardNonce]
638+*nonceBucket9 : [ForwardNonce]
639+*nonceBucket10 : [ForwardNonce]
640+*nonceBucket11 : [ForwardNonce]
641+*nonceBucket12 : [ForwardNonce]
642+*nonceBucket13 : [ForwardNonce]
643+*nonceBucket14 : [ForwardNonce]
579−*forwardNonces : [ForwardNonce]
644+*nonceBucket15 : [ForwardNonce]
581-- Federation verify per-IP rate limiting (tighter than general IP rate)
646-- Federation verify per-IP rate limiting (tighter than general IP rate)
582*verifyIpRates : [IpRateEntry]
647*verifyIpRates : [IpRateEntry]
@@ -759,6 +824,10 @@ route MaintenanceApi where
759 POST {pubkey: PubkeyHex} /maintenance/sessions/revoke-all headers {authorization: Text}
824 POST {pubkey: PubkeyHex} /maintenance/sessions/revoke-all headers {authorization: Text}
760 -> {ok: Bool, revokedCount: Int}
825 -> {ok: Bool, revokedCount: Int}
761 = AdminRevokeSessions
826 = AdminRevokeSessions
827+ -- Purge all pending messages for a specific pubkey (inbox bombing response)
828+ POST {pubkey: PubkeyHex} /maintenance/messages/purge headers {authorization: Text}
829+ -> {ok: Bool, purgedCount: Int}
830+ = AdminPurgeMessages
762 -- Read current effective rate limits (compile-time defaults + runtime overrides)
831 -- Read current effective rate limits (compile-time defaults + runtime overrides)
763 GET /maintenance/rate-limits headers {authorization: Text}
832 GET /maintenance/rate-limits headers {authorization: Text}
764 -> {ok: Bool, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int}
833 -> {ok: Bool, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int}
@@ -785,7 +854,7 @@ route StatsApi where
785 -> {peers: [PeerStatus], ok: Bool}
854 -> {peers: [PeerStatus], ok: Bool}
786 = StatsPeers
855 = StatsPeers
787 GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text}
856 GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text}
788− -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, activePolls: Int, ok: Bool}
857+ -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, forwardDetails: [{server: Text, retries: Int, createdAt: Timestamp}], activePolls: Int, ok: Bool}
789 = StatsKey
858 = StatsKey
791route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | MaintenanceApi | HealthApi | StatsApi
860route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | MaintenanceApi | HealthApi | StatsApi
@@ -1452,6 +1521,17 @@ transformAllPresenceBuckets = \f -> do
1452 b14 <- *presenceBucket14; set *presenceBucket14 = f b14
1521 b14 <- *presenceBucket14; set *presenceBucket14 = f b14
1453 b15 <- *presenceBucket15; set *presenceBucket15 = f b15
1522 b15 <- *presenceBucket15; set *presenceBucket15 = f b15
1524+-- Scan presence buckets one at a time for local entries nearing expiry.
1525+-- Avoids materializing all 16 buckets when few keys are near expiry.
1526+findNearExpiryPresence : Int -> Timestamp -> IO [PresenceEntry]
1527+findNearExpiryPresence = \b t ->
1528+ if b > 15 then yield []
1530+ bucket <- readPresenceBucket b
1531+ let nearExpiry = do p <- bucket; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p
1532+ rest <- findNearExpiryPresence (b + 1) t
1533+ yield (union nearExpiry rest)
1455-- ============================================
1535-- ============================================
1456-- Message Bucket Helpers
1536-- Message Bucket Helpers
1457-- ============================================
1537-- ============================================
@@ -1721,6 +1801,343 @@ checkMessageDup : [Message] -> Text -> PubkeyHex -> Bool
1721checkMessageDup = \messages messageId toKey ->
1801checkMessageDup = \messages messageId toKey ->
1722 messageId != "" && any messages (\m -> m.messageId == messageId && m.toKey == toKey)
1802 messageId != "" && any messages (\m -> m.messageId == messageId && m.toKey == toKey)
1804+-- ============================================
1805+-- Forward Nonce Bucket Helpers
1806+-- ============================================
1807+-- Forward nonces sharded into 16 buckets by pubkeyBucket(toKey).
1808+-- Hot-path (handleForwardMessage, sendMessageCore) scans one bucket (~1/16 of nonces).
1810+readNonceBucket : Int -> IO [ForwardNonce]
1811+readNonceBucket = \b ->
1812+ if b == 0 then do s <- *nonceBucket0; yield s
1813+ else if b == 1 then do s <- *nonceBucket1; yield s
1814+ else if b == 2 then do s <- *nonceBucket2; yield s
1815+ else if b == 3 then do s <- *nonceBucket3; yield s
1816+ else if b == 4 then do s <- *nonceBucket4; yield s
1817+ else if b == 5 then do s <- *nonceBucket5; yield s
1818+ else if b == 6 then do s <- *nonceBucket6; yield s
1819+ else if b == 7 then do s <- *nonceBucket7; yield s
1820+ else if b == 8 then do s <- *nonceBucket8; yield s
1821+ else if b == 9 then do s <- *nonceBucket9; yield s
1822+ else if b == 10 then do s <- *nonceBucket10; yield s
1823+ else if b == 11 then do s <- *nonceBucket11; yield s
1824+ else if b == 12 then do s <- *nonceBucket12; yield s
1825+ else if b == 13 then do s <- *nonceBucket13; yield s
1826+ else if b == 14 then do s <- *nonceBucket14; yield s
1827+ else do s <- *nonceBucket15; yield s
1829+writeNonceBucket : Int -> [ForwardNonce] -> IO ()
1830+writeNonceBucket = \b nonces ->
1831+ if b == 0 then set *nonceBucket0 = nonces
1832+ else if b == 1 then set *nonceBucket1 = nonces
1833+ else if b == 2 then set *nonceBucket2 = nonces
1834+ else if b == 3 then set *nonceBucket3 = nonces
1835+ else if b == 4 then set *nonceBucket4 = nonces
1836+ else if b == 5 then set *nonceBucket5 = nonces
1837+ else if b == 6 then set *nonceBucket6 = nonces
1838+ else if b == 7 then set *nonceBucket7 = nonces
1839+ else if b == 8 then set *nonceBucket8 = nonces
1840+ else if b == 9 then set *nonceBucket9 = nonces
1841+ else if b == 10 then set *nonceBucket10 = nonces
1842+ else if b == 11 then set *nonceBucket11 = nonces
1843+ else if b == 12 then set *nonceBucket12 = nonces
1844+ else if b == 13 then set *nonceBucket13 = nonces
1845+ else if b == 14 then set *nonceBucket14 = nonces
1846+ else set *nonceBucket15 = nonces
1848+readAllNonces : IO [ForwardNonce]
1850+ n0 <- *nonceBucket0
1851+ n1 <- *nonceBucket1
1852+ n2 <- *nonceBucket2
1853+ n3 <- *nonceBucket3
1854+ n4 <- *nonceBucket4
1855+ n5 <- *nonceBucket5
1856+ n6 <- *nonceBucket6
1857+ n7 <- *nonceBucket7
1858+ n8 <- *nonceBucket8
1859+ n9 <- *nonceBucket9
1860+ n10 <- *nonceBucket10
1861+ n11 <- *nonceBucket11
1862+ n12 <- *nonceBucket12
1863+ n13 <- *nonceBucket13
1864+ n14 <- *nonceBucket14
1865+ n15 <- *nonceBucket15
1866+ yield (union n0 (union n1 (union n2 (union n3 (union n4 (union n5 (union n6 (union n7 (union n8 (union n9 (union n10 (union n11 (union n12 (union n13 (union n14 n15)))))))))))))))
1868+countAllNonces : Int -> IO Int
1869+countAllNonces = \b ->
1870+ if b > 15 then yield 0
1872+ bucket <- readNonceBucket b
1873+ rest <- countAllNonces (b + 1)
1874+ yield (count bucket + rest)
1876+-- Write nonces to all 16 buckets via single-pass partitioning
1877+writeAllNonces : [ForwardNonce] -> IO ()
1878+writeAllNonces = \nonces -> do
1879+ let partitioned = fold (\acc n ->
1880+ let b = pubkeyBucket n.toKey in
1881+ {b0: if b == 0 then union [n] acc.b0 else acc.b0,
1882+ b1: if b == 1 then union [n] acc.b1 else acc.b1,
1883+ b2: if b == 2 then union [n] acc.b2 else acc.b2,
1884+ b3: if b == 3 then union [n] acc.b3 else acc.b3,
1885+ b4: if b == 4 then union [n] acc.b4 else acc.b4,
1886+ b5: if b == 5 then union [n] acc.b5 else acc.b5,
1887+ b6: if b == 6 then union [n] acc.b6 else acc.b6,
1888+ b7: if b == 7 then union [n] acc.b7 else acc.b7,
1889+ b8: if b == 8 then union [n] acc.b8 else acc.b8,
1890+ b9: if b == 9 then union [n] acc.b9 else acc.b9,
1891+ b10: if b == 10 then union [n] acc.b10 else acc.b10,
1892+ b11: if b == 11 then union [n] acc.b11 else acc.b11,
1893+ b12: if b == 12 then union [n] acc.b12 else acc.b12,
1894+ b13: if b == 13 then union [n] acc.b13 else acc.b13,
1895+ b14: if b == 14 then union [n] acc.b14 else acc.b14,
1896+ b15: if b == 15 then union [n] acc.b15 else acc.b15})
1897+ {b0: [], b1: [], b2: [], b3: [], b4: [], b5: [], b6: [], b7: [], b8: [], b9: [], b10: [], b11: [], b12: [], b13: [], b14: [], b15: []} nonces
1898+ writeNonceBucket 0 partitioned.b0
1899+ writeNonceBucket 1 partitioned.b1
1900+ writeNonceBucket 2 partitioned.b2
1901+ writeNonceBucket 3 partitioned.b3
1902+ writeNonceBucket 4 partitioned.b4
1903+ writeNonceBucket 5 partitioned.b5
1904+ writeNonceBucket 6 partitioned.b6
1905+ writeNonceBucket 7 partitioned.b7
1906+ writeNonceBucket 8 partitioned.b8
1907+ writeNonceBucket 9 partitioned.b9
1908+ writeNonceBucket 10 partitioned.b10
1909+ writeNonceBucket 11 partitioned.b11
1910+ writeNonceBucket 12 partitioned.b12
1911+ writeNonceBucket 13 partitioned.b13
1912+ writeNonceBucket 14 partitioned.b14
1913+ writeNonceBucket 15 partitioned.b15
1915+-- Prune nonces across all 16 buckets
1916+pruneAllNonceBuckets : Int -> Timestamp -> IO ()
1917+pruneAllNonceBuckets = \b t ->
1918+ if b > 15 then yield {}
1920+ bucket <- readNonceBucket b
1921+ writeNonceBucket b (do n <- bucket; where t - n.seenAt < forwardNonceRetention; yield n)
1922+ pruneAllNonceBuckets (b + 1) t
1924+-- ============================================
1925+-- Forward Bucket Helpers
1926+-- ============================================
1927+-- Forwards sharded into 16 buckets by pubkeyBucket(toKey).
1929+readFwdBucket : Int -> IO [ForwardEntry]
1930+readFwdBucket = \b ->
1931+ if b == 0 then do s <- *fwdBucket0; yield s
1932+ else if b == 1 then do s <- *fwdBucket1; yield s
1933+ else if b == 2 then do s <- *fwdBucket2; yield s
1934+ else if b == 3 then do s <- *fwdBucket3; yield s
1935+ else if b == 4 then do s <- *fwdBucket4; yield s
1936+ else if b == 5 then do s <- *fwdBucket5; yield s
1937+ else if b == 6 then do s <- *fwdBucket6; yield s
1938+ else if b == 7 then do s <- *fwdBucket7; yield s
1939+ else if b == 8 then do s <- *fwdBucket8; yield s
1940+ else if b == 9 then do s <- *fwdBucket9; yield s
1941+ else if b == 10 then do s <- *fwdBucket10; yield s
1942+ else if b == 11 then do s <- *fwdBucket11; yield s
1943+ else if b == 12 then do s <- *fwdBucket12; yield s
1944+ else if b == 13 then do s <- *fwdBucket13; yield s
1945+ else if b == 14 then do s <- *fwdBucket14; yield s
1946+ else do s <- *fwdBucket15; yield s
1948+writeFwdBucket : Int -> [ForwardEntry] -> IO ()
1949+writeFwdBucket = \b fwds ->
1950+ if b == 0 then set *fwdBucket0 = fwds
1951+ else if b == 1 then set *fwdBucket1 = fwds
1952+ else if b == 2 then set *fwdBucket2 = fwds
1953+ else if b == 3 then set *fwdBucket3 = fwds
1954+ else if b == 4 then set *fwdBucket4 = fwds
1955+ else if b == 5 then set *fwdBucket5 = fwds
1956+ else if b == 6 then set *fwdBucket6 = fwds
1957+ else if b == 7 then set *fwdBucket7 = fwds
1958+ else if b == 8 then set *fwdBucket8 = fwds
1959+ else if b == 9 then set *fwdBucket9 = fwds
1960+ else if b == 10 then set *fwdBucket10 = fwds
1961+ else if b == 11 then set *fwdBucket11 = fwds
1962+ else if b == 12 then set *fwdBucket12 = fwds
1963+ else if b == 13 then set *fwdBucket13 = fwds
1964+ else if b == 14 then set *fwdBucket14 = fwds
1965+ else set *fwdBucket15 = fwds
1967+readAllForwards : IO [ForwardEntry]
1968+readAllForwards = do
1979+ f10 <- *fwdBucket10
1980+ f11 <- *fwdBucket11
1981+ f12 <- *fwdBucket12
1982+ f13 <- *fwdBucket13
1983+ f14 <- *fwdBucket14
1984+ f15 <- *fwdBucket15
1985+ yield (union f0 (union f1 (union f2 (union f3 (union f4 (union f5 (union f6 (union f7 (union f8 (union f9 (union f10 (union f11 (union f12 (union f13 (union f14 f15)))))))))))))))
1987+countAllForwards : Int -> IO Int
1988+countAllForwards = \b ->
1989+ if b > 15 then yield 0
1991+ bucket <- readFwdBucket b
1992+ rest <- countAllForwards (b + 1)
1993+ yield (count bucket + rest)
1995+writeAllForwards : [ForwardEntry] -> IO ()
1996+writeAllForwards = \fwds -> do
1997+ let partitioned = fold (\acc f ->
1998+ let b = pubkeyBucket f.toKey in
1999+ {b0: if b == 0 then union [f] acc.b0 else acc.b0,
2000+ b1: if b == 1 then union [f] acc.b1 else acc.b1,
2001+ b2: if b == 2 then union [f] acc.b2 else acc.b2,
2002+ b3: if b == 3 then union [f] acc.b3 else acc.b3,
2003+ b4: if b == 4 then union [f] acc.b4 else acc.b4,
2004+ b5: if b == 5 then union [f] acc.b5 else acc.b5,
2005+ b6: if b == 6 then union [f] acc.b6 else acc.b6,
2006+ b7: if b == 7 then union [f] acc.b7 else acc.b7,
2007+ b8: if b == 8 then union [f] acc.b8 else acc.b8,
2008+ b9: if b == 9 then union [f] acc.b9 else acc.b9,
2009+ b10: if b == 10 then union [f] acc.b10 else acc.b10,
2010+ b11: if b == 11 then union [f] acc.b11 else acc.b11,
2011+ b12: if b == 12 then union [f] acc.b12 else acc.b12,
2012+ b13: if b == 13 then union [f] acc.b13 else acc.b13,
2013+ b14: if b == 14 then union [f] acc.b14 else acc.b14,
2014+ b15: if b == 15 then union [f] acc.b15 else acc.b15})
2015+ {b0: [], b1: [], b2: [], b3: [], b4: [], b5: [], b6: [], b7: [], b8: [], b9: [], b10: [], b11: [], b12: [], b13: [], b14: [], b15: []} fwds
2016+ writeFwdBucket 0 partitioned.b0
2017+ writeFwdBucket 1 partitioned.b1
2018+ writeFwdBucket 2 partitioned.b2
2019+ writeFwdBucket 3 partitioned.b3
2020+ writeFwdBucket 4 partitioned.b4
2021+ writeFwdBucket 5 partitioned.b5
2022+ writeFwdBucket 6 partitioned.b6
2023+ writeFwdBucket 7 partitioned.b7
2024+ writeFwdBucket 8 partitioned.b8
2025+ writeFwdBucket 9 partitioned.b9
2026+ writeFwdBucket 10 partitioned.b10
2027+ writeFwdBucket 11 partitioned.b11
2028+ writeFwdBucket 12 partitioned.b12
2029+ writeFwdBucket 13 partitioned.b13
2030+ writeFwdBucket 14 partitioned.b14
2031+ writeFwdBucket 15 partitioned.b15
2033+-- ============================================
2034+-- AckedIds Bucket Helpers
2035+-- ============================================
2036+readAckedBucket : Int -> IO [AckedEntry]
2037+readAckedBucket = \b ->
2038+ if b == 0 then do s <- *ackedBucket0; yield s else if b == 1 then do s <- *ackedBucket1; yield s
2039+ else if b == 2 then do s <- *ackedBucket2; yield s else if b == 3 then do s <- *ackedBucket3; yield s
2040+ else if b == 4 then do s <- *ackedBucket4; yield s else if b == 5 then do s <- *ackedBucket5; yield s
2041+ else if b == 6 then do s <- *ackedBucket6; yield s else if b == 7 then do s <- *ackedBucket7; yield s
2042+ else if b == 8 then do s <- *ackedBucket8; yield s else if b == 9 then do s <- *ackedBucket9; yield s
2043+ else if b == 10 then do s <- *ackedBucket10; yield s else if b == 11 then do s <- *ackedBucket11; yield s
2044+ else if b == 12 then do s <- *ackedBucket12; yield s else if b == 13 then do s <- *ackedBucket13; yield s
2045+ else if b == 14 then do s <- *ackedBucket14; yield s else do s <- *ackedBucket15; yield s
2047+writeAckedBucket : Int -> [AckedEntry] -> IO ()
2048+writeAckedBucket = \b entries ->
2049+ if b == 0 then set *ackedBucket0 = entries else if b == 1 then set *ackedBucket1 = entries
2050+ else if b == 2 then set *ackedBucket2 = entries else if b == 3 then set *ackedBucket3 = entries
2051+ else if b == 4 then set *ackedBucket4 = entries else if b == 5 then set *ackedBucket5 = entries
2052+ else if b == 6 then set *ackedBucket6 = entries else if b == 7 then set *ackedBucket7 = entries
2053+ else if b == 8 then set *ackedBucket8 = entries else if b == 9 then set *ackedBucket9 = entries
2054+ else if b == 10 then set *ackedBucket10 = entries else if b == 11 then set *ackedBucket11 = entries
2055+ else if b == 12 then set *ackedBucket12 = entries else if b == 13 then set *ackedBucket13 = entries
2056+ else if b == 14 then set *ackedBucket14 = entries else set *ackedBucket15 = entries
2058+readAllAcked : IO [AckedEntry]
2060+ a0 <- *ackedBucket0; a1 <- *ackedBucket1; a2 <- *ackedBucket2; a3 <- *ackedBucket3
2061+ a4 <- *ackedBucket4; a5 <- *ackedBucket5; a6 <- *ackedBucket6; a7 <- *ackedBucket7
2062+ a8 <- *ackedBucket8; a9 <- *ackedBucket9; a10 <- *ackedBucket10; a11 <- *ackedBucket11
2063+ a12 <- *ackedBucket12; a13 <- *ackedBucket13; a14 <- *ackedBucket14; a15 <- *ackedBucket15
2064+ yield (union a0 (union a1 (union a2 (union a3 (union a4 (union a5 (union a6 (union a7 (union a8 (union a9 (union a10 (union a11 (union a12 (union a13 (union a14 a15)))))))))))))))
2066+writeAllAcked : [AckedEntry] -> IO ()
2067+writeAllAcked = \entries -> do
2068+ let p = fold (\acc e -> let b = pubkeyBucket e.toKey in
2069+ {b0: if b == 0 then union [e] acc.b0 else acc.b0, b1: if b == 1 then union [e] acc.b1 else acc.b1,
2070+ b2: if b == 2 then union [e] acc.b2 else acc.b2, b3: if b == 3 then union [e] acc.b3 else acc.b3,
2071+ b4: if b == 4 then union [e] acc.b4 else acc.b4, b5: if b == 5 then union [e] acc.b5 else acc.b5,
2072+ b6: if b == 6 then union [e] acc.b6 else acc.b6, b7: if b == 7 then union [e] acc.b7 else acc.b7,
2073+ b8: if b == 8 then union [e] acc.b8 else acc.b8, b9: if b == 9 then union [e] acc.b9 else acc.b9,
2074+ b10: if b == 10 then union [e] acc.b10 else acc.b10, b11: if b == 11 then union [e] acc.b11 else acc.b11,
2075+ b12: if b == 12 then union [e] acc.b12 else acc.b12, b13: if b == 13 then union [e] acc.b13 else acc.b13,
2076+ b14: if b == 14 then union [e] acc.b14 else acc.b14, b15: if b == 15 then union [e] acc.b15 else acc.b15})
2077+ {b0: [], b1: [], b2: [], b3: [], b4: [], b5: [], b6: [], b7: [], b8: [], b9: [], b10: [], b11: [], b12: [], b13: [], b14: [], b15: []} entries
2078+ writeAckedBucket 0 p.b0; writeAckedBucket 1 p.b1; writeAckedBucket 2 p.b2; writeAckedBucket 3 p.b3
2079+ writeAckedBucket 4 p.b4; writeAckedBucket 5 p.b5; writeAckedBucket 6 p.b6; writeAckedBucket 7 p.b7
2080+ writeAckedBucket 8 p.b8; writeAckedBucket 9 p.b9; writeAckedBucket 10 p.b10; writeAckedBucket 11 p.b11
2081+ writeAckedBucket 12 p.b12; writeAckedBucket 13 p.b13; writeAckedBucket 14 p.b14; writeAckedBucket 15 p.b15
2083+countAllAcked : Int -> IO Int
2084+countAllAcked = \b -> if b > 15 then yield 0 else do
2085+ bucket <- readAckedBucket b; rest <- countAllAcked (b + 1); yield (count bucket + rest)
2087+-- ============================================
2088+-- DeliveredIds Bucket Helpers
2089+-- ============================================
2090+readDelivBucket : Int -> IO [DeliveredEntry]
2091+readDelivBucket = \b ->
2092+ if b == 0 then do s <- *delivBucket0; yield s else if b == 1 then do s <- *delivBucket1; yield s
2093+ else if b == 2 then do s <- *delivBucket2; yield s else if b == 3 then do s <- *delivBucket3; yield s
2094+ else if b == 4 then do s <- *delivBucket4; yield s else if b == 5 then do s <- *delivBucket5; yield s
2095+ else if b == 6 then do s <- *delivBucket6; yield s else if b == 7 then do s <- *delivBucket7; yield s
2096+ else if b == 8 then do s <- *delivBucket8; yield s else if b == 9 then do s <- *delivBucket9; yield s
2097+ else if b == 10 then do s <- *delivBucket10; yield s else if b == 11 then do s <- *delivBucket11; yield s
2098+ else if b == 12 then do s <- *delivBucket12; yield s else if b == 13 then do s <- *delivBucket13; yield s
2099+ else if b == 14 then do s <- *delivBucket14; yield s else do s <- *delivBucket15; yield s
2101+writeDelivBucket : Int -> [DeliveredEntry] -> IO ()
2102+writeDelivBucket = \b entries ->
2103+ if b == 0 then set *delivBucket0 = entries else if b == 1 then set *delivBucket1 = entries
2104+ else if b == 2 then set *delivBucket2 = entries else if b == 3 then set *delivBucket3 = entries
2105+ else if b == 4 then set *delivBucket4 = entries else if b == 5 then set *delivBucket5 = entries
2106+ else if b == 6 then set *delivBucket6 = entries else if b == 7 then set *delivBucket7 = entries
2107+ else if b == 8 then set *delivBucket8 = entries else if b == 9 then set *delivBucket9 = entries
2108+ else if b == 10 then set *delivBucket10 = entries else if b == 11 then set *delivBucket11 = entries
2109+ else if b == 12 then set *delivBucket12 = entries else if b == 13 then set *delivBucket13 = entries
2110+ else if b == 14 then set *delivBucket14 = entries else set *delivBucket15 = entries
2112+readAllDelivered : IO [DeliveredEntry]
2113+readAllDelivered = do
2114+ d0 <- *delivBucket0; d1 <- *delivBucket1; d2 <- *delivBucket2; d3 <- *delivBucket3
2115+ d4 <- *delivBucket4; d5 <- *delivBucket5; d6 <- *delivBucket6; d7 <- *delivBucket7
2116+ d8 <- *delivBucket8; d9 <- *delivBucket9; d10 <- *delivBucket10; d11 <- *delivBucket11
2117+ d12 <- *delivBucket12; d13 <- *delivBucket13; d14 <- *delivBucket14; d15 <- *delivBucket15
2118+ yield (union d0 (union d1 (union d2 (union d3 (union d4 (union d5 (union d6 (union d7 (union d8 (union d9 (union d10 (union d11 (union d12 (union d13 (union d14 d15)))))))))))))))
2120+writeAllDelivered : [DeliveredEntry] -> IO ()
2121+writeAllDelivered = \entries -> do
2122+ let p = fold (\acc e -> let b = pubkeyBucket e.toKey in
2123+ {b0: if b == 0 then union [e] acc.b0 else acc.b0, b1: if b == 1 then union [e] acc.b1 else acc.b1,
2124+ b2: if b == 2 then union [e] acc.b2 else acc.b2, b3: if b == 3 then union [e] acc.b3 else acc.b3,
2125+ b4: if b == 4 then union [e] acc.b4 else acc.b4, b5: if b == 5 then union [e] acc.b5 else acc.b5,
2126+ b6: if b == 6 then union [e] acc.b6 else acc.b6, b7: if b == 7 then union [e] acc.b7 else acc.b7,
2127+ b8: if b == 8 then union [e] acc.b8 else acc.b8, b9: if b == 9 then union [e] acc.b9 else acc.b9,
2128+ b10: if b == 10 then union [e] acc.b10 else acc.b10, b11: if b == 11 then union [e] acc.b11 else acc.b11,
2129+ b12: if b == 12 then union [e] acc.b12 else acc.b12, b13: if b == 13 then union [e] acc.b13 else acc.b13,
2130+ b14: if b == 14 then union [e] acc.b14 else acc.b14, b15: if b == 15 then union [e] acc.b15 else acc.b15})
2131+ {b0: [], b1: [], b2: [], b3: [], b4: [], b5: [], b6: [], b7: [], b8: [], b9: [], b10: [], b11: [], b12: [], b13: [], b14: [], b15: []} entries
2132+ writeDelivBucket 0 p.b0; writeDelivBucket 1 p.b1; writeDelivBucket 2 p.b2; writeDelivBucket 3 p.b3
2133+ writeDelivBucket 4 p.b4; writeDelivBucket 5 p.b5; writeDelivBucket 6 p.b6; writeDelivBucket 7 p.b7
2134+ writeDelivBucket 8 p.b8; writeDelivBucket 9 p.b9; writeDelivBucket 10 p.b10; writeDelivBucket 11 p.b11
2135+ writeDelivBucket 12 p.b12; writeDelivBucket 13 p.b13; writeDelivBucket 14 p.b14; writeDelivBucket 15 p.b15
2137+countAllDelivered : Int -> IO Int
2138+countAllDelivered = \b -> if b > 15 then yield 0 else do
2139+ bucket <- readDelivBucket b; rest <- countAllDelivered (b + 1); yield (count bucket + rest)
1724-- ============================================
2141-- ============================================
1725-- Federation In-Flight Tracking
2142-- Federation In-Flight Tracking
1726-- ============================================
2143-- ============================================
@@ -2119,18 +2536,18 @@ pruneExpired = \t -> do
2119 -- Prune federation state
2536 -- Prune federation state
2120 atomic do
2537 atomic do
2121 let retainCutoff = t - messageMaxAge
2538 let retainCutoff = t - messageMaxAge
2122− forwards <- *forwards
2539+ forwards <- readAllForwards
2123 -- Clear all forwards when federation is disabled (they can never be retried)
2540 -- Clear all forwards when federation is disabled (they can never be retried)
2124− set *forwards = if not federationEnabled then []
2541+ writeAllForwards (if not federationEnabled then []
2125− else do f <- forwards; where f.createdAt > retainCutoff; yield f
2542+ else do f <- forwards; where f.createdAt > retainCutoff; yield f)
2126− ackedIds <- *ackedIds
2543+ ackedIds <- readAllAcked
2127− set *ackedIds = do a <- ackedIds; where a.ackedAt > retainCutoff; yield a
2544+ writeAllAcked (do a <- ackedIds; where a.ackedAt > retainCutoff; yield a)
2128 forwardFailures <- *forwardFailures
2545 forwardFailures <- *forwardFailures
2129 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f
2546 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f
2130 ackOverflow <- *ackOverflow
2547 ackOverflow <- *ackOverflow
2131 set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a
2548 set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a
2132− deliveredIds <- *deliveredIds
2549+ deliveredIds <- readAllDelivered
2133− set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > retainCutoff; yield d
2550+ writeAllDelivered (do d <- deliveredIds; where d.deliveredAt > retainCutoff; yield d)
2134 -- Prune rate limiters (split into smaller atomic blocks to reduce lock contention)
2551 -- Prune rate limiters (split into smaller atomic blocks to reduce lock contention)
2135 atomic do
2552 atomic do
2136 sendRates <- *sendRates
2553 sendRates <- *sendRates
@@ -2189,9 +2606,8 @@ pruneExpired = \t -> do
2189 -- Prune expired federation verification tokens
2606 -- Prune expired federation verification tokens
2190 pending <- *pendingVerifications
2607 pending <- *pendingVerifications
2191 set *pendingVerifications = do p <- pending; where t - p.createdAt < verificationTokenExpiry; yield p
2608 set *pendingVerifications = do p <- pending; where t - p.createdAt < verificationTokenExpiry; yield p
2192− -- Prune expired forward nonces (extended replay protection)
2609+ -- Prune expired forward nonces per bucket (extended replay protection)
2193− forwardNonces <- *forwardNonces
2610+ pruneAllNonceBuckets 0 t
2194− set *forwardNonces = do n <- forwardNonces; where t - n.seenAt < forwardNonceRetention; yield n
2195 -- Prune gossip event rate entries
2611 -- Prune gossip event rate entries
2196 gossipEventRates <- *gossipEventRates
2612 gossipEventRates <- *gossipEventRates
2197 set *gossipEventRates = do r <- gossipEventRates; where t - r.windowStart < gossipEventRateWindowMs; yield r
2613 set *gossipEventRates = do r <- gossipEventRates; where t - r.windowStart < gossipEventRateWindowMs; yield r
@@ -2432,9 +2848,10 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
2432 then onUnauth
2848 then onUnauth
2433 else onAuth pubkey
2849 else onAuth pubkey
2435−removeForward = \mid srv -> atomic do
2851+removeForward = \mid srv toKey -> atomic do
2436− forwards <- *forwards
2852+ let b = pubkeyBucket toKey
2853+ fwdBkt <- readFwdBucket b
2437− set *forwards = forwardsExcluding forwards srv [mid]
2854+ writeFwdBucket b (forwardsExcluding fwdBkt srv [mid])
2439-- Reclaim origin messages after federation ack confirms delivery.
2856-- Reclaim origin messages after federation ack confirms delivery.
2440-- Batched by message bucket: all entries for the same bucket are processed in one
2857-- Batched by message bucket: all entries for the same bucket are processed in one
@@ -2451,13 +2868,13 @@ ackCleanupByBucket = \bucketIds entries t -> case headOf bucketIds of
2451 atomic do
2868 atomic do
2452 -- All entries in this batch share pubkeyBucket == b, so only that presence bucket is needed
2869 -- All entries in this batch share pubkeyBucket == b, so only that presence bucket is needed
2453 bucketPresence <- readPresenceBucket b
2870 bucketPresence <- readPresenceBucket b
2454− forwards <- *forwards
2871+ fwdBkt <- readFwdBucket b
2455 msgs <- readMessageBucket b
2872 msgs <- readMessageBucket b
2456 tc <- *totalMessageCount
2873 tc <- *totalMessageCount
2457 rc <- readRcBucket b
2874 rc <- readRcBucket b
2458 let cleanup = fold (\acc rf ->
2875 let cleanup = fold (\acc rf ->
2459 let hasLocal = any bucketPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in
2876 let hasLocal = any bucketPresence (\p -> p.pubkey == rf.toKey && isLocalServer p.server && p.expiresAt > t) in
2460− let hasOtherFwds = any forwards (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey) in
2877+ let hasOtherFwds = any fwdBkt (\f -> f.messageId == rf.messageId && f.toKey == rf.toKey) in
2461 if hasLocal || hasOtherFwds then acc
2878 if hasLocal || hasOtherFwds then acc
2462 else case headOf (do m <- acc.msgs; where m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod; yield m) of
2879 else case headOf (do m <- acc.msgs; where m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod; yield m) of
2463 Nothing {} -> acc
2880 Nothing {} -> acc
@@ -2476,11 +2893,20 @@ retryBackoff : Int -> Int<Ms>
2476retryBackoff = \failures ->
2893retryBackoff = \failures ->
2477 min retryBackoffMax (retryBackoffBase * pow2 (max 0 (failures - 1)))
2894 min retryBackoffMax (retryBackoffBase * pow2 (max 0 (failures - 1)))
2896+-- Exponential backoff with deterministic per-server jitter to prevent thundering herd
2897+-- when multiple peers recover simultaneously. Jitter derived from first hex char of
2898+-- server name (0-15) adds 0-47% of base delay, spreading retries across the window.
2899+retryBackoffFor : Int -> Text -> Int<Ms>
2900+retryBackoffFor = \failures server ->
2901+ let base = retryBackoff failures in
2902+ let jitterFrac = hexCharIndex (take 1 server) in
2903+ min retryBackoffMax (base + base * jitterFrac / 32)
2479shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
2905shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
2480shouldRetryServer = \server t failures ->
2906shouldRetryServer = \server t failures ->
2481 case headOf (do f <- failures; where f.server == server; yield f) of
2907 case headOf (do f <- failures; where f.server == server; yield f) of
2482 Nothing {} -> True {}
2908 Nothing {} -> True {}
2483− Just {value: f} -> t - f.firstFailedAt < maxServerDowntime && t - f.failedAt >= retryBackoff f.failures
2909+ Just {value: f} -> t - f.firstFailedAt < maxServerDowntime && t - f.failedAt >= retryBackoffFor f.failures server
2485-- Extended retry check: also skip servers with hung in-flight requests or on the blocklist
2911-- Extended retry check: also skip servers with hung in-flight requests or on the blocklist
2486shouldRetryServerEx : Text -> Timestamp -> [ForwardFailure] -> [FederationInFlightEntry] -> [Text] -> Bool
2912shouldRetryServerEx : Text -> Timestamp -> [ForwardFailure] -> [FederationInFlightEntry] -> [Text] -> Bool
@@ -2513,14 +2939,15 @@ clearFailure = \server -> do
2513 set *lastFederationSuccessAt = t
2939 set *lastFederationSuccessAt = t
2515-- Atomically increment a forward entry's retry count; drop if maxForwardRetries exceeded
2941-- Atomically increment a forward entry's retry count; drop if maxForwardRetries exceeded
2516−incrementForwardRetries = \messageId toServer -> atomic do
2942+incrementForwardRetries = \messageId toServer toKey -> atomic do
2517− forwards <- *forwards
2943+ let b = pubkeyBucket toKey
2944+ fwdBkt <- readFwdBucket b
2518− let shouldDrop = any forwards (\f -> f.messageId == messageId && f.toServer == toServer && f.retries + 1 >= maxForwardRetries)
2945+ let shouldDrop = any fwdBkt (\f -> f.messageId == messageId && f.toServer == toServer && f.retries + 1 >= maxForwardRetries)
2519 if shouldDrop
2946 if shouldDrop
2521− set *forwards = do f <- forwards; where not (f.messageId == messageId && f.toServer == toServer); yield f
2948+ writeFwdBucket b (do f <- fwdBkt; where not (f.messageId == messageId && f.toServer == toServer); yield f)
2522 logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
2949 logWarn ("federation: dropping forward " ++ messageId ++ " to " ++ toServer ++ " after " ++ show maxForwardRetries ++ " retries")
2523− else set *forwards = do f <- forwards; yield (if f.messageId == messageId && f.toServer == toServer then {messageId: f.messageId, toServer: f.toServer, createdAt: f.createdAt, toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, retries: f.retries + 1} else f)
2950+ else writeFwdBucket b (do f <- fwdBkt; yield (if f.messageId == messageId && f.toServer == toServer then {messageId: f.messageId, toServer: f.toServer, createdAt: f.createdAt, toKey: f.toKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, originServer: f.originServer, retries: f.retries + 1} else f))
2525retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
2952retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
2526retryForwardsForServer = \entries remaining ->
2953retryForwardsForServer = \entries remaining ->
@@ -2542,15 +2969,15 @@ retryForwardsForServer = \entries remaining ->
2542 Err e -> do
2969 Err e -> do
2543 logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
2970 logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)
2544 recordFailure f.toServer
2971 recordFailure f.toServer
2545− incrementForwardRetries f.messageId f.toServer
2972+ incrementForwardRetries f.messageId f.toServer f.toKey
2546 -- Network error: stop retrying this server
2973 -- Network error: stop retrying this server
2547 Ok response -> do
2974 Ok response -> do
2548 if response.ok then do
2975 if response.ok then do
2549− removeForward f.messageId f.toServer
2976+ removeForward f.messageId f.toServer f.toKey
2550 clearFailure f.toServer
2977 clearFailure f.toServer
2552 logWarn ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)
2979 logWarn ("federation: forward to " ++ f.toServer ++ " rejected for " ++ f.messageId ++ ": " ++ response.error)
2553− incrementForwardRetries f.messageId f.toServer
2980+ incrementForwardRetries f.messageId f.toServer f.toKey
2554 retryForwardsForServer rest (remaining - 1)
2981 retryForwardsForServer rest (remaining - 1)
2556-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
2983-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
@@ -2572,7 +2999,7 @@ retryPendingForwards = \pending -> do
2572-- When a remote key comes online, create forward entries for queued messages and send them
2999-- When a remote key comes online, create forward entries for queued messages and send them
2573forwardQueuedMessages = \targets t -> do
3000forwardQueuedMessages = \targets t -> do
2574 newForwards <- atomic do
3001 newForwards <- atomic do
2575− forwards <- *forwards
3002+ forwards <- readAllForwards
2576 -- Read only the presence buckets needed for target pubkeys (not all 16)
3003 -- Read only the presence buckets needed for target pubkeys (not all 16)
2577 let targetPresenceBuckets = uniqueBucketsFor targets (\target -> target.pubkey)
3004 let targetPresenceBuckets = uniqueBucketsFor targets (\target -> target.pubkey)
2578 relevantPresence <- readSelectedPresenceBuckets targetPresenceBuckets
3005 relevantPresence <- readSelectedPresenceBuckets targetPresenceBuckets
@@ -2582,9 +3009,9 @@ forwardQueuedMessages = \targets t -> do
2582 let neededBuckets = uniqueBucketsFor remoteTargets (\target -> target.pubkey)
3009 let neededBuckets = uniqueBucketsFor remoteTargets (\target -> target.pubkey)
2583 relevantMessages <- readSelectedBuckets neededBuckets
3010 relevantMessages <- readSelectedBuckets neededBuckets
2584 let candidates = forwardCandidates remoteTargets relevantMessages forwards t
3011 let candidates = forwardCandidates remoteTargets relevantMessages forwards t
3012+ let totalFwdCount <- countAllForwards 0
2585− let capped = takeRelation (maxForwardsEntries - count forwards) candidates
3013+ let capped = takeRelation (maxForwardsEntries - totalFwdCount) candidates
2586− let updatedForwards = union forwards capped
3014+ writeAllForwards (union forwards capped)
2587− set *forwards = updatedForwards
2588 yield capped
3015 yield capped
2589 let servers = collectPeerServers newForwards (\x -> x.toServer)
3016 let servers = collectPeerServers newForwards (\x -> x.toServer)
2590 failures <- *forwardFailures
3017 failures <- *forwardFailures
@@ -2617,27 +3044,27 @@ sendBatchedAcksToServer = \serverAcks server ->
2617 recordFailure server
3044 recordFailure server
2618 -- Increment retry counters; drop entries exceeding maxAckRetries to deliveredIds
3045 -- Increment retry counters; drop entries exceeding maxAckRetries to deliveredIds
2619 dropped <- atomic do
3046 dropped <- atomic do
2620− ackedIds <- *ackedIds
3047+ ackedIds <- readAllAcked
2621 let isTarget = \a -> a.originServer == server && hasMessage a.messageId a.toKey batch
3048 let isTarget = \a -> a.originServer == server && hasMessage a.messageId a.toKey batch
2622 let toDrop = do a <- ackedIds; where isTarget a && a.retries + 1 >= maxAckRetries; yield a
3049 let toDrop = do a <- ackedIds; where isTarget a && a.retries + 1 >= maxAckRetries; yield a
2623 let dropCount = count toDrop
3050 let dropCount = count toDrop
2624− set *ackedIds = do a <- ackedIds
3051+ writeAllAcked (do a <- ackedIds
2625 where not (isTarget a && a.retries + 1 >= maxAckRetries)
3052 where not (isTarget a && a.retries + 1 >= maxAckRetries)
2626− yield (if isTarget a then {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1} else a)
3053+ yield (if isTarget a then {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1} else a))
2627 when (dropCount > 0) (do
3054 when (dropCount > 0) (do
2628− deliveredIds <- *deliveredIds
3055+ deliveredIds <- readAllDelivered
2629 let newDelivered = do d <- toDrop; yield {messageId: d.messageId, toKey: d.toKey, deliveredAt: d.ackedAt}
3056 let newDelivered = do d <- toDrop; yield {messageId: d.messageId, toKey: d.toKey, deliveredAt: d.ackedAt}
2630 let combined = union deliveredIds newDelivered
3057 let combined = union deliveredIds newDelivered
2631− set *deliveredIds = if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined)
3058+ writeAllDelivered (if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined))
2632 yield dropCount
3059 yield dropCount
2633 when (dropped > 0)
3060 when (dropped > 0)
2634 (logWarn ("federation: dropped " ++ show dropped ++ " ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
3061 (logWarn ("federation: dropped " ++ show dropped ++ " ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
2635 Ok _ -> do
3062 Ok _ -> do
2636 clearFailure server
3063 clearFailure server
2637 atomic do
3064 atomic do
2638− ackedIds <- *ackedIds
3065+ ackedIds <- readAllAcked
2639 let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch
3066 let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch
2640− set *ackedIds = updatedAckedIds)
3067+ writeAllAcked updatedAckedIds)
2642-- Send delivery acknowledgments back to origin servers and remove acked entries on success
3069-- Send delivery acknowledgments back to origin servers and remove acked entries on success
2643sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
3070sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
@@ -2667,15 +3094,15 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
2667 Err e -> do
3094 Err e -> do
2668 logWarn ("federation: evicted ack to " ++ server ++ " failed: " ++ show e)
3095 logWarn ("federation: evicted ack to " ++ server ++ " failed: " ++ show e)
2669 overflow <- atomic do
3096 overflow <- atomic do
2670− ackedIds <- *ackedIds
3097+ ackedIds <- readAllAcked
2671 let reinsert = do a <- acksForServer evictedToRetry server; yield {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1}
3098 let reinsert = do a <- acksForServer evictedToRetry server; yield {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1}
2672 let combined = union ackedIds reinsert
3099 let combined = union ackedIds reinsert
2673 let excess = count combined - maxAckedIds
3100 let excess = count combined - maxAckedIds
2674 if excess <= 0
3101 if excess <= 0
2675− then do set *ackedIds = combined; yield []
3102+ then do writeAllAcked combined; yield []
2677 let eviction = evictSmallest combined (max 0 excess) (\a -> a.ackedAt)
3104 let eviction = evictSmallest combined (max 0 excess) (\a -> a.ackedAt)
2678− set *ackedIds = eviction.kept
3105+ writeAllAcked eviction.kept
2679 yield eviction.evicted
3106 yield eviction.evicted
2680 when (count overflow > 0) (do
3107 when (count overflow > 0) (do
2681 dropped <- atomic do
3108 dropped <- atomic do
@@ -2687,12 +3114,12 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
2688 let overflowEviction = evictSmallest updated (max 0 overflowExcess) (\a -> a.ackedAt)
3115 let overflowEviction = evictSmallest updated (max 0 overflowExcess) (\a -> a.ackedAt)
2689 set *ackOverflow = overflowEviction.kept
3116 set *ackOverflow = overflowEviction.kept
2690− deliveredIds <- *deliveredIds
3117+ deliveredIds <- readAllDelivered
2691 let newDelivered = do a <- overflowEviction.evicted; yield {messageId: a.messageId, toKey: a.toKey, deliveredAt: a.ackedAt}
3118 let newDelivered = do a <- overflowEviction.evicted; yield {messageId: a.messageId, toKey: a.toKey, deliveredAt: a.ackedAt}
2692 let combinedDelivered = union deliveredIds newDelivered
3119 let combinedDelivered = union deliveredIds newDelivered
2693 let deliveredExcess = count combinedDelivered - maxDeliveredIds
3120 let deliveredExcess = count combinedDelivered - maxDeliveredIds
2694− set *deliveredIds = if deliveredExcess <= 0 then combinedDelivered
3121+ writeAllDelivered (if deliveredExcess <= 0 then combinedDelivered
2695− else (evictSmallest combinedDelivered (max 0 deliveredExcess) (\d -> d.deliveredAt)).kept
3122+ else (evictSmallest combinedDelivered (max 0 deliveredExcess) (\d -> d.deliveredAt)).kept)
2696 when (deliveredExcess > 0)
3123 when (deliveredExcess > 0)
2697 (logWarn ("ackMessages: deliveredIds at capacity (" ++ show maxDeliveredIds ++ "), " ++ show deliveredExcess ++ " oldest dedup entries permanently removed for " ++ server))
3124 (logWarn ("ackMessages: deliveredIds at capacity (" ++ show maxDeliveredIds ++ "), " ++ show deliveredExcess ++ " oldest dedup entries permanently removed for " ++ server))
2698 yield overflowExcess
3125 yield overflowExcess
@@ -2728,10 +3155,10 @@ retryOverflowAcks = \overflow -> do
2728 where not (isTarget a && a.retries + 1 >= maxAckRetries)
3155 where not (isTarget a && a.retries + 1 >= maxAckRetries)
2729 yield (if isTarget a then {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1} else a)
3156 yield (if isTarget a then {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1} else a)
2730 when (dropCount > 0) (do
3157 when (dropCount > 0) (do
2731− deliveredIds <- *deliveredIds
3158+ deliveredIds <- readAllDelivered
2732 let newDelivered = do d <- toDrop; yield {messageId: d.messageId, toKey: d.toKey, deliveredAt: d.ackedAt}
3159 let newDelivered = do d <- toDrop; yield {messageId: d.messageId, toKey: d.toKey, deliveredAt: d.ackedAt}
2733 let combined = union deliveredIds newDelivered
3160 let combined = union deliveredIds newDelivered
2734− set *deliveredIds = if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined)
3161+ writeAllDelivered (if count combined > maxDeliveredIds then (evictSmallest combined (count combined - maxDeliveredIds) (\d -> d.deliveredAt)).kept else combined))
2735 yield dropCount
3162 yield dropCount
2736 when (dropped > 0)
3163 when (dropped > 0)
2737 (logWarn ("federation: dropped " ++ show dropped ++ " overflow ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
3164 (logWarn ("federation: dropped " ++ show dropped ++ " overflow ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))
@@ -2801,9 +3228,9 @@ pruneAndRetry = \t -> do
2801 when (msgDrift != 0)
3228 when (msgDrift != 0)
2802 (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))
3229 (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))
2803 else yield {}
3230 else yield {}
2804− pendingForwards <- atomic do f <- *forwards; yield f
3231+ pendingForwards <- atomic do f <- readAllForwards; yield f
2805 ackState <- atomic do
3232 ackState <- atomic do
2806− ackedIds2 <- *ackedIds
3233+ ackedIds2 <- readAllAcked
2807 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a
3234 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a
2808 ackOverflow <- *ackOverflow
3235 ackOverflow <- *ackOverflow
2809 let pendingOverflow = do a <- ackOverflow; where not (isLocalServer a.originServer); yield a
3236 let pendingOverflow = do a <- ackOverflow; where not (isLocalServer a.originServer); yield a
@@ -2811,22 +3238,33 @@ pruneAndRetry = \t -> do
2811 refreshResult <- atomic do
3238 refreshResult <- atomic do
2812 lastRefresh <- *lastPresenceRefresh
3239 lastRefresh <- *lastPresenceRefresh
2813 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval
3240 let shouldRefresh = federationEnabled && t - lastRefresh >= presenceRefreshInterval
2814− -- Early return: skip expensive readAllPresence + readAllSessions on ~9/10 prune cycles
3241+ -- Early return: skip expensive reads on ~9/10 prune cycles
2815 if not shouldRefresh then yield {extendedActive: [], nearExpiryActive: [], refreshPeers: [], shouldRefresh: False {}}
3242 if not shouldRefresh then yield {extendedActive: [], nearExpiryActive: [], refreshPeers: [], shouldRefresh: False {}}
2817− allPresence <- readAllPresence
3244+ -- Optimized: scan presence buckets one at a time to find near-expiry local entries,
2818− allSessions <- readAllSessions
3245+ -- then only read session buckets for the pubkeys found. Avoids materializing all 16
2819− -- Extend presence for keys with active sessions nearing expiry (sliding window)
3246+ -- presence + 16 session buckets when few keys are near expiry (common case).
2820− let localNearExpiry = do p <- allPresence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p
3247+ localNearExpiry <- findNearExpiryPresence 0 t
3248+ if count localNearExpiry == 0 then do
3249+ knownPeers <- *knownPeerServers
3250+ set *lastPresenceRefresh = t
3251+ yield {extendedActive: [], nearExpiryActive: [], refreshPeers: knownPeers, shouldRefresh: True {}}
3253+ -- Only read session buckets for pubkeys that have near-expiry presence
3254+ let nearExpiryPubkeys = fold (\acc p -> if any acc (\a -> a == p.pubkey) then acc else union acc [p.pubkey]) [] localNearExpiry
3255+ allSessions <- readAllSessions
2821− let toExtend = do p <- localNearExpiry; where any allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
3256+ let toExtend = do p <- localNearExpiry; where any allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
2822− let notExtended = do p <- localNearExpiry; where none allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
3257+ let notExtended = do p <- localNearExpiry; where none allSessions (\s -> s.pubkey == p.pubkey && s.expiresAt > t); yield p
2823− when (count toExtend > 0) (do
3258+ when (count toExtend > 0) (do
3259+ -- Only write to the affected presence buckets (not all 16)
3260+ let affectedBuckets = uniqueBucketsFor toExtend (\e -> e.pubkey)
3261+ affectedPresence <- readSelectedPresenceBuckets affectedBuckets
2824− let updatedPresence = do pp <- allPresence; yield (if any toExtend (\e -> e.pubkey == pp.pubkey && isLocalServer pp.server) then {pubkey: pp.pubkey, server: pp.server, expiresAt: t + sessionExpiry, timestamp: t} else pp)
3262+ let updatedPresence = do pp <- affectedPresence; yield (if any toExtend (\e -> e.pubkey == pp.pubkey && isLocalServer pp.server) then {pubkey: pp.pubkey, server: pp.server, expiresAt: t + sessionExpiry, timestamp: t} else pp)
2825− writeAllPresence updatedPresence
3263+ writeSelectedPresenceBuckets affectedBuckets updatedPresence
2826− set *presenceVersion = t)
3264+ set *presenceVersion = t)
2827− knownPeers <- *knownPeerServers
3265+ knownPeers <- *knownPeerServers
2828− set *lastPresenceRefresh = t
3266+ set *lastPresenceRefresh = t
2829− yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: knownPeers, shouldRefresh: True {}}
3267+ yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: knownPeers, shouldRefresh: True {}}
2830 -- Reset federation latency metrics periodically for windowed averages (prevents
3268 -- Reset federation latency metrics periodically for windowed averages (prevents
2831 -- lifetime-cumulative average from masking current conditions)
3269 -- lifetime-cumulative average from masking current conditions)
2832 atomic do
3270 atomic do
@@ -2935,14 +3373,16 @@ ackPolledMessages = \pubkey cursor t -> do
2935 recipientCounts <- readRcBucket b
3373 recipientCounts <- readRcBucket b
2936 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
3374 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
2937 let ackedMsgIds = do m <- toAck; yield m.messageId
3375 let ackedMsgIds = do m <- toAck; yield m.messageId
2938− forwards <- *forwards
3376+ let fb = pubkeyBucket pubkey
3377+ fwdBkt <- readFwdBucket fb
2939− set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
3378+ writeFwdBucket fb (do f <- fwdBkt; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
2940 let ackedFederated = federatedAckEntries toAck t
3379 let ackedFederated = federatedAckEntries toAck t
2941− ackedIds <- *ackedIds
3380+ let ab = pubkeyBucket pubkey
3381+ ackedBkt <- readAckedBucket ab
2942− let combinedAckedIds = union ackedIds ackedFederated
3382+ let combinedAckedIds = union ackedBkt ackedFederated
2943 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
3383 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
2944 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
3384 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
2945− set *ackedIds = eviction.kept
3385+ writeAckedBucket ab eviction.kept
2946 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
3386 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
2947 else yield {pendingAcks: [], evictedEntries: []}
3387 else yield {pendingAcks: [], evictedEntries: []}
@@ -3382,9 +3822,9 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
3383 let msgBucket = pubkeyBucket to
3823 let msgBucket = pubkeyBucket to
3384 bucket <- readMessageBucket msgBucket
3824 bucket <- readMessageBucket msgBucket
3385− ackedIds <- *ackedIds
3825+ ackedBkt <- readAckedBucket msgBucket
3386− deliveredIds <- *deliveredIds
3826+ delivBkt <- readDelivBucket msgBucket
3387− let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedIds || hasMessage effectiveMsgId to deliveredIds
3827+ let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt
3388 if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}
3828 if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}
3390 sendRates <- *sendRates
3830 sendRates <- *sendRates
@@ -3425,20 +3865,24 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->
3425 -- if the message expires from messages/ackedIds/deliveredIds and a federation
3865 -- if the message expires from messages/ackedIds/deliveredIds and a federation
3426 -- peer forwards it back. Without this, the nonce check in handleForwardMessage
3866 -- peer forwards it back. Without this, the nonce check in handleForwardMessage
3427 -- would miss locally-originated messages entirely.
3867 -- would miss locally-originated messages entirely.
3428− fwdNonces <- *forwardNonces
3868+ -- Sharded: writes only to the bucket for this recipient's pubkey.
3869+ let nb = pubkeyBucket to
3870+ fwdNonces <- readNonceBucket nb
3429 let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, originServer: "", seenAt: t}]
3871 let updatedNonces = union fwdNonces [{messageId: effectiveMsgId, toKey: to, originServer: "", seenAt: t}]
3430− -- O(n) cutoff eviction: remove expired nonces first, avoiding O(n*k) evictSmallest on hot path
3872+ let maxPerBucket = maxForwardNonces / 16 + 1
3431− set *forwardNonces = if count updatedNonces > maxForwardNonces
3873+ writeNonceBucket nb (if count updatedNonces > maxPerBucket
3432− then evictNonces updatedNonces maxForwardNonces forwardNonceRetention t
3874+ then evictNonces updatedNonces maxPerBucket forwardNonceRetention t
3433− else updatedNonces
3875+ else updatedNonces)
3434 let toPresence = presenceForKey recipientPresence to t
3876 let toPresence = presenceForKey recipientPresence to t
3435 let localOnline = hasLocalPresence toPresence
3877 let localOnline = hasLocalPresence toPresence
3436 let remoteServers = collectPeerServers toPresence (\p -> p.server)
3878 let remoteServers = collectPeerServers toPresence (\p -> p.server)
3437 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
3879 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do
3438− forwards <- *forwards
3880+ let fbi = pubkeyBucket to
3881+ fwdBkt <- readFwdBucket fbi
3439− 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}
3882+ let newFwds = do s <- remoteServers; where none fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == s); yield {messageId: effectiveMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0}
3883+ totalFwdCount <- countAllForwards 0
3440− let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds
3884+ let cappedFwds = takeRelation (maxForwardsEntries - totalFwdCount) newFwds
3441− set *forwards = union forwards cappedFwds
3885+ writeFwdBucket fbi (union fwdBkt cappedFwds)
3442 -- Return whether any forwards were actually enqueued (vs dropped by capacity cap)
3886 -- Return whether any forwards were actually enqueued (vs dropped by capacity cap)
3443 yield (count cappedFwds > 0)
3887 yield (count cappedFwds > 0)
3444 else yield (False {})
3888 else yield (False {})
@@ -3452,8 +3896,8 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
3452 when (count remoteServers > 0 && federationEnabled) (
3896 when (count remoteServers > 0 && federationEnabled) (
3453 fork (sendToEach remoteServers (\server -> do
3897 fork (sendToEach remoteServers (\server -> do
3454 stillPending <- atomic do
3898 stillPending <- atomic do
3455− forwards <- *forwards
3899+ fwdBkt <- readFwdBucket (pubkeyBucket to)
3456− yield (any forwards (\f -> f.messageId == effectiveMsgId && f.toServer == server))
3900+ yield (any fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == server))
3457 when stillPending (do
3901 when stillPending (do
3458 fwdT <- now
3902 fwdT <- now
3459 fwdAuth <- generateFederationAuth "forward"
3903 fwdAuth <- generateFederationAuth "forward"
@@ -3469,7 +3913,7 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->
3469 recordFailure server
3913 recordFailure server
3470 Ok response ->
3914 Ok response ->
3471 if response.ok then do
3915 if response.ok then do
3472− removeForward effectiveMsgId server
3916+ removeForward effectiveMsgId server to
3473 clearFailure server
3917 clearFailure server
3475 logWarn ("federation: forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))
3919 logWarn ("federation: forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))
@@ -3482,8 +3926,8 @@ sendBatchForwardsSequential = \msgs server ->
3482 Just {value: r} -> do
3926 Just {value: r} -> do
3483 let rest = do m <- msgs; where m.messageId != r.messageId; yield m
3927 let rest = do m <- msgs; where m.messageId != r.messageId; yield m
3484 stillPending <- atomic do
3928 stillPending <- atomic do
3485− forwards <- *forwards
3929+ fwdBkt <- readFwdBucket (pubkeyBucket r.toKey)
3486− yield (any forwards (\f -> f.messageId == r.messageId && f.toServer == server))
3930+ yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))
3487 when stillPending (do
3931 when stillPending (do
3488 fwdT <- now
3932 fwdT <- now
3489 fwdAuth <- generateFederationAuth "forward"
3933 fwdAuth <- generateFederationAuth "forward"
@@ -3499,7 +3943,7 @@ sendBatchForwardsSequential = \msgs server ->
3499 recordFailure server
3943 recordFailure server
3500 Ok response ->
3944 Ok response ->
3501 if response.ok then do
3945 if response.ok then do
3502− removeForward r.messageId server
3946+ removeForward r.messageId server r.toKey
3503 clearFailure server
3947 clearFailure server
3505 logWarn ("federation: forward to " ++ server ++ " rejected for " ++ r.messageId ++ ": " ++ response.error))
3949 logWarn ("federation: forward to " ++ server ++ " rejected for " ++ r.messageId ++ ": " ++ response.error))
@@ -3595,22 +4039,23 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
3595 let inMailbox = any bucket (\m -> m.messageId == messageId && m.toKey == to)
4039 let inMailbox = any bucket (\m -> m.messageId == messageId && m.toKey == to)
3596 if inMailbox then yield "pending"
4040 if inMailbox then yield "pending"
3598− -- Check if being forwarded to a remote server
4042+ -- Check if being forwarded to a remote server (bucket read: O(n/16))
3599− forwards <- *forwards
4043+ fwdBkt <- readFwdBucket (pubkeyBucket to)
3600− let inForwards = any forwards (\f -> f.messageId == messageId && f.toKey == to)
4044+ let inForwards = any fwdBkt (\f -> f.messageId == messageId && f.toKey == to)
3601 if inForwards then yield "forwarding"
4045 if inForwards then yield "forwarding"
3603 -- Check if already delivered (acked or in terminal delivered set)
4047 -- Check if already delivered (acked or in terminal delivered set)
3604− ackedIds <- *ackedIds
4048+ ackedBkt <- readAckedBucket msgBucket
3605− deliveredIds <- *deliveredIds
4049+ delivBkt <- readDelivBucket msgBucket
3606− let inAcked = hasMessage messageId to ackedIds
4050+ let inAcked = hasMessage messageId to ackedBkt
3607− let inDelivered = hasMessage messageId to deliveredIds
4051+ let inDelivered = hasMessage messageId to delivBkt
3608 if inAcked || inDelivered then yield "delivered"
4052 if inAcked || inDelivered then yield "delivered"
3610 -- Extended check: forward nonces have 60-day retention (2x message lifetime)
4054 -- Extended check: forward nonces have 60-day retention (2x message lifetime)
3611− -- and 100k capacity, providing delivery evidence after ackedIds/deliveredIds are pruned
4055+ -- and 100k capacity, providing delivery evidence after ackedIds/deliveredIds are pruned.
3612− forwardNonces <- *forwardNonces
4056+ -- Sharded: reads only the bucket for this recipient's pubkey.
4057+ nonceBkt <- readNonceBucket (pubkeyBucket to)
3613− let inNonces = any forwardNonces (\n -> n.messageId == messageId && n.toKey == to)
4058+ let inNonces = any nonceBkt (\n -> n.messageId == messageId && n.toKey == to)
3614 if inNonces then yield "delivered"
4059 if inNonces then yield "delivered"
3615 else yield "unknown"
4060 else yield "unknown"
3616 yield (respond {status: result, authorized: True {}}))
4061 yield (respond {status: result, authorized: True {}}))
@@ -3664,9 +4109,13 @@ handleRevoke = \targetToken authorization clientIp respond -> do
3664 Nothing {} -> yield (respond {ok: False {}, authorized: True {}})
4109 Nothing {} -> yield (respond {ok: False {}, authorized: True {}})
3665 Just {value: r} -> do
4110 Just {value: r} -> do
3666 when r.gossipOffline (do
4111 when r.gossipOffline (do
4112+ isDraining <- *draining
4114+ then logInfo ("revoke: last session for " ++ r.pubkey ++ " (drain active, skipping redundant offline gossip)")
3667− logInfo ("revoke: last session for " ++ r.pubkey ++ ", gossiping offline")
4116+ logInfo ("revoke: last session for " ++ r.pubkey ++ ", gossiping offline")
3668− when (federationEnabled && count r.peerServers > 0)
4117+ when (federationEnabled && count r.peerServers > 0)
3669− (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
4118+ (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
3670 when (not r.gossipOffline) (logInfo ("revoke: session revoked for " ++ r.pubkey))
4119 when (not r.gossipOffline) (logInfo ("revoke: session revoked for " ++ r.pubkey))
3671 yield (respond {ok: True {}, authorized: True {}}))
4120 yield (respond {ok: True {}, authorized: True {}}))
@@ -3719,9 +4168,15 @@ handleLogout = \authorization clientIp respond -> do
3719 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
4168 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})
3720 Just {value: r} -> do
4169 Just {value: r} -> do
3721 when r.gossipOffline (do
4170 when r.gossipOffline (do
4171+ -- Skip offline gossip during drain: handleDrain already broadcast offline for all local keys,
4172+ -- so per-key gossip here would be a redundant federation round-trip.
4173+ isDraining <- *draining
4175+ then logInfo ("logout: last session for " ++ r.pubkey ++ " (drain active, skipping redundant offline gossip)")
3722− logInfo ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")
4177+ logInfo ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")
3723− when (federationEnabled && count r.peerServers > 0)
4178+ when (federationEnabled && count r.peerServers > 0)
3724− (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
4179+ (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))
3725 yield (respond {ok: True {}, authorized: True {}})
4180 yield (respond {ok: True {}, authorized: True {}})
3727-- POST /federation/verify: Confirm a token was issued by this server.
4182-- POST /federation/verify: Confirm a token was issued by this server.
@@ -3906,13 +4361,14 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3906 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: True {}}
4361 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: True {}}
3908 bucket <- readMessageBucket msgBucket
4363 bucket <- readMessageBucket msgBucket
3909− ackedIds <- *ackedIds
4364+ ackedBkt <- readAckedBucket msgBucket
3910− deliveredIds <- *deliveredIds
4365+ delivBkt <- readDelivBucket msgBucket
3911− forwardNonces <- *forwardNonces
4366+ -- Sharded nonce check: reads only the bucket for this recipient's pubkey (O(n/16) vs O(n)).
4367+ nonceBkt <- readNonceBucket (pubkeyBucket toKey)
3912 -- Nonce check ignores originServer: a (messageId, toKey) pair is globally unique,
4368 -- Nonce check ignores originServer: a (messageId, toKey) pair is globally unique,
3913 -- so replays from any origin should be rejected. This also catches locally-originated
4369 -- so replays from any origin should be rejected. This also catches locally-originated
3914 -- messages (originServer="") whose nonces were recorded by sendMessageCore.
4370 -- messages (originServer="") whose nonces were recorded by sendMessageCore.
3915− let isDup = checkMessageDup bucket messageId toKey || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds || any forwardNonces (\n -> n.messageId == messageId && n.toKey == toKey)
4371+ let isDup = checkMessageDup bucket messageId toKey || hasMessage messageId toKey ackedBkt || hasMessage messageId toKey delivBkt || any nonceBkt (\n -> n.messageId == messageId && n.toKey == toKey)
3916 if isDup
4372 if isDup
3917 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
4373 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
@@ -3946,13 +4402,15 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
3946 set *totalMessageCount = totalMsgCount + 1
4402 set *totalMessageCount = totalMsgCount + 1
3947 writeRcBucket msgBucket (adjustRecipientCount recipientCounts toKey 1)
4403 writeRcBucket msgBucket (adjustRecipientCount recipientCounts toKey 1)
3948 signalMessageForKey toKey seqTs
4404 signalMessageForKey toKey seqTs
3949− -- Record forward nonce for replay protection beyond message/ack dedup lifetime
4405+ -- Record forward nonce for replay protection beyond message/ack dedup lifetime.
3950− fwdNonces <- *forwardNonces
4406+ -- Sharded: writes only to the bucket for this recipient's pubkey.
4407+ let fnb = pubkeyBucket toKey
4408+ fwdNonces <- readNonceBucket fnb
3951 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
4409 let updatedNonces = union fwdNonces [{messageId: messageId, toKey: toKey, originServer: originServer, seenAt: t}]
3952− -- Evict expired nonces first (O(n)), then 1% LRU fallback to minimize replay window
4410+ let maxPerBucket = maxForwardNonces / 16 + 1
3953− set *forwardNonces = if count updatedNonces > maxForwardNonces
4411+ writeNonceBucket fnb (if count updatedNonces > maxPerBucket
3954− then evictNonces updatedNonces maxForwardNonces forwardNonceRetention t
4412+ then evictNonces updatedNonces maxPerBucket forwardNonceRetention t
3955− else updatedNonces
4413+ else updatedNonces)
3956 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
4414 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}
3957 if result.noPresence then do
4415 if result.noPresence then do
3958 logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
4416 logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")
@@ -3983,14 +4441,20 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
3983 -- Clear forward entries for the acking server. Extract removed forwards' toKey
4441 -- Clear forward entries for the acking server. Extract removed forwards' toKey
3984 -- so we can reclaim origin messages that are no longer needed.
4442 -- so we can reclaim origin messages that are no longer needed.
3985 removedForwards <- atomic do
4443 removedForwards <- atomic do
3986− forwards <- *forwards
4444+ forwards <- readAllForwards
3987 let removed = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
4445 let removed = do f <- forwards; where f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid); yield f
3988− set *forwards = forwardsExcluding forwards fromServer dedupedIds
4446+ writeAllForwards (forwardsExcluding forwards fromServer dedupedIds)
3989 yield removed
4447 yield removed
3990 let ackedCount = count removedForwards
4448 let ackedCount = count removedForwards
3991− -- Deferred origin message cleanup: queue for processing in the next prune cycle
4449+ -- Guard: warn if peer acks messages we had no forward entries for. This can indicate
3992− -- after ackGracePeriod elapses. Replaces fork+sleep which accumulated sleeping
4450+ -- a peer acking messages it never received (e.g., the forward response was lost but
3993− -- goroutines under high federation ack volume.
4451+ -- the message was never stored), which would cause premature origin message deletion.
4452+ let unmatched = count dedupedIds - ackedCount
4453+ when (unmatched > 0)
4454+ (logWarn ("federation/ack: " ++ fromServer ++ " acked " ++ show unmatched ++ " messageIds with no matching forward entries — skipping cleanup for those"))
4455+ -- Deferred origin message cleanup: only queue cleanup for messages that actually had
4456+ -- matching forward entries. Messages acked without a forward entry are NOT cleaned up,
4457+ -- preventing premature deletion when a peer acks messages it never actually received.
3994 when (ackedCount > 0) (atomic do
4458 when (ackedCount > 0) (atomic do
3995 cleanups <- *pendingAckCleanups
4459 cleanups <- *pendingAckCleanups
3996 let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
4460 let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}
@@ -4010,6 +4474,13 @@ handleDrain = \authorization respond -> do
4010 let authorized = verifyMaintenanceAuth authorization
4474 let authorized = verifyMaintenanceAuth authorization
4011 if not authorized then yield (respond {ok: False {}, drained: 0})
4475 if not authorized then yield (respond {ok: False {}, drained: 0})
4477+ -- Idempotency: if already draining, return success without re-running drain logic.
4478+ -- Prevents duplicate gossip broadcasts, snapshot exports, and flush forks.
4479+ alreadyDraining <- *draining
4480+ if alreadyDraining then do
4481+ logInfo "drain: already draining, skipping"
4482+ yield (respond {ok: True {}, drained: 0})
4013 drainResult <- atomic do
4484 drainResult <- atomic do
4014 set *draining = True {}
4485 set *draining = True {}
4015 allPresence <- readAllPresence
4486 allPresence <- readAllPresence
@@ -4023,8 +4494,8 @@ handleDrain = \authorization respond -> do
4023 let drainedCount = count drainResult.localEntries
4494 let drainedCount = count drainResult.localEntries
4024 -- Read pending federation state atomically (fix: was read outside atomic)
4495 -- Read pending federation state atomically (fix: was read outside atomic)
4025 flushState <- atomic do
4496 flushState <- atomic do
4026− f <- *forwards
4497+ f <- readAllForwards
4027− a <- *ackedIds
4498+ a <- readAllAcked
4028 let pa = do x <- a; where not (isLocalServer x.originServer); yield x
4499 let pa = do x <- a; where not (isLocalServer x.originServer); yield x
4029 yield {forwards: f, pendingAcks: pa}
4500 yield {forwards: f, pendingAcks: pa}
4030 let pendingForwardCount = count flushState.forwards
4501 let pendingForwardCount = count flushState.forwards
@@ -4033,11 +4504,11 @@ handleDrain = \authorization respond -> do
4033 -- Output as structured JSON with [SNAPSHOT_DATA] prefix for log-shipping pipelines.
4504 -- Output as structured JSON with [SNAPSHOT_DATA] prefix for log-shipping pipelines.
4034 -- This is the last chance to persist state before the server stops.
4505 -- This is the last chance to persist state before the server stops.
4035 drainSnapshot <- atomic do
4506 drainSnapshot <- atomic do
4036− a <- *ackedIds
4507+ a <- readAllAcked
4037 ao <- *ackOverflow
4508 ao <- *ackOverflow
4038− d <- *deliveredIds
4509+ d <- readAllDelivered
4039 ff <- *forwardFailures
4510 ff <- *forwardFailures
4040− fn <- *forwardNonces
4511+ fn <- readAllNonces
4041 bp <- *blockedPeers
4512 bp <- *blockedPeers
4042 yield {version: snapshotVersion, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp}
4513 yield {version: snapshotVersion, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp}
4043 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
4514 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)
@@ -4128,7 +4599,7 @@ handleHealth = \authorization respond -> do
4128 lastWd <- *lastWatchdogAt
4599 lastWd <- *lastWatchdogAt
4129 sa <- *startedAt
4600 sa <- *startedAt
4130 presenceCount <- *totalPresenceCount
4601 presenceCount <- *totalPresenceCount
4131− forwards <- *forwards
4602+ forwardCount <- countAllForwards 0
4132 sessionCount <- *totalSessionCount
4603 sessionCount <- *totalSessionCount
4133 activePolls <- *activePolls
4604 activePolls <- *activePolls
4134 tmc <- *totalMessageCount
4605 tmc <- *totalMessageCount
@@ -4141,7 +4612,7 @@ handleHealth = \authorization respond -> do
4141 -- Capacity pressure: flag when any capped collection exceeds 90% full
4612 -- Capacity pressure: flag when any capped collection exceeds 90% full
4142 let capacityOk = tmc * 10 <= maxMessages * 9
4613 let capacityOk = tmc * 10 <= maxMessages * 9
4143 && presenceCount * 10 <= maxPresenceEntries * 9
4614 && presenceCount * 10 <= maxPresenceEntries * 9
4144− && count forwards * 10 <= maxForwardsEntries * 9
4615+ && forwardCount * 10 <= maxForwardsEntries * 9
4145 && sessionCount * 10 <= maxSessions * 9
4616 && sessionCount * 10 <= maxSessions * 9
4146 && count activePolls * 10 <= maxActivePolls * 9
4617 && count activePolls * 10 <= maxActivePolls * 9
4147 -- Federation: healthy if disabled, in warmup, no peer issues, or recent success
4618 -- Federation: healthy if disabled, in warmup, no peer issues, or recent success
@@ -4152,7 +4623,7 @@ handleHealth = \authorization respond -> do
4152 -- has been exported within the staleness window (federation state will be lost on restart)
4623 -- has been exported within the staleness window (federation state will be lost on restart)
4153 let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup
4624 let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup
4154 || (lastSnap > 0 && t - lastSnap <= snapshotStalenessWarning)
4625 || (lastSnap > 0 && t - lastSnap <= snapshotStalenessWarning)
4155− || (tmc == 0 && presenceCount == 0 && count forwards == 0)
4626+ || (tmc == 0 && presenceCount == 0 && forwardCount == 0)
4156 -- Watchdog: healthy if recently bumped or in warmup (watchdog starts after 10x prune interval)
4627 -- Watchdog: healthy if recently bumped or in warmup (watchdog starts after 10x prune interval)
4157 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
4628 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)
4158 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
4629 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),
@@ -4202,24 +4673,24 @@ handleStats = \authorization respond -> do
4202 core <- atomic do
4673 core <- atomic do
4203 sc <- *totalSessionCount
4674 sc <- *totalSessionCount
4204 pc <- *totalPresenceCount
4675 pc <- *totalPresenceCount
4205− f <- *forwards
4676+ f <- readAllForwards
4206 c <- *challenges
4677 c <- *challenges
4207− a <- *ackedIds
4678+ a <- readAllAcked
4208 ao <- *ackOverflow
4679 ao <- *ackOverflow
4209− d <- *deliveredIds
4680+ d <- readAllDelivered
4210 ff <- *forwardFailures
4681 ff <- *forwardFailures
4211 ap <- *activePolls
4682 ap <- *activePolls
4212 tmc <- *totalMessageCount
4683 tmc <- *totalMessageCount
4213 rc <- countAllRcEntries 0
4684 rc <- countAllRcEntries 0
4214 fi <- *federationInFlight
4685 fi <- *federationInFlight
4215 rg <- *recentGossip
4686 rg <- *recentGossip
4216− fn <- *forwardNonces
4687+ fnCount <- countAllNonces 0
4217 pac <- *pendingAckCleanups
4688 pac <- *pendingAckCleanups
4218 bp <- *blockedPeers
4689 bp <- *blockedPeers
4219 ger <- *gossipEventRates
4690 ger <- *gossipEventRates
4220 vir <- *verifyIpRates
4691 vir <- *verifyIpRates
4221 ukir <- *uniqueKeyIpRates
4692 ukir <- *uniqueKeyIpRates
4222− yield {sc: sc, pc: pc, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg, fn: fn, pac: pac, bp: bp, ger: ger, vir: vir, ukir: ukir}
4693+ yield {sc: sc, pc: pc, f: f, c: c, a: a, ao: ao, d: d, ff: ff, ap: ap, tmc: tmc, rc: rc, fi: fi, rg: rg, fnCount: fnCount, pac: pac, bp: bp, ger: ger, vir: vir, ukir: ukir}
4223 counters <- atomic do
4694 counters <- atomic do
4224 sr <- *sendRates
4695 sr <- *sendRates
4225 cr <- *challengeRates
4696 cr <- *challengeRates
@@ -4256,7 +4727,7 @@ handleStats = \authorization respond -> do
4256 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
4727 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
4257 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt
4728 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt
4258 yield (respond {ok: True {},
4729 yield (respond {ok: True {},
4259− capacity: {sessions: sessions, sessionsMax: maxSessions, sessionsPct: sessPct, messages: messages, messagesMax: maxMessages, messagesPct: msgPct, totalMessageCount: messages, presence: presence, presenceMax: maxPresenceEntries, presencePct: presPct, forwards: fwds, forwardsMax: maxForwardsEntries, forwardsPct: fwdPct, challenges: count core.c, challengesMax: maxChallenges, ackedIds: count core.a, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: count core.d, deliveredIdsMax: maxDeliveredIds, forwardNonces: count core.fn, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
4730+ capacity: {sessions: sessions, sessionsMax: maxSessions, sessionsPct: sessPct, messages: messages, messagesMax: maxMessages, messagesPct: msgPct, totalMessageCount: messages, presence: presence, presenceMax: maxPresenceEntries, presencePct: presPct, forwards: fwds, forwardsMax: maxForwardsEntries, forwardsPct: fwdPct, challenges: count core.c, challengesMax: maxChallenges, ackedIds: count core.a, ackedIdsMax: maxAckedIds, ackOverflow: count core.ao, ackOverflowMax: maxAckOverflow, deliveredIds: count core.d, deliveredIdsMax: maxDeliveredIds, forwardNonces: core.fnCount, forwardNoncesMax: maxForwardNonces, activePolls: count core.ap, activePollsMax: maxActivePolls, recipientCounts: count core.rc},
4260 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca},
4731 federation: {enabled: federationEnabled, inFlight: count core.fi, hungSlots: hungFedSlots, avgLatencyMs: fedAvg, maxLatencyMs: counters.flm, callCount: counters.flc, forwardFailures: count core.ff, oldestForwardAge: oldestFwdAge, oldestAckAge: oldestAckAge, recentGossip: count core.rg, blockedPeers: count core.bp, drainCompletedAt: counters.dca},
4261 rates: {send: count counters.sr, challenge: count counters.cr, federation: count counters.fr, federationPair: count counters.fpr, global: counters.grc, ip: count counters.ir, lookup: count counters.lr, recv: count counters.rr, pair: count counters.prr, gossipEvent: count core.ger, verifyIp: count core.vir, uniqueKeyIp: count core.ukir, pendingAckCleanups: count core.pac},
4732 rates: {send: count counters.sr, challenge: count counters.cr, federation: count counters.fr, federationPair: count counters.fpr, global: counters.grc, ip: count counters.ir, lookup: count counters.lr, recv: count counters.rr, pair: count counters.prr, gossipEvent: count core.ger, verifyIp: count core.vir, uniqueKeyIp: count core.ukir, pendingAckCleanups: count core.pac},
4262 timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr}})
4733 timing: {pruneCount: counters.pc, pruneRunning: counters.pr, heartbeatCount: counters.hc, uptime: t - counters.sa, lastPruneAt: counters.lp, lastHeartbeatAt: counters.lh, draining: counters.dr}})
@@ -4271,7 +4742,7 @@ handleStatsPeers = \authorization respond -> do
4271 if authorized
4742 if authorized
4273 failures <- *forwardFailures
4744 failures <- *forwardFailures
4274− let peers = do f <- failures; yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoff f.failures, retriable: shouldRetryServer f.server t failures}
4745+ let peers = do f <- failures; yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoffFor f.failures f.server, retriable: shouldRetryServer f.server t failures}
4275 yield (respond {peers: peers, ok: True {}})
4746 yield (respond {peers: peers, ok: True {}})
4276 else yield (respond {peers: [], ok: False {}})
4747 else yield (respond {peers: [], ok: False {}})
@@ -4288,16 +4759,18 @@ handleStatsKey = \pubkey authorization respond -> do
4288 -- Single bucket read: all entries for this pubkey are in the same presence bucket
4759 -- Single bucket read: all entries for this pubkey are in the same presence bucket
4289 keyPresenceBkt <- readPresenceBucket (pubkeyBucket pubkey)
4760 keyPresenceBkt <- readPresenceBucket (pubkeyBucket pubkey)
4290 allSessions <- readAllSessions
4761 allSessions <- readAllSessions
4291− forwards <- *forwards
4762+ forwards <- readFwdBucket (pubkeyBucket pubkey)
4292 activePolls <- *activePolls
4763 activePolls <- *activePolls
4293 let keyMessages = count (do m <- keyBucket; where m.toKey == pubkey; yield m)
4764 let keyMessages = count (do m <- keyBucket; where m.toKey == pubkey; yield m)
4294 let keyPresence = do p <- keyPresenceBkt; where p.pubkey == pubkey && p.expiresAt > t; yield {server: p.server, expiresAt: p.expiresAt}
4765 let keyPresence = do p <- keyPresenceBkt; where p.pubkey == pubkey && p.expiresAt > t; yield {server: p.server, expiresAt: p.expiresAt}
4295 let keySessions = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
4766 let keySessions = count (do s <- allSessions; where s.pubkey == pubkey; yield s)
4296− let keyForwards = count (do f <- forwards; where f.toKey == pubkey; yield f)
4767+ let keyForwardEntries = do f <- forwards; where f.toKey == pubkey; yield f
4768+ let keyForwards = count keyForwardEntries
4769+ let keyForwardDetails = do f <- keyForwardEntries; yield {server: f.toServer, retries: f.retries, createdAt: f.createdAt}
4297 let keyPolls = count (do p <- activePolls; where p.pubkey == pubkey; yield p)
4770 let keyPolls = count (do p <- activePolls; where p.pubkey == pubkey; yield p)
4298− yield {messages: keyMessages, presence: keyPresence, sessions: keySessions, forwards: keyForwards, activePolls: keyPolls}
4771+ yield {messages: keyMessages, presence: keyPresence, sessions: keySessions, forwards: keyForwards, forwardDetails: keyForwardDetails, activePolls: keyPolls}
4299− yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})
4772+ yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, forwardDetails: result.forwardDetails, activePolls: result.activePolls, ok: True {}})
4300− else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})
4773+ else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, forwardDetails: [], activePolls: 0, ok: False {}})
4302emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: []}
4775emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: []}
@@ -4312,12 +4785,12 @@ handleSnapshot = \authorization respond -> do
4312 t <- now
4785 t <- now
4313 result <- atomic do
4786 result <- atomic do
4314 p <- readAllPresence
4787 p <- readAllPresence
4315− a <- *ackedIds
4788+ a <- readAllAcked
4316 ao <- *ackOverflow
4789 ao <- *ackOverflow
4317− d <- *deliveredIds
4790+ d <- readAllDelivered
4318 ff <- *forwardFailures
4791 ff <- *forwardFailures
4319− f <- *forwards
4792+ f <- readAllForwards
4320− fn <- *forwardNonces
4793+ fn <- readAllNonces
4321 bp <- *blockedPeers
4794 bp <- *blockedPeers
4322 yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}
4795 yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}
4323 atomic do set *lastSnapshotExportAt = t
4796 atomic do set *lastSnapshotExportAt = t
@@ -4362,33 +4835,33 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD
4362 set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)
4835 set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)
4363 set *presenceVersion = t
4836 set *presenceVersion = t
4364 -- Union dedup state, respecting capacity limits (using validated data)
4837 -- Union dedup state, respecting capacity limits (using validated data)
4365− ackedIds <- *ackedIds
4838+ ackedIds <- readAllAcked
4366 let mergedAcked = union ackedIds validAcked
4839 let mergedAcked = union ackedIds validAcked
4367 let cappedAcked = takeRelation maxAckedIds mergedAcked
4840 let cappedAcked = takeRelation maxAckedIds mergedAcked
4368− set *ackedIds = cappedAcked
4841+ writeAllAcked cappedAcked
4369 ackOverflow <- *ackOverflow
4842 ackOverflow <- *ackOverflow
4370 let mergedOverflow = union ackOverflow validOverflow
4843 let mergedOverflow = union ackOverflow validOverflow
4371 set *ackOverflow = takeRelation maxAckOverflow mergedOverflow
4844 set *ackOverflow = takeRelation maxAckOverflow mergedOverflow
4372− deliveredIds <- *deliveredIds
4845+ deliveredIds <- readAllDelivered
4373 let mergedDelivered = union deliveredIds deliveredIdsData
4846 let mergedDelivered = union deliveredIds deliveredIdsData
4374 let cappedDelivered = takeRelation maxDeliveredIds mergedDelivered
4847 let cappedDelivered = takeRelation maxDeliveredIds mergedDelivered
4375− set *deliveredIds = cappedDelivered
4848+ writeAllDelivered cappedDelivered
4376 -- Restore peer failure records
4849 -- Restore peer failure records
4377 forwardFailures <- *forwardFailures
4850 forwardFailures <- *forwardFailures
4378 let mergedFailures = mergeByKey (\a b -> a.server == b.server) (\x -> x.failedAt) forwardFailures forwardFailuresData
4851 let mergedFailures = mergeByKey (\a b -> a.server == b.server) (\x -> x.failedAt) forwardFailures forwardFailuresData
4379 let cappedFailures = takeRelation maxForwardFailures mergedFailures
4852 let cappedFailures = takeRelation maxForwardFailures mergedFailures
4380 set *forwardFailures = cappedFailures
4853 set *forwardFailures = cappedFailures
4381 -- Restore forward entries (dedup by messageId + toServer, keep fresher on conflict; using validated data)
4854 -- Restore forward entries (dedup by messageId + toServer, keep fresher on conflict; using validated data)
4382− forwards <- *forwards
4855+ existingForwards <- readAllForwards
4383− let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) forwards validForwards
4856+ let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) existingForwards validForwards
4384 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
4857 let cappedForwards = takeRelation maxForwardsEntries mergedForwards
4385− set *forwards = cappedForwards
4858+ writeAllForwards cappedForwards
4386 -- Restore forward nonces (dedup by messageId + toKey + originServer, keep fresher on conflict)
4859 -- Restore forward nonces (dedup by messageId + toKey + originServer, keep fresher on conflict)
4387− forwardNonces <- *forwardNonces
4860+ existingNonces <- readAllNonces
4388 let validNonces = do n <- forwardNoncesData; where n.seenAt <= t && t - n.seenAt < forwardNonceRetention; yield n
4861 let validNonces = do n <- forwardNoncesData; where n.seenAt <= t && t - n.seenAt < forwardNonceRetention; yield n
4389− let mergedNonces = mergeByKey (\a b -> a.messageId == b.messageId && a.toKey == b.toKey && a.originServer == b.originServer) (\x -> x.seenAt) forwardNonces validNonces
4862+ let mergedNonces = mergeByKey (\a b -> a.messageId == b.messageId && a.toKey == b.toKey && a.originServer == b.originServer) (\x -> x.seenAt) existingNonces validNonces
4390 let cappedNonces = takeRelation maxForwardNonces mergedNonces
4863 let cappedNonces = takeRelation maxForwardNonces mergedNonces
4391− set *forwardNonces = cappedNonces
4864+ writeAllNonces cappedNonces
4392 -- Restore blocked peers (union with existing, cap at maxBlockedPeers)
4865 -- Restore blocked peers (union with existing, cap at maxBlockedPeers)
4393 blocked <- *blockedPeers
4866 blocked <- *blockedPeers
4394 let mergedBlocked = fold (\acc b -> if any acc (\x -> x == b) then acc else union acc [b]) blocked blockedPeersData
4867 let mergedBlocked = fold (\acc b -> if any acc (\x -> x == b) then acc else union acc [b]) blocked blockedPeersData
@@ -4431,14 +4904,15 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
4431 recipientCounts <- readRcBucket b
4904 recipientCounts <- readRcBucket b
4432 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
4905 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))
4433 let ackedMsgIds = do m <- toAck; yield m.messageId
4906 let ackedMsgIds = do m <- toAck; yield m.messageId
4434− forwards <- *forwards
4907+ let fb2 = pubkeyBucket pubkey
4908+ fwdBkt2 <- readFwdBucket fb2
4435− set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
4909+ writeFwdBucket fb2 (do f <- fwdBkt2; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
4436 let ackedFederated = federatedAckEntries toAck t
4910 let ackedFederated = federatedAckEntries toAck t
4437− ackedIds <- *ackedIds
4911+ ackedBkt <- readAckedBucket b
4438− let combinedAckedIds = union ackedIds ackedFederated
4912+ let combinedAckedIds = union ackedBkt ackedFederated
4439 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
4913 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)
4440 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
4914 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
4441− set *ackedIds = eviction.kept
4915+ writeAckedBucket b eviction.kept
4442 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
4916 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}
4443 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}
4917 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}
4444 fork (when federationEnabled (do
4918 fork (when federationEnabled (do
@@ -4503,10 +4977,11 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
4503 set *totalMessageCount = max 0 (tc - removed)
4977 set *totalMessageCount = max 0 (tc - removed)
4504 rc <- readRcBucket b
4978 rc <- readRcBucket b
4505 writeRcBucket b (adjustRecipientCount rc pubkey (0 - removed))
4979 writeRcBucket b (adjustRecipientCount rc pubkey (0 - removed))
4506− -- Cancel pending forwards for deleted messages
4980+ -- Cancel pending forwards for deleted messages (bucket read: O(n/16))
4507 let deletedMsgIds = do m <- toDelete; yield m.messageId
4981 let deletedMsgIds = do m <- toDelete; yield m.messageId
4508− forwards <- *forwards
4982+ let dfb = pubkeyBucket pubkey
4983+ fwdBkt <- readFwdBucket dfb
4509− set *forwards = do f <- forwards; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f
4984+ writeFwdBucket dfb (do f <- fwdBkt; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
4510 yield removed
4985 yield removed
4511 else yield 0
4986 else yield 0
4512 yield (respond {ok: True {}, authorized: True {}, deletedCount: deletedCount}))
4987 yield (respond {ok: True {}, authorized: True {}, deletedCount: deletedCount}))
@@ -4555,7 +5030,7 @@ sendBatchInner = \senderKey messages t acc ->
4555 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
5030 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)
4556 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
5031 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))
4557 result <- if t - msg.timestamp > messageMaxAge
5032 result <- if t - msg.timestamp > messageMaxAge
4558− then yield {messageId: "", status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, retryAfterMs: 0}
5033+ then yield {messageId: effectiveMsgId, status: MsRejected {}, remoteServers: [], toKey: "", blob: "", ts: 0, retryAfterMs: 0}
4560 let clampedTs = clampTimestamp msg.timestamp t
5035 let clampedTs = clampTimestamp msg.timestamp t
4561 sendResult <- atomic do
5036 sendResult <- atomic do
@@ -4677,6 +5152,36 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do
4677 (fork (sendBatchedGossip result.peerServers [{eventType: GossipOffline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))
5152 (fork (sendBatchedGossip result.peerServers [{eventType: GossipOffline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))
4678 yield (respond {ok: True {}, revokedCount: result.revokedCount})
5153 yield (respond {ok: True {}, revokedCount: result.revokedCount})
5155+-- POST /maintenance/messages/purge: Purge all pending messages for a specific pubkey.
5156+-- Use for inbox-bombing abuse response: clears the recipient's mailbox without
5157+-- requiring their session token. Also cancels pending forwards for purged messages.
5158+handleAdminPurgeMessages = \pubkey authorization respond -> do
5159+ let authorized = verifyMaintenanceAuth authorization
5160+ if not authorized then yield (respond {ok: False {}, purgedCount: 0})
5163+ purgedCount <- atomic do
5164+ let b = pubkeyBucket pubkey
5165+ bucket <- readMessageBucket b
5166+ let toDelete = do m <- bucket; where m.toKey == pubkey; yield m
5167+ let removed = count toDelete
5168+ if removed > 0 then do
5169+ writeMessageBucket b (do m <- bucket; where m.toKey != pubkey; yield m)
5170+ tc <- *totalMessageCount
5171+ set *totalMessageCount = max 0 (tc - removed)
5172+ rc <- readRcBucket b
5173+ writeRcBucket b (adjustRecipientCount rc pubkey (0 - removed))
5174+ -- Cancel pending forwards for purged messages (bucket read: O(n/16))
5175+ let deletedMsgIds = do m <- toDelete; yield m.messageId
5176+ let pfb = pubkeyBucket pubkey
5177+ fwdBkt <- readFwdBucket pfb
5178+ writeFwdBucket pfb (do f <- fwdBkt; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)
5181+ when (purgedCount > 0)
5182+ (logInfo ("admin: purged " ++ show purgedCount ++ " messages for " ++ pubkey))
5183+ yield (respond {ok: True {}, purgedCount: purgedCount})
4680-- GET /maintenance/rate-limits: Return current effective rate limits.
5185-- GET /maintenance/rate-limits: Return current effective rate limits.
4681handleGetRateLimits = \authorization respond -> do
5186handleGetRateLimits = \authorization respond -> do
4682 let authorized = verifyMaintenanceAuth authorization
5187 let authorized = verifyMaintenanceAuth authorization
@@ -4737,6 +5242,7 @@ requestPath = \req -> case req of
4737 PingPeer {} -> "POST /maintenance/ping"
5242 PingPeer {} -> "POST /maintenance/ping"
4738 AdminListSessions {} -> "GET /maintenance/sessions"
5243 AdminListSessions {} -> "GET /maintenance/sessions"
4739 AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"
5244 AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"
5245+ AdminPurgeMessages {} -> "POST /maintenance/messages/purge"
4740 GetRateLimits {} -> "GET /maintenance/rate-limits"
5246 GetRateLimits {} -> "GET /maintenance/rate-limits"
4741 SetRateLimits {} -> "POST /maintenance/rate-limits"
5247 SetRateLimits {} -> "POST /maintenance/rate-limits"
4742 Health {} -> "GET /health"
5248 Health {} -> "GET /health"
@@ -4773,6 +5279,7 @@ requestClientIp = \req -> case req of
4773 PingPeer {} -> ""
5279 PingPeer {} -> ""
4774 AdminListSessions {} -> ""
5280 AdminListSessions {} -> ""
4775 AdminRevokeSessions {} -> ""
5281 AdminRevokeSessions {} -> ""
5282+ AdminPurgeMessages {} -> ""
4776 GetRateLimits {} -> ""
5283 GetRateLimits {} -> ""
4777 SetRateLimits {} -> ""
5284 SetRateLimits {} -> ""
4778 Health {} -> ""
5285 Health {} -> ""
@@ -4812,6 +5319,7 @@ serveInner = \req -> case req of
4812 PingPeer {server, authorization, respond} -> handlePingPeer server authorization respond
5319 PingPeer {server, authorization, respond} -> handlePingPeer server authorization respond
4813 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
5320 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
4814 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
5321 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
5322+ AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond
4815 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
5323 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
4816 SetRateLimits {globalRequest, ipRate, sendRate, federationRate, authGlobal, recvRate, authorization, respond} -> handleSetRateLimits globalRequest ipRate sendRate federationRate authGlobal recvRate authorization respond
5324 SetRateLimits {globalRequest, ipRate, sendRate, federationRate, authGlobal, recvRate, authorization, respond} -> handleSetRateLimits globalRequest ipRate sendRate federationRate authGlobal recvRate authorization respond
4817 Health {authorization, respond} -> handleHealth authorization respond
5325 Health {authorization, respond} -> handleHealth authorization respond
@@ -4958,13 +5466,13 @@ backgroundSnapshot = do
4958 tmc <- *totalMessageCount
5466 tmc <- *totalMessageCount
4959 sc <- *totalSessionCount
5467 sc <- *totalSessionCount
4960 pc <- *totalPresenceCount
5468 pc <- *totalPresenceCount
4961− f <- *forwards
5469+ fwdCount <- countAllForwards 0
4962− a <- *ackedIds
5470+ ackedCount <- countAllAcked 0
4963 ao <- *ackOverflow
5471 ao <- *ackOverflow
4964− d <- *deliveredIds
5472+ deliveredCount <- countAllDelivered 0
4965 ff <- *forwardFailures
5473 ff <- *forwardFailures
4966 pac <- *pendingAckCleanups
5474 pac <- *pendingAckCleanups
4967− yield {messages: tmc, sessions: sc, presence: pc, forwards: count f, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff, pendingAckCleanups: count pac}
5475+ yield {messages: tmc, sessions: sc, presence: pc, forwards: fwdCount, ackedIds: ackedCount, ackOverflow: count ao, deliveredIds: deliveredCount, forwardFailures: count ff, pendingAckCleanups: count pac}
4968 logInfo ("snapshot: messages=" ++ show result.messages ++ " sessions=" ++ show result.sessions ++ " presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " ackOverflow=" ++ show result.ackOverflow ++ " deliveredIds=" ++ show result.deliveredIds ++ " forwardFailures=" ++ show result.forwardFailures ++ " pendingAckCleanups=" ++ show result.pendingAckCleanups)
5476 logInfo ("snapshot: messages=" ++ show result.messages ++ " sessions=" ++ show result.sessions ++ " presence=" ++ show result.presence ++ " forwards=" ++ show result.forwards ++ " ackedIds=" ++ show result.ackedIds ++ " ackOverflow=" ++ show result.ackOverflow ++ " deliveredIds=" ++ show result.deliveredIds ++ " forwardFailures=" ++ show result.forwardFailures ++ " pendingAckCleanups=" ++ show result.pendingAckCleanups)
4969 t <- now
5477 t <- now
4970 -- Periodic full snapshot auto-export: emit [SNAPSHOT_DATA] JSON to stdout so a
5478 -- Periodic full snapshot auto-export: emit [SNAPSHOT_DATA] JSON to stdout so a
@@ -4975,12 +5483,12 @@ backgroundSnapshot = do
4975 when (hasState && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do
5483 when (hasState && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do
4976 exportData <- atomic do
5484 exportData <- atomic do
4977 p <- readAllPresence
5485 p <- readAllPresence
4978− a <- *ackedIds
5486+ a <- readAllAcked
4979 ao <- *ackOverflow
5487 ao <- *ackOverflow
4980− d <- *deliveredIds
5488+ d <- readAllDelivered
4981 ff <- *forwardFailures
5489 ff <- *forwardFailures
4982− f <- *forwards
5490+ f <- readAllForwards
4983− fn <- *forwardNonces
5491+ fn <- readAllNonces
4984 bp <- *blockedPeers
5492 bp <- *blockedPeers
4985 yield {version: snapshotVersion, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}
5493 yield {version: snapshotVersion, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}
4986 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
5494 println ("[SNAPSHOT_DATA] " ++ toJson exportData)
@@ -5047,18 +5555,19 @@ startupProbeNext = \peers probed ->
5047 Just {value: server} -> do
5555 Just {value: server} -> do
5048 let rest = do p <- peers; where p != server; yield p
5556 let rest = do p <- peers; where p != server; yield p
5049 startTime <- now
5557 startTime <- now
5050− auth <- generateFederationAuth "gossip"
5558+ -- Unauthenticated probe: tests TCP/TLS/HTTP connectivity without generating
5559+ -- federation auth tokens. Peers reject with "unauthorized" (expected), which
5560+ -- still confirms reachability. Avoids leaking single-use tokens to potentially
5561+ -- untrusted peers during startup (DNS spoofing window before first real gossip).
5051 result <- fetchFederation server (RecvGossip {
5562 result <- fetchFederation server (RecvGossip {
5052− authorization: auth,
5563+ authorization: "",
5053 events: [], hopsRemaining: 1, fromServer: serverHostN,
5564 events: [], hopsRemaining: 1, fromServer: serverHostN,
5054 requestTimestamp: startTime, correlationId: ""})
5565 requestTimestamp: startTime, correlationId: ""})
5055 endTime <- now
5566 endTime <- now
5056 let latency = endTime - startTime
5567 let latency = endTime - startTime
5057 case result of
5568 case result of
5058 Err e -> logWarn ("startup: peer " ++ server ++ " unreachable (" ++ show e ++ ") — check DNS/firewall/TLS")
5569 Err e -> logWarn ("startup: peer " ++ server ++ " unreachable (" ++ show e ++ ") — check DNS/firewall/TLS")
5060− then logInfo ("startup: peer " ++ server ++ " ok (" ++ show latency ++ "ms)")
5570+ Ok r -> logInfo ("startup: peer " ++ server ++ " reachable (" ++ show latency ++ "ms)")
5061− else logWarn ("startup: peer " ++ server ++ " rejected (" ++ r.error ++ ", " ++ show latency ++ "ms)")
5062 startupProbeNext rest (probed + 1)
5571 startupProbeNext rest (probed + 1)
5064main = do
5573main = do