460356e · skrepka · git.iko.soy

git.iko.soy/skrepka

Implement 12 server.knot improvements: forward retry jitter, batch messageId return, drain optimization, startup probe security, stats enhancement, admin message purge, cleanup guards, presence refresh optimization, and 16-way sharding of forward nonces, forwards, and acked/delivered collections for O(n/16) hot-path performance

commit460356e7a5caeae263311a8e296acc8459ba1dc2 authoriko <mail@iko.soy> dateSat 25 Apr 2026, 13:39 (6 months ago) parenta77ed7c

1 file changed+687−178

compare with master

unifiedsplit

server.knot+687−178

server.knot+687−178

@@ -431,10 +431,43 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}

431431-- Monotonic counter ensuring unique receivedAt ordering even within the same millisecond

432432*seqCounter : Timestamp

433433

434−*forwards : [ForwardEntry]

435−

436−-- Messages acked by local recipients, pending federation ack to origin server

437−*ackedIds : [AckedEntry]

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]

451+*fwdBucket15 : [ForwardEntry]

452+

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]

470+*ackedBucket15 : [AckedEntry]

438471

439472*lastPrunedAt : Timestamp

440473

@@ -477,8 +510,24 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}

477510-- Overflow buffer: acks evicted from ackedIds before they could be sent to origin

478511*ackOverflow : [AckedEntry]

479512

480−-- Final dedup set: messages confirmed delivered (ack sent or overflow drained)

481−*deliveredIds : [DeliveredEntry]

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]

530+*delivBucket15 : [DeliveredEntry]

482531

483532*federationRates : [FederationRateEntry]

484533

@@ -575,8 +624,24 @@ type BatchResult = {messageId: OptionalMessageId, status: MessageStatus}

575624-- scope binds the token to a specific request type (gossip/forward/ack) to prevent cross-use

576625*pendingVerifications : [{token: Text, createdAt: Timestamp, scope: Text}]

577626

578−-- Forward nonce dedup: (messageId, toKey, originServer, seenAt) for replay protection beyond message lifetime

579−*forwardNonces : [ForwardNonce]

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]

644+*nonceBucket15 : [ForwardNonce]

580645

581646-- Federation verify per-IP rate limiting (tighter than general IP rate)

582647*verifyIpRates : [IpRateEntry]

@@ -759,6 +824,10 @@ route MaintenanceApi where

759824 POST {pubkey: PubkeyHex} /maintenance/sessions/revoke-all headers {authorization: Text}

760825 -> {ok: Bool, revokedCount: Int}

761826 = 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

762831 -- Read current effective rate limits (compile-time defaults + runtime overrides)

763832 GET /maintenance/rate-limits headers {authorization: Text}

764833 -> {ok: Bool, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int}

@@ -785,7 +854,7 @@ route StatsApi where

785854 -> {peers: [PeerStatus], ok: Bool}

786855 = StatsPeers

787856 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}

789858 = StatsKey

790859

@@ -1452,6 +1521,17 @@ transformAllPresenceBuckets = \f -> do

14521521 b14 <- *presenceBucket14; set *presenceBucket14 = f b14

14531522 b15 <- *presenceBucket15; set *presenceBucket15 = f b15

14541523

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 []

1529+ else do

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)

1534+

14551535-- ============================================

14561536-- Message Bucket Helpers

14571537-- ============================================

@@ -1721,6 +1801,343 @@ checkMessageDup : [Message] -> Text -> PubkeyHex -> Bool

17211801checkMessageDup = \messages messageId toKey ->

17221802 messageId != "" && any messages (\m -> m.messageId == messageId && m.toKey == toKey)

17231803

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).

1809+

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

1828+

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

1847+

1848+readAllNonces : IO [ForwardNonce]

1849+readAllNonces = do

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)))))))))))))))

1867+

1868+countAllNonces : Int -> IO Int

1869+countAllNonces = \b ->

1870+ if b > 15 then yield 0

1871+ else do

1872+ bucket <- readNonceBucket b

1873+ rest <- countAllNonces (b + 1)

1874+ yield (count bucket + rest)

1875+

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

1914+

1915+-- Prune nonces across all 16 buckets

1916+pruneAllNonceBuckets : Int -> Timestamp -> IO ()

1917+pruneAllNonceBuckets = \b t ->

1918+ if b > 15 then yield {}

1919+ else do

1920+ bucket <- readNonceBucket b

1921+ writeNonceBucket b (do n <- bucket; where t - n.seenAt < forwardNonceRetention; yield n)

1922+ pruneAllNonceBuckets (b + 1) t

1923+

1924+-- ============================================

1925+-- Forward Bucket Helpers

1926+-- ============================================

1927+-- Forwards sharded into 16 buckets by pubkeyBucket(toKey).

1928+

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

1947+

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

1966+

1967+readAllForwards : IO [ForwardEntry]

1968+readAllForwards = do

1969+ f0 <- *fwdBucket0

1970+ f1 <- *fwdBucket1

1971+ f2 <- *fwdBucket2

1972+ f3 <- *fwdBucket3

1973+ f4 <- *fwdBucket4

1974+ f5 <- *fwdBucket5

1975+ f6 <- *fwdBucket6

1976+ f7 <- *fwdBucket7

1977+ f8 <- *fwdBucket8

1978+ f9 <- *fwdBucket9

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)))))))))))))))

1986+

1987+countAllForwards : Int -> IO Int

1988+countAllForwards = \b ->

1989+ if b > 15 then yield 0

1990+ else do

1991+ bucket <- readFwdBucket b

1992+ rest <- countAllForwards (b + 1)

1993+ yield (count bucket + rest)

1994+

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

2032+

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

2046+

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

2057+

2058+readAllAcked : IO [AckedEntry]

2059+readAllAcked = do

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)))))))))))))))

2065+

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

2082+

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)

2086+

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

2100+

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

2111+

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)))))))))))))))

2119+

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

2136+

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)

2140+

17242141-- ============================================

17252142-- Federation In-Flight Tracking

17262143-- ============================================

@@ -2119,18 +2536,18 @@ pruneExpired = \t -> do

21192536 -- Prune federation state

21202537 atomic do

21212538 let retainCutoff = t - messageMaxAge

2122− forwards <- *forwards

2539+ forwards <- readAllForwards

21232540 -- Clear all forwards when federation is disabled (they can never be retried)

2124− set *forwards = if not federationEnabled then []

2125− else do f <- forwards; where f.createdAt > retainCutoff; yield f

2126− ackedIds <- *ackedIds

2127− set *ackedIds = do a <- ackedIds; where a.ackedAt > retainCutoff; yield a

