From 3a6df8a17108b77eb77cc929acdf7b00a3d8943f Mon Sep 17 00:00:00 2001 From: Avi Basnet Date: Fri, 4 Sep 2026 00:29:10 +0000 Subject: [PATCH 1/2] [tinker] Keep an undelivered sample result alive for the SDK's retry retrieve_future started the short retrieved-TTL clock as soon as the response object was built, even for a poll the SDK had already abandoned after its 45s client timeout. uvicorn drops the send to a dead client silently, the sweeper evicted the result 120s later, and the SDK's retry of the same request_id got 404 "Future not found", which the SDK treats as fatal. This is the 128x128 failure seen against j316chuck/SkyRL#18. Start the clock only if request.is_disconnected() is false, and raise the retrieved TTL to 300s so it outlasts the SDK's worst-case re-poll gap (45s timeout plus up to 30s backoff, twice). tests/tinker/ test_retrieve_future_lost_response.py reproduces the chain under a real uvicorn socket; it fails without this change and passes with it. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Avi Basnet --- skyrl/tinker/api.py | 9 +- skyrl/tinker/external_future_store.py | 6 +- tests/tinker/test_external_future_store.py | 8 + .../test_retrieve_future_lost_response.py | 159 ++++++++++++++++++ 4 files changed, 178 insertions(+), 4 deletions(-) create mode 100644 tests/tinker/test_retrieve_future_lost_response.py diff --git a/skyrl/tinker/api.py b/skyrl/tinker/api.py index 5432555af5..22c6a5b1a9 100644 --- a/skyrl/tinker/api.py +++ b/skyrl/tinker/api.py @@ -1515,8 +1515,13 @@ async def retrieve_future(request: RetrieveFutureRequest, req: Request): else: response = raw_json_response(result_data) # Start the retry-grace clock now that the response is built and about to - # be sent, so a large result is never evicted mid-delivery. - if found_in_memory: + # be sent, so a large result is never evicted mid-delivery -- but only if + # this client is still there to receive it. The SDK abandons a poll after + # 45s and retries the same request_id; if the result lands after that, + # this handler wakes on a dead connection (uvicorn drops the send + # silently) and starting the short clock here would let the sweeper + # evict a result nobody received, turning the retry into a 404. + if found_in_memory and not await req.is_disconnected(): external_future_store.mark_retrieved(request_id) return response diff --git a/skyrl/tinker/external_future_store.py b/skyrl/tinker/external_future_store.py index e69e6999d4..1e58b4b1bc 100644 --- a/skyrl/tinker/external_future_store.py +++ b/skyrl/tinker/external_future_store.py @@ -44,8 +44,10 @@ class ExternalFutureStore: # retry following a lost HTTP response still finds it. Measured from # delivery (mark_retrieved), never from the in-store read: a large result # can spend minutes being serialized and sent, and starting the clock at - # read would evict it mid-delivery. - _RETRIEVED_TTL_SECONDS = 120.0 + # read would evict it mid-delivery. The SDK re-polls after a 45s client + # timeout plus up to 30s of backoff, so two consecutive misses span 150s; + # the grace has to outlast that. + _RETRIEVED_TTL_SECONDS = 300.0 # Completed but not yet delivered — governs the read/serialize/send window # and clients that never come back. _COMPLETED_TTL_SECONDS = 600.0 diff --git a/tests/tinker/test_external_future_store.py b/tests/tinker/test_external_future_store.py index 4178dc54dc..c85f9dcac9 100644 --- a/tests/tinker/test_external_future_store.py +++ b/tests/tinker/test_external_future_store.py @@ -42,6 +42,11 @@ def _sample_input(seq_id: int) -> types.SampleInput: ) +async def _still_connected() -> bool: + """Stand-in for ``Request.is_disconnected`` on a client that is still waiting.""" + return False + + class _CompletingForwarder: def __init__(self, store: ExternalFutureStore): self.store = store @@ -140,6 +145,7 @@ async def test_sustained_model_path_rollouts_training_futures_and_heartbeats(fut ) ), headers={}, + is_disconnected=_still_connected, ) async with AsyncSession(engine) as session: @@ -294,6 +300,7 @@ def serialize_result_in_thread(request_type, result_data): ) ), headers={"accept": api.PROTO_CONTENT_TYPE}, + is_disconnected=_still_connected, ) responses = await asyncio.gather( @@ -643,6 +650,7 @@ async def test_retrieve_future_serializes_in_memory_result_as_proto(future_store ) ), headers={"accept": "application/x-protobuf, application/json"}, + is_disconnected=_still_connected, ) response = await api.retrieve_future(api.RetrieveFutureRequest(request_id=str(request_id)), request) diff --git a/tests/tinker/test_retrieve_future_lost_response.py b/tests/tinker/test_retrieve_future_lost_response.py new file mode 100644 index 0000000000..a8ce73eb1a --- /dev/null +++ b/tests/tinker/test_retrieve_future_lost_response.py @@ -0,0 +1,159 @@ +"""A result whose delivery the client never received must survive for the SDK's retry. + +Reproduces the 128x128 failure Chuck hit against PR j316chuck/SkyRL#18 (same +code as main): the SDK polls ``retrieve_future`` with a 45s client timeout and +gives up; the result lands afterwards; the abandoned handler still builds a +response and starts the short *retrieved* TTL clock even though nobody got the +bytes; the sweeper evicts the entry; the SDK's retry of the same request_id +gets ``404 Future not found``, which the SDK treats as fatal. + +The server runs under a real uvicorn socket so the abandoned poll is a genuine +TCP disconnect, exactly as with the SDK. TTLs are shortened so the whole chain +takes a few seconds. +""" + +import asyncio +import sys +from contextlib import suppress +from types import SimpleNamespace + +import aiohttp +import pytest +import pytest_asyncio +import uvicorn +from sqlalchemy.ext.asyncio import create_async_engine +from sqlmodel import SQLModel + +from skyrl.tinker import api, types +from skyrl.tinker.config import EngineConfig +from skyrl.tinker.db_models import ( + RequestStatus, + enable_sqlite_wal, + get_async_database_url, +) +from skyrl.tinker.external_future_store import ExternalFutureStore + +BASE_MODEL = "test-model" +RETRIEVED_TTL_SECONDS = 1.0 +SWEEP_INTERVAL_SECONDS = 0.2 + + +class _GatedForwarder: + """Completes each forwarded sample only once the test releases it.""" + + def __init__(self, store: ExternalFutureStore): + self.store = store + self.release = asyncio.Event() + + async def call_and_store_result(self, request_id, sample_req, model_id, checkpoint_id, *, base_model=None): + await self.release.wait() + result = types.SampleOutput( + sequences=[types.GeneratedSequence(stop_reason="length", tokens=[1, 2, 3], logprobs=[-0.1, -0.2, -0.3])] + ) + await self.store.complete(request_id, result, RequestStatus.COMPLETED) + + +@pytest_asyncio.fixture() +async def served_app(tmp_path, monkeypatch): + """The real API app on a real uvicorn socket, with app.state wired the way the lifespan does.""" + monkeypatch.setattr(ExternalFutureStore, "_RETRIEVED_TTL_SECONDS", RETRIEVED_TTL_SECONDS) + monkeypatch.setattr(ExternalFutureStore, "_SWEEP_INTERVAL_SECONDS", SWEEP_INTERVAL_SECONDS) + + engine = create_async_engine(get_async_database_url(f"sqlite:///{tmp_path / 'tinker.db'}")) + enable_sqlite_wal(engine.sync_engine) + async with engine.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + + store = ExternalFutureStore() + await store.start() + forwarder = _GatedForwarder(store) + + state = api.app.state + state.engine_config = EngineConfig(base_model=BASE_MODEL) + state.db_engine = engine + state.future_waiters = {} + state.future_poller = asyncio.create_task(api.poll_futures(engine, state.future_waiters, poll_interval_sec=0.01)) + state.proto_serialization_lock = asyncio.Lock() + state.db_write_lock = asyncio.Lock() + state.sampling_model_cache = {} + state.sampling_model_cache_lock = asyncio.Lock() + state.validated_sampler_checkpoints = set() + state.sampler_checkpoint_validation_lock = asyncio.Lock() + state.external_future_store = store + state.external_inference_client = forwarder + + config = uvicorn.Config(api.app, host="127.0.0.1", port=0, log_level="warning", lifespan="off") + server = uvicorn.Server(config) + serve_task = asyncio.create_task(server.serve()) + while not server.started: + await asyncio.sleep(0.01) + port = server.servers[0].sockets[0].getsockname()[1] + + yield SimpleNamespace(url=f"http://127.0.0.1:{port}/api/v1", store=store, forwarder=forwarder) + + server.should_exit = True + await serve_task + state.future_poller.cancel() + with suppress(asyncio.CancelledError): + await state.future_poller + await store.close() + await engine.dispose() + + +@pytest.mark.asyncio +@pytest.mark.skipif(sys.platform != "linux", reason="relies on uvicorn disconnect handling over a real socket") +async def test_retry_after_client_abandoned_poll_is_served(served_app): + payload = { + "num_samples": 1, + "prompt": {"chunks": [{"type": "encoded_text", "tokens": [1, 2, 3]}]}, + "sampling_params": {"max_tokens": 3, "temperature": 1.0, "seed": 0}, + "base_model": BASE_MODEL, + } + async with aiohttp.ClientSession() as client: + async with client.post(f"{served_app.url}/asample", json=payload) as resp: + assert resp.status == 200 + request_id = (await resp.json())["request_id"] + + # The SDK's retrieve_future poll times out client-side (45s in the SDK) + # while the result is still pending, and the connection is closed. + with pytest.raises(asyncio.TimeoutError): + await client.post( + f"{served_app.url}/retrieve_future", + json={"request_id": request_id}, + timeout=aiohttp.ClientTimeout(total=0.3), + ) + await asyncio.sleep(0.2) # let the server observe the disconnect + + # The result arrives after the client gave up. The abandoned handler + # wakes, builds a response nobody will receive, and must NOT start the + # short retrieved-TTL clock. + served_app.forwarder.release.set() + await asyncio.sleep(RETRIEVED_TTL_SECONDS + 3 * SWEEP_INTERVAL_SECONDS) + + # The SDK retries the same request_id once its backoff elapses. + async with client.post(f"{served_app.url}/retrieve_future", json={"request_id": request_id}) as resp: + body = await resp.text() + assert resp.status == 200, f"retry of an undelivered result got {resp.status}: {body}" + assert types.SampleOutput.model_validate_json(body).sequences[0].tokens == [1, 2, 3] + + +@pytest.mark.asyncio +@pytest.mark.skipif(sys.platform != "linux", reason="relies on uvicorn disconnect handling over a real socket") +async def test_delivered_result_still_expires_on_retrieved_ttl(served_app): + """A result the client actually received is reclaimed on the short clock as before.""" + payload = { + "num_samples": 1, + "prompt": {"chunks": [{"type": "encoded_text", "tokens": [1, 2, 3]}]}, + "sampling_params": {"max_tokens": 3, "temperature": 1.0, "seed": 0}, + "base_model": BASE_MODEL, + } + served_app.forwarder.release.set() + async with aiohttp.ClientSession() as client: + async with client.post(f"{served_app.url}/asample", json=payload) as resp: + request_id = (await resp.json())["request_id"] + async with client.post(f"{served_app.url}/retrieve_future", json={"request_id": request_id}) as resp: + assert resp.status == 200 + await resp.read() + assert int(request_id) in served_app.store._entries + await asyncio.sleep(RETRIEVED_TTL_SECONDS + 3 * SWEEP_INTERVAL_SECONDS) + assert int(request_id) not in served_app.store._entries From 36188da4a70b9bb4fcd8902183695b1668880ef4 Mon Sep 17 00:00:00 2001 From: avigyabb <98926738+avigyabb@users.noreply.github.com> Date: Fri, 4 Sep 2026 17:30:54 -0700 Subject: [PATCH 2/2] [tinker] Survive completion bursts at the socket layer (accept backlog, keep-alive) (#2163) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stack 4/7. `uvicorn.run`: `backlog=SKYRL_HTTP_CONNECTION_LIMIT` (50k, as on Chuck's branch; the effective value is capped by `net.core.somaxconn`, raise it to match) and `timeout_keep_alive=75`. With 131072 outstanding samples and 212k-token results, each 2048-result completion burst kept the event loop busy ~16 s; uvicorn's 5 s keep-alive then closed every idle client connection, all clients reconnected at once and the 2048-entry accept backlog overflowed, refusing 109k of 131072 requests. With these settings the same run completed 130917 of 131072 with zero forwarding errors and zero reconnects (earlier runs on the 5 s keep-alive showed hundreds to thousands). Neither setting is needed for correctness: the SDK retries refused or dropped connections. They avoid the reconnect storm rather than fix a failure, and with the SDK's per-client in-flight cap the burst that overflowed the backlog does not occur. An earlier revision of this PR also exposed `sample_max_concurrent_requests` from `EngineConfig` via `/client/config`; that was dropped as unnecessary (its default equalled the SDK default, and no measured run depended on it). **Stack** (each PR retargets to `main` as the one below merges) 1. #2160 2. #2161 3. #2162 4. #2163 5. #2164 6. #2165 7. #2166 🤖 Generated with [Claude Code](https://claude.com/claude-code) --- > [!NOTE] > **Low Risk** > Server-only uvicorn listen/keep-alive defaults; no API or auth behavior change, with SDK retries as a fallback. > > **Overview** > The Tinker API server now passes **uvicorn** socket tuning so large completion bursts do not trigger mass client reconnects and accept-queue overflows. > > **`timeout_keep_alive`** is set to **75s** (via `HTTP_KEEP_ALIVE_TIMEOUT_SECONDS`) instead of uvicorn’s **5s** default, so idle SDK connections stay open while the event loop is busy for many seconds during bursts. > > **`backlog`** is set to **`SKYRL_HTTP_CONNECTION_LIMIT`** (default **50k**, overridable by env), so pending connections queue in the kernel instead of being refused when the loop cannot accept fast enough (effective cap is **`net.core.somaxconn`**). > > These are **reliability/performance** knobs, not correctness fixes—the SDK already retries refused or dropped connections. > > Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit 27c838e8696610616a6593926df0835e00bd8428. Bugbot is set up for automated code reviews on this repo. Configure [here](https://www.cursor.com/dashboard/bugbot). Signed-off-by: Avi Basnet Co-authored-by: Claude Fable 5.1 --- skyrl/tinker/api.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/skyrl/tinker/api.py b/skyrl/tinker/api.py index 22c6a5b1a9..a7e4e1f450 100644 --- a/skyrl/tinker/api.py +++ b/skyrl/tinker/api.py @@ -32,6 +32,7 @@ from sqlmodel import SQLModel, func, select from sqlmodel.ext.asyncio.session import AsyncSession +from skyrl.env_vars import SKYRL_HTTP_CONNECTION_LIMIT from skyrl.tinker import types from skyrl.tinker.config import EngineConfig, add_model, config_to_argv from skyrl.tinker.db_models import ( @@ -71,6 +72,12 @@ # How long retrieve_future waits for a result before returning 408 RETRIEVE_FUTURE_TIMEOUT_SECONDS = 300 +# Idle keep-alive for client connections. Under a burst of completions the +# event loop can be busy for many seconds; with uvicorn's 5s default every +# idle SDK connection is closed during such a burst and all clients reconnect +# at once, overflowing the accept backlog. Hold connections across bursts. +HTTP_KEEP_ALIVE_TIMEOUT_SECONDS = 75 + # How often poll_futures looks for newly finished requests. A single query # covers every waiter, so this can stay tight without the load scaling up with # the number of in-flight requests. @@ -1854,4 +1861,13 @@ async def root(): # Store config in app.state so lifespan can access it app.state.engine_config = engine_config - uvicorn.run(app, host=args.host, port=args.port, log_config=get_uvicorn_log_config()) + uvicorn.run( + app, + host=args.host, + port=args.port, + log_config=get_uvicorn_log_config(), + # Pending connections queue in the kernel while the loop is busy instead + # of being refused (effective value is capped by net.core.somaxconn). + backlog=SKYRL_HTTP_CONNECTION_LIMIT, + timeout_keep_alive=HTTP_KEEP_ALIVE_TIMEOUT_SECONDS, + )