Skip to main content

max / synckit

Assert the retry, backoff and jitter math directly retry_delay, jittered and parse_retry_after were only ever reached through the request loop, and the one nearby test restated the exponential formula instead of calling anything. Cover them head-on: the Retry-After path with its 60-second cap, the jittered exponential window per attempt, the sub-5ms span-rounds-to-zero case, and that the jitter actually spreads. Add the Http arm of is_transient over all four reqwest error kinds (builder, redirect, decode, connect), read_body_capped's Content-Length reject and its at-the-cap boundary, and the empty-string fallback in read_text_capped. Tighten reconnect_delay's test from "grows then caps" to the exact 1s, 2s, 4s series with a per-attempt jitter window, plus a spread check.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-23 22:13 UTC
Signed with PGP, not checked
Commit: c9af4da9ad58ae4ff48a879a526671d689ccb255
Parent: a06324d
2 files changed, +349 insertions, -9 deletions
M synckit-client/src/client/helpers.rs +313
@@ -1588,4 +1588,317 @@
1588 1588 .unwrap();
1589 1589 assert_eq!(decrypted.data.unwrap(), plaintext);
1590 1590 }
1591 +
1592 + // ── Backoff, jitter and Retry-After ──
1593 +
1594 + fn ensure_crypto_provider() {
1595 + static PROVIDER: std::sync::Once = std::sync::Once::new();
1596 + PROVIDER.call_once(|| {
1597 + let _ = rustls::crypto::ring::default_provider().install_default();
1598 + });
1599 + }
1600 +
1601 + fn server_err(status: u16, retry_after_secs: Option<u64>) -> SyncKitError {
1602 + SyncKitError::Server {
1603 + status,
1604 + message: String::new(),
1605 + retry_after_secs,
1606 + }
1607 + }
1608 +
1609 + #[test]
1610 + fn retry_delay_uses_retry_after_capped_at_sixty_seconds() {
1611 + assert_eq!(
1612 + retry_delay(&server_err(429, Some(1)), 0),
1613 + Duration::from_secs(1)
1614 + );
1615 + assert_eq!(
1616 + retry_delay(&server_err(429, Some(7)), 3),
1617 + Duration::from_secs(7)
1618 + );
1619 + assert_eq!(
1620 + retry_delay(&server_err(429, Some(59)), 0),
1621 + Duration::from_secs(59)
1622 + );
1623 + assert_eq!(
1624 + retry_delay(&server_err(429, Some(60)), 0),
1625 + Duration::from_mins(1)
1626 + );
1627 + assert_eq!(
1628 + retry_delay(&server_err(429, Some(61)), 0),
1629 + Duration::from_mins(1)
1630 + );
1631 + assert_eq!(
1632 + retry_delay(&server_err(503, Some(86_400)), 2),
1633 + Duration::from_mins(1)
1634 + );
1635 + }
1636 +
1637 + #[test]
1638 + fn retry_delay_without_retry_after_is_jittered_exponential() {
1639 + let errors = [
1640 + server_err(503, None),
1641 + server_err(429, None),
1642 + SyncKitError::Internal("no retry-after on this variant".to_string()),
1643 + ];
1644 + for attempt in 0..4u32 {
1645 + let base = BASE_DELAY * 2u32.pow(attempt);
1646 + let low = base.mul_f64(0.8);
1647 + let high = base.mul_f64(1.2);
1648 + for err in &errors {
1649 + for _ in 0..32 {
1650 + let delay = retry_delay(err, attempt);
1651 + assert!(
1652 + delay >= low && delay <= high,
1653 + "attempt {attempt}: {delay:?} outside [{low:?}, {high:?}]"
1654 + );
1655 + }
1656 + }
1657 + }
1658 + }
1659 +
1660 + #[test]
1661 + fn retry_delay_grows_with_the_attempt() {
1662 + // The windows for consecutive attempts do not overlap, so the ordering
1663 + // holds for every draw rather than on average.
1664 + let err = server_err(503, None);
1665 + for attempt in 0..3u32 {
1666 + let here = retry_delay(&err, attempt);
1667 + let next = retry_delay(&err, attempt + 1);
1668 + assert!(
1669 + next > here,
1670 + "attempt {attempt}: {next:?} not above {here:?}"
1671 + );
1672 + }
1673 + }
1674 +
1675 + #[test]
1676 + fn jittered_stays_within_twenty_percent_of_base() {
1677 + for base_ms in [5u64, 10, 1_000, 4_000, 60_000] {
1678 + let base = Duration::from_millis(base_ms);
1679 + let span = base_ms / 5;
1680 + let low = Duration::from_millis(base_ms - span);
1681 + let high = Duration::from_millis(base_ms + span);
1682 + for _ in 0..128 {
1683 + let got = jittered(base);
1684 + assert!(
1685 + got >= low && got <= high,
1686 + "{got:?} outside [{low:?}, {high:?}]"
1687 + );
1688 + }
1689 + }
1690 + }
1691 +
1692 + #[test]
1693 + fn jittered_actually_spreads() {
1694 + let base = Duration::from_secs(1);
1695 + let first = jittered(base);
1696 + assert!(
1697 + (0..256).any(|_| jittered(base) != first),
1698 + "256 draws all returned {first:?}, so nothing is being spread"
1699 + );
1700 + }
1701 +
1702 + #[test]
1703 + fn jittered_returns_base_when_the_span_rounds_to_zero() {
1704 + // Under 5ms the 20% span truncates to 0 and the base is returned as-is.
1705 + for base_ms in 0..5u64 {
1706 + let base = Duration::from_millis(base_ms);
1707 + assert_eq!(jittered(base), base);
1708 + }
1709 + assert_eq!(jittered(Duration::ZERO), Duration::ZERO);
1710 + }
1711 +
1712 + /// Fetch a response carrying the given `Retry-After` header value (or none).
1713 + /// The server is returned alongside so it outlives the response.
1714 + async fn response_with_retry_after(
1715 + header: Option<&str>,
1716 + ) -> (wiremock::MockServer, reqwest::Response) {
1717 + ensure_crypto_provider();
1718 + let server = wiremock::MockServer::start().await;
1719 + let mut template = wiremock::ResponseTemplate::new(429);
1720 + if let Some(value) = header {
1721 + template = template.insert_header("retry-after", value);
1722 + }
1723 + wiremock::Mock::given(wiremock::matchers::any())
1724 + .respond_with(template)
1725 + .mount(&server)
1726 + .await;
1727 + let resp = reqwest::get(server.uri()).await.unwrap();
1728 + (server, resp)
1729 + }
1730 +
1731 + #[tokio::test]
1732 + async fn parse_retry_after_reads_a_numeric_header() {
1733 + let (_server, resp) = response_with_retry_after(Some("5")).await;
1734 + assert_eq!(parse_retry_after(&resp), Some(5));
1735 + }
1736 +
1737 + #[tokio::test]
1738 + async fn parse_retry_after_does_not_cap() {
1739 + // The 60s cap belongs to retry_delay, not to parsing.
1740 + let (_server, resp) = response_with_retry_after(Some("3600")).await;
1741 + assert_eq!(parse_retry_after(&resp), Some(3600));
1742 + }
1743 +
1744 + #[tokio::test]
1745 + async fn parse_retry_after_is_none_when_absent() {
1746 + let (_server, resp) = response_with_retry_after(None).await;
1747 + assert_eq!(parse_retry_after(&resp), None);
1748 + }
1749 +
1750 + #[tokio::test]
1751 + async fn parse_retry_after_is_none_for_a_non_numeric_header() {
1752 + for value in [
1753 + "",
1754 + "soon",
1755 + "5s",
1756 + "-3",
1757 + "1.5",
1758 + "Wed, 21 Oct 2015 07:28:00 GMT",
1759 + ] {
1760 + let (_server, resp) = response_with_retry_after(Some(value)).await;
1761 + assert_eq!(parse_retry_after(&resp), None, "header {value:?}");
1762 + }
1763 + }
1764 +
1765 + #[tokio::test]
1766 + async fn parse_retry_after_is_none_for_zero() {
1767 + let (_server, resp) = response_with_retry_after(Some("0")).await;
1768 + assert_eq!(parse_retry_after(&resp), None);
1769 + }
1770 +
1771 + #[tokio::test]
1772 + async fn check_response_carries_retry_after_onto_the_error() {
1773 + let (_server, resp) = response_with_retry_after(Some("12")).await;
1774 + let err = check_response(resp).await.unwrap_err();
1775 + match err {
1776 + SyncKitError::Server {
1777 + status,
1778 + retry_after_secs,
1779 + ..
1780 + } => {
1781 + assert_eq!(status, 429);
1782 + assert_eq!(retry_after_secs, Some(12));
1783 + }
1784 + other => panic!("expected a Server error, got {other:?}"),
1785 + }
1786 + // And the retry loop then uses it verbatim.
1787 + assert_eq!(
1788 + retry_delay(&server_err(429, Some(12)), 0),
1789 + Duration::from_secs(12)
1790 + );
1791 + }
1792 +
1793 + // ── is_transient over the Http arm ──
1794 +
1795 + #[tokio::test]
1796 + async fn is_transient_over_reqwest_error_kinds() {
1797 + ensure_crypto_provider();
1798 +
1799 + // A builder error is a programming mistake: an invalid header value.
1800 + let builder = reqwest::Client::new()
1801 + .get("http://example.invalid/")
1802 + .header("x-bad", "line\nbreak")
1803 + .build()
1804 + .unwrap_err();
1805 + assert!(builder.is_builder());
1806 + assert!(!is_transient(&SyncKitError::Http(builder)));
1807 +
1808 + // A redirect loop is permanent.
1809 + let server = wiremock::MockServer::start().await;
1810 + let uri = server.uri();
1811 + wiremock::Mock::given(wiremock::matchers::any())
1812 + .respond_with(
1813 + wiremock::ResponseTemplate::new(302).insert_header("location", uri.as_str()),
1814 + )
1815 + .mount(&server)
1816 + .await;
1817 + let redirect = reqwest::get(server.uri()).await.unwrap_err();
1818 + assert!(redirect.is_redirect());
1819 + assert!(!is_transient(&SyncKitError::Http(redirect)));
1820 +
1821 + // A body that will not decode is permanent. The uncapped reader is
1822 + // banned in production code; a decode error is what this test needs and
1823 + // the body is eight bytes long.
1824 + let json_server = wiremock::MockServer::start().await;
1825 + wiremock::Mock::given(wiremock::matchers::any())
1826 + .respond_with(wiremock::ResponseTemplate::new(200).set_body_string("not json"))
1827 + .mount(&json_server)
1828 + .await;
1829 + #[allow(clippy::disallowed_methods)]
1830 + let decode = {
1831 + let resp = reqwest::get(json_server.uri()).await.unwrap();
1832 + resp.json::<u32>().await.unwrap_err()
1833 + };
1834 + assert!(decode.is_decode());
1835 + assert!(!is_transient(&SyncKitError::Http(decode)));
1836 +
1837 + // A connect failure is the transient case the arm exists for.
1838 + let connect = reqwest::get("http://127.0.0.1:1/").await.unwrap_err();
1839 + assert!(connect.is_connect());
1840 + assert!(!connect.is_builder() && !connect.is_redirect() && !connect.is_decode());
1841 + assert!(is_transient(&SyncKitError::Http(connect)));
1842 + }
1843 +
1844 + // ── read_body_capped ──
1845 +
1846 + async fn body_of_len(len: usize) -> (wiremock::MockServer, reqwest::Response) {
1847 + ensure_crypto_provider();
1848 + let server = wiremock::MockServer::start().await;
1849 + wiremock::Mock::given(wiremock::matchers::any())
1850 + .respond_with(wiremock::ResponseTemplate::new(200).set_body_bytes(vec![b'x'; len]))
1851 + .mount(&server)
1852 + .await;
1853 + let resp = reqwest::get(server.uri()).await.unwrap();
1854 + (server, resp)
1855 + }
1856 +
1857 + #[tokio::test]
1858 + async fn read_body_capped_returns_a_body_under_the_cap() {
1859 + let (_server, resp) = body_of_len(100).await;
1860 + let bytes = read_body_capped(resp, 1024).await.unwrap();
1861 + assert_eq!(bytes.len(), 100);
1862 + assert!(bytes.iter().all(|&b| b == b'x'));
1863 + }
1864 +
1865 + #[tokio::test]
1866 + async fn read_body_capped_allows_a_body_exactly_at_the_cap() {
1867 + // The check is `>`, so a body the size of the cap is legal.
1868 + let (_server, resp) = body_of_len(256).await;
1869 + let bytes = read_body_capped(resp, 256).await.unwrap();
1870 + assert_eq!(bytes.len(), 256);
1871 + }
1872 +
1873 + #[tokio::test]
1874 + async fn read_body_capped_rejects_an_oversized_content_length() {
1875 + let (_server, resp) = body_of_len(2048).await;
1876 + assert_eq!(resp.content_length(), Some(2048));
1877 + let err = read_body_capped(resp, 1024).await.unwrap_err();
1878 + match err {
1879 + SyncKitError::Internal(msg) => {
1880 + assert!(
1881 + msg.contains("2048"),
1882 + "message did not name the length: {msg}"
1883 + );
1884 + assert!(msg.contains("1024"), "message did not name the cap: {msg}");
1885 + }
1886 + other => panic!("expected Internal, got {other:?}"),
1887 + }
1888 + }
1889 +
1890 + #[tokio::test]
1891 + async fn read_text_capped_falls_back_to_empty_on_an_oversized_body() {
1892 + let (_server, resp) = body_of_len(2048).await;
1893 + assert_eq!(read_text_capped(resp, 16).await, String::new());
1894 + }
1895 +
1896 + #[tokio::test]
1897 + async fn read_json_capped_propagates_the_cap_error() {
1898 + let (_server, resp) = body_of_len(2048).await;
1899 + let err = read_json_capped::<serde_json::Value>(resp, 16)
1900 + .await
1901 + .unwrap_err();
1902 + assert!(matches!(err, SyncKitError::Internal(_)));
1903 + }
1591 1904 }
M synckit-client/src/client/subscribe.rs +36 -9
@@ -303,14 +303,41 @@
303 303 }
304 304
305 305 #[test]
306 - fn reconnect_delay_grows_then_caps() {
307 - // Within +/-20% jitter, the delay grows with the attempt and never
308 - // exceeds the cap plus jitter headroom.
309 - let d1 = reconnect_delay(1);
310 - let d6 = reconnect_delay(6);
311 - assert!(d1 >= Duration::from_millis(1600) && d1 <= Duration::from_millis(2400));
312 - assert!(d6 <= MAX_RECONNECT_DELAY + Duration::from_secs(12));
313 - // Deep attempts stay capped (exponent is clamped).
314 - assert!(reconnect_delay(100) <= MAX_RECONNECT_DELAY + Duration::from_secs(12));
306 + fn reconnect_delay_follows_the_documented_series() {
307 + // 1s, 2s, 4s, ... capped at MAX_RECONNECT_DELAY, each draw within
308 + // +/-20% of its own step rather than merely below the cap.
309 + let expected = |attempt: u32| {
310 + Duration::from_secs(1)
311 + .saturating_mul(2u32.saturating_pow(attempt.min(6)))
312 + .min(MAX_RECONNECT_DELAY)
313 + };
314 + assert_eq!(expected(0), Duration::from_secs(1));
315 + assert_eq!(expected(1), Duration::from_secs(2));
316 + assert_eq!(expected(2), Duration::from_secs(4));
317 + assert_eq!(expected(3), Duration::from_secs(8));
318 + assert_eq!(expected(4), Duration::from_secs(16));
319 + assert_eq!(expected(5), Duration::from_secs(32));
320 + assert_eq!(expected(6), MAX_RECONNECT_DELAY);
321 +
322 + for attempt in [0u32, 1, 2, 3, 4, 5, 6, 7, 100, u32::MAX] {
323 + let base = expected(attempt);
324 + let span = base / 5;
325 + for _ in 0..32 {
326 + let got = reconnect_delay(attempt);
327 + assert!(
328 + got >= base.saturating_sub(span) && got <= base.saturating_add(span),
329 + "attempt {attempt}: {got:?} outside {base:?} +/-20%"
330 + );
331 + }
332 + }
333 + }
334 +
335 + #[test]
336 + fn reconnect_delay_is_jittered() {
337 + let first = reconnect_delay(3);
338 + assert!(
339 + (0..256).any(|_| reconnect_delay(3) != first),
340 + "256 draws all returned {first:?}, so the backoff is not spread"
341 + );
315 342 }
316 343 }