2541+ writeAllForwards (if not federationEnabled then []

2542+ else do f <- forwards; where f.createdAt > retainCutoff; yield f)

2543+ ackedIds <- readAllAcked

2544+ writeAllAcked (do a <- ackedIds; where a.ackedAt > retainCutoff; yield a)

21282545 forwardFailures <- *forwardFailures

21292546 set *forwardFailures = do f <- forwardFailures; where t - f.failedAt < maxServerDowntime; yield f

21302547 ackOverflow <- *ackOverflow

21312548 set *ackOverflow = do a <- ackOverflow; where a.ackedAt > retainCutoff; yield a

2132− deliveredIds <- *deliveredIds

2133− set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > retainCutoff; yield d

2549+ deliveredIds <- readAllDelivered

2550+ writeAllDelivered (do d <- deliveredIds; where d.deliveredAt > retainCutoff; yield d)

21342551 -- Prune rate limiters (split into smaller atomic blocks to reduce lock contention)

21352552 atomic do

21362553 sendRates <- *sendRates

@@ -2189,9 +2606,8 @@ pruneExpired = \t -> do

21892606 -- Prune expired federation verification tokens

21902607 pending <- *pendingVerifications

21912608 set *pendingVerifications = do p <- pending; where t - p.createdAt < verificationTokenExpiry; yield p

2192− -- Prune expired forward nonces (extended replay protection)

2193− forwardNonces <- *forwardNonces

2194− set *forwardNonces = do n <- forwardNonces; where t - n.seenAt < forwardNonceRetention; yield n

2609+ -- Prune expired forward nonces per bucket (extended replay protection)

2610+ pruneAllNonceBuckets 0 t

21952611 -- Prune gossip event rate entries

21962612 gossipEventRates <- *gossipEventRates

21972613 set *gossipEventRates = do r <- gossipEventRates; where t - r.windowStart < gossipEventRateWindowMs; yield r

@@ -2432,9 +2848,10 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do

24322848 then onUnauth

24332849 else onAuth pubkey

24342850

2435−removeForward = \mid srv -> atomic do

2436− forwards <- *forwards

2437− set *forwards = forwardsExcluding forwards srv [mid]

2851+removeForward = \mid srv toKey -> atomic do

2852+ let b = pubkeyBucket toKey

2853+ fwdBkt <- readFwdBucket b

2854+ writeFwdBucket b (forwardsExcluding fwdBkt srv [mid])

24382855

24392856-- Reclaim origin messages after federation ack confirms delivery.

24402857-- 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

24512868 atomic do

24522869 -- All entries in this batch share pubkeyBucket == b, so only that presence bucket is needed

24532870 bucketPresence <- readPresenceBucket b

2454− forwards <- *forwards

2871+ fwdBkt <- readFwdBucket b

24552872 msgs <- readMessageBucket b

24562873 tc <- *totalMessageCount

24572874 rc <- readRcBucket b

24582875 let cleanup = fold (\acc rf ->

24592876 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

24612878 if hasLocal || hasOtherFwds then acc

24622879 else case headOf (do m <- acc.msgs; where m.messageId == rf.messageId && m.toKey == rf.toKey && t - m.receivedAt >= ackGracePeriod; yield m) of

24632880 Nothing {} -> acc

@@ -2476,11 +2893,20 @@ retryBackoff : Int -> Int<Ms>

24762893retryBackoff = \failures ->

24772894 min retryBackoffMax (retryBackoffBase * pow2 (max 0 (failures - 1)))

24782895

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)

2904+

24792905shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool

24802906shouldRetryServer = \server t failures ->

24812907 case headOf (do f <- failures; where f.server == server; yield f) of

24822908 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

24842910

24852911-- Extended retry check: also skip servers with hung in-flight requests or on the blocklist

24862912shouldRetryServerEx : Text -> Timestamp -> [ForwardFailure] -> [FederationInFlightEntry] -> [Text] -> Bool

@@ -2513,14 +2939,15 @@ clearFailure = \server -> do

25132939 set *lastFederationSuccessAt = t

25142940

25152941-- Atomically increment a forward entry's retry count; drop if maxForwardRetries exceeded

2516−incrementForwardRetries = \messageId toServer -> atomic do

2517− forwards <- *forwards

2518− let shouldDrop = any forwards (\f -> f.messageId == messageId && f.toServer == toServer && f.retries + 1 >= maxForwardRetries)

2942+incrementForwardRetries = \messageId toServer toKey -> atomic do

2943+ let b = pubkeyBucket toKey

2944+ fwdBkt <- readFwdBucket b

2945+ let shouldDrop = any fwdBkt (\f -> f.messageId == messageId && f.toServer == toServer && f.retries + 1 >= maxForwardRetries)

25192946 if shouldDrop

25202947 then do

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)

25222949 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))

25242951

25252952retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}

25262953retryForwardsForServer = \entries remaining ->

@@ -2542,15 +2969,15 @@ retryForwardsForServer = \entries remaining ->

25422969 Err e -> do

25432970 logWarn ("federation: forward to " ++ f.toServer ++ " failed for " ++ f.messageId ++ ": " ++ show e)

25442971 recordFailure f.toServer

2545− incrementForwardRetries f.messageId f.toServer

2972+ incrementForwardRetries f.messageId f.toServer f.toKey

25462973 -- Network error: stop retrying this server

25472974 Ok response -> do

25482975 if response.ok then do

2549− removeForward f.messageId f.toServer

2976+ removeForward f.messageId f.toServer f.toKey

25502977 clearFailure f.toServer

25512978 else do

25522979 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

25542981 retryForwardsForServer rest (remaining - 1)

25552982

25562983-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.

@@ -2572,7 +2999,7 @@ retryPendingForwards = \pending -> do

25722999-- When a remote key comes online, create forward entries for queued messages and send them

25733000forwardQueuedMessages = \targets t -> do

25743001 newForwards <- atomic do

2575− forwards <- *forwards

3002+ forwards <- readAllForwards

25763003 -- Read only the presence buckets needed for target pubkeys (not all 16)

25773004 let targetPresenceBuckets = uniqueBucketsFor targets (\target -> target.pubkey)

25783005 relevantPresence <- readSelectedPresenceBuckets targetPresenceBuckets

@@ -2582,9 +3009,9 @@ forwardQueuedMessages = \targets t -> do

25823009 let neededBuckets = uniqueBucketsFor remoteTargets (\target -> target.pubkey)

25833010 relevantMessages <- readSelectedBuckets neededBuckets

25843011 let candidates = forwardCandidates remoteTargets relevantMessages forwards t

