fix: randomize per-client H2 shard start offsets

Avoid every new SDK client landing first wait/upload on global shard 0 when transports are process-shared.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: Reflex

src/runloop_api_client/_base_client.py

@@ -8,6 +8,7 @@
 import asyncio
 import inspect
 import logging
+import secrets
 import weakref
 import platform
 import warnings
@@ -178,8 +179,8 @@ async def aclose(self) -> None:
 
 # Sharded H2 bulkheads: long-polls and file transfers stay off the API control-plane
 # connection. Each shard index maps to its own shared transport (≈ one H2 connection).
-# Per-client round-robin spreads concurrent requests across shards. Removable once
-# httpcore respects stream capacity when opening connections.
+# Per-client round-robin (random start offset) spreads concurrent requests across
+# shards. Removable once httpcore respects stream capacity when opening connections.
 _shared_sync_background_transports: dict[int, _SharedTransport] = {}
 _shared_sync_transfer_transports: dict[int, _SharedTransport] = {}
 _shared_async_background_transports: weakref.WeakKeyDictionary[
@@ -1054,8 +1055,9 @@ def __init__(
         self._bulkhead_lock = threading.Lock()
         self._background_pool_shards = background_pool_shards
         self._transfer_pool_shards = transfer_pool_shards
-        self._background_next = 0
-        self._transfer_next = 0
+        # Random start avoids every short-lived SDK client pinning global shard 0.
+        self._background_next = secrets.randbelow(background_pool_shards)
+        self._transfer_next = secrets.randbelow(transfer_pool_shards)
         # Custom http_client owns the full transport stack; don't invent sibling pools.
         self._isolate_workload_pools = http_client is None
 
@@ -1786,8 +1788,9 @@ def __init__(
         self._transfer_clients = {}
         self._background_pool_shards = background_pool_shards
         self._transfer_pool_shards = transfer_pool_shards
-        self._background_next = 0
-        self._transfer_next = 0
+        # Random start avoids every short-lived SDK client pinning global shard 0.
+        self._background_next = secrets.randbelow(background_pool_shards)
+        self._transfer_next = secrets.randbelow(transfer_pool_shards)
         # Custom http_client owns the full transport stack; don't invent sibling pools.
         self._isolate_workload_pools = http_client is None
 

src/runloop_api_client/_client.py

@@ -98,7 +98,8 @@ def __init__(
         # Set to False to create a private connection pool (old behavior).
         shared_http_pool: bool = True,
         # Separate H2 connection pools for long-polls (/wait_for_status) and file
-        # transfers. Each shard ≈ one H2 connection; requests round-robin per client.
+        # transfers. Each shard ≈ one H2 connection; requests round-robin per client
+        # from a randomized starting offset (avoids every new client hitting shard 0).
         background_pool_shards: int = DEFAULT_BACKGROUND_POOL_SHARDS,
         transfer_pool_shards: int = DEFAULT_TRANSFER_POOL_SHARDS,
         # Enable or disable schema validation for data returned by the API.
@@ -406,7 +407,8 @@ def __init__(
         # Set to False to create a private connection pool (old behavior).
         shared_http_pool: bool = True,
         # Separate H2 connection pools for long-polls (/wait_for_status) and file
-        # transfers. Each shard ≈ one H2 connection; requests round-robin per client.
+        # transfers. Each shard ≈ one H2 connection; requests round-robin per client
+        # from a randomized starting offset (avoids every new client hitting shard 0).
         background_pool_shards: int = DEFAULT_BACKGROUND_POOL_SHARDS,
         transfer_pool_shards: int = DEFAULT_TRANSFER_POOL_SHARDS,
         # Enable or disable schema validation for data returned by the API.

tests/test_transfer_client.py

@@ -119,6 +119,18 @@ def test_round_robin_spreads_requests_across_shards() -> None:
         client.close()
 
 
+def test_many_clients_first_requests_use_both_shards() -> None:
+    clients = [_make_client(background_pool_shards=2) for _ in range(40)]
+    try:
+        req = httpx.Request("POST", f"{base_url}/v1/devboxes/dbx_1/wait_for_status")
+        seen_transports = {id(c._send_client_for_request(req)._transport) for c in clients}  # type: ignore[attr-defined]
+        assert len(seen_transports) == 2
+        assert set(_base_mod._shared_sync_background_transports) == {0, 1}
+    finally:
+        for c in clients:
+            c.close()
+
+
 def test_custom_http_client_skips_isolation() -> None:
     custom = httpx.Client()
     client = _make_client(http_client=custom)
@@ -137,19 +149,24 @@ def test_round_robin_is_per_client_while_transports_are_shared() -> None:
     c2 = _make_client(background_pool_shards=2)
     try:
         req = httpx.Request("POST", f"{base_url}/v1/devboxes/dbx_shared/wait_for_status")
-        # Each SDK client has its own counter; both start at shard 0.
+        start1 = c1._background_next
+        start2 = c2._background_next
         t1 = c1._send_client_for_request(req)
         t2 = c2._send_client_for_request(req)
         assert t1 is not t2
-        assert t1._transport is t2._transport  # type: ignore[attr-defined]
-        assert c1._background_next == 1
-        assert c2._background_next == 1
+        # Same starting shard → same shared transport; different → different shards.
+        if start1 % 2 == start2 % 2:
+            assert t1._transport is t2._transport  # type: ignore[attr-defined]
+        else:
+            assert t1._transport is not t2._transport  # type: ignore[attr-defined]
+        assert c1._background_next == start1 + 1
+        assert c2._background_next == start2 + 1
 
         # Advancing one client does not affect the other.
         t1b = c1._send_client_for_request(req)
         assert t1b is not t1
-        assert c1._background_next == 2
-        assert c2._background_next == 1
+        assert c1._background_next == start1 + 2
+        assert c2._background_next == start2 + 1
     finally:
         c1.close()
         c2.close()