diff --git a/CHANGELOG.md b/CHANGELOG.md index dfc9da6..0125ba5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `stocks.quotes()` now requests `stocks/quotes/?symbols=...` instead of the deprecated `stocks/bulkquotes/`; the method, its parameters and its output are unchanged (#74) - **BREAKING**: `UserRateLimits` speaks in API credits, as the product does (SDK requirements §8.1): `requests_limit` → `credit_limit`, `requests_remaining` → `credits_remaining`, `requests_consumed` → `credits_consumed`, `requests_reset` → `reset_time`. The old names are gone, with no aliases; its string form now reads `Credits used X/Y, remaining: Z, reset at: ` (#48) - The pre-flight credit check reads a private, thread-safe tracker fed by every response that carries credit headers, error answers included; out-of-order responses (an older reset window, a higher remaining in the same window) are ignored, and every response logs its credits at DEBUG. An envelope with a `credit_limit` of zero is not recorded: the API answers a missing or unknown token with demo data and a `0/0` envelope from another window, which would otherwise tell the check this account has no credits. `client.utilities.user()` is asked for precisely to learn the balance, so its answer replaces the state rather than being weighed against it (#49) +- A client sends at most 50 requests at a time, the concurrent-request limit the API documents (it states that requests above it get a `429`, SDK requirements §12). Every request of the client takes one of 50 slots for the length of its HTTP exchange, whichever endpoint, fan-out or thread sends it, and a request that finds none waits for a slot instead of being sent. Before, `stocks.candles()` and `options.quotes()` each ran 50 workers of their own, so two calls at the same time put 100 requests on the wire. A retry waiting out its backoff holds no slot. The cap is fixed and has no constructor parameter, and it belongs to the client: two clients on the same account have 50 slots each, so share one client between threads rather than building one per thread (#65) ### Added diff --git a/README.md b/README.md index 990b302..29d7294 100644 --- a/README.md +++ b/README.md @@ -905,8 +905,10 @@ The SDK uses Pydantic for input validation: The SDK uses concurrent requests for efficient data fetching in specific scenarios: -- **`stocks.candles()`**: For intraday resolutions (minutely/hourly), large date ranges are automatically split into year-long, non-overlapping chunks and fetched concurrently (up to 50 concurrent requests by default) -- **`options.quotes()`**: Multiple option symbols are fetched concurrently (up to 50 concurrent requests by default) +- **`stocks.candles()`**: For intraday resolutions (minutely/hourly), large date ranges are automatically split into year-long, non-overlapping chunks and fetched concurrently +- **`options.quotes()`**: Multiple option symbols are fetched concurrently + +A client sends at most 50 requests at a time, the concurrent-request limit the API documents (it states that requests above it get a `429`). Every request takes one of the client's 50 slots for as long as its HTTP exchange lasts, whichever endpoint, fan-out or thread sends it, and a request that finds none waits for a slot to free instead of being sent. A retry waiting out its backoff holds no slot. The cap is fixed and has no constructor parameter. It belongs to the client: two clients on the same account have 50 slots each, so share one client between threads rather than building one per thread. When concurrent requests are used, responses are automatically merged into a single result. A chunk or symbol with no data is left out of the merge; if every part has no data the call returns an empty result, and if the API answered with nothing usable at all a `MarketdataHttpError` is raised. diff --git a/docs/options.md b/docs/options.md index 648cd37..9e3719a 100644 --- a/docs/options.md +++ b/docs/options.md @@ -575,7 +575,7 @@ if csv_file: ### `quotes()` -Fetches options quotes for one or more option symbols. This method includes API status checking, automatic retry logic, and supports concurrent requests for multiple symbols (up to 50 concurrent requests by default). Each symbol's request retries on its own, so a failed symbol never re-sends the healthy ones. A symbol whose response body cannot be decoded raises `ParseError`, as in every other method, on every output format; in CSV output the merged file keeps the header the API sent (the requested `columns`, the human-readable names) and every row of every symbol. On the other formats the symbols are merged on the model's columns among those the API sent, in the order it sent them (the request order under `columns=`), and a symbol whose answer lacks one of them, or carries one that is not a list as long as its others, raises `ParseError` as well, since merging it would put the next symbol's values on its row. +Fetches options quotes for one or more option symbols. This method includes API status checking, automatic retry logic, and supports concurrent requests for multiple symbols (within the client's 50 request slots). Each symbol's request retries on its own, so a failed symbol never re-sends the healthy ones. A symbol whose response body cannot be decoded raises `ParseError`, as in every other method, on every output format; in CSV output the merged file keeps the header the API sent (the requested `columns`, the human-readable names) and every row of every symbol. On the other formats the symbols are merged on the model's columns among those the API sent, in the order it sent them (the request order under `columns=`), and a symbol whose answer lacks one of them, or carries one that is not a list as long as its others, raises `ParseError` as well, since merging it would put the next symbol's values on its row. > **Note:** The `symbols` parameter can be passed as the first positional argument or as a keyword argument. All other parameters must be keyword-only. @@ -612,7 +612,7 @@ Fetches options quotes for one or more option symbols. This method includes API - If `output_format=OutputFormat.CSV`: A string containing the filename where CSV data was written (merged from all requested symbols) - Raises a `BaseMarketdataException` subclass if an error occurs (rate limits, validation errors, request failures, no valid responses received, etc.); see the [README](../README.md#error-handling) -> **Note:** The method returns the requested output, an empty result when the API has no data for every symbol (`404` with `s: "no_data"`, see the [README](../README.md#no-data-is-not-an-error)), or raises. The method uses concurrent requests (up to 50 concurrent requests by default) to fetch quotes for multiple symbols efficiently. +> **Note:** The method returns the requested output, an empty result when the API has no data for every symbol (`404` with `s: "no_data"`, see the [README](../README.md#no-data-is-not-an-error)), or raises. The method uses concurrent requests, within the client's 50 request slots (see the [README](../README.md#concurrent-requests)), to fetch quotes for multiple symbols efficiently. #### Examples diff --git a/docs/stocks.md b/docs/stocks.md index d9ac434..d391e7f 100644 --- a/docs/stocks.md +++ b/docs/stocks.md @@ -320,7 +320,7 @@ print(df) ### `candles()` -Fetches stock candles (OHLCV data) for a symbol with support for various timeframes and date ranges. This method includes API status checking and automatic retry logic. For intraday resolutions (minutely/hourly), large date ranges are automatically split into year-long chunks and fetched concurrently using up to 50 concurrent requests. Each chunk's request retries on its own, so a failed chunk never re-sends the others. +Fetches stock candles (OHLCV data) for a symbol with support for various timeframes and date ranges. This method includes API status checking and automatic retry logic. For intraday resolutions (minutely/hourly), large date ranges are automatically split into year-long chunks and fetched concurrently, within the client's 50 request slots (see the [README](../README.md#concurrent-requests)). Each chunk's request retries on its own, so a failed chunk never re-sends the others. > **Note:** The `symbol` parameter can be passed as the first positional argument or as a keyword argument. All other parameters must be keyword-only. @@ -331,8 +331,8 @@ Fetches stock candles (OHLCV data) for a symbol with support for various timefra - Numeric with unit: `"1"`, `"15M"`, `"1H"`, `"1D"`, `"1W"`, `"1M"`, `"1Y"` - Unit only: `"M"`, `"H"`, `"D"`, `"W"`, `"M"`, `"Y"` - Descriptive: `"minutely"`, `"hourly"`, `"daily"`, `"weekly"`, `"monthly"`, `"yearly"` -- `from_date` (datetime.date, optional): The start date to fetch candles for. When both `from_date` and `to_date` are provided, the date range is automatically split into year-long chunks and fetched concurrently for intraday resolutions (minutely/hourly) using up to 50 concurrent requests. For non-intraday resolutions (daily/weekly/monthly/yearly), the entire date range is fetched in a single request without splitting. -- `to_date` (datetime.date, optional): The end date to fetch candles for. When both `from_date` and `to_date` are provided, the date range is automatically split into year-long chunks and fetched concurrently for intraday resolutions (minutely/hourly) using up to 50 concurrent requests. For non-intraday resolutions (daily/weekly/monthly/yearly), the entire date range is fetched in a single request without splitting. +- `from_date` (datetime.date, optional): The start date to fetch candles for. When both `from_date` and `to_date` are provided, the date range is automatically split into year-long chunks and fetched concurrently for intraday resolutions (minutely/hourly), within the client's 50 request slots. For non-intraday resolutions (daily/weekly/monthly/yearly), the entire date range is fetched in a single request without splitting. +- `to_date` (datetime.date, optional): The end date to fetch candles for. When both `from_date` and `to_date` are provided, the date range is automatically split into year-long chunks and fetched concurrently for intraday resolutions (minutely/hourly), within the client's 50 request slots. For non-intraday resolutions (daily/weekly/monthly/yearly), the entire date range is fetched in a single request without splitting. - `countback` (int, optional): The number of candles to fetch (alternative to date range) - `extended` (bool, optional): Whether to fetch extended candles (pre-market and after-hours data) - `adjust_splits` (bool, optional): Whether to adjust for stock splits. Uses API alias `adjustsplits`. @@ -368,7 +368,7 @@ Fetches stock candles (OHLCV data) for a symbol with support for various timefra When both `from_date` and `to_date` are provided, the method behavior depends on the resolution: -- **For intraday resolutions (minutely/hourly)**: The date range is automatically split into year-long chunks, fetched concurrently using a thread pool executor (up to 50 concurrent requests), and merged into a single response. Chunks never share a calendar day, so no candle is requested twice. This allows efficient fetching of large historical date ranges without manual pagination. +- **For intraday resolutions (minutely/hourly)**: The date range is automatically split into year-long chunks, fetched concurrently using a thread pool executor within the client's 50 request slots, and merged into a single response. Chunks never share a calendar day, so no candle is requested twice. This allows efficient fetching of large historical date ranges without manual pagination. - **For non-intraday resolutions (daily/weekly/monthly/yearly)**: The entire date range is fetched in a single request without splitting. diff --git a/src/marketdata/client.py b/src/marketdata/client.py index 5145c45..a5a4f1d 100644 --- a/src/marketdata/client.py +++ b/src/marketdata/client.py @@ -1,3 +1,4 @@ +import threading import time from importlib.metadata import version from logging import DEBUG, INFO, Logger @@ -21,6 +22,7 @@ from marketdata.input_types.base import UserUniversalAPIParams from marketdata.internal_settings import ( HEADER_REQUEST_ID, + MAX_CONCURRENT_REQUESTS, MAX_CREDIT_WINDOW_SECONDS, MAX_RETRY_ATTEMPTS, NO_TOKEN_VALUE, @@ -53,6 +55,18 @@ def __init__( logger: Logger = None, max_retries: int = MAX_RETRY_ATTEMPTS, ): + """Build the client and, when it has a token, seed the credit tracker. + + Args: + token: The API token. Falls back to the ``MARKETDATA_TOKEN`` + setting; with neither, the client starts in demo mode. + logger: The logger to write to. The SDK's own when omitted. + max_retries: Retries after the first attempt of each request. + + Raises: + ValueError: ``max_retries`` is negative. + BaseMarketdataException: The start-up ``/user/`` request failed. + """ if max_retries < 0: raise ValueError("max_retries must be >= 0") self.token = token or settings.marketdata_token @@ -75,7 +89,12 @@ def __init__( self.client = self._get_client() self.default_params = UserUniversalAPIParams() - # The private credit tracker behind the pre-flight check (#49); the + # One budget for every request of this client, whichever endpoint or + # fan-out sends it. It exists before `_setup_rate_limits`, whose + # `/user/` request already takes a slot. + self._request_slots = threading.BoundedSemaphore(MAX_CONCURRENT_REQUESTS) + + # The private credit tracker behind the pre-flight check; the # request-scoped numbers travel with each result (marketdata.get_meta). self._rate_limits = RateLimitTracker() self._setup_rate_limits() @@ -298,6 +317,38 @@ def _make_request( response_log_level: int = INFO, **kwargs, ) -> Response: + """Send one HTTP request and map its outcome to the SDK's exceptions. + + The request holds one of the client's request slots only for the HTTP + exchange itself: it waits for a free slot after the pre-flight check, + and a retry sleeping through its backoff holds none. + + Args: + method: The HTTP method. + url: The path, relative to the base URL. + check_rate_limits: Refuse the request when the account is known to + have no credits left. + part_of_result: Record the response's metadata in the caller's + scope. + include_api_version: Prefix the path with the API version. + authoritative_credits: Let this response replace the tracked credit + state instead of being weighed against it. + response_log_level: The level of the line logged for the response. + **kwargs: Passed to ``httpx.Client.request``. + + Returns: + The response, including a 404 without ``errmsg``, which is the + API's "no data" answer. + + Raises: + RateLimitError: The pre-flight check refused the request, or the + API answered 429. + NetworkError: There was no usable answer (a connection failure, a + timeout, a protocol or proxy error). + ParseError: The body does not match its ``Content-Encoding``. + MarketdataHttpError: The API answered with an error status; the + subclass names it. + """ if self.token is NO_TOKEN_VALUE: check_rate_limits = False @@ -311,7 +362,8 @@ def _make_request( # No `timeout=` here: the client carries the fixed one (§10), # so there is one place to read it and no argument a resource # could use to give itself a longer bound. - response = self.client.request(method, url, **kwargs) + with self._request_slots: + response = self.client.request(method, url, **kwargs) except DecodingError as exc: # The API answered but the body does not match its Content-Encoding # (an intercepting proxy): the answer is unusable, not missing. diff --git a/src/tests/test_request_pool.py b/src/tests/test_request_pool.py new file mode 100644 index 0000000..fcb32e2 --- /dev/null +++ b/src/tests/test_request_pool.py @@ -0,0 +1,535 @@ +"""One request pool per client (SDK requirements §12): at most 50 requests in +flight, shared by every endpoint and every fan-out, and a freed slot goes to the +next request at once.""" + +import collections +import threading +import time + +import httpx +import pytest + +from marketdata.api_status import API_STATUS_DATA +from marketdata.client import MarketDataClient +from marketdata.exceptions import NetworkError, ServerError +from marketdata.input_types.base import OutputFormat + +PRICES_URL = "https://api.marketdata.app/v1/stocks/prices/" +MARKETS_STATUS_URL = "https://api.marketdata.app/v1/markets/status/" +PRICES_CSV = ( + "symbol,mid,change,changepct,updated\nAAPL,280.02,-0.68,-0.0024,1764950546\n" +) +FAN_OUT_URL = r".*/v1/(options/quotes|stocks/candles)/.*" +POOL_SIZE = 50 +CALLERS = 120 +ENOUGH = 1000 +WAIT = 10.0 +HOLD = 20.0 +FINISH = 15.0 + + +class Gate: + """Holds every request inside the mocked transport and counts them.""" + + def __init__(self, answer): + """Build a gate that lets nothing through until told to. + + Args: + answer: Builds the response of a request once the gate lets it go. + """ + self._answer = answer + self._changed = threading.Condition() + self._permits = threading.Semaphore(0) + self.current = 0 + self.started = 0 + self.peak = 0 + + def handle(self, request): + """Count a request in, hold it until it is let go, and count it out. + + Args: + request: The request the client sent. + + Returns: + The response built by ``answer``. + """ + with self._changed: + self.current += 1 + self.started += 1 + self.peak = max(self.peak, self.current) + self._changed.notify_all() + try: + self._permits.acquire(timeout=HOLD) + finally: + with self._changed: + self.current -= 1 + self._changed.notify_all() + return self._answer(request) + + def let_go(self, count): + """Let held requests go. + + Args: + count: How many requests to let go. + """ + self._permits.release(count) + + def wait_for(self, condition): + """Wait until a condition on the gate holds. + + Args: + condition: Takes the gate and says whether the wait is over. + + Returns: + Whether the condition held before ``WAIT`` seconds went by. + """ + with self._changed: + return self._changed.wait_for(lambda: condition(self), WAIT) + + def settle(self, quiet=0.3): + """Wait until the number of held requests stops changing. + + Args: + quiet: How long the number must stay the same, in seconds. + + Returns: + The number of held requests once it has been steady that long. + """ + deadline = time.monotonic() + WAIT + with self._changed: + while time.monotonic() < deadline: + before = self.current + if not self._changed.wait_for(lambda: self.current != before, quiet): + break + return self.current + + +class Callers: + """Runs calls in daemon threads and keeps what each returned or raised. + + The threads are daemons so that a call stuck waiting for a slot cannot keep + the test process alive. + """ + + def __init__(self, calls): + """Start one thread per call. + + Args: + calls: The callables to run, each in a thread of its own. + """ + self.outcomes = [None] * len(calls) + self._threads = [ + threading.Thread(target=self._run, args=(index, call), daemon=True) + for index, call in enumerate(calls) + ] + for thread in self._threads: + thread.start() + + def _run(self, index, call): + """Run one call and keep its outcome. + + Args: + index: Where the outcome goes in ``outcomes``. + call: The callable to run. + """ + try: + self.outcomes[index] = call() + except BaseException as exc: + self.outcomes[index] = exc + + def finish(self): + """Wait for every call to end. + + Returns: + What each call returned or raised, in the order they were given. + + Raises: + AssertionError: A call is still running after ``FINISH`` seconds. + """ + deadline = time.monotonic() + FINISH + for thread in self._threads: + thread.join(max(0.0, deadline - time.monotonic())) + stuck = sum(thread.is_alive() for thread in self._threads) + assert not stuck, f"{stuck} calls still waiting after {FINISH} seconds" + return self.outcomes + + +def raised(outcomes): + """Pick the exceptions out of what a set of calls produced. + + Args: + outcomes: What each call returned or raised. + + Returns: + The outcomes that are exceptions. + """ + return [outcome for outcome in outcomes if isinstance(outcome, BaseException)] + + +def connect_timeout(request): + """Fail as a request that never connects does. + + Args: + request: The request the client sent. + + Raises: + httpx.ConnectTimeout: Always. + """ + raise httpx.ConnectTimeout("slow") + + +def connection_refused(request): + """Fail as a request the server refuses does. + + Args: + request: The request the client sent. + + Raises: + httpx.ConnectError: Always. + """ + raise httpx.ConnectError("refused") + + +def unexpected_failure(request): + """Fail with an exception the client does not know. + + Args: + request: The request the client sent. + + Raises: + RuntimeError: Always. + """ + raise RuntimeError("unexpected") + + +def service_unavailable(request): + """Answer a request with a 503. + + Args: + request: The request the client sent. + + Returns: + A 503 response. + """ + return httpx.Response(503, json={}) + + +@pytest.fixture(autouse=True) +def _no_sleep(monkeypatch): + """Skip the retry backoff.""" + monkeypatch.setattr("time.sleep", lambda *_: None) + + +@pytest.fixture(autouse=True) +def _synchronous_status_refresh(monkeypatch): + """Refresh ``/status/`` inline, so no helper thread holds a slot while a test + counts them.""" + monkeypatch.setattr( + API_STATUS_DATA, "_trigger_async_refresh", lambda c: API_STATUS_DATA.refresh(c) + ) + + +def test_120_callers_never_put_more_than_50_requests_in_flight( + load_json, respx_mock, client +): + """Callers beyond the 50th wait for a slot instead of reaching the API.""" + body = load_json("stocks_prices_response_200") + gate = Gate(lambda request: httpx.Response(200, json=body)) + respx_mock.get(PRICES_URL).mock(side_effect=gate.handle) + + callers = Callers( + [ + lambda: client.stocks.prices("AAPL", output_format=OutputFormat.JSON) + for _ in range(CALLERS) + ] + ) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + settled = gate.settle() + gate.let_go(ENOUGH) + outcomes = callers.finish() + + assert settled == POOL_SIZE, ( + f"{settled} requests in flight at once, the pool allows {POOL_SIZE}" + ) + assert gate.peak <= POOL_SIZE + assert gate.started == CALLERS + assert not raised(outcomes) + + +def test_two_fan_outs_started_together_share_the_pool(load_json, respx_mock, client): + """A ``stocks.candles`` and an ``options.quotes`` running at once draw from + one budget of 50 requests.""" + quotes = load_json("options_quotes_response_200") + candles = load_json("stocks_candles_response_200") + + def answer(request): + """Answer with the body of the endpoint the request is for. + + Args: + request: The request the client sent. + + Returns: + A 200 response. + """ + body = candles if "/stocks/candles/" in request.url.path else quotes + return httpx.Response(200, json=body) + + gate = Gate(answer) + respx_mock.get(url__regex=FAN_OUT_URL).mock(side_effect=gate.handle) + symbols = [f"AAPL250117C{150000 + 500 * i:08d}" for i in range(60)] + start = threading.Barrier(2, timeout=WAIT) + + def quote_symbols(): + """Quote 60 symbols once both fan-outs are ready.""" + start.wait() + return client.options.quotes(symbols, output_format=OutputFormat.JSON) + + def fetch_candles(): + """Fetch 60 years of hourly candles, one request per year, once both + fan-outs are ready.""" + start.wait() + return client.stocks.candles( + "AAPL", + resolution="H", + from_date="1966-01-01", + to_date="2026-01-01", + output_format=OutputFormat.JSON, + ) + + callers = Callers([quote_symbols, fetch_candles]) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + settled = gate.settle() + gate.let_go(ENOUGH) + outcomes = callers.finish() + + assert settled == POOL_SIZE, ( + f"{settled} requests in flight at once, the pool allows {POOL_SIZE}" + ) + assert gate.peak <= POOL_SIZE + assert not raised(outcomes) + + +def test_a_freed_slot_starts_the_next_request_without_waiting_for_the_others( + load_json, respx_mock, client +): + """With 50 requests in flight and one waiting, one completion starts the + waiting one while the other 49 are still running.""" + body = load_json("stocks_prices_response_200") + gate = Gate(lambda request: httpx.Response(200, json=body)) + respx_mock.get(PRICES_URL).mock(side_effect=gate.handle) + + callers = Callers( + [ + lambda: client.stocks.prices("AAPL", output_format=OutputFormat.JSON) + for _ in range(POOL_SIZE + 1) + ] + ) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + assert gate.settle() == POOL_SIZE + gate.let_go(1) + + started = gate.wait_for(lambda g: g.started == POOL_SIZE + 1) + in_flight = gate.current + gate.let_go(ENOUGH) + outcomes = callers.finish() + + assert started, "the waiting request did not start when a slot freed" + assert in_flight == POOL_SIZE + assert not raised(outcomes) + + +@pytest.mark.parametrize( + "output_format", + [ + OutputFormat.INTERNAL, + OutputFormat.JSON, + OutputFormat.CSV, + OutputFormat.DATAFRAME, + ], + ids=["internal", "json", "csv", "dataframe"], +) +def test_the_output_format_does_not_change_what_the_pool_allows( + output_format, load_json, respx_mock, client +): + """Calls asking for any of the four output formats draw from the same 50 + slots: with the pool full, the 10 beyond it wait and every call is served. + + Args: + output_format: The output format every call asks for. + """ + body = load_json("stocks_prices_response_200") + + def answer(request): + """Answer with a body in the format the request asked for. + + Args: + request: The request the client sent. + + Returns: + A 200 response, CSV when the request asked for CSV and JSON + otherwise. + """ + if request.url.params.get("format") == "csv": + return httpx.Response(200, text=PRICES_CSV) + return httpx.Response(200, json=body) + + def call(index): + """Make one ``stocks.prices`` call in the output format of the test. + + Args: + index: Which call of the burst this is. It names the CSV file, which + is created exclusively, so no two calls share one. + + Returns: + What the call returned. + """ + extra = ( + {"filename": f"prices-{index}.csv"} if output_format.name == "CSV" else {} + ) + return client.stocks.prices("AAPL", output_format=output_format, **extra) + + gate = Gate(answer) + respx_mock.get(PRICES_URL).mock(side_effect=gate.handle) + + callers = Callers( + [lambda index=index: call(index) for index in range(POOL_SIZE + 10)] + ) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + settled = gate.settle() + gate.let_go(ENOUGH) + outcomes = callers.finish() + + assert settled == POOL_SIZE, ( + f"{settled} requests in flight at once, the pool allows {POOL_SIZE}" + ) + assert gate.peak <= POOL_SIZE + assert gate.started == POOL_SIZE + 10 + assert not raised(outcomes) + + +@pytest.mark.parametrize( + ("fails", "ends_in"), + [ + (connect_timeout, NetworkError), + (connection_refused, NetworkError), + (unexpected_failure, RuntimeError), + (service_unavailable, ServerError), + ], + ids=["timeout", "connection-error", "unexpected-exception", "5xx-after-retries"], +) +def test_calls_waiting_for_a_slot_go_on_when_the_requests_ahead_fail( + fails, ends_in, respx_mock, client +): + """With the pool full of requests that fail, the 10 calls waiting behind them + still run and fail in turn. + + Args: + fails: What the mocked transport does with each request once the gate + lets it go. + ends_in: The exception every call ends in. + """ + gate = Gate(fails) + respx_mock.get(PRICES_URL).mock(side_effect=gate.handle) + + callers = Callers( + [ + lambda: client.stocks.prices("AAPL", output_format=OutputFormat.JSON) + for _ in range(POOL_SIZE + 10) + ] + ) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + settled = gate.settle() + gate.let_go(ENOUGH) + outcomes = callers.finish() + + assert settled == POOL_SIZE + assert all(isinstance(outcome, ends_in) for outcome in outcomes) + + +def test_retries_asleep_in_their_backoff_do_not_hold_up_another_call( + load_json, respx_mock, client, monkeypatch +): + """With 50 calls asleep between two attempts, one more call still goes + through.""" + body = load_json("stocks_prices_response_200") + attempts = collections.Counter() + attempts_lock = threading.Lock() + asleep = threading.Semaphore(0) + wake_up = threading.Event() + + def answer(request): + """Refuse the first request of each ``SYM`` symbol with a 503. + + Args: + request: The request the client sent. + + Returns: + A 503 for the first request of a ``SYM`` symbol, a 200 otherwise. + """ + symbol = request.url.params["symbols"] + with attempts_lock: + attempts[symbol] += 1 + first = attempts[symbol] == 1 + if first and symbol.startswith("SYM"): + return httpx.Response(503, json={}) + return httpx.Response(200, json=body) + + def sleep_until_woken(seconds): + """Stand in for the backoff: sleep until the test wakes the call up. + + Args: + seconds: How long the retry asked to wait. Ignored. + """ + asleep.release() + wake_up.wait(HOLD) + + monkeypatch.setattr("time.sleep", sleep_until_woken) + respx_mock.get(PRICES_URL).mock(side_effect=answer) + + retrying = Callers( + [ + lambda symbol=f"SYM{index}": client.stocks.prices( + symbol, output_format=OutputFormat.JSON + ) + for index in range(POOL_SIZE) + ] + ) + for _ in range(POOL_SIZE): + assert asleep.acquire(timeout=WAIT), "a retry never reached its backoff" + other = Callers( + [lambda: client.stocks.prices("AAPL", output_format=OutputFormat.JSON)] + ).finish() + wake_up.set() + outcomes = retrying.finish() + + assert not raised(other) + assert not raised(outcomes) + + +def test_a_client_with_every_slot_taken_does_not_hold_up_another_client( + load_json, respx_mock, client +): + """A client whose 50 slots are all taken leaves a second client's call free + to go through.""" + prices = load_json("stocks_prices_response_200") + markets_status = load_json("markets_status_response_200") + gate = Gate(lambda request: httpx.Response(200, json=prices)) + respx_mock.get(PRICES_URL).mock(side_effect=gate.handle) + respx_mock.get(MARKETS_STATUS_URL).respond(json=markets_status, status_code=200) + other = MarketDataClient(token="test") + + saturating = Callers( + [ + lambda: client.stocks.prices("AAPL", output_format=OutputFormat.JSON) + for _ in range(POOL_SIZE) + ] + ) + assert gate.wait_for(lambda g: g.current >= POOL_SIZE), "the pool never filled" + on_other = Callers( + [lambda: other.markets.status(output_format=OutputFormat.JSON)] + ).finish() + gate.let_go(ENOUGH) + outcomes = saturating.finish() + + assert not raised(on_other) + assert not raised(outcomes)