2585− let capped = takeRelation (maxForwardsEntries - count forwards) candidates

2586− let updatedForwards = union forwards capped

2587− set *forwards = updatedForwards

3012+ let totalFwdCount <- countAllForwards 0

3013+ let capped = takeRelation (maxForwardsEntries - totalFwdCount) candidates

3014+ writeAllForwards (union forwards capped)

25883015 yield capped

25893016 let servers = collectPeerServers newForwards (\x -> x.toServer)

25903017 failures <- *forwardFailures

@@ -2617,27 +3044,27 @@ sendBatchedAcksToServer = \serverAcks server ->

26173044 recordFailure server

26183045 -- Increment retry counters; drop entries exceeding maxAckRetries to deliveredIds

26193046 dropped <- atomic do

2620− ackedIds <- *ackedIds

3047+ ackedIds <- readAllAcked

26213048 let isTarget = \a -> a.originServer == server && hasMessage a.messageId a.toKey batch

26223049 let toDrop = do a <- ackedIds; where isTarget a && a.retries + 1 >= maxAckRetries; yield a

26233050 let dropCount = count toDrop

2624− set *ackedIds = do a <- ackedIds

3051+ writeAllAcked (do a <- ackedIds

26253052 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))

26273054 when (dropCount > 0) (do

2628− deliveredIds <- *deliveredIds

3055+ deliveredIds <- readAllDelivered

26293056 let newDelivered = do d <- toDrop; yield {messageId: d.messageId, toKey: d.toKey, deliveredAt: d.ackedAt}

26303057 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))

26323059 yield dropCount

26333060 when (dropped > 0)

26343061 (logWarn ("federation: dropped " ++ show dropped ++ " ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))

26353062 Ok _ -> do

26363063 clearFailure server

26373064 atomic do

2638− ackedIds <- *ackedIds

3065+ ackedIds <- readAllAcked

26393066 let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch

2640− set *ackedIds = updatedAckedIds)

3067+ writeAllAcked updatedAckedIds)

26413068

26423069-- Send delivery acknowledgments back to origin servers and remove acked entries on success

26433070sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}

@@ -2667,15 +3094,15 @@ retryEvictedAcks = \evictedEntries pendingAcks ->

26673094 Err e -> do

26683095 logWarn ("federation: evicted ack to " ++ server ++ " failed: " ++ show e)

26693096 overflow <- atomic do

2670− ackedIds <- *ackedIds

3097+ ackedIds <- readAllAcked

26713098 let reinsert = do a <- acksForServer evictedToRetry server; yield {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1}

26723099 let combined = union ackedIds reinsert

26733100 let excess = count combined - maxAckedIds

26743101 if excess <= 0

2675− then do set *ackedIds = combined; yield []

3102+ then do writeAllAcked combined; yield []

26763103 else do

26773104 let eviction = evictSmallest combined (max 0 excess) (\a -> a.ackedAt)

2678− set *ackedIds = eviction.kept

3105+ writeAllAcked eviction.kept

26793106 yield eviction.evicted

26803107 when (count overflow > 0) (do

26813108 dropped <- atomic do

@@ -2687,12 +3114,12 @@ retryEvictedAcks = \evictedEntries pendingAcks ->

26873114 else do

26883115 let overflowEviction = evictSmallest updated (max 0 overflowExcess) (\a -> a.ackedAt)

26893116 set *ackOverflow = overflowEviction.kept

2690− deliveredIds <- *deliveredIds

3117+ deliveredIds <- readAllDelivered

26913118 let newDelivered = do a <- overflowEviction.evicted; yield {messageId: a.messageId, toKey: a.toKey, deliveredAt: a.ackedAt}

26923119 let combinedDelivered = union deliveredIds newDelivered

26933120 let deliveredExcess = count combinedDelivered - maxDeliveredIds

2694− set *deliveredIds = if deliveredExcess <= 0 then combinedDelivered

2695− else (evictSmallest combinedDelivered (max 0 deliveredExcess) (\d -> d.deliveredAt)).kept

3121+ writeAllDelivered (if deliveredExcess <= 0 then combinedDelivered

3122+ else (evictSmallest combinedDelivered (max 0 deliveredExcess) (\d -> d.deliveredAt)).kept)

26963123 when (deliveredExcess > 0)

26973124 (logWarn ("ackMessages: deliveredIds at capacity (" ++ show maxDeliveredIds ++ "), " ++ show deliveredExcess ++ " oldest dedup entries permanently removed for " ++ server))

26983125 yield overflowExcess

@@ -2728,10 +3155,10 @@ retryOverflowAcks = \overflow -> do

27283155 where not (isTarget a && a.retries + 1 >= maxAckRetries)

27293156 yield (if isTarget a then {messageId: a.messageId, toKey: a.toKey, ackedAt: a.ackedAt, originServer: a.originServer, retries: a.retries + 1} else a)

27303157 when (dropCount > 0) (do

2731− deliveredIds <- *deliveredIds

3158+ deliveredIds <- readAllDelivered

27323159 let newDelivered = do d <- toDrop; yield {messageId: d.messageId, toKey: d.toKey, deliveredAt: d.ackedAt}

27333160 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))

27353162 yield dropCount

27363163 when (dropped > 0)

27373164 (logWarn ("federation: dropped " ++ show dropped ++ " overflow ack entries to " ++ server ++ " after " ++ show maxAckRetries ++ " retries"))

@@ -2801,9 +3228,9 @@ pruneAndRetry = \t -> do

28013228 when (msgDrift != 0)

28023229 (logInfo ("reconciliation: corrected message drift: " ++ show msgDrift))

28033230 else yield {}

2804− pendingForwards <- atomic do f <- *forwards; yield f

3231+ pendingForwards <- atomic do f <- readAllForwards; yield f

28053232 ackState <- atomic do

2806− ackedIds2 <- *ackedIds

3233+ ackedIds2 <- readAllAcked

28073234 let pendingAcks = do a <- ackedIds2; where not (isLocalServer a.originServer); yield a

28083235 ackOverflow <- *ackOverflow

28093236 let pendingOverflow = do a <- ackOverflow; where not (isLocalServer a.originServer); yield a

@@ -2811,22 +3238,33 @@ pruneAndRetry = \t -> do

28113238 refreshResult <- atomic do

28123239 lastRefresh <- *lastPresenceRefresh

28133240 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

28153242 if not shouldRefresh then yield {extendedActive: [], nearExpiryActive: [], refreshPeers: [], shouldRefresh: False {}}

28163243 else do

2817− allPresence <- readAllPresence

2818− allSessions <- readAllSessions

2819− -- Extend presence for keys with active sessions nearing expiry (sliding window)

