From 2bf66dfa3ba1d658be164741774a1becd68dd4e8 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Sat, 19 Sep 2026 08:18:32 +0000 Subject: [PATCH 1/2] feat(client,transport): expose httpx connection-pool limits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `models.run` is awaited server-side, so each concurrent run holds its pooled connection for the whole generation. The pool was always httpx's default (`max_connections=100`) and nothing on the public surface could change it, so a caller fanning out past 100 concurrent runs had its 101st call queue behind a generation and — if nothing freed up — end in `httpx.PoolTimeout` after the run timeout rather than in a result. `Comfy`/`AsyncComfy` (and `ComfyLow`/`AsyncComfyLow` underneath) now take a keyword-only `limits: httpx.Limits | None = None`. `None` is dropped rather than forwarded — httpx types the parameter as a `Limits` and does not accept `None` — so the default path stays httpx's own default exactly as before. An injected `client=` owns its pool and ignores `limits`, as it already does `timeout`. --- CHANGELOG.md | 9 +++++++ README.md | 17 ++++++++++++ src/comfy_low/transport.py | 28 ++++++++++++++++++-- src/comfy_sdk/client.py | 23 ++++++++++++++++ tests/test_models_namespace.py | 48 ++++++++++++++++++++++++++++++++++ 5 files changed, 123 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a81f44d..96f52cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,15 @@ the fuller account of each version, including verification notes. ## [Unreleased] +### Added + +- `Comfy(limits=...)` / `AsyncComfy(limits=...)` — an `httpx.Limits` sizing the connection pool + the client's namespaces share. The default is unchanged (httpx's own, 100 connections). Raise + `max_connections` when fanning out more than 100 concurrent `models.run` calls: each holds its + pooled connection for the whole generation, so past 100 the next call waits on a generation and + can end in `httpx.PoolTimeout`. `ComfyLow`/`AsyncComfyLow` take the same keyword; an injected + `client=` owns its own pool and ignores it, as it already does `timeout`. + ### Fixed - **`except RouterError` now catches every Comfy Router refusal.** `insufficient_credits`, diff --git a/README.md b/README.md index f3daeda..1b4b008 100644 --- a/README.md +++ b/README.md @@ -509,6 +509,23 @@ indefinitely. Each call also sends a fresh `Idempotency-Key`, so an accidental exact resend is rejected by the server instead of billing a second generation; pass `idempotency_key=` to choose the value yourself. +Holding the connection that long also means holding a *pooled* one: a run +occupies its slot in the client's httpx connection pool for the whole +generation. That pool is httpx's default — 100 connections — so a fan-out past +100 concurrent runs queues the 101st behind a generation finishing, and if +nothing frees up it waits out the same 10-minute timeout and then raises +`httpx.PoolTimeout` rather than returning a result. Size the pool up front +instead of discovering that as a ten-minute stall: + +```python +client = Comfy(limits=httpx.Limits(max_connections=250)) +``` + +Raising the client's pool is not a licence to exceed the concurrency your +account is allowed: the server may also bound how many requests a customer has +in flight at once, and a pool wider than that limit just moves the queue from +your process to theirs. + ### `models.submit` — queue it, collect it later `run` holds one connection open until the generation is finished. When the diff --git a/src/comfy_low/transport.py b/src/comfy_low/transport.py index c975d41..cefb85b 100644 --- a/src/comfy_low/transport.py +++ b/src/comfy_low/transport.py @@ -621,12 +621,26 @@ def __init__( *, client: httpx.Client | None = None, timeout: float | None = 30.0, + limits: httpx.Limits | None = None, client_info: str | None = None, router_base_url: str = ROUTER_BASE_URL, ) -> None: + """``limits`` sizes the connection pool of the client built here. + + ``None`` means httpx's own default (100 connections, 20 kept alive) — + it is not forwarded, because ``httpx.Client`` types ``limits`` as a + ``Limits`` and does not accept ``None`` for it. + + Passing ``client=`` hands pool ownership to the caller, so ``limits`` + is ignored on that path exactly as ``timeout`` already is: the injected + client was built with its own. + """ self._p = _Prepared(base_url, api_key, client_info, router_base_url) self._own_client = client is None - self._client = client or httpx.Client(timeout=timeout, follow_redirects=True) + kwargs: dict[str, Any] = {"timeout": timeout, "follow_redirects": True} + if limits is not None: + kwargs["limits"] = limits + self._client = client or httpx.Client(**kwargs) # -- configuration ---------------------------------------------------- # Read-only views of the settings this transport was built with, so a layer @@ -1109,12 +1123,22 @@ def __init__( *, client: httpx.AsyncClient | None = None, timeout: float | None = 30.0, + limits: httpx.Limits | None = None, client_info: str | None = None, router_base_url: str = ROUTER_BASE_URL, ) -> None: + """``limits`` sizes the pool of the client built here — see :class:`ComfyLow`. + + Same two rules: ``None`` means httpx's own default rather than being + forwarded, and an injected ``client=`` owns its pool, so ``limits`` is + ignored there. + """ self._p = _Prepared(base_url, api_key, client_info, router_base_url) self._own_client = client is None - self._client = client or httpx.AsyncClient(timeout=timeout, follow_redirects=True) + kwargs: dict[str, Any] = {"timeout": timeout, "follow_redirects": True} + if limits is not None: + kwargs["limits"] = limits + self._client = client or httpx.AsyncClient(**kwargs) # -- configuration (mirrors :class:`ComfyLow`) ------------------------- @property diff --git a/src/comfy_sdk/client.py b/src/comfy_sdk/client.py index 3bc6b9f..bfd7d1d 100644 --- a/src/comfy_sdk/client.py +++ b/src/comfy_sdk/client.py @@ -43,6 +43,8 @@ from typing import Any from urllib.parse import urlsplit +import httpx + from comfy_low.errors import ApiError from comfy_low.transport import ROUTER_BASE_URL, AsyncComfyLow, ComfyLow, origin @@ -230,6 +232,15 @@ class Comfy: It does not govern this client's own ``submit``/``run`` (the workflow surface), whose 429 handling follows the server's own ``Retry-After`` instead. + + ``limits`` configures the httpx connection pool every namespace on this + client shares; the default is httpx's own (100 connections, 20 kept + alive). Raise ``max_connections`` when fanning out more than 100 + concurrent ``models.run`` calls, because each one holds its pooled + connection for the whole generation — past that the next call waits for a + generation to finish and eventually fails with ``httpx.PoolTimeout``:: + + Comfy(limits=httpx.Limits(max_connections=250)) """ def __init__( @@ -237,6 +248,7 @@ def __init__( *, api_key: str | None = None, timeout: float | None = 30.0, + limits: httpx.Limits | None = None, client_info: str | None = None, retry: RetryPolicy = DEFAULT_RETRY, ) -> None: @@ -246,6 +258,7 @@ def __init__( base_url, key, timeout=timeout, + limits=limits, client_info=client_info, router_base_url=_resolve_router_base_url(), ) @@ -394,6 +407,14 @@ class AsyncComfy: Same credential resolution (explicit ``api_key`` → ``COMFY_API_KEY`` → :class:`~comfy_sdk.exceptions.MissingApiKey` on Comfy Cloud) and the same key-free ``repr``. + + ``limits`` configures the shared httpx connection pool exactly as on + :class:`Comfy`, and matters more here: fanning out past httpx's default of + 100 connections is easier to do with ``asyncio.gather`` than with threads, + and every awaited ``models.run`` holds its connection for the whole + generation:: + + AsyncComfy(limits=httpx.Limits(max_connections=250)) """ def __init__( @@ -401,6 +422,7 @@ def __init__( *, api_key: str | None = None, timeout: float | None = 30.0, + limits: httpx.Limits | None = None, client_info: str | None = None, retry: RetryPolicy = DEFAULT_RETRY, ) -> None: @@ -410,6 +432,7 @@ def __init__( base_url, key, timeout=timeout, + limits=limits, client_info=client_info, router_base_url=_resolve_router_base_url(), ) diff --git a/tests/test_models_namespace.py b/tests/test_models_namespace.py index b8197da..4fd1167 100644 --- a/tests/test_models_namespace.py +++ b/tests/test_models_namespace.py @@ -12,6 +12,7 @@ import httpx import comfy_sdk +from comfy_low.transport import ComfyLow from comfy_sdk import AsyncComfy, Comfy from comfy_sdk.models import AsyncModels, Models @@ -70,6 +71,53 @@ async def test_a_timeout_change_on_the_async_client_is_visible_through_models(se assert client.models.timeout.read == 1.25 +# --- the pool the namespace shares -------------------------------------- +# +# `models.run` holds its pooled connection for the whole generation, so the +# pool's ceiling is what caps a fan-out of concurrent runs. These reach into +# httpx/httpcore internals (`_transport._pool._max_connections`) because httpx +# exposes no public getter for the limits a client was built with; the +# alternative is asserting nothing about whether `limits=` arrived at all. + + +def test_limits_sizes_the_pool_the_client_builds(server) -> None: + with Comfy(limits=httpx.Limits(max_connections=250)) as client: + assert client._low._client._transport._pool._max_connections == 250 + + +async def test_limits_sizes_the_pool_the_async_client_builds(server) -> None: + async with AsyncComfy(limits=httpx.Limits(max_connections=250)) as client: + assert client._low._client._transport._pool._max_connections == 250 + + +def test_no_limits_leaves_httpxs_own_default_pool(server) -> None: + # `limits=None` is dropped rather than forwarded — httpx types the + # parameter as a `Limits` and does not accept `None` — so the default path + # has to land on httpx's own DEFAULT_LIMITS, 100 connections. + with Comfy() as client: + assert client._low._client._transport._pool._max_connections == 100 + + +async def test_no_limits_leaves_the_async_client_httpxs_default_pool(server) -> None: + async with AsyncComfy() as client: + assert client._low._client._transport._pool._max_connections == 100 + + +def test_an_injected_client_keeps_its_own_pool(server) -> None: + # Injecting `client=` hands pool ownership to the caller, so `limits` is + # ignored there exactly as `timeout` already is: the pool stays the + # injected client's default 100, not the 5 asked for here. + with httpx.Client() as injected: + low = ComfyLow( + server.base_url, + "ck_test", + client=injected, + limits=httpx.Limits(max_connections=5), + ) + assert low._client is injected + assert low._client._transport._pool._max_connections == 100 + + def test_models_sends_the_host_clients_credentials(server) -> None: server.state.require_auth = True with Comfy(api_key="k-first") as client: From 9a3af4e5f49084f735694d33a2a2dba67e499037 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Sun, 20 Sep 2026 06:28:46 +0000 Subject: [PATCH 2/2] fix(tests): reconcile the credits_used tripwire now that the spec declares it The vendored spec sync (86d8f2c) added X-Comfy-Credits-Used to runRouterModel's declared 200 headers after the credits_used lift landed as an intentionally-undeclared header, so test_an_undeclared_lift_stays_undeclared_until_someone_reconciles_it started failing on main itself -- exactly the reconciliation its docstring calls for. Move credits_used into _CONTRACT_HEADER_LIFTS and drop the now-satisfied tripwire. test_the_lift_actually_reads_the_declared_name's generic "x" sample doesn't round-trip for credits_used, which normalises non-decimal values to None -- give it a decimal sample instead so the test still proves the lift reads the declared name. --- tests/test_router_spec_contract.py | 39 +++++++----------------------- 1 file changed, 9 insertions(+), 30 deletions(-) diff --git a/tests/test_router_spec_contract.py b/tests/test_router_spec_contract.py index 4e51f76..6b0af95 100644 --- a/tests/test_router_spec_contract.py +++ b/tests/test_router_spec_contract.py @@ -294,11 +294,15 @@ def test_the_bound_path_has_exactly_the_two_segments_the_binding_fills() -> None "dropped_params": "X-Comfy-Router-Dropped-Params", "replayed": "Idempotent-Replayed", "request_id": "X-Comfy-Request-Id", + "credits_used": "X-Comfy-Credits-Used", } -#: Lifted by the SDK but NOT declared on the contract's 200 -- see the tripwire -#: test at the bottom of this file. -_UNDECLARED_HEADER_LIFTS = {"credits_used": "X-Comfy-Credits-Used"} +#: Header value to use for the "present" case in +#: test_the_lift_actually_reads_the_declared_name. "x" round-trips fine for +#: the pass-through lifts, but credits_used parses its header as a decimal, +#: so a non-numeric sample would normalise to None same as absent and the +#: test would pass for the wrong reason. +_SAMPLE_HEADER_VALUES: dict[str, str] = {"credits_used": "1.25"} def _declared_run_response_headers() -> set[str]: @@ -333,34 +337,9 @@ def test_the_lift_actually_reads_the_declared_name(field: str, header: str) -> N to fail. """ absent = getattr(_run_result({}, {}), field) - present = getattr(_run_result({}, {header: "x"}), field) + sample = _SAMPLE_HEADER_VALUES.get(field, "x") + present = getattr(_run_result({}, {header: sample}), field) assert present != absent, ( f"_run_result ignored {header!r}: RouterRunResult.{field} read {absent!r} both with " f"the header and without it, so the lift is reading some other name." ) - - -@pytest.mark.parametrize(("field", "header"), sorted(_UNDECLARED_HEADER_LIFTS.items())) -def test_an_undeclared_lift_stays_undeclared_until_someone_reconciles_it( - field: str, header: str -) -> None: - """Tripwire, and deliberately asserting the *absence*. - - ``credits_used`` is lifted from a header the vendored contract does not - declare anywhere -- the 200's only cost headers are the - ``X-Committed-Spend-*`` trio, which is a different quantity (USD cents of - in-flight commitment, not the price of this run). Nothing in the suite can - catch a wrong name here, because every test configures its stub to emit the - exact literal the lift reads. - - That gap is tracked, not accepted. This test fails the moment a spec sync - declares the header, which is the signal to move the entry up into - ``_CONTRACT_HEADER_LIFTS`` and get it pinned like the rest. It also fails - if the header is declared under a *different* name for the same quantity, - because the reconciliation is the same either way. - """ - declared = _declared_run_response_headers() - assert header not in declared, ( - f"the vendored spec now declares {header!r}: move {field!r} from " - f"_UNDECLARED_HEADER_LIFTS into _CONTRACT_HEADER_LIFTS so it is pinned." - )