2820− let localNearExpiry = do p <- allPresence; where isLocalServer p.server && p.expiresAt > t && p.expiresAt - t < presenceRefreshThreshold; yield p

2821− 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

2823− when (count toExtend > 0) (do

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)

2825− writeAllPresence updatedPresence

2826− set *presenceVersion = t)

2827− knownPeers <- *knownPeerServers

2828− set *lastPresenceRefresh = t

2829− yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: knownPeers, shouldRefresh: True {}}

3244+ -- Optimized: scan presence buckets one at a time to find near-expiry local entries,

3245+ -- then only read session buckets for the pubkeys found. Avoids materializing all 16

3246+ -- presence + 16 session buckets when few keys are near expiry (common case).

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 {}}

3252+ else do

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

3256+ let toExtend = do p <- localNearExpiry; where any 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

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

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)

3263+ writeSelectedPresenceBuckets affectedBuckets updatedPresence

3264+ set *presenceVersion = t)

3265+ knownPeers <- *knownPeerServers

3266+ set *lastPresenceRefresh = t

3267+ yield {extendedActive: toExtend, nearExpiryActive: notExtended, refreshPeers: knownPeers, shouldRefresh: True {}}

28303268 -- Reset federation latency metrics periodically for windowed averages (prevents

28313269 -- lifetime-cumulative average from masking current conditions)

28323270 atomic do

@@ -2935,14 +3373,16 @@ ackPolledMessages = \pubkey cursor t -> do

29353373 recipientCounts <- readRcBucket b

29363374 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))

29373375 let ackedMsgIds = do m <- toAck; yield m.messageId

2938− forwards <- *forwards

2939− set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f

3376+ let fb = pubkeyBucket pubkey

3377+ fwdBkt <- readFwdBucket fb

3378+ writeFwdBucket fb (do f <- fwdBkt; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)

29403379 let ackedFederated = federatedAckEntries toAck t

2941− ackedIds <- *ackedIds

2942− let combinedAckedIds = union ackedIds ackedFederated

3380+ let ab = pubkeyBucket pubkey

3381+ ackedBkt <- readAckedBucket ab

3382+ let combinedAckedIds = union ackedBkt ackedFederated

29433383 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)

29443384 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)

2945− set *ackedIds = eviction.kept

3385+ writeAckedBucket ab eviction.kept

29463386 yield {pendingAcks: ackedFederated, evictedEntries: eviction.evicted}

29473387 else yield {pendingAcks: [], evictedEntries: []}

29483388

@@ -3382,9 +3822,9 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->

33823822 else do

33833823 let msgBucket = pubkeyBucket to

33843824 bucket <- readMessageBucket msgBucket

3385− ackedIds <- *ackedIds

3386− deliveredIds <- *deliveredIds

3387− let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedIds || hasMessage effectiveMsgId to deliveredIds

3825+ ackedBkt <- readAckedBucket msgBucket

3826+ delivBkt <- readDelivBucket msgBucket

3827+ let isDup = checkMessageDup bucket effectiveMsgId to || hasMessage effectiveMsgId to ackedBkt || hasMessage effectiveMsgId to delivBkt

33883828 if isDup then yield {status: MsDuplicate {}, remoteServers: [], retryAfterMs: 0}

33893829 else do

33903830 sendRates <- *sendRates

@@ -3425,20 +3865,24 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effectiveMsgId t ->

34253865 -- if the message expires from messages/ackedIds/deliveredIds and a federation

34263866 -- peer forwards it back. Without this, the nonce check in handleForwardMessage

34273867 -- 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

34293871 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

3431− set *forwardNonces = if count updatedNonces > maxForwardNonces

3432− then evictNonces updatedNonces maxForwardNonces forwardNonceRetention t

3433− else updatedNonces

3872+ let maxPerBucket = maxForwardNonces / 16 + 1

3873+ writeNonceBucket nb (if count updatedNonces > maxPerBucket

3874+ then evictNonces updatedNonces maxPerBucket forwardNonceRetention t

3875+ else updatedNonces)

34343876 let toPresence = presenceForKey recipientPresence to t

34353877 let localOnline = hasLocalPresence toPresence

34363878 let remoteServers = collectPeerServers toPresence (\p -> p.server)

34373879 let forwardsEnqueued = if federationEnabled && count remoteServers > 0 then do

3438− forwards <- *forwards

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}

3440− let cappedFwds = takeRelation (maxForwardsEntries - count forwards) newFwds

3441− set *forwards = union forwards cappedFwds

3880+ let fbi = pubkeyBucket to

3881+ fwdBkt <- readFwdBucket fbi

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

3884+ let cappedFwds = takeRelation (maxForwardsEntries - totalFwdCount) newFwds

3885+ writeFwdBucket fbi (union fwdBkt cappedFwds)

34423886 -- Return whether any forwards were actually enqueued (vs dropped by capacity cap)

34433887 yield (count cappedFwds > 0)

34443888 else yield (False {})

@@ -3452,8 +3896,8 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->

34523896 when (count remoteServers > 0 && federationEnabled) (

34533897 fork (sendToEach remoteServers (\server -> do

34543898 stillPending <- atomic do

3455− forwards <- *forwards

3456− yield (any forwards (\f -> f.messageId == effectiveMsgId && f.toServer == server))

3899+ fwdBkt <- readFwdBucket (pubkeyBucket to)

3900+ yield (any fwdBkt (\f -> f.messageId == effectiveMsgId && f.toServer == server))

34573901 when stillPending (do

34583902 fwdT <- now

34593903 fwdAuth <- generateFederationAuth "forward"

@@ -3469,7 +3913,7 @@ forkSendForwards = \effectiveMsgId to encryptedBlob clampedTs remoteServers ->

34693913 recordFailure server

34703914 Ok response ->

34713915 if response.ok then do

3472− removeForward effectiveMsgId server

3916+ removeForward effectiveMsgId server to

34733917 clearFailure server

34743918 else

34753919 logWarn ("federation: forward to " ++ server ++ " rejected for " ++ effectiveMsgId ++ ": " ++ response.error)))))

@@ -3482,8 +3926,8 @@ sendBatchForwardsSequential = \msgs server ->

34823926 Just {value: r} -> do

34833927 let rest = do m <- msgs; where m.messageId != r.messageId; yield m

34843928 stillPending <- atomic do

3485− forwards <- *forwards

3486− yield (any forwards (\f -> f.messageId == r.messageId && f.toServer == server))

3929+ fwdBkt <- readFwdBucket (pubkeyBucket r.toKey)

3930+ yield (any fwdBkt (\f -> f.messageId == r.messageId && f.toServer == server))

34873931 when stillPending (do

34883932 fwdT <- now

34893933 fwdAuth <- generateFederationAuth "forward"

@@ -3499,7 +3943,7 @@ sendBatchForwardsSequential = \msgs server ->

34993943 recordFailure server

35003944 Ok response ->

35013945 if response.ok then do

3502− removeForward r.messageId server

3946+ removeForward r.messageId server r.toKey

35033947 clearFailure server

35043948 else

35053949 logWarn ("federation: forward to " ++ server ++ " rejected for " ++ r.messageId ++ ": " ++ response.error))

@@ -3595,22 +4039,23 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do

35954039 let inMailbox = any bucket (\m -> m.messageId == messageId && m.toKey == to)

35964040 if inMailbox then yield "pending"

35974041 else do

3598− -- Check if being forwarded to a remote server

3599− forwards <- *forwards

3600− let inForwards = any forwards (\f -> f.messageId == messageId && f.toKey == to)

4042+ -- Check if being forwarded to a remote server (bucket read: O(n/16))

4043+ fwdBkt <- readFwdBucket (pubkeyBucket to)

4044+ let inForwards = any fwdBkt (\f -> f.messageId == messageId && f.toKey == to)

36014045 if inForwards then yield "forwarding"

36024046 else do

36034047 -- Check if already delivered (acked or in terminal delivered set)

3604− ackedIds <- *ackedIds

3605− deliveredIds <- *deliveredIds

3606− let inAcked = hasMessage messageId to ackedIds

3607− let inDelivered = hasMessage messageId to deliveredIds

4048+ ackedBkt <- readAckedBucket msgBucket

4049+ delivBkt <- readDelivBucket msgBucket

4050+ let inAcked = hasMessage messageId to ackedBkt

4051+ let inDelivered = hasMessage messageId to delivBkt

36084052 if inAcked || inDelivered then yield "delivered"

36094053 else do

36104054 -- Extended check: forward nonces have 60-day retention (2x message lifetime)

3611− -- and 100k capacity, providing delivery evidence after ackedIds/deliveredIds are pruned

3612− forwardNonces <- *forwardNonces

3613− let inNonces = any forwardNonces (\n -> n.messageId == messageId && n.toKey == to)

4055+ -- and 100k capacity, providing delivery evidence after ackedIds/deliveredIds are pruned.

4056+ -- Sharded: reads only the bucket for this recipient's pubkey.

4057+ nonceBkt <- readNonceBucket (pubkeyBucket to)

4058+ let inNonces = any nonceBkt (\n -> n.messageId == messageId && n.toKey == to)

36144059 if inNonces then yield "delivered"

36154060 else yield "unknown"

36164061 yield (respond {status: result, authorized: True {}}))

@@ -3664,9 +4109,13 @@ handleRevoke = \targetToken authorization clientIp respond -> do

36644109 Nothing {} -> yield (respond {ok: False {}, authorized: True {}})

36654110 Just {value: r} -> do

36664111 when r.gossipOffline (do

3667− logInfo ("revoke: last session for " ++ r.pubkey ++ ", gossiping offline")

3668− when (federationEnabled && count r.peerServers > 0)

3669− (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))

4112+ isDraining <- *draining

4113+ if isDraining

4114+ then logInfo ("revoke: last session for " ++ r.pubkey ++ " (drain active, skipping redundant offline gossip)")

4115+ else do

4116+ logInfo ("revoke: last session for " ++ r.pubkey ++ ", gossiping offline")

4117+ when (federationEnabled && count r.peerServers > 0)

4118+ (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))

36704119 when (not r.gossipOffline) (logInfo ("revoke: session revoked for " ++ r.pubkey))

36714120 yield (respond {ok: True {}, authorized: True {}}))

36724121

@@ -3719,9 +4168,15 @@ handleLogout = \authorization clientIp respond -> do

37194168 Nothing {} -> yield (respond {ok: False {}, authorized: False {}})

37204169 Just {value: r} -> do

37214170 when r.gossipOffline (do

3722− logInfo ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")

3723− when (federationEnabled && count r.peerServers > 0)

3724− (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))

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

4174+ if isDraining

4175+ then logInfo ("logout: last session for " ++ r.pubkey ++ " (drain active, skipping redundant offline gossip)")

4176+ else do

4177+ logInfo ("logout: last session for " ++ r.pubkey ++ ", gossiping offline")

4178+ when (federationEnabled && count r.peerServers > 0)

4179+ (fork (sendBatchedGossip r.peerServers [{eventType: GossipOffline {}, pubkey: r.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops)))

37254180 yield (respond {ok: True {}, authorized: True {}})

37264181

37274182-- POST /federation/verify: Confirm a token was issued by this server.

@@ -3906,13 +4361,14 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

39064361 then yield {forward: ForwardRejected {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: True {}}

39074362 else do

39084363 bucket <- readMessageBucket msgBucket

3909− ackedIds <- *ackedIds

3910− deliveredIds <- *deliveredIds

3911− forwardNonces <- *forwardNonces

4364+ ackedBkt <- readAckedBucket msgBucket

4365+ delivBkt <- readDelivBucket msgBucket

4366+ -- Sharded nonce check: reads only the bucket for this recipient's pubkey (O(n/16) vs O(n)).

4367+ nonceBkt <- readNonceBucket (pubkeyBucket toKey)

39124368 -- Nonce check ignores originServer: a (messageId, toKey) pair is globally unique,

39134369 -- so replays from any origin should be rejected. This also catches locally-originated

39144370 -- 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)

39164372 if isDup

39174373 then yield {forward: ForwardDuplicate {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}

39184374 else do

@@ -3946,13 +4402,15 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

39464402 set *totalMessageCount = totalMsgCount + 1

39474403 writeRcBucket msgBucket (adjustRecipientCount recipientCounts toKey 1)

39484404 signalMessageForKey toKey seqTs

3949− -- Record forward nonce for replay protection beyond message/ack dedup lifetime

3950− fwdNonces <- *forwardNonces

4405+ -- Record forward nonce for replay protection beyond message/ack dedup lifetime.

4406+ -- Sharded: writes only to the bucket for this recipient's pubkey.

4407+ let fnb = pubkeyBucket toKey

4408+ fwdNonces <- readNonceBucket fnb

39514409 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

3953− set *forwardNonces = if count updatedNonces > maxForwardNonces

3954− then evictNonces updatedNonces maxForwardNonces forwardNonceRetention t

3955− else updatedNonces

4410+ let maxPerBucket = maxForwardNonces / 16 + 1

4411+ writeNonceBucket fnb (if count updatedNonces > maxPerBucket

4412+ then evictNonces updatedNonces maxPerBucket forwardNonceRetention t

4413+ else updatedNonces)

39564414 yield {forward: ForwardStored {}, globalCapReached: False {}, recvRateLimited: False {}, noPresence: False {}}

39574415 if result.noPresence then do

39584416 logWarn ("forward: rejected for " ++ toKey ++ " — no active presence on this server (from " ++ fromServer ++ ")")

@@ -3983,14 +4441,20 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre

39834441 -- Clear forward entries for the acking server. Extract removed forwards' toKey

39844442 -- so we can reclaim origin messages that are no longer needed.

39854443 removedForwards <- atomic do

3986− forwards <- *forwards

4444+ forwards <- readAllForwards

39874445 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)

39894447 yield removed

39904448 let ackedCount = count removedForwards

3991− -- Deferred origin message cleanup: queue for processing in the next prune cycle

3992− -- after ackGracePeriod elapses. Replaces fork+sleep which accumulated sleeping

3993− -- goroutines under high federation ack volume.

4449+ -- Guard: warn if peer acks messages we had no forward entries for. This can indicate

4450+ -- a peer acking messages it never received (e.g., the forward response was lost but

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.

39944458 when (ackedCount > 0) (atomic do

39954459 cleanups <- *pendingAckCleanups

39964460 let newCleanups = do f <- removedForwards; yield {messageId: f.messageId, toKey: f.toKey, cleanupAfter: t + ackGracePeriod}

@@ -4010,6 +4474,13 @@ handleDrain = \authorization respond -> do

40104474 let authorized = verifyMaintenanceAuth authorization

40114475 if not authorized then yield (respond {ok: False {}, drained: 0})

40124476 else do

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})

4483+ else do

40134484 drainResult <- atomic do

40144485 set *draining = True {}

40154486 allPresence <- readAllPresence

@@ -4023,8 +4494,8 @@ handleDrain = \authorization respond -> do

40234494 let drainedCount = count drainResult.localEntries

40244495 -- Read pending federation state atomically (fix: was read outside atomic)

40254496 flushState <- atomic do

4026− f <- *forwards

4027− a <- *ackedIds

4497+ f <- readAllForwards

4498+ a <- readAllAcked

40284499 let pa = do x <- a; where not (isLocalServer x.originServer); yield x

40294500 yield {forwards: f, pendingAcks: pa}

40304501 let pendingForwardCount = count flushState.forwards

@@ -4033,11 +4504,11 @@ handleDrain = \authorization respond -> do

40334504 -- Output as structured JSON with [SNAPSHOT_DATA] prefix for log-shipping pipelines.

40344505 -- This is the last chance to persist state before the server stops.

40354506 drainSnapshot <- atomic do

4036− a <- *ackedIds

4507+ a <- readAllAcked

40374508 ao <- *ackOverflow

4038− d <- *deliveredIds

4509+ d <- readAllDelivered

40394510 ff <- *forwardFailures

4040− fn <- *forwardNonces

4511+ fn <- readAllNonces

40414512 bp <- *blockedPeers

40424513 yield {version: snapshotVersion, presence: drainResult.localEntries, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: flushState.forwards, forwardNonces: fn, blockedPeers: bp}

40434514 println ("[SNAPSHOT_DATA] " ++ toJson drainSnapshot)

@@ -4128,7 +4599,7 @@ handleHealth = \authorization respond -> do

41284599 lastWd <- *lastWatchdogAt

41294600 sa <- *startedAt

41304601 presenceCount <- *totalPresenceCount

4131− forwards <- *forwards

4602+ forwardCount <- countAllForwards 0

41324603 sessionCount <- *totalSessionCount

41334604 activePolls <- *activePolls

41344605 tmc <- *totalMessageCount

@@ -4141,7 +4612,7 @@ handleHealth = \authorization respond -> do

41414612 -- Capacity pressure: flag when any capped collection exceeds 90% full

41424613 let capacityOk = tmc * 10 <= maxMessages * 9

41434614 && presenceCount * 10 <= maxPresenceEntries * 9

4144− && count forwards * 10 <= maxForwardsEntries * 9

4615+ && forwardCount * 10 <= maxForwardsEntries * 9

41454616 && sessionCount * 10 <= maxSessions * 9

41464617 && count activePolls * 10 <= maxActivePolls * 9

41474618 -- Federation: healthy if disabled, in warmup, no peer issues, or recent success

@@ -4152,7 +4623,7 @@ handleHealth = \authorization respond -> do

41524623 -- has been exported within the staleness window (federation state will be lost on restart)

41534624 let snapOk = maintenanceSecret == "" || snapshotStalenessWarning == 0 || warmup

41544625 || (lastSnap > 0 && t - lastSnap <= snapshotStalenessWarning)

4155− || (tmc == 0 && presenceCount == 0 && count forwards == 0)

4626+ || (tmc == 0 && presenceCount == 0 && forwardCount == 0)

41564627 -- Watchdog: healthy if recently bumped or in warmup (watchdog starts after 10x prune interval)

41574628 let wdOk = warmup || t - sa < healthWatchdogMaxAge || (lastWd > 0 && t - lastWd < healthWatchdogMaxAge)

41584629 yield {pruneOk: warmup || (t - lastPrune < healthPruneMaxAge && not pruneStuck),

@@ -4202,24 +4673,24 @@ handleStats = \authorization respond -> do

42024673 core <- atomic do

42034674 sc <- *totalSessionCount

42044675 pc <- *totalPresenceCount

4205− f <- *forwards

4676+ f <- readAllForwards

42064677 c <- *challenges

4207− a <- *ackedIds

4678+ a <- readAllAcked

42084679 ao <- *ackOverflow

4209− d <- *deliveredIds

4680+ d <- readAllDelivered

42104681 ff <- *forwardFailures

42114682 ap <- *activePolls

42124683 tmc <- *totalMessageCount

42134684 rc <- countAllRcEntries 0

42144685 fi <- *federationInFlight

42154686 rg <- *recentGossip

4216− fn <- *forwardNonces

4687+ fnCount <- countAllNonces 0

42174688 pac <- *pendingAckCleanups

42184689 bp <- *blockedPeers

42194690 ger <- *gossipEventRates

42204691 vir <- *verifyIpRates

42214692 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}

42234694 counters <- atomic do

42244695 sr <- *sendRates

42254696 cr <- *challengeRates

@@ -4256,7 +4727,7 @@ handleStats = \authorization respond -> do

42564727 let oldestFwdAge = case minBy core.f (\x -> x.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt

42574728 let oldestAckAge = case minBy core.a (\x -> x.ackedAt) of Nothing {} -> 0; Just {value: x} -> t - x.ackedAt

42584729 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},

42604731 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},

42614732 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},

42624733 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

42714742 if authorized

42724743 then do

42734744 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}

42754746 yield (respond {peers: peers, ok: True {}})

42764747 else yield (respond {peers: [], ok: False {}})

42774748

@@ -4288,16 +4759,18 @@ handleStatsKey = \pubkey authorization respond -> do

42884759 -- Single bucket read: all entries for this pubkey are in the same presence bucket

42894760 keyPresenceBkt <- readPresenceBucket (pubkeyBucket pubkey)

42904761 allSessions <- readAllSessions

4291− forwards <- *forwards

4762+ forwards <- readFwdBucket (pubkeyBucket pubkey)

42924763 activePolls <- *activePolls

42934764 let keyMessages = count (do m <- keyBucket; where m.toKey == pubkey; yield m)

42944765 let keyPresence = do p <- keyPresenceBkt; where p.pubkey == pubkey && p.expiresAt > t; yield {server: p.server, expiresAt: p.expiresAt}

42954766 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}

42974770 let keyPolls = count (do p <- activePolls; where p.pubkey == pubkey; yield p)

4298− yield {messages: keyMessages, presence: keyPresence, sessions: keySessions, forwards: keyForwards, activePolls: keyPolls}

4299− yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})

4300− else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})

4771+ yield {messages: keyMessages, presence: keyPresence, sessions: keySessions, forwards: keyForwards, forwardDetails: keyForwardDetails, activePolls: keyPolls}

4772+ yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, forwardDetails: result.forwardDetails, activePolls: result.activePolls, ok: True {}})

4773+ else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, forwardDetails: [], activePolls: 0, ok: False {}})

43014774

43024775emptySnapshot = {ok: False {}, version: 0, presence: [], ackedIds: [], ackOverflow: [], deliveredIds: [], forwardFailures: [], forwards: [], forwardNonces: [], blockedPeers: []}

43034776

@@ -4312,12 +4785,12 @@ handleSnapshot = \authorization respond -> do

43124785 t <- now

43134786 result <- atomic do

43144787 p <- readAllPresence

4315− a <- *ackedIds

4788+ a <- readAllAcked

43164789 ao <- *ackOverflow

4317− d <- *deliveredIds

4790+ d <- readAllDelivered

43184791 ff <- *forwardFailures

4319− f <- *forwards

4320− fn <- *forwardNonces

4792+ f <- readAllForwards

4793+ fn <- readAllNonces

43214794 bp <- *blockedPeers

43224795 yield {presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}

43234796 atomic do set *lastSnapshotExportAt = t

@@ -4362,33 +4835,33 @@ handleRestore = \version presenceData ackedIdsData ackOverflowData deliveredIdsD

43624835 set *knownPeerServers = collectPeerServers cappedPresence (\p -> p.server)

43634836 set *presenceVersion = t

43644837 -- Union dedup state, respecting capacity limits (using validated data)

4365− ackedIds <- *ackedIds

4838+ ackedIds <- readAllAcked

43664839 let mergedAcked = union ackedIds validAcked

43674840 let cappedAcked = takeRelation maxAckedIds mergedAcked

4368− set *ackedIds = cappedAcked

4841+ writeAllAcked cappedAcked

43694842 ackOverflow <- *ackOverflow

43704843 let mergedOverflow = union ackOverflow validOverflow

43714844 set *ackOverflow = takeRelation maxAckOverflow mergedOverflow

4372− deliveredIds <- *deliveredIds

4845+ deliveredIds <- readAllDelivered

43734846 let mergedDelivered = union deliveredIds deliveredIdsData

43744847 let cappedDelivered = takeRelation maxDeliveredIds mergedDelivered

4375− set *deliveredIds = cappedDelivered

4848+ writeAllDelivered cappedDelivered

43764849 -- Restore peer failure records

43774850 forwardFailures <- *forwardFailures

43784851 let mergedFailures = mergeByKey (\a b -> a.server == b.server) (\x -> x.failedAt) forwardFailures forwardFailuresData

43794852 let cappedFailures = takeRelation maxForwardFailures mergedFailures

43804853 set *forwardFailures = cappedFailures

43814854 -- Restore forward entries (dedup by messageId + toServer, keep fresher on conflict; using validated data)

4382− forwards <- *forwards

4383− let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) forwards validForwards

4855+ existingForwards <- readAllForwards

4856+ let mergedForwards = mergeByKey (\a b -> a.messageId == b.messageId && a.toServer == b.toServer) (\x -> x.createdAt) existingForwards validForwards

43844857 let cappedForwards = takeRelation maxForwardsEntries mergedForwards

4385− set *forwards = cappedForwards

4858+ writeAllForwards cappedForwards

43864859 -- Restore forward nonces (dedup by messageId + toKey + originServer, keep fresher on conflict)

4387− forwardNonces <- *forwardNonces

4860+ existingNonces <- readAllNonces

43884861 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

43904863 let cappedNonces = takeRelation maxForwardNonces mergedNonces

4391− set *forwardNonces = cappedNonces

4864+ writeAllNonces cappedNonces

43924865 -- Restore blocked peers (union with existing, cap at maxBlockedPeers)

43934866 blocked <- *blockedPeers

43944867 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

44314904 recipientCounts <- readRcBucket b

44324905 writeRcBucket b (adjustRecipientCount recipientCounts pubkey (0 - ackCount))

44334906 let ackedMsgIds = do m <- toAck; yield m.messageId

4434− forwards <- *forwards

4435− set *forwards = do f <- forwards; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f

4907+ let fb2 = pubkeyBucket pubkey

4908+ fwdBkt2 <- readFwdBucket fb2

4909+ writeFwdBucket fb2 (do f <- fwdBkt2; where none ackedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)

44364910 let ackedFederated = federatedAckEntries toAck t

4437− ackedIds <- *ackedIds

4438− let combinedAckedIds = union ackedIds ackedFederated

4911+ ackedBkt <- readAckedBucket b

4912+ let combinedAckedIds = union ackedBkt ackedFederated

44394913 let excessCount = max 0 (count combinedAckedIds - maxAckedIds)

44404914 let eviction = evictSmallest combinedAckedIds excessCount (\a -> a.ackedAt)

4441− set *ackedIds = eviction.kept

4915+ writeAckedBucket b eviction.kept

44424916 yield {ackedCount: ackCount, pendingAcks: ackedFederated, evictedEntries: eviction.evicted}

44434917 else yield {ackedCount: 0, pendingAcks: [], evictedEntries: []}

44444918 fork (when federationEnabled (do

@@ -4503,10 +4977,11 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do

45034977 set *totalMessageCount = max 0 (tc - removed)

45044978 rc <- readRcBucket b

45054979 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))

45074981 let deletedMsgIds = do m <- toDelete; yield m.messageId

4508− forwards <- *forwards

4509− set *forwards = do f <- forwards; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f

4982+ let dfb = pubkeyBucket pubkey

4983+ fwdBkt <- readFwdBucket dfb

4984+ writeFwdBucket dfb (do f <- fwdBkt; where none deletedMsgIds (\mid -> f.messageId == mid && f.toKey == pubkey); yield f)

45104985 yield removed

45114986 else yield 0

45124987 yield (respond {ok: True {}, authorized: True {}, deletedCount: deletedCount}))

@@ -4555,7 +5030,7 @@ sendBatchInner = \senderKey messages t acc ->

45555030 when (msg.timestamp < t - messageMaxAge / 2 && t - msg.timestamp <= messageMaxAge)

45565031 (logWarn ("sendBatch: stale timestamp (" ++ show ((t - msg.timestamp) / 86400000) ++ " days old) for " ++ effectiveMsgId))

45575032 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}

45595034 else do

45605035 let clampedTs = clampTimestamp msg.timestamp t

45615036 sendResult <- atomic do

@@ -4677,6 +5152,36 @@ handleAdminRevokeSessions = \pubkey authorization respond -> do

46775152 (fork (sendBatchedGossip result.peerServers [{eventType: GossipOffline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl}] gossipOriginHops))

46785153 yield (respond {ok: True {}, revokedCount: result.revokedCount})

46795154

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})

5161+ else do

5162+ t <- now

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)

5179+ yield removed

5180+ else yield 0

5181+ when (purgedCount > 0)

5182+ (logInfo ("admin: purged " ++ show purgedCount ++ " messages for " ++ pubkey))

5183+ yield (respond {ok: True {}, purgedCount: purgedCount})

5184+

46805185-- GET /maintenance/rate-limits: Return current effective rate limits.

46815186handleGetRateLimits = \authorization respond -> do

46825187 let authorized = verifyMaintenanceAuth authorization

@@ -4737,6 +5242,7 @@ requestPath = \req -> case req of

47375242 PingPeer {} -> "POST /maintenance/ping"

47385243 AdminListSessions {} -> "GET /maintenance/sessions"

47395244 AdminRevokeSessions {} -> "POST /maintenance/sessions/revoke-all"

5245+ AdminPurgeMessages {} -> "POST /maintenance/messages/purge"

47405246 GetRateLimits {} -> "GET /maintenance/rate-limits"

47415247 SetRateLimits {} -> "POST /maintenance/rate-limits"

47425248 Health {} -> "GET /health"

@@ -4773,6 +5279,7 @@ requestClientIp = \req -> case req of

47735279 PingPeer {} -> ""

47745280 AdminListSessions {} -> ""

47755281 AdminRevokeSessions {} -> ""

5282+ AdminPurgeMessages {} -> ""

47765283 GetRateLimits {} -> ""

47775284 SetRateLimits {} -> ""

47785285 Health {} -> ""

@@ -4812,6 +5319,7 @@ serveInner = \req -> case req of

48125319 PingPeer {server, authorization, respond} -> handlePingPeer server authorization respond

48135320 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond

48145321 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond

5322+ AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond

48155323 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond

48165324 SetRateLimits {globalRequest, ipRate, sendRate, federationRate, authGlobal, recvRate, authorization, respond} -> handleSetRateLimits globalRequest ipRate sendRate federationRate authGlobal recvRate authorization respond

48175325 Health {authorization, respond} -> handleHealth authorization respond

@@ -4958,13 +5466,13 @@ backgroundSnapshot = do

49585466 tmc <- *totalMessageCount

49595467 sc <- *totalSessionCount

49605468 pc <- *totalPresenceCount

4961− f <- *forwards

4962− a <- *ackedIds

5469+ fwdCount <- countAllForwards 0

5470+ ackedCount <- countAllAcked 0

49635471 ao <- *ackOverflow

4964− d <- *deliveredIds

5472+ deliveredCount <- countAllDelivered 0

49655473 ff <- *forwardFailures

49665474 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}

49685476 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)

49695477 t <- now

49705478 -- Periodic full snapshot auto-export: emit [SNAPSHOT_DATA] JSON to stdout so a

@@ -4975,12 +5483,12 @@ backgroundSnapshot = do

49755483 when (hasState && (lastAutoExport == 0 || t - lastAutoExport >= autoSnapshotExportInterval)) (do

49765484 exportData <- atomic do

49775485 p <- readAllPresence

4978− a <- *ackedIds

5486+ a <- readAllAcked

49795487 ao <- *ackOverflow

4980− d <- *deliveredIds

5488+ d <- readAllDelivered

49815489 ff <- *forwardFailures

4982− f <- *forwards

4983− fn <- *forwardNonces

5490+ f <- readAllForwards

5491+ fn <- readAllNonces

49845492 bp <- *blockedPeers

49855493 yield {version: snapshotVersion, presence: p, ackedIds: a, ackOverflow: ao, deliveredIds: d, forwardFailures: ff, forwards: f, forwardNonces: fn, blockedPeers: bp}

49865494 println ("[SNAPSHOT_DATA] " ++ toJson exportData)

@@ -5047,18 +5555,19 @@ startupProbeNext = \peers probed ->

50475555 Just {value: server} -> do

50485556 let rest = do p <- peers; where p != server; yield p

50495557 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).

50515562 result <- fetchFederation server (RecvGossip {

5052− authorization: auth,

5563+ authorization: "",

50535564 events: [], hopsRemaining: 1, fromServer: serverHostN,

50545565 requestTimestamp: startTime, correlationId: ""})

50555566 endTime <- now

50565567 let latency = endTime - startTime

50575568 case result of

50585569 Err e -> logWarn ("startup: peer " ++ server ++ " unreachable (" ++ show e ++ ") — check DNS/firewall/TLS")

5059− Ok r -> if r.ok

5060− then logInfo ("startup: peer " ++ server ++ " ok (" ++ show latency ++ "ms)")

5061− else logWarn ("startup: peer " ++ server ++ " rejected (" ++ r.error ++ ", " ++ show latency ++ "ms)")

5570+ Ok r -> logInfo ("startup: peer " ++ server ++ " reachable (" ++ show latency ++ "ms)")

50625571 startupProbeNext rest (probed + 1)

50635572

50645573